[Jetty] Subscribed readers to the notifier and handed wakes to lang.

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

@@ -31,6 +31,18 @@ service StreamService {
   // Waits until the stream notifications this process sent are out, for at most 10 seconds. A
   // notification goes in the background, so a process calls it before it stops.
   rpc FlushNotifications(FlushNotificationsRequest) returns (FlushNotificationsResponse);
+  // Attaches a Workflow's reader of a stream to the stream's notifier, so the notifier wakes the
+  // Workflow with a `StreamWake` on its next Workflow Task. A reader calls it before its first
+  // read. The process attaches each subscription once and answers a repeat from memory, and the
+  // server takes a repeat from another process as the same attach.
+  rpc Subscribe(SubscribeRequest) returns (SubscribeResponse);
+  // Detaches a Workflow's reader that stops before the stream ends.
+  rpc Unsubscribe(UnsubscribeRequest) returns (UnsubscribeResponse);
+  // Waits on the stream's notifier, for a client reader whose read armed the stream. 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 with `arm`. After the
+  // wait limit, that read finds records whose notification was lost, or arms again.
+  rpc WaitForNotification(WaitForNotificationRequest) returns (WaitForNotificationResponse);
   // What the store can do beyond the log every store gives, and its limits.
   rpc DescribeStore(DescribeStoreRequest) returns (DescribeStoreResponse);
 }
@@ -256,6 +268,61 @@ message FlushNotificationsRequest {}
 
 message FlushNotificationsResponse {}
 
+message SubscribeRequest {
+  // The stream to read. The reader's Workflow is in the stream's namespace.
+  StreamAddress stream = 1;
+  // The reader's Workflow and run.
+  string workflow_id = 2;
+  string run_id = 3;
+  // Names the reader in its run. Lang derives it from the reader's call number, so replay gives
+  // the same one. At most 256 bytes.
+  string subscription_id = 4;
+  // Counts the times the reader subscribed again after a wake with `resubscribe`. Each value is a
+  // new attach, since the server answers a repeat of a dropped attach as done. Lang counts it from
+  // the wakes, so replay gives the same one.
+  uint32 generation = 5;
+}
+
+message SubscribeResponse {
+  // The first run of the stream's run chain, which keys the notifier the reader attached to.
+  string chain_first_run_id = 1;
+  // Whether this call sent the attach. False when this process attached the reader before.
+  bool attached = 2;
+}
+
+message UnsubscribeRequest {
+  // The stream, the reader's run, the subscription and its generation, as the last Subscribe
+  // named them.
+  StreamAddress stream = 1;
+  string run_id = 2;
+  string subscription_id = 3;
+  uint32 generation = 4;
+}
+
+message UnsubscribeResponse {}
+
+message WaitForNotificationRequest {
+  // The stream the reader reads.
+  StreamAddress stream = 1;
+  // The `wake_after` of the read that armed the stream.
+  int64 after_counter = 2;
+}
+
+message WaitForNotificationResponse {
+  // The highest counter the notifier has seen.
+  int64 counter = 1;
+  // The position that came with `counter`.
+  string position = 2;
+  // The wait reached its limit with no counter above `after_counter` and no close.
+  bool timed_out = 3;
+  // The stream closed. The reader reads to its end.
+  bool closed = 4;
+  // The result the stream closed with, when it gave one.
+  temporal.api.common.v1.Payload close_result = 5;
+  // The failure the stream closed with, when it closed with one.
+  temporal.api.failure.v1.Failure close_failure = 6;
+}
+
 message DeleteOwnerRequest {
   string namespace = 1;
   StreamOwnerKind owner_kind = 2;

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

@@ -140,6 +140,9 @@ message WorkflowActivationJob {
         ResolveNexusOperationStart resolve_nexus_operation_start = 15;
         // A nexus operation resolved.
         ResolveNexusOperation resolve_nexus_operation = 16;
+        // A stream that one of the Workflow's readers subscribed to moved or closed. Numbers 17
+        // and 18 are left to the external stream replay job and to Nexus operation progress.
+        ResolveStreamWake resolve_stream_wake = 19;
         // Remove the workflow identified by the [WorkflowActivation] containing this job from the
         // cache after performing the activation. It is guaranteed that this will be the only job
         // in the activation if present.
@@ -384,6 +387,34 @@ message ResolveNexusOperation {
     nexus.NexusOperationResult result = 2;
 }
 
+// A wake for one of the Workflow's stream readers, as the server folded it onto this Workflow
+// Task's scheduled event.
+//
+// Core issues at most one per subscription per Workflow Task, in the task's first activation. It
+// comes from History, so replay issues the same job at the same point. A wake says that the
+// stream moved, not what it holds, so the reader reads the stream itself.
+message ResolveStreamWake {
+    // The subscription id the reader subscribed with.
+    string subscription_id = 1;
+    // The position that came with `counter`, in the producer's terms.
+    string position = 2;
+    // The highest counter the stream's notifier has seen. The reader reads when it is above the
+    // `wake_after` of the read that armed the stream.
+    int64 counter = 3;
+    // The metadata that came with `counter`.
+    map<string, string> metadata = 4;
+    // The notifier asks the reader to read again whatever the counter, after a quiet period.
+    bool recheck = 5;
+    // The stream closed. The reader reads to its end, then sees the outcome.
+    bool closed = 6;
+    // The result the stream closed with, when it gave one.
+    temporal.api.common.v1.Payload close_result = 7;
+    // The failure the stream closed with, when it closed with one.
+    temporal.api.failure.v1.Failure close_failure = 8;
+    // The notifier dropped the subscription. The reader subscribes again.
+    bool resubscribe = 9;
+}
+
 message RemoveFromCache {
     string message = 1;
 

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

@@ -33,6 +33,12 @@ message Success {
     repeated uint32 used_internal_flags = 6;
     // The versioning behavior this workflow is currently using
     temporal.api.enums.v1.VersioningBehavior versioning_behavior = 7;
+    // Asks Core to complete the workflow task now if local activities are still running, as it
+    // does at a local activity heartbeat. The completion carries the commands and markers so far
+    // and asks the server for a new task, and the local activities run on. Lang sets it from a
+    // count or a clock, so Core ignores it while replaying. Without running local activities,
+    // the task completes as it would anyway.
+    bool heartbeat_task = 8;
 }
 
 // Failure to activate or execute a workflow

crates/protos/src/protos/mod.rs

@@ -81,6 +81,7 @@ pub mod coresdk {
                     commands: v,
                     used_internal_flags: vec![],
                     versioning_behavior: VersioningBehavior::Unspecified.into(),
+                    heartbeat_task: false,
                 }
             }
         }
@@ -1254,6 +1255,9 @@ pub mod coresdk {
                         workflow_activation_job::Variant::ResolveNexusOperation(_) => {
                             write!(f, "ResolveNexusOperation")
                         }
+                        workflow_activation_job::Variant::ResolveStreamWake(w) => {
+                            write!(f, "ResolveStreamWake({}, {})", w.subscription_id, w.counter)
+                        }
                     }
                 }
             }

