feat(kv): add DiscardPolicy knob (default New, opt-in Old) — v0.7.2

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

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};

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?;

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<usize>,
+    /// 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).

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)"
+    );
+}