Let a Redis workflow reader start at the tail or at the newest N.

temporalio/contrib/external_workflow_streams/__init__.py

@@ -100,6 +100,7 @@
     Offset,
     OffsetComparator,
     RecordKind,
+    StartAtTail,
     StreamRecord,
 )
 from temporalio.contrib.external_workflow_streams._wake import WakeRequest
@@ -128,6 +129,7 @@
     "ExternalStreamProducerTopic",
     "ExternalStreamSubscription",
     "ExternalStreamTopic",
+    "StartAtTail",
     "IdempotencyKey",
     "Offset",
     "OffsetComparator",

temporalio/contrib/external_workflow_streams/_api.py

@@ -37,6 +37,7 @@
 from temporalio.contrib.external_workflow_streams._record import (
     Cursor,
     Offset,
+    StartAtTail,
     StreamRecord,
 )
 from temporalio.contrib.external_workflow_streams._wake import channel_for
@@ -98,6 +99,7 @@ def register(
         stream_key: StreamKey,
         idle_timeout: timedelta,
         start_cursor: Cursor | None = None,
+        start_at_tail: StartAtTail | None = None,
     ) -> None:
         """Registers a wait with the Worker's subscription manager.
 
@@ -279,7 +281,10 @@ class ExternalStreamTopic(Generic[AnyType]):
     options: ExternalStreamOptions
 
     def subscribe(
-        self, *, start_cursor: Cursor | None = None
+        self,
+        *,
+        start_cursor: Cursor | None = None,
+        start_at_tail: StartAtTail | None = None,
     ) -> ExternalStreamSubscription[AnyType]:
         """Starts a new subscription and returns its async iterator.
 