crates/sdk-core/CHANGELOG.md

@@ -66,6 +66,26 @@ relevant information.
   finds the owner ended closes only the store. The notifier's owner check closes the notifier.
 * Experimental: `StreamStore` has `close_topic_with` and `close_stream`, which carry the close
   outcome to a wrapper that tells the notifier.
+* Experimental: the new `Subscribe` and `Unsubscribe` stream calls attach a Workflow's reader to
+  a stream's notifier as a `workflow_target`, and detach it. The attach's request id is
+  `jetty/<run id>/<subscription id>`, so a rerun attaches as the same reader. After a wake with
+  `resubscribe`, lang raises `SubscribeRequest.generation`, which makes a new attach. A process
+  attaches each subscription once and answers a repeat from memory. The attach names the chain
+  that holds the Workflow id now, and on an empty stream it arms the stream, so the first append
+  wakes the Workflow.
+* Experimental: the new `WaitForNotification` stream call waits on a stream's notifier with the
+  server's `PollStreamNotifier` long poll, for a client reader whose read armed the stream. It
+  returns when the notifier's counter passes the read's `wake_after`, when the stream closed, or
+  with `timed_out` at the server's wait limit. The reader then reads again with `arm`.
+* Experimental: a stream wake for one of a Workflow's readers reaches lang as a
+  `ResolveStreamWake` job, at most one per subscription per workflow task, folded from the
+  `stream_wakes` on the task's scheduled events. Replay delivers the same jobs in the same
+  activations. A task scheduled with wakes ends a workflow task heartbeat chain, so it keeps an
+  activation of its own on replay.
+* Experimental: `Success.heartbeat_task` on a workflow activation completion asks Core to
+  complete the workflow task now while local activities run, as at a local activity heartbeat.
+  The task carries the commands and markers so far and asks for a new task. Core ignores the
+  flag while replaying, since lang sets it from a count or a clock.
 * Experimental: the stream service's `RetireProducer` call drops one producer attempt's dedupe
   state on a stream, which the Redis store keeps as one field of the topic's meta hash until
   retention. A producer that writes once per short-lived attempt calls it after the attempt's

crates/sdk-core/src/core_tests/mod.rs

@@ -4,6 +4,8 @@ mod event_groups;
 mod external_streams;
 mod queries;
 mod replay_flag;
+mod stream_wakes;
+mod task_heartbeat;
 mod updates;
 mod workers;
 mod workflow_cancels;