diff --git a/Cargo.lock b/Cargo.lock index dfa9bec..3181eec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -143,7 +143,7 @@ checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" [[package]] name = "beyond-slipstream" -version = "0.7.1" +version = "0.7.2" dependencies = [ "aho-corasick", "async-nats", diff --git a/Cargo.toml b/Cargo.toml index 4e675a7..ad3c4b3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ rust-version = "1.92" [package] name = "beyond-slipstream" -version = "0.7.1" +version = "0.7.2" edition.workspace = true license.workspace = true rust-version.workspace = true diff --git a/src/lib.rs b/src/lib.rs index da44759..171a67b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -54,6 +54,8 @@ pub use snapshot::{AppendLogSnapshot, SnapshotStore}; pub use snapshot_fjall::{FjallConfig, FjallReader, FjallSnapshot}; #[cfg(feature = "rocksdb")] pub use snapshot_rocksdb::{RocksDbConfig, RocksDbReader, RocksDbSnapshot}; -pub use stores::{Connection, ConnectionCapabilities, KvStore, StorageType, StoreConfig}; +pub use stores::{ + Connection, ConnectionCapabilities, DiscardPolicy, KvStore, StorageType, StoreConfig, +}; #[cfg(feature = "transport")] pub use transport::{ArtifactTransport, ObjectStoreTransport, PublishOutcome, run_export_round}; diff --git a/src/nats.rs b/src/nats.rs index e9376c8..16c466e 100644 --- a/src/nats.rs +++ b/src/nats.rs @@ -12,7 +12,7 @@ use tracing::{debug, error, info, warn}; use crate::kv::{ KvEntry, KvError, KvPurge, KvReader, KvUpdate, KvWatcher, KvWriter, VersionToken, WatchCursor, }; -use crate::stores::{Connection, ConnectionCapabilities, KvStore, StoreConfig}; +use crate::stores::{Connection, ConnectionCapabilities, DiscardPolicy, KvStore, StoreConfig}; /// Default per-operation timeout for NATS KV ops. async-nats's request/response /// futures don't fail in-flight requests when the underlying TCP connection @@ -183,6 +183,7 @@ pub(crate) async fn create_kv_bucket_raw( max_bytes: i64, history: i64, max_age_nanos: i64, + discard: DiscardPolicy, num_replicas: usize, ) -> Result<(), KvError> { let stream_name = format!("KV_{}", bucket); @@ -200,7 +201,7 @@ pub(crate) async fn create_kv_bucket_raw( "deny_delete": false, "deny_purge": false, "allow_direct": true, - "discard": "new", + "discard": discard.as_nats(), "num_replicas": num_replicas, "retention": "limits" }); @@ -446,31 +447,40 @@ impl NatsConnection { // slow or distant — `?`-propagating the timeout here would skip the // exact workaround built for it. If the connection is genuinely dead, // the raw path's own `timed()` bounds surface that promptly anyway. - match timed(js.create_key_value(kv_config)).await { - Ok(Ok(kv)) => return Ok(kv), - Ok(Err(e)) => { - warn!( - bucket = config.name, - error = ?e, - "create_key_value failed, trying raw JetStream API" - ); - } - Err(_) => { - warn!( - bucket = config.name, - timeout = ?KV_OP_TIMEOUT, - "create_key_value timed out, trying raw JetStream API" - ); + // + // EXCEPTION: `discard:old` cannot go through the normal path — + // async-nats' `kv::Config` exposes no discard field, so `create_key_value` + // would silently produce a `discard:new` bucket. Skip straight to the raw + // stream API (the only place the policy is expressible) for that case. + if config.discard == DiscardPolicy::New { + match timed(js.create_key_value(kv_config)).await { + Ok(Ok(kv)) => return Ok(kv), + Ok(Err(e)) => { + warn!( + bucket = config.name, + error = ?e, + "create_key_value failed, trying raw JetStream API" + ); + } + Err(_) => { + warn!( + bucket = config.name, + timeout = ?KV_OP_TIMEOUT, + "create_key_value timed out, trying raw JetStream API" + ); + } } } - // Try raw JetStream API as fallback + // Raw JetStream API: the fallback for `discard:new`, and the ONLY path for + // `discard:old`. create_kv_bucket_raw( client, &config.name, max_bytes, history, max_age_nanos, + config.discard, config.num_replicas.unwrap_or(1), ) .await?; diff --git a/src/stores.rs b/src/stores.rs index dcc58ec..72e99db 100644 --- a/src/stores.rs +++ b/src/stores.rs @@ -14,6 +14,37 @@ pub enum StorageType { Persistent, } +/// What a bounded bucket does when it reaches `max_bytes` (NATS-specific). +/// +/// This is a real semantic choice, not a tuning knob — it decides what the bucket +/// gives up at capacity: +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum DiscardPolicy { + /// **Reject** new writes when full, preserving every existing entry (NATS + /// `discard:new`). Correct for **config** buckets (certs, configs read as the + /// source of truth) where silently dropping a live value is unacceptable — but + /// it FREEZES writes at capacity (err 10077). Pair it with `max_age` so the + /// bucket is trimmed before it ever fills. + #[default] + New, + /// **Evict the oldest** messages when full (NATS `discard:old`): a hard size + /// ceiling that never rejects. Correct for high-churn **log** buckets whose + /// consumers hold the durable fold (e.g. routing origins): the bucket is a + /// bounded change-feed, not the source of truth, so an evicted entry is + /// recovered from the consumer's fold (and the `CursorExpired` resync path), + /// while writers never freeze. + Old, +} + +impl DiscardPolicy { + pub(crate) fn as_nats(self) -> &'static str { + match self { + DiscardPolicy::New => "new", + DiscardPolicy::Old => "old", + } + } +} + /// Configuration for creating a store. #[derive(Debug, Clone, Default)] pub struct StoreConfig { @@ -33,6 +64,12 @@ pub struct StoreConfig { /// Number of stream replicas for the bucket (NATS cluster mode). /// Defaults to 1 (single replica). Set to 3 for production HA clusters. pub num_replicas: Option, + /// Behavior at `max_bytes` (NATS-specific, ignored by other stores). Defaults + /// to [`DiscardPolicy::New`] (reject) so config buckets never silently drop a + /// live value; set [`DiscardPolicy::Old`] for size-bounded log buckets (routing + /// origins) that must never freeze writers. Only applied at bucket *creation* — + /// an existing bucket's policy is left untouched. + pub discard: DiscardPolicy, } /// A named KV store (bucket/namespace/database). diff --git a/tests/integration.rs b/tests/integration.rs index d9d31bd..fd35900 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -13,8 +13,8 @@ use std::sync::Arc; use std::time::Duration; use slipstream::{ - Connection, KvError, KvStore, KvUpdate, KvWriter, NatsConnection, NatsConnectionConfig, - StoreConfig, VersionToken, WatchCursor, + Connection, DiscardPolicy, KvError, KvStore, KvUpdate, KvWriter, NatsConnection, + NatsConnectionConfig, StoreConfig, VersionToken, WatchCursor, }; use tokio::sync::mpsc; use tokio::time::timeout; @@ -1682,3 +1682,66 @@ async fn export_lease_complete_publishes_outcome() { assert!(record.completed_at_unix.is_some()); assert_eq!(record.holder_id, "node-a"); } + +/// The `discard` knob: `DiscardPolicy::Old` creates an evict-oldest bucket (a +/// size ceiling that never rejects — right for high-churn routing-origin logs), +/// while the default stays `discard:new` (reject-when-full, right for config +/// buckets). Asserted by reading the created stream's config back from NATS. +#[tokio::test] +async fn discard_policy_old_evicts_default_rejects() { + use async_nats::jetstream::stream::DiscardPolicy as NatsDiscard; + + let nats = TestNats::start().await; + let conn = nats.connect().await; + + // Opt-in: a size-bounded, evict-oldest log bucket. + conn.store_with_config(StoreConfig { + name: "origins-old".to_string(), + max_bytes: Some(1024 * 1024), + max_history: Some(1), + discard: DiscardPolicy::Old, + ..Default::default() + }) + .await + .expect("open discard:old store"); + + // Default: unchanged reject-when-full behavior (back-compat). + conn.store_with_config(StoreConfig { + name: "config-new".to_string(), + max_bytes: Some(1024 * 1024), + ..Default::default() + }) + .await + .expect("open default store"); + + let js = async_nats::jetstream::new(async_nats::connect(&nats.url).await.expect("raw connect")); + let old = js + .get_stream("KV_origins-old") + .await + .expect("get old stream") + .info() + .await + .expect("old info") + .config + .discard; + let new = js + .get_stream("KV_config-new") + .await + .expect("get new stream") + .info() + .await + .expect("new info") + .config + .discard; + + assert_eq!( + old, + NatsDiscard::Old, + "DiscardPolicy::Old must create an evict-oldest bucket" + ); + assert_eq!( + new, + NatsDiscard::New, + "the default must remain discard:new (back-compat)" + ); +}