diff --git a/.gitignore b/.gitignore index 8cd439e05..f16269ea9 100644 --- a/.gitignore +++ b/.gitignore @@ -15,3 +15,6 @@ temporalio/bridge/temporal_sdk_bridge* tags /.claude tmpclaude-* + +# nexgen builds that scripts/gen_stream_service.py installs +/.nexgen-streams/ diff --git a/scripts/_wit/stream-service.wit b/scripts/_wit/stream-service.wit index 50fed2c50..ed94f21a2 100644 --- a/scripts/_wit/stream-service.wit +++ b/scripts/_wit/stream-service.wit @@ -12,8 +12,6 @@ world system { /// @nexus.endpoint "__temporal_system" /// @nexus.service-name "temporal.api.streamservice.v1.StreamService" /// @nexus.experimental -/// @nexus.handle "stream-handle" keys="stream" -/// @nexus.handle "stream-producer" keys="stream producer" interface stream-service { use nexus:temporal-types/model@1.0.0.{ duration, @@ -141,7 +139,7 @@ interface stream-service { /// @nexus.doc "The place of this Record in the Stream. Ordered and unique, with gaps." /// @nexus.offset "stream-offset" offset: option, - /// @nexus.doc "Empty on a SUPERSEDED, close or IDLE Record, and for the default Topic." + /// @nexus.doc "Empty on CLOSE, IDLE and SUPERSEDED Records." topic: option, producer: option, /// @nexus.doc "Set on a data Record." diff --git a/scripts/gen_stream_service.py b/scripts/gen_stream_service.py index 234f86a20..be287b764 100644 --- a/scripts/gen_stream_service.py +++ b/scripts/gen_stream_service.py @@ -2,8 +2,9 @@ The WIT and the protos are copies of the temporalio-streams crate's, in scripts/_wit and scripts/_proto. Run scripts/gen_protos.py first for the proto modules. No nexgen release has the -offset, one-way, long-poll, handle and client features yet, so NEX_GEN_STREAMS_BIN must name an -advanced build of moetemp/nexgen moe/brook/v3/n9-handles. +offset, one-way, long-poll and client features yet, so the script installs nexgen at +NEX_GEN_REVISION from moetemp/nexgen. NEX_GEN_STREAMS_BIN may name a build of that revision +instead. """ import os @@ -30,12 +31,37 @@ ) output_dir = base_dir / "temporalio" / "contrib" / "streams" / "system" +NEX_GEN_REPOSITORY = "https://github.com/moetemp/nexgen" +# The head of moe/brook/v3/n7-long-poll-loop. A branch moves, so the pin is the commit. +NEX_GEN_REVISION = "64ee48a79abe18077d727f5067a6302c45c25c9d" +nex_gen_root = base_dir / ".nexgen-streams" / NEX_GEN_REVISION + def nex_gen_command() -> list[str]: - bin_path = os.environ.get("NEX_GEN_STREAMS_BIN") - if not bin_path: - raise RuntimeError("set NEX_GEN_STREAMS_BIN to a nexgen build with --client") - return [bin_path] + if bin_path := os.environ.get("NEX_GEN_STREAMS_BIN"): + return [bin_path] + binary = nex_gen_root / "bin" / "nexgen" + if not binary.exists(): + subprocess.check_call( + [ + "cargo", + "install", + "--locked", + "nexgen", + "--git", + NEX_GEN_REPOSITORY, + "--rev", + NEX_GEN_REVISION, + "--features", + "advanced", + "--root", + str(nex_gen_root), + ], + # The git CLI honours the user's URL rewrites and credentials, which cargo's + # built-in client does not. + env={**os.environ, "CARGO_NET_GIT_FETCH_WITH_CLI": "true"}, + ) + return [str(binary)] def api_protos() -> list[Path]: @@ -99,6 +125,10 @@ def generate_stream_service() -> None: if __name__ == "__main__": - print("Generating the Stream service bindings...", file=sys.stderr) + print( + f"Generating the Stream service bindings with nexgen {NEX_GEN_REPOSITORY}" + f"@{NEX_GEN_REVISION}...", + file=sys.stderr, + ) generate_stream_service() print("Done", file=sys.stderr) diff --git a/temporalio/contrib/streams/system/client.py b/temporalio/contrib/streams/system/client.py index d76a8c93c..e8164de3a 100644 --- a/temporalio/contrib/streams/system/client.py +++ b/temporalio/contrib/streams/system/client.py @@ -5,22 +5,15 @@ import asyncio import dataclasses import datetime -import email.message -import ssl import typing -import urllib.error -import urllib.parse -import urllib.request import google.protobuf.json_format import google.protobuf.message import temporalio.api.common.v1 -import temporalio.api.streamservice.v1.stream_service_pb2 import temporalio.converter from .models import ( - AppendRecord, AppendRecordsRequest, AppendRecordsResponse, CloseStreamRequest, @@ -29,11 +22,8 @@ CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse, - Producer, ReadRecordsRequest, ReadRecordsResponse, - StreamRef, - WorkflowTaskEpoch, _AppendRecordsRequestTransferTypeConverter, _AppendRecordsResponseTransferTypeConverter, _CloseStreamRequestTransferTypeConverter, @@ -123,156 +113,6 @@ def _proto_message( return message -class HttpTransport: - """A transport that posts each call to a Nexus HTTP endpoint. - - Each call goes to ``{base_url}/{service}/{operation}``. The payload maps to - HTTP content as the Temporal server maps it: ``json/plain`` is - ``application/json``, a proto payload names its message type, and any - other payload travels whole as ``application/x-temporal-payload``. - """ - - def __init__( - self, - base_url: str, - *, - headers: dict[str, str] | None = None, - timeout: float = 90.0, - ssl_context: ssl.SSLContext | None = None, - ) -> None: - """Create an HTTP transport. - - Args: - base_url: The endpoint's base URL, before the service name. - headers: Added to every request. - timeout: Seconds to wait for each answer. Keep it longer than the - longest wait that a long poll asks for. - ssl_context: TLS settings, such as a private certificate authority. - """ - self._base_url = base_url.rstrip("/") - self._headers = dict(headers or {}) - self._timeout = timeout - self._ssl_context = ssl_context - - async def call( - self, - *, - service: str, - operation: str, - input: temporalio.api.common.v1.Payload, - ) -> temporalio.api.common.v1.Payload: - """Post one call and return its answer.""" - url = "/".join( - [ - self._base_url, - urllib.parse.quote(service, safe=""), - urllib.parse.quote(operation, safe=""), - ] - ) - content_type, body = _payload_to_http(input) - headers = dict(self._headers) - if content_type is not None: - headers["Content-Type"] = content_type - request = urllib.request.Request(url, data=body, headers=headers, method="POST") - return await asyncio.to_thread(self._send, request) - - def _send( - self, request: urllib.request.Request - ) -> temporalio.api.common.v1.Payload: - try: - with urllib.request.urlopen( - request, timeout=self._timeout, context=self._ssl_context - ) as response: - status = int(response.status) - content_type = response.headers.get("Content-Type") - body = bytes(response.read()) - except urllib.error.HTTPError as error: - detail = bytes(error.read()).decode(errors="replace") - retryable_header = error.headers.get("Nexus-Request-Retryable") - if retryable_header is not None: - retryable = retryable_header.strip().lower() == "true" - else: - retryable = error.code == 429 or error.code >= 500 - raise TransportError( - f"{request.full_url} answered {error.code}: {detail}", - retryable=retryable, - ) from error - except urllib.error.URLError as error: - raise TransportError( - f"{request.full_url} failed: {error.reason}", retryable=True - ) from error - if status not in (200, 204): - raise TransportError( - f"{request.full_url} answered {status}, which is not a synchronous result" - ) - return _http_to_payload(content_type, body) - - -def _payload_to_http( - payload: temporalio.api.common.v1.Payload, -) -> tuple[str | None, bytes]: - metadata = dict(payload.metadata) - encoding = metadata.get("encoding", b"").decode() - message_type = metadata.get("messageType", b"").decode() - if encoding == "json/plain" and len(metadata) == 1: - return "application/json", payload.data - if encoding == "json/protobuf" and len(metadata) == 2 and message_type: - return ( - f'application/json; format=protobuf; message-type="{message_type}"', - payload.data, - ) - if encoding == "binary/protobuf" and len(metadata) == 2 and message_type: - return f'application/x-protobuf; message-type="{message_type}"', payload.data - if encoding == "binary/plain" and len(metadata) == 1: - return "application/octet-stream", payload.data - if encoding == "binary/null" and len(metadata) == 1: - return None, b"" - return "application/x-temporal-payload", payload.SerializeToString() - - -def _http_to_payload( - content_type: str | None, body: bytes -) -> temporalio.api.common.v1.Payload: - if not content_type: - if body: - raise TransportError("the answer has a body but no content type") - return temporalio.api.common.v1.Payload(metadata={"encoding": b"binary/null"}) - header = email.message.Message() - header["Content-Type"] = content_type - media_type = header.get_content_type() - message_type = header.get_param("message-type") - message_type = message_type if isinstance(message_type, str) else "" - if media_type == "application/x-temporal-payload": - payload = temporalio.api.common.v1.Payload() - _ = payload.ParseFromString(body) - return payload - if media_type == "application/json" and header.get_param("format") == "protobuf": - return temporalio.api.common.v1.Payload( - metadata={ - "encoding": b"json/protobuf", - "messageType": message_type.encode(), - }, - data=body, - ) - if media_type == "application/json": - return temporalio.api.common.v1.Payload( - metadata={"encoding": b"json/plain"}, data=body - ) - if media_type == "application/x-protobuf": - return temporalio.api.common.v1.Payload( - metadata={ - "encoding": b"binary/protobuf", - "messageType": message_type.encode(), - }, - data=body, - ) - if media_type == "application/octet-stream": - return temporalio.api.common.v1.Payload( - metadata={"encoding": b"binary/plain"}, data=body - ) - raise TransportError(f"the answer has an unknown content type {content_type!r}") - - class StreamServiceClient: """Calls the `temporal.api.streamservice.v1.StreamService` operations over a transport. @@ -437,185 +277,3 @@ async def describe_stream( _proto_message(answer, converter.transfer_type), DescribeStreamResponse, ) - - def stream(self, *, stream: StreamRef | None) -> StreamHandle: - """Bind ``stream`` and return a ``StreamHandle``.""" - return StreamHandle(self, stream=stream) - - def stream_producer( - self, *, stream: StreamRef | None, producer: Producer | None - ) -> StreamProducer: - """Bind ``stream``, ``producer`` and return a ``StreamProducer``.""" - return StreamProducer(self, stream=stream, producer=producer) - - -class StreamHandle: - """Binds ``stream`` for the ``StreamServiceClient`` operations that take it.""" - - def __init__( - self, client: StreamServiceClient, *, stream: StreamRef | None - ) -> None: - """Create a handle over ``client``.""" - self._client = client - self._stream = stream - - def producer(self, *, producer: Producer | None) -> StreamProducer: - """Bind ``producer`` and return a ``StreamProducer``.""" - return StreamProducer(self._client, stream=self._stream, producer=producer) - - async def append_records( - self, - *, - producer: Producer | None = None, - sequence: int | None = None, - records: list[AppendRecord] | None = None, - previous_completed_task: WorkflowTaskEpoch | None = None, - failed_tasks: list[int] | None = None, - withdraw_run: str | None = None, - ) -> AppendRecordsResponse: - """In Workflow code, the Brook transport puts this call into a buffer.""" - request = AppendRecordsRequest( - stream=self._stream, - producer=producer, - sequence=sequence, - records=records or [], - previous_completed_task=previous_completed_task, - failed_tasks=failed_tasks or [], - withdraw_run=withdraw_run, - ) - return await self._client.append_records(request) - - async def close_stream( - self, - *, - producer: Producer | None = None, - result: typing.Any | None = None, - failure: BaseException | None = None, - previous_completed_task: WorkflowTaskEpoch | None = None, - failed_tasks: list[int] | None = None, - ) -> CloseStreamResponse: - request = CloseStreamRequest( - stream=self._stream, - producer=producer, - result=result, - failure=failure, - previous_completed_task=previous_completed_task, - failed_tasks=failed_tasks or [], - ) - return await self._client.close_stream(request) - - async def read_records( - self, - *, - from_offset: int | None = None, - start: temporalio.api.streamservice.v1.stream_service_pb2.ReadStart.ValueType - | None = None, - topics: list[str] | None = None, - topic_prefixes: list[str] | None = None, - max_records: int | None = None, - wait: datetime.timedelta | None = None, - create_wait: datetime.timedelta | None = None, - idle_timeout: datetime.timedelta | None = None, - ) -> ReadRecordsResponse: - """Not for Workflow code in Brook. A long poll. NOT_FOUND for an uncreated Stream.""" - request = ReadRecordsRequest( - stream=self._stream, - from_offset=from_offset, - start=start, - topics=topics or [], - topic_prefixes=topic_prefixes or [], - max_records=max_records, - wait=wait, - create_wait=create_wait, - idle_timeout=idle_timeout, - ) - return await self._client.read_records(request) - - async def read_records_until_records( - self, - *, - from_offset: int | None = None, - start: temporalio.api.streamservice.v1.stream_service_pb2.ReadStart.ValueType - | None = None, - topics: list[str] | None = None, - topic_prefixes: list[str] | None = None, - max_records: int | None = None, - create_wait: datetime.timedelta | None = None, - idle_timeout: datetime.timedelta | None = None, - deadline: datetime.timedelta, - ) -> ReadRecordsResponse: - """Call the caller's ``read_records_until_records`` with the bound members.""" - request = ReadRecordsRequest( - stream=self._stream, - from_offset=from_offset, - start=start, - topics=topics or [], - topic_prefixes=topic_prefixes or [], - max_records=max_records, - create_wait=create_wait, - idle_timeout=idle_timeout, - ) - return await self._client.read_records_until_records(request, deadline=deadline) - - async def describe_stream( - self, - ) -> DescribeStreamResponse: - request = DescribeStreamRequest( - stream=self._stream, - ) - return await self._client.describe_stream(request) - - -class StreamProducer: - """Binds ``stream``, ``producer`` for the ``StreamServiceClient`` operations that take it.""" - - def __init__( - self, - client: StreamServiceClient, - *, - stream: StreamRef | None, - producer: Producer | None, - ) -> None: - """Create a handle over ``client``.""" - self._client = client - self._stream = stream - self._producer = producer - - async def append_records( - self, - *, - sequence: int | None = None, - records: list[AppendRecord] | None = None, - previous_completed_task: WorkflowTaskEpoch | None = None, - failed_tasks: list[int] | None = None, - withdraw_run: str | None = None, - ) -> AppendRecordsResponse: - """In Workflow code, the Brook transport puts this call into a buffer.""" - request = AppendRecordsRequest( - stream=self._stream, - producer=self._producer, - sequence=sequence, - records=records or [], - previous_completed_task=previous_completed_task, - failed_tasks=failed_tasks or [], - withdraw_run=withdraw_run, - ) - return await self._client.append_records(request) - - async def close_stream( - self, - *, - result: typing.Any | None = None, - failure: BaseException | None = None, - previous_completed_task: WorkflowTaskEpoch | None = None, - failed_tasks: list[int] | None = None, - ) -> CloseStreamResponse: - request = CloseStreamRequest( - stream=self._stream, - producer=self._producer, - result=result, - failure=failure, - previous_completed_task=previous_completed_task, - failed_tasks=failed_tasks or [], - ) - return await self._client.close_stream(request) diff --git a/tests/contrib/streams/test_bindings.py b/tests/contrib/streams/test_bindings.py index 60837ab8d..cdbdf1701 100644 --- a/tests/contrib/streams/test_bindings.py +++ b/tests/contrib/streams/test_bindings.py @@ -65,11 +65,9 @@ async def test_long_poll_carries_the_offset_until_records_come(): stream_pb.ReadRecordsResponse(next_offset=8), stream_pb.ReadRecordsResponse(records=[record], next_offset=10), ) - handle = client.StreamServiceClient(transport).stream( - stream=models.StreamRef(id="wf:run") - ) - answer = await handle.read_records_until_records( - from_offset=1, deadline=timedelta(seconds=5) + answer = await client.StreamServiceClient(transport).read_records_until_records( + models.ReadRecordsRequest(stream=models.StreamRef(id="wf:run"), from_offset=1), + deadline=timedelta(seconds=5), ) assert answer.records and answer.records[0].offset == 9 first, second = ( diff --git a/tests/contrib/streams/test_redis.py b/tests/contrib/streams/test_redis.py index bf52ed02a..6e9c8d9b6 100644 --- a/tests/contrib/streams/test_redis.py +++ b/tests/contrib/streams/test_redis.py @@ -15,7 +15,7 @@ from temporalio import activity, workflow from temporalio.api.common.v1 import Payload -from temporalio.client import Client +from temporalio.client import Client, WorkflowFailureError from temporalio.common import RawValue, RetryPolicy from temporalio.contrib import streams from temporalio.contrib.streams.proto import v1 as ext_pb @@ -900,3 +900,52 @@ def record(method: str, request: Any) -> bytes: for ref in refs: assert ref.workflow_id == handle.id assert ref.first_execution_run_id == handle.first_execution_run_id + + +@workflow.defn +class AlwaysFails: + @workflow.run + async def run(self, non_retryable: bool) -> None: + attempt = workflow.info().attempt + streams.workflow_stream().topic("t", type=str).publish(f"run-{attempt}") + raise ApplicationError(f"attempt {attempt} fails", non_retryable=non_retryable) + + +@pytest.mark.parametrize("non_retryable", [False, True]) +async def test_final_failure_under_a_retry_policy( + stream_client: Client, non_retryable: bool +): + async with new_worker(stream_client, AlwaysFails) as worker: + handle = await stream_client.start_workflow( + AlwaysFails.run, + non_retryable, + id=f"wf-{uuid.uuid4()}", + task_queue=worker.task_queue, + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=10), maximum_attempts=3 + ), + ) + with pytest.raises(WorkflowFailureError): + await handle.result() + reader = streams.for_handle( + handle, end_with_run=True, wait_for_create=timedelta(seconds=20) + ) + items = [item async for item in reader.subscribe(type=str)] + records = [item for item in items if isinstance(item, streams.Record)] + superseded = [item for item in items if isinstance(item, streams.Superseded)] + closes = [item for item in items if isinstance(item, streams.Closed)] + # Only the run that ends the chain closes the Stream, with its failure, and it is last. + assert len(closes) == 1 and items[-1] is closes[0] + assert closes[0].failure is not None + if non_retryable: + # A non-retryable failure ends the chain on attempt 1, so it closes at once. + assert [r.data for r in records] == ["run-1"] + assert superseded == [] + return + # Attempts 1 and 2 left the Stream open, so attempts 2 and 3 could write. Each retry run + # withdraws the run before it. + assert [r.data for r in records] == ["run-1", "run-2", "run-3"] + assert len(superseded) == 2 + assert [r.data for r in records if superseded[0].withdraws(r)] == ["run-1"] + assert [r.data for r in records if superseded[1].withdraws(r)] == ["run-2"] + assert not any(s.withdraws(records[-1]) for s in superseded)