[Brook] Pinned crate s15 and dropped the buffered writer calls.

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

@@ -1,4 +1,4 @@
-// Draft for the Brook v2 API proposal. It backs stream-service.wit in the same directory.
+// Draft for the Brook API proposal. It backs crates/streams/nexus/stream-service.wit.
 // For the draft, every message and enum is in one package. A real change would split them
 // into temporal/api/stream/v1, temporal/api/enums/v1 and temporal/api/streamservice/v1.
 syntax = "proto3";

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

@@ -87,42 +87,31 @@ message FlushRunBufferResponse {
     int32 records = 1;
 }
 
-// Method `OpenWriter`, asynchronous. A buffered writer for an Activity or a client Producer.
+// Method `OpenWriter`, asynchronous. A writer for an Activity or a client Producer, which keeps
+// its epoch and `sequence` order. A writer of an Activity attempt above 1 first sends an empty
+// batch, so the new epoch withdraws the earlier attempt's Records.
 message OpenWriterRequest {
     string namespace = 1;
     temporal.api.streamservice.v1.StreamRef stream = 2;
     temporal.api.streamservice.v1.Producer producer = 3;
-    // How often the writer flushes. Empty is 100 ms.
-    google.protobuf.Duration batch_interval = 4;
-    // A flush sends at most this many Records in one batch. Zero is 1000.
+    // The interval flush and its buffer limit are gone. Lang buffers on its own.
+    reserved 4, 7;
+    reserved "batch_interval", "max_buffered_records";
+    // A write sends at most this many Records in one batch. Zero is 1000.
     int32 max_batch_size = 5;
-    // How long a flush retries a Store failure. Empty is 10 s.
+    // How long a write retries a Store failure. Empty is 10 s.
     google.protobuf.Duration max_retry_duration = 6;
-    // A publish past this many buffered Records fails with `stream_buffer_full`. Zero is 10000.
-    int32 max_buffered_records = 7;
 }
 
 message OpenWriterResponse {
     uint64 writer_id = 1;
 }
 
