Skip to content

Commit 830ffa9

Browse files
committed
Merged N4 into N5.
2 parents 5fdb603 + 6e305dd commit 830ffa9

14 files changed

Lines changed: 438 additions & 278 deletions

File tree

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
Add experimental `temporalio.contrib.streams` so Activities and clients can append ordered records to a Workflow's stream in Redis, and outside readers can follow it from a cursor. The `temporalio-streams` crate runs the store and the reader rules in process, so every SDK reads the same records.
22
A Workflow publishes to its own stream with `stream_producer`. The Worker sends each live activation's records with `PublishWorkflowTask` before it completes the activation, so readers see them while the Workflow Task runs and nothing goes to History. If the store fails the append, the Workflow Task fails with `StreamOutputNotAppended`. A topic the store reports closed drops its records with a warning and counts `workflow_stream_output_dropped`.
33
A Workflow's producer attempt is `(prefix << 16) | retry`, where `prefix` is the WorkflowTaskStarted event id of the run's last completed Workflow Task and `retry` counts that task's reruns since, from 1. So a read reports `SUPERSEDED` only for a rerun of the same task. `SupersessionTracker` tells a reader which earlier records each `SUPERSEDED` withdraws.
4-
A Local Activity publishes to its Workflow's stream with `activity_handle`, under a producer identity that survives its reruns.
4+
A Local Activity publishes to its Workflow's stream with `stream_producer`, under a producer identity that survives its reruns. Its attempt comes from the Workflow Task facts its Workflow stamps on the call, with no server call, and a reader withdraws its records when that task never completed.
55
A read of a Workflow that Temporal doesn't know yet answers empty for `owner_start_grace` (60 seconds unless set), so a reader can start before its owner.

‎temporalio/bridge/Cargo.lock‎

Lines changed: 1 addition & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎temporalio/bridge/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,7 @@ url = "2.2"
9191
# makes it use the submodule's crates, so the bridge sees one `Connection` type.
9292
[dependencies.temporalio-streams]
9393
git = "https://github.com/moetemp/sdk-streams"
94-
rev = "a08e6f32ccdebc64a8d208ac9aa1b1549e85479c"
94+
rev = "02ef7af32246475e209980bb87e11f72923c7801"
9595
default-features = false
9696
features = ["envconfig", "redis"]
9797

‎temporalio/bridge/proto/streams/streams_pb2.py‎

Lines changed: 128 additions & 128 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎temporalio/bridge/proto/streams/streams_pb2.pyi‎

Lines changed: 25 additions & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎temporalio/bridge/proto/streams/v1/message_pb2.pyi‎

Lines changed: 7 additions & 6 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎temporalio/contrib/streams/README.md‎

Lines changed: 28 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -132,29 +132,37 @@ an identity of its own:
132132
plugin's Worker interceptor stamps the number on each call in a header, so
133133
it stays the same across reruns, and across a retry that backs off longer
134134
than `local_retry_threshold`, which gets a new Activity id.
135-
- **Attempt.** Before the first append, the producer asks the server for
136-
the run's History length `h` and the pending Workflow Task's attempt `t`.
137-
With the Local Activity's attempt `a`, it writes attempt
138-
`(h * 100_000 + t) * 10_000 + a`. Each rerun runs in a later Workflow Task
139-
attempt, so the attempt grows, and readers see `SUPERSEDED` before the
140-
rerun's records. A retry inside one attempt is written once, as for any
141-
producer.
135+
- **Attempt.** The interceptor also stamps the Workflow Task facts of the
136+
activation that schedules the call. Core takes them from History. The
137+
task attempt is `prefix << 16 | retry`, the same value the Workflow's own
138+
records use. With the Local Activity's attempt `a`, the producer writes
139+
attempt `task attempt << 14 | min(a, 16383)`. Each rerun runs in a later
140+
Workflow Task attempt, so the attempt grows, and readers see `SUPERSEDED`
141+
before the rerun's records. A retry inside one attempt is written once, as
142+
for any producer.
143+
- **Failed tasks.** The records carry the started event id of the call's
144+
Workflow Task attempt and the run's last completed tasks, as the
145+
Workflow's records do. If that task fails and its retry doesn't run the
146+
call again, the next record of the run on the topic shows that the task
147+
never completed. Readers then see `SUPERSEDED` for the call's records.
148+
149+
The attempt needs no server call, so the first record goes out at once. The
150+
stream runtime's owner check describes the Workflow, but it caches the
151+
answer per owner run, so the run's Local Activities share it.
142152

143153
Known limits:
144154

145-
- **One describe call.** Each Local Activity attempt makes one
146-
`DescribeWorkflowExecution` call before its first append, so its first
147-
record waits one round trip to the server. Later appends make no call. The
148-
stream runtime's owner check also describes the Workflow, but it caches
149-
the answer per owner run for a minute, so the run's Local Activities share
150-
it.
151-
- **Describe permission.** The Worker's identity needs permission to
152-
describe the Workflow (`DescribeWorkflowExecution` on its namespace), both
153-
for the attempt and for the stream runtime's owner check. Without it, the
154-
first append fails.
155-
- **Attempt range.** A Local Activity past attempt 9,999, or in a Workflow
156-
Task past attempt 99,999, raises `StreamUnsupportedError` on its first
157-
append.
155+
- **Old failed tasks.** Each record lists only the run's last eight
156+
completed tasks. If more than seven tasks complete before the next record
157+
on the topic, the reader can't prove the call's task failed, and its
158+
records stand.
159+
- **Rerun after a heartbeat.** If the Workflow Task completed as a
160+
heartbeat while the call ran and a later task fails, the rerun schedules
161+
the call again while it replays. It gets the first run's task facts, so
162+
its first attempt writes the same producer attempt. The same content is
163+
taken as a resend. Other content is refused with `StreamProducerError`,
164+
which fails that Local Activity attempt, and the local retry writes a
165+
higher attempt that supersedes the first run.
158166

