diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index 5bda9c839..fd7ca09ac 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -58,6 +58,22 @@ streams and the doorbells as the consumer group `streams-driver` (`XREADGROUP`, SLOTS`. The slot streams have no expiry, and a driver empties them. The logged append script has its own hash. The plain one is unchanged, and an SDK's provider writes no entries. +**Archive hold.** With `archive_lease` set, a driver copies each topic into an archive, and the +store holds what the archive doesn't hold yet. The floors hash keeps the holder `archive`. The +append script makes it at a topic's first write as `0-1|`, so it holds the whole +topic. `advance.lua` moves it to the archive's newest position with no minimum, renews the lease, +extends the keys' expiry to it, and keeps the position in the meta's `archived` field. The trim +reads the hold as it reads a reader's, so retention drops only what the archive holds. When the +archive stops for longer than the lease, the hold ends and the trim goes on, and nothing blocks +appends. The append script also adds the log key to `:{}:a`, a sorted set +in the slot of the chain's notification stream, scored by the time of the write. A close adds it +with score `0`, so the archive takes closed topics first. `advance.lua` takes it out once the +archive holds the meta's `last`, and the archive's release takes it out after a closed topic's +end. The archive needs the notification log. The archived append script has its own hash. With an +archive, an application also needs `~temporal-streams:{*}:a +zadd +hexists`, and a driver also +needs `+zrange +zrem` on those keys. A delete takes a deleted topic's mark out of the set, so it +also needs `+zrem` there. + **Retired producers.** A field stays until a write's trim finds it older than the retention, so a topic written by many short-lived producers holds one field per producer attempt of the last retention window, about 235 bytes each. `RetireProducer` deletes one attempt's field with diff --git a/crates/streams/lua/advance.lua b/crates/streams/lua/advance.lua new file mode 100644 index 000000000..08dd9cc1c --- /dev/null +++ b/crates/streams/lua/advance.lua @@ -0,0 +1,34 @@ +local function extend(key, ms, create) + local ttl = redis.call('PTTL', key) + if (ttl >= 0 or (create and ttl == -1)) and ttl < ms then + redis.call('PEXPIRE', key, ms) + end +end +local lease, grace = tonumber(ARGV[3]), tonumber(ARGV[4]) +if redis.call('EXISTS', KEYS[3]) == 0 then + -- The topic is gone, so its mark would only hold back the slot's other topics. + if KEYS[4] then + redis.call('ZREM', KEYS[4], KEYS[2]) + end + return 0 +end +-- Two drivers can archive from the same start. The slower one's advance must not move the hold +-- back, nor hold a closed topic again after the archive released it. +local archived = redis.call('HGET', KEYS[3], 'archived') +if archived and archived ~= '' then + if id_below(ARGV[2], archived) then + return 0 + end + if ARGV[2] == archived and redis.call('HGET', KEYS[3], 'closed') then + return 0 + end +end +redis.call('HSET', KEYS[1], ARGV[1], ARGV[2] .. '|' .. (clock_ms() + lease)) +redis.call('HSET', KEYS[3], 'archived', ARGV[2]) +if KEYS[4] and redis.call('HGET', KEYS[3], 'last') == ARGV[2] then + redis.call('ZREM', KEYS[4], KEYS[2]) +end +extend(KEYS[1], lease, true) +extend(KEYS[2], lease, false) +extend(KEYS[3], lease + grace, false) +return 1 diff --git a/crates/streams/lua/close.lua b/crates/streams/lua/close.lua index dedde1cee..8723ab66d 100644 --- a/crates/streams/lua/close.lua +++ b/crates/streams/lua/close.lua @@ -6,4 +6,7 @@ if not close then redis.call('HSET', KEYS[1], 'close', close) end redis.call('PEXPIRE', KEYS[1], ARGV[2]) +if KEYS[2] then + redis.call('ZADD', KEYS[2], 0, ARGV[3]) +end return close diff --git a/crates/streams/lua/release.lua b/crates/streams/lua/release.lua index 3841c125a..c11b67e3a 100644 --- a/crates/streams/lua/release.lua +++ b/crates/streams/lua/release.lua @@ -1,4 +1,7 @@ local held_value = redis.call('HGET', KEYS[1], ARGV[1]) +if KEYS[2] then + redis.call('ZREM', KEYS[2], ARGV[3]) +end if not held_value then return 0 end diff --git a/crates/streams/src/redis/keys.rs b/crates/streams/src/redis/keys.rs index c93d353a1..7345e573e 100644 --- a/crates/streams/src/redis/keys.rs +++ b/crates/streams/src/redis/keys.rs @@ -32,6 +32,7 @@ //! :{}:n slot s's notification entries //! :{}:bell //! doorbell d, which slots s with s % doorbells == d ring +//! :{}:a slot s's topics with records to archive, with an archive //! ``` //! //! Each part, the prefix included, is percent-encoded, so a `:` or a brace in an id cannot make @@ -111,6 +112,11 @@ pub(crate) fn notifications(prefix: &str, slot: u16) -> String { format!("{}:{{{}}}:n", part(prefix), slots::tag(slot)) } +/// Slot `slot`'s set of log keys with records to archive, scored by their newest write. +pub(crate) fn archive_pending(prefix: &str, slot: u16) -> String { + format!("{}:{{{}}}:a", part(prefix), slots::tag(slot)) +} + /// Doorbell `doorbell`'s stream, of `doorbells`. The doorbells spread over the slot range, so /// a cluster's primaries share them. pub(crate) fn doorbell(prefix: &str, doorbell: u32, doorbells: u32) -> String { diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index 6894421f9..3aa1ad8d7 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -35,6 +35,8 @@ const TOMBSTONE_GRACE: Duration = Duration::from_secs(30 * 24 * 60 * 60); const CLOSED_FIELD: &str = "closed"; /// The meta field that holds the topic's first close, a serialized `StoredClose`. const CLOSE_FIELD: &str = "close"; +/// The meta field that holds the newest position the archive holds. +const ARCHIVED_FIELD: &str = "archived"; const DELETE_BATCH: usize = 500; /// How to reach Redis and how long streams live there. @@ -63,6 +65,11 @@ pub struct RedisStoreOptions { pub notification_log: bool, /// How many doorbells the notification log rings, each a stream that a driver blocks on. pub notify_doorbells: u32, + /// With an archive, how long the archive's hold on a topic lasts after the archive last + /// moved it. Each topic's first append makes the hold, so retention drops only what the + /// archive holds while the archive keeps up. When the archive stops for longer, the hold + /// ends and retention goes on. Needs `notification_log`, since a driver archives. + pub archive_lease: Option, } impl RedisStoreOptions { @@ -79,6 +86,7 @@ impl RedisStoreOptions { floor_release_grace: Duration::from_secs(24 * 60 * 60), notification_log: false, notify_doorbells: DEFAULT_NOTIFY_DOORBELLS, + archive_lease: None, } } } @@ -96,6 +104,7 @@ pub struct RedisStore { floor_lease_ms: u64, floor_release_grace_ms: u64, log: Option>, + archive_lease_ms: Option, } impl std::fmt::Debug for RedisStore { @@ -126,6 +135,11 @@ impl RedisStore { options.retention ))); } + if options.archive_lease.is_some() && !options.notification_log { + return Err(StreamError::refused( + "an archive needs the notification log, since a driver archives", + )); + } // The scripts trim below `now - retention` as a stream id, which can't be negative. let now_ms = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -176,6 +190,7 @@ impl RedisStore { .floor_lease .map_or(retention_ms, |lease| millis(lease).max(1)), floor_release_grace_ms: millis(options.floor_release_grace), + archive_lease_ms: options.archive_lease.map(|lease| millis(lease).max(1)), }) } @@ -200,6 +215,27 @@ impl RedisStore { error.with_closed_by(closed_by) } + /// Takes deleted keys out of their slot's archive marks. A mark of a deleted topic stays + /// lowest in its slot, and enough of them stop the slot's other topics from being archived. + async fn drop_archive_marks(&self, keys: &[String]) -> StreamResult<()> { + if self.archive_lease_ms.is_none() { + return Ok(()); + } + let mut by_slot: HashMap> = HashMap::new(); + for key in keys { + by_slot.entry(conn::slot(key)).or_default().push(key); + } + for (slot, keys) in by_slot { + let _: u64 = redis::cmd("ZREM") + .arg(keys::archive_pending(&self.prefix, slot)) + .arg(keys) + .query_async(&mut self.shared.clone()) + .await + .mapped(Call::Write)?; + } + Ok(()) + } + fn keys(&self, chain: Option<&ChainId>) -> StreamResult { let chain = chain .ok_or_else(|| StreamError::refused("a store call needs the chain it is about"))?; @@ -528,9 +564,10 @@ impl StreamStore for RedisStore { return Err(StreamError::refused("an append needs the batch's digest")); } let keys = self.keys(request.chain.as_ref())?; - let mut script = match self.log { - Some(_) => scripts::APPEND_LOGGED_SCRIPT.prepare_invoke(), - None => scripts::APPEND_SCRIPT.prepare_invoke(), + let mut script = match (&self.log, self.archive_lease_ms) { + (Some(_), Some(_)) => scripts::APPEND_ARCHIVED_SCRIPT.prepare_invoke(), + (Some(_), None) => scripts::APPEND_LOGGED_SCRIPT.prepare_invoke(), + (None, _) => scripts::APPEND_SCRIPT.prepare_invoke(), }; script .key(keys.log(&request.topic)) @@ -540,6 +577,9 @@ impl StreamStore for RedisStore { if self.log.is_some() { script.key(keys::notifications(&self.prefix, keys.slot())); } + if self.archive_lease_ms.is_some() { + script.key(keys::archive_pending(&self.prefix, keys.slot())); + } if request.chain.as_ref().is_some_and(ChainKeys::keeps_topics) { script.key(keys.topics()); } @@ -549,6 +589,9 @@ impl StreamStore for RedisStore { .arg(session_field(&request.producer_id, request.attempt)) .arg(request.sequence) .arg(hex(&request.digest)); + if let Some(lease) = self.archive_lease_ms { + script.arg(lease); + } for record in &request.records { script.arg(record.as_slice()); } @@ -746,16 +789,65 @@ impl StreamStore for RedisStore { } async fn release_floor(&self, chain: &ChainId, topic: &str, holder: &str) -> StreamResult<()> { - scripts::RELEASE_SCRIPT - .key(ChainKeys::new(&self.prefix, chain).floors(topic)) + let keys = ChainKeys::new(&self.prefix, chain); + let mut script = scripts::RELEASE_SCRIPT.key(keys.floors(topic)); + // The archive releases once it holds a closed topic to its end, so the topic leaves the + // set to archive in the same step. + if holder == crate::ARCHIVE_HOLDER && self.archive_lease_ms.is_some() { + script.key(keys::archive_pending(&self.prefix, keys.slot())); + } + script .arg(holder) .arg(self.floor_release_grace_ms) + .arg(keys.log(topic)) .invoke_async::(&mut self.shared.clone()) .await .mapped(Call::Write) .map(drop) } + async fn advance_archive( + &self, + chain: &ChainId, + topic: &str, + position: &str, + ) -> StreamResult<()> { + let Some(lease) = self.archive_lease_ms else { + return Err(StreamError::unsupported("this store keeps no archive hold")); + }; + entry(position)?; + let keys = ChainKeys::new(&self.prefix, chain); + scripts::ADVANCE_SCRIPT + .key(keys.floors(topic)) + .key(keys.log(topic)) + .key(keys.meta(topic)) + .key(keys::archive_pending(&self.prefix, keys.slot())) + .arg(crate::ARCHIVE_HOLDER) + .arg(position) + .arg(lease) + .arg(self.grace_ms) + .invoke_async::(&mut self.shared.clone()) + .await + .mapped(Call::Write) + .map(drop) + } + + async fn archived(&self, chain: &ChainId, topic: &str) -> StreamResult> { + if self.archive_lease_ms.is_none() { + return Ok(None); + } + let (exists, archived): (bool, Option) = redis::pipe() + .exists(ChainKeys::new(&self.prefix, chain).meta(topic)) + .hget( + ChainKeys::new(&self.prefix, chain).meta(topic), + ARCHIVED_FIELD, + ) + .query_async(&mut self.shared.clone()) + .await + .mapped(Call::Read)?; + Ok(exists.then(|| archived.unwrap_or_default())) + } + async fn close_chain(&self, chain: &ChainId) -> StreamResult<()> { let keys = ChainKeys::new(&self.prefix, chain); redis::pipe() @@ -779,13 +871,18 @@ impl StreamStore for RedisStore { topic: &str, close: StoredClose, ) -> StreamResult { - let meta = ChainKeys::new(&self.prefix, chain).meta(topic); + let keys = ChainKeys::new(&self.prefix, chain); // The meta is the topic's tombstone, so the mark and the outcome live as long as the topic // is known. - let stored = scripts::CLOSE_SCRIPT - .key(meta) + let mut script = scripts::CLOSE_SCRIPT.key(keys.meta(topic)); + // A closed topic gets nothing new, so the archive takes it first. + if self.archive_lease_ms.is_some() { + script.key(keys::archive_pending(&self.prefix, keys.slot())); + } + let stored = script .arg(close.encode_to_vec()) .arg(self.retention_ms + self.grace_ms) + .arg(keys.log(topic)) .invoke_async::>(&mut self.shared.clone()) .await .mapped(Call::Write)?; @@ -876,6 +973,7 @@ impl StreamStore for RedisStore { .await .mapped(Call::Write)?; deleted += removed; + self.drop_archive_marks(batch).await?; } if next == "0" { break; diff --git a/crates/streams/src/redis/scripts.rs b/crates/streams/src/redis/scripts.rs index a0387d2a9..c2b9353e9 100644 --- a/crates/streams/src/redis/scripts.rs +++ b/crates/streams/src/redis/scripts.rs @@ -43,6 +43,60 @@ pub(crate) static APPEND_LOGGED: LazyLock = LazyLock::new(|| { pub(crate) static APPEND_LOGGED_SCRIPT: LazyLock