@@ -298,6 +303,11 @@ def subscribe(
                 ``BEGINNING`` on a first execution. A boundary the Workflow
                 names is recorded in the marker's header like the restored one,
                 so it must be derived deterministically: replay names it again.
+            start_at_tail: Start at the stream's tail instead, or at its newest
+                ``last`` records. The Worker resolves the boundary against the
+                store after this Workflow Task and records it with the
+                subscription, so replay starts where the live run did rather
+                than asking the store again. Exclusive with ``start_cursor``.
         """
         state = _run_state()
         if state.runtime is None:
@@ -314,6 +324,8 @@ def subscribe(
         start: dict[str, Any] = (
             {} if start_cursor is None else {"start_cursor": start_cursor}
         )
+        if start_at_tail is not None:
+            start["start_at_tail"] = start_at_tail
         state.runtime.register(
             wait_id=wait_id,
             stream_key=stream_key,

temporalio/contrib/external_workflow_streams/_backend.py

@@ -275,6 +275,23 @@ def compare_offsets(self, left: Offset, right: Offset) -> int:
         inside a validation loop.
         """
 
+    async def tail_cursor(self, key: StreamKey, *, before_last: int = 0) -> Cursor:
+        """The boundary the newest ``before_last`` records begin after.
+
+        With ``before_last=0`` it is the boundary after the newest record, so a
+        read from it sees only what is appended after this call. ``BEGINNING``
+        when the stream holds fewer records than asked for. Resolved on the
+        Worker, never on the Workflow thread, and recorded with the
+        subscription so replay reads it from the marker instead of asking
+        again. A provider that cannot answer leaves this as it is, and a
+        subscription that asks for a tail start fails when the Worker resolves
+        it.
+        """
+        raise NotImplementedError(
+            f"{type(self).__name__} cannot resolve a start {before_last} records "
+            f"before the tail of {key}; subscribe with a cursor instead"
+        )
+
     # --- waking -------------------------------------------------------------
 
     wake_transport: WakeTransport = "auto"

temporalio/contrib/external_workflow_streams/_record.py

@@ -11,7 +11,29 @@
 from dataclasses import dataclass, field
 from typing import Final, Protocol
 
+
+@dataclass(frozen=True)
+class StartAtTail:
+    """A subscription start the Worker resolves against the store: the tail.
+
+    ``last`` is how many of the newest records the read begins with; zero
+    means the boundary after the newest record, so the read sees only what
+    is appended after it was opened. The Workflow thread cannot ask the store
+    where the tail is, so the runtime records the request and the Worker
+    resolves it before the watcher starts, writing the boundary into the
+    marker beside the subscription as it does a cursor the Workflow named.
+    """
+
+    last: int = 0
+
+    def __post_init__(self) -> None:
+        """Refuse a negative count."""
+        if self.last < 0:
+            raise ValueError(f"last must not be negative, got {self.last}")
+
+
 __all__ = [
+    "StartAtTail",
     "AFTER",
     "BEGINNING",
     "Cursor",

temporalio/contrib/external_workflow_streams/_runtime.py

@@ -88,6 +88,7 @@
     BEGINNING,
     Cursor,
     RecordKind,
+    StartAtTail,
     StreamRecord,
 )
 from temporalio.contrib.external_workflow_streams._replay import ReplayPlan
@@ -200,6 +201,14 @@ class _SubscriptionState:
     generation: int = 0
     """Increments each time this wait re-enters the blocked state."""
 
+    start_pending: bool = False
+    """The start is a tail the Worker has yet to resolve against the store.
+
+    Until it does, the cursors hold ``BEGINNING`` as a placeholder, no watcher
+    runs, and the binding stays out of every header and bindings frame, so the
+    marker never records the placeholder as where this wait began.
+    """
+
     announced: bool = False
     """Whether the *current* annotation has carried this subscription's binding.
 
@@ -291,6 +300,9 @@ def __init__(
         #: would give replay whatever the stream holds now (ADR-022).
         self._continuation = continuation
         self._subscriptions: dict[int, _SubscriptionState] = {}
+        #: Tail starts registered in the activation under way, resolved by the
+        #: Worker once it returns. See :meth:`resolve_pending_starts`.
+        self._pending_starts: dict[int, StartAtTail] = {}
         #: Where the *current* annotation begins, per wait. Captured when the
         #: annotation begins rather than when its header is first needed --
         #: lazily reading the delivery cursor would let a record delivered
@@ -604,7 +616,7 @@ def _reserve_bytes(self) -> int:
         late = {
             wait_id: self._binding(state)
             for wait_id, state in sorted(self._subscriptions.items())
-            if not state.announced
+            if not state.announced and not state.start_pending
         }
         if late and self._accumulator is not None:
             # Only with an accumulator: without one the header is about to carry
@@ -1134,13 +1146,16 @@ def register(
         stream_key: StreamKey,
         idle_timeout: timedelta | None = None,
         start_cursor: Cursor | None = None,
+        start_at_tail: StartAtTail | None = None,
     ) -> None:
         """Registers a wait with the Worker's manager. Non-blocking, no I/O.
 
         The start cursor defaults to what the predecessor Run committed for this
         ``wait_id``, so a chain resumes where it left off without the Workflow
         code saying anything about it -- and a first execution gets ``BEGINNING``
-        from the same path.
+        from the same path. A ``start_at_tail`` is resolved by the Worker after
+        this activation and recorded with the subscription; on replay the
+        recorded boundary is taken from the marker and the store is not asked.
 
         Raises:
             ExternalStreamCapacityError: This subscription set cannot be recorded
@@ -1161,7 +1176,20 @@ def register(
                 "external streams are not configured on this Worker; pass "
                 "external_stream_backend=... to the Worker"
             )
-        if start_cursor is None:
+        if start_at_tail is not None and start_cursor is not None:
+            raise ValueError(
+                "a subscription starts at a cursor or at the tail, not at both"
+            )
+        pending = False
+        if start_at_tail is not None:
+            recorded = (self._replay_bindings or {}).get(wait_id)
+            if recorded is not None:
+                # The live run resolved it and the marker carries the answer.
+                start_cursor = recorded.start_cursor
+            else:
+                pending = True
+                start_cursor = BEGINNING
+        elif start_cursor is None:
             start_cursor = self.restored_start(wait_id, stream_key.stream_name)
         if self._replay_bindings is not None:
             # A subscription made while a marker is being replayed -- which is
@@ -1179,6 +1207,7 @@ def register(
             delivery_cursor=start_cursor,
             consumption_cursor=start_cursor,
             idle_timeout=idle_timeout or self._default_idle_timeout,
+            start_pending=pending,
         )
         self._subscriptions[wait_id] = state
         # A subscription created part-way through an annotation begins at its
@@ -1194,16 +1223,57 @@ def register(
             del self._annotation_start[wait_id]
             raise
         self._update_reserve()
-        self._manager.register(
-            run_id=self._run_id,
-            wait_id=wait_id,
-            stream_key=stream_key,
-            start_cursor=start_cursor,
-        )
+        if pending:
+            assert start_at_tail is not None
+            self._pending_starts[wait_id] = start_at_tail
+        else:
+            self._manager.register(
+                run_id=self._run_id,
+                wait_id=wait_id,
+                stream_key=stream_key,
+                start_cursor=start_cursor,
+            )
         # A registration alone is replay-visible: it is what puts the stream in
         # the annotation header, without which replay cannot start.
         self._observed_this_activation = True
 
+    async def resolve_pending_starts(self) -> None:
+        """Resolves every tail start the activation just finished registered.
+
+        On the Worker's loop, between the activation and its completion. The
+        Workflow thread can do no I/O, and the boundary has to be fixed before
+        anything is delivered on the wait, so the marker records it beside the
+        subscription and replay reads it there instead of asking the store,
+        which by then holds something else.
+        """
+        if not self._pending_starts:
+            return
+        backend = self._backend
+        if backend is None:
+            raise RuntimeError(
+                "external streams are not configured on this Worker; pass "
+                "external_stream_backend=... to the Worker"
+            )
+        pending, self._pending_starts = self._pending_starts, {}
+        for wait_id, tail in pending.items():
+            state = self._subscriptions.get(wait_id)
+            if state is None or not state.start_pending:
+                continue
+            cursor = await backend.tail_cursor(state.stream_key, before_last=tail.last)
+            state.start_cursor = cursor
+            state.delivery_cursor = cursor
+            state.consumption_cursor = cursor
+            state.start_pending = False
+            self._annotation_start[wait_id] = cursor
+            self._check_annotation_capacity(wait_id)
+            self._update_reserve()
+            self._manager.register(
+                run_id=self._run_id,
+                wait_id=wait_id,
+                stream_key=state.stream_key,
+                start_cursor=cursor,
+            )
+
     def drain(self, wait_id: int, max_records: int | None = None) -> list[StreamRecord]:
         """Pops buffered records. Performs no I/O and never blocks.
 
@@ -2026,7 +2096,7 @@ def _announce_late_subscriptions(self) -> None:
         late = {
             wait_id: state
             for wait_id, state in sorted(self._subscriptions.items())
-            if not state.announced
+            if not state.announced and not state.start_pending
         }
         if not late:
             return
@@ -2053,11 +2123,13 @@ def _header(self) -> AnnotationHeader:
         wrong store for all the others.
         """
         for state in self._subscriptions.values():
-            state.announced = True
+            if not state.start_pending:
+                state.announced = True
         return AnnotationHeader(
             streams={
                 wait_id: self._binding(state)
                 for wait_id, state in sorted(self._subscriptions.items())
+                if not state.start_pending
             },
         )
 
@@ -2072,6 +2144,7 @@ def _header_preview(self) -> AnnotationHeader:
             streams={
                 wait_id: self._binding(state)
                 for wait_id, state in sorted(self._subscriptions.items())
+                if not state.start_pending
             },
         )
 

temporalio/streams/providers/redis.py

@@ -47,10 +47,13 @@
 - Cursors are ``redis:<ms>-<seq>`` and name an entry of the topic's log, so a
   cursor a workflow reader returned seeds an outside read and the other way
   round. A reader opened without a cursor starts where the chain's
-  predecessor run committed, which is the transport's own rule. An outside
-  read positions ``END`` and ``last=N`` against the log on the first step of
-  the generator, since the call itself cannot reach the store. A workflow
-  reader starts at a cursor or at the beginning.
+  predecessor run committed, which is the transport's own rule. ``END`` and
+  ``last=N`` are positioned against the log when the read starts: outside,
+  on the first step of the generator, since the call itself cannot reach the
+  store; inside a workflow, by the worker right after the Workflow Task that
+  opened the subscription, which records the entry it resolved in the marker
+  beside the subscription, so replay and a cold start read it from History
+  and never ask the log again.
 - A workflow's streams are keyed by the chain's first run, so a handle
   follows continue-as-new by construction and ``run_id`` only decides whose
   close ends a read.
@@ -134,6 +137,7 @@
     ChainKeyMismatchError,
     ExternalStreamProducer,
     Offset,
+    StartAtTail,
     StreamDirection,
     WakeNotAcknowledgedError,
     WorkflowChainKey,
@@ -907,14 +911,12 @@ def open_reader(
     ) -> ReadSource:
         _require_topic(topic)
         check_read_start(after, last)
+        subscribe = self._input.topic(topic, type=bytes).subscribe
         if last is not None or after == END:
             # The tail is where the log is when the worker looks, which the
-            # workflow thread cannot see.
-            raise StreamUnsupportedError(
-                "a workflow reader on the redis provider starts at a cursor or at "
-                "the beginning"
-            )
-        subscribe = self._input.topic(topic, type=bytes).subscribe
+            # workflow thread cannot see: the transport has the worker resolve
+            # it after this task and record the entry with the subscription.
+            return _RedisReadSource(subscribe(start_at_tail=StartAtTail(last or 0)))
         position = _position(after)
         # Without a position the transport resumes where the chain's
         # predecessor run committed; with one, that is where the wait starts

temporalio/worker/_workflow.py

@@ -650,6 +650,11 @@ async def _handle_activation(
                     raise deadlock_exc from None
 
             output_runtime = self._external_stream_runtimes.get(act.run_id)
+            if output_runtime is not None and completion.HasField("successful"):
+                # A subscription opened at a stream's tail is positioned here,
+                # off the Workflow thread and before the completion goes out, so
+                # the marker records the boundary the watcher starts from.
+                await output_runtime.resolve_pending_starts()
             if (
                 output_runtime is not None
                 and completion.HasField("successful")

tests/contrib/external_workflow_streams/test_api.py

@@ -76,10 +76,14 @@ def register(
         stream_key: StreamKey,
         idle_timeout: timedelta,
         start_cursor: Cursor | None = None,
+        start_at_tail: Any = None,
     ) -> None:
         self.registrations.append((wait_id, stream_key))
         self.idle_timeouts[wait_id] = idle_timeout
-        self.start_cursors[wait_id] = start_cursor
+        # A tail start stands in for the cursor the Worker would resolve.
+        self.start_cursors[wait_id] = (
+            start_cursor if start_at_tail is None else start_at_tail
+        )
 
     def drain(self, wait_id: int, max_records: int | None = None) -> list[StreamRecord]:
         buffered = self.buffers.get(wait_id, [])

tests/contrib/external_workflow_streams/test_registry.py

@@ -104,6 +104,7 @@ def test_supported_entry_points_are_public() -> None:
         "PrecedingWriteFailedError",
         "RecordKind",
         "RedisStreamBackend",
+        "StartAtTail",
         "StreamBackend",
         "StreamDecodeError",
         "StreamError",