diff --git a/tests/contrib/streams/test_live_activity_owner.py b/tests/contrib/streams/test_live_activity_owner.py new file mode 100644 index 000000000..123f9bb35 --- /dev/null +++ b/tests/contrib/streams/test_live_activity_owner.py @@ -0,0 +1,327 @@ +"""Streams that a standalone Activity owns, against a server that serves them. + +A standalone Activity owns a stream through its run. Each attempt writes as +a new producer attempt, the Activity closes the stream itself, and when the +run ends with no close, the server's notifier closes the stream with the +run's outcome. The stock dev server has no stream notifier, so every case +here skips there. ``scripts/streams_live_lane.sh`` runs them on a server +that has it, on memory, standalone Redis and a Redis Cluster. +""" + +from __future__ import annotations + +import asyncio +import uuid +from datetime import timedelta +from typing import Any + +import pytest + +from temporalio import activity, workflow +from temporalio.api.workflowservice.v1 import ResetActivityExecutionRequest +from temporalio.bridge.proto.streams import ReadRequest +from temporalio.client import Client, WorkflowFailureError +from temporalio.common import RetryPolicy +from temporalio.contrib.streams import ( + BEGINNING, + Cursor, + RecordKind, + StreamRef, + activity_handle, + stream_handle, + topic, + track_supersession, + workflow_reader, +) +from temporalio.contrib.streams._plugin import store_for_client +from temporalio.contrib.streams._ref import stream_address +from temporalio.exceptions import ApplicationError, TerminatedError +from tests.contrib.streams._support import connect_with, new_workflow_id +from tests.contrib.streams.test_live_notifications import ( + skip_without_notifier, + store, +) +from tests.contrib.streams.test_outside_handles import Waits, running_owner +from tests.helpers import new_worker + +TOKENS = topic("tokens", str) +# The lane checks owners every 2 seconds and debounces each stream's +# deliveries, so a close by the owner check lands in a few seconds. +OWNER_CHECK = 30.0 + + +@pytest.fixture(params=["memory", "redis", "cluster"]) +async def notifying(request: pytest.FixtureRequest, client: Client) -> Client: + """A client whose store notifies, on a server with the notifier.""" + await skip_without_notifier(client) + return await connect_with(client, store(request.param)) + + +@workflow.defn +class Consume: + """Reads an Activity's stream to its end and says what it saw.""" + + @workflow.run + async def run(self, ref: StreamRef) -> list[str]: + seen: list[str] = [] + reader = workflow_reader(ref, topic=TOKENS) + async with reader: + async for batch in reader: + for record in batch.records: + if record.kind is RecordKind.DATA: + seen.append(str(record.value)) + else: + seen.append(record.kind.name) + return seen + + +def _ref(handle: Any) -> StreamRef: + return StreamRef.for_activity(handle.id, run_id=handle.run_id, topic=TOKENS) + + +async def _read(client: Client, ref: StreamRef) -> tuple[list[str], list[str]]: + """What a client read sees, and what each ``SUPERSEDED`` withdrew.""" + seen: list[str] = [] + withdrawn: list[str] = [] + read = track_supersession(stream_handle(client, ref).read()) + async for record, gone in read: + withdrawn.extend(str(r.value) for r in gone) + assert not record.stale + seen.append( + str(record.value) if record.kind is RecordKind.DATA else record.kind.name + ) + return seen, withdrawn + + +async def _close_result(client: Client, ref: StreamRef) -> Any: + """The close result the stream runtime's read gives once the stream ended.""" + service = await store_for_client(client)._service_for(client) + response = await service.read( + ReadRequest(stream=stream_address(client.namespace, ref), after="$end") + ) + assert response.done + if not response.HasField("close_result"): + return None + return client.data_converter.payload_converter.from_payloads( + [response.close_result] + )[0] + + +# Steps 1 to 5: a retry supersedes, and both readers see the close result. + + +@activity.defn +async def produce() -> str: + stream = activity_handle() + out = stream.producer(topic=TOKENS) + if activity.info().attempt == 1: + await out.append("a1", "a2", "a3") + raise ApplicationError("retry me") + await out.append("b1", "b2") + await stream.close(topic=TOKENS, result="done") + return "ok" + + +async def test_live_both_readers_follow_a_standalone_activity_across_its_retry( + notifying: Client, +): + async with new_worker(notifying, Consume, activities=[produce]) as worker: + owner = await notifying.start_activity( + produce, + id=new_workflow_id(), + task_queue=worker.task_queue, + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy( + initial_interval=timedelta(seconds=1), maximum_attempts=2 + ), + ) + ref = _ref(owner) + reader = await notifying.start_workflow( + Consume.run, ref, id=new_workflow_id(), task_queue=worker.task_queue + ) + seen, withdrawn = await asyncio.wait_for(_read(notifying, ref), 60) + assert await owner.result() == "ok" + expected = ["a1", "a2", "a3", "SUPERSEDED", "b1", "b2"] + assert seen == expected + assert withdrawn == ["a1", "a2", "a3"] + # Python readers expose no close result, so it is read from the stream runtime. + assert await _close_result(notifying, ref) == "done" + assert await asyncio.wait_for(reader.result(), 60) == expected + + +# Step 6: a standalone Activity follows a Workflow's stream and resumes from +# its heartbeat after a retry. + +_followed: list[list[str]] = [] + + +@activity.defn +async def follow(workflow_id: str) -> int: + details = activity.info().heartbeat_details + after = Cursor(details[0]) if details else BEGINNING + seen: list[str] = [] + _followed.append(seen) + stream = stream_handle(activity.client(), workflow_id, topic=TOKENS) + async for record in stream.read(after=after): + seen.append(str(record.value)) + activity.heartbeat(record.cursor.token) + if activity.info().attempt == 1 and record.value == "p2": + raise ApplicationError("dropped mid-stream") + return len(seen) + + +async def test_live_a_standalone_activity_follows_a_stream_from_its_heartbeat( + notifying: Client, +): + _followed.clear() + async with new_worker(notifying, Waits, activities=[follow]) as worker: + workflow_id = await running_owner(notifying, worker.task_queue) + stream = stream_handle(notifying, workflow_id, topic=TOKENS) + producer = stream.producer(producer_id="p", attempt=1) + await producer.append("p1", "p2", "p3") + follower = await notifying.start_activity( + follow, + workflow_id, + id=new_workflow_id(), + task_queue=worker.task_queue, + start_to_close_timeout=timedelta(seconds=60), + retry_policy=RetryPolicy( + initial_interval=timedelta(seconds=1), maximum_attempts=2 + ), + ) + # The retry reads past the heartbeat and sleeps until the next record. + await asyncio.sleep(3) + await producer.append("p4") + await stream.close() + assert await asyncio.wait_for(follower.result(), 60) == 2 + assert _followed == [["p1", "p2"], ["p3", "p4"]] + await notifying.get_workflow_handle(workflow_id).signal(Waits.finish) + + +# Step 7: an Activity that fails for good with no close. + + +@activity.defn +async def fails_hard() -> None: + await activity_handle().producer(topic=TOKENS).append("x1") + raise ApplicationError("boom", type="Boom", non_retryable=True) + + +async def test_live_an_activity_that_fails_with_no_close_fails_the_workflow_reader( + notifying: Client, +): + async with new_worker(notifying, Consume, activities=[fails_hard]) as worker: + owner = await notifying.start_activity( + fails_hard, + id=new_workflow_id(), + task_queue=worker.task_queue, + start_to_close_timeout=timedelta(seconds=10), + ) + ref = _ref(owner) + reader = await notifying.start_workflow( + Consume.run, ref, id=new_workflow_id(), task_queue=worker.task_queue + ) + seen, _ = await asyncio.wait_for(_read(notifying, ref), OWNER_CHECK) + assert seen == ["x1"] + with pytest.raises(WorkflowFailureError) as raised: + await asyncio.wait_for(reader.result(), OWNER_CHECK) + cause = raised.value.cause + assert isinstance(cause, ApplicationError) + assert (cause.type, cause.message) == ("Boom", "boom") + + +# Step 8: a reset after attempt 2 of 3. + +_calls: dict[str, int] = {} +_second_failed: dict[str, asyncio.Event] = {} + + +@activity.defn +async def writes_until_reset() -> None: + key = activity.info().activity_id + _calls[key] = call = _calls.get(key, 0) + 1 + stream = activity_handle() + await stream.producer(topic=TOKENS).append(f"r{call}") + if call == 1: + raise ApplicationError("retry me", next_retry_delay=timedelta(seconds=1)) + if call == 2: + _second_failed[key].set() + raise ApplicationError("reset me", next_retry_delay=timedelta(minutes=5)) + await stream.close(topic=TOKENS, result="after reset") + + +async def test_live_an_attempt_after_a_reset_supersedes_with_no_divergence( + notifying: Client, +): + activity_id = new_workflow_id() + _second_failed[activity_id] = asyncio.Event() + async with new_worker( + notifying, Consume, activities=[writes_until_reset] + ) as worker: + owner = await notifying.start_activity( + writes_until_reset, + id=activity_id, + task_queue=worker.task_queue, + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy(maximum_attempts=3), + ) + ref = _ref(owner) + reader = await notifying.start_workflow( + Consume.run, ref, id=new_workflow_id(), task_queue=worker.task_queue + ) + await asyncio.wait_for(_second_failed[activity_id].wait(), 30) + await asyncio.sleep(0.5) + await notifying.workflow_service.reset_activity_execution( + ResetActivityExecutionRequest( + namespace=notifying.namespace, + activity_id=activity_id, + run_id=owner.run_id or "", + identity="lane", + request_id=str(uuid.uuid4()), + ) + ) + await asyncio.wait_for(owner.result(), 30) + expected = ["r1", "SUPERSEDED", "r2", "SUPERSEDED", "r3"] + seen, withdrawn = await asyncio.wait_for(_read(notifying, ref), 30) + assert seen == expected + assert withdrawn == ["r1", "r2"] + assert await _close_result(notifying, ref) == "after reset" + assert await asyncio.wait_for(reader.result(), 30) == expected + + +# Step 9: a terminated Activity. + + +@activity.defn +async def runs_until_terminated() -> None: + await activity_handle().producer(topic=TOKENS).append("t1") + while True: + activity.heartbeat() + await asyncio.sleep(0.2) + + +async def test_live_a_terminated_activity_fails_the_workflow_reader( + notifying: Client, +): + async with new_worker( + notifying, Consume, activities=[runs_until_terminated] + ) as worker: + owner = await notifying.start_activity( + runs_until_terminated, + id=new_workflow_id(), + task_queue=worker.task_queue, + start_to_close_timeout=timedelta(seconds=60), + heartbeat_timeout=timedelta(seconds=5), + ) + ref = _ref(owner) + reader = await notifying.start_workflow( + Consume.run, ref, id=new_workflow_id(), task_queue=worker.task_queue + ) + read = asyncio.ensure_future(_read(notifying, ref)) + await asyncio.sleep(3) + await owner.terminate(reason="stop it") + seen, _ = await asyncio.wait_for(read, OWNER_CHECK) + assert seen == ["t1"] + with pytest.raises(WorkflowFailureError) as raised: + await asyncio.wait_for(reader.result(), OWNER_CHECK) + assert isinstance(raised.value.cause, TerminatedError) diff --git a/tests/contrib/streams/test_live_notifications.py b/tests/contrib/streams/test_live_notifications.py index 35055f916..d2c96ab91 100644 --- a/tests/contrib/streams/test_live_notifications.py +++ b/tests/contrib/streams/test_live_notifications.py @@ -22,25 +22,31 @@ import pytest from temporalio import workflow +from temporalio.api.common.v1 import Payloads from temporalio.api.enums.v1 import EventType, StreamOwnerKind from temporalio.api.stream.v1 import StreamReference, StreamWake from temporalio.api.workflowservice.v1 import ( DescribeStreamNotifierRequest, DescribeStreamNotifierResponse, ) -from temporalio.client import Client +from temporalio.bridge.proto.streams import ReadRangeRequest +from temporalio.client import Client, WorkflowFailureError from temporalio.contrib.streams import ( RecordKind, + StreamPosition, StreamRefusedError, StreamStorePlugin, StreamUnsupportedError, + close_workflow_stream, stream_handle, + stream_producer, topic, workflow_reader, ) +from temporalio.contrib.streams._replay_reads import POSITIONS_TYPE, ReadStates from temporalio.contrib.streams.memory import MemoryStreams from temporalio.contrib.streams.redis import RedisStreams -from temporalio.exceptions import ApplicationError +from temporalio.exceptions import ActivityError, ApplicationError from temporalio.worker import Replayer from tests.contrib.streams._support import ( connect_with, @@ -505,3 +511,377 @@ async def _completed_tasks(handle: Any) -> int: e.event_type == EventType.EVENT_TYPE_WORKFLOW_TASK_COMPLETED for e in history.events ) + + +async def test_live_a_workflow_reader_on_an_empty_stream_wakes_on_the_first_append( + notifying: Client, +): + async with new_worker(notifying, Waits, Reads) as worker: + owner = await running_owner(notifying, worker.task_queue) + reader_id = new_workflow_id() + reader = await notifying.start_workflow( + Reads.run, args=[owner, 1], id=reader_id, task_queue=worker.task_queue + ) + await _sleeping(notifying, reader_id, owner) + await ( + stream_handle(notifying, owner, topic=TOKENS) + .producer(producer_id="p", attempt=1) + .append("only") + ) + assert await asyncio.wait_for(reader.result(), 20) == ["only"] + await notifying.get_workflow_handle(owner).signal(Waits.finish) + + +async def test_live_two_readers_wake_from_one_notification(notifying: Client): + async with new_worker(notifying, Waits, Reads) as worker: + owner = await running_owner(notifying, worker.task_queue) + ids = [new_workflow_id(), new_workflow_id()] + readers = [ + await notifying.start_workflow( + Reads.run, args=[owner, 1], id=i, task_queue=worker.task_queue + ) + for i in ids + ] + for i in ids: + await _sleeping(notifying, i, owner) + handle = stream_handle(notifying, owner, topic=TOKENS) + client_records = handle.read() + client_first = asyncio.ensure_future(client_records.__anext__()) + await asyncio.sleep(3) + await handle.producer(producer_id="p", attempt=1).append("all") + for reader in readers: + assert await asyncio.wait_for(reader.result(), 20) == ["all"] + assert (await asyncio.wait_for(client_first, 10)).value == "all" + await client_records.aclose() + notifier = await describe_notifier(notifying, owner) + assert notifier is not None and notifier.counter > 0 + await notifying.get_workflow_handle(owner).signal(Waits.finish) + + +@workflow.defn +class PublishesAndCloses: + def __init__(self) -> None: + self.how: str | None = None + + @workflow.run + async def run(self) -> None: + writer = stream_producer(TOKENS) + writer.append("a") + writer.append("b") + await workflow.wait_condition(lambda: self.how is not None) + if self.how == "error": + await close_workflow_stream( + TOKENS, error=ApplicationError("the agent gave up", type="GaveUp") + ) + else: + await close_workflow_stream(TOKENS, result="done") + await workflow.wait_condition(lambda: self.how == "end") + + @workflow.signal + def end(self, how: str) -> None: + self.how = how + + +@pytest.mark.parametrize("how", ["result", "error"]) +async def test_live_a_workflow_close_ends_both_readers_with_its_outcome( + notifying: Client, how: str +): + async with new_worker(notifying, PublishesAndCloses, Reads) as worker: + owner_id = new_workflow_id() + owner = await notifying.start_workflow( + PublishesAndCloses.run, id=owner_id, task_queue=worker.task_queue + ) + reader_id = new_workflow_id() + reader = await notifying.start_workflow( + Reads.run, args=[owner_id, 0], id=reader_id, task_queue=worker.task_queue + ) + await _sleeping(notifying, reader_id, owner_id) + client_read = asyncio.ensure_future( + _read_to_the_end(stream_handle(notifying, owner_id, topic=TOKENS)) + ) + await owner.signal(PublishesAndCloses.end, how) + records, ended_with = await asyncio.wait_for(client_read, 30) + assert [r.kind for r in records] == [ + RecordKind.DATA, + RecordKind.DATA, + RecordKind.FINISH, + ] + if how == "error": + assert isinstance(ended_with, ApplicationError) + assert ended_with.type == "GaveUp" + else: + assert ended_with is None + if how == "error": + with pytest.raises(WorkflowFailureError) as raised: + await asyncio.wait_for(reader.result(), 30) + cause = raised.value.cause + assert isinstance(cause, ApplicationError) and cause.type == "GaveUp" + else: + assert await asyncio.wait_for(reader.result(), 30) == ["a", "b", "FINISH"] + await owner.signal(PublishesAndCloses.end, "end") + await owner.result() + + +async def _read_to_the_end(handle: Any) -> tuple[list[Any], BaseException | None]: + """The records of a client read, and the error it ended with, if any.""" + records: list[Any] = [] + try: + async for record in handle.read(): + records.append(record) + except ApplicationError as error: + return records, error + return records, None + + +@workflow.defn +class ReadsAcrossContinueAsNew: + @workflow.run + async def run( + self, owner: str, seen: list[str], after: StreamPosition | None + ) -> list[str]: + reader = workflow_reader(owner, topic=TOKENS, after=after or StreamPosition("")) + async with reader: + batch = await reader.__anext__() + seen = seen + [str(r.value) for r in batch.records] + position = batch.position + if len(seen) >= 2: + return seen + workflow.continue_as_new(args=[owner, seen, position]) + + +async def test_live_a_position_resumes_the_reader_after_continue_as_new( + notifying: Client, +): + async with new_worker(notifying, Waits, ReadsAcrossContinueAsNew) as worker: + owner = await running_owner(notifying, worker.task_queue) + producer = stream_handle(notifying, owner, topic=TOKENS).producer( + producer_id="p", attempt=1 + ) + await producer.append("one") + reader = await notifying.start_workflow( + ReadsAcrossContinueAsNew.run, + args=[owner, [], None], + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + await asyncio.sleep(4) + await producer.append("two") + assert await asyncio.wait_for(reader.result(), 30) == ["one", "two"] + await notifying.get_workflow_handle(owner).signal(Waits.finish) + + +async def test_live_a_recheck_finds_a_record_whose_notification_was_lost( + client: Client, +): + # The producer's process never notifies, so only the notifier's recheck + # (streamnotifier.recheckInitial, a few seconds in the lane) wakes the reader. + reading, silent = await two_processes(client) + async with new_worker(reading, Waits, Reads) as worker: + owner = await running_owner(reading, worker.task_queue) + reader_id = new_workflow_id() + reader = await reading.start_workflow( + Reads.run, args=[owner, 1], id=reader_id, task_queue=worker.task_queue + ) + await _sleeping(reading, reader_id, owner) + await ( + stream_handle(silent, owner, topic=TOKENS) + .producer(producer_id="p", attempt=1) + .append("found") + ) + assert await asyncio.wait_for(reader.result(), 30) == ["found"] + await reading.get_workflow_handle(owner).signal(Waits.finish) + + +# Positions-only Workflow readers. + + +@workflow.defn +class ReadsFromTheStore: + """Reads with replay_from_store until the stream ends, then says what it saw.""" + + @workflow.run + async def run(self, owner: str) -> list[str]: + seen: list[str] = [] + async with workflow_reader( + owner, topic=TOKENS, replay_from_store=True + ) as reader: + async for batch in reader: + seen.extend( + str(r.value) if r.kind is RecordKind.DATA else r.kind.name + for r in batch.records + ) + return seen + + +def _positions_payloads(history: Any) -> list[Any]: + out = [] + for event in history.events: + if not event.HasField("marker_recorded_event_attributes"): + continue + for payloads in event.marker_recorded_event_attributes.details.values(): + for payload in payloads.payloads: + if payload.metadata.get("messageType") == POSITIONS_TYPE: + out.append(payload) + return out + + +async def _read_from_the_store(client: Client) -> tuple[Any, str, Any]: + """Run a positions-only reader over an owner that publishes and closes.""" + # The owner runs on a Worker with a cache. With none, an eviction can + # cancel its close's Local Activity, and the task's rerun shows readers a + # SUPERSEDED before the second FINISH. + async with ( + new_worker(client, PublishesAndCloses) as owners, + new_worker(client, ReadsFromTheStore, max_cached_workflows=0) as worker, + ): + owner_id = new_workflow_id() + owner = await client.start_workflow( + PublishesAndCloses.run, id=owner_id, task_queue=owners.task_queue + ) + reader = await client.start_workflow( + ReadsFromTheStore.run, + owner_id, + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + await asyncio.sleep(3) + await owner.signal(PublishesAndCloses.end, "result") + # With no cache, every task replays the reads before it, through the stream runtime. + assert await asyncio.wait_for(reader.result(), 30) == ["a", "b", "FINISH"] + await owner.signal(PublishesAndCloses.end, "end") + await owner.result() + return await reader.fetch_history(), owner_id, reader + + +async def test_live_a_positions_only_reader_records_ranges_and_replays_from_the_store( + notifying: Client, +): + history, _, _ = await _read_from_the_store(notifying) + positions = _positions_payloads(history) + assert positions and all(p.ByteSize() < 2000 for p in positions) + store = store_for(notifying) + store._batches.clear() + await Replayer(workflows=[ReadsFromTheStore], plugins=[store]).replay_workflow( + history + ) + + +async def test_live_a_replay_that_finds_other_records_is_nondeterminism( + notifying: Client, +): + history, _, _ = await _read_from_the_store(notifying) + store = store_for(notifying) + store._batches.clear() + for payload in _positions_payloads(history): + parts = Payloads.FromString(payload.data) + range = ReadRangeRequest.FromString(parts.payloads[0].data) + if range.stored_records: + range.records_digest = bytes(32) + parts.payloads[0].data = range.SerializeToString() + payload.data = parts.SerializeToString() + # The task fails with the nondeterminism cause. The Replayer reports a + # failure that the SDK found as a failed activation, not as Core's + # NondeterminismError. + with pytest.raises(Exception) as raised: + await Replayer(workflows=[ReadsFromTheStore], plugins=[store]).replay_workflow( + history + ) + assert "replayed other records" in str(raised.value) + assert "NonDeterministicError" in str(raised.value) + + +async def test_live_a_replay_of_a_lost_range_fails_the_task(notifying: Client): + history, owner_id, _ = await _read_from_the_store(notifying) + store = store_for(notifying) + store._batches.clear() + await store.delete_workflow_streams(notifying.namespace, owner_id) + with pytest.raises(Exception) as raised: + await Replayer(workflows=[ReadsFromTheStore], plugins=[store]).replay_workflow( + history + ) + # A lost range fails the task for a retry, not as nondeterminism. + assert "StreamReplayRangeLostError" in str(raised.value) + assert "NonDeterministicError" not in str(raised.value) + + +def _values(batch: Any) -> list[str]: + return [str(r.value) for r in batch.records if r.kind is RecordKind.DATA] + + +@workflow.defn +class ReadsThroughATimeout: + """Reads one batch, times out a read on the idle stream, then reads again.""" + + def __init__(self) -> None: + self.seen: list[str] = [] + + @workflow.run + async def run(self, owner: str) -> list[str]: + reader = workflow_reader(owner, topic=TOKENS, idle_wait=timedelta(seconds=5)) + async with reader: + self.seen.extend(_values(await anext(reader))) + try: + await asyncio.wait_for(anext(reader), 1) + except (asyncio.TimeoutError, ActivityError): + self.seen.append("timed out") + self.seen.extend(_values(await anext(reader))) + return self.seen + + @workflow.query + def progress(self) -> list[str]: + return self.seen + + +async def test_live_a_read_after_a_timed_out_read_replays_from_the_store( + notifying: Client, +): + async with new_worker(notifying, Waits, ReadsThroughATimeout) as worker: + owner = await running_owner(notifying, worker.task_queue) + stream = stream_handle(notifying, owner, topic=TOKENS) + await stream.producer(producer_id="p", attempt=1).append("a") + reader = await notifying.start_workflow( + ReadsThroughATimeout.run, + owner, + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + + async def timed_out() -> bool: + return "timed out" in await reader.query(ReadsThroughATimeout.progress) + + await until(timed_out) + # A new attempt reads as a SUPERSEDED and then "b". + await stream.producer(producer_id="p", attempt=2).append("b") + assert await asyncio.wait_for(reader.result(), 30) == ["a", "timed out", "b"] + await notifying.get_workflow_handle(owner).signal(Waits.finish) + # A Worker that never ran the reader replays it with no batch cached and + # no answer state kept. + store = store_for(notifying) + store._batches.clear() + store._read_states = ReadStates() + await Replayer(workflows=[ReadsThroughATimeout], plugins=[store]).replay_workflow( + await reader.fetch_history() + ) + + +async def test_live_a_workflow_reader_waits_for_an_owner_that_starts_later( + notifying: Client, +): + # Inside the owner start grace, the stream runtime answers an idle read with no chain + # and attaches nothing, so the reader reads again on a timer. + async with new_worker(notifying, Waits, Reads) as worker: + owner = new_workflow_id() + reader = await notifying.start_workflow( + Reads.run, + args=[owner, 1], + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + await asyncio.sleep(3) + await notifying.start_workflow( + Waits.run, id=owner, task_queue=worker.task_queue + ) + stream = stream_handle(notifying, owner, topic=TOKENS) + await stream.producer(producer_id="p", attempt=1).append("first") + assert await asyncio.wait_for(reader.result(), 30) == ["first"] + await notifying.get_workflow_handle(owner).signal(Waits.finish)