feat(kv): add KvPurge for byte-reclaiming key purge (NATS rollup)

Cargo.toml

@@ -8,7 +8,7 @@ rust-version = "1.92"
 
 [package]
 name = "beyond-slipstream"
-version = "0.7.0"
+version = "0.7.1"
 edition.workspace = true
 license.workspace = true
 rust-version.workspace = true

src/kv.rs

@@ -380,6 +380,24 @@ pub trait KvTtl: KvWriter {
     ) -> Result<VersionToken, KvError>;
 }
 
+/// Purge support - optional, for stores that can reclaim a key's storage.
+///
+/// Unlike [`KvWriter::delete`] (which writes a delete marker) and
+/// [`KvWriter::delete_with_version`] (which writes an empty-value tombstone),
+/// `purge` removes a key *and reclaims its bytes*. On NATS this issues a
+/// rollup (`Nats-Rollup: sub`) that drops all prior revisions of the subject,
+/// so the bytes stop counting against the stream's `max_bytes`.
+///
+/// Use this to bound a bucket that has no `max_age`: dead keys deleted with
+/// `delete`/`delete_with_version` accumulate forever, but purged keys are
+/// reclaimed.
+#[async_trait]
+pub trait KvPurge: KvWriter {
+    /// Purge a key, reclaiming its storage. Idempotent: purging an absent key
+    /// is not an error.
+    async fn purge(&self, key: &str) -> Result<(), KvError>;
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/lib.rs

@@ -45,7 +45,8 @@ pub use applied::{BatchConfig, ExportRequest, WatchScope, watch_applied};
 pub use artifact::{ARTIFACT_SCHEMA_VERSION, ArtifactFile, ExportManifest, MANIFEST_FILE};
 pub use export_lease::{ExportLease, LeaseGuard, LeaseRecord};
 pub use kv::{
-    KvEntry, KvError, KvReader, KvTtl, KvUpdate, KvWatcher, KvWriter, VersionToken, WatchCursor,
+    KvEntry, KvError, KvPurge, KvReader, KvTtl, KvUpdate, KvWatcher, KvWriter, VersionToken,
+    WatchCursor,
 };
 pub use nats::{NatsConnection, NatsConnectionConfig, nats_connect};
 pub use snapshot::{AppendLogSnapshot, SnapshotStore};

src/nats.rs

@@ -10,7 +10,7 @@ use tokio::sync::{RwLock, mpsc::Sender};
 use tracing::{debug, error, info, warn};
 
 use crate::kv::{
-    KvEntry, KvError, KvReader, KvUpdate, KvWatcher, KvWriter, VersionToken, WatchCursor,
+    KvEntry, KvError, KvPurge, KvReader, KvUpdate, KvWatcher, KvWriter, VersionToken, WatchCursor,
 };
 use crate::stores::{Connection, ConnectionCapabilities, KvStore, StoreConfig};
 
@@ -642,6 +642,9 @@ impl Connection for NatsConnection {
             // callers that branch on this flag down a path that can never
             // succeed. Flip to `true` together with the `KvTtl` impl.
             ttl: false,
+            // Byte-reclaiming purge IS implemented for NATS (rollup delete) and
+            // vended via `KvStore::purge_writer`.
+            purge: true,
             cas: true,
             transactions: false,
             // 0 = unlimited from this layer's perspective: we impose no cap, but
@@ -689,6 +692,12 @@ impl KvStore for NatsKvStore {
             kv: self.kv.clone(),
         }))
     }
+
+    fn purge_writer(&self) -> Option<Arc<dyn KvPurge>> {
+        Some(Arc::new(NatsKvWriterImpl {
+            kv: self.kv.clone(),
+        }))
+    }
 }
 
 struct NatsKvReader {
@@ -1592,6 +1601,19 @@ impl KvWriter for NatsKvWriterImpl {
     }
 }
 
