samples: workflow_stream: add truncating-ticker scenario

Adds a fourth scenario for long-running workflows that need to bound
their event log: the workflow publishes events at a fixed cadence and
calls self.stream.truncate(...) periodically to keep only the most
recent entries.

The runner subscribes twice — fast and slow — to make the trade
visible: the fast subscriber sees every offset in order; the slow one
falls behind a truncation, has its iterator transparently jump forward
to the new base offset, and shows the offset gap that intermediate
events fell into. This is the model for high-volume long-running
streams: bounded log size, slow consumers may miss intermediate events
but always see the most recent state.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: jssmith

workflow_stream/README.md

@@ -54,7 +54,20 @@ This directory has two scenarios sharing one Worker.
   backend service or scheduled job pushing events into a workflow it
   didn't itself start.
 
-`run_worker.py` registers all three workflows and the activity.
+**Scenario 4 — bounded log via `truncate()`:**
+
+* `workflows/ticker_workflow.py` — a long-running workflow that
+  publishes events at a fixed cadence and calls
+  `self.stream.truncate(...)` periodically to bound log growth, keeping
+  only the most recent N entries.
+* `run_truncating_ticker.py` — runs a fast subscriber and a slow
+  subscriber side by side. The fast one keeps up and sees every offset
+  in order; the slow one sleeps between iterations, falls behind a
+  truncation, and silently jumps forward to the new base offset. The
+  output makes the trade visible: bounded log size in exchange for
+  intermediate events being invisible to slow consumers.
+
+`run_worker.py` registers all four workflows and the activity.
 
 ## Run it
 
@@ -68,6 +81,8 @@ uv run workflow_stream/run_publisher.py
 uv run workflow_stream/run_reconnecting_subscriber.py
 # or
 uv run workflow_stream/run_external_publisher.py