-// Method `Publish`, synchronous. Adds encoded Records to a writer's buffer.
-message PublishRequest {
-    uint64 writer_id = 1;
-    repeated temporal.api.streamservice.v1.AppendRecord records = 2;
-    // Start a flush at once.
-    bool force_flush = 3;
-}
-
-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.
+// Method `WriteRecords`, asynchronous. Appends Records at once for an open writer 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. A
+// failure after the Store retries fails the writer, since the batch may be in the Store.
 message WriteRecordsRequest {
     uint64 writer_id = 1;
     repeated temporal.api.streamservice.v1.AppendRecord records = 2;
@@ -134,33 +123,24 @@ message WriteRecordsResponse {
     int64 last_offset = 2;
 }
 
-// Method `FlushWriter`, asynchronous. Waits until the Records published so far are in the Store.
-message FlushWriterRequest {
-    uint64 writer_id = 1;
-}
-
-message FlushWriterResponse {
-    // The offset of the newest Record the writer wrote. Zero if it wrote none.
-    int64 last_offset = 1;
-}
-
-// Method `CloseWriter`, asynchronous. Flushes, then drops the writer. It doesn't close the
-// Stream.
+// Method `CloseWriter`, asynchronous. Waits for a write in flight, then drops the writer. It
+// doesn't close the Stream. It fails with the writer's error if a write failed.
 message CloseWriterRequest {
     uint64 writer_id = 1;
 }
 
 message CloseWriterResponse {
+    // The offset of the newest Record the writer wrote. Zero if it wrote none.
     int64 last_offset = 1;
 }
 
-// Method `CloseWriters`, asynchronous. Flushes and drops every open writer, for example when the
-// plugin's client or Worker shuts down, so no flusher task stays behind.
+// Method `CloseWriters`, asynchronous. Drops every open writer, for example when the plugin's
+// client or Worker shuts down.
 message CloseWritersRequest {
 }
 
 message CloseWritersResponse {
-    // The writers whose last flush failed.
+    // The writers that had failed.
     int32 failed = 1;
 }
 

scripts/_wit/stream-service.wit

@@ -1,8 +1,8 @@
-// Draft. The Stream service as a System Nexus service, for the Brook v2 API proposal.
-// The records are backed by the messages in stream_service.proto (same directory), which
-// is a draft of temporal/api/streamservice/v1/. For the draft, every message and enum
-// is in one package. The spec carries no per-language names. Each language gets its
-// idiomatic shape from a hand-written facade over the generated code.
+// Draft. The Stream service as a System Nexus service, for the Brook API proposal. The records
+// are backed by the messages in ../protos/temporal/api/streamservice/v1/stream_service.proto.
+// This file is the source. Each SDK copies it and generates its bindings with nexgen. The spec
+// carries no per-language names. Each language gets its idiomatic shape from a hand-written
+// facade over the generated code.
 package temporal:nexus@1.0.0;
 
 world system {
@@ -33,11 +33,11 @@ interface stream-service {
   record stream-ref {
     /// @nexus.doc "The Stream id. 1 to 4096 bytes after encoding."
     id: string,
-    /// @nexus.doc "The Workflow that owns the Stream. Empty on a standalone, client or Activity ref."
+    /// @nexus.doc "The owning Workflow. Empty on a standalone, client or Activity ref."
     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."
+    /// @nexus.doc "Registered backend name, empty for the default. The Store never reads it."
     backend: option<string>,
   }
 
@@ -123,9 +123,9 @@ interface stream-service {
     previous-completed-task: option<workflow-task-epoch>,
     /// @nexus.doc "Workflow only. Started event ids of failed or timed out Workflow Tasks."
     failed-tasks: list<s64>,
-    /// @nexus.doc "Workflow retry only. The previous run. Every Record whose Producer carries it is withdrawn."
+    /// @nexus.doc "Workflow retry only. The previous run, all of whose Records are withdrawn."
     withdraw-run: option<string>,
-    /// @nexus.doc "The run of the withdrawing Producer, so a Reader applies the Workflow rule without parsing ids. Empty for a client."
+    /// @nexus.doc "The withdrawing Producer's run, so no Reader parses ids. Empty for a client."
     run-id: option<string>,
   }
 
@@ -151,16 +151,16 @@ interface stream-service {
     close-failure: option<failure>,
   }
 
-  /// @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.doc "Create a Stream idempotently. ALREADY_EXISTS for another Retention or 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."
     stream-id: string,
     /// @nexus.doc "The age after which a Record is dropped. Empty is the Store default."
     retention: option<duration>,
-    /// @nexus.doc "Workflow only. A retry run's first batch or create: the previous run, whose Records it withdraws."
+    /// @nexus.doc "Retry run only. The previous run, whose Records this 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."
+    /// @nexus.doc "Close with no result after this long with no append. Empty never auto-closes."
     auto-close: option<duration>,
     /// @nexus.omit
     namespace: placeholder,
@@ -185,7 +185,8 @@ interface stream-service {
 
   // The meta key of each Stream has a format version v. Lua refuses a write to a Stream with a
   // higher v with FAILED_PRECONDITION and the detail stream_format_too_new.
-  /// @nexus.doc "Append one batch. FAILED_PRECONDITION if closed, fenced or stream_format_too_new. NOT_FOUND if absent."
+  /// @nexus.doc
+  ///   "Append. FAILED_PRECONDITION if closed, fenced or format too new. NOT_FOUND if absent."
   /// @nexus.proto "temporal.api.streamservice.v1.AppendRecordsRequest"
   record append-records-request {
     %stream: option<stream-ref>,
@@ -198,7 +199,7 @@ interface stream-service {
     previous-completed-task: option<workflow-task-epoch>,
     /// @nexus.doc "Workflow only. Failed or timed out tasks since this run's last flush here."
     failed-tasks: list<s64>,
-    /// @nexus.doc "Workflow only. A retry run's first batch or create: the previous run, whose Records it withdraws."
+    /// @nexus.doc "Retry run only. The previous run, whose Records this withdraws."
     withdraw-run: option<string>,
     /// @nexus.omit
     namespace: placeholder,
@@ -242,7 +243,7 @@ interface stream-service {
   /// @nexus.proto "temporal.api.streamservice.v1.ReadRecordsRequest"
   record read-records-request {
     %stream: option<stream-ref>,
-    /// @nexus.doc "Start at this offset. Empty uses start. INVALID_ARGUMENT if past the newest offset + 1."
+    /// @nexus.doc "Start offset. Empty uses start. INVALID_ARGUMENT past the newest offset + 1."
     /// @nexus.offset "stream-offset"
     from-offset: option<s64>,
     /// @nexus.doc "Where to start when from-offset is empty. Unspecified is the beginning."
@@ -292,7 +293,8 @@ interface stream-service {
   /// @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 the run's buffer, and the Workflow facade drops the response."
+  /// @nexus.doc
+  ///   "In Workflow code, Brook buffers this call in the run, and the facade drops the answer."
   /// @nexus.operation name="AppendRecords"
   append-records: func(request: append-records-request) -> append-records-response;
 

temporalio/api/streamservice/v1/stream_service_pb2.pyi

@@ -1,7 +1,7 @@
 """
 @generated by mypy-protobuf.  Do not edit manually!
 isort:skip_file
-Draft for the Brook v2 API proposal. It backs stream-service.wit in the same directory.
+Draft for the Brook API proposal. It backs crates/streams/nexus/stream-service.wit.
 For the draft, every message and enum is in one package. A real change would split them
 into temporal/api/stream/v1, temporal/api/enums/v1 and temporal/api/streamservice/v1.
 """

temporalio/api/streamservice/v1/stream_service_pb2_grpc.pyi

@@ -1,7 +1,7 @@
 """
 @generated by mypy-protobuf.  Do not edit manually!
 isort:skip_file
-Draft for the Brook v2 API proposal. It backs stream-service.wit in the same directory.
+Draft for the Brook API proposal. It backs crates/streams/nexus/stream-service.wit.
 For the draft, every message and enum is in one package. A real change would split them
 into temporal/api/stream/v1, temporal/api/enums/v1 and temporal/api/streamservice/v1.
 """

temporalio/bridge/Cargo.toml

@@ -94,7 +94,7 @@ url = "2.2"
 # crates, so the bridge sees one type of each.
 [dependencies.temporalio-streams]
 git = "https://github.com/moetemp/sdk-streams"
-rev = "6254405544278746d98ac4bbb1f748ce174d1dc2"
+rev = "daeb5fc2a13e1f3457a461f109e7f5b8eb31951c"
 default-features = false
 optional = true
 

temporalio/contrib/streams/__init__.py

@@ -35,7 +35,6 @@
 )
 from temporalio.contrib.streams._errors import (
     ReadOnlyContextError,
-    StreamBufferFullError,
     StreamClosedError,
     StreamError,
     StreamExpiredError,
@@ -65,7 +64,6 @@
     "ReadOnlyContextError",
     "Record",
     "StreamBackend",
-    "StreamBufferFullError",
     "StreamClient",
     "StreamClosedError",
     "StreamDefinition",

temporalio/contrib/streams/_errors.py

@@ -73,10 +73,6 @@ class StreamOffsetError(StreamError):
     """The offset is past the newest one plus one, or Retention removed it."""
 
 
-class StreamBufferFullError(StreamError):
-    """The writer's buffer is full. Flush, or publish less often."""
-
-
 class StreamUnsupportedError(StreamError):
     """The Store lacks a capability that the call needs. The message ends with its name."""
 
@@ -96,7 +92,6 @@ class ReadOnlyContextError(StreamError):
     "stream_fenced": StreamFencedError,
     "stream_storage": StreamStorageError,
     "stream_offset": StreamOffsetError,
-    "stream_buffer_full": StreamBufferFullError,
     "stream_unsupported": StreamUnsupportedError,
 }
 

temporalio/contrib/streams/proto/v1/__init__.py

@@ -9,12 +9,8 @@
     DrainClosesResponse,
     FlushRunBufferRequest,
     FlushRunBufferResponse,
-    FlushWriterRequest,
-    FlushWriterResponse,
     OpenWriterRequest,
     OpenWriterResponse,
-    PublishRequest,
-    PublishResponse,
     RedisStreamConfig,
     StreamErrorDetails,
     WorkflowCall,
@@ -33,12 +29,8 @@
     "DrainClosesResponse",
     "FlushRunBufferRequest",
     "FlushRunBufferResponse",
-    "FlushWriterRequest",
-    "FlushWriterResponse",
     "OpenWriterRequest",
     "OpenWriterResponse",
-    "PublishRequest",
-    "PublishResponse",
     "RedisStreamConfig",
     "StreamErrorDetails",
     "WorkflowCall",