diff --git a/changelog/added/amentiferous-refugia.md b/changelog/added/amentiferous-refugia.md new file mode 100644 index 000000000..8ad223e26 --- /dev/null +++ b/changelog/added/amentiferous-refugia.md @@ -0,0 +1,2 @@ +Core: Experimental: the stream wakes the server puts on a Workflow Task's scheduled event reach the SDK as job 17, `WorkflowTaskAttachments`, one `Any(StreamWake)` per attachment. Replay delivers the same attachments in the same activations. A task scheduled with attachments ends a workflow task heartbeat chain, so it keeps an activation of its own on replay. A heartbeat fold that would reach the live task ends at the last completed task. +Core: Experimental: `Success.heartbeat_task` on a workflow activation completion asks Core to complete the workflow task now while local activities run, as at a local activity heartbeat. The task carries the commands and markers so far and asks for a new task. Core ignores the flag while replaying, since the SDK sets it from a count or a clock. diff --git a/changelog/added/drowsy-dormouse.md b/changelog/added/drowsy-dormouse.md new file mode 100644 index 000000000..753317b28 --- /dev/null +++ b/changelog/added/drowsy-dormouse.md @@ -0,0 +1,9 @@ +Experimental `temporalio.contrib.streams`: a Worker built from a client that carries a stream store waits, when it stops, for up to 10 seconds for the stream notifications the stream runtime still has out. `StreamStorePlugin.flush_notifications()` does the same for a process that holds only a client. +Experimental `temporalio.contrib.streams`: a store tells each stream's notifier on the server when the stream moves, after every append, and the append answers only after its notification went. It is on by default, and `StreamStorePlugin.notify_on_append(False)` turns it off. `StreamHandle.close()` and `StreamStorePlugin.close_stream()` close a stream with a result or an `error=`, which reaches every reader that sleeps on the stream. +Experimental `temporalio.contrib.streams`: an idle `StreamHandle.read()` sleeps on the stream's notifier on the server instead of holding a store read open. After `idle_wait` (2 seconds by default) with no record it waits for a notification or the close, and reads again at the server's wait limit. On a server without the notifier it waits in the store as before. +Experimental `temporalio.contrib.streams`: a `StreamHandle.read()` of a topic closed with `error=` raises that error after the last record. +Experimental `temporalio.contrib.streams`: `workflow_reader()` reads a stream in Workflow code, one Local Activity per batch, with every record kind including `SUPERSEDED`. An idle reader sleeps on the stream's notifier with no Local Activity and no timer, and `position` lets the next run resume after Continue-as-New. It needs a server with the stream notifier and Workflow wake target. +Experimental `temporalio.contrib.streams`: a `workflow_reader()` ends its Workflow Task after 100 commands or 2 seconds by default (`TaskCutoff`), so a busy reader doesn't hold back the Workflow's other work. It asks Core to complete the task through the new `heartbeat_task` flag while its read runs on. +Experimental `temporalio.contrib.streams`: `close_workflow_stream(topic, result=, error=)` closes a Workflow's own stream from Workflow code. The run's `FINISH` goes with the activation, and the Worker appends it before it starts the Local Activity that closes the store and the stream's notifier, so every reader ends and sleeping readers see the outcome. +Experimental `temporalio.contrib.streams`: `StreamClosedError.closed_by` names who closed the stream. After a failed Workflow Task, the stream runtime drops the rerun's records on a topic the run closed itself. +Experimental `temporalio.contrib.streams`: `workflow_reader()` records each read's range and digest in History in place of its records by default, and replay reads the range from the store again before Workflow code sees it. `replay_from_store=False` records each batch in History instead. Other records fail the Workflow Task as nondeterministic, and a lost range fails it for a retry with `StreamReplayRangeLostError`. `RedisStreams` takes `floor_lease` and `floor_release_grace`. diff --git a/changelog/added/ringing-heron.md b/changelog/added/ringing-heron.md new file mode 100644 index 000000000..2af26f01b --- /dev/null +++ b/changelog/added/ringing-heron.md @@ -0,0 +1 @@ +Experimental `temporalio.contrib.streams`: a `workflow_reader()` reads first and attaches to the stream's notifier only when a read comes back empty, then reads once more with no wait before it sleeps. diff --git a/changelog/added/standalone-heron.md b/changelog/added/standalone-heron.md new file mode 100644 index 000000000..5528821f2 --- /dev/null +++ b/changelog/added/standalone-heron.md @@ -0,0 +1,3 @@ +Experimental `temporalio.contrib.streams`: a standalone Activity owns a stream. In a standalone Activity, `activity_handle()` is the Activity's own stream, and it writes as `@`. `StreamRef.for_activity` opens it, a read ends when the Activity run ends, and `delete_activity_streams` deletes it. +Experimental `temporalio.contrib.streams`: `start_activity_with_stream` starts an Activity that owns a stream of its own. The Worker closes the stream when the Workflow sees the Activity end, and the server's notifier closes it after its graces when that close never comes. +Experimental `temporalio.contrib.streams`: an Activity's producer attempt is `started_unix_ms << 11 | min(attempt, 2047)`, from the time the server started the attempt and its Temporal attempt. A retry and an attempt after a reset start later, so they supersede the earlier attempt, and the key fits 53 bits. diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock index 35314d57a..1a2e59585 100644 --- a/temporalio/bridge/Cargo.lock +++ b/temporalio/bridge/Cargo.lock @@ -2926,7 +2926,7 @@ dependencies = [ [[package]] name = "temporalio-streams" version = "0.1.0" -source = "git+https://github.com/moetemp/sdk-streams?rev=16592b73e5fa198c6f35cfbf22368fe566cca793#16592b73e5fa198c6f35cfbf22368fe566cca793" +source = "git+https://github.com/moetemp/sdk-streams?rev=b7637ed8e25abade941c89bc6c3139671a6fcb72#b7637ed8e25abade941c89bc6c3139671a6fcb72" dependencies = [ "async-trait", "aws-lc-rs", diff --git a/temporalio/bridge/Cargo.toml b/temporalio/bridge/Cargo.toml index d3e71ab82..85534c6fb 100644 --- a/temporalio/bridge/Cargo.toml +++ b/temporalio/bridge/Cargo.toml @@ -91,7 +91,7 @@ url = "2.2" # makes it use the submodule's crates, so the bridge sees one `Connection` type. [dependencies.temporalio-streams] git = "https://github.com/moetemp/sdk-streams" -rev = "16592b73e5fa198c6f35cfbf22368fe566cca793" +rev = "b7637ed8e25abade941c89bc6c3139671a6fcb72" default-features = false features = ["envconfig", "redis"] diff --git a/temporalio/bridge/proto/streams/streams_pb2.pyi b/temporalio/bridge/proto/streams/streams_pb2.pyi index 8233d88b5..92feee530 100644 --- a/temporalio/bridge/proto/streams/streams_pb2.pyi +++ b/temporalio/bridge/proto/streams/streams_pb2.pyi @@ -638,7 +638,7 @@ class AppendRecord(google.protobuf.message.Message): kind: temporalio.bridge.proto.streams.v1.message_pb2.StreamRecordKind.ValueType @property def body(self) -> temporalio.api.common.v1.message_pb2.Payload: - """Set on DATA only, after lang's payload codec and external storage.""" + """Set on DATA, and optional on LINK, after lang's payload codec and external storage.""" content_hash: builtins.bytes """The SHA-256 of the body as the payload converter made it, before the codec. A codec that encrypts with a fresh nonce makes a retry's bytes differ, so a retry is recognized by this. diff --git a/temporalio/contrib/streams/README.md b/temporalio/contrib/streams/README.md index 5b6a931bf..f810e7496 100644 --- a/temporalio/contrib/streams/README.md +++ b/temporalio/contrib/streams/README.md @@ -5,9 +5,10 @@ A stream is an ordered log that a Workflow owns. The Workflow publishes to it from Workflow code, and its Activities and any client append to it. Outside readers follow it from a cursor. Records live in the application's -Redis and never pass through Temporal or its History. The stream runtime, -the `temporalio-streams` crate from the sdk-streams repository, runs the -store for every SDK, so Python reads what any other SDK writes. Publishes and +Redis and never pass through Temporal or its History. The +stream runtime, the `temporalio-streams` crate from the sdk-streams +repository, runs the store for every SDK, so Python reads what any other SDK +writes. Publishes and appends add nothing to History. ## Register @@ -29,6 +30,12 @@ client = await Client.connect( ) ``` +The store tells the server's stream notifier after each append, which is +what wakes a reader that sleeps on the stream. An append answers once its +notification went. Keep it on in every process that appends. In a process +whose streams have no sleeping reader, `RedisStreams(...).notify_on_append(False)` +turns it off. + Redis 7.0 or later is required. Use `maxmemory-policy noeviction`: an evicting Redis drops whole stream keys under memory pressure, and the stream runtime logs a warning when it sees another policy. For tests and local @@ -38,11 +45,13 @@ in the process. ## Path B: an Activity streams to its Workflow An Activity writes to the stream of the Workflow that scheduled it, as -itself: its Activity id with the scheduling run id, and its Temporal -attempt. The run id keeps two runs of a chain apart, since Activity ids -restart in each run. When a retry starts writing, readers first see a -`SUPERSEDED` record, so they can drop what this Activity's failed attempt -wrote. +itself: its Activity id with the scheduling run id. Its producer attempt is +`started_ms << 11 | min(attempt, 2047)`, from the time the server started +the attempt and its Temporal attempt, so a retry and an attempt after a +reset write a higher one. The run id keeps two runs of a chain apart, since +Activity ids restart in each run. When a retry starts writing, readers +first see a `SUPERSEDED` record, so they can drop what this Activity's +failed attempt wrote. ```python TOKENS = topic("tokens", str) @@ -99,9 +108,6 @@ can't grow it either. A `SUPERSEDED` past that withdraws only the records still held. Call `forget(producer_id)` once a producer can't retry, to drop it sooner. -A future Workflow reader can't drop superseded records silently, because -replay must see the same records, so it will deliver `SUPERSEDED` too. - Every read follows the Workflow's run chain across Continue-as-New and ends when the chain closes. A reader can't stop at one run. A Workflow id can start a new chain after the old one closed, and a handle opens the newest @@ -112,8 +118,6 @@ To resume later, keep the last `record.cursor` and pass it back as in the store before it sleeps on the stream's notifier, so a longer `read(idle_wait=...)` holds a store connection longer and sleeps less often. -The sample runs in full in `tests/contrib/streams/samples/`. - ## A Local Activity streams to its Workflow A Local Activity calls `stream_producer` the same way, and its records @@ -156,19 +160,182 @@ Known limit: on the topic, the reader can't prove the call's task failed, and its records stand. +## Streams owned by a standalone Activity + +A standalone Activity is one that a client starts with `start_activity`, +with no Workflow. It owns streams of its own. In a standalone Activity, +`activity_handle()` is the Activity's own stream, and the producer and +close calls are the ones a Workflow Activity uses. + +```python +@activity.defn +async def generate(prompt: str) -> str: + stream = activity_handle() + out = stream.producer(topic=TOKENS) + async for token in model(prompt): + await out.append(token) + await stream.close(topic=TOKENS, result="done") + return "ok" + +handle = await client.start_activity(generate, "hi", id="gen-1", ...) +ref = StreamRef.for_activity(handle.id, run_id=handle.run_id, topic=TOKENS) +async for record in stream_handle(client, ref).read(): + ... +``` + +- **Owner.** The stream belongs to one Activity run. The ref names the + Activity id and its run. A ref with no run names the newest run of the id + when a read starts, and the read stays on that run. A new run of the same + Activity id has new streams. +- **Readers.** A client reads with `stream_handle(client, ref)`, and + Workflow code with `workflow_reader(ref)`, as for a Workflow's stream. +- **Producer.** The Activity writes as `@`. Retries, + backoff, pause and reset keep the stream open. +- **Attempt.** Each attempt writes the producer attempt + `started_ms << 11 | min(attempt, 2047)`, where `started_ms` is the Unix + time in milliseconds at which the server started the attempt. A retry and + an attempt after a reset start later, so readers see `SUPERSEDED` before + their records. An attempt that timed out but still writes started + earlier, so its late records are `stale`. Workflow Activities use the same + rule. +- **Close.** The Activity closes its stream with `close(result=)` or + `close(error=)`. The SDK closes nothing when the Activity returns, so make + the close the last step. When the run ends with no close, the stream's + notifier on the server closes it within its owner check interval (30 + seconds by default). A completed run gives a null result. A failed, + canceled, terminated or timed-out run gives the run's own failure, which a + Workflow reader raises after its last batch. A client reader that the + notifier did not wake ends with no outcome. +- **Codec.** Records and closes go through the codec with an + `ActivitySerializationContext` that holds only the namespace and the + Activity id, since a reader can build no more from the ref. +- **Server.** The server must run standalone Activities + (`activity.enableStandalone`) and the stream notifier. +- **Deleting.** `delete_activity_streams(namespace, activity_id)` removes + the streams of every run of an Activity id. + +A standalone Activity reads and follows other streams as a client does. To +go on after a retry, heartbeat the cursor of each record it handled, and +read after the cursor in the heartbeat details: + +```python +@activity.defn +async def follow(workflow_id: str) -> int: + details = activity.info().heartbeat_details + after = Cursor(details[0]) if details else BEGINNING + seen = 0 + async for record in stream_handle(activity.client(), workflow_id).read( + after=after + ): + seen += 1 + activity.heartbeat(record.cursor.token) + return seen +``` + +Known limits: + +- **Clock skew.** The attempt uses the server's attempt start time. After a + failover to a cluster whose clock is behind by more than the retry + interval, a new attempt can look older, and its records are `stale`. +- **Workflow Activities.** A Workflow Activity writes to its Workflow's + stream unless the Workflow started it to own one. See the next section. + +## Streams owned by a Workflow Activity + +An Activity that a Workflow schedules can own a stream too, for example a +tool call of an agent. The Workflow starts it with +`start_activity_with_stream`, which takes the arguments of +`workflow.start_activity` and a `topic`, and returns the handle and the +stream's ref. In that Activity, `activity_handle()` is the Activity's own +stream. + +```python +@workflow.defn +class Agent: + @workflow.run + async def run(self, call: ToolCall) -> str: + handle, ref = start_activity_with_stream( + run_tool, call, activity_id=f"tool-{call.id}", + start_to_close_timeout=timedelta(minutes=5), topic=TOKENS, + ) + return await handle + +@activity.defn +async def run_tool(call: ToolCall) -> str: + out = stream_producer(TOKENS) + async for token in tool(call): + await out.append(token) + return "ok" + +ref = StreamRef.for_workflow_activity( + agent_id, run_id=run_id, activity_id="tool-42", topic=TOKENS +) +async for record in stream_handle(client, ref).read(): + ... +``` + +- **Owner.** The stream belongs to one Activity in one Workflow run. The ref + names the Workflow id, the run that scheduled the Activity and the + Activity id. The run is required, since Activity ids such as `"1"` repeat + in each run of a chain. +- **Picking the stream.** `start_activity_with_stream` marks the Activity + with the header `__temporal_stream_owner`. With the mark, + `activity_handle()` is the Activity's own stream. Without it, it is the + Workflow's stream. `activity_handle(scope="activity")` and + `activity_handle(scope="workflow")` pick one. A Local Activity can't own a + stream, and `scope="activity"` raises `StreamUnsupportedError` there. +- **Producer.** The Activity writes as `@`, with the + attempt rule of a standalone Activity. A retry and an Activity reset + write to the same stream, and readers see `SUPERSEDED`. +- **Close.** The Activity need not close its stream. When the Workflow sees + the Activity end, its Worker closes each topic the Activity wrote. A + completed Activity gives a null result. A failed, timed-out or canceled + Activity gives the failure that History holds, which a Workflow reader + raises after its last batch. The Activity may close a topic itself first, + with its own result. +- **Backstop.** If the Workflow's Worker never closes the stream, for + example since it stopped right after the Activity ended, the stream's + notifier closes it with a generic failure. It does this once the + Activity is no longer pending, after `streamnotifier.ownerStartGrace` + from the notifier's start and `streamnotifier.ownerCloseGrace` from the + first check that found the Activity gone. A read that starts before the + Activity is scheduled waits for `owner_start_grace`. +- **Runs.** Continue-as-New and a Workflow reset make a new run, so the + rerun Activity writes a new stream, and a cursor of the old stream is + refused on it. A second Activity with the same id in the same run writes + to a closed stream, and its first append raises `StreamClosedError`. +- **Codec.** Records and closes go through the codec with an + `ActivitySerializationContext` that holds the namespace, the Activity id + and the Workflow id, and no other field. +- **Deleting.** `delete_workflow_streams` removes the streams of the + Workflow's Activities too. + +Known limits: + +- **Cancel.** With cancellation types `TRY_CANCEL` and `ABANDON`, the + Workflow sees the cancel at once, while the Activity may still run. The + stream closes then with the canceled failure, and later appends raise + `StreamClosedError`. +- **Owner check cost.** The owner check reads the run's pending + Activities. A run with thousands of them makes each check larger. +- **Codec keys.** A codec that keys on the Activity type or task queue + can't decode the close outcome. + ## Link records and followers -A stream can hold a reference to another Workflow's stream, as a `LINK` -record. A follower reads the stream and each stream its links open in one -order. So an agent writes a link per child agent in place of copying the -child's records, and a reader still sees one ordered view. +A stream can hold a reference to another stream, as a `LINK` record. A +follower reads the stream and each stream its links open in one order. So +an agent writes a link per tool call or child agent in place of copying +their records, and a reader still sees one ordered view. ```python -# Workflow code (agent). Links go out with the activation, as appended values do. +# Workflow code (agent). Links go out with the activation, as published values do. events = stream_producer(EVENTS) -events.link(StreamRef.for_workflow(child_id), name="sub-3", after=child_cursor) # OPEN -reply = await child.execute_update(...) -events.join(name="sub-3", until=reply.turn_end) # JOIN, the agent waited here +handle, tool_ref = start_activity_with_stream(run_tool, call, topic=TOKENS, ...) +events.link(tool_ref, name=call.id) # OPEN +events.link(StreamRef.for_workflow(child_id), name="sub-3", after=child_cursor) +await handle +events.join(name=call.id) # JOIN, the agent waited here # Client code. One ordered view of the agent and every stream it links. async for item in stream_handle(client, agent_id, topic=EVENTS).follow(): @@ -183,31 +350,33 @@ async for item in stream_handle(client, agent_id, topic=EVENTS).follow(): - **Writing.** `StreamProducer.link(ref, *, name, after=None, value=None)` and `WorkflowStreamProducer.link(...)` write an `OPEN`. The follower reads the linked stream strictly after `after`, or from its oldest record. - `join(*, name, until=None)` writes a `JOIN`. It says the producer waited - for the link up to `until`, or to the linked stream's end. A link names a - Workflow's stream only. Any other ref raises `ValueError`. + `join(*, name, until=None)` writes a `JOIN`. It says the producer waited for the link + up to `until`, or to the linked stream's end. A tool stream needs no `JOIN` + for its segment to end, since its stream closes with the call. - **Mixed batches.** `append` also takes a `StreamLink` next to plain values, so `append(event, StreamLink(LinkEvent.OPEN, ref, name))` writes the event and its link as one batch, with one dedupe identity. -- **Following.** `StreamHandle.follow()` and `ActivityStreamHandle.follow()` - read a stream and the streams it links, from client or Activity code. - Workflow code can't follow. - **Order.** Each stream keeps its own order. A linked stream's records never come before their `OPEN`. A record that comes after a `JOIN` waits until the linked stream reached the join point or its end. Other records come in the order they were read. No clock decides the order. - **Link reports.** Items with `link_state` say what happened to a link. - `OPENED` when a segment starts. `DONE` when it ends. `UNAVAILABLE` with - the error when it can't be read, for example a Workflow that never - started, after `owner_start_grace`. `SKIPPED` with a reason: `depth` past - `max_depth` (4), `limit` past `max_open` (16), `cycle` for a link back to - a stream on the path, `open` for a stream already open, and `endpoint` - for a stream in another namespace or behind a Nexus endpoint. A linked - stream never fails the follow. + `OPENED` when a segment starts. `DONE` with the linked stream's close + outcome when it ends. `UNAVAILABLE` with the error when it can't be read, + for example a Workflow that never started, after `owner_start_grace`. + `SKIPPED` with a reason: `depth` past `max_depth` (4), `limit` past + `max_open` (16), `cycle` for a link back to a stream on the path, `open` + for a stream already open, and `endpoint` for a stream behind another + Nexus endpoint. A linked stream never fails the follow. - **Positions.** Each record keeps its own stream's cursor. The last item of each batch carries a `FollowPosition`, which resumes the root and every open segment with no loss and no repeat. To resume from an earlier item, resume from the previous batch end. +- **Workflow followers.** `workflow_follower(ref, topic=EVENTS)` gives the + same items in batches, one Local Activity per step, and sleeps on the + wakes of every open source. Each open source is one subscription of the + Workflow. By default each step's marker holds its ranges and a digest, + and replay reads them again through the stream runtime. - **Retries.** An attempt owns the `OPEN`s it writes. A `SUPERSEDED` ends each segment the withdrawn attempt opened, with `DONE` and reason `withdrawn`. The same `OPEN` from a later attempt of that producer, with @@ -216,23 +385,22 @@ async for item in stream_handle(client, agent_id, topic=EVENTS).follow(): again shows each linked record once. Another `OPEN` of the name reads the stream from its `after`. - **Plaintext.** A link's ref, name and cursors are stored as plaintext, as - Workflow ids are in History. A link's value goes through the codec, so put - secrets there. + Workflow and Activity ids are in History. A link's value goes through the + codec, so put secrets there. Known limits: - **Size.** The follow position grows with the open segments. With the default bounds it stays below 64 KiB. -- **Waits.** An idle follow holds one store wait per open source. -- **Old readers.** A reader of an earlier release reads a `LINK` as a - record it can't decode. +- **Workflow followers.** A Workflow follower with more open sources than + its free wake sources gets `SKIPPED` with reason `limit` for the rest. - **Withdrawn joins.** A `JOIN` from a withdrawn attempt stays in effect, since its segment ended before the `SUPERSEDED`. ## A Workflow publishes to its own stream `stream_producer` is one function for Workflow code and Activity code. In -an Activity it returns `activity_handle().producer(topic=...)`, and +an Activity it returns `activity_handle(scope).producer(topic=...)`, and `append` is awaited. In a Workflow it returns the Workflow Producer, and `append` returns at once. Workflow code can't do I/O, so the Workflow Producer hands its records to the Worker with the activation that appended @@ -308,6 +476,36 @@ Known limits: publishes with the store, so a Workflow change that publishes other records replays without an error. +## A reader sleeps while the stream is idle + +On a server with the stream notifier, an idle reader sleeps instead of holding a store read open. A +client's `StreamHandle.read()` waits on the notifier with a long poll. +Workflow code reads with `workflow_reader()`, one Local Activity per batch, +and wakes on a Workflow Task when a producer appends: + +```python +@workflow.defn +class Follows: + @workflow.run + async def run(self, owner: str) -> None: + async with workflow_reader(owner, topic=EVENTS) as reader: + async for batch in reader: + for record in batch.records: + ... +``` + +By default History keeps each read's range and digest, not its records, and +replay reads them from the store again. The store keeps what the reader read +for `floor_lease` (the retention by default) after its last read or replay, +so a run that goes longer than that should Continue-as-New. +`replay_from_store=False` records each batch in History, for a History that +replays with no store. See +[Positions or recorded](docs/notifications.md#positions-or-recorded) to choose. + +A Workflow closes its own stream with `close_workflow_stream(topic, +result=..., error=...)`, and each sleeping reader sees the outcome. See +[Readers that sleep](docs/notifications.md). + ## What to know - **Retries deduplicate.** An append that fails with @@ -316,12 +514,17 @@ Known limits: producer state, such as `RedisStreams(plain=True)`, the copy lands and every reader drops it. `StreamStorePlugin.capabilities` says which. - **Retention.** A stream keeps records for `retention` (7 days by default, - set on `RedisStreams`) and dies that long after its last write. A cursor + set on `RedisStreams`) and dies that long after its + last write. A cursor past dropped records raises `StreamExpiredError`. - **Closing.** Once the Workflow's run chain ends, appends are refused with - `StreamClosedError`, after a short closing window. On a store without - producer state, a late append can still answer success, and no reader - delivers it, since every read ends at the first `CLOSE` record. + `StreamClosedError`, after a short closing window. Close earlier with + `close_workflow_stream` in the Workflow or `close()` on a handle. A + Workflow doesn't see a close from outside through its producer: the stream runtime drops + its later records on that topic, and the run goes on. So close a topic + from the Workflow that owns it, or have readers check how it closed. On a + store without producer state, a late append can still answer success, and + no reader delivers it, since every read ends at the first `CLOSE` record. - **Readers before the owner.** A read of a Workflow that Temporal doesn't know yet waits for `owner_start_grace` (60 seconds by default, set on `RedisStreams` or `MemoryStreams`), counted from the read's first owner @@ -330,12 +533,14 @@ Known limits: the grace, not at once. Appends and `latest()` don't wait. Zero turns the grace off. - **Deleting.** Deleting a Workflow does not delete its streams. - `RedisStreams.delete_workflow_streams` removes them at once. -- **Not in this release.** A Workflow can't read a stream. Streams owned - by an Activity or by no one raise `StreamUnsupportedError`. + `delete_workflow_streams` on the store removes them at once, and + `delete_activity_streams` removes the streams of a standalone Activity. +- **Not in this release.** Streams owned by no one raise + `StreamUnsupportedError`. ## Further reading +- [Readers that sleep](docs/notifications.md) - [Store capabilities and guarantees](docs/store-capabilities.md) - [Redis store: guarantees and setup](docs/redis-guarantees.md) - [Stream wire contract](docs/wire-contract.md) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index a7d82933a..deb8cb0fd 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -1,9 +1,10 @@ -"""Streams: an ordered log a Workflow owns, written from many places. +"""Streams: an ordered log a Workflow or an Activity owns, written from many places. .. warning:: This package is experimental and may change in future versions. -A Workflow owns a stream. Producers outside Workflow code, such as the +A Workflow or an Activity owns a stream. Producers outside +Workflow code, such as the Workflow's Activities and clients, append to it, and outside readers consume it from a cursor. Records live in a store the application runs, which the stream runtime reaches for every SDK. Record bodies never reach the server. @@ -46,19 +47,29 @@ activation, and nothing goes to History. - An Activity writes to the stream of the Workflow that scheduled it with :func:`stream_producer` too, and reads it with :func:`activity_handle`. - It writes as itself: its Activity id with the scheduling run id, and its - Temporal attempt. A Local Activity writes as its call number in the run, - with an attempt that grows across its reruns. -- Any process holding a client reaches a Workflow's stream with + It writes as itself: its Activity id with the scheduling run id, and an + attempt from its start time and its Temporal attempt. A Local Activity + writes as its call number in the run, with an attempt that grows across + its reruns. In a standalone Activity, :func:`stream_producer` and + :func:`activity_handle` reach the Activity's own stream, and the Activity + writes as its id with its run id. An Activity that Workflow code started + with :func:`start_activity_with_stream` owns a stream too, which + :func:`activity_handle` gives it, and which closes when the Workflow sees + the Activity end. +- Any process holding a client reaches a stream with :func:`stream_handle`, and writes with a producer id and an attempt of its own. -Every read follows the Workflow's run chain across Continue-as-New and ends -when the chain closes. A reader can't pin one run. +Every read of a Workflow's stream follows the Workflow's run chain across +Continue-as-New and ends when the chain closes. A reader can't pin one run. +A read of a standalone Activity's stream ends when the Activity run ends. A +read of the stream of an Activity that a Workflow scheduled ends when the +scheduling run sees the Activity end. -In this release only a Workflow owns a stream, and only Activities and -clients read one. The other owner kinds raise -:class:`StreamUnsupportedError`. +Workflow code reads a stream with :func:`workflow_reader`, one Local +Activity per batch, and sleeps on the stream's notifier while the stream is +idle. Only a Workflow or an Activity owns a stream. The other owner kinds +raise :class:`StreamUnsupportedError`. The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body`` diff --git a/temporalio/contrib/streams/docs/migration.md b/temporalio/contrib/streams/docs/migration.md index 4534b01d9..42c688355 100644 --- a/temporalio/contrib/streams/docs/migration.md +++ b/temporalio/contrib/streams/docs/migration.md @@ -41,12 +41,11 @@ Differences to plan for: ## What does not map -- **Reading inside the Workflow.** A Workflow cannot read a stream in this - release. If a Workflow reads its own Workflow Stream today, keep that - part on Workflow Streams, or send the Workflow what it needs, for example - as a Signal. -- **Activity-owned and standalone streams.** Only a Workflow owns a stream - in this release. +- **Reading inside the Workflow.** A Workflow reads a stream with + `workflow_reader`, one Local Activity per batch, and sleeps on the + stream's notifier while the stream is idle. That needs a server with the + stream notifier. On a server without it, keep that part on Workflow + Streams, or send the Workflow what it needs, for example as a Signal. - **A server-sent events bridge.** Streams ships none. Serve readers from your own endpoint with `stream_handle(...).read(...)`. - **Records already published.** They stay where they are, in the diff --git a/temporalio/contrib/streams/docs/notifications.md b/temporalio/contrib/streams/docs/notifications.md new file mode 100644 index 000000000..6c66d13a2 --- /dev/null +++ b/temporalio/contrib/streams/docs/notifications.md @@ -0,0 +1,285 @@ +# Readers that sleep + +A reader that waits for the next record can sleep while the stream is idle, +instead of holding a store read open. The server's stream notifier wakes it +when a producer appends, or when the stream closes. Two readers sleep this +way in one namespace: + +- A client reader, `StreamHandle.read()`, waits on the notifier with a long + poll. +- A Workflow reader, `workflow_reader()`, reads in batches through Local + Activities and wakes on a Workflow Task. + +Both need a server with the stream notifier, and producers that notify. + +## Server setup + +The notifier is a CHASM component. Turn it on with these dynamic config +settings: + +```yaml +history.enableChasm: + - value: true +history.enableCHASMCallbacks: + - value: true +streamnotifier.enabled: + - value: true +``` + +These settings change how readers wake. The defaults are right for most +deployments: + +| Setting | Default | What it does | +|---------|---------|--------------| +| `streamnotifier.pollMaxWait` | 60 s | How long a client reader's poll waits. At the limit the reader reads again | +| `streamnotifier.recheckInitial` | 1 min | When the notifier first asks an idle Workflow reader to read again | +| `streamnotifier.recheckMax` | 5 min | The longest gap between two such asks. The gap doubles up to it | +| `streamwake.minInterval` | 1 s | The shortest time between two Workflow Tasks that only carry wakes | +| `history.maximumStreamWakeSources` | 32 | How many readers one Workflow execution can have at once. A reader frees its place when it detaches, when its attach fails, or when its run takes its close or resubscribe wake | +| `streamnotifier.maxCallbacks` | 2,000 | How many Workflow readers one stream can have | +| `streamnotifier.ownerCheckInterval` | 30 s | How often the notifier checks a stream's owner | +| `streamnotifier.idleTimeoutNoCallbacks` | 1 h | How long a notifier with no reader stays | + +## Notifications are on by default + +Every store notifies after each append. An append answers only after its +notification went, or after a few failed tries, so a producer that stops +before then gets no answer, and its retry notifies again. Keep it on in +every process that appends or reads. A client reader sleeps on the notifier +only when its own store has it on. A process whose streams have no sleeping +reader can turn it off: + +```python +store = RedisStreams("redis://localhost:6379").notify_on_append(False) +client = await Client.connect("localhost:7233", plugins=[store]) +``` + +On a server without the stream notifier, the first notification fails and +the process stops notifying. A Worker built from the +client waits for the notifications still out when it stops. A process that +holds only a client calls `await store.flush_notifications()` before it +exits. + +## When the stream runtime notifies + +1. A reader reads and waits up to its idle time, 2 seconds by default. A + read that finds records returns them at once. +2. A read that waits its whole idle time and finds nothing answers + `wake_after`, the counter of its position. +3. The reader sleeps until the notifier's counter is above `wake_after`, the + stream closes, or a time limit passes. A Workflow reader asks the + notifier for this one wake in the same Local Activity as the empty read. + A client reader holds a long poll. +4. Every append notifies the notifier, from the producer's process. + Notifications for one stream fold into one call in flight, and a failed + last notification is tried twice more. +5. The notifier wakes every sleeping reader with one notification. A reader + that reads records asks for nothing, so it gets no wake while it works. + +The server drops a notification for a stream with no notifier, and a +reader's attach or wait starts the notifier. So a reader reads first, and +attaches only when a read comes back empty, then reads once more before it +sleeps, which catches an append that landed before the notifier existed. + +A Workflow reader attaches as an idle target, which gets no wake until it +waits. Each wait gets at most one wake, and uses it up. A wait whose counter +the stream passed already wakes the reader at once. So a Workflow reader +gets one wake per idle stretch, whatever the append rate, and the records +that land while it reads cost no Workflow Task. Its next read finds them. + +The notifier delivers the first notification of a window at once and the +newest one at the window's end (`streamnotifier.debounceWindow`, 1 second by +default). A wait that the window holds back wakes the reader at the +window's end. + +## A client reader + +`StreamHandle.read()` keeps its API. With notifications on, an idle read +sleeps on the notifier: + +```python +handle = stream_handle(client, workflow_id, topic=EVENTS) +async for record in handle.read(): + ... +``` + +The sleep holds one gRPC long poll, not a store connection. At the poll +limit the reader reads again. That read finds records whose notification +was lost, or comes back empty and sleeps again. On a server without the notifier, the +reader waits in the store instead, as without notifications. + +## A Workflow reader + +```python +@workflow.defn +class Connector: + @workflow.run + async def run(self, channel: str, after: StreamPosition | None) -> None: + async with workflow_reader( + channel, topic=MESSAGES, after=after or BEGINNING + ) as reader: + async for batch in reader: + for record in batch.records: + if record.kind is RecordKind.DATA: + await workflow.execute_activity(post, record.value, ...) + if workflow.info().is_continue_as_new_suggested(): + workflow.continue_as_new(args=[channel, batch.position]) +``` + +- Each read is one Local Activity, which adds one marker event when its + Workflow Task completes. By default the marker holds the read's range, and + replay reads the range from the store again (see below). +- Each read names the run as the subscriber. Only a read that comes back + empty attaches the run to the stream's notifier, in the same Local + Activity, and the stream runtime reads once more after the attach, so it finds a record + that landed before the attach. A reader that never goes idle never + attaches. While the stream is idle, the reader waits with no Local + Activity and no timer. The server records each wake on the next Workflow + Task's scheduled event, so a wake adds no event of its own. +- The reader gets one wake per idle stretch. Each empty read asks the + notifier for the next wake, and a read that finds records asks for none. + So a reader that works on a batch, for example in an Activity, gets no + Workflow Task from the appends that land meanwhile. +- A batch holds every record kind, `SUPERSEDED` and `FINISH` too, since + replay must give your code the same records. `SupersessionTracker` names + the records each `SUPERSEDED` withdraws. +- `batch.position` carries the cursor and the stream runtime's read state. Pass it as + `after=` in the next run, for example after Continue-as-New. +- `async with` detaches a reader that stops before the stream ends. +- A reader that finds records on every read ends its Workflow Task after 100 + commands or 2 seconds, so your Activities and timers do not wait for the + reads. Change it with `cutoff=TaskCutoff(commands=..., time=...)`. +- A read that fails for good, for example on an owner that does not exist, + fails with an `ApplicationError` whose type names the stream error. +- Each marker has a summary that says what the read did, for example + `stream read tokens after the start` or `stream unsubscribe tokens`. The UI + shows it on the marker, since every read marker has the same Activity type. +- The read and close Activities run only as the Local Activities that the + stream API schedules. Workflow code that calls one as a normal Activity + gets a non-retryable `StreamUnsupportedError`. + +### Positions or recorded + +A Workflow reader keeps what it read in one of two ways. Choose per use case. + +| | Positions (default) | Recorded | +|---|---|---| +| Option | `replay_from_store=True` | `replay_from_store=False` | +| History per read | The range, a digest and the stream runtime's read state | Every record of the batch | +| Replay reads | The store, through the store plugin | History only | +| Replay needs | The store, and the reader's hold on the range | Nothing outside History | +| Run length | Continue-as-New before `floor_lease` passes | Bounded by History size | + +Use positions for a reader that reads a lot or reads large records, such as +a connector that follows a busy stream. Use recorded batches for a History +that has to replay with no store, for example one you ship to another team +or keep past the store's retention, or for a run that waits longer than +`floor_lease` between reads. + +### Replay from the store + +The store is the record of a stream, so History does not keep a second copy +of what a Workflow reader read. By default, `replay_from_store=True`, each +read's marker holds the read's range, the stream runtime's digest of its records, and +The stream runtime's read state after the read. The state holds the newest attempt of +each producer the read saw, up to 16 KiB. A read starts from the state the +reader's previous read ended with, so its marker leaves that out, and +replay takes it from the previous marker. A read counts only once it +returned, so a failed or canceled read leaves no gap. The reader's first +read keeps the state it started from, the one `after=` gave, and so does +the first read of a reset run, since the markers before it are another +run's. So a marker is a +few hundred bytes for a stream with a few producers, whatever the batch +holds, and up to about 17 KiB for a stream with many, such as one that +Local Activities write to, which has a producer per call. A stream that a +`stream_producer` publishes to has one producer per run, `@`, so its +state stays small. At 17 KiB, 1,000 reads add about 17 MiB to History. + +- On replay, the Worker reads the same range from the store again, before + Workflow code sees the batch. Workflow code gets the same records in both + modes, and no I/O runs in it. The live task takes the batch the Local + Activity read, from memory. +- Replay decodes a reader's markers in order. A process that sees a later + read before the one ahead of it, which a Worker that replays the full + History never does, fails the Workflow Task, and the retry replays from + the start. +- Other records in the range fail the Workflow Task as nondeterministic. A + range the store lost, to retention or a delete, fails the task for a + retry, so an operator can restore the store, reset the Workflow, or go + back to recorded batches. +- The store holds the records the reader read against retention, per + reader run. Each read, and each replay of a read, renews the hold for + `floor_lease` (the retention by default, set on `RedisStreams`). The + reader never ends the hold itself, not at the stream's end and not at + `close()`, since the run may replay its reads until it ends. +- So a run can replay a read only while the hold lasts. A run that goes + longer than `floor_lease` with no read and no replay, for example an + entity Workflow that read a stream and then waits on Signals for longer, + loses the range. Its next replay, after a Worker restart, an eviction or + a query, raises `StreamReplayRangeLostError`, and the Workflow Task fails + on each retry until an operator resets the run. A long-running reader + should Continue-as-New well before `floor_lease` passes, or pass + `replay_from_store=False`. +- A positions-only reader pins every record after its first read until its + run ends. Plan a Continue-as-New for a long reader. Each holder costs + store memory and append time. `reader.continue_as_new_suggested()` says + when the records it read pass 256 MiB, or a day passed since its first + read. Pass `reader.position` as `after=` to the next run. +- Replay needs the store. A `Replayer` needs the store plugin and a client + in its process that connected the same store. Without them, the read + raises `StreamUnsupportedError`, which names the opt-out. +- `workflow_reader(..., replay_from_store=False)` records each batch in + History instead, for a History that replays with no store, such as one you + ship to another team or keep past the store's retention. It costs History + size for each record read. +- The mode is per reader and fixed for a run. Histories written in one mode + replay in that mode, so a change of the default or of the option applies + to new runs. + +The store plugin does the replay step in the client's payload codec: it +wraps the codec the client has and passes every other payload through. + +## Closing a stream + + + +| Who closes | How | +|------------|-----| +| The owner Workflow | `await close_workflow_stream(topic, result=..., error=...)` | +| An Activity | `await activity_handle().close(result, error=...)` | +| A client | `await stream_handle(client, workflow_id).close(result, error=...)` | +| Nobody | The owner's run chain ends. The notifier's owner check closes the stream | + +`close_workflow_stream` puts the run's `FINISH` in the activation, after +what the run published before. The stream runtime appends it before it starts the close's +Local Activity, which closes the store and the notifier. A Worker with no +Workflow cache can rerun the task of a close whose Local Activity was still +in flight, since an eviction cancels it. Readers then see `SUPERSEDED` +before the second `FINISH`. +Every read ends after the last record. A sleeping reader wakes. The store +keeps the first close's outcome and hands it to every reader that reaches +the end, so a client read raises the close's error after its last record, +and a Workflow reader raises it after its last batch. A close with a result +ends the read with no error. + +A close from an Activity or a client is not visible to the owner +Workflow's producer. The stream runtime drops the Workflow's later records on that topic +and the run goes on. Close from the owner Workflow, or have +readers check how the stream closed. + +Close explicitly. Without a close, the notifier's owner check closes the +stream up to `ownerCheckInterval` plus `ownerCloseGrace` after the owner +ended, and a client reader ends at its next read. + +## A lost notification + +A producer can stop after its append and before its notification lands. +Its append then gets no answer, so a retry appends again, which writes +nothing twice and notifies. A producer that never retries, or a process +with notifications off, leaves the record to the next append by any +producer, or to a time-based check: + +- A client reader reads again at the poll limit (60 seconds). +- The notifier asks an idle Workflow reader to read again after 1 minute, + doubling to 5 minutes. That costs about 290 Workflow Tasks a day per idle + reader. diff --git a/temporalio/contrib/streams/docs/redis-guarantees.md b/temporalio/contrib/streams/docs/redis-guarantees.md index a3c2edc51..ea57ad390 100644 --- a/temporalio/contrib/streams/docs/redis-guarantees.md +++ b/temporalio/contrib/streams/docs/redis-guarantees.md @@ -43,6 +43,11 @@ reader's cursor can name an entry the new primary never had, and new entries can get lower ids, so the reader skips them. Read again from `latest()` after a known failover. +OSS Redis, Valkey and ElastiCache replicate asynchronously. A failover can +lose acked records, closes and dedupe entries. A positions-only Workflow +reader then fails replay as nondeterministic. Use MemoryDB, or +`replay_from_store=False`. + Recommended for production: `appendonly yes` with `appendfsync everysec` or `always`, at least one replica, and the settings below. @@ -61,7 +66,8 @@ Recommended for production: `appendonly yes` with `appendfsync everysec` or error they can retry. - **Enough memory for retention.** A stream keeps records for `retention` (7 days by default) after they were written. Size Redis for your write - rate times the retention. + rate times the retention. For a longer retention, see + [Long retention](#long-retention). ## Retention and cleanup @@ -82,6 +88,23 @@ The store cleans up after itself, with no job to run: the keyspace with `SCAN`, so run it rarely. Run it only after the chain closed and its producers stopped: it also removes the close flag and the dedupe state, so a late append writes a fresh log. + `delete_activity_streams(namespace, activity_id)` does the same for the + streams of every run of a standalone Activity id. + +## Long retention + +Redis keeps every record in memory. So on OSS Redis and Valkey, and on +Amazon ElastiCache, retention is bound by memory: write rate times +`retention` must fit in `maxmemory`. AOF and RDB do not lift that bound. +They copy memory to disk, and a restart loads it all back into memory. Do +not raise `retention` on these servers to keep streams for observability. + +On AWS, Amazon MemoryDB for Valkey is the production target for the store. +It runs the store's scripts unchanged, and its multi-AZ write log keeps +acknowledged records through a failover. Its data tiering moves cold keys +to SSD, but it moves a topic's whole log, and keeps a log above 128 MiB in +memory. See `crates/streams/docs/redis.md` in the sdk-streams repository for its +access string. ## When a Workflow closes @@ -105,13 +128,16 @@ stream: the next run keeps publishing to it. Temporal does not authorize or meter record traffic. Records go straight between your applications and your Redis, so access is Redis's job. -Readers write nothing, so a reader needs only read access to Redis, and the -Temporal permission to describe the Workflow. A read holds one Redis -connection while it waits, from a pool the stream runtime keeps per node, so -set `blocking_reads_per_node` on `RedisStreams` to the streams a process -follows at once. A read that finds the pool full waits for a connection -until its own poll ends, then answers empty, so a process that follows more -streams than the pool holds sees records late rather than failing. +Readers write no records and no close flags. A client reader needs only read +access to Redis, and the Temporal permission to describe the Workflow. A +`workflow_reader` with `replay_from_store=True` also writes its hold in the +topic's `:floors` hash, so its Worker needs the writer rules. A read holds one +Redis connection while it waits, from a pool the stream runtime keeps per +node, so set `blocking_reads_per_node` on `RedisStreams` to the streams a +process follows at once. A read that finds the pool full waits for a +connection until its own poll ends, then answers empty, so a process that +follows more streams than the pool holds sees records late rather than +failing. Every key of a namespace's streams starts with `:{:`, where `` is the store's `key_prefix` diff --git a/temporalio/contrib/streams/docs/store-capabilities.md b/temporalio/contrib/streams/docs/store-capabilities.md index f170780af..d67523bab 100644 --- a/temporalio/contrib/streams/docs/store-capabilities.md +++ b/temporalio/contrib/streams/docs/store-capabilities.md @@ -44,8 +44,8 @@ application kept. |---|---|---|---| | Producer state | `producer_state` | A retry of the newest batch writes nothing and answers the first positions. A changed batch, or one with an old sequence, fails with `StreamProducerError`. An append after a close fails with `StreamClosedError`. | A retry can land a second copy, which readers drop. The producer can get the second copy's positions. A changed or stale batch lands with no error, and readers keep the first copy. An append that races a close can answer success, and no reader delivers it. A Workflow then gets no `workflow_stream_output_dropped` metric for it. | | Kept last position | `kept_last` | The store knows a log's last position after retention dropped all of it, for 30 days. So a read tells an expired cursor from an empty stream. | The stream runtime decides expiry from the cursor's own record. Retention drops only the oldest records, so if that record is still there, nothing after it is lost. If it is gone, the read fails with `StreamExpiredError`, also when nothing after it was lost. | -| Retention holds | `holds` | A holder keeps records from retention. | No holder can keep records. | -| Change feed | `change_feed` | One consumer reads every append in order. | Producers notify after each append. | +| Retention holds | `holds` | A `workflow_reader` with `replay_from_store=True` holds the records it read against retention. | A positions-only read works only on a store with a `replay_horizon`. Otherwise its first read fails with `StreamUnsupportedError`, and the reader needs `replay_from_store=False`. | +| Change feed | `change_feed` | The notification log keeps a notification for each append, so a producer that stops after its append still wakes its readers. | Producers notify after each append, and an append answers only after its notification went. A producer that stops before then gets no answer, and its retry notifies. A producer that never retries leaves the reader to its next recheck. | | Cheap logs | `cheap_logs` | One log per topic of each run chain. | The stream runtime would share a few logs between streams. No store in this release needs it. | | Delete by owner | `delete_owner` | `delete_workflow_streams` removes a Workflow's streams. | `delete_workflow_streams` raises `StreamUnsupportedError`. Records go at retention. | | Topic listing | `topic_listing` | A closer finds every topic of a run chain. | Producers write a `TOPIC` record to the chain's control log, so a closer finds the topics there. | diff --git a/temporalio/contrib/streams/docs/wire-contract.md b/temporalio/contrib/streams/docs/wire-contract.md index 3699773a8..c93b96557 100644 --- a/temporalio/contrib/streams/docs/wire-contract.md +++ b/temporalio/contrib/streams/docs/wire-contract.md @@ -32,6 +32,11 @@ Workflow's run chain: (Python's `urllib.parse.quote(part, safe="")`). So no part can contain `:`, `{` or `}`, and no two streams share a key. +The streams of one standalone Activity run share the hash tag +`{:!a::}`, with the same suffixes. The +`!a` part is written as it is. Every other part is percent-encoded, which +writes `!` as `%21`, so no Workflow id makes this tag. + | Suffix | Type | Holds | |---|---|---| | `:t:` | stream | The log of one topic | @@ -156,10 +161,11 @@ publishes, when the Worker evicts the run, and at Worker shutdown. A cursor token is `redis::`. - `` is the first 8 lowercase hex characters of the SHA-256 - over namespace, owner kind (`workflow`), Workflow id and topic, each as - UTF-8 bytes prefixed by its length as 8 bytes big-endian. The run id is - not part of it, so a cursor stays valid across Continue-as-New. So it - also can't tell two chains of one Workflow id apart. + over namespace, owner kind (`workflow` or `activity`), the Workflow id + or the Activity id, and topic, each as UTF-8 bytes prefixed by its + length as 8 bytes big-endian. The run id is not part of it, so a cursor + stays valid across Continue-as-New. So it also can't tell two chains of + one Workflow id apart. - The empty token is `BEGINNING` and `$end` is `END`. A reader refuses a token from another store or with another stream