+# or
+uv run workflow_stream/run_truncating_ticker.py
 ```
 
 Expected output on the basic publisher side:

workflow_stream/run_truncating_ticker.py

@@ -0,0 +1,83 @@
+"""Truncating ticker: bounded log + slow vs. fast subscribers.
+
+The ``TickerWorkflow`` publishes ``count`` events at a fixed interval,
+calling ``self.stream.truncate(...)`` periodically to bound log
+growth. This script subscribes twice — once fast, once slow — and
+prints both side-by-side so the trade is visible:
+
+* The fast subscriber keeps up and sees every published offset in
+  order.
+* The slow subscriber sleeps between iterations. When a truncation
+  runs past its position, the iterator silently jumps forward to the
+  new base offset — the slow subscriber's offsets jump too, and
+  intermediate events are not visible to it.
+
+This is the bounded-log model: log size is capped, slow consumers may
+miss intermediate events, but they always see the most recent state.
+For long-running workflows pushing high event volumes this is usually
+the right trade — pair with set-semantic events where each event
+carries enough state to make missing the prior ones recoverable.
+
+Run the worker first (``uv run workflow_stream/run_worker.py``), then::
+
+    uv run workflow_stream/run_truncating_ticker.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+
+from temporalio.client import Client
+from temporalio.contrib.workflow_stream import WorkflowStreamClient
+
+from workflow_stream.shared import (
+    TASK_QUEUE,
+    TOPIC_TICK,
+    TickerInput,
+    TickEvent,
+)
+from workflow_stream.workflows.ticker_workflow import TickerWorkflow
+
+
+SLOW_SUBSCRIBER_DELAY_S = 1.5
+
+
+async def main() -> None:
+    client = await Client.connect("localhost:7233")
+
+    workflow_id = f"workflow-stream-ticker-{uuid.uuid4().hex[:8]}"
+    handle = await client.start_workflow(
+        TickerWorkflow.run,
+        TickerInput(
+            count=20,
+            keep_last=3,
+            truncate_every=5,
+            interval_ms=400,
+        ),
+        id=workflow_id,
+        task_queue=TASK_QUEUE,
+    )
+
+    stream = WorkflowStreamClient.create(client, workflow_id)
+
+    async def fast_subscriber() -> None:
+        async for item in stream.subscribe([TOPIC_TICK], result_type=TickEvent):
+            print(f"[fast]  offset={item.offset:3d}  n={item.data.n}")
+
+    async def slow_subscriber() -> None:
+        async for item in stream.subscribe([TOPIC_TICK], result_type=TickEvent):
+            print(f"[SLOW]  offset={item.offset:3d}  n={item.data.n}")
+            await asyncio.sleep(SLOW_SUBSCRIBER_DELAY_S)
+
+    # Both iterators exit normally when the workflow completes. No
+    # terminal sentinel is needed — see the doc's "When the Workflow
+    # run completes" note.
+    await asyncio.gather(fast_subscriber(), slow_subscriber())
+
+    result = await handle.result()
+    print(f"\nworkflow result: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

workflow_stream/run_worker.py

@@ -11,6 +11,7 @@
 from workflow_stream.workflows.hub_workflow import HubWorkflow
 from workflow_stream.workflows.order_workflow import OrderWorkflow
 from workflow_stream.workflows.pipeline_workflow import PipelineWorkflow
+from workflow_stream.workflows.ticker_workflow import TickerWorkflow
 
 
 async def main() -> None:
@@ -19,7 +20,7 @@ async def main() -> None:
     worker = Worker(
         client,
         task_queue=TASK_QUEUE,
-        workflows=[HubWorkflow, OrderWorkflow, PipelineWorkflow],
+        workflows=[HubWorkflow, OrderWorkflow, PipelineWorkflow, TickerWorkflow],
         activities=[charge_card],
     )
     await worker.run()

workflow_stream/shared.py

@@ -14,6 +14,7 @@
 TOPIC_STATUS = "status"
 TOPIC_PROGRESS = "progress"
 TOPIC_NEWS = "news"
+TOPIC_TICK = "tick"
 
 
 @dataclass
@@ -58,6 +59,21 @@ class NewsEvent:
     headline: str
 
 
+@dataclass
+class TickerInput:
+    count: int = 20
+    keep_last: int = 3
+    truncate_every: int = 5
+    interval_ms: int = 400
+    # Carries stream state across continue-as-new. None on a fresh start.
+    stream_state: WorkflowStreamState | None = None
+
+
+@dataclass
+class TickEvent:
+    n: int
+
+
 T = TypeVar("T")
 
 

workflow_stream/workflows/ticker_workflow.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from datetime import timedelta
+
+from temporalio import workflow
+from temporalio.contrib.workflow_stream import WorkflowStream
+
+from workflow_stream.shared import (
+    TOPIC_TICK,
+    TickEvent,
+    TickerInput,
+)
+
+
+@workflow.defn
+class TickerWorkflow:
+    """Long-running ticker that bounds its event log via ``truncate``.
+
+    Long-running workflows that publish high volumes of events would
+    otherwise grow their event log unboundedly. This workflow shows
+    the truncation pattern: every ``truncate_every`` events, drop
+    everything except the last ``keep_last`` entries by calling
+    ``self.stream.truncate(safe_offset)``.
+
+    Subscribers that fall behind a truncation jump forward to the new
+    base offset transparently (the iterator handles the
+    ``TruncatedOffset`` error internally), so consumers stay live but
+    may not see every intermediate event. That is the trade: bounded
+    log size in exchange for at-best-effort delivery to slow
+    consumers.
+
+    To compute the truncation offset the workflow tracks its own
+    published count. ``WorkflowStream`` does not expose a workflow-side
+    head-offset accessor, but the running count plus the carried
+    ``base_offset`` (in continue-as-new chains) is sufficient.
+    """
+
+    @workflow.init
+    def __init__(self, input: TickerInput) -> None:
+        self.stream = WorkflowStream(prior_state=input.stream_state)
+        # Running count of events published by THIS run. To compute a
+        # global offset, add the prior_state's base_offset (omitted
+        # here — this sample doesn't continue-as-new).
+        self._published = 0
+
+    @workflow.run
+    async def run(self, input: TickerInput) -> str:
+        for n in range(input.count):
+            self.stream.publish(TOPIC_TICK, TickEvent(n=n))
+            self._published += 1
+            await workflow.sleep(timedelta(milliseconds=input.interval_ms))
+            if (
+                self._published % input.truncate_every == 0
+                and self._published > input.keep_last
+            ):
+                # Drop everything except the last `keep_last` entries.
+                truncate_to = self._published - input.keep_last
+                self.stream.truncate(truncate_to)
+        return f"ticker emitted {self._published} events"