[Brook] Wrote Activity and client Records through the generated Writer.

pyproject.toml

@@ -188,20 +188,25 @@ exclude = [
   'temporalio/nexus/system/workflow_service',
   'temporalio/contrib/streams/proto',
   'temporalio/contrib/streams/system',
+  'temporalio/contrib/streams/writer_system',
 ]
 
 [[tool.mypy.overrides]]
 module = "temporalio.nexus.system.workflow_service.*"
 ignore_errors = true
 
 [[tool.mypy.overrides]]
-module = ["temporalio.contrib.streams.proto.*", "temporalio.contrib.streams.system.*"]
+module = [
+  "temporalio.contrib.streams.proto.*",
+  "temporalio.contrib.streams.system.*",
+  "temporalio.contrib.streams.writer_system.*",
+]
 ignore_errors = true
 
 [tool.pydocstyle]
 convention = "google"
 # https://github.com/PyCQA/pydocstyle/issues/363#issuecomment-625563088
-match_dir = "^(?!(docs|scripts|tests|api|proto|system|_support|notifications|\\.)).*"
+match_dir = "^(?!(docs|scripts|tests|api|proto|system|writer_system|_support|notifications|\\.)).*"
 add_ignore = [
   # We like to wrap at a certain number of chars, even long summary sentences.
   # https://github.com/PyCQA/pydocstyle/issues/184

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

@@ -5,6 +5,13 @@ syntax = "proto3";
 
 package temporal.api.streamservice.v1;
 
+option go_package = "go.temporal.io/api/streamservice/v1;streamservice";
+option java_package = "io.temporal.api.streamservice.v1";
+option java_multiple_files = true;
+option java_outer_classname = "StreamServiceProto";
+option ruby_package = "Temporalio::Api::StreamService::V1";
+option csharp_namespace = "Temporalio.Api.StreamService.V1";
+
 import "google/protobuf/duration.proto";
 import "google/protobuf/timestamp.proto";
 import "temporal/api/common/v1/message.proto";

scripts/_wit/stream-writer.wit

@@ -0,0 +1,98 @@
+// Internal. Not a Temporal API, and no server serves it. The crate's Worker extension serves
+// these calls in process, backed by the messages in ../protos/temporal/sdk/streams/v1/
+// extension.proto. Lang reaches them through an extension-call transport, from Activities and
+// clients only. So the interface is client-only and has no endpoint, which keeps it out of the
+// System Nexus path. The writer keeps the Producer epoch and the sequence in the crate. This file
+// is the source. Each SDK copies it beside stream-service.wit and generates its bindings with
+// nexgen.
+package temporal:streams@1.0.0;
+
+world writer {
+  export stream-writer;
+}
+
+/// @nexus.service-name "temporal.sdk.streams.v1.StreamWriter"
+/// @nexus.client-only
+/// @nexus.experimental
+interface stream-writer {
+  use nexus:temporal-types/model@1.0.0.{duration, placeholder};
+  use temporal:nexus/stream-service@1.0.0.{append-record, producer, stream-ref};
+
+  /// @nexus.doc "Open a writer for one Producer of one Stream."
+  /// @nexus.proto "temporal.sdk.streams.v1.OpenWriterRequest"
+  record open-writer-request {
+    %stream: option<stream-ref>,
+    producer: option<producer>,
+    /// @nexus.doc "A write sends at most this many Records in one batch. Zero is 1000."
+    max-batch-size: option<s32>,
+    /// @nexus.doc "How long a write retries a Store failure. Empty is 10 s."
+    max-retry-duration: option<duration>,
+    /// @nexus.omit
+    namespace: placeholder,
+  }
+
+  /// @nexus.proto "temporal.sdk.streams.v1.WriteRecordsRequest"
+  record write-records-request {
+    writer-id: option<u64>,
+    records: list<append-record>,
+  }
+
+  /// @nexus.proto "temporal.sdk.streams.v1.WriteRecordsResponse"
+  record write-records-response {
+    /// @nexus.doc "The offset of the first Record. Zero when the request has no Record."
+    first-offset: option<s64>,
+    /// @nexus.doc "The offset of the last Record. Zero when the request has no Record."
+    last-offset: option<s64>,
+  }
+
+  /// @nexus.proto "temporal.sdk.streams.v1.CloseWriterRequest"
+  record close-writer-request {
+    writer-id: option<u64>,
+  }
+
+  /// @nexus.proto "temporal.sdk.streams.v1.CloseWriterResponse"
+  record close-writer-response {
+    /// @nexus.doc "The offset of the newest Record the writer wrote. Zero if it wrote none."
+    last-offset: option<s64>,
+  }
+
+  /// @nexus.proto "temporal.sdk.streams.v1.CloseWritersRequest"
+  record close-writers-request {
+  }
+
+  /// @nexus.proto "temporal.sdk.streams.v1.CloseWritersResponse"
+  record close-writers-response {
+    /// @nexus.doc "The writers that had failed."
+    failed: option<s32>,
+  }
+
+  // A writer never changes. Offsets are values that its methods return.
+  /// @nexus.doc "An open writer of one Producer. The crate keeps its epoch and sequence."
+  resource writer {
+    constructor(writer-id: u64);
+
+    /// @nexus.doc "Write Records at once. The answer has the first and the last offset."
+    /// @nexus.operation "write-records"
+    write-records: func(records: list<append-record>) -> write-records-response;
+
+    /// @nexus.doc "Wait for a write in flight, then drop the writer. The Stream stays open."
+    /// @nexus.operation "close-writer"
+    close: func() -> close-writer-response;
+  }
+
+  /// @nexus.proto "temporal.sdk.streams.v1.OpenWriterResponse"
+  type writer-result = own<writer>;
+
+  /// @nexus.operation name="OpenWriter"
+  open-writer: func(request: open-writer-request) -> writer-result;
+
+  /// @nexus.operation name="WriteRecords"
+  write-records: func(request: write-records-request) -> write-records-response;
+
+  /// @nexus.operation name="CloseWriter"
+  close-writer: func(request: close-writer-request) -> close-writer-response;
+
+  /// @nexus.doc "Drop every open writer, for example at shutdown."
+  /// @nexus.operation name="CloseWriters"
+  close-writers: func(request: close-writers-request) -> close-writers-response;
+}

scripts/gen_stream_service.py

@@ -18,6 +18,7 @@
 
 base_dir = Path(__file__).parent.parent
 wit_path = base_dir / "scripts" / "_wit" / "stream-service.wit"
+writer_wit_path = base_dir / "scripts" / "_wit" / "stream-writer.wit"
 model_wit_path = (
     gen_protos.api_proto_dir / "nexus" / "deps" / "nexus-temporal-types" / "model.wit"
 )
@@ -30,10 +31,12 @@
     / "stream_service.proto"
 )
 output_dir = base_dir / "temporalio" / "contrib" / "streams" / "system"