+#[async_trait]
+impl KvPurge for NatsKvWriterImpl {
+    async fn purge(&self, key: &str) -> Result<(), KvError> {
+        // Rollup purge (`Nats-Rollup: sub`): drops all prior revisions of the
+        // subject, reclaiming bytes against `max_bytes` — unlike `delete`, which
+        // only appends a marker. Idempotent: purging an absent key is a no-op.
+        timed(self.kv.purge(key))
+            .await?
+            .map_err(|e| KvError::OperationFailed(e.to_string()))?;
+        Ok(())
+    }
+}
+
 impl std::fmt::Debug for NatsConnection {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("NatsConnection")
@@ -1925,6 +1947,76 @@ mod floor_guard_tests {
         (js, kv)
     }
 
+    /// `KvPurge::purge` must reclaim bytes against `max_bytes` — unlike
+    /// `delete`/`delete_with_version`, which only append markers. This is the
+    /// in-repo twin of the "does purge actually free bytes?" gate: fill a
+    /// bucket, purge half the keys, assert the stream's byte count drops.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn purge_reclaims_bytes() {
+        use crate::kv::KvPurge;
+
+        let server = start_server().await;
+        let client = async_nats::connect(&server.url).await.unwrap();
+        let js = async_nats::jetstream::new(client);
+        let kv = js
+            .create_key_value(async_nats::jetstream::kv::Config {
+                bucket: "purge".into(),
+                history: 1,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        // Fill with sizable values across many distinct keys.
+        let val = vec![b'x'; 4096];
+        for i in 0..50u32 {
+            kv.put(format!("k{i}"), val.clone().into()).await.unwrap();
+        }
+        let before = js
+            .get_stream("KV_purge")
+            .await
+            .unwrap()
+            .info()
+            .await
+            .unwrap()
+            .state
+            .bytes;
+
+        // Purge half the keys through the KvPurge impl.
+        let writer = NatsKvWriterImpl { kv: kv.clone() };
+        for i in 0..25u32 {
+            writer.purge(&format!("k{i}")).await.unwrap();
+        }
+
+        // Purge is a rollup: prior revisions of each subject are removed, so the
+        // stream's byte count must fall. (A residual purge marker may remain per
+        // subject — far smaller than the 4KiB value — so we assert a strict drop,
+        // not zero.) Poll briefly in case the server reflects reclamation async.
+        let mut after = before;
+        for _ in 0..20 {
+            after = js
+                .get_stream("KV_purge")
+                .await
+                .unwrap()
+                .info()
+                .await
+                .unwrap()
+                .state
+                .bytes;
+            if after < before {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+        assert!(
+            after < before,
+            "purge must reclaim bytes: before={before} after={after}"
+        );
+
+        // Purge is idempotent: re-purging an absent key is not an error.
+        writer.purge("k0").await.unwrap();
+    }
+
     /// TRUE POSITIVE: the watch was clamped past evicted revisions (purge
     /// advanced first_seq beyond the frontier) — the first gapped delivery
     /// must trip the guard BEFORE the entry is processed, never silently

src/stores.rs

@@ -2,7 +2,7 @@ use async_trait::async_trait;
 use std::sync::Arc;
 use std::time::Duration;
 
-use crate::kv::{KvError, KvReader, KvWatcher, KvWriter};
+use crate::kv::{KvError, KvPurge, KvReader, KvWatcher, KvWriter};
 
 /// Storage type for a store.
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
@@ -52,6 +52,15 @@ pub trait KvStore: Send + Sync {
     fn writer(&self) -> Option<Arc<dyn KvWriter>> {
         None
     }
+
+    /// Get the purge interface (if supported).
+    ///
+    /// Purge reclaims a key's bytes, unlike `writer().delete()` which only
+    /// writes a marker. See [`KvPurge`]. Returns `None` for backends without
+    /// byte-reclaiming purge.
+    fn purge_writer(&self) -> Option<Arc<dyn KvPurge>> {
+        None
+    }
 }
 
 /// Capabilities a store connection may support.
@@ -63,6 +72,8 @@ pub struct ConnectionCapabilities {
     pub prefix_watch: bool,
     /// Supports TTL on keys.
     pub ttl: bool,
+    /// Supports byte-reclaiming purge (rollup delete). NATS: true.
+    pub purge: bool,
     /// Supports atomic compare-and-swap.
     pub cas: bool,
     /// Supports multi-key transactions.