[FROZEN][Bridge][Optional] Restored the notification log and the stream driver.

.github/workflows/ci.yml

@@ -89,3 +89,12 @@ jobs:
         env:
           STREAMS_REQUIRE: redis
           STREAMS_REDIS_CLUSTER_ADMIN: "1"
+
+  optional-restore:
+    name: The optional layers' restore commits copy 4aebbd4a
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - run: scripts/check-optional-restore.py

crates/streams/docs/managed-redis.md

@@ -40,18 +40,23 @@ MemoryDB is the AWS production target for the Redis store.
 MemoryDB takes users through its API, not `ACL SETUSER`. An access string has no password, no
 `reset` rules and no `config` rule
 (https://docs.aws.amazon.com/memorydb/latest/devguide/clusters.acls.html). This is the access
-string for one namespace's applications, with owner deletes. The
+string for one namespace's applications, with the notification log on and owner deletes. The
 conformance suite runs as a user with exactly these rules on local Redis.
 
 ```text
-on ~temporal-streams:{my-ns:* -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hlen +hincrby +hdel +hscan +sadd +smembers +rpush +lrange +exists +del +unlink +scan +pexpire +pttl +time +info +ping +hello +client|setinfo +cluster|slots
+on ~temporal-streams:{my-ns:* ~temporal-streams:{*}:n ~temporal-streams:{*}:bell -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hlen +hincrby +hdel +hscan +sadd +smembers +rpush +lrange +exists +del +unlink +scan +pexpire +pttl +time +info +ping +hello +client|setinfo +cluster|slots
 ```
 
-It differs from the `ACL SETUSER` rules in [redis.md](redis.md) in three ways.
+It differs from the `ACL SETUSER` rules in [redis.md](redis.md) in four ways.
 
 - It has no `>secret`. The password goes in the user's authentication mode.
 - It has no `resetkeys` and no `resetchannels`. A new user has no keys and no channels.
 - It has no `+config|get`. The store goes on without the eviction check.
+- The doorbell keys use `~`, not `%W~`. We do not know if MemoryDB takes key selectors with
+  permissions. `~` gives read access too, which the application does not need.
+
+Without the notification log, drop the two `{*}` key patterns. A driver also needs
+`+xreadgroup +xautoclaim +xack +xdel +xgroup|create` on the `{*}` keys.
 
 Create the user and its ACL with the AWS CLI. Then give the ACL to the cluster.
 
@@ -72,8 +77,8 @@ aws memorydb create-acl --acl-name streams --user-names streams-app
 - **Serverless.** It supports `EVAL`, `EVALSHA`, `SCRIPT LOAD`, the stream and consumer group
   commands the store uses, `HSCAN`, `SCAN`, `UNLINK` and `TIME`. `CLUSTER SLOTS` answers one
   shard with every slot
-  (https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/SupportedCommands.html). We did not
-  run the suite on it.
+  (https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/SupportedCommands.html). The driver
+  then sweeps every slot on one node. We did not run the suite on it.
 - **Access.** `CONFIG` and `ACL SETUSER` are restricted. ElastiCache users take an access string
   through the API, so use the MemoryDB access string above.
 

crates/streams/docs/redis.md

@@ -43,6 +43,24 @@ doesn't read the floors hash, so its appends can trim records that a positions-o
 topic that a positions-only reader reads must take its appends only through this crate's store.
 Other providers on such a topic are unsupported.
 
+**Notification log.** With `notification_log` on, the append script also adds an entry to the
+notification stream of the chain's Cluster slot, `<prefix>:{<tag>}:n`, where the tag is the one
+`testdata/redis/notify_slots.json` pins for that slot. So the log, the meta and the entry share a
+slot, one script writes them, and a reshard moves them together. A standalone server uses the same
+keys. An entry holds `k`, the log key, and `p`, the append's last entry id. The stream keeps about
+the newest 10,000 entries, also ones a driver took and has not acked. So a driver that falls more
+than 10,000 appends behind on a slot loses the oldest entries, and a quiet stream that shares the
+slot with a busy one can lose its only entry. Its readers then wake on their recheck. A repeat of
+the newest batch adds none. The script answers the stream's length too, and when it is 1, the store
+adds the slot to doorbell `d = s % doorbells`, at
+`<prefix>:{<tag of slot d * 16384 / doorbells>}:bell`, apart from the script, so a lost ring is
+possible. The doorbells spread over the slot range, so a cluster's primaries share them, and an
+application and its drivers must agree on the number of doorbells. Drivers read the slot streams
+and the doorbells as the consumer group `streams-driver` (`XREADGROUP`, `XAUTOCLAIM`, `XACK`,
+`XDEL`, `XGROUP CREATE`), and sweep with `XLEN` over each primary's slots from `CLUSTER 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.
+
 **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
@@ -94,30 +112,37 @@ ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetcha
 
 A cluster client reads the topology with `CLUSTER SLOTS` first. An ACL user lives on one node,
 so on a cluster the user must exist on every node. Deleting an owner's streams also needs
-`+scan`. `+config|get` only reads `maxmemory-policy`, and the store goes on when a server
-refuses it.
+`+scan`. With the notification log on, an application also needs `~temporal-streams:{*}:n
+%W~temporal-streams:{*}:bell`, since its appends write the slot streams. Those entries name
+other namespaces' streams and positions, never records. A driver needs those keys with read and
+write, and `+xreadgroup +xautoclaim +xack +xdel +xgroup|create +xlen +xadd +cluster|slots`.
+`+config|get` only reads `maxmemory-policy`, and the store goes on when a server refuses it.
 
 **Plain mode.** `RedisStoreOptions::plain` runs the store with no Lua. An append is a
 `MULTI` of `XADD`s that trim by age, a read is `XREAD` alone, and no meta and no close flag is
 kept. So the store declares neither producer state nor a kept last position, and the service takes
 their fallbacks: readers drop copies and end at the first `CLOSE` record, and a read decides
 expiry from the record at its cursor. It exists so tests run those fallbacks on a real store.
 Listing a chain's topics scans every primary, as a delete does, and needs `+scan`. With no script
-there is no floor hold, so plain mode declares no holds.
+there is no notification log and no floor hold, so plain mode declares neither holds nor a
+change feed, and a config that asks for a log is refused.
 
 **Tests.** The Redis tests run when `STREAMS_REDIS_URL` names a standalone server and
 `STREAMS_REDIS_CLUSTER_URL` a cluster seed. `scripts/redis-test-servers.sh` starts both, the
-cluster with one replica per primary. The conformance suite runs on each, once more as a user
-with exactly the ACL above, and once in plain mode with the cases of the fallbacks.
-`tests/fallbacks.rs` adds faults on plain Redis too. The cross-implementation test also needs
-`STREAMS_BROOK_PY`, a Python SDK checkout with its own Redis provider. When `STREAMS_REQUIRE`
-names `redis`, a Redis test with no URL set fails and does not skip.
+cluster with one replica per primary. `STREAMS_REDIS_CLUSTER_ADMIN=1` lets the driver tests move
+a slot and fail a primary over, so set it only for a cluster of the tests' own. The
+conformance suite runs on each, once more as a user with exactly the ACL above, and once in
+plain mode with the cases of the fallbacks. `tests/fallbacks.rs` adds faults on plain Redis
+too. The cross-implementation test also needs `STREAMS_BROOK_PY`, a Python SDK checkout with its
+own Redis provider. When `STREAMS_REQUIRE` names `redis`, a Redis test with no URL set fails and
+does not skip.
 
 **Managed Redis.** [managed-redis.md](managed-redis.md) covers Amazon MemoryDB and ElastiCache,
 with the access string that those services take in place of `ACL SETUSER`.
 
 **Kvrocks lane.** `STREAMS_KVROCKS_URL` names an Apache Kvrocks server, which keeps its data on
 disk. The script starts one when it finds a `kvrocks` binary or Docker. The conformance suite runs
-on it standalone, with the store cases and owner deletes. This lane is a test only. We make no
-support claim for Kvrocks. Kvrocks has no ACL users and runs no Redis Cluster, so those lanes do
-not run.
+on it standalone, with the store cases, owner deletes and the notification log. This lane is a
+test only. We make no support claim for Kvrocks. Its doorbell case does not run, since Kvrocks
+answers an empty `XREADGROUP` where Redis fails with `NOGROUP`, so a driver never makes the
+doorbell's group. Kvrocks has no ACL users and runs no Redis Cluster, so those lanes do not run.

crates/streams/protos/temporal/sdk/core/streams/streams.proto

@@ -54,6 +54,10 @@ service StreamService {
   // unsupported when the config turns no notifier client on.
   rpc DescribeStreamNotifier(temporal.api.workflowservice.v1.DescribeStreamNotifierRequest)
       returns (temporal.api.workflowservice.v1.DescribeStreamNotifierResponse);
+  // Runs a stream driver over the store's notification log until the caller cancels the call.
+  // It answers only when the server takes no stream notifications, or when the store keeps no
+  // log. Each driver of one store needs its own consumer name.
+  rpc RunStreamDriver(RunStreamDriverRequest) returns (RunStreamDriverResponse);
   // Reads again the exact range an earlier read answered, for a positions-only Workflow reader on
   // replay, and checks it against that read's digest. Waits for nothing and checks no owner.
   rpc ReadRange(ReadRangeRequest) returns (ReadResponse);
@@ -104,6 +108,10 @@ message StreamStoreConfig {
   // owner does. After it the read fails as NOT_FOUND. Unset waits 60 seconds, and zero fails at
   // once.
   google.protobuf.Duration owner_start_grace = 6;
+  // Each append writes a notification entry in the same store step, and producers send no
+  // notifications. A `StreamDriver` reads the entries and notifies the server, so readers wake
+  // only while one runs. Off by default. A store without a notification log refuses it.
+  bool notification_log = 7;
 }
 
 // Streams in a Redis or Redis Cluster that the service connects to itself.
@@ -127,6 +135,9 @@ message RedisStoreConfig {
   google.protobuf.Duration floor_lease = 7;
   // How long a hold lasts after its reader released it. Unset keeps one day.
   google.protobuf.Duration floor_release_grace = 8;
+  // With `StreamStoreConfig.notification_log`, how many doorbell streams the log rings, each one
+  // a driver blocks on. Zero uses 16.
+  uint32 notify_doorbells = 9;
   // Runs with no Lua, so the store keeps no producer state and no last position, and the service
   // takes their fallbacks. For tests of the fallbacks on a real store.
   bool plain = 10;
@@ -615,6 +626,21 @@ message WaitForNotificationResponse {
   temporal.api.failure.v1.Failure close_failure = 6;
 }
 
+message RunStreamDriverRequest {
+  // The driver's name in the log's consumer group, unique among the store's drivers.
+  string consumer = 1;
+  // How often the driver looks at every slot that holds entries. Unset uses 30 seconds.
+  google.protobuf.Duration sweep_interval = 2;
+  // How long another driver's unacked entries wait before this one takes them. Unset uses 60
+  // seconds.
+  google.protobuf.Duration claim_idle = 3;
+  // The most entries one read takes, and so the most streams one notify call names. Zero uses
+  // 100.
+  uint32 batch = 4;
+}
+
+message RunStreamDriverResponse {}
+
 message DeleteOwnerRequest {
   string namespace = 1;
   StreamOwnerKind owner_kind = 2;

crates/streams/scripts/redis-test-servers.sh

@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 # Starts the Redis servers the stream store tests run against: one standalone server, and a
-# cluster of three primaries with one replica each.
+# cluster of three primaries with one replica each, so the driver tests can fail a primary over.
 # Uses a local redis-server when there is one, and docker on a Linux host otherwise, since a
 # cluster's nodes must reach each other on the addresses they announce. STREAMS_CLUSTER_BASE moves
 # the six cluster ports.