samples: workflow_streams: drop temp-file resume offset; add stats column

jssmith · jssmith · commit 553bfdbcd81b · 2026-05-02T17:53:21.000-07:00
The reconnecting-subscriber demo previously persisted its resume offset
to a temp file between phases. Inside one process that's theatrical:
the disconnect/reconnect shape comes from creating a fresh Client +
WorkflowStreamClient with from_offset=N, not from where N happens to be
stored. Replace the file with a local int and a comment about durable
storage in production (a DB row keyed by user_id/run_id, etc.).

Restructure output around a stats column so the demo conveys what's
happening to the stream at all times, not just between phases. A
background poller calls WorkflowStreamClient.get_offset() throughout
and emits a heartbeat line once a second; every emit prints current
proc/avail/pend in a left column followed by the phase or event
message. Watching pend grow during the disconnect window and shrink
again as phase 2 catches up is the demo's core point.
diff --git a/workflow_streams/README.md b/workflow_streams/README.md
@@ -130,22 +130,27 @@ Expected output on the basic publisher side:
 workflow result: charge-order-1
 ```
 
-Expected output on the reconnecting subscriber side (note the offsets
-are continuous across the disconnect — no events lost, none duplicated):
+Expected output on the reconnecting subscriber side. Each line carries
+a stats column on the left (`proc`, `avail`, `pend`) and a phase /
+event message on the right; a background poller emits a `·` heartbeat
+once a second. Offsets are continuous across the disconnect — no
+events lost, none duplicated:
 
 ```