+# The writer is client-only, so it has a caller and its resource class, and no Workflow code.
+writer_output_dir = base_dir / "temporalio" / "contrib" / "streams" / "writer_system"
 
 NEX_GEN_REPOSITORY = "https://github.com/moetemp/nexgen"
 # The head of moe/brook/v3/n8-stream-resource. A branch moves, so the pin is the commit.
-NEX_GEN_REVISION = "10fc35b40fcf48651c25e654a260bc71585f1429"
+NEX_GEN_REVISION = "204b566ee996939b165ccd1d18c8c432bfdcc0dd"
 nex_gen_root = base_dir / ".nexgen-streams" / NEX_GEN_REVISION
 
 
@@ -73,20 +76,44 @@ def api_protos() -> list[Path]:
 
 
 def generate_stream_service() -> None:
-    for path in (wit_path, model_wit_path, proto_path):
+    for path in (wit_path, writer_wit_path, model_wit_path, proto_path):
         if not path.exists():
             raise RuntimeError(f"missing generator input: {path}")
     with tempfile.TemporaryDirectory(dir=base_dir) as temp_dir:
         descriptor_path = Path(temp_dir) / "stream_service.bin"
+        # The extension messages live in temporalio.contrib.streams.proto, so the descriptor
+        # names their file by that module path.
+        extension_root = Path(temp_dir) / "extension"
+        extension_path = (
+            extension_root
+            / "temporalio"
+            / "contrib"
+            / "streams"
+            / "proto"
+            / "v1"
+            / "extension.proto"
+        )
+        extension_path.parent.mkdir(parents=True)
+        shutil.copy(
+            gen_protos.additional_proto_dir
+            / "temporal"
+            / "sdk"
+            / "streams"
+            / "v1"
+            / "extension.proto",
+            extension_path,
+        )
         subprocess.check_call(
             [
                 sys.executable,
                 "-mgrpc_tools.protoc",
                 f"--proto_path={gen_protos.api_proto_dir}",
                 f"--proto_path={gen_protos.additional_proto_dir}",
+                f"--proto_path={extension_root}",
                 "--include_imports",
                 f"--descriptor_set_out={descriptor_path}",
                 str(proto_path),
+                str(extension_path),
                 # The model WIT names API types, so their descriptors must be in the set.
                 *map(str, api_protos()),
             ]
@@ -109,19 +136,38 @@ def generate_stream_service() -> None:
                 str(output_dir),
             ]
         )
-    subprocess.check_call(
-        [
-            sys.executable,
-            "-m",
-            "ruff",
-            "check",
-            "--select",
-            "I",
-            "--fix",
-            str(output_dir),
-        ]
-    )
-    subprocess.check_call([sys.executable, "-m", "ruff", "format", str(output_dir)])
+        shutil.rmtree(writer_output_dir, ignore_errors=True)
+        subprocess.check_call(
+            [
+                *nex_gen_command(),
+                "python",
+                str(writer_wit_path),
+                str(wit_path),
+                str(model_wit_path),
+                "--native-api",
+                "--client",
+                "--support-package",
+                "temporalio.nexus.system._support",
+                "--descriptors",
+                str(descriptor_path),
+                "--output",
+                str(writer_output_dir),
+            ]
+        )
+    for directory in (output_dir, writer_output_dir):
+        subprocess.check_call(
+            [
+                sys.executable,
+                "-m",
+                "ruff",
+                "check",
+                "--select",
+                "I",
+                "--fix",
+                str(directory),
+            ]
+        )
+        subprocess.check_call([sys.executable, "-m", "ruff", "format", str(directory)])
 
 
 if __name__ == "__main__":