[Brook] Pinned the v4 crate and regenerated the Stream bindings.

scripts/_proto/temporal/api/streamservice/v1/stream_service.proto

@@ -17,6 +17,10 @@ message StreamRef {
     // client or Activity ref. The Store keys on id only and never reads these.
     string workflow_id = 2;
     string first_execution_run_id = 3;
+    // The name of the backend that holds the Stream, as the Worker and client config register
+    // it, so every process routes a ref it receives. Empty is the default backend. The Store
+    // never reads it.
+    string backend = 4;
 }
 
 enum ProducerKind {
@@ -69,9 +73,9 @@ enum StreamRecordKind {
     // Lua flags the first batch of a new epoch. The read makes this Record from the flag.
     STREAM_RECORD_KIND_SUPERSEDED = 2;
     STREAM_RECORD_KIND_CLOSE = 3;
-    // The last Record of a read that saw no new Record for idle_timeout. Its offset is the
-    // newest offset read + 1, the place to resume from.
-    STREAM_RECORD_KIND_IDLE = 4;
+    // A Reader that stops after a quiet period sets its own timer and resumes from next_offset.
+    reserved 4;
+    reserved "STREAM_RECORD_KIND_IDLE";
 }
 
 message Supersession {
@@ -101,7 +105,8 @@ message StreamRecord {
     temporal.api.failure.v1.Failure close_failure = 8;
 }
 
-// Idempotent for one id and one Retention. ALREADY_EXISTS for another Retention.
+// Idempotent for one id, one Retention and one auto_close. ALREADY_EXISTS for another
+// Retention or another auto_close.
 message CreateStreamRequest {
     string namespace = 1;
     // Required from Workflow code.
@@ -110,6 +115,11 @@ message CreateStreamRequest {
     google.protobuf.Duration retention = 3;
     // Workflow only. A retry run's first create: the previous run, whose Records it withdraws.
     string withdraw_run = 4;
+    // Close the Stream with no result when no Record was appended for this long, counted from
+    // the create. Empty never closes on its own. It is separate from Retention and the safety
+    // TTL. The Store applies it on the next access, so no call sees the Stream open after the
+    // period. UNIMPLEMENTED (detail stream_unsupported) on a Store that can't.
+    google.protobuf.Duration auto_close = 5;
 }
 
 enum StreamStatus {
@@ -180,9 +190,8 @@ message ReadRecordsRequest {
     // How long to wait for a Stream that nobody created yet. Empty fails at once with
     // NOT_FOUND. At most 60 s.
     google.protobuf.Duration create_wait = 8;
-    // End the read with an IDLE Record when no new Record comes for this long. Empty never
-    // ends on idle. The facade counts it across reads, since one read waits at most 60 s.
-    google.protobuf.Duration idle_timeout = 9;
+    reserved 9;
+    reserved "idle_timeout";
     // Also read the Topics that start with one of these. A Record passes when its Topic is in
     // topics or starts with a prefix. Both empty reads every Topic.
     repeated string topic_prefixes = 10;
@@ -205,4 +214,6 @@ message DescribeStreamResponse {
     // Zero when the Stream has no Record.
     int64 last_offset = 2;
     google.protobuf.Duration retention = 3;
+    // The auto_close of the create. Empty when the Stream has none.
+    google.protobuf.Duration auto_close = 4;
 }

scripts/_proto/temporal/sdk/streams/v1/extension.proto

@@ -27,6 +27,9 @@ message RedisStreamConfig {
     int32 blocking_reads_per_node = 8;
     // Empty is 5 s.
     google.protobuf.Duration response_timeout = 9;
+    // The backend name that lang registered this extension under. The crate puts it in each
+    // StreamRef it builds, such as the one in a create answer.
+    string backend = 10;
 }
 
 // Method `BufferWorkflowCalls`, synchronous. Lang calls it at the end of each Activation of a
@@ -61,6 +64,29 @@ message WorkflowCall {
 message BufferWorkflowCallsResponse {
 }
 
+// Method `FlushRunBuffer`, asynchronous. Lang calls it before it runs a Local Activity, also
+// before each retry that Core runs in the same task. The crate writes the run's buffered
+// creates and appends in the epoch of the Workflow Task that dispatched the Local Activity, so
+// the run's Records come before the Local Activity's. Closes still wait for an accepted
+// completion. The task's completion writes only what came later, with the numbering going on,
+// and sends the task's withdrawal facts to each Stream this call wrote. A Local Activity of a
+// task that already completed, or of a task older than an earlier flush, writes nothing. When
+// the call fails, the run's next completion fails with the same error, so the retry withdraws
+// what was written. Lang then fails the Local Activity attempt.
+message FlushRunBufferRequest {
+    string namespace = 1;
+    string run_id = 2;
+    // `ActivityTask.Start.workflow_task` (field 20) of the Local Activity.
+    temporal.api.streamservice.v1.WorkflowTaskEpoch workflow_task = 3;
+    // How long the write may take. Empty is 10 s.
+    google.protobuf.Duration timeout = 4;
+}
+
+message FlushRunBufferResponse {
+    // The Records the call wrote. Zero when the run had nothing buffered.
+    int32 records = 1;
+}
+
 // Method `OpenWriter`, asynchronous. A buffered writer for an Activity or a client Producer.
 message OpenWriterRequest {
     string namespace = 1;
@@ -91,6 +117,23 @@ message PublishRequest {
 message PublishResponse {
 }
 
+// Method `WriteRecords`, asynchronous. Appends Records at once for an open writer, after the
+// Records that `Publish` buffered, and answers when they are in the Store. It keeps the writer's
+// Producer epoch and `sequence` order, and a Store failure resends the same batch, so it goes in
+// once. Records past the Store's batch limit go as more batches, in order. An awaited publish of
+// an Activity or a client uses this call, so it needs no interval flush. A failure after the
+// Store retries fails the writer, as a failed flush does.
+message WriteRecordsRequest {
+    uint64 writer_id = 1;
+    repeated temporal.api.streamservice.v1.AppendRecord records = 2;
+}
+
+message WriteRecordsResponse {
+    // The offsets of the first and the last Record. Zero when the request has no Record.
+    int64 first_offset = 1;
+    int64 last_offset = 2;
+}
+
 // Method `FlushWriter`, asynchronous. Waits until the Records published so far are in the Store.
 message FlushWriterRequest {
     uint64 writer_id = 1;

scripts/_wit/stream-service.wit

@@ -26,9 +26,8 @@ interface stream-service {
     nanos: option<s32>,
   }
 
-  // The default Stream of a Workflow has the id <workflow id>:<first run id>. A Stream that
-  // the Workflow creates with a name has <workflow id>:<first run id>:<name>. Each part is
-  // percent-encoded. So a caller that knows the first run id builds the id with no call.
+  // A Stream that a Workflow creates has the id <workflow id>:<first run id>:<name>. Each part
+  // is percent-encoded. So a caller that knows the first run id builds the id with no call.
   /// @nexus.doc "Identifies one Stream. The id is opaque to callers."
   /// @nexus.proto "temporal.api.streamservice.v1.StreamRef"
   record stream-ref {
@@ -38,6 +37,8 @@ interface stream-service {
     workflow-id: option<string>,
     /// @nexus.doc "The first run id of the owning run chain. Empty when workflow-id is."
     first-execution-run-id: option<string>,
+    /// @nexus.doc "The name of the backend that holds the Stream, as the Worker and client config register it. Empty is the default backend. The Store never reads it."
+    backend: option<string>,
   }
 
   /// @nexus.proto "temporal.api.streamservice.v1.ProducerKind"
@@ -104,7 +105,6 @@ interface stream-service {
     stream-record-kind-data,
     stream-record-kind-superseded,
     stream-record-kind-close,
-    stream-record-kind-idle,
   }
 
   // For an Activity, Local Activity or client Producer: withdraw the Records of this
@@ -139,19 +139,19 @@ interface stream-service {
     /// @nexus.doc "The place of this Record in the Stream. Ordered and unique, with gaps."
     /// @nexus.offset "stream-offset"
     offset: option<s64>,
-    /// @nexus.doc "Empty on CLOSE, IDLE and SUPERSEDED Records."
+    /// @nexus.doc "Empty on CLOSE and SUPERSEDED Records."
     topic: option<string>,
     producer: option<producer>,
     /// @nexus.doc "Set on a data Record."
     body: option<payload>,
     /// @nexus.doc "Set on a SUPERSEDED Record. Lua flags the batch, and the read makes it."
     supersession: option<supersession>,
-    /// @nexus.doc "Set on a close Record with an outcome. An IDLE Record carries only the offset, the newest read + 1."
+    /// @nexus.doc "Set on a close Record with an outcome."
     close-result: option<payload>,
     close-failure: option<failure>,
   }
 
-  /// @nexus.doc "Create a Stream. Idempotent for one id and one Retention. ALREADY_EXISTS for another Retention."
+  /// @nexus.doc "Create a Stream. Idempotent for one id, one Retention and one auto-close. ALREADY_EXISTS for another Retention or another auto-close."
   /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamRequest"
   record create-stream-request {
     /// @nexus.doc "Required from Workflow code. Empty elsewhere asks the service for an id."
@@ -160,6 +160,8 @@ interface stream-service {
     retention: option<duration>,
     /// @nexus.doc "Workflow only. A retry run's first batch or create: the previous run, whose Records it withdraws."
     withdraw-run: option<string>,
+    /// @nexus.doc "Close the Stream with no result when no Record was appended for this long. Empty never closes on its own. It is separate from Retention."
+    auto-close: option<duration>,
     /// @nexus.omit
     namespace: placeholder,
   }
@@ -255,8 +257,6 @@ interface stream-service {
     wait: option<duration>,
     /// @nexus.doc "How long to wait for an uncreated Stream. Empty fails at once. At most 60 s."
     create-wait: option<duration>,
-    /// @nexus.doc "End the read with an IDLE Record when no new Record comes for this long. Empty never ends on idle. The facade counts it across reads."
-    idle-timeout: option<duration>,
     /// @nexus.omit
     namespace: placeholder,
   }
@@ -285,18 +285,18 @@ interface stream-service {
     /// @nexus.offset "stream-offset"
     last-offset: option<s64>,
     retention: option<duration>,
+    /// @nexus.doc "The auto-close of the create. Empty when the Stream has none."
+    auto-close: option<duration>,
   }
 
   /// @nexus.operation name="CreateStream"
   create-stream: func(request: create-stream-request) -> create-stream-response;
 
-  /// @nexus.doc "In Workflow code, the Brook transport puts this call into a buffer."
+  /// @nexus.doc "In Workflow code, the Brook transport puts this call into the run's buffer, and the Workflow facade drops the response."
   /// @nexus.operation name="AppendRecords"
-  /// @nexus.one-way
   append-records: func(request: append-records-request) -> append-records-response;
 
   /// @nexus.operation name="CloseStream"
-  /// @nexus.one-way
   close-stream: func(request: close-stream-request) -> close-stream-response;
 
   /// @nexus.doc "Not for Workflow code in Brook. A long poll. NOT_FOUND for an uncreated Stream."
@@ -308,7 +308,6 @@ interface stream-service {
   ///   done-field="done"
   ///   offset-field="from-offset"
   ///   next-offset-field="next-offset"
-  ///   idle-timeout-field="idle-timeout"
   read-records: func(request: read-records-request) -> read-records-response;
 
   /// @nexus.operation name="DescribeStream"

scripts/gen_stream_service.py

@@ -2,7 +2,7 @@
 
 The WIT and the protos are copies of the temporalio-streams crate's, in scripts/_wit and
 scripts/_proto. Run scripts/gen_protos.py first for the proto modules. No nexgen release has the
-offset, one-way, long-poll and client features yet, so the script installs nexgen at
+offset, long-poll and client features yet, so the script installs nexgen at
 NEX_GEN_REVISION from moetemp/nexgen. NEX_GEN_STREAMS_BIN may name a build of that revision
 instead.
 """
@@ -33,7 +33,7 @@
 
 NEX_GEN_REPOSITORY = "https://github.com/moetemp/nexgen"
 # The head of moe/brook/v3/n7-long-poll-loop. A branch moves, so the pin is the commit.
-NEX_GEN_REVISION = "64ee48a79abe18077d727f5067a6302c45c25c9d"
+NEX_GEN_REVISION = "25b9c1be8f51245618297415aa51df6928c58504"
 nex_gen_root = base_dir / ".nexgen-streams" / NEX_GEN_REVISION