[Jetty] Held unarchived Redis records against the trim.

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|<now + lease>`, 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 `<prefix>:{<tag of slot s>}: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

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

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

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

crates/streams/src/redis/keys.rs

@@ -32,6 +32,7 @@
 //! <prefix>:{<tag of slot s>}:n      slot s's notification entries
 //! <prefix>:{<tag of slot d * 16384 / doorbells>}:bell
 //!                                    doorbell d, which slots s with s % doorbells == d ring
+//! <prefix>:{<tag of slot s>}: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 {

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<Duration>,
 }
 
 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<Arc<log::RedisLog>>,
+    archive_lease_ms: Option<u64>,
 }
 
 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<u16, Vec<&String>> = 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<ChainKeys> {
         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::<i64>(&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::<i64>(&mut self.shared.clone())
+            .await
+            .mapped(Call::Write)
+            .map(drop)
+    }
+
+    async fn archived(&self, chain: &ChainId, topic: &str) -> StreamResult<Option<String>> {
+        if self.archive_lease_ms.is_none() {
+            return Ok(None);
+        }
+        let (exists, archived): (bool, Option<String>) = 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<StoredClose> {
-        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::<Vec<u8>>(&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;

crates/streams/src/redis/scripts.rs

@@ -43,6 +43,60 @@ pub(crate) static APPEND_LOGGED: LazyLock<String> = LazyLock::new(|| {
 
 pub(crate) static APPEND_LOGGED_SCRIPT: LazyLock<Script> =
     LazyLock::new(|| Script::new(&APPEND_LOGGED));
+
+/// What a store with an archive adds before the batch: the archive's hold on a topic's first
+/// write, from the topic's start, with the lease in `ARGV[6]`. So retention can't drop a record
+/// before the archive's first pass. The holds hash lives at least as long as the lease, since a
+/// reader's hold only lengthens its life and a shorter one must not take the archive's hold away.
+const ARCHIVE_HOLD: &str = "if redis.call('EXISTS', KEYS[2]) == 0 and \
+redis.call('HEXISTS', KEYS[4], 'archive') == 0 then
+  local lease = tonumber(ARGV[6])
+  redis.call('HSET', KEYS[4], 'archive', '0-1|' .. (now_ms() + lease))
+  if redis.call('PTTL', KEYS[4]) < lease then
+    redis.call('PEXPIRE', KEYS[4], lease)
+  end
+end
+revive(KEYS[1], KEYS[2])
+";
+
+/// What a store with an archive adds after the notification entry: the log key in the slot's
+/// set of topics with records to archive (`KEYS[6]`), scored by the time of its newest write.
+const ARCHIVE_MARK: &str = "redis.call('ZADD', KEYS[6], now_ms(), KEYS[1])
+return {first, last, redis.call('XLEN', KEYS[5])}
+";
+
+/// [APPEND_LOGGED] with the archive's hold and mark. The lease takes `ARGV[6]`, so the records
+/// start at `ARGV[7]`, and the topic set moves to `KEYS[7]`.
+pub(crate) static APPEND_ARCHIVED: LazyLock<String> = LazyLock::new(|| {
+    let edits = [
+        ("revive(KEYS[1], KEYS[2])\n", ARCHIVE_HOLD),
+        ("for i = 6, #ARGV do", "for i = 7, #ARGV do"),
+        ("local topics = KEYS[6]\n", "local topics = KEYS[7]\n"),
+        (
+            "return {first, last, redis.call('XLEN', KEYS[5])}\n",
+            ARCHIVE_MARK,
+        ),
+    ];
+    let mut script = APPEND_LOGGED.clone();
+    for (was, now) in edits {
+        assert_eq!(script.matches(was).count(), 1, "{was}");
+        script = script.replacen(was, now, 1);
+    }
+    assert_eq!(script.matches("#ARGV - 5").count(), 2);
+    script.replace("#ARGV - 5", "#ARGV - 6")
+});
+
+pub(crate) static APPEND_ARCHIVED_SCRIPT: LazyLock<Script> =
+    LazyLock::new(|| Script::new(&APPEND_ARCHIVED));
+
+/// Moves the archive's hold forward and keeps the position as the meta's `archived`. When the
+/// archive holds the topic's last record, the topic leaves the slot's set to archive.
+pub(crate) static ADVANCE_SCRIPT: LazyLock<Script> = LazyLock::new(|| {
+    Script::new(concat!(
+        include_str!("../../lua/floor.lua"),
+        include_str!("../../lua/advance.lua")
+    ))
+});
 // Marks the topic closed and keeps its first close's outcome in one step, so two closes can't both
 // find no outcome and the second overwrite the first.
 pub(crate) static CLOSE_SCRIPT: LazyLock<Script> =
@@ -124,6 +178,25 @@ end
         );
     }
 
+    #[test]
+    fn the_archived_append_differs_from_the_logged_one_in_its_hold_and_mark_only() {
+        let mut was = APPEND_ARCHIVED
+            .replacen(ARCHIVE_HOLD, "revive(KEYS[1], KEYS[2])\n", 1)
+            .replacen(
+                ARCHIVE_MARK,
+                "return {first, last, redis.call('XLEN', KEYS[5])}\n",
+                1,
+            )
+            .replacen("local topics = KEYS[7]\n", "local topics = KEYS[6]\n", 1)
+            .replacen("for i = 7, #ARGV do", "for i = 6, #ARGV do", 1);
+        was = was.replace("#ARGV - 6", "#ARGV - 5");
+        assert_eq!(was, *APPEND_LOGGED);
+        assert_eq!(
+            APPEND_ARCHIVED_SCRIPT.get_hash(),
+            "bb8e042129e393c24f89efa5b81339d87090e8c1"
+        );
+    }
+
     #[test]
     fn the_append_script_is_the_one_before_the_floor_with_the_floor_edits_only() {
         assert_eq!(