engine/src/workers/stream/adapters/redis_adapter.rs (1)

423-470: Consider adding a Redis-specific integration test for stream names with colons.

The KV adapter has a dedicated test (list_all_stream_preserves_stream_names_with_colons), but the Redis adapter lacks an equivalent test. While the shared helpers are unit-tested in mod.rs, an integration test for Redis would provide confidence that the SCAN-based list_all_stream correctly handles encoded keys in a real Redis environment.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@engine/src/workers/stream/adapters/redis_adapter.rs` around lines 423 - 470,
Add an integration test that exercises RedisAdapter::list_all_stream to ensure
stream names containing colons are preserved: create a test (e.g.,
list_all_stream_preserves_stream_names_with_colons_redis) that boots a real
Redis instance or uses the test Redis fixture, writes keys using the same
storage key encoding (so parse_stream_storage_key can decode them) for a stream
name containing colons and multiple group IDs, then call list_all_stream() and
assert the returned StreamMetadata contains the original stream id (with colons)
and the expected sorted groups; target the Redis adapter code path
(list_all_stream in redis_adapter.rs) and use the same connection/publisher
setup the adapter uses so SCAN is exercised end-to-end.

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

engine/src/workers/stream/adapters/redis_adapter.rs (1)

409-420: Use SCAN instead of KEYS in list_groups to avoid Redis blocking.

Line 414 uses KEYS, which blocks Redis under larger keyspaces. list_all_stream already uses cursor-based scanning in the same file; list_groups should follow the same pattern for consistency.

♻️ Proposed refactor

     async fn list_groups(&self, stream_name: &str) -> anyhow::Result<Vec<String>> {
         let mut conn = self.publisher.lock().await;
         let prefix = stream_storage_prefix(stream_name);
         let pattern = format!("{prefix}*");
-
-        match conn.keys::<_, Vec<String>>(pattern).await {
-            Ok(keys) => Ok(keys
-                .into_iter()
-                .filter_map(|key| key.strip_prefix(&prefix).map(|s| s.to_string()))
-                .collect()),
-            Err(e) => Err(anyhow::anyhow!("Failed to list groups from Redis: {}", e)),
-        }
+        use std::collections::BTreeSet;
+        let mut cursor = 0u64;
+        let mut groups = BTreeSet::new();
+
+        loop {
+            let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
+                .arg(cursor)
+                .arg("MATCH")
+                .arg(&pattern)
+                .query_async(&mut *conn)
+                .await
+                .map_err(|e| anyhow::anyhow!("Failed to scan groups from Redis: {}", e))?;
+
+            for key in keys {
+                if let Some(group) = key.strip_prefix(&prefix) {
+                    groups.insert(group.to_string());
+                }
+            }
+
+            cursor = next_cursor;
+            if cursor == 0 {
+                break;
+            }
+        }
+
+        Ok(groups.into_iter().collect())
     }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@engine/src/workers/stream/adapters/redis_adapter.rs` around lines 409 - 420,
The list_groups function uses Redis KEYS (via conn.keys) which can block large
keyspaces; replace it with a cursor-based SCAN loop like list_all_stream:
acquire the same publisher lock, build the same prefix and pattern
(stream_storage_prefix and pattern = format!("{prefix}*")), then iterate SCAN
with MATCH pattern and a reasonable COUNT (e.g. 1000), accumulating returned
keys until cursor == 0, map each key via strip_prefix(&prefix).map(|s|
s.to_string()) and return the Vec<String>; propagate any Redis errors as
anyhow::Error similarly to the existing Err branch.