diff --git a/Cargo.lock b/Cargo.lock index 1ef38f5..dfa9bec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -143,7 +143,7 @@ checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" [[package]] name = "beyond-slipstream" -version = "0.7.0" +version = "0.7.1" dependencies = [ "aho-corasick", "async-nats", diff --git a/Cargo.toml b/Cargo.toml index 2356021..4e675a7 100644 --- a/Cargo.toml +++ b/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 diff --git a/src/kv.rs b/src/kv.rs index 00b93ef..a0a39df 100644 --- a/src/kv.rs +++ b/src/kv.rs @@ -380,6 +380,24 @@ pub trait KvTtl: KvWriter { ) -> Result; } +/// 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::*; diff --git a/src/lib.rs b/src/lib.rs index cb95982..da44759 100644 --- a/src/lib.rs +++ b/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}; diff --git a/src/nats.rs b/src/nats.rs index 192446a..e9376c8 100644 --- a/src/nats.rs +++ b/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> { + 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 diff --git a/src/stores.rs b/src/stores.rs index 9de036e..dcc58ec 100644 --- a/src/stores.rs +++ b/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> { 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> { + 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.