mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-31 01:09:23 +00:00
feat: add an opt-in NATS JetStream publish path for the notify and audit targets (#4634)
feat(targets): add an opt-in NATS JetStream publish path for the notify and audit targets The NATS notify and audit targets publish through NATS Core, which returns before the server has durably accepted the message. A broker restart or a connection drop between the publish and the flush loses the event, even though the send queue has already cleared it, and no acknowledgement gates that clear. An opt-in JetStream publish path clears a queued event only after the server returns a durable PublishAck, so delivery is at-least-once across a broker restart or a reconnect. It applies to both the notify and audit NATS targets, is off by default, and is byte-identical to the NATS Core path when disabled. The path includes durable store-and-forward, a stable dedup id sent as the Nats-Msg-Id header so a replayed event is collapsed by the stream duplicate window, pre-flight stream validation, and a bounded failed-events store for terminally-failed and retry-exhausted events. Three configuration keys per target select it: JETSTREAM_ENABLE, JETSTREAM_STREAM_NAME, and JETSTREAM_ACK_TIMEOUT_SECS, under the RUSTFS_NOTIFY_NATS_ and RUSTFS_AUDIT_NATS_ prefixes. The on-disk batch filename separator changes from colon to underscore so batch names are valid on Windows filesystems, with transparent read-back of files written under the previous separator. The migration affects the shared queue store for every target type and lands with this feature because the store gains its first Windows-exercised paths here. Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
escapecode
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,856 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::plugin::PluginEvent;
|
||||
use crate::{
|
||||
StoreError, Target,
|
||||
arn::TargetID,
|
||||
error::TargetError,
|
||||
runtime::tls::{
|
||||
ReloadableTargetTls, TargetTlsGeneration, TargetTlsInputSet, TlsReloadAdapter, config::ReloadApplyMode,
|
||||
validate_tls_material,
|
||||
},
|
||||
store::{FailedEventStore, Key, QueueStore, Store},
|
||||
target::{
|
||||
ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
|
||||
TargetTlsState, TargetType, build_queued_payload_with_records, build_target_tls_fingerprint, is_connectivity_error,
|
||||
open_target_queue_store_typed, persist_queued_payload_to_store, redacted_secret,
|
||||
},
|
||||
};
|
||||
use async_trait::async_trait;
|
||||
use rustfs_config::{
|
||||
NATS_CREDENTIALS_FILE, NATS_JETSTREAM_ACK_TIMEOUT_DEFAULT_SECS, NATS_TLS_CA, NATS_TLS_CLIENT_CERT, NATS_TLS_CLIENT_KEY,
|
||||
};
|
||||
use std::fmt;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::str::FromStr;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::Duration;
|
||||
use tokio::sync::Mutex;
|
||||
use tracing::{error, info, instrument, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
mod jetstream;
|
||||
mod publish_error;
|
||||
mod validation;
|
||||
|
||||
use jetstream::{CachedJetStreamContext, drain_jetstream_context};
|
||||
use publish_error::{classify_nats_flush_error, classify_nats_publish_error};
|
||||
|
||||
pub(crate) use jetstream::resolve_dedup_id;
|
||||
pub(crate) use validation::{validate_jetstream_settings, validate_jetstream_stream};
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct NATSArgs {
|
||||
pub enable: bool,
|
||||
pub address: String,
|
||||
pub subject: String,
|
||||
pub username: String,
|
||||
pub password: String,
|
||||
pub token: String,
|
||||
pub credentials_file: String,
|
||||
pub tls_ca: String,
|
||||
pub tls_client_cert: String,
|
||||
pub tls_client_key: String,
|
||||
pub tls_required: bool,
|
||||
pub queue_dir: String,
|
||||
pub queue_limit: u64,
|
||||
// JetStream publish settings. Absent maps to off and the defaults.
|
||||
pub jetstream_enable: Option<bool>,
|
||||
pub jetstream_stream_name: Option<String>,
|
||||
pub jetstream_ack_timeout_secs: Option<u64>,
|
||||
pub target_type: TargetType,
|
||||
}
|
||||
|
||||
impl fmt::Debug for NATSArgs {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.debug_struct("NATSArgs")
|
||||
.field("enable", &self.enable)
|
||||
.field("address", &self.address)
|
||||
.field("subject", &self.subject)
|
||||
.field("username", &self.username)
|
||||
.field("password", &redacted_secret(&self.password))
|
||||
.field("token", &redacted_secret(&self.token))
|
||||
.field("credentials_file", &redacted_secret(&self.credentials_file))
|
||||
.field("tls_ca", &self.tls_ca)
|
||||
.field("tls_client_cert", &self.tls_client_cert)
|
||||
.field("tls_client_key", &redacted_secret(&self.tls_client_key))
|
||||
.field("tls_required", &self.tls_required)
|
||||
.field("queue_dir", &self.queue_dir)
|
||||
.field("queue_limit", &self.queue_limit)
|
||||
.field("jetstream_enable", &self.jetstream_enable)
|
||||
.field("jetstream_stream_name", &self.jetstream_stream_name)
|
||||
.field("jetstream_ack_timeout_secs", &self.jetstream_ack_timeout_secs)
|
||||
.field("target_type", &self.target_type)
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
impl NATSArgs {
|
||||
pub fn validate(&self) -> Result<(), TargetError> {
|
||||
if !self.enable {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
validate_nats_address(&self.address)?;
|
||||
validate_nats_auth(self)?;
|
||||
|
||||
if self.subject.trim().is_empty() || self.subject.chars().any(char::is_whitespace) {
|
||||
return Err(TargetError::Configuration(
|
||||
"NATS subject cannot be empty or contain whitespace".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
if !self.credentials_file.is_empty() && !Path::new(&self.credentials_file).is_absolute() {
|
||||
return Err(TargetError::Configuration(format!("{NATS_CREDENTIALS_FILE} must be an absolute path")));
|
||||
}
|
||||
if !self.tls_ca.is_empty() && !Path::new(&self.tls_ca).is_absolute() {
|
||||
return Err(TargetError::Configuration(format!("{NATS_TLS_CA} must be an absolute path")));
|
||||
}
|
||||
if !self.tls_client_cert.is_empty() && !Path::new(&self.tls_client_cert).is_absolute() {
|
||||
return Err(TargetError::Configuration(format!("{NATS_TLS_CLIENT_CERT} must be an absolute path")));
|
||||
}
|
||||
if !self.tls_client_key.is_empty() && !Path::new(&self.tls_client_key).is_absolute() {
|
||||
return Err(TargetError::Configuration(format!("{NATS_TLS_CLIENT_KEY} must be an absolute path")));
|
||||
}
|
||||
if self.tls_client_cert.is_empty() != self.tls_client_key.is_empty() {
|
||||
return Err(TargetError::Configuration(
|
||||
"NATS tls_client_cert and tls_client_key must be specified together".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
if !self.queue_dir.is_empty() && !Path::new(&self.queue_dir).is_absolute() {
|
||||
return Err(TargetError::Configuration("NATS queue directory must be an absolute path".to_string()));
|
||||
}
|
||||
|
||||
validate_jetstream_settings(
|
||||
self.jetstream_enable.unwrap_or(false),
|
||||
self.jetstream_stream_name.as_deref().unwrap_or_default(),
|
||||
&self.queue_dir,
|
||||
self.jetstream_ack_timeout_secs,
|
||||
)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub fn validate_nats_address(address: &str) -> Result<async_nats::ServerAddr, TargetError> {
|
||||
let server = async_nats::ServerAddr::from_str(address)
|
||||
.map_err(|e| TargetError::Configuration(format!("Invalid NATS address: {e}")))?;
|
||||
|
||||
if server.has_user_pass() {
|
||||
return Err(TargetError::Configuration("NATS address must not embed username or password".to_string()));
|
||||
}
|
||||
|
||||
Ok(server)
|
||||
}
|
||||
|
||||
fn validate_nats_auth(args: &NATSArgs) -> Result<(), TargetError> {
|
||||
let mut auth_methods = 0usize;
|
||||
|
||||
if !args.token.is_empty() {
|
||||
auth_methods += 1;
|
||||
}
|
||||
|
||||
if !args.credentials_file.is_empty() {
|
||||
auth_methods += 1;
|
||||
}
|
||||
|
||||
let has_user = !args.username.is_empty();
|
||||
let has_password = !args.password.is_empty();
|
||||
if has_user || has_password {
|
||||
if has_user != has_password {
|
||||
return Err(TargetError::Configuration(
|
||||
"NATS username and password must be specified together".to_string(),
|
||||
));
|
||||
}
|
||||
auth_methods += 1;
|
||||
}
|
||||
|
||||
if auth_methods > 1 {
|
||||
return Err(TargetError::Configuration(
|
||||
"NATS supports only one auth method at a time: token, username/password, or credentials_file".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Returns true when the target sends credentials over a connection that does not require TLS, which
|
||||
/// would transmit the secrets in cleartext (backlog#983). TLS is active when tls_required is set or
|
||||
/// the address uses the tls:// scheme.
|
||||
fn nats_sends_credentials_without_tls(args: &NATSArgs) -> bool {
|
||||
let has_auth = !args.token.is_empty() || !args.credentials_file.is_empty() || !args.username.is_empty();
|
||||
if !has_auth {
|
||||
return false;
|
||||
}
|
||||
let scheme_is_tls = args.address.trim_start().to_ascii_lowercase().starts_with("tls://");
|
||||
!(args.tls_required || scheme_is_tls)
|
||||
}
|
||||
|
||||
pub async fn connect_nats(args: &NATSArgs) -> Result<async_nats::Client, TargetError> {
|
||||
args.validate()?;
|
||||
|
||||
let mut options = async_nats::ConnectOptions::new().require_tls(args.tls_required);
|
||||
|
||||
if !args.token.is_empty() {
|
||||
options = options.token(args.token.clone());
|
||||
} else if !args.username.is_empty() {
|
||||
options = options.user_and_password(args.username.clone(), args.password.clone());
|
||||
} else if !args.credentials_file.is_empty() {
|
||||
options = options
|
||||
.credentials_file(&args.credentials_file)
|
||||
.await
|
||||
.map_err(|e| TargetError::Configuration(format!("Failed to load NATS credentials file: {e}")))?;
|
||||
}
|
||||
|
||||
if !args.tls_ca.is_empty() {
|
||||
options = options.add_root_certificates(PathBuf::from(&args.tls_ca));
|
||||
}
|
||||
if !args.tls_client_cert.is_empty() {
|
||||
options = options.add_client_certificate(PathBuf::from(&args.tls_client_cert), PathBuf::from(&args.tls_client_key));
|
||||
}
|
||||
|
||||
options
|
||||
.connect(args.address.clone())
|
||||
.await
|
||||
.map_err(|e| TargetError::Network(format!("Failed to connect to NATS server: {e}")))
|
||||
}
|
||||
|
||||
pub struct NATSTarget<E>
|
||||
where
|
||||
E: PluginEvent,
|
||||
{
|
||||
id: TargetID,
|
||||
args: NATSArgs,
|
||||
client: Arc<Mutex<Option<async_nats::Client>>>,
|
||||
/// Cached JetStream context, one per target shared across clones so a single acker and semaphore
|
||||
/// serve every clone. Read out under the lock before each publish await, never held across it.
|
||||
/// Carries the stream-validation verdict for the bound connection.
|
||||
jetstream_context: Arc<Mutex<Option<CachedJetStreamContext>>>,
|
||||
tls_state: Arc<parking_lot::Mutex<TargetTlsState>>,
|
||||
/// When set, the coordinator drives TLS reload, otherwise inline fingerprint change detection.
|
||||
tls_adapter: Option<TlsReloadAdapter<async_nats::Client>>,
|
||||
/// The concrete queue store, held typed so the target projects both the generic Store handle and
|
||||
/// its own failed-events capability from a single store. Shared across clones through the Arc.
|
||||
store: Option<Arc<QueueStore<QueuedPayload>>>,
|
||||
connected: AtomicBool,
|
||||
delivery_counters: Arc<TargetDeliveryCounters>,
|
||||
_phantom: std::marker::PhantomData<E>,
|
||||
}
|
||||
|
||||
impl<E> NATSTarget<E>
|
||||
where
|
||||
E: PluginEvent,
|
||||
{
|
||||
pub fn clone_box(&self) -> Box<dyn Target<E> + Send + Sync> {
|
||||
Box::new(NATSTarget::<E> {
|
||||
id: self.id.clone(),
|
||||
args: self.args.clone(),
|
||||
client: Arc::clone(&self.client),
|
||||
jetstream_context: Arc::clone(&self.jetstream_context),
|
||||
tls_state: Arc::clone(&self.tls_state),
|
||||
tls_adapter: self.tls_adapter.clone(),
|
||||
store: self.store.clone(),
|
||||
connected: AtomicBool::new(self.connected.load(Ordering::SeqCst)),
|
||||
delivery_counters: Arc::clone(&self.delivery_counters),
|
||||
_phantom: std::marker::PhantomData,
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(args), fields(target_id_as_string = %id))]
|
||||
pub fn new(id: String, args: NATSArgs) -> Result<Self, TargetError> {
|
||||
args.validate()?;
|
||||
if args.enable && nats_sends_credentials_without_tls(&args) {
|
||||
warn!(
|
||||
target_id = %id,
|
||||
address = %args.address,
|
||||
"NATS target sends authentication credentials without TLS; secrets are transmitted in cleartext. Enable tls_required or use a tls:// address."
|
||||
);
|
||||
}
|
||||
let target_id = TargetID::new(id, ChannelTargetType::Nats.as_str().to_string());
|
||||
let queue_store = open_target_queue_store_typed(
|
||||
&args.queue_dir,
|
||||
args.queue_limit,
|
||||
args.target_type,
|
||||
ChannelTargetType::Nats.as_str(),
|
||||
&target_id,
|
||||
"Failed to open store for NATS target",
|
||||
)?
|
||||
.map(Arc::new);
|
||||
|
||||
Ok(Self {
|
||||
id: target_id,
|
||||
args,
|
||||
client: Arc::new(Mutex::new(None)),
|
||||
jetstream_context: Arc::new(Mutex::new(None)),
|
||||
tls_state: Arc::new(parking_lot::Mutex::new(TargetTlsState::default())),
|
||||
tls_adapter: None,
|
||||
store: queue_store,
|
||||
connected: AtomicBool::new(false),
|
||||
delivery_counters: Arc::new(TargetDeliveryCounters::default()),
|
||||
_phantom: std::marker::PhantomData,
|
||||
})
|
||||
}
|
||||
|
||||
async fn invalidate_cached_client_connection(&self) {
|
||||
*self.client.lock().await = None;
|
||||
}
|
||||
|
||||
async fn get_or_connect(&self) -> Result<async_nats::Client, TargetError> {
|
||||
// Adapter-managed path: use the material directly from the TLS reload adapter.
|
||||
if let Some(adapter) = &self.tls_adapter {
|
||||
let client: async_nats::Client = (*adapter.current_material()).clone();
|
||||
|
||||
// Ensure the client is also stored locally so that close() can drain it.
|
||||
{
|
||||
let mut guard = self.client.lock().await;
|
||||
*guard = Some(client.clone());
|
||||
}
|
||||
return Ok(client);
|
||||
}
|
||||
|
||||
// Inline fingerprint fallback path (no coordinator).
|
||||
let next_fingerprint =
|
||||
build_target_tls_fingerprint(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key).await?;
|
||||
let tls_changed = {
|
||||
let tls_state_guard = self.tls_state.lock();
|
||||
tls_state_guard.needs_update(&next_fingerprint)
|
||||
};
|
||||
if tls_changed {
|
||||
self.invalidate_cached_client_connection().await;
|
||||
}
|
||||
|
||||
{
|
||||
let guard = self.client.lock().await;
|
||||
if let Some(client) = guard.as_ref() {
|
||||
return Ok(client.clone());
|
||||
}
|
||||
}
|
||||
|
||||
let client = connect_nats(&self.args).await?;
|
||||
client
|
||||
.flush()
|
||||
.await
|
||||
.map_err(|e| TargetError::Network(format!("Failed to flush NATS connection: {e}")))?;
|
||||
self.connected.store(true, Ordering::SeqCst);
|
||||
|
||||
let mut client_guard = self.client.lock().await;
|
||||
let shared = client_guard.get_or_insert_with(|| client.clone()).clone();
|
||||
|
||||
// Swap in a context built from the winning client while the client lock is held, so client
|
||||
// and context swap as a unit and no clone reads a context bound to the old client. Only the
|
||||
// JetStream path caches a context. The previous context is drained after the locks release.
|
||||
let previous_context = if tls_changed && self.jetstream_enabled() {
|
||||
let ack_timeout = self.ack_timeout();
|
||||
let mut context_guard = self.jetstream_context.lock().await;
|
||||
let mut context = async_nats::jetstream::new(shared.clone());
|
||||
context.set_timeout(ack_timeout);
|
||||
context_guard.replace(CachedJetStreamContext::new(context))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
drop(client_guard);
|
||||
|
||||
// Advance the recorded fingerprint only after the reconnect and flush succeed, so a rotation
|
||||
// whose first reconnect fails stays detected and the next success still rebuilds the context.
|
||||
if tls_changed {
|
||||
self.tls_state.lock().refresh(next_fingerprint);
|
||||
}
|
||||
|
||||
drain_jetstream_context(previous_context.map(|cached| cached.context)).await;
|
||||
Ok(shared)
|
||||
}
|
||||
|
||||
fn jetstream_enabled(&self) -> bool {
|
||||
self.args.jetstream_enable.unwrap_or(false)
|
||||
}
|
||||
|
||||
fn ack_timeout(&self) -> Duration {
|
||||
Duration::from_secs(
|
||||
self.args
|
||||
.jetstream_ack_timeout_secs
|
||||
.unwrap_or(NATS_JETSTREAM_ACK_TIMEOUT_DEFAULT_SECS),
|
||||
)
|
||||
}
|
||||
|
||||
fn build_queued_payload(&self, event: &EntityTarget<E>) -> Result<QueuedPayload, TargetError> {
|
||||
let mut queued = build_queued_payload_with_records(event, vec![event.clone()])?;
|
||||
if self.jetstream_enabled() {
|
||||
// Mint the dedup id once at enqueue so it is identical across every retry and replay and
|
||||
// distinct across entries. The body-hash fallback only covers entries queued before enable.
|
||||
queued.meta.dedup_id = Uuid::new_v4().to_string();
|
||||
}
|
||||
Ok(queued)
|
||||
}
|
||||
|
||||
async fn send_body(&self, body: Vec<u8>) -> Result<(), TargetError> {
|
||||
let client = self.get_or_connect().await?;
|
||||
if let Err(e) = client.publish(self.args.subject.clone(), body.into()).await {
|
||||
let err = classify_nats_publish_error(&e);
|
||||
if is_connectivity_error(&err) {
|
||||
self.invalidate_cached_client_connection().await;
|
||||
self.connected.store(false, Ordering::SeqCst);
|
||||
}
|
||||
return Err(err);
|
||||
}
|
||||
|
||||
// publish only enqueues the message on the client's outbound channel. Flush to confirm the
|
||||
// message reached the server before delivery is treated as successful (backlog#971).
|
||||
if let Err(e) = client.flush().await {
|
||||
let err = classify_nats_flush_error(&e);
|
||||
self.invalidate_cached_client_connection().await;
|
||||
self.connected.store(false, Ordering::SeqCst);
|
||||
return Err(err);
|
||||
}
|
||||
|
||||
self.delivery_counters.record_success();
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl<E> Target<E> for NATSTarget<E>
|
||||
where
|
||||
E: PluginEvent,
|
||||
{
|
||||
fn id(&self) -> TargetID {
|
||||
self.id.clone()
|
||||
}
|
||||
|
||||
async fn is_active(&self) -> Result<bool, TargetError> {
|
||||
// With JetStream enabled the health answer covers the stream as well as the connection, so a
|
||||
// reachable broker with a failing stream reports the validation error. The verdict is cached
|
||||
// and reset on a reconnect, TLS rotation, wrong-stream ack, or stream-not-found outcome, so a
|
||||
// reset forces a live lookup on the next check.
|
||||
if self.jetstream_enabled() {
|
||||
self.validated_jetstream_context().await?;
|
||||
}
|
||||
let client = self.get_or_connect().await?;
|
||||
client
|
||||
.flush()
|
||||
.await
|
||||
.map_err(|e| TargetError::Network(format!("NATS health check failed: {e}")))?;
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
async fn save(&self, event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
|
||||
let queued = match self.build_queued_payload(&event) {
|
||||
Ok(queued) => queued,
|
||||
Err(err) => {
|
||||
self.delivery_counters.record_final_failure();
|
||||
return Err(err);
|
||||
}
|
||||
};
|
||||
|
||||
if let Some(store) = &self.store {
|
||||
if let Err(e) = persist_queued_payload_to_store(store.as_ref(), &queued) {
|
||||
self.delivery_counters.record_final_failure();
|
||||
return Err(e);
|
||||
}
|
||||
Ok(())
|
||||
} else {
|
||||
if let Err(err) = self.send_body(queued.body).await {
|
||||
self.delivery_counters.record_final_failure();
|
||||
return Err(err);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
async fn send_raw_from_store(&self, key: Key, body: Vec<u8>, meta: QueuedPayloadMeta) -> Result<(), TargetError> {
|
||||
if self.jetstream_enabled() {
|
||||
let dedup_id = resolve_dedup_id(&meta.dedup_id, &key);
|
||||
self.publish_jetstream(body, &dedup_id).await
|
||||
} else {
|
||||
self.send_body(body).await
|
||||
}
|
||||
}
|
||||
|
||||
async fn close(&self) -> Result<(), TargetError> {
|
||||
// Drain the cached context's in-flight acks before dropping it so the async-nats acker exits.
|
||||
// Drain the context before the client, since the acker rides the client connection.
|
||||
let context = {
|
||||
let mut guard = self.jetstream_context.lock().await;
|
||||
guard.take()
|
||||
};
|
||||
// Bound the drain by the ack timeout so an unresponsive broker cannot stall close. On elapse
|
||||
// the client drain below still runs.
|
||||
if tokio::time::timeout(self.ack_timeout(), drain_jetstream_context(context.map(|cached| cached.context)))
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
warn!(target_id = %self.id, "Timed out draining JetStream acks on close, proceeding to close the client");
|
||||
}
|
||||
|
||||
let client = {
|
||||
let mut guard = self.client.lock().await;
|
||||
guard.take()
|
||||
};
|
||||
self.tls_state.lock().reset();
|
||||
self.connected.store(false, Ordering::SeqCst);
|
||||
if let Some(client) = client {
|
||||
client
|
||||
.drain()
|
||||
.await
|
||||
.map_err(|e| TargetError::Network(format!("Failed to drain NATS client: {e}")))?;
|
||||
}
|
||||
// The durable queue store stays on disk, so a queued entry survives close and replays. Close
|
||||
// releases the cached handles, never the entries.
|
||||
info!(target_id = %self.id, "NATS target closed");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
|
||||
self.store
|
||||
.as_deref()
|
||||
.map(|store| store as &(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync))
|
||||
}
|
||||
|
||||
fn failed_store(&self) -> Option<&dyn FailedEventStore> {
|
||||
self.store.as_deref().map(|store| store as &dyn FailedEventStore)
|
||||
}
|
||||
|
||||
async fn handle_terminal_failure(
|
||||
&self,
|
||||
store: &(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send),
|
||||
key: &Key,
|
||||
error: &TargetError,
|
||||
retry_count: u32,
|
||||
) -> bool {
|
||||
let Some(failed_store) = self.failed_store() else {
|
||||
error!(
|
||||
target_id = %self.id,
|
||||
replay_key = %key,
|
||||
"NATS target has no failed-events store for the terminal move"
|
||||
);
|
||||
return false;
|
||||
};
|
||||
match jetstream::move_entry_to_failed_store(store, failed_store, &self.id, key, error, retry_count).await {
|
||||
Ok(()) => true,
|
||||
Err(move_err) => {
|
||||
error!(
|
||||
target_id = %self.id,
|
||||
error = %move_err,
|
||||
replay_key = %key,
|
||||
"Failed to move event to the failed-events store"
|
||||
);
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn clone_dyn(&self) -> Box<dyn Target<E> + Send + Sync> {
|
||||
self.clone_box()
|
||||
}
|
||||
|
||||
async fn init(&self) -> Result<(), TargetError> {
|
||||
if !self.is_enabled() {
|
||||
return Ok(());
|
||||
}
|
||||
let _ = self.get_or_connect().await?;
|
||||
if self.jetstream_enabled() {
|
||||
let cached = self.jetstream_context().await?;
|
||||
validate_jetstream_stream(&cached.context, &self.args, &self.id.to_string(), Some(&cached.validation_logged)).await?;
|
||||
// Record the verdict on the context so the first publish does not repeat the stream
|
||||
// lookup that just passed.
|
||||
cached.stream_validated.store(true, Ordering::Release);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn is_enabled(&self) -> bool {
|
||||
self.args.enable
|
||||
}
|
||||
|
||||
fn delivery_snapshot(&self) -> TargetDeliverySnapshot {
|
||||
self.delivery_counters.snapshot(
|
||||
self.store.as_deref().map_or(0, |store| store.len() as u64),
|
||||
self.failed_store().map_or(0, |failed_store| failed_store.failed_len() as u64),
|
||||
)
|
||||
}
|
||||
|
||||
fn record_final_failure(&self) {
|
||||
self.delivery_counters.record_final_failure();
|
||||
}
|
||||
}
|
||||
|
||||
/// Coordinated TLS hot-reload implementation for NATS targets.
|
||||
///
|
||||
/// The coordinator calls these methods on a background poll loop to detect
|
||||
/// TLS file changes and rebuild the NATS client without restarting.
|
||||
#[async_trait]
|
||||
impl<E> ReloadableTargetTls for NATSTarget<E>
|
||||
where
|
||||
E: PluginEvent,
|
||||
{
|
||||
type Material = async_nats::Client;
|
||||
|
||||
fn tls_input_set(&self) -> TargetTlsInputSet {
|
||||
TargetTlsInputSet {
|
||||
ca_path: self.args.tls_ca.clone(),
|
||||
client_cert_path: self.args.tls_client_cert.clone(),
|
||||
client_key_path: self.args.tls_client_key.clone(),
|
||||
target_label: format!("nats:{}", self.id.id),
|
||||
}
|
||||
}
|
||||
|
||||
async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
|
||||
connect_nats(&self.args).await
|
||||
}
|
||||
|
||||
async fn apply_tls_material(
|
||||
&self,
|
||||
_generation: TargetTlsGeneration,
|
||||
material: Arc<Self::Material>,
|
||||
_mode: ReloadApplyMode,
|
||||
) -> Result<(), TargetError> {
|
||||
let mut guard = self.client.lock().await;
|
||||
*guard = Some((*material).clone());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn validate_tls_files(&self) -> Result<(), TargetError> {
|
||||
validate_tls_material(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) mod test_support {
|
||||
use super::*;
|
||||
use async_nats::jetstream;
|
||||
use rustfs_s3_types::EventName;
|
||||
|
||||
// Absolute on Linux, macOS, and Windows. temp_dir needs no filesystem to exist for a
|
||||
// validation-only test, and Path::is_absolute stays true across platforms.
|
||||
pub(crate) fn nats_queue_dir() -> String {
|
||||
std::env::temp_dir().join("rustfs-nats-queue").to_string_lossy().into_owned()
|
||||
}
|
||||
|
||||
pub(crate) fn base_args() -> NATSArgs {
|
||||
NATSArgs {
|
||||
enable: true,
|
||||
address: "nats://127.0.0.1:4222".to_string(),
|
||||
subject: "rustfs.events".to_string(),
|
||||
username: String::new(),
|
||||
password: String::new(),
|
||||
token: String::new(),
|
||||
credentials_file: String::new(),
|
||||
tls_ca: String::new(),
|
||||
tls_client_cert: String::new(),
|
||||
tls_client_key: String::new(),
|
||||
tls_required: false,
|
||||
queue_dir: String::new(),
|
||||
queue_limit: 0,
|
||||
jetstream_enable: None,
|
||||
jetstream_stream_name: None,
|
||||
jetstream_ack_timeout_secs: None,
|
||||
target_type: TargetType::NotifyEvent,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn jetstream_args(target_type: TargetType) -> NATSArgs {
|
||||
NATSArgs {
|
||||
jetstream_enable: Some(true),
|
||||
jetstream_stream_name: Some("RUSTFS_EVENTS".to_string()),
|
||||
jetstream_ack_timeout_secs: Some(30),
|
||||
target_type,
|
||||
..base_args()
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn nats_target(args: NATSArgs) -> NATSTarget<String> {
|
||||
NATSTarget::<String> {
|
||||
id: TargetID::new("test-target".to_string(), ChannelTargetType::Nats.as_str().to_string()),
|
||||
args,
|
||||
client: Arc::new(Mutex::new(None)),
|
||||
jetstream_context: Arc::new(Mutex::new(None)),
|
||||
tls_state: Arc::new(parking_lot::Mutex::new(TargetTlsState::default())),
|
||||
tls_adapter: None,
|
||||
store: None,
|
||||
connected: AtomicBool::new(false),
|
||||
delivery_counters: Arc::new(TargetDeliveryCounters::default()),
|
||||
_phantom: std::marker::PhantomData,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn sample_event() -> EntityTarget<String> {
|
||||
EntityTarget {
|
||||
object_name: "folder/object.txt".to_string(),
|
||||
bucket_name: "bucket-a".to_string(),
|
||||
event_name: EventName::ObjectCreatedPut,
|
||||
data: "payload-data".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn sample_stored_key(name: &str) -> Key {
|
||||
Key {
|
||||
name: name.to_string(),
|
||||
extension: ".event".to_string(),
|
||||
item_count: 1,
|
||||
compress: false,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn nats_target_with_store(args: NATSArgs, queue_dir: &str) -> NATSTarget<String> {
|
||||
let mut configured = args;
|
||||
configured.queue_dir = queue_dir.to_string();
|
||||
NATSTarget::<String>::new("store-target".to_string(), configured).expect("target with store builds")
|
||||
}
|
||||
|
||||
/// A stream configuration that passes every assertion: it captures the configured subject, returns
|
||||
/// acks, accepts writes, and deduplicates well beyond the retry lifetime.
|
||||
pub(crate) fn writable_stream_config(subject: &str) -> jetstream::stream::Config {
|
||||
jetstream::stream::Config {
|
||||
name: "RUSTFS_EVENTS".to_string(),
|
||||
subjects: vec![subject.to_string()],
|
||||
no_ack: false,
|
||||
sealed: false,
|
||||
duplicate_window: Duration::from_secs(600),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
// Live-broker behaviour tests. They require a NATS server with JetStream enabled and are ignored by
|
||||
// default. To run locally:
|
||||
//
|
||||
// docker run -d --name rustfs-nats-test -p 4222:4222 nats:2 -js
|
||||
// cargo test -p rustfs-targets --lib -- --ignored nats::tests::tls_change
|
||||
//
|
||||
// Override the server URL with RUSTFS_TEST_NATS_URL.
|
||||
pub(crate) fn broker_url() -> String {
|
||||
std::env::var("RUSTFS_TEST_NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string())
|
||||
}
|
||||
|
||||
/// Log sink for asserting a specific line is written. The subscriber writes into a shared
|
||||
/// buffer the assertion reads back.
|
||||
#[derive(Clone, Default)]
|
||||
pub(crate) struct CapturedLog(Arc<parking_lot::Mutex<Vec<u8>>>);
|
||||
|
||||
impl CapturedLog {
|
||||
pub(crate) fn contents(&self) -> String {
|
||||
String::from_utf8_lossy(&self.0.lock()).into_owned()
|
||||
}
|
||||
}
|
||||
|
||||
impl std::io::Write for CapturedLog {
|
||||
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
||||
self.0.lock().extend_from_slice(buf);
|
||||
Ok(buf.len())
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> std::io::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturedLog {
|
||||
type Writer = CapturedLog;
|
||||
|
||||
fn make_writer(&'a self) -> Self::Writer {
|
||||
self.clone()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::target::REDACTED_SECRET;
|
||||
use crate::target::nats::test_support::*;
|
||||
|
||||
#[test]
|
||||
fn debug_redacts_nats_secret_fields() {
|
||||
let args = NATSArgs {
|
||||
password: "nats-password".to_string(),
|
||||
token: "nats-token".to_string(),
|
||||
credentials_file: "/etc/rustfs/nats.creds".to_string(),
|
||||
tls_client_key: "/etc/rustfs/nats.key".to_string(),
|
||||
..base_args()
|
||||
};
|
||||
|
||||
let rendered = format!("{args:?}");
|
||||
|
||||
assert!(!rendered.contains("nats-password"));
|
||||
assert!(!rendered.contains("nats-token"));
|
||||
assert!(!rendered.contains("/etc/rustfs/nats.creds"));
|
||||
assert!(!rendered.contains("/etc/rustfs/nats.key"));
|
||||
assert!(rendered.contains(REDACTED_SECRET));
|
||||
assert!(rendered.contains("rustfs.events"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_nats_rejects_multiple_auth_methods() {
|
||||
let args = NATSArgs {
|
||||
token: "abc".to_string(),
|
||||
username: "user".to_string(),
|
||||
password: "pass".to_string(),
|
||||
..base_args()
|
||||
};
|
||||
assert!(args.validate().is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_nats_rejects_relative_queue_dir() {
|
||||
let args = NATSArgs {
|
||||
queue_dir: "relative/path".to_string(),
|
||||
..base_args()
|
||||
};
|
||||
assert!(args.validate().is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nats_credentials_without_tls_is_detected() {
|
||||
// Token auth over a plaintext nats:// address without tls_required leaks the credential (backlog#983).
|
||||
let insecure = NATSArgs {
|
||||
token: "secret-token".to_string(),
|
||||
tls_required: false,
|
||||
..base_args()
|
||||
};
|
||||
assert!(nats_sends_credentials_without_tls(&insecure));
|
||||
|
||||
// tls_required protects the credentials.
|
||||
let with_tls = NATSArgs {
|
||||
tls_required: true,
|
||||
..insecure.clone()
|
||||
};
|
||||
assert!(!nats_sends_credentials_without_tls(&with_tls));
|
||||
|
||||
// A tls:// address also counts as protected.
|
||||
let tls_scheme = NATSArgs {
|
||||
address: "tls://127.0.0.1:4222".to_string(),
|
||||
..insecure
|
||||
};
|
||||
assert!(!nats_sends_credentials_without_tls(&tls_scheme));
|
||||
|
||||
// No credentials configured: nothing to leak.
|
||||
let no_auth = base_args();
|
||||
assert!(!nats_sends_credentials_without_tls(&no_auth));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn flag_off_stored_meta_is_byte_identical_to_pre_feature() {
|
||||
// With the flag off the payload carries no dedup id, so its encoded meta matches a pre-feature entry.
|
||||
let disabled = nats_target(base_args());
|
||||
let off_payload = disabled.build_queued_payload(&sample_event()).expect("payload builds");
|
||||
assert!(off_payload.meta.dedup_id.is_empty(), "no dedup id is minted with the flag off");
|
||||
|
||||
let meta_json = serde_json::to_string(&off_payload.meta).expect("meta serializes");
|
||||
assert!(!meta_json.contains("dedup_id"), "the absent dedup id is skipped on serialization");
|
||||
|
||||
// No JetStream context is built on the flag-off path.
|
||||
assert!(disabled.jetstream_context.lock().await.is_none(), "no context is built with the flag off");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,346 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::jetstream::STREAM_NOT_FOUND_DETAIL;
|
||||
use crate::error::TargetError;
|
||||
use async_nats::jetstream::context::PublishErrorKind;
|
||||
|
||||
/// Maps an async-nats publish error to a typed TargetError with the retryable-versus-terminal
|
||||
/// classification, so the replay loop retries recoverable errors instead of dropping them. The
|
||||
/// detail comes from the fixed vocabulary in publish_error_detail, never from raw error text.
|
||||
pub(crate) fn classify_publish_error(err: &async_nats::jetstream::context::PublishError) -> TargetError {
|
||||
let retryable = match err.kind() {
|
||||
PublishErrorKind::TimedOut
|
||||
| PublishErrorKind::BrokenPipe
|
||||
| PublishErrorKind::MaxAckPending
|
||||
| PublishErrorKind::StreamNotFound => true,
|
||||
PublishErrorKind::MaxPayloadExceeded | PublishErrorKind::WrongLastMessageId | PublishErrorKind::WrongLastSequence => {
|
||||
false
|
||||
}
|
||||
PublishErrorKind::Other => !other_error_is_terminal(err),
|
||||
};
|
||||
TargetError::JetStreamPublish {
|
||||
retryable,
|
||||
detail: publish_error_detail(err),
|
||||
}
|
||||
}
|
||||
|
||||
/// Fixed diagnostic detail for a publish failure. Each error kind maps to a constant label, and a
|
||||
/// server rejection carries the numeric JetStream error code and status. Never the library rendering
|
||||
/// or any server-returned text, so the detail is safe to persist in a failed-store entry.
|
||||
fn publish_error_detail(err: &async_nats::jetstream::context::PublishError) -> String {
|
||||
use async_nats::jetstream::Error as JetStreamApiError;
|
||||
match err.kind() {
|
||||
PublishErrorKind::StreamNotFound => STREAM_NOT_FOUND_DETAIL.to_string(),
|
||||
PublishErrorKind::TimedOut => "publish timed out".to_string(),
|
||||
PublishErrorKind::BrokenPipe => "broken pipe".to_string(),
|
||||
PublishErrorKind::MaxAckPending => "max ack pending reached".to_string(),
|
||||
PublishErrorKind::MaxPayloadExceeded => "max payload exceeded".to_string(),
|
||||
PublishErrorKind::WrongLastMessageId => "wrong last message id".to_string(),
|
||||
PublishErrorKind::WrongLastSequence => "wrong last sequence".to_string(),
|
||||
PublishErrorKind::Other => {
|
||||
match std::error::Error::source(err).and_then(|source| source.downcast_ref::<JetStreamApiError>()) {
|
||||
Some(api_error) => format!("server error code {} status {}", api_error.error_code().0, api_error.code()),
|
||||
None => "publish failed".to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// HTTP-style 5xx status range. A rejection reporting a 5xx status is a server-side condition and
|
||||
/// stays retryable regardless of its error code.
|
||||
const SERVER_ERROR_STATUS_RANGE: std::ops::Range<usize> = 500..600;
|
||||
|
||||
/// Classifies an Other publish error against a terminal allowlist. Terminal only for an explicit
|
||||
/// permanent error code, STREAM_SEALED being the sole current member. Everything else is retryable,
|
||||
/// so only a recognized permanent rejection enters the failed store.
|
||||
fn other_error_is_terminal(err: &async_nats::jetstream::context::PublishError) -> bool {
|
||||
use async_nats::jetstream::{Error as JetStreamApiError, ErrorCode};
|
||||
|
||||
let Some(api_error) = std::error::Error::source(err).and_then(|source| source.downcast_ref::<JetStreamApiError>()) else {
|
||||
return false;
|
||||
};
|
||||
|
||||
// A 5xx status is a server-side condition, so it stays retryable even when the error code is a
|
||||
// permanent one.
|
||||
if SERVER_ERROR_STATUS_RANGE.contains(&api_error.code()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
matches!(api_error.error_code(), ErrorCode::STREAM_SEALED)
|
||||
}
|
||||
|
||||
/// Classifies a NATS publish failure by its typed error kind. Send means the outbound channel is
|
||||
/// closed and is retriable. Payload and subject violations are permanent request-level errors
|
||||
/// (backlog#971, backlog#973).
|
||||
pub(crate) fn classify_nats_publish_error(err: &async_nats::PublishError) -> TargetError {
|
||||
use async_nats::PublishErrorKind;
|
||||
match err.kind() {
|
||||
PublishErrorKind::Send => TargetError::NotConnected,
|
||||
PublishErrorKind::MaxPayloadExceeded | PublishErrorKind::InvalidSubject => {
|
||||
TargetError::Request(format!("Failed to publish NATS message: {err}"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Classifies a NATS flush failure. Both kinds mean the message was not confirmed on the wire, so
|
||||
/// the event is kept for replay (backlog#971, backlog#973).
|
||||
pub(crate) fn classify_nats_flush_error(err: &async_nats::client::FlushError) -> TargetError {
|
||||
use async_nats::client::FlushErrorKind;
|
||||
match err.kind() {
|
||||
FlushErrorKind::SendError | FlushErrorKind::FlushError => TargetError::NotConnected,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::target::nats::jetstream::{ACK_STREAM_MISMATCH_DETAIL, ATTEMPT_DEADLINE_DETAIL};
|
||||
use crate::target::nats::validation::gate_publish_on_stream_validation;
|
||||
use async_nats::jetstream::context::{PublishError, PublishErrorKind};
|
||||
use std::sync::atomic::AtomicBool;
|
||||
|
||||
#[test]
|
||||
fn nats_publish_error_classification() {
|
||||
use async_nats::PublishErrorKind;
|
||||
// Send means the outbound channel is gone: retriable connectivity error.
|
||||
let send_err = async_nats::PublishError::new(PublishErrorKind::Send);
|
||||
assert!(matches!(classify_nats_publish_error(&send_err), TargetError::NotConnected));
|
||||
|
||||
// Payload and subject violations are permanent request-level errors.
|
||||
let payload_err = async_nats::PublishError::new(PublishErrorKind::MaxPayloadExceeded);
|
||||
assert!(matches!(classify_nats_publish_error(&payload_err), TargetError::Request(_)));
|
||||
let subject_err = async_nats::PublishError::new(PublishErrorKind::InvalidSubject);
|
||||
assert!(matches!(classify_nats_publish_error(&subject_err), TargetError::Request(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nats_flush_error_classification() {
|
||||
use async_nats::client::{FlushError, FlushErrorKind};
|
||||
for kind in [FlushErrorKind::SendError, FlushErrorKind::FlushError] {
|
||||
assert!(matches!(classify_nats_flush_error(&FlushError::new(kind)), TargetError::NotConnected));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_publish_error_marks_recoverable_kinds_retryable() {
|
||||
for kind in [
|
||||
PublishErrorKind::TimedOut,
|
||||
PublishErrorKind::BrokenPipe,
|
||||
PublishErrorKind::MaxAckPending,
|
||||
PublishErrorKind::StreamNotFound,
|
||||
] {
|
||||
let classified = classify_publish_error(&PublishError::new(kind));
|
||||
match classified {
|
||||
TargetError::JetStreamPublish { retryable, .. } => assert!(retryable, "{kind} is retryable"),
|
||||
other => panic!("expected JetStreamPublish, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_publish_error_marks_terminal_kinds_terminal() {
|
||||
for kind in [
|
||||
PublishErrorKind::MaxPayloadExceeded,
|
||||
PublishErrorKind::WrongLastMessageId,
|
||||
PublishErrorKind::WrongLastSequence,
|
||||
] {
|
||||
let classified = classify_publish_error(&PublishError::new(kind));
|
||||
match classified {
|
||||
TargetError::JetStreamPublish { retryable, .. } => assert!(!retryable, "{kind} is terminal"),
|
||||
other => panic!("expected JetStreamPublish, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_publish_error_other_without_source_is_retryable() {
|
||||
// An undecodable Other carries no code to recognize as permanent, so it stays on the retry path.
|
||||
let classified = classify_publish_error(&PublishError::new(PublishErrorKind::Other));
|
||||
match classified {
|
||||
TargetError::JetStreamPublish { retryable, .. } => assert!(retryable, "an undecodable Other is retryable"),
|
||||
other => panic!("expected JetStreamPublish, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_publish_error_unknown_code_is_retryable() {
|
||||
// A decoded rejection carrying a code outside the terminal allowlist is retryable.
|
||||
let api_error: async_nats::jetstream::Error =
|
||||
serde_json::from_value(serde_json::json!({"code": 400, "err_code": 10999, "description": "an unrecognized code"}))
|
||||
.expect("api error deserializes");
|
||||
let publish_error = PublishError::with_source(PublishErrorKind::Other, api_error);
|
||||
match classify_publish_error(&publish_error) {
|
||||
TargetError::JetStreamPublish { retryable, .. } => assert!(retryable, "an unknown error code is retryable"),
|
||||
other => panic!("expected JetStreamPublish, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_publish_error_other_with_transient_code_is_retryable() {
|
||||
// Server conditions that each clear on their own. None is in the terminal allowlist, so each
|
||||
// is retryable. The status is 400 in every case, so retryability comes from the allowlist
|
||||
// policy rather than the 5xx status range.
|
||||
for (err_code, condition) in [
|
||||
(10002, "account resources exceeded"),
|
||||
(10008, "cluster not available"),
|
||||
(10009, "cluster not leader"),
|
||||
(10028, "memory resources exceeded"),
|
||||
(10039, "jetstream not enabled for account"),
|
||||
(10040, "cluster peer not a member"),
|
||||
(10041, "raft general error"),
|
||||
(10076, "jetstream not enabled"),
|
||||
(10118, "stream offline"),
|
||||
(10194, "stream offline with a reason"),
|
||||
(10202, "server member change in flight"),
|
||||
] {
|
||||
let api_error: async_nats::jetstream::Error =
|
||||
serde_json::from_value(serde_json::json!({"code": 400, "err_code": err_code, "description": condition}))
|
||||
.expect("api error deserializes");
|
||||
let publish_error = PublishError::with_source(PublishErrorKind::Other, api_error);
|
||||
let classified = classify_publish_error(&publish_error);
|
||||
match classified {
|
||||
TargetError::JetStreamPublish { retryable, .. } => {
|
||||
assert!(retryable, "transient code {err_code} ({condition}) is retryable")
|
||||
}
|
||||
other => panic!("expected JetStreamPublish for code {err_code}, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_publish_error_uncoded_server_error_status_is_retryable() {
|
||||
// A rejection with no err_code and a 5xx status is a server-side condition, kept on the retry path.
|
||||
for status in [500, 503] {
|
||||
let api_error: async_nats::jetstream::Error =
|
||||
serde_json::from_value(serde_json::json!({"code": status, "description": "insufficient resources"}))
|
||||
.expect("api error deserializes");
|
||||
let publish_error = PublishError::with_source(PublishErrorKind::Other, api_error);
|
||||
let classified = classify_publish_error(&publish_error);
|
||||
match classified {
|
||||
TargetError::JetStreamPublish { retryable, .. } => {
|
||||
assert!(retryable, "an uncoded {status} status is retryable")
|
||||
}
|
||||
other => panic!("expected JetStreamPublish for status {status}, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_publish_error_other_with_terminal_code_is_terminal() {
|
||||
// STREAM_SEALED (10109) is the sole terminal-allowlist member and its 400 status keeps it terminal.
|
||||
let api_error: async_nats::jetstream::Error =
|
||||
serde_json::from_value(serde_json::json!({"code": 400, "err_code": 10109, "description": "stream sealed"}))
|
||||
.expect("api error deserializes");
|
||||
let publish_error = PublishError::with_source(PublishErrorKind::Other, api_error);
|
||||
match classify_publish_error(&publish_error) {
|
||||
TargetError::JetStreamPublish { retryable, .. } => assert!(!retryable, "a sealed stream is terminal"),
|
||||
other => panic!("expected JetStreamPublish, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_publish_error_uncoded_4xx_is_retryable() {
|
||||
// An uncoded 4xx rejection carries no permanent code, so the allowlist policy leaves it on the retry path.
|
||||
let api_error: async_nats::jetstream::Error =
|
||||
serde_json::from_value(serde_json::json!({"code": 400, "description": "bad request without an error code"}))
|
||||
.expect("api error deserializes");
|
||||
let publish_error = PublishError::with_source(PublishErrorKind::Other, api_error);
|
||||
match classify_publish_error(&publish_error) {
|
||||
TargetError::JetStreamPublish { retryable, .. } => assert!(retryable, "an uncoded 4xx rejection is retryable"),
|
||||
other => panic!("expected JetStreamPublish, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_publish_error_terminal_code_with_server_error_status_is_retryable() {
|
||||
// Deliberate policy: a 5xx status classifies retryable even when the error code is terminal.
|
||||
let api_error: async_nats::jetstream::Error =
|
||||
serde_json::from_value(serde_json::json!({"code": 503, "err_code": 10109, "description": "stream sealed"}))
|
||||
.expect("api error deserializes");
|
||||
let publish_error = PublishError::with_source(PublishErrorKind::Other, api_error);
|
||||
let classified = classify_publish_error(&publish_error);
|
||||
match classified {
|
||||
TargetError::JetStreamPublish { retryable, .. } => {
|
||||
assert!(retryable, "a 5xx status keeps the STREAM_SEALED code on the retry path")
|
||||
}
|
||||
other => panic!("expected JetStreamPublish, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_publish_error_detail_omits_server_body() {
|
||||
// A server rejection contributes its numeric code and status, never the server-returned description.
|
||||
let api_error: async_nats::jetstream::Error = serde_json::from_value(
|
||||
serde_json::json!({"code": 503, "err_code": 10008, "description": "secret-token-abc123 in description"}),
|
||||
)
|
||||
.expect("api error deserializes");
|
||||
let classified = classify_publish_error(&PublishError::with_source(PublishErrorKind::Other, api_error));
|
||||
match classified {
|
||||
TargetError::JetStreamPublish { detail, .. } => {
|
||||
assert_eq!(detail, "server error code 10008 status 503");
|
||||
}
|
||||
other => panic!("expected JetStreamPublish, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classified_publish_details_use_the_fixed_vocabulary() {
|
||||
// Every classification path draws its detail from fixed labels and numeric fields. The pattern
|
||||
// (lowercase letters, digits, spaces, underscores, colons) rejects raw error text.
|
||||
let is_fixed = |detail: &str| {
|
||||
!detail.is_empty()
|
||||
&& detail
|
||||
.chars()
|
||||
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || matches!(c, ' ' | '_' | ':'))
|
||||
};
|
||||
|
||||
let mut details = vec![ATTEMPT_DEADLINE_DETAIL.to_string(), ACK_STREAM_MISMATCH_DETAIL.to_string()];
|
||||
for kind in [
|
||||
PublishErrorKind::StreamNotFound,
|
||||
PublishErrorKind::TimedOut,
|
||||
PublishErrorKind::BrokenPipe,
|
||||
PublishErrorKind::MaxAckPending,
|
||||
PublishErrorKind::MaxPayloadExceeded,
|
||||
PublishErrorKind::WrongLastMessageId,
|
||||
PublishErrorKind::WrongLastSequence,
|
||||
PublishErrorKind::Other,
|
||||
] {
|
||||
match classify_publish_error(&PublishError::new(kind)) {
|
||||
TargetError::JetStreamPublish { detail, .. } => details.push(detail),
|
||||
other => panic!("expected JetStreamPublish, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
let api_error: async_nats::jetstream::Error =
|
||||
serde_json::from_value(serde_json::json!({"code": 503, "err_code": 10077, "description": "raw server text"}))
|
||||
.expect("api error deserializes");
|
||||
match classify_publish_error(&PublishError::with_source(PublishErrorKind::Other, api_error)) {
|
||||
TargetError::JetStreamPublish { detail, .. } => details.push(detail),
|
||||
other => panic!("expected JetStreamPublish, got {other:?}"),
|
||||
}
|
||||
|
||||
// The validation gate detail must stay fixed even when the verdict error carries text outside the vocabulary.
|
||||
let stream_validated = AtomicBool::new(false);
|
||||
let verdict = Err(TargetError::Configuration("Raw Display TEXT with (punctuation)".to_string()));
|
||||
match gate_publish_on_stream_validation(verdict, &stream_validated) {
|
||||
Err(TargetError::JetStreamPublish { detail, .. }) => details.push(detail),
|
||||
other => panic!("expected a JetStreamPublish gate error, got {other:?}"),
|
||||
}
|
||||
|
||||
for detail in details {
|
||||
assert!(is_fixed(&detail), "detail falls outside the fixed vocabulary: {detail:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,405 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::NATSArgs;
|
||||
use super::jetstream::retry_lifetime;
|
||||
use crate::error::TargetError;
|
||||
use async_nats::jetstream;
|
||||
use rustfs_config::{
|
||||
NATS_JETSTREAM_ACK_TIMEOUT_DEFAULT_SECS, NATS_JETSTREAM_ACK_TIMEOUT_MAX_SECS, NATS_JETSTREAM_ACK_TIMEOUT_MIN_SECS,
|
||||
NATS_JETSTREAM_ACK_TIMEOUT_SECS, NATS_JETSTREAM_ENABLE, NATS_JETSTREAM_STREAM_NAME, NATS_QUEUE_DIR,
|
||||
};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::Duration;
|
||||
use tracing::{debug, error, info};
|
||||
|
||||
/// Validates the JetStream publish settings shared by the config-time and connect-time validators,
|
||||
/// so both reject the same combinations.
|
||||
pub(crate) fn validate_jetstream_settings(
|
||||
enable: bool,
|
||||
stream_name: &str,
|
||||
queue_dir: &str,
|
||||
ack_timeout_secs: Option<u64>,
|
||||
) -> Result<(), TargetError> {
|
||||
if let Some(secs) = ack_timeout_secs
|
||||
&& !(NATS_JETSTREAM_ACK_TIMEOUT_MIN_SECS..=NATS_JETSTREAM_ACK_TIMEOUT_MAX_SECS).contains(&secs)
|
||||
{
|
||||
return Err(TargetError::Configuration(format!(
|
||||
"{NATS_JETSTREAM_ACK_TIMEOUT_SECS} must be between {NATS_JETSTREAM_ACK_TIMEOUT_MIN_SECS} and {NATS_JETSTREAM_ACK_TIMEOUT_MAX_SECS} seconds"
|
||||
)));
|
||||
}
|
||||
|
||||
if !enable {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if stream_name.trim().is_empty() {
|
||||
return Err(TargetError::Configuration(format!(
|
||||
"{NATS_JETSTREAM_STREAM_NAME} is required when {NATS_JETSTREAM_ENABLE} is on"
|
||||
)));
|
||||
}
|
||||
|
||||
if queue_dir.trim().is_empty() {
|
||||
return Err(TargetError::Configuration(format!(
|
||||
"{NATS_QUEUE_DIR} is required when {NATS_JETSTREAM_ENABLE} is on"
|
||||
)));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Detail carried by the retryable publish error raised when the stream-validation gate closes. The
|
||||
/// full validation cause is logged at error level inside the validator.
|
||||
pub(crate) const STREAM_VALIDATION_FAILED_DETAIL: &str = "stream validation failed";
|
||||
|
||||
/// Applies a stream-validation verdict on the publish path. A pass records the verdict so later
|
||||
/// publishes skip re-validation. A failure returns a retryable JetStream publish error carrying the
|
||||
/// fixed label and leaves the verdict unvalidated.
|
||||
pub(crate) fn gate_publish_on_stream_validation(
|
||||
verdict: Result<(), TargetError>,
|
||||
stream_validated: &AtomicBool,
|
||||
) -> Result<(), TargetError> {
|
||||
match verdict {
|
||||
Ok(()) => {
|
||||
stream_validated.store(true, Ordering::Release);
|
||||
Ok(())
|
||||
}
|
||||
Err(_) => Err(TargetError::JetStreamPublish {
|
||||
retryable: true,
|
||||
detail: STREAM_VALIDATION_FAILED_DETAIL.to_string(),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
/// Reports whether a publish subject is captured by a stream subject filter, honouring the NATS
|
||||
/// wildcard tokens: a single-token wildcard matches one subject token, a trailing multi-token
|
||||
/// wildcard matches one or more remaining tokens, any other token matches literally.
|
||||
fn subject_filter_captures(filter: &str, subject: &str) -> bool {
|
||||
let mut filter_tokens = filter.split('.');
|
||||
let mut subject_tokens = subject.split('.');
|
||||
|
||||
loop {
|
||||
match (filter_tokens.next(), subject_tokens.next()) {
|
||||
(Some(">"), Some(_)) => return true,
|
||||
(Some("*"), Some(_)) => continue,
|
||||
(Some(filter_token), Some(subject_token)) if filter_token == subject_token => continue,
|
||||
(None, None) => return true,
|
||||
_ => return false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Validates the configured JetStream stream from a single get_stream lookup when enabled. Asserts
|
||||
/// the stream exists, captures the subject, returns acks, accepts writes, and keeps a duplicate
|
||||
/// window that covers one retry cycle. The client-facing error is generic and every concrete cause
|
||||
/// is logged server-side. The stream is never created, a missing stream is an error.
|
||||
///
|
||||
/// When validation_logged is set, the first passing validation logs at info and later revalidations
|
||||
/// of the same context log at debug, so a persistent wrong-stream loop does not flood the log.
|
||||
pub(crate) async fn validate_jetstream_stream(
|
||||
context: &jetstream::Context,
|
||||
args: &NATSArgs,
|
||||
target_label: &str,
|
||||
validation_logged: Option<&AtomicBool>,
|
||||
) -> Result<(), TargetError> {
|
||||
if !args.jetstream_enable.unwrap_or(false) {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let stream_name = args.jetstream_stream_name.as_deref().unwrap_or_default();
|
||||
if stream_name.is_empty() {
|
||||
error!(target = %target_label, "JetStream enabled without a configured stream name");
|
||||
return Err(TargetError::Configuration("JetStream stream is not configured".to_string()));
|
||||
}
|
||||
|
||||
let stream = context.get_stream(stream_name).await.map_err(|err| {
|
||||
error!(
|
||||
target = %target_label,
|
||||
stream = %stream_name,
|
||||
error = %err,
|
||||
"JetStream stream lookup failed, the stream must be pre-provisioned"
|
||||
);
|
||||
TargetError::Configuration("configured JetStream stream is unavailable".to_string())
|
||||
})?;
|
||||
|
||||
assert_stream_writable(&stream.cached_info().config, args, target_label)?;
|
||||
// Info on a context's first passing validation, debug on later revalidations of the same context.
|
||||
let first_pass = validation_logged
|
||||
.map(|logged| !logged.swap(true, Ordering::AcqRel))
|
||||
.unwrap_or(true);
|
||||
if first_pass {
|
||||
info!(
|
||||
target = %target_label,
|
||||
stream = %stream_name,
|
||||
"JetStream stream validation succeeded"
|
||||
);
|
||||
} else {
|
||||
debug!(
|
||||
target = %target_label,
|
||||
stream = %stream_name,
|
||||
"JetStream stream revalidation succeeded"
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Asserts the retrieved stream configuration captures the subject, returns acks, accepts writes, and
|
||||
/// keeps a duplicate window that covers one retry cycle. The client-facing error is generic, the
|
||||
/// concrete cause logged server-side.
|
||||
fn assert_stream_writable(config: &jetstream::stream::Config, args: &NATSArgs, target_label: &str) -> Result<(), TargetError> {
|
||||
let stream_name = args.jetstream_stream_name.as_deref().unwrap_or_default();
|
||||
let subject = &args.subject;
|
||||
let subject_bound = config.subjects.iter().any(|filter| subject_filter_captures(filter, subject));
|
||||
if !subject_bound {
|
||||
error!(
|
||||
target = %target_label,
|
||||
stream = %stream_name,
|
||||
subject = %subject,
|
||||
"configured subject is not captured by the JetStream stream subjects"
|
||||
);
|
||||
return Err(TargetError::Configuration(
|
||||
"configured JetStream stream does not capture the subject".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
if config.no_ack {
|
||||
error!(
|
||||
target = %target_label,
|
||||
stream = %stream_name,
|
||||
"JetStream stream has no_ack set, publishes would never acknowledge"
|
||||
);
|
||||
return Err(TargetError::Configuration(
|
||||
"configured JetStream stream does not acknowledge writes".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
if config.sealed {
|
||||
error!(
|
||||
target = %target_label,
|
||||
stream = %stream_name,
|
||||
"JetStream stream is sealed, writes are rejected"
|
||||
);
|
||||
return Err(TargetError::Configuration("configured JetStream stream is sealed".to_string()));
|
||||
}
|
||||
|
||||
let ack_timeout = Duration::from_secs(
|
||||
args.jetstream_ack_timeout_secs
|
||||
.unwrap_or(NATS_JETSTREAM_ACK_TIMEOUT_DEFAULT_SECS),
|
||||
);
|
||||
let required_window = retry_lifetime(ack_timeout);
|
||||
if config.duplicate_window < required_window {
|
||||
error!(
|
||||
target = %target_label,
|
||||
stream = %stream_name,
|
||||
configured_window_secs = config.duplicate_window.as_secs(),
|
||||
required_window_secs = required_window.as_secs(),
|
||||
"JetStream stream duplicate_window is below the retry lifetime, a late retry would be delivered twice"
|
||||
);
|
||||
return Err(TargetError::Configuration(
|
||||
"configured JetStream stream duplicate window is too small for the retry lifetime".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::target::TargetType;
|
||||
use crate::target::nats::test_support::*;
|
||||
|
||||
#[test]
|
||||
fn jetstream_settings_disabled_passes() {
|
||||
validate_jetstream_settings(false, "", "", None).expect("disabled jetstream validates");
|
||||
validate_jetstream_settings(false, "", "", Some(30)).expect("disabled jetstream ignores stream and queue");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn jetstream_settings_ack_timeout_range() {
|
||||
for secs in [
|
||||
NATS_JETSTREAM_ACK_TIMEOUT_MIN_SECS,
|
||||
NATS_JETSTREAM_ACK_TIMEOUT_DEFAULT_SECS,
|
||||
NATS_JETSTREAM_ACK_TIMEOUT_MAX_SECS,
|
||||
] {
|
||||
validate_jetstream_settings(false, "", "", Some(secs)).expect("in-range ack timeout accepted");
|
||||
}
|
||||
for secs in [
|
||||
NATS_JETSTREAM_ACK_TIMEOUT_MIN_SECS - 1,
|
||||
NATS_JETSTREAM_ACK_TIMEOUT_MAX_SECS + 1,
|
||||
] {
|
||||
let err = validate_jetstream_settings(false, "", "", Some(secs)).expect_err("out-of-range ack timeout rejected");
|
||||
assert!(err.to_string().contains("between"));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn jetstream_settings_requires_stream_name() {
|
||||
let err = validate_jetstream_settings(true, " ", &nats_queue_dir(), Some(30))
|
||||
.expect_err("empty stream name rejected when enabled");
|
||||
assert!(err.to_string().contains(NATS_JETSTREAM_STREAM_NAME));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn jetstream_settings_requires_queue_dir() {
|
||||
let err =
|
||||
validate_jetstream_settings(true, "RUSTFS_EVENTS", "", Some(30)).expect_err("empty queue_dir rejected when enabled");
|
||||
assert!(err.to_string().contains(NATS_QUEUE_DIR));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn jetstream_settings_enabled_with_stream_and_queue_passes() {
|
||||
validate_jetstream_settings(true, "RUSTFS_EVENTS", &nats_queue_dir(), Some(30))
|
||||
.expect("a stream name and queue dir accept enabling jetstream");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn natsargs_validate_rejects_enabled_jetstream_without_stream() {
|
||||
let args = NATSArgs {
|
||||
queue_dir: nats_queue_dir(),
|
||||
jetstream_enable: Some(true),
|
||||
jetstream_stream_name: None,
|
||||
..base_args()
|
||||
};
|
||||
let err = args.validate().expect_err("enabled jetstream without a stream is rejected");
|
||||
assert!(err.to_string().contains(NATS_JETSTREAM_STREAM_NAME));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn subject_filter_matches_literal_and_wildcards() {
|
||||
assert!(subject_filter_captures("rustfs.events", "rustfs.events"));
|
||||
assert!(subject_filter_captures("rustfs.*", "rustfs.events"));
|
||||
assert!(subject_filter_captures("rustfs.>", "rustfs.events.created"));
|
||||
assert!(subject_filter_captures("rustfs.*.created", "rustfs.events.created"));
|
||||
assert!(!subject_filter_captures("rustfs.events", "rustfs.audit"));
|
||||
assert!(!subject_filter_captures("rustfs.*", "rustfs.events.created"));
|
||||
assert!(!subject_filter_captures("rustfs.events", "rustfs.events.created"));
|
||||
assert!(!subject_filter_captures("rustfs.events.created", "rustfs.events"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assert_stream_writable_enforces_the_worst_case_window_at_the_default_ack_timeout() {
|
||||
// The default 30s ack timeout requires a 274s window. One second below is rejected, exactly at it is accepted.
|
||||
let args = jetstream_args(TargetType::NotifyEvent);
|
||||
assert_eq!(retry_lifetime(Duration::from_secs(30)), Duration::from_secs(274));
|
||||
|
||||
let mut below = writable_stream_config(&args.subject);
|
||||
below.duplicate_window = Duration::from_secs(273);
|
||||
let err = assert_stream_writable(&below, &args, "test").expect_err("a window below the worst case is rejected");
|
||||
assert!(matches!(err, TargetError::Configuration(_)));
|
||||
|
||||
let mut at = writable_stream_config(&args.subject);
|
||||
at.duplicate_window = Duration::from_secs(274);
|
||||
assert_stream_writable(&at, &args, "test").expect("a window at the worst case is accepted");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assert_stream_writable_accepts_a_valid_stream() {
|
||||
for target_type in [TargetType::NotifyEvent, TargetType::AuditLog] {
|
||||
let args = jetstream_args(target_type);
|
||||
let config = writable_stream_config(&args.subject);
|
||||
assert_stream_writable(&config, &args, "test")
|
||||
.unwrap_or_else(|err| panic!("a writable stream passes for {target_type:?}: {err}"));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assert_stream_writable_passes_when_subject_is_bound_by_wildcard() {
|
||||
let args = jetstream_args(TargetType::NotifyEvent);
|
||||
let mut config = writable_stream_config(&args.subject);
|
||||
config.subjects = vec!["rustfs.>".to_string()];
|
||||
// The configured subject rustfs.events is captured by the rustfs.> filter.
|
||||
assert_stream_writable(&config, &args, "test").expect("a wildcard-bound subject passes");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assert_stream_writable_rejects_a_subject_not_bound() {
|
||||
let args = jetstream_args(TargetType::NotifyEvent);
|
||||
let mut config = writable_stream_config(&args.subject);
|
||||
config.subjects = vec!["some.other.subject".to_string()];
|
||||
let err = assert_stream_writable(&config, &args, "test").expect_err("an unbound subject is rejected");
|
||||
assert!(matches!(err, TargetError::Configuration(_)), "the error is a generic configuration error");
|
||||
assert!(
|
||||
!err.to_string().contains("some.other.subject"),
|
||||
"the client message omits the stream subjects"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assert_stream_writable_rejects_no_ack() {
|
||||
let args = jetstream_args(TargetType::NotifyEvent);
|
||||
let mut config = writable_stream_config(&args.subject);
|
||||
config.no_ack = true;
|
||||
let err = assert_stream_writable(&config, &args, "test").expect_err("a no_ack stream is rejected");
|
||||
assert!(matches!(err, TargetError::Configuration(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assert_stream_writable_rejects_sealed() {
|
||||
let args = jetstream_args(TargetType::NotifyEvent);
|
||||
let mut config = writable_stream_config(&args.subject);
|
||||
config.sealed = true;
|
||||
let err = assert_stream_writable(&config, &args, "test").expect_err("a sealed stream is rejected");
|
||||
assert!(matches!(err, TargetError::Configuration(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assert_stream_writable_rejects_a_too_small_duplicate_window() {
|
||||
let args = jetstream_args(TargetType::NotifyEvent);
|
||||
let mut config = writable_stream_config(&args.subject);
|
||||
// Below the worst-case retry lifetime at the default ack timeout, so a late retry would deliver twice.
|
||||
config.duplicate_window = Duration::from_secs(90);
|
||||
let err = assert_stream_writable(&config, &args, "test").expect_err("a too-small window is rejected");
|
||||
assert!(matches!(err, TargetError::Configuration(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assert_stream_writable_rejects_a_zero_duplicate_window() {
|
||||
let args = jetstream_args(TargetType::NotifyEvent);
|
||||
let mut config = writable_stream_config(&args.subject);
|
||||
// The window defaults to zero (dedup disabled) when the operator leaves it unset.
|
||||
config.duplicate_window = Duration::ZERO;
|
||||
let err = assert_stream_writable(&config, &args, "test").expect_err("a zero window is rejected");
|
||||
assert!(matches!(err, TargetError::Configuration(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_validation_gate_blocks_the_publish_and_classifies_retryable() {
|
||||
// A failed validation returns before any publish call, classified retryable so the entry is never lost.
|
||||
let stream_validated = AtomicBool::new(false);
|
||||
let verdict = Err(TargetError::Configuration("configured JetStream stream is unavailable".to_string()));
|
||||
let err = gate_publish_on_stream_validation(verdict, &stream_validated).expect_err("a failed validation closes the gate");
|
||||
match err {
|
||||
TargetError::JetStreamPublish { retryable, detail } => {
|
||||
assert!(retryable, "a validation failure is retryable");
|
||||
assert_eq!(detail, STREAM_VALIDATION_FAILED_DETAIL, "the fixed label is the detail");
|
||||
}
|
||||
other => panic!("expected a retryable JetStreamPublish, got {other:?}"),
|
||||
}
|
||||
assert!(
|
||||
!stream_validated.load(Ordering::SeqCst),
|
||||
"the verdict stays closed so the next attempt re-validates"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_validation_gate_pass_is_recorded_with_the_context() {
|
||||
let stream_validated = AtomicBool::new(false);
|
||||
gate_publish_on_stream_validation(Ok(()), &stream_validated).expect("a passing validation opens the gate");
|
||||
assert!(
|
||||
stream_validated.load(Ordering::SeqCst),
|
||||
"the recorded verdict lets later publishes skip re-validation"
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user