feat: add the Stream protos and WIT to the vendored API by moetemp · Pull Request #1733 · temporalio/sdk-rust

moetemp · 2026-10-30T21:25:41Z

What was changed

This PR adds the Stream protos and the Stream WIT to Core's vendored API, and compiles the protos in temporalio-protos. Core itself gets no Stream code.

The three Stream protos and nexus/stream-service.wit of feat(brook): add the Stream service messages (as experimental) api#895 sit in crates/protos/protos/api_upstream, as the subtree will have them once that PR merges. A subtree update then replaces them.
crates/protos/build.rs compiles the protos. temporalio-protos exposes them as temporal::api::stream::v1 and temporal::api::streamservice::v1, and the Stream enums are in temporal::api::enums::v1.

Why?

The Stream protos and the WIT are the contract with every SDK, so they come first. A lang SDK generates its Streams API from the WIT, and sdk-python reads these files through its Core submodule. The Stream crate that the bridges link is #1799, which sits on this branch.

Checklist

How was this tested:
- cargo check -p temporalio-protos on this head, which builds the new modules.

Sushisource

Overall I think it would be really helpful to at least include the top-level Core APIs in this PR to see how these messages get used (even if they're stubbed out for now). Then we could merge that into a branch. I don't think we should merge to main until there's some kind of usable implementation.

Sushisource · 2026-11-04T19:28:00Z

+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 `FreeWriter`. Releases one handle of the writer with no wait, for a handle that lang
+// drops, such as from a finalizer. The bridge may also call it with no async runtime. A write in
+// flight still completes. The writer's error, if any, is lost.
+message FreeWriterRequest {
+    uint64 writer_id = 1;
+}


I'm not sure these really need to be defined in proto at all since the types are so simple. We often don't define everything as protos, esp. simple stuff.

And free probably doesn't even need to be a dedicated top level method. It can be implemented in terms of Droping refs that the lang bridge holds.

I kept them as protos so every bridge can use one generic call. A bridge passes a method name and the request bytes to the crate's StreamRuntime::call, and gets the response bytes back. So the Python, TypeScript and .NET bridges have no Stream code, and a new method needs no bridge change. That only works if every request and response is a proto, even one with one field, like CloseWriterRequest.

FreeWriter comes from the same choice. Lang holds only the writer_id number, not a Rust object. When Python's garbage collector drops a writer, there's no Rust value to Drop, so a weakref.finalize callback sends FreeWriter with the id.

The other way is a writer object in each bridge, which Drop releases. Then we can remove FreeWriter, but each of the three bridges needs its own writer binding. Which one do you prefer?

Yeah, interesting. I did consider doing similar stuff for other methods. You incur some extra overhead doing it this way but it's probably immaterial.

The more important thing would be the Drop semantics I think. Using Drop is pretty foolproof whereas needing to explicitly call finalize is much easier to forget.

I don't have a strong preference for now. This is all internal so we can change it later. I'm fine doing the runtime dispatch to keep bridge changes minimal for now

Thanks. I'll keep the runtime dispatch and FreeWriter for now, so the bridges stay thin. We can move to Drop-based writer handles later if the finalize call proves easy to forget.

Each message and field says what it is for. A create names its Stream only by its handle, the protos keep no reserved fields, and SEQUENCE and OFFSET are SEQUENCE_TOO_LOW and OFFSET_TOO_HIGH.

Each kind has its own epoch fields, and the case picks the withdrawal rule, so StreamProducerKind is gone. The protos match temporalio/sdk-rust#1733.

The protos match temporalio/sdk-rust#1733 again.

A corrupt Store isn't retryable, unlike an unavailable one. A close carries no value, so a read gives no close item and the close takes no result or failure. The protos match temporalio/sdk-rust#1733.

A ref is the data that code passes, and the stream resource, the object with calls, is the handle, as the API review decided. The protos match temporalio/sdk-rust#1733.

moetemp

I agree it should not land on main before there is a usable implementation. I'll keep this PR open, as use it as the base branch. I'll stack the runtime object and its calls on this one, so you can see how lang uses each message. This push also takes the changes of the Nov. 4 API review. A create fails if the Stream exists unless it asks for USE_EXISTING, a request id keeps a retried create safe, and the SDKs create only standalone Streams.

moetemp · 2026-11-05T05:41:16Z

+
+// Experimental. Might change in the future.
+// Why a Stream call failed. Each reason has one gRPC status code.
+enum StreamErrorReason {


Code acts on fewer of them. The SDKs give an error class only to the reasons that code can act on, like StreamClosedError to stop writing, StreamFencedError for an attempt that a retry replaced, or StreamStorageUnavailableError as the one to retry. The others, such as SEQUENCE_TOO_LOW, FORMAT_TOO_NEW, INVALID, UNKNOWN_WRITER and the new STORAGE_CORRUPT, raise the base class and are mostly for logs. The enum doc says that now (9adfe1e). We could still fold SEQUENCE_TOO_LOW into SEQ_DIVERGENT and UNKNOWN_WRITER into INVALID, since nobody handles them apart. Do you think it's worth it?

moetemp · 2026-11-05T05:41:20Z

+    reserved "auto_close";
+    string namespace = 1;
+    // Required, and never empty.
+    string stream_id = 2;


Thanks. It was left over from before the ref carried the backend. The create now gives its Stream only as stream_ref, in 9adfe1e. Also, 259bf7d renames StreamHandle to StreamRef, since after the API review, the data is called a ref and the object with calls is called a handle. owner isn't quite the same as the ref's owner, though. The ref's owner is the run chain, and owner is the Producer of the creating run, with its epoch. Neither is set for now, since we create only standalone Streams for now.

moetemp · 2026-11-05T17:42:20Z

+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 `FreeWriter`. Releases one handle of the writer with no wait, for a handle that lang
+// drops, such as from a finalizer. The bridge may also call it with no async runtime. A write in
+// flight still completes. The writer's error, if any, is lost.
+message FreeWriterRequest {
+    uint64 writer_id = 1;
+}


I kept them as protos so every bridge can use one generic call. A bridge passes a method name and the request bytes to the crate's StreamRuntime::call, and gets the response bytes back. So the Python, TypeScript and .NET bridges have no Stream code, and a new method needs no bridge change. That only works if every request and response is a proto, even one with one field, like CloseWriterRequest.

FreeWriter comes from the same choice. Lang holds only the writer_id number, not a Rust object. When Python's garbage collector drops a writer, there's no Rust value to Drop, so a weakref.finalize callback sends FreeWriter with the id.

The other way is a writer object in each bridge, which Drop releases. Then we can remove FreeWriter, but each of the three bridges needs its own writer binding. Which one do you prefer?

Sushisource

Looking much cleaner. A few more things.

Sushisource · 2026-11-06T21:17:30Z

+// The ordered attempt of a Producer. A higher epoch supersedes the Records of a lower epoch of the
+// same Producer. The case is the kind of the Producer. It decides how the Store orders the epochs,
+// and which earlier Records a higher epoch withdraws.
+message ProducerEpoch {


Given that this is only used inside of Producer, this could just get inlined into there and it would resolve the fact that run_id doesn't exist for clients because it can move inside the oneof.

Done in 4584348. ProducerEpoch is gone, and Producer has the kind oneof directly.

Sushisource · 2026-11-06T21:18:56Z

+    // Workflow and Local Activity only. The run's last completed Workflow Task, from a Local
+    // Activity's start. Its started_event_id is 0 before the first. The write withdraws the run's
+    // Local Activity Records whose Workflow Task never completed.
+    WorkflowTaskEpoch previous_completed_task = 2;
+    // Workflow and Local Activity only. The started event ids of the run's failed or timed out
+    // Workflow Tasks. The write withdraws the run's Local Activity Records of these tasks.
+    repeated int64 failed_tasks = 3;
+    // Retry run only. The previous run, on the retry run's first write. Every Record whose
+    // Producer carries it is withdrawn.
+    string withdraw_run = 4;


All these "only" statements are another indication we could be putting things inside of variants. IMO there's very, very rarely a good reason to allow the representation of invalid states like that in proto.

Same change in 4584348. The producer-kind fields moved into the Producer oneof above. For the writer, the task facts and withdraw_run move into a RunWithdrawal message that only a Workflow or Local Activity write sets. I kept them in one message and not a oneof, since a retry run's first Local Activity write has both the task facts and withdraw_run. If you'd rather forbid the client case in the type, I can split StreamWriter by kind too, but that repeats the Producer oneof, so I left it for now.

Update: I removed the withdrawal machinery. RunWithdrawal, the writer's withdrawal fields, StreamSuperseded, and the superseded read item are gone, and a write only has its Producer. Supersession moved to an opt-in client helper, with_superseded(), computed from the Producer epoch, so a Store keeps no supersession facts.

moetemp

@Sushisource Thanks for the review.

moetemp · 2026-11-07T00:16:39Z

+// The ordered attempt of a Producer. A higher epoch supersedes the Records of a lower epoch of the
+// same Producer. The case is the kind of the Producer. It decides how the Store orders the epochs,
+// and which earlier Records a higher epoch withdraws.
+message ProducerEpoch {


Done in 4584348. ProducerEpoch is gone, and Producer has the kind oneof directly.

moetemp · 2026-11-07T00:16:40Z

+    // Workflow and Local Activity only. The run's last completed Workflow Task, from a Local
+    // Activity's start. Its started_event_id is 0 before the first. The write withdraws the run's
+    // Local Activity Records whose Workflow Task never completed.
+    WorkflowTaskEpoch previous_completed_task = 2;
+    // Workflow and Local Activity only. The started event ids of the run's failed or timed out
+    // Workflow Tasks. The write withdraws the run's Local Activity Records of these tasks.
+    repeated int64 failed_tasks = 3;
+    // Retry run only. The previous run, on the retry run's first write. Every Record whose
+    // Producer carries it is withdrawn.
+    string withdraw_run = 4;


Same change in 4584348. The producer-kind fields moved into the Producer oneof above. For the writer, the task facts and withdraw_run move into a RunWithdrawal message that only a Workflow or Local Activity write sets. I kept them in one message and not a oneof, since a retry run's first Local Activity write has both the task facts and withdraw_run. If you'd rather forbid the client case in the type, I can split StreamWriter by kind too, but that repeats the Producer oneof, so I left it for now.

moetemp · 2026-11-07T00:16:42Z

+
+// Experimental. Might change in the future.
+// Why a Stream call failed. Each reason has one gRPC status code.
+enum StreamErrorReason {


I kept the reasons distinct on the wire, since each one has its own gRPC status code that native Streams exposes. Bundling them would drop that code. As we discussed in the API review session, let's collapse the SDK error classes to the few that code acts on, and kept the proto reason so the rest can still be told apart in a log. Wdyt?

moetemp · 2026-11-07T00:16:44Z

+    reserved "auto_close";
+    string namespace = 1;
+    // Required, and never empty.
+    string stream_id = 2;


I dropped StreamRef.owner and the create/close owner fields. The protos are updated to match (api#895 and the vendored copy).

moetemp · 2026-11-07T00:16:45Z

+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 `FreeWriter`. Releases one handle of the writer with no wait, for a handle that lang
+// drops, such as from a finalizer. The bridge may also call it with no async runtime. A write in
+// flight still completes. The writer's error, if any, is lost.
+message FreeWriterRequest {
+    uint64 writer_id = 1;
+}


Thanks. I'll keep the runtime dispatch and FreeWriter for now, so the bridges stay thin. We can move to Drop-based writer handles later if the finalize call proves easy to forget.

Sushisource · 2026-11-10T23:36:58Z

+    reserved 2, 3;
+    reserved "result", "failure";


reserved stuff still showing up

They're gone now.

Sushisource · 2026-11-10T23:37:37Z

+
+// Experimental. Might change in the future.
+// Why a Stream call failed. Each reason has one gRPC status code.
+enum StreamErrorReason {


That's fine but a lot of these do map to the same gRPC code, no? Those could potentially be combined too but it's not a big deal either way

These are the files of the Stream API that temporalio/api#895 adds, byte for byte. An update of the API subtree replaces them once that PR merges.

The Stream crate and the SDK bridges take the Stream messages from the protos crate, like every other API message.

The Stream WIT of temporalio/api#895 has append and close on the Workflow target again, and this is its copy.

The file is a copy of the Stream WIT of the API repo, which got three methods that the SDK implements.

moetemp requested a review from a team as a code owner October 30, 2026 21:25

moetemp mentioned this pull request Oct 30, 2026

[Brook K1] Added the temporalio-streams crate with the Stream protos. moetemp/sdk-rust#212

Closed

moetemp added the skip-changelog PR changes do not require changelog updates label Oct 30, 2026

moetemp changed the title ~~feat(brook): add the temporalio-streams crate with the Stream protos~~ feat: add the temporalio-streams crate with the Stream protos Nov 4, 2026

Sushisource reviewed Nov 4, 2026

View reviewed changes

moetemp added a commit to temporalio/api that referenced this pull request Nov 5, 2026

Stopped using name as a verb in the Stream comments.

cabddc7

The protos match temporalio/sdk-rust#1733 again.

This was referenced Nov 5, 2026

[Brook B1] Added the streams bridge feature that links the temporalio-streams crate. moetemp/sdk-python#370

Closed

[Brook S1] Added the Redis Store's config proto and its generated code. moetemp/sdk-python#371

Closed

moetemp commented Nov 5, 2026

View reviewed changes

moetemp requested a review from Sushisource November 5, 2026 17:45

moetemp mentioned this pull request Nov 5, 2026

feat: add the Stream runtime API with stubbed calls #1762

Closed

Sushisource reviewed Nov 6, 2026

View reviewed changes

moetemp commented Nov 8, 2026

View reviewed changes

moetemp requested a review from Sushisource November 8, 2026 20:52

moetemp mentioned this pull request Nov 10, 2026

Streams crate: combined branch for review moetemp/sdk-rust#228

Draft

Sushisource reviewed Nov 10, 2026

View reviewed changes

moetemp mentioned this pull request Nov 11, 2026

feat(brook): add the Stream service messages (as experimental) temporalio/api#895

Draft

moetemp changed the title ~~feat: add the temporalio-streams crate with the Stream protos~~ feat: add the Stream protos and WIT to the vendored API Nov 13, 2026

moetemp added 2 commits November 13, 2026 17:30

Vendored the Stream protos and WIT from the api repo.

6312e2d

These are the files of the Stream API that temporalio/api#895 adds, byte for byte. An update of the API subtree replaces them once that PR merges.

Compiled the Stream protos in the protos crate.

1607a9f

The Stream crate and the SDK bridges take the Stream messages from the protos crate, like every other API message.

moetemp force-pushed the moe/streams-crate-protos branch from bf47727 to 1607a9f Compare November 14, 2026 00:31

This was referenced Nov 14, 2026

feat: add temporalio-streams, the Stream runtime that SDK bridges link #1799

Draft

feat: add the Streams API, with Stream Stores on Redis and in memory temporalio/sdk-python#2129

Draft

Followed the API's Workflow Stream with append and close.

e623681

The Stream WIT of temporalio/api#895 has append and close on the Workflow target again, and this is its copy.

moetemp requested a review from Sushisource November 14, 2026 06:07

moetemp added this pull request to stack #1800 November 14, 2026 06:07

Followed the API's Stream WIT with take, filter and reduce.

9aac66e

The file is a copy of the Stream WIT of the API repo, which got three methods that the SDK implements.

Conversation

moetemp commented Oct 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What was changed

Why?

Checklist

Uh oh!

Sushisource left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

moetemp left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Sushisource left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

moetemp left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

moetemp commented Oct 30, 2026 •

edited

Loading