159167
## Streams owned by a standalone Activity
160168

‎temporalio/contrib/streams/_handles.py‎

Lines changed: 24 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,9 @@
1414
runs of a chain. A standalone Activity has no scheduling Workflow. It owns a
1515
stream of its own and writes it as ``<Activity id>@<Activity run id>``. A
1616
Local Activity writes as ``__temporal_local_<call number>@<scheduling run id>``,
17-
with an attempt that grows across its reruns (see ``_local_activity``). Its
18-
attempt doesn't fit the Activity form's 32 bits, so it writes as a named
19-
producer.
17+
with an attempt that grows across its reruns and the task facts its Workflow
18+
stamped (see ``_local_activity``). Its attempt doesn't fit the Activity form's
19+
32 bits, so it writes as a named producer.
2020
2121
Any process holding a client reaches a Workflow's stream by Workflow id and
2222
writes with a producer id and attempt of its own.
@@ -131,11 +131,7 @@ def producer_id(self) -> str:
131131

132132
@property
133133
def attempt(self) -> int:
134-
"""The attempt this producer writes.
135-
136-
In a Local Activity it is 0 until the first append asks the server
137-
for it.
138-
"""
134+
"""The attempt this producer writes."""
139135
return self._attempt
140136

141137
async def append(self, *values: T) -> Cursor:
@@ -223,8 +219,6 @@ async def join(self, *, name: str, until: Cursor | None = None) -> Cursor:
223219
async def _write(
224220
self, kind: RecordKind, values: Any, link: StreamLink | None = None
225221
) -> Cursor:
226-
if self._local_call is not None:
227-
self._attempt = await self._local_call.attempt(self._handle._client)
228222
converter = self._handle._converter
229223
wires = [
230224
to_wire(
@@ -251,6 +245,13 @@ async def _write(
251245
request.named.CopyFrom(
252246
NamedProducer(producer_id=self._producer_id, attempt=self._attempt)
253247
)
248+
if self._local_call is not None:
249+
request.named.task_started_event_id = (
250+
self._local_call.task_started_event_id
251+
)
252+
request.named.completed_started_event_ids.extend(
253+
self._local_call.completed_started_event_ids
254+
)
254255
service = await self._handle._plugin._service_for(self._handle._client)
255256
response = await call(service.append(request))
256257
self._sequence += len(wires)
@@ -768,12 +769,14 @@ def producer(
768769
769770
In a Local Activity the producer id is
770771
``__temporal_local_<call number>@<scheduling run id>``, where the number counts
771-
the run's Local Activity calls in order. Before the first append the
772-
producer asks the server for the run's History length ``h`` and the
773-
pending Workflow Task's attempt ``t``, and with the Local Activity's
774-
attempt ``a`` writes attempt ``(h * 100_000 + t) * 10_000 + a``. So a
772+
the run's Local Activity calls in order. The attempt is ``task attempt
773+
<< 14 | min(a, 16383)``. The task attempt is ``prefix << 16 | retry``
774+
of the Workflow Task that scheduled the call, as the Workflow's own
775+
records write it, and ``a`` is the Local Activity's attempt. So a
775776
rerun after a Worker crash or a failed Workflow Task, and a local
776-
retry, each write a higher attempt.
777+
retry, each write a higher attempt. The records carry the task's
778+
started event id and the run's last completed tasks, so a reader
779+
withdraws them when the task never completed.
777780
778781
Every call in one attempt returns the same producer for a topic, so
779782
two parts of the Activity that each ask for it number their records
@@ -788,7 +791,7 @@ def producer(
788791
self._inner,
789792
name,
790793
f"{local_call.producer_id_prefix}@{identity.run_id}",
791-
0,
794+
local_call.attempt(identity.attempt),
792795
local_call=local_call,
793796
)
794797
else:
@@ -840,17 +843,17 @@ def activity_handle(
840843
841844
A Local Activity writes the Workflow's stream too, while the Workflow
842845
Task that runs it is still open. Its attempt is not durable, so the
843-
store plugin's Worker interceptor numbers each call for a producer id
844-
that survives a rerun, and its first append asks the server for an
845-
attempt that grows across reruns. See :meth:`ActivityStreamHandle.producer`.
846+
store plugin's Worker interceptor stamps each call with a number for a
847+
producer id that survives a rerun, and with the facts of its Workflow
848+
Task for an attempt that grows across reruns. See
849+
:meth:`ActivityStreamHandle.producer`.
846850
847851
Raises:
848852
StreamUnsupportedError: A standalone Activity got no run id from
849853
the server, or asked for ``scope="workflow"``. A Local Activity
850854
asked for ``scope="activity"``, since it has no server state
851855
that could end its stream. Or it is a Local Activity that the
852-
Worker's stream interceptor did not number. Its first append
853-
raises it when its History or attempts are too long to encode.
856+
Worker's stream interceptor did not stamp.
854857
ValueError: ``scope`` is not ``"activity"``, ``"workflow"`` or
855858
``None``, or no stream store is registered on the Worker's
856859
client.

0 commit comments

Comments
 (0)