[Jetty] Added the stream driver.

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

@@ -40,9 +40,16 @@ service StreamService {
   rpc Unsubscribe(UnsubscribeRequest) returns (UnsubscribeResponse);
   // Waits on the stream's notifier, for a client reader whose read came back empty. Returns when
   // the notifier's counter is above `after_counter`, when the stream closed, or at the server's
-  // wait limit (`streamnotifier.pollMaxWait`). The reader then reads again. After the wait limit,
-  // that read finds records whose notification was lost, or comes back empty again.
+  // wait limit (`streamnotifier.pollMaxWait`). The poll starts the notifier, and Core reads the
+  // stream once more just after it sends the poll, since a record that landed before the notifier
+  // existed woke no one. A newer record there answers at once. The reader then reads again.
+  // After the wait limit, that read finds records whose notification was lost, or comes back
+  // empty again.
   rpc WaitForNotification(WaitForNotificationRequest) returns (WaitForNotificationResponse);
+  // 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);
@@ -367,6 +374,10 @@ message SubscribeResponse {
   string chain_first_run_id = 1;
   // Whether this call sent the attach. False when this process attached the reader before.
   bool attached = 2;
+  // The notification counter of the stream's newest record, read after the attach landed, zero
+  // for an empty stream. A record that landed before the notifier existed woke no one, so a
+  // reader whose last read stopped below this counter reads once more before it sleeps.
+  int64 newest_counter = 3;
 }
 
 message UnsubscribeRequest {
@@ -402,6 +413,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/sdk-core/CHANGELOG.md

@@ -58,6 +58,17 @@ relevant information.
   `<prefix>:{<slot tag>}:n`, and Core rings one of `notify_doorbells` (16 when zero) doorbell
   streams when that stream takes its first entry. `testdata/redis/notify_slots.json` pins the slot
   tags for other SDKs. Drivers read as the consumer group `streams-driver`.
+* Experimental: the new `RunStreamDriver` stream call runs a stream driver until the caller
+  cancels it. The driver blocks on each doorbell and sweeps every busy slot every
+  `sweep_interval` (30 seconds). It folds each drain to the newest counter per stream, sends one
+  `NotifyStreams` call per namespace, and acks the entries only once the call landed. It takes
+  entries another driver left unacked for `claim_idle` (60 seconds). Each driver needs its own
+  `consumer`. `StreamDriver` runs the same in Rust.
+* Experimental: a reader reads first and subscribes only when a read comes back empty, since a
+  notification for a stream with no notifier is dropped. `SubscribeResponse.newest_counter` names
+  the newest record after the attach landed, so a reader whose read stopped below it reads once
+  more before it sleeps. `WaitForNotification` reads the stream once more just after it sends the
+  poll, and answers at once for a record that landed before the notifier existed.
 * Experimental: the new `FlushNotifications` stream call waits up to 10 seconds for the
   notifications still out. A process calls it before it stops.
 * Experimental: a server without the stream notifier answers the first notifier call with

crates/sdk-core/src/streams.rs

@@ -59,14 +59,42 @@ impl NotifierClient for ConnectionNotifierClient {
         &self,
         notifications: Vec<StreamNotification>,
     ) -> Result<(), NotifyError> {
+        self.notify_streams_each(notifications)
+            .await?
+            .into_iter()
+            .collect()
+    }
+
+    async fn notify_streams_each(
+        &self,
+        notifications: Vec<StreamNotification>,
+    ) -> Result<Vec<Result<(), NotifyError>>, NotifyError> {
+        let namespaces: Vec<_> = notifications
+            .iter()
+            .map(|n| n.chain.namespace.clone())
+            .collect();
+        let mut results: Vec<_> = namespaces.iter().map(|_| Ok(())).collect();
         for request in notify_streams_requests(notifications, self.connection.identity()) {
-            self.connection
+            // A request keeps the order of its namespace's notifications.
+            let indexes: Vec<_> = (0..namespaces.len())
+                .filter(|&index| namespaces[index] == request.namespace)
+                .collect();
+            let response = self
+                .connection
                 .workflow_service()
                 .notify_streams(request.into_request())
                 .await
-                .map_err(notify_error)?;
+                .map_err(notify_error)?
+                .into_inner();
+            // A server that answers no results applied every entry.
+            for (index, result) in indexes.into_iter().zip(response.results) {
+                if result.code != tonic::Code::Ok as i32 {
+                    let code = tonic::Code::from(result.code);
+                    results[index] = Err(notify_error(tonic::Status::new(code, result.message)));
+                }
+            }
         }
-        Ok(())
+        Ok(results)
     }
 
     async fn attach_workflow(&self, attachment: WorkflowAttachment) -> Result<(), NotifyError> {

crates/streams/Cargo.toml

@@ -20,6 +20,7 @@ redis = ["dep:redis", "dep:crc16"]
 [dependencies]
 async-trait = "0.1"
 crc16 = { version = "0.4", optional = true }
+gethostname = "1.0.2"
 lru = "0.18"
 prost = { workspace = true }
 redis = { version = "1.7", optional = true, default-features = false, features = [
@@ -31,7 +32,7 @@ redis = { version = "1.7", optional = true, default-features = false, features =
 ] }
 sha2 = "0.10"
 thiserror = { workspace = true }
-tokio = { version = "1.47", default-features = false, features = ["sync", "time"] }
+tokio = { version = "1.47", default-features = false, features = ["macros", "rt", "sync", "time"] }
 tracing = "0.1"
 
 [dependencies.temporalio-common]

crates/streams/docs/redis.md

@@ -104,7 +104,9 @@ expiry from the record at its cursor. It exists so tests run those fallbacks on
 Listing a chain's topics scans every primary, as a delete does, and needs `+scan`.
 
 **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
+`STREAMS_REDIS_CLUSTER_URL` a cluster seed. `scripts/redis-test-servers.sh` starts both, the
+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 whose

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

@@ -1,14 +1,17 @@
 #!/usr/bin/env bash
 # Starts the Redis servers the stream store tests run against: one standalone server, and a
-# cluster of three primaries. 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.
+# 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.
 #
 #   STREAMS_REDIS_URL=redis://localhost:6384/0
 #   STREAMS_REDIS_CLUSTER_URL=redis://localhost:7101/0
 set -euo pipefail
 
 STANDALONE=${STREAMS_STANDALONE_PORT:-6384}
-CLUSTER=(7101 7102 7103)
+BASE=${STREAMS_CLUSTER_BASE:-7101}
+CLUSTER=($BASE $((BASE + 1)) $((BASE + 2)) $((BASE + 3)) $((BASE + 4)) $((BASE + 5)))
 DATA=${STREAMS_REDIS_DATA:-${TMPDIR:-/tmp}/streams-redis}
 mkdir -p "$DATA"
 
@@ -33,15 +36,15 @@ for port in "${CLUSTER[@]}"; do
   start "$port" --cluster-enabled yes --cluster-config-file nodes.conf
 done
 for _ in $(seq 50); do
-  redis-cli -p "${CLUSTER[2]}" ping >/dev/null 2>&1 && break
+  redis-cli -p "${CLUSTER[5]}" ping >/dev/null 2>&1 && break
   sleep 0.1
 done
 if ! redis-cli -p "${CLUSTER[0]}" cluster info | grep -q cluster_state:ok; then
   nodes=()
   for port in "${CLUSTER[@]}"; do
     nodes+=("127.0.0.1:$port")
   done
-  redis-cli --cluster create "${nodes[@]}" --cluster-replicas 0 --cluster-yes >/dev/null
+  redis-cli --cluster create "${nodes[@]}" --cluster-replicas 1 --cluster-yes >/dev/null
   for _ in $(seq 50); do
     redis-cli -p "${CLUSTER[0]}" cluster info | grep -q cluster_state:ok && break
     sleep 0.1