-[phase 1] connecting and reading first few events
-  offset= 0  stage=validating
-  offset= 1  stage=loading data
-[phase 1] persisted resume offset=2 -> /tmp/...; disconnecting
-
-[phase 2] reconnecting and resuming from persisted offset
-  offset= 2  stage=transforming
-  offset= 3  stage=writing output
-  offset= 4  stage=verifying
-  offset= 5  stage=complete
-
-workflow result: pipeline workflow-stream-pipeline-... done
+proc= 0  avail= 0  pend= 0     │ started workflow-stream-pipeline-...
+proc= 0  avail= 1  pend= 1     │ [phase 1] connecting
+proc= 1  avail= 1  pend= 0     │   offset= 0  stage=validating
+proc= 2  avail= 2  pend= 0     │   offset= 1  stage=loading data
+proc= 2  avail= 2  pend= 0     │ [phase 1] disconnecting
+proc= 2  avail= 3  pend= 1     │ ·
+proc= 2  avail= 3  pend= 1     │ ·
+proc= 2  avail= 4  pend= 2     │ ·
+proc= 2  avail= 4  pend= 2     │ [phase 2] reconnecting
+proc= 3  avail= 4  pend= 1     │   offset= 2  stage=transforming
+proc= 4  avail= 4  pend= 0     │   offset= 3  stage=writing output
+proc= 5  avail= 5  pend= 0     │   offset= 4  stage=verifying
+proc= 6  avail= 6  pend= 0     │   offset= 5  stage=complete
+proc= 6  avail= 6  pend= 0     │ workflow result: pipeline ... done
 ```
 
 ## Notes
diff --git a/workflow_streams/run_reconnecting_subscriber.py b/workflow_streams/run_reconnecting_subscriber.py
@@ -1,4 +1,4 @@
-"""Reconnecting subscriber: persist offset, disconnect, resume.
+"""Reconnecting subscriber: read a few events, disconnect, resume.
 
 Demonstrates the central Workflow Streams use case: a consumer can
 disappear mid-stream — page refresh, server restart, laptop closed —
@@ -8,7 +8,15 @@
 
 The script runs the pattern in two phases inside one process to keep
 the demo short. The same code shape works across actual process
-restarts because the resume offset is persisted to disk between phases.
+restarts because the resume offset is durable in the workflow, not in
+the consumer.
+
+Output is one line per emit, with current stream stats in a left column
+and a phase / event message in a right column. A background poller
+calls ``WorkflowStreamClient.get_offset()`` for the whole demo and
+emits a heartbeat line once a second so you can watch ``pending``
+(``available - processed``) grow while the consumer is disconnected
+and shrink as phase 2 catches up.
 
 Run the worker first (``uv run workflow_streams/run_worker.py``), then::
 
@@ -18,9 +26,8 @@
 from __future__ import annotations
 
 import asyncio
-import tempfile
 import uuid
-from pathlib import Path
+from dataclasses import dataclass
 
 from temporalio.client import Client
 from temporalio.contrib.workflow_streams import WorkflowStreamClient
@@ -37,6 +44,36 @@
 # Picked small enough that the workflow is still running after.
 PHASE_1_EVENTS = 2
 
+# How long to stay disconnected.
+DISCONNECT_SECONDS = 3.0
+
+# Background poller cadence. The poller refreshes state.available this
+# often and emits a heartbeat line once per HEARTBEAT_SECONDS.
+POLL_INTERVAL_SECONDS = 0.25
+HEARTBEAT_SECONDS = 1.0
+
+# Width of the stats column. Picked to fit the longest stats string.
+LEFT_WIDTH = 30
+
+
+@dataclass
+class State:
+    processed: int = 0
+    available: int = 0
+
+    @property
+    def pending(self) -> int:
+        return max(0, self.available - self.processed)
+
+
+def emit(state: State, message: str) -> None:
+    left = (
+        f"proc={state.processed:>2}  "
+        f"avail={state.available:>2}  "
+        f"pend={state.pending:>2}"
+    )
+    print(f"{left:<{LEFT_WIDTH}}│ {message}", flush=True)
+
 
 async def main() -> None:
     client = await Client.connect("localhost:7233")
@@ -49,58 +86,81 @@ async def main() -> None:
         task_queue=TASK_QUEUE,
     )
 
-    # Where the consumer remembers its position. In a real BFF or UI
-    # backend this would be a database row keyed by (user_id, run_id);
-    # a temp file keeps the sample self-contained.
-    offset_path = Path(tempfile.gettempdir()) / f"{workflow_id}.offset"
-
-    # ---- Phase 1: connect, read a couple of events, persist offset, disconnect.
-    print("[phase 1] connecting and reading first few events")
+    # In a real BFF the resume offset lives in durable storage keyed by
+    # (user_id, run_id) — a database row, a Redis key, etc. For an
+    # in-process demo a State.processed attribute works the same way.
+    state = State()
     stream = WorkflowStreamClient.create(client, workflow_id)
-    seen = 0
-    next_offset = 0
-    async for item in stream.subscribe([TOPIC_STATUS], result_type=StageEvent):
-        print(f"  offset={item.offset:2d}  stage={item.data.stage}")
-        # Persist *one past* the offset just consumed. On resume we want
-        # the *next* unseen event, not the one we already showed.
-        next_offset = item.offset + 1
-        offset_path.write_text(str(next_offset))
-        seen += 1
-        if seen >= PHASE_1_EVENTS:
-            break
-
-    print(
-        f"[phase 1] persisted resume offset={next_offset} -> {offset_path}; disconnecting\n"
-    )
-    # The async for loop exits the subscribe() iterator. Any background
-    # poll Update is cancelled. The workflow keeps running in the
-    # background, accumulating events into its log.
-    await asyncio.sleep(3)  # let the workflow publish more in our absence
-
-    # ---- Phase 2: reconnect, read persisted offset, resume from there.
-    print("[phase 2] reconnecting and resuming from persisted offset")
-    resume_from = int(offset_path.read_text())
-    # A brand-new client and stream object — same shape as a different
-    # process picking up where the first one left off.
-    client2 = await Client.connect("localhost:7233")
-    stream2 = WorkflowStreamClient.create(client2, workflow_id)
-    async for item in stream2.subscribe(
-        [TOPIC_STATUS],
-        from_offset=resume_from,
-        result_type=StageEvent,
-    ):
-        print(f"  offset={item.offset:2d}  stage={item.data.stage}")
-        # Continue persisting after each event so a second crash here
-        # would also resume cleanly.
-        offset_path.write_text(str(item.offset + 1))
-        if item.data.stage == "complete":
-            break
-
-    result = await handle.result()
-    print(f"\nworkflow result: {result}")
-    # Clean up the offset file; in a real consumer you'd retain it as
-    # long as the user might reconnect.
-    offset_path.unlink(missing_ok=True)
+    emit(state, f"started {workflow_id}")
+
+    stop = asyncio.Event()
+
+    async def poller() -> None:
+        """Refresh state.available; emit a heartbeat line once a second."""
+        loop = asyncio.get_running_loop()
+        last_emit = loop.time()
+        while not stop.is_set():
+            try:
+                state.available = await stream.get_offset()
+            except Exception:
+                pass
+            now = loop.time()
+            if now - last_emit >= HEARTBEAT_SECONDS:
+                emit(state, "·")
+                last_emit = now
+            try:
+                await asyncio.wait_for(
+                    stop.wait(), timeout=POLL_INTERVAL_SECONDS
+                )
+            except asyncio.TimeoutError:
+                pass
+
+    poller_task = asyncio.create_task(poller())
+    try:
+        # ---- Phase 1: connect, read a couple of events, "disconnect".
+        emit(state, "[phase 1] connecting")
+        seen = 0
+        async for item in stream.subscribe(
+            [TOPIC_STATUS], result_type=StageEvent
+        ):
+            # Remember *one past* the offset just consumed: on resume we
+            # want the next unseen event, not the one we already showed.
+            state.processed = item.offset + 1
+            emit(state, f"  offset={item.offset:2d}  stage={item.data.stage}")
+            seen += 1
+            if seen >= PHASE_1_EVENTS:
+                break
+        emit(state, "[phase 1] disconnecting")
+
+        # ---- Disconnect window: nobody reads. The workflow keeps
+        # publishing — `pend` grows on the heartbeat lines as the offset
+        # advances past `processed`.
+        await asyncio.sleep(DISCONNECT_SECONDS)
+
+        # ---- Phase 2: brand-new client + stream, resume from saved
+        # offset. Same shape as a different process picking up where the
+        # first one left off.
+        emit(state, "[phase 2] reconnecting")
+        client2 = await Client.connect("localhost:7233")
+        stream2 = WorkflowStreamClient.create(client2, workflow_id)
+        async for item in stream2.subscribe(
+            [TOPIC_STATUS],
+            from_offset=state.processed,
+            result_type=StageEvent,
+        ):
+            state.processed = item.offset + 1
+            emit(state, f"  offset={item.offset:2d}  stage={item.data.stage}")
+            if item.data.stage == "complete":
+                break
+
+        result = await handle.result()
+        emit(state, f"workflow result: {result}")
+    finally:
+        stop.set()
+        try:
+            await poller_task
+        except asyncio.CancelledError:
+            pass
 
 
 if __name__ == "__main__":
