[Brook] Added the Redis stream store behind the redis feature.

.gitattributes

@@ -1 +1,3 @@
 CHANGELOG.md merge=union
+# The Redis scripts are hashed byte for byte, so a CRLF checkout would change their SHA-1.
+crates/streams/lua/*.lua text eol=lf

crates/streams/Cargo.toml

@@ -14,11 +14,20 @@ categories = ["development-tools"]
 [features]
 default = []
 # The Redis store. Off by default, so a build that needs only the trait takes no Redis client.
-redis = []
+# TLS uses the rustls provider the process installs, the same one Core uses.
+redis = ["dep:redis", "dep:crc16"]
 
 [dependencies]
 async-trait = "0.1"
+crc16 = { version = "0.4", optional = true }
 prost = { workspace = true }
+redis = { version = "1.7", optional = true, default-features = false, features = [
+    "cluster-async",
+    "connection-manager",
+    "script",
+    "streams",
+    "tokio-rustls-comp",
+] }
 sha2 = "0.10"
 thiserror = { workspace = true }
 tokio = { version = "1.47", default-features = false, features = ["sync", "time"] }
@@ -30,7 +39,8 @@ version = "~1.1.0"
 default-features = false
 
 [dev-dependencies]
-tokio = { version = "1.47", features = ["macros", "rt", "time"] }
+serde_json = { workspace = true }
+tokio = { version = "1.47", features = ["io-util", "macros", "net", "rt", "rt-multi-thread", "time"] }
 
 [lints]
 workspace = true

crates/streams/lua/append.lua

@@ -0,0 +1,38 @@
+
+local retention, grace = tonumber(ARGV[1]), tonumber(ARGV[2])
+local field, digest_arg = ARGV[3], ARGV[5]
+local held = redis.call('HGET', KEYS[2], field)
+local sequence = tonumber(ARGV[4])
+if held then
+  local held_sequence, held_count, first, last, digest =
+    string.match(held, '^(%d+)|(%d+)|([^|]+)|([^|]+)|(%x+)$')
+  held_sequence = tonumber(held_sequence)
+  local next_sequence = held_sequence + tonumber(held_count)
+  if sequence == held_sequence then
+    if digest == digest_arg then
+      return {first, last}
+    end
+    return redis.error_reply('STREAMS_DIVERGENT sequence ' .. ARGV[4] ..
+      ' was already written with different content')
+  end
+  if sequence < next_sequence then
+    return redis.error_reply('STREAMS_STALE sequence ' .. ARGV[4] ..
+      ' is below the next one expected, ' .. next_sequence)
+  end
+end
+if redis.call('HGET', KEYS[3], 'closed') then
+  return redis.error_reply('STREAMS_CLOSED the Workflow that owns this stream ' ..
+    'has closed')
+end
+revive(KEYS[1], KEYS[2])
+local first, last
+for i = 6, #ARGV do
+  last = redis.call('XADD', KEYS[1], '*', 'r', ARGV[i])
+  if not first then
+    first = last
+  end
+end
+redis.call('HSET', KEYS[2], field,
+  ARGV[4] .. '|' .. (#ARGV - 5) .. '|' .. first .. '|' .. last .. '|' .. digest_arg)
+keep(KEYS[1], KEYS[2], #ARGV - 5, last, retention, grace)
+return {first, last}

crates/streams/lua/keep.lua

@@ -0,0 +1,39 @@
+
+local function now_ms()
+  local time = redis.call('TIME')
+  return tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000)
+end
+local function revive(log, meta)
+  if redis.call('EXISTS', log) == 0 then
+    local last = redis.call('HGET', meta, 'last')
+    if last then
+      redis.call('HSET', meta, 'trimmed', last)
+    end
+  end
+end
+local function sweep_sessions(meta, floor_ms)
+  local cursor = redis.call('HGET', meta, 'hwscan') or '0'
+  local found = redis.call('HSCAN', meta, cursor, 'MATCH', 'hw:*', 'COUNT', 50)
+  local fields = found[2]
+  for i = 1, #fields, 2 do
+    local last_ms = string.match(fields[i + 1], '|(%d+)%-%d+|[^|]+$')
+    if last_ms and tonumber(last_ms) < floor_ms then
+      redis.call('HDEL', meta, fields[i])
+    end
+  end
+  redis.call('HSET', meta, 'hwscan', found[1])
+end
+local function keep(log, meta, added, last, retention, grace)
+  local floor_ms = now_ms() - retention
+  local floor = floor_ms .. '-0'
+  local doomed = redis.call('XREVRANGE', log, '(' .. floor, '-', 'COUNT', 1)
+  if #doomed > 0 then
+    redis.call('XTRIM', log, 'MINID', floor)
+    redis.call('HSET', meta, 'trimmed', doomed[1][1])
+  end
+  redis.call('PEXPIRE', log, retention)
+  redis.call('HINCRBY', meta, 'added', added)
+  redis.call('HSET', meta, 'last', last)
+  redis.call('PEXPIRE', meta, retention + grace)
+  sweep_sessions(meta, floor_ms)
+end

crates/streams/lua/promote.lua

@@ -0,0 +1,35 @@
+
+local items = redis.call('LRANGE', KEYS[1], 0, -1)
+local retention, grace = tonumber(ARGV[1]), tonumber(ARGV[2])
+local slots = {}
+for i = 4, #ARGV do
+  slots[ARGV[i]] = {log = KEYS[2 * (i - 3) + 1], meta = KEYS[2 * (i - 3) + 2], added = 0}
+end
+for i = 1, #items, 2 do
+  if not slots[items[i]] then
+    return redis.error_reply('STREAMS_TOPIC the stage holds topic ' .. items[i] ..
+      ', which the promotion did not name')
+  end
+end
+local was_pending = redis.call('HDEL', KEYS[2], ARGV[3])
+if #items == 0 then
+  if was_pending == 1 then
+    return -1
+  end
+  return 0
+end
+for _, slot in pairs(slots) do
+  revive(slot.log, slot.meta)
+end
+for i = 1, #items, 2 do
+  local slot = slots[items[i]]
+  slot.last = redis.call('XADD', slot.log, '*', 'r', items[i + 1])
+  slot.added = slot.added + 1
+end
+for _, slot in pairs(slots) do
+  if slot.added > 0 then
+    keep(slot.log, slot.meta, slot.added, slot.last, retention, grace)
+  end
+end
+redis.call('DEL', KEYS[1])
+return #items / 2

crates/streams/lua/stage.lua

@@ -0,0 +1,10 @@
+
+local chunk = 1000
+for i = 4, #ARGV, chunk do
+  redis.call('RPUSH', KEYS[1], unpack(ARGV, i, math.min(i + chunk - 1, #ARGV)))
+end
+redis.call('PEXPIRE', KEYS[1], ARGV[1])
+redis.call('HSET', KEYS[2], ARGV[2], ARGV[3])
+if redis.call('PTTL', KEYS[2]) < tonumber(ARGV[1]) then
+  redis.call('PEXPIRE', KEYS[2], ARGV[1])
+end

crates/streams/src/lib.rs

@@ -14,8 +14,12 @@ mod error;
 mod memory;
 mod reader;
 mod record;
+#[cfg(feature = "redis")]
+mod redis;
 mod store;
 
+#[cfg(feature = "redis")]
+pub use self::redis::{RedisStore, RedisStoreOptions};
 pub use cursor::{BEGINNING, END, cursor_position, mint_cursor, stream_hash};
 pub use error::{StreamError, StreamResult};
 pub use memory::MemoryStore;

crates/streams/src/memory.rs

@@ -200,6 +200,9 @@ impl StreamStore for MemoryStore {
         if request.records.is_empty() {
             return Err(StreamError::refused("an append needs at least one record"));
         }
+        if request.digest.is_empty() {
+            return Err(StreamError::refused("an append needs the batch's digest"));
+        }
         let mut chains = self.chains.lock().unwrap();
         let chain = chains
             .entry(chain_key(chain_of(request.chain.as_ref())?))

crates/streams/src/reader.rs

@@ -228,7 +228,13 @@ pub async fn read_page(
 
 /// A stored record as a reader delivers it. Fails on bytes that are no record, and on a kind no
 /// writer stores, such as the `SUPERSEDED` that only readers synthesize.
+///
+/// Empty bytes are no record either. Every stored record names its topic, so it is never empty,
+/// and a store hands over an entry it holds without a record as empty bytes.
 fn parse(bytes: &[u8]) -> Result<StreamRecord, String> {
+    if bytes.is_empty() {
+        return Err("the entry holds no stream record".to_string());
+    }
     let mut record = StreamRecord::decode(bytes).map_err(|error| error.to_string())?;
     match StreamRecordKind::try_from(record.kind) {
         Ok(StreamRecordKind::Data | StreamRecordKind::Finish) => {}
@@ -679,6 +685,21 @@ mod tests {
         assert!(error.message.contains("kind 3"), "{error}");
     }
 
+    #[tokio::test]
+    async fn rule_20_4_an_entry_without_a_record_fails_with_its_cursor() {
+        let log = Log::default();
+        log.write("p", 1, b"1");
+        let empty = log.push(Vec::new());
+        log.write("p", 1, b"3");
+        let first = read(&log, BEGINNING, &[]).await.unwrap();
+        assert_eq!(shape(&first), [(cursor("1"), DATA, 1, false)]);
+        let error = read(&log, &first.cursor, &first.state).await.unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::Record);
+        assert_eq!(error.cursor, Some(cursor(&empty)));
+        let after = read(&log, &cursor(&empty), &first.state).await.unwrap();
+        assert_eq!(shape(&after), [(cursor("3"), DATA, 1, false)]);
+    }
+
     #[tokio::test]
     async fn an_unspecified_kind_reads_as_data() {
         let log = Log::default();