Repository navigation
Conversation
The Streams API moves into temporalio.common.streams, so the module becomes a package with the same contents.
The first-class API lives next to the converter it registers with. The paths of the generators and the generated imports follow.
A Stream's lifecycle is its owner's to manage, so a run end closes nothing. drain() still lets unawaited writes finish before the run ends.
Only the run-end close on Redis used them, along with workflow.Info.execution_expiration_time.
Backends go in stream_backends= of Client.connect, a Worker takes its client's, and the Replayer and WorkflowEnvironment take them too. temporalio.workflow, Client.stream and WorkflowHandle.stream replace StreamClient, and Redis moves to temporalio.contrib.stream.
A read is stream-level only, so Topic objects lose subscribe and pass to the Stream's subscribe as topics. The proto field names stay.
The crate reserves the auto_close fields, so the pb2 and the system bindings lose them. The generators write into the new package.
s23 drops auto_close from the Stores and the protos, and c8 drops the cache-off eviction reason that only the run-end close used.
…iewed-api The Brook layout, API and removals win. The native backend, the Workflow reads and the Standalone Activity create move into temporalio.common.streams, and the crate and sdk-core pins stay at j8's until they move to k3 and c10.
Every client, Worker and Replayer has it, and a registered NativeStreamBackend only configures it. A client builds its native runtime at its first Stream call, and a Worker loads Streams only when it has Workflows or Activities.
The server closes a native Stream when its run chain ends, and Redis ignores the field. The owner check of an empty page is now the backend's, so the close-with-run flag goes.
A Standalone Activity owns its Stream on the native backend, which closes it when the Activity ends. A client binds it from the Activity handle with no call.
k3 merges s23 into the native crate, and c10 merges c8 into the native sdk-rust. The vendored protos and WIT already equal k3's, and the bindings regenerate with no change.
A replay reads each page of a native Workflow read again, and only the caller knows which server keeps the Streams. The Replayer takes the client's backends when it has none.
The native lane registers nothing, and the examples follow the Jetty HLD. Examples that need the native server run with Redis or a memory Store named redis in place of Redis.
A Worker with Workflows runs them, and they enable no polling of the task queue's Activities and stay out of the errors about the user's Activities. So a Worker that uses no Streams polls and fails as before.
read_spec takes any Topic object, so the base is no longer used in its own module, as SyncTopicBase isn't. basedpyright and mypy are clean.
A Local Activity close outlived a Workflow Task that failed after it, so the retry's create failed the run. The block end now leaves the Stream open in a cron run and on a failure that a retry run follows.
A ClientStream kept a failed or cancelled Describe forever, and the handle map kept a failed one with its handle. A found id now stays as text, and a failed call goes.
A Replayer serves one backend as a Worker does, a Worker configures its backends after its plugins, and a generator of backends stays. A backend in plugins= fails before the connect, and the per-client calls are private.
A Worker serves one backend, so the backend-move recipe can't run. A continued run reads its Streams' backends after @workflow.init.
Redis keeps its regular Activity close, and a backend that closes another way overrides _close_in_workflow. Workers that share a task queue all register the Redis backend, since it polls Activities there.
The server carries out the close once the Workflow Task commits, so a task that fails after the close leaves the Stream open for its retry. drain() awaits it as it awaits the writes.
…ure. A cron chain writes on to the Stream in its next run, and so does a retry run after a failure under a retry policy. An explicit close() still closes.
…iewed-api p35's close hook and registration fixes win, and j9 keeps a Worker's many backends. The native backend overrides the close hook with its System Nexus command, and only a Worker with a Redis-kind backend registers the regular close Activity.
A Local Activity in the task of the close runs, the task fails, and the Stream stays open until the retry closes it once, through the Activity on Redis and the command on native.
On native the close is a CloseStream operation in History, not an Activity, and the server closes the Stream at the terminate. p35's block tests cover the cron and retry cases.
The server drops the cancel of a closed run, so a wake that the run's last task cancelled kept its parked read and a slot of stream.maxParkedReads.
An entry left only on success, a rerun under a new epoch added another, and the count bound let up to 2 GiB stay.
The close of a Workflow's Stream is a regular Activity, so such a Worker alone left every close waiting. Another Worker on the queue may run it, so it is a warning.
Wakes and the native close share the run's 30 pending Nexus operations. Past the limit the server failed every task of the run, so a close never went out.
The create took the expired status as a success, and the error came only at the first publish.
A plain TypeError ended the read with next_offset before the Record, so every resume failed on it again.
The close at the block end always raised StreamNotOwnerError and hid the block's own error.
An empty id failed only at the first call, with a bare AssertionError.
Nothing withdraws a client's Records, so a long UI read grew its withdrawal map with no end.
subscribe takes offset: int | End | None, and nothing in Brook or Jetty calls from_rpc_error.
The key is in History, so a run that never creates, binds or receives a Stream adds nothing there. A Local Activity keeps its key always, since its headers stay out of History.
Retention is a Record's age and trims open Streams, a trimmed offset raises StreamExpiredError, and aclose from another task raises.
The server keeps a Workflow reader's range of a client's Stream and refuses an append that the size cap would make drop it, so a reader no longer stops at half the Retention. StreamTooLargeError names that refusal.
The Brook crate and the Jetty server both answer a Record or a batch over the Store's limit with it, so both stacks raise one class.
The vendored extension.proto and stream-service.wit changed in comments only, so only their docstrings in the generated code moved.
The crate refuses a Local Activity's close of a Workflow's Stream, so a close by id from one is refused too. The vendored proto and WIT differ in comments only.
The code is in the private temporalio.common._streams, and temporalio.common loads its names on first use, since the Stream modules import temporalio.client and temporalio.workflow.
The model's converter keeps payloads raw for the Topic types, so the failure kept RawValue details, as p35 fixed for client reads.
Only that Activity closes it. As with a Workflow's Stream, the block end of async with would raise and hide the block's own error.
The native server finds a Workflow's Stream only by its owner fields, so the by-id case p35 added answers stream_not_found there.
The write takes its place in the Stream's order at the call, also with no await. An error that no caller gets raises at the run's end, as with publish_nowait.
1 of 3 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR merges Brook p35 into Jetty, makes native the default Stream backend, and fixes the Jetty findings of an independent review.
Jetty stores Streams in the Temporal server (t15). The SDK review made native the default with no registration. This PR also ports Jetty's Workflow reads and Standalone Activity Streams to the reviewed API.
What changed?
From p35. The merge adds the reviewed API and its review fixes.
WorkflowTopic.publishstarts the write when called and returns itsasyncio.Future[int]. A failed write that nobody awaits or checks fails the run at the run end. The Streams API is intemporalio.common, with the code in the privatetemporalio.common._streams.NativeStreamBackendandWorkflowItemsare intemporalio.commontoo.Native default.
NativeStreamBackendconfigures it, and a non-native backend namednativeis refused. The default isbackend=, else the one backend withdefault=True, else native.CloseStreamcommand, which commits with its Workflow Task. A Workflow create sendsclose_with_run = true, and the server closes the Stream at the Run Chain end.Jetty APIs.
Client.create_stream(definition, *, id, retention=None, backend=None)creates a client-owned Stream on the default backend. The block end ofasync withcloses it.WorkflowStream.subscribetakesoffsetandtopic_prefixesand returnsWorkflowItems.WorkflowTopic.subscribeis gone.temporalio.activity.create_streamandActivityHandle.streamserve Standalone Activity Streams on native.Replayer(stream_client=...)reads native pages again on the connection of that client.Review fixes.
activity.create_streamleaves the Stream open. The server closes it at the Activity end with the final result, and a retry can create it again. A Workflow can read a Standalone Activity's Stream.async withon a client handle of an Activity's Stream raises before the block runs.WAIT_REQUESTED, and the run end waits until the server has each cancel. When a run has 20 wakes, a read waits on a timer, as on Redis. So wakes and the native close stay under the per-run limit of pending Nexus operations.StreamConfigError, not the run. A digest mismatch of the codec step on replay is reported as nondeterminism. A Workflow read'sCloseditem has its status, and its failure is decoded with the user's converter.__temporal_stream_describeLocal Activity is gone. An append that would drop a reader's range raisesStreamTooLargeError.Pins and CI.
abd8955band sdk-core c102de24aef.changelog/added/jaunty-jetty.mdfor the native backend. With native the default, every Worker runs the Worker hook of the native backend. So the upstream interpreter-pool test on 3.14 needs p35'sinterceptorsfix.Known limits.
NondeterminismErrorinworkflow_failure_exception_types, a replay digest mismatch doesn't fail the run. The task keeps the plain failure cause. A Core change would fix this.Closedfailure.Why?
The independent review of Brook and Jetty found that an Activity's block end broke its retries, and that dead wakes filled the server's parked read slots. It also found that a disabled server failed the run, not the task, and that a Workflow reader of a client Stream could lose its range.
How did you test it?
Each review fix has a test that fails without it, apart from the docstring fixes. At the head,
pytest tests/contrib/streamspasses on the memory, Redis and native lanes, with none skipped. The native lane ran against a server built from t159f912df3. With no native target, the native tests skip and the others pass. The worker, client, plugin, testing, Activity, Workflow, Nexus and converter suites pass.poe lintandpoe gen-docsare clean.On the fork CI, every job passes except
check-protosand the arm64 jobs.check-protoswaits on a nexgen release with temporalio/nexgen#260, which merged after v0.2.10. The arm64 jobs have no runner on the fork.The superproject branch
temporalio/internal-ai-prototypes2026/10/ai198-streamspins the head,37a0a22c.