Added server-side streams for workflows and external clients. - #2
Open
moedash wants to merge 72 commits into
Open
Added server-side streams for workflows and external clients.#2moedash wants to merge 72 commits into
moedash wants to merge 72 commits into
Conversation
Derives a server-side stream primitive from Temporal's storage invariants: the stream gets its own history-node branch, appends do not schedule workflow tasks, and readers own their cursor. Keeps the CHASM component O(1) so it does not depend on partial reads.
Max's external-store prototype and Johann's dedicated-facet prototype both reach further than our design did in places, so the comparison is recorded and the twelve resulting changes are folded into both docs. The load-bearing one is a correctness fix: an in-workflow consumer must record its delivered range on every task, including empty ones, or replay can hand it items it did not have. Riding WorkflowTaskCompleted instead of a new event makes the idle case free.
The Native Streams design reuses the history-node store for stream payloads instead of standing up a parallel one, which rests on two properties that were only ever read off the code. Both now have tests on SQLite and Postgres: a non-proto blob round-trips byte-identical on a branch whose tree ID is not a run ID, and the transaction-ID chain rejects a stale node left by a shrinking retry once the frontier has moved past it. The stale-node case is asserted on the raw read path as well, since that is the path a stream uses and it carries no contiguity check of its own. A negative control confirms the assertion is load-bearing.
The neighbouring suite methods use underscores but predate the lint base rev, so only new code is held to ST1003 and ST1020.
The canonical option list now lives in Notion, so design-comparison.md says so and narrows its own scope to our design against the prototypes that exist as running code. Option 7 and Max's task/python-sdk-streaming branch are the same approach, so they are compared as one. Option 7 targets bucket 2, which makes the claim that nothing else did stale; the high-level design said that and no longer does. Took two things from it: an explicit flush marker so a consumer does not wait out an idle timeout, and a per-topic sequence alongside the global offset.
…ign. Codex independently reached nearly the same architecture, and found a real flaw in ours. history_node partitions on tree_id alone, which is safe for workflow history because history is capped and unsafe for a stream because it is not. Streams now roll to a new tree every bucket_size offsets; the bucket is arithmetic so there is still no index. Also filled three gaps it exposed: continue-as-new for an attached stream needs the successor to inherit the state and the old child to redirect, reset must carry the current head forward rather than rewind, and a missing committed node has to read as DataLoss rather than as an empty stream. Group commit moves to deferred, matching their reasoning that the case for it rests on Cassandra numbers this prototype does not measure.
Stage 0 of AI-198. Reproduces the shipped pattern, batched Signals in and a long-polling Update out, and measures history events, history bytes, persistence operations, and latency across flush interval and subscriber count. Nothing to compare a native path against without it. Latency is stamped at generation rather than at flush. Stamping at flush measures only the server round trip and hides the batching delay, which is the dominant term and the whole reason a shorter interval is wanted. Two ceilings fell out that matter more than the cost curve. history.maxInFlightUpdates caps concurrent subscribers at 10 and starves rather than degrades past it. history.maxTotalUpdates gives a workflow 2000 reads for its entire life, so at 100ms batching five subscribers exhaust it in under a minute and force continue-as-new for reasons that have nothing to do with the application.
Stage 1 of AI-198. The component holds only the frontier: head and base offsets, the transaction chain, producer dedup, and consumer pins. It is O(producers + consumers) rather than O(messages), which is what keeps it off the CHASM partial-read path. Payload bytes go to the history-node store on trees of the stream's own, rolling to a new tree every bucket_size offsets. That is not cosmetic: the Cassandra table partitions on tree_id alone, which is safe for workflow history because history is capped and unsafe for a stream because it is not. Buckets are arithmetic and tree IDs derive from them, so nothing indexes them. The component stages appends rather than writing them. Keeping the staging separate from the transaction is what will let an append ride a workflow's own commit later without the component knowing. Transaction IDs come from the caller rather than from stream state: a retry has to carry a higher ID than the attempt it replaces, and a counter derived from committed state would hand the retry the same one. The bucket-boundary test earned its place twice. It found a real bug where the result cap was passed through as a storage page size, which made the store read past the end of an empty result. Its negative control confirms the transaction chain, not luck, is what rejects a stale node in a freshly started tree.
Eight shard-routed RPCs, with the request shapes declared in the library rather than in the public API, since streams have no public API yet. Nesting them under frontend_request matches the other CHASM libraries, so promoting them later is a package move rather than a redesign. PollMessages is categorised as a long poll so it lands in the right quota bucket, though it does not block yet. ListStreams is left out: it needs a visibility field and search attributes on the component, which belongs with the lifecycle work rather than here, and nothing on the benchmark path reads it.
History registers the library and serves the service; frontend resolves the namespace name and forwards over the layered client, which routes to the shard owning the stream. The frontend registers the service on its gRPC server directly rather than embedding a handler in WorkflowHandler, because StreamService is not part of the public API. Appends are serialized per stream in the handler. The node has to be durable before the frontier advances, so it is written outside the transition that advances it, and two concurrent writers could otherwise commit in a different order than they wrote: whichever node carried the higher transaction ID would win the read regardless of which writer actually committed. Serializing removes the interleaving. It is a stopgap for the CHASM transaction hook, and it is noted as such in the handler rather than left for someone to discover.
Reads starting inside a batch silently dropped everything before the batch boundary. A node is addressed by the first offset of its batch, so a read has to begin at the node containing the offset, not at the node whose ID equals it. Batch size is now bounded on write, which bounds how far back a read steps; the bound is a correctness mechanism, not only admission control. A negative control confirms the test catches it. Producer and consumer maps come back nil after deserialization, so appends against a reloaded stream failed on a nil map write. Guarded at each write rather than relying on the constructor. Truncation, close, dedup, topic filtering, and per-producer fencing are covered end to end through the frontend.
A caught-up reader now parks on chasm.PollComponent until the head moves past it or the stream closes. The predicate is monotonic as that API requires: the head only advances and closed never clears. Expiry of the server's own budget returns an empty response rather than an error, so a reader cannot confuse a quiet stream with a failure. A reader that is already behind is never parked. The tail cache keeps recently appended batches in memory so a reader at the tail is served without a database read. That is what makes fan-out cheap: readers at the tail cost a copy each rather than a range scan each, which is the difference between a subscriber ceiling and none. Two properties it has to hold. It caches only after the commit, because a write whose commit failed can be superseded by a retry carrying different bytes at the same offsets. And a partial hit is a miss: returning only the tail of a requested range would look like a short read, which is the shape of a silent data loss. Group commit stays deferred. The case for it rests on Cassandra numbers this prototype does not measure.
A capped stream now advances its own readable floor at the end of a successful append rather than through a sweeper. The append transition is already writing, so folding the check into it costs nothing and keeps the cap tight instead of eventually true. Truncation reclaims whole buckets. That is the point of bucketing: a storage partition is dropped outright rather than leaving a tombstone per message. Deletion always follows the commit that moved the floor, so a failure leaks storage to reclaim later rather than losing data a reader can still ask for. The cap yields to a registered workflow consumer. That consumer's history records an offset range it must re-read on replay, so storage grows rather than the consumer losing data underneath it. Close now records when and arms a retention task; deletion removes the log before the execution, since the other order would drop the only record of which buckets exist and leak them permanently. Close became a pure function returning when to schedule, with the caller arming the task. That keeps the component testable without a live context, matching how appends already stage rather than persist. max_bytes is removed from the lifecycle proto rather than left accepted and silently ignored. Enforcing it needs per-bucket byte accounting that does not exist yet.
Streams now carry a visibility field and ListStreams answers from visibility on the frontend. Without it the only way to reach a stream is to already know its ID, which is not something an operator can work with. It is deliberately not on the history handler: the query is against visibility rather than any one stream, so there is no business ID to route on and nothing a shard could answer. Unit tests build the component directly instead of through NewStream, which now needs a live context to wire the visibility field. Construction through the real path stays covered end to end.
Persistence metrics and rate limits are keyed on caller info carried by the context, not on the request. The RPC path sets it through interceptors, but the stream writes its log directly, so those calls carried no caller name at all. They were escaping namespace rate limiting and priority, and going uncounted in per-namespace metrics. Found while building the native half of the benchmark, which reported zero persistence operations per message for a path that demonstrably writes on every append. Worth stating plainly: the first version of that number was a fiction, and only the raw per-operation breakdown exposed it. The retention task needs the same treatment, since it runs outside any request.
Without one, every append and every poll resolved the stream's current run through persistence. That lookup dominated: at 100ms batching the native path spent 302 of 423 persistence operations on it, which made an otherwise free read cost a database call and left the design no cheaper than the pattern it replaces. Supplying the run id, which CreateStream already returns, takes the same workload from 423 operations to 121 against the baseline's 408. The remaining native cost is exactly what the design predicts: one log append and one frontier update per flush, and nothing at all for reads. Optional, and defaulting to the current run, matching how run ids work elsewhere in the API.
Same workload through both designs, sharing workload generation, latency stamping, and metric capture, because a benchmark whose halves generate load differently measures the harness rather than the design. At 100ms batching with 25 subscribers the current pattern delivers 19% of expected messages with an 11.2 second p99. The native path delivers 100% with a 104ms p99 at a twentieth of the persistence cost per message. The shape matters more than the ratio: native cost per message barely moves between 1 and 25 subscribers because reads cost no persistence operations at all, while the current pattern's rises sevenfold over the same range. Both measurement bugs are recorded in the results rather than quietly fixed, since each produced a confident wrong number that no failing test would have caught.
…cked. The Command attributes oneof is closed and the module cache is read only, so a workflow cannot publish to a stream without changing the public API module. The patch against temporalio/api is exact and applies cleanly; what does not work is generating from it. buf.gen.yaml runs a plugin from a directory absent from the repository, and the published module is a reshaped artifact rather than the checkout. A local replace directive would make this branch unbuildable for anyone else, which defeats a prototype meant to be reviewed. The unblock is a branch on temporalio/api, which is a shared repository and not mine to push unilaterally. Recording it rather than leaving the next attempt to rediscover it.
A producer standing in for an LLM-calling activity appends tokens, and a browser watches them arrive over SSE with a bridge in the middle doing what an application backend would. It uses no SDK and no workflow, which is the point of the path it exercises: tokens come from an activity rather than from workflow code, so the demo does not wait on the workflow-facing API that Paths A and C still need. The reconnect story falls out of the design rather than being built: the reader owns its offset, so resuming means passing the offset back and the server remembering nothing about the reader.
Paths A and C need a command type and two fields that cannot exist without changing the public API module, so the module is forked to moedash/api and pinned here by pseudo-version. Building a usable fork was most of the work. A clean clone cannot generate at all, plain buf generate leaves an enum prefix the published module strips, and the pipeline does not emit several packages the module ships. All of that is written down rather than left for the next person. The round-trip test marshals the new shapes rather than only compiling against them, since a field added without its descriptor compiles and then silently drops on the wire.
I had this wrong. temporalio/api is protos-only by design and temporalio/api-go is the Go module, carrying api as a submodule and regenerating with make update-proto. My earlier note claimed a clean clone of api "cannot generate", which is backwards: it is not meant to. So the previous fork was a hand-assembled module built by copying pieces out of the published artifact. It compiled, but it could not be regenerated and would have misled anyone who opened it. Replaced with the supported shape: a proto-only branch on moedash/api, and moedash/api-go regenerated from it through its own make targets with the submodule pointed at that branch. The proto diff is now 65 lines with no vendored Go alongside it.
The stream is a co-located subcomponent of the workflow, so its frontier advances in the workflow task's own commit. The test asserts what that buys: publishing writes no history event at all. A command handler runs under the state lock with no context for I/O, so it cannot write the log itself. It stages the write, the task handler collects it, and RespondWorkflowTaskCompleted flushes before the commit that makes the offsets visible. A crash between the two leaves nodes at or past the frontier, which no reader can observe, and the retried task stages them again. Reusing a transaction id across workflow task attempts is safe here in a way it is not for an external producer: replay is deterministic, so a reused id lands identical bytes at the same node, which the store treats as an idempotent overwrite. The hazard the external path guards against is differing content under an equal id, which replay cannot produce. Three things this turned up. The stream package had to split, because the service half imports the history service and the workflow library cannot import that; the component now lives in chasm/lib/stream and the service in chasm/lib/stream/service. The command attribute validator rejects unknown command types by design, to make new commands declare whether they close the workflow, so this one is listed as non-closing. And a visibility component must be an immediate child of a CHASM root, so an attached stream carries none. An attached stream is not yet readable through the stream API: it has no standalone id to route on, and reaching it needs a path-addressed component reference that CHASM does not expose. The test asserts that gap rather than tolerating it, so it will fail and say so when it closes.
The poll response is built once per delivery, so it cannot carry slices for the prior tasks a cache miss replays. The Go SDK builds its per-task message index from History on the replay path and never does I/O inside the replay loop, so the bytes have to be reassembled server-side on the History read rather than fetched by the worker.
Both pairs keep a second copy of the message so matching can forward a task without deserializing it, and the invariant that every field number means the same thing on both sides lives only in a proto comment. A mismatch would stay invisible until sendRawHistoryBetweenInternalServices is enabled.
The field has to exist on four messages, because both the history and the matching response keep a wire-compatible twin for the raw-history path, and then be copied by hand at each of the three hops. The guard test covers the proto halves; the copies are still hand-written.
The cursor sits under the consuming workflow rather than on the stream so that folding in a delivered range commits with the event recording it. On the stream it would be a cross-execution write, and a crash between the two would redeliver or skip. An empty range still counts as pending, because a task that observed nothing is a fact replay has to reproduce.
A failed attempt never commits, so the stream's last committed id does not move, and the next attempt anchored on the same completed event id wrote the same node under the same transaction id. Storage keys a node by that pair, so the rows collapse and the survivor is whichever reached the database last rather than whichever attempt committed.
RespondWorkflowTaskCompleted can return the next workflow task inline, and every current SDK asks it to. That path builds its response from its own handler, which neither asked for a range nor copied the field, so a subscribed workflow got an empty task whenever one was returned that way. Subscription resolution now waits until the task is known to have succeeded. A failed command returns no error, so the durable RegisterConsumer write went ahead and then the workflow's own side rolled back, leaving a pin on another execution's stream with no cursor behind it. Publishing checks the batch against the payload size limit. Unchecked it failed in the flush instead, which the worker replays and re-issues forever.
A sticky task's history starts at the previous task's completion, and that event carries the range the task already consumed. Reassembly ran over it and attached the payloads again, so the workflow read the same messages twice on every task boundary.
The server no-opped a second subscribe to the same stream and wrote nothing. Commands are matched against command-generated events by position, so the machine that issued it stayed unmatched and every later command went to the wrong event. It registers nothing now but still records, reporting where the cursor actually is. A redelivered range is handed over exactly as staged. The re-read recomputed the byte cap, so a short read gave the worker less than the completion was about to record. HasPendingStreamData runs on every transaction close for every workflow, so it resolves the root component once instead of asking whether it exists and then asking for it.
Event sizes move by a byte or two on varint widths, so demanding the two payload-size arms match exactly passed on luck. A leak would cost six figures across a hundred batches, so the bound is set well below that and well above the noise. The demo binary was committed by accident and is ignored now.
moedash
force-pushed
the
moe/AI-198-server-side-streams
branch
from
August 28, 2026 20:37
7e9c756 to
628d509
Compare
A stream a workflow owns is a subcomponent of that execution, so it has no id of its own and the id-addressed read could not reach it. Publishing from a workflow was therefore write-only from anywhere else, which rules out the case the feature exists for: an agent emits tokens and a UI reads them. It is addressed by its owner and its name instead, and routed on the owner so the call lands on the shard that holds both the frontier and the log. A stream the workflow has not published to yet reads as an empty one rather than as an error, because a reader arriving before the first event is ordinary and a parked reader has to wake when the stream appears.
A workflow brackets a turn, but the tokens come from the activity making the model call, and an activity has no Workflow Task to ride. Both producers now write one ordered log: the workflow inside its own commit, at no transition cost, and everything else in a transition of its own per batch. Creating the stream is what the first append from outside has to pay for, because the collection id and bucket size it writes under are decided then. Only the first one pays it.
Nothing can be added to a stream inside a closed execution, from the workflow's own task or from any other producer, so a reader parked on one was waiting for a message that could never arrive. It is now told the stream is finished, and it can still read everything published before the end. A message also carries the offset it sits at. A topic filter leaves gaps in the sequence, so a reader that resumes between messages cannot count its way back to a position.
A stream log takes writes from a workflow task and from producers outside it, and the two drew transaction ids from unrelated sequences. The store resolves a contested node by keeping the highest id it has seen and dropping everything below, so two sequences on one log lose data rather than order it: a node left by a write that never committed outranks every later write numbered below it, and the reader is not told, it just stops seeing new messages. Both producers now number from the shard. Ids only increase, so an uncommitted node is superseded rather than dominant, and two attempts of one workflow task can no longer land on the same key, which the design required from the start and the event-id derivation could not give. The staged flush is skipped once the task has failed, so a task that will not advance the frontier writes no bytes at all. The renumbering against already-read state is gone from the external paths: that state can be stale by commit time, so it protected nothing, and two writers that both took it would collide. Found by an outside review. The suite could not see it because SQLite resolves a repeated key with REPLACE, where MySQL and Postgres fail the write.
moedash
force-pushed
the
moe/AI-198-server-side-streams
branch
from
September 2, 2026 05:29
f7e6525 to
0e74266
Compare
Two defects an outside review found, unrelated except that both were invisible to the tests. A stream's log rows carried a cleanup tag the history scavenger read as a workflow identity. It looked the namespace up, found none, took the branch for garbage and deleted it. On by default, sixty days old by default, so a live stream would lose everything past that. Branches that are not an execution's history now say so, and the scavenger leaves them to whoever wrote them. The subscribe event was written in the flush, which runs after every command, so it landed behind the events of commands issued later. Every SDK matches issued commands against their events by position, so a workflow that subscribed and then did anything else in the same task would fail its first replay. The event is now written where the command is, and the flush fills in the start offset it could not know yet. Only the tests subscribing last kept this hidden.
The frontier push resolves each consumer through the local shard controller, so a consumer whose shard lives on another host fails every time. Its known head never advances, delivery clips to it, and the workflow simply never receives anything. That was logged at warning level per consumer and then dropped, which is why a single-host test cluster showed nothing. Retrying does not make it work across hosts. Routing these through the history client the way signals are is the actual fix, and that is a larger change. This makes the failure visible and keeps one unreachable consumer from stopping the others in the same pass.
Registering a consumer pin and telling a consumer the frontier moved both resolved refs through the local shard controller, which refuses a shard this host does not own. Both are now RPCs on the stream service, routed on the execution they act upon, so they land wherever it lives. The pin still goes on before the cursor, because the routed call returns before the local update runs, so the ordering that guarantee rests on is unchanged. The third step cannot be done this way. A subscribe issued from workflow code registers its pin inside the workflow task's own transaction, and a cross-shard call cannot live there. Moving it out makes the pin asynchronous and gives up the ordering. That is a design question rather than plumbing and is written up beside the design documents.
Per-partition cost was the biggest named risk to a Temporal-owned stream and had never been run. It has been now, on the schema a dedicated facet would use. Cassandra is not the obstacle it was feared to be: a 100k-token session is 5.84 MB, well inside guidance. Row count binds before size does, so buckets should be sized against the unbatched case, around 10,000 offsets rather than the 100,000 in use. It also found a gap that applies to the current substrate just as much: a bucket has an offset budget and no byte budget, so a stream of large batches can build a partition twice the guidance out of cells that are themselves too big. A bucket has to roll on whichever budget is reached first. And it measured a reason to move that is not about correctness. Keying a row by the offset its batch starts at positions a read in one row. The node-id mapping cannot, so it reads back up to a full batch of rows before the target on every poll, for every reader.
The log lived in history_node, which was the fastest substrate to prototype on and the wrong one to keep. Its contested-node rule is written for one writer per branch, its rows look like workflow history to anything that inspects them, and its replication path runs through version histories a stream has none of. Three of the six blocking defects an outside review found came from that choice. A row is now keyed by the offset its batch starts at. That single change is what the substrate is for: a write is idempotent, a retry addresses the row it wrote before, and there is nothing for an uncommitted write to outrank. The frontier the stream component commits is the only thing deciding what a reader may see, which is what the design always said and could not previously rely on. It also positions in one row. Finding the batch holding an arbitrary offset is an indexed lookup rather than reading a full batch of rows backwards on every poll, which is measured in streaming-cassandra-partitioning.md. SQLite, MySQL, Postgres and Cassandra all implement it, and the Cassandra DDL is validated against a running server. Only SQLite is exercised by tests here. The whole stream functional suite passes unchanged, which is the point: this moves where the bytes live and nothing else. The transaction id is now vestigial and still threaded through the component. Removing it is a follow-up, kept separate so this change stays reviewable.
An offset-keyed row is idempotent, so there is nothing for a second id to order. What the id used to guard against, a writer working from a frontier that has since moved, is what ExpectedOffset checks, and that check can see the committed state where the id could not. Gone from the component, its state, the command handler options and both external append paths. The test that asserted the id must advance goes with it; the case it protected is covered by the expected-offset test beside it.
The floor waited for the slowest reader so that a consumer could always re-read a range its history recorded. Nothing released it when that consumer finished, so a stream with a cap kept everything for as long as a consumer had ever been registered. It was not protecting readers, it was disabling the cap. Registering a consumer now says who to wake and nothing more, and truncation applies the cap it was asked for. A consumer that falls behind is told: the polling path already refused a read below the base, and delivery now does too. Delivery used not to, which was the real hazard, because it would have handed the workflow whatever survived and left a hole nothing could see. This also removes the reason subscribing from workflow code had to reach another shard while holding the execution lock. That was the last of the three cross-execution steps and the one that could not be routed, so it stops being a problem rather than being solved with machinery. Six tests asserted the old floor. They assert the new behaviour instead, since what changed is a guarantee and not an implementation.
The component's state already replicates: a stream's frontier and cursors are CHASM nodes and ride the sync-versioned-transition message. Only the log rows do not, and on the old substrate there was nowhere to put them, because replication reads history through version histories a stream had none of. Keying a row by the offset its batch starts at changes that. Applying a row is idempotent, rows are independent, and a row says which offsets it covers, so the payload can be shipped rather than fetched back the way history events are. Carrying the rows in the message that already carries the frontier removes the ordering question instead of answering it. Split them and the frontier can arrive first, leaving a standby holding offsets whose bytes never landed, found at failover. Not built. What needs deciding first is bigger than the mechanism: a size cap on the most important message in the replication path, and what happens to a stream written from both sides of a failover, which idempotent-by-offset turns into silent last-writer-wins.
The mechanism said to carry the rows with the frontier and left unsaid how the sender knows which rows those are. The state delta is versioned and the rows are not, so it does not follow from what is already there. Two ways, and the choice turns on something other than the obvious. Having the receiver report the offset it holds through needs a back-channel that does not exist, but it makes the byte cap free: a message that cannot carry the range carries a prefix, and the next watermark resumes there. Versioning each range as a child node needs no protocol change and costs a node per batch, which for a token stream is the state this design keeps small everywhere else, and it still needs continuation built separately.
The note designed a private replication protocol for the stream log. The reason it needed one is that CHASM has no way for a component to own data it does not store inline, so the framework does not know the bytes exist and cannot replicate, reclaim or account for them. History solved this bespoke with branch tokens. Asking for the general version is the right order, since building the private one first makes it harder to ask. Reframed on the assumption that the store is external and shared rather than the execution database. That assumption removes any chance of the framework reaching the bytes through an existing persistence path, and it puts the current interface in the wrong place: the three methods sit on ExecutionStore, which presumes the log is shard-local.
The baseline signalled the workflow done before waiting for consumers to drain, so consumers polled a completing execution for up to fifteen seconds and counted their own doomed polls as rejections and their client backoff as delivery latency. The comment above it already described the correct order. Rejections fall to 1 per cell on both halves with the order fixed. The native half assigned a constant zero to the history columns under a comment claiming a measured zero was the point. That workload has no workflow at all, so there is nothing to measure, which is a different claim and now renders as such.
The sink lives in the worker process, so it does not rewind when a Workflow task replays, and it was stamping an observation per replay rather than per token. Only the Option 5 half was replaying, so only its counts drifted, but the sink was equally wrong on both sides. Counted by token identity rather than by an SDK replay flag. The token carries its own send time, so it is unique and the check is exact. A replay flag would not work here, since a sticky cache hit is reported as a replay and guarding on it would drop live observations instead of duplicate ones.
CHASM event storage with replication and lifecycle is planned for H2 and owned elsewhere, so the request this file made is withdrawn. Nothing here should be built to substitute for it. Also recorded the prior art, which matters more than the answer. A CHASM streaming component already exists on seandan/streaming with the same component path, and its chasm.Map of payload per message hits the mutable state ceiling that a year-old note on it already called out. The same note enumerated the three ways out, including the two this prototype arrived at on its own. Not having read the existing discussion first is what made that duplicate work.
Both halves rebuilt in release. The old numbers ran on a debug sdk-core on both sides, which was never stated and is most of why Option 7 showed a 6 to 8 second tail it does not have. Three runs each, because the first two Option 5 runs disagreed by more than 2x on p50 and history events, so one run per side was not defensible. Option 5 delivery is paced by workflow task scheduling and varies with it; Option 7 streams into a task it holds open, which Core warns about at 8 seconds per task, and its latency barely varies. Exactly-once holds on both sides in all six runs, 800 observed for 800 published, which closes the 813 question with the sink counting correctly. Also fixed the Redis scrape, which shelled into a container that need not exist and returned an empty dict on failure, reporting the cost that moved out of Temporal as zero.
The batches are now data nodes keyed by the offset each one starts at, so the payload and the frontier commit in one transaction and replication comes from UpdatedChasmNodes rather than needing a protocol of its own. That collapses a lot. The append no longer previews itself against a detached copy to work out what to write, because there is nothing separate to write. The staging and flush between the command handler and the commit are gone, and so are the buckets: they bounded a storage partition, and mutable state is not one. Deleting a stream now takes its payload with it instead of sweeping buckets first. Path C delivery is not converted yet. It reads a stream in another execution straight from the store, which is no longer written, so in-workflow consumption returns nothing until that path is routed.
It existed so N readers at the tail cost N copies rather than N range scans against the database. The payload is component state now, so a read never reaches the database in the first place and the cache was written on every append and consulted by nothing.
Delivery and replay reassembly both read a stream's payload where it now lives. An owned stream is read from the consumer's own component, and during delivery that is the one already loaded, which is the only safe order while its task is being built. Replay runs after the execution lock is released, so it reads both kinds back through the engine. Both resolve through the local shard controller, so a stream on a shard this host does not own is not reachable. That is the same limit the other cross-execution steps had before they were routed, and it is the next thing this path needs.
Persistence cost per message halved, because an append is one mutable state write rather than a log write plus a frontier update, and the slope across subscriber count is now flat rather than nearly flat. Two things the three runs show that one run would not. Option 7 stalls for about eight seconds in one run of three, so its tail is intermittent rather than absent; a debug build made it show every time and that is why it looked structural. And Option 5's latency and its history cost are the same dial: across the runs the median latency and the event count move inversely and their product is roughly constant, so the spread is the trade moving rather than noise.
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
1 participant
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 adds server-side streams: a durable, offset-addressed log that lives beside Workflow
History instead of inside it.
An append schedules no Workflow Task. Each reader holds its own cursor, so a reader costs the
writer nothing. Payloads never enter History. A workflow can publish to a stream, consume one,
or both, and a process outside Temporal can do the same over gRPC.
Today the same job needs Signals in and a polling Update out. That batches at seconds, grows
against the 50MB history cap, tops out at 10 subscribers, and becomes unreadable once the
workflow closes.
How it works
A stream gets its own branch in the existing
history_nodetable, so no new storage engine isneeded. Offsets map to node ids by bucket arithmetic. The transaction-id chain the store already
maintains is what drops a stale node after a retry, and reads clip to
HeadOffset, so bytes aredurable before the frontier makes them visible.
Consuming inside a workflow records only the offset range on
WorkflowTaskCompleted. Replayre-reads the same range from the stream, so the data stays out of History but the run stays
deterministic.
Cost
TestStreamPublishHistoryCostmeasures this. It runs a workflow whose single Workflow Taskcarries N publish commands, then differences against an N=0 run so the fixed events cancel.
The event is a fixed 41 bytes. A hundredfold increase in payload size does not move it, which is
the claim that bodies stay out of History, measured rather than read off the code.
Batching is free, because the event is per call. Against Signals that is 2.7x cheaper at 20-byte
messages and 51x at 2KB, and the gap grows with payload size.
The binding limit is the event count, not the size: 51,200 events is the
limit.historyCount.errorceiling, reached while using 2MB of the 50MB size budget. So a workflow is bounded at 51,200
publish calls with no bound on the messages inside them.
Layout
chasm/lib/stream/holds the component, the log helpers and the stream service.chasm/lib/workflow/stream_commands.gohandlesAddStreamMessagesandSubscribeStream.service/history/api/recordworkflowtaskstarted/stream_slices.godelivers ranges andre-supplies them on replay.
service/history/api/respondworkflowtaskcompleted/stream_appends.goflushes staged log writesand resolves subscriptions before the commit.
Three companion branches carry the protos and the SDK support:
moedash/api,moedash/api-go,moedash/sdk-rustandmoedash/sdk-python.What is not done
limit.historyMaxPageSizeevents replays without its data. This needs fetch-on-demand rather than riding the response.
Fixing it means tagging on
WorkflowTaskStartedrather thanWorkflowTaskCompleted, across theserver and
sdk-core.max_itemsdoes notenforce while such a consumer exists.