From 1bc9c7e261c9bc4b5f30e2ddf8d46e930ebaddd5 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 18 Oct 2026 00:00:26 -0700 Subject: [PATCH 1/2] Served the stream service from a Worker's store. --- temporalio/contrib/streams/_body.py | 33 +- temporalio/contrib/streams/nexus/__init__.py | 11 + temporalio/contrib/streams/nexus/_handler.py | 441 +++++++++++++++++++ tests/contrib/streams/test_nexus_service.py | 327 ++++++++++++++ 4 files changed, 805 insertions(+), 7 deletions(-) create mode 100644 temporalio/contrib/streams/nexus/_handler.py create mode 100644 tests/contrib/streams/test_nexus_service.py diff --git a/temporalio/contrib/streams/_body.py b/temporalio/contrib/streams/_body.py index a10bea9219..e1dfb8c513 100644 --- a/temporalio/contrib/streams/_body.py +++ b/temporalio/contrib/streams/_body.py @@ -1,12 +1,18 @@ """What the SDK owes a record's body between the converter and the store. -The payload converter turns a value into the body and stops there. Every other -payload the SDK sends then passes through the payload codec and external -storage, and a stream body owes the same, or a codec-protected deployment would -leak plaintext through its streams. The stream runtime never sees a plaintext -body, so both hashes that need one are taken here, before the codec. A codec -that encrypts with a fresh nonce makes every retry's bytes differ, so a store -that compared encoded bytes would refuse a retry as a divergent write. +The payload converter turns a value into the body and stops there. Every +other payload the SDK sends then passes through the payload codec and +external storage, and a stream body owes the same, or a codec-protected +deployment would leak plaintext through its streams. On the direct paths, a +Workflow's writer and the outside handles, the stream runtime never sees a +plaintext body, so both hashes that need one are taken here, before the codec. +A codec that encrypts with a fresh nonce makes every retry's bytes differ, so a +store that compared encoded bytes would refuse a retry as a divergent write. + +The stream service's Nexus path is different. The stream runtime gets +the caller's plain bodies, takes both hashes over the bytes the caller sent, +and asks the handler to run the codec. Only the codec step uses this module +there. """ from __future__ import annotations @@ -63,6 +69,19 @@ async def encode_bodies( return await converter._external_store_payload_sequence(encoded) +async def decode_bodies( + converter: DataConverter, payloads: Sequence[Payload] +) -> list[Payload]: + """Undo :func:`encode_bodies` on bodies read back from the store, together. + + Raises: + RuntimeError: A body is a claim and ``converter`` has no external + storage to redeem it with. + """ + retrieved = await converter._external_retrieve_payload_sequence(payloads) + return await converter._decode_payload_sequence(retrieved) + + async def decode_body(converter: DataConverter, payload: Payload) -> Payload: """Undo :func:`encode_bodies` on a body read back from the store. diff --git a/temporalio/contrib/streams/nexus/__init__.py b/temporalio/contrib/streams/nexus/__init__.py index f3e64d4e46..386f3a24cd 100644 --- a/temporalio/contrib/streams/nexus/__init__.py +++ b/temporalio/contrib/streams/nexus/__init__.py @@ -24,6 +24,11 @@ nothing beyond the standard library. * ``TemporalStreamsForwarder``, which hands each operation's input to the stream runtime as the contract's JSON transfer form. + +:class:`TemporalStreamsHandler` is the one hand-written piece. It runs its +authorizer and the namespace's payload codec, the parts only the SDK has, and +the stream runtime runs every rule of the contract on the store of its Worker's +client. """ from temporalio.contrib.streams import StreamRef @@ -40,6 +45,10 @@ HTTPStatusError, TemporalStreamsHttpClient, ) +from temporalio.contrib.streams.nexus._handler import ( + StreamAccess, + TemporalStreamsHandler, +) __all__ = [ "AppendInput", @@ -48,8 +57,10 @@ "ReadInput", "ReadOutput", "RecordWire", + "StreamAccess", "StreamCursor", "StreamRef", "TemporalStreams", + "TemporalStreamsHandler", "TemporalStreamsHttpClient", ] diff --git a/temporalio/contrib/streams/nexus/_handler.py b/temporalio/contrib/streams/nexus/_handler.py new file mode 100644 index 0000000000..c34fe76422 --- /dev/null +++ b/temporalio/contrib/streams/nexus/_handler.py @@ -0,0 +1,441 @@ +"""The stream service's handler, which the stream runtime serves.""" + +from __future__ import annotations + +import dataclasses +import json +from collections.abc import Awaitable, Callable +from datetime import datetime, timezone +from typing import Any, Literal + +import nexusrpc +import nexusrpc.handler +from google.protobuf.json_format import MessageToDict + +import temporalio.api.failure.v1 +import temporalio.nexus +from temporalio.api.common.v1 import Payload +from temporalio.api.enums.v1 import NexusHandlerErrorRetryBehavior +from temporalio.bridge.proto.streams import ( + CodecResult, + CodecWork, + NexusCall, + NexusFailure, + NexusLimits, + NexusResumeRequest, +) +from temporalio.client import Client +from temporalio.contrib.streams._body import decode_bodies, decode_body, encode_bodies +from temporalio.contrib.streams._plugin import call, store_for_client +from temporalio.contrib.streams._ref import StreamRef, owner_context +from temporalio.contrib.streams.nexus._generated import ( + AppendInput, + AppendOutput, + FollowInput, + FollowOutput, + FollowRangeInput, + ReadInput, + ReadOutput, + ReadRangeInput, + TemporalStreams, +) +from temporalio.contrib.streams.nexus._generated.forwarder import ( + TemporalStreamsForwarder, +) +from temporalio.converter import DataConverter + +__all__ = ["StreamAccess", "TemporalStreamsHandler"] + +# The Nexus failure metadata type under which the details are a +# temporal.api.failure.v1.Failure, which the failure converter reads back as is. +_TEMPORAL_FAILURE_TYPE = "temporal.api.failure.v1.Failure" +# The server sets it on ExecuteSyncNexusOperation from the authenticated +# namespace, and refuses it from the client. +_CALLER_NAMESPACE_HEADER = "temporal-caller-namespace" + + +@dataclasses.dataclass(frozen=True) +class StreamAccess: + """One call the stream service is about to serve, for an authorizer. + + .. warning:: + This API is experimental. + """ + + operation: Literal["append", "read", "read_range", "follow", "follow_range"] + """What the caller asks to do. A ``read_range`` reads again the records of + an earlier read that named a floor holder, for a Workflow that replays it. + A ``follow`` reads the stream and every stream its links open in this + namespace, so an authorizer that allows it allows those streams too. A + ``follow_range`` reads an earlier follow again.""" + + stream: StreamRef + """The stream the call names. + + Check ``stream.kind`` before an owner field. A Workflow's stream names + ``workflow_id``, and a standalone Activity's names ``activity_id`` with + an empty ``workflow_id``. The stream of an Activity that a Workflow + scheduled names ``workflow_id``, ``activity_id`` and the scheduling run + in ``chain_run_id``. + """ + + context: nexusrpc.handler.StartOperationContext + """The Nexus call, with the caller's headers.""" + + producer_id: str | None = None + """The producer an append writes as. ``None`` for a read.""" + + attempt: int | None = None + """The producer attempt an append writes. ``None`` for a read.""" + + floor_holder: str | None = None + """The holder a read or a ``read_range`` names, so the store keeps the + records it read for the caller's replay. ``None`` when the call names none. + A Bridge reader names ``//``, but the + caller chooses it, so treat it as a claim. With :attr:`caller_namespace` + set, the handler refuses a holder outside that namespace before it asks + the authorizer.""" + + caller_namespace: str | None = None + """The caller's namespace, as the server vouches for it. Only + ``ExecuteSyncNexusOperation``, which ``ClientTransport`` calls, sets + it. ``None`` on the HTTP ingress and on a Workflow's Nexus call.""" + + +@nexusrpc.handler.service_handler(service=TemporalStreams) +class TemporalStreamsHandler(TemporalStreamsForwarder): + """Serves the stream service over the store of its Worker's client. + + Register it on a Worker with ``nexus_service_handlers=[handler]`` and + point a Nexus endpoint at the Worker's task queue. The Worker's client + must carry a stream store, and a stream lives in the client's namespace. + + The stream runtime serves each call and runs every rule of the contract. + The handler runs only what the stream runtime can't: ``authorize``, and + the namespace's payload codec from the Worker's client. The store keeps + each body encoded with that codec, and a read decodes the bodies, so + Workflow code that reads gets plain bodies with no codec of its own. The + stream runtime holds those plain bodies in memory during the call only. + It never stores or logs them. + + The handler holds no state between calls. A reader carries the stream + runtime's read state in ``ReadInput.state``, so any Worker on the + endpoint's task queue can serve its next call, and retries are + deduplicated by the store. + + A read answers with at most ``max_records`` records and, past its first + record, at most ``max_answer_bytes`` of records as they cross, so a read's + result stays below the server's blob size limit. A record whose bytes + would take more than ``max_record_bytes`` in a read result can never be + read through the service, so an append with one is refused. A record + already in the store that is larger, a record the stream runtime can't + parse, and a record whose body the codec can't decode each cross as an + error at their cursor, which a reader goes on past. Sizes count what crosses: the record + bytes as base64, about 4/3 of them. + + A read that names a floor holder holds its records in the store and + answers ``replay``, the range of the records that crossed. A + ``read_range`` with that range answers the same records again, with no + byte budget, for a caller Workflow that replays the read. + + The handler writes and reads with its Worker's client and that client's + store, whatever caller sent the call. So every caller that may reach the + endpoint gets the Worker's store rights, for every stream in the + namespace and as any producer. Pass ``authorize`` to check each call: it + gets a :class:`StreamAccess` and answers whether to serve it, and a + refused call fails as unauthorized. Without ``authorize``, the handler + serves reads to any caller and refuses every append as unauthorized, + since an append could write as any producer. A ``nexusrpc.HandlerError`` it raises + crosses as it is, so it can give the caller a reason. Any other exception + crosses as a retryable ``INTERNAL`` error. + + Parked reads hold Nexus task slots for up to their ``wait_ms``. With many + readers, serve appends and reads on separate task queues or endpoints, so + an append doesn't wait for a slot behind parked reads, and give the + Worker more Nexus pollers than the default (``nexus_task_poller_behavior``). + """ + + def __init__( + self, + *, + max_answer_bytes: int = 1 << 20, + max_record_bytes: int = 3 << 19, + authorize: Callable[[StreamAccess], Awaitable[bool]] | None = None, + ) -> None: + """Serve the store of the Worker's client. + + Raises: + ValueError: ``max_answer_bytes`` or ``max_record_bytes`` is not + positive. + """ + if max_answer_bytes <= 0: + raise ValueError("max_answer_bytes must be positive") + if max_record_bytes <= 0: + raise ValueError("max_record_bytes must be positive") + self._limits = NexusLimits( + max_answer_bytes=max_answer_bytes, max_record_bytes=max_record_bytes + ) + self._authorize = authorize + + @nexusrpc.handler.sync_operation + async def append( + self, ctx: nexusrpc.handler.StartOperationContext, input: AppendInput + ) -> AppendOutput: + """Write one batch, or finish the producer.""" + await self._check( + StreamAccess( + "append", + input.stream, + ctx, + producer_id=input.producer_id, + attempt=input.attempt, + ) + ) + return await super().append(ctx, input) + + @nexusrpc.handler.sync_operation + async def read( + self, ctx: nexusrpc.handler.StartOperationContext, input: ReadInput + ) -> ReadOutput: + """Answer with the records after the caller's cursor, waiting for some.""" + await self._check( + StreamAccess("read", input.stream, ctx, floor_holder=input.floor_holder) + ) + return await super().read(ctx, input) + + @nexusrpc.handler.sync_operation + async def read_range( + self, ctx: nexusrpc.handler.StartOperationContext, input: ReadRangeInput + ) -> ReadOutput: + """Answer again the records of an earlier read, for a Workflow that replays it.""" + await self._check( + StreamAccess( + "read_range", input.stream, ctx, floor_holder=input.floor_holder + ) + ) + return await super().read_range(ctx, input) + + @nexusrpc.handler.sync_operation + async def follow( + self, ctx: nexusrpc.handler.StartOperationContext, input: FollowInput + ) -> FollowOutput: + """Answer with the items of a stream and of the streams its links open.""" + await self._check( + StreamAccess("follow", input.stream, ctx, floor_holder=input.floor_holder) + ) + return await super().follow(ctx, input) + + @nexusrpc.handler.sync_operation + async def follow_range( + self, ctx: nexusrpc.handler.StartOperationContext, input: FollowRangeInput + ) -> FollowOutput: + """Answer again the items of an earlier follow, for a Workflow that replays it.""" + await self._check( + StreamAccess( + "follow_range", input.stream, ctx, floor_holder=input.floor_holder + ) + ) + return await super().follow_range(ctx, input) + + async def forward( + self, + operation: str, + ctx: nexusrpc.handler.StartOperationContext, + transfer: Any, + ) -> Any: + """Serve one operation in the stream runtime, which asks for the codec.""" + return await serve_in_core( + temporalio.nexus.client(), + operation, + transfer, + self._limits, + ctx.request_deadline, + ) + + async def _check(self, access: StreamAccess) -> None: + access = dataclasses.replace( + access, caller_namespace=_caller_namespace(access.context) + ) + holder = access.floor_holder + if ( + access.caller_namespace is not None + and holder is not None + and not holder.startswith(f"{access.caller_namespace}/") + ): + # A holder pins records against retention, so a caller may hold + # them only under its own namespace. + raise nexusrpc.HandlerError( + f"floor holder {holder!r} is not under the caller's namespace " + f"{access.caller_namespace!r}", + type=nexusrpc.HandlerErrorType.UNAUTHORIZED, + retryable_override=False, + ) + if self._authorize is None: + if access.operation != "append": + return + raise nexusrpc.HandlerError( + "this stream service refuses appends, since its handler has no " + "authorize check; pass authorize= to TemporalStreamsHandler to " + "serve them", + type=nexusrpc.HandlerErrorType.UNAUTHORIZED, + retryable_override=False, + ) + try: + allowed = await self._authorize(access) + except nexusrpc.HandlerError: + # The authorizer chose what the caller sees. + raise + except Exception as error: + # The authorizer is user code, so its failure crosses spelled out, + # with its stack frames left on this side. + raise _handler_error(_internal(error)) from None + if allowed: + return + stream = access.stream + if stream.kind == "activity": + owner = f"Activity {stream.activity_id!r}" + elif stream.kind == "workflow_activity": + owner = ( + f"Activity {stream.activity_id!r} of Workflow {stream.workflow_id!r}" + ) + else: + owner = f"Workflow {stream.workflow_id!r}" + raise nexusrpc.HandlerError( + f"the caller may not {access.operation} stream {stream.topic!r} of {owner}", + type=nexusrpc.HandlerErrorType.UNAUTHORIZED, + retryable_override=False, + ) + + +async def serve_in_core( + client: Client, + operation: str, + transfer: Any, + limits: NexusLimits, + deadline: datetime | None, +) -> Any: + """Serve one stream service operation on the store of ``client``. + + ``transfer`` is the input's JSON transfer form, and the answer is the + output's. The stream runtime runs every rule of the contract and asks for + the codec of ``client`` when a body crosses it. + + Raises: + nexusrpc.HandlerError: The stream runtime refused the call. + ValueError: No stream store is registered on ``client``. + """ + service = await store_for_client(client)._nexus_service_for(client) + request = NexusCall( + input=json.dumps(transfer).encode(), + limits=limits, + namespace=client.namespace, + ) + if deadline is not None: + if deadline.tzinfo is None: + deadline = deadline.replace(tzinfo=timezone.utc) + request.deadline.FromDatetime(deadline) + if operation == "append": + step = await call(service.nexus_append(request)) + elif operation == "read_range": + step = await call(service.nexus_read_range(request)) + elif operation == "follow": + step = await call(service.nexus_follow(request)) + elif operation == "follow_range": + step = await call(service.nexus_follow_range(request)) + else: + step = await call(service.nexus_read(request)) + while step.HasField("codec"): + results = await _run_codec(client.data_converter, step.codec) + step = await call( + service.nexus_resume( + NexusResumeRequest( + continuation=step.codec.continuation, results=results + ) + ) + ) + if step.HasField("failure"): + raise _handler_error(step.failure) + return json.loads(step.output) + + +async def _run_codec(converter: DataConverter, work: CodecWork) -> list[CodecResult]: + # Every body of one owner's streams uses that owner's context. + converter = converter.with_context( + owner_context(work.namespace, work.workflow_id, work.activity_id) + ) + payloads = list(work.payloads) + if work.encode: + try: + encoded = await encode_bodies(converter, payloads) + except Exception as error: + # An append's batch is one write, so one failed body fails it. + failed = CodecResult(error=f"{type(error).__name__}: {error}") + return [failed] * len(payloads) + return [CodecResult(payload=payload) for payload in encoded] + try: + # One call for the page, since a remote codec pays a round trip each. + decoded_page = await decode_bodies(converter, payloads) + except Exception: + pass + else: + return [CodecResult(payload=payload) for payload in decoded_page] + results: list[CodecResult] = [] + for payload in payloads: + # Each record decodes on its own, so a body the codec refuses crosses + # as an error at its cursor and the rest of the read goes on. + try: + decoded: Payload = await decode_body(converter, payload) + except Exception as error: + results.append(CodecResult(error=f"{type(error).__name__}: {error}")) + else: + results.append(CodecResult(payload=decoded)) + return results + + +def _caller_namespace(ctx: nexusrpc.handler.StartOperationContext) -> str | None: + for key, value in ctx.headers.items(): + if key.lower() == _CALLER_NAMESPACE_HEADER: + return value + return None + + +def _internal(error: Exception) -> NexusFailure: + # The same shape the stream runtime gives a stream condition, with the + # error's class as the cause's type and no stack. + name = type(error).__name__ + return NexusFailure( + handler_error_type=nexusrpc.HandlerErrorType.INTERNAL.name, + retryable=True, + message=f"{name}: {error}", + failure=temporalio.api.failure.v1.Failure( + cause=temporalio.api.failure.v1.Failure( + message=str(error), + application_failure_info=temporalio.api.failure.v1.ApplicationFailureInfo( + type=name + ), + ), + nexus_handler_failure_info=temporalio.api.failure.v1.NexusHandlerFailureInfo( + type=nexusrpc.HandlerErrorType.INTERNAL.name, + retry_behavior=NexusHandlerErrorRetryBehavior.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_RETRYABLE, + ), + ), + ) + + +def _handler_error(failure: NexusFailure) -> nexusrpc.HandlerError: + try: + kind = nexusrpc.HandlerErrorType[failure.handler_error_type] + except KeyError: + # A type this release doesn't know, which only a stream runtime from + # another release sends. + kind = nexusrpc.HandlerErrorType.INTERNAL + return nexusrpc.HandlerError( + failure.message, + type=kind, + retryable_override=failure.retryable, + original_failure=nexusrpc.Failure( + message=failure.message, + metadata={"type": _TEMPORAL_FAILURE_TYPE}, + details=MessageToDict(failure.failure), + ), + ) diff --git a/tests/contrib/streams/test_nexus_service.py b/tests/contrib/streams/test_nexus_service.py new file mode 100644 index 0000000000..a9851d97c7 --- /dev/null +++ b/tests/contrib/streams/test_nexus_service.py @@ -0,0 +1,327 @@ +"""The stream service, served by a Worker and called over the Nexus HTTP ingress. + +Every case runs against the stock dev server, started here with an HTTP port, +through a Nexus endpoint the case creates and deletes, so the request goes +through the same ingress a remote caller's does. Core serves the contract and +its own tests pin each rule. These cases check the whole path end to end, and +pin what only the SDK decides: the authorizer, the payload codec and its +serialization context, and the failure that crosses. + +The cases are parametrized over ``BACKINGS``, the stores the service runs +over. Each yields a :class:`Backing`. +""" + +from __future__ import annotations + +import json +import uuid +from collections.abc import AsyncIterator, Awaitable, Callable +from contextlib import asynccontextmanager +from dataclasses import dataclass +from datetime import timedelta +from typing import Any + +import pytest +import pytest_asyncio + +import temporalio.api.nexus.v1 +import temporalio.api.operatorservice.v1 +from temporalio import workflow +from temporalio.api.common.v1 import Payload +from temporalio.bridge.proto.streams.v1 import StreamRecord as WireRecord +from temporalio.client import Client +from temporalio.contrib.streams import ( + StreamRef, + StreamStorePlugin, +) +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.contrib.streams.nexus import ( + AppendInput, + AppendOutput, + HTTPStatusError, + ReadInput, + ReadOutput, + StreamAccess, + TemporalStreamsHandler, + TemporalStreamsHttpClient, +) +from temporalio.converter import ( + DataConverter, +) +from temporalio.testing import WorkflowEnvironment +from temporalio.worker import PollerBehaviorSimpleMaximum, Worker +from tests import DEV_SERVER_DOWNLOAD_VERSION +from tests.helpers import find_free_port + +_SERVICE = "temporal.sdk.streams.v1.TemporalStreams" + + +@dataclass +class Server: + """The dev server these cases share, and its Nexus HTTP ingress.""" + + env: WorkflowEnvironment + http: str + + @property + def client(self) -> Client: + return self.env.client + + +@pytest_asyncio.fixture(scope="module") # type: ignore[reportUntypedFunctionDecorator] +async def server() -> AsyncIterator[Server]: + port = find_free_port() + env = await WorkflowEnvironment.start_local( + dev_server_extra_args=[ + "--http-port", + str(port), + "--dynamic-config-value", + 'system.system.refreshNexusEndpointsMinWait="0s"', + "--dynamic-config-value", + "activity.enableStandalone=true", + ], + dev_server_download_version=DEV_SERVER_DOWNLOAD_VERSION, + ) + try: + yield Server(env, f"http://127.0.0.1:{port}") + finally: + await env.shutdown() + + +@dataclass +class Backing: + """One store the service runs over, and a client that carries it.""" + + name: str + store: StreamStorePlugin + client: Client + host: Callable[[str], Awaitable[None]] + """Starts the Workflow that owns a stream, since Core checks that the + owner exists before a stream is written.""" + + +@workflow.defn +class StreamOwner: + """Owns a stream until it is told to finish.""" + + def __init__(self) -> None: + self.done = False + + @workflow.run + async def run(self) -> None: + await workflow.wait_condition(lambda: self.done) + + @workflow.signal + def finish(self) -> None: + self.done = True + + +async def connect( + client: Client, store: StreamStorePlugin, converter: DataConverter | None = None +) -> Client: + """A client of ``client``'s server that carries ``store``.""" + return await Client.connect( + client.service_client.config.target_host, + namespace=client.namespace, + data_converter=converter or DataConverter.default, + plugins=[store], + ) + + +def with_converter(client: Client, converter: DataConverter) -> Client: + """``client`` on its own connection, with another data converter.""" + config = client.config() + config["data_converter"] = converter + return Client(**config) + + +@asynccontextmanager +async def _memory_backing(client: Client) -> AsyncIterator[Backing]: + store = MemoryStreams() + connected = await connect(client, store) + + async def host(workflow_id: str) -> None: + # No Worker runs it, so it stays open for the case. + await connected.start_workflow( + StreamOwner.run, id=workflow_id, task_queue=f"unpolled-{uuid.uuid4().hex}" + ) + + yield Backing("memory", store, connected, host) + + +BACKINGS: dict[str, Callable[[Client], Any]] = {"memory": _memory_backing} + + +async def allow_all(_access: StreamAccess) -> bool: + return True + + +@dataclass +class Service: + """A handler served by a Worker, and a caller on its endpoint.""" + + caller: TemporalStreamsHttpClient + handler: TemporalStreamsHandler + backing: Backing + client: Client + endpoint: str + + +@asynccontextmanager +async def serve( + server: Server, + backing: Backing, + *, + client: Client | None = None, + **handler_options: Any, +) -> AsyncIterator[Service]: + """Serve ``backing`` behind a fresh endpoint, deleted on the way out. + + The handler's Worker runs on ``client``, the backing's by default. It + serves every call unless ``authorize`` is given. + """ + handler_options.setdefault("authorize", allow_all) + handler = TemporalStreamsHandler(**handler_options) + task_queue = f"streams-service-{uuid.uuid4().hex}" + endpoint_name = f"streams-{uuid.uuid4().hex}" + worker_client = backing.client if client is None else client + async with Worker( + worker_client, + task_queue=task_queue, + nexus_service_handlers=[handler], + # As the handler recommends: with the default five pollers over the + # dev server's partitions, an append behind parked reads was seen to + # wait for one of them to finish. + nexus_task_poller_behavior=PollerBehaviorSimpleMaximum(20), + ): + created = await server.client.operator_service.create_nexus_endpoint( + temporalio.api.operatorservice.v1.CreateNexusEndpointRequest( + spec=temporalio.api.nexus.v1.EndpointSpec( + name=endpoint_name, + target=temporalio.api.nexus.v1.EndpointTarget( + worker=temporalio.api.nexus.v1.EndpointTarget.Worker( + namespace=server.client.namespace, + task_queue=task_queue, + ) + ), + ) + ) + ) + endpoint = created.endpoint + try: + yield Service( + TemporalStreamsHttpClient( + f"{server.http}/nexus/endpoints/{endpoint.id}/services/{_SERVICE}" + ), + handler, + backing, + worker_client, + endpoint_name, + ) + finally: + await server.client.operator_service.delete_nexus_endpoint( + temporalio.api.operatorservice.v1.DeleteNexusEndpointRequest( + id=endpoint.id, version=endpoint.version + ) + ) + + +@pytest_asyncio.fixture(params=sorted(BACKINGS)) # type: ignore[reportUntypedFunctionDecorator] +async def backing( + request: pytest.FixtureRequest, server: Server +) -> AsyncIterator[Backing]: + async with BACKINGS[request.param](server.client) as found: + yield found + + +@pytest_asyncio.fixture # type: ignore[reportUntypedFunctionDecorator] +async def service(server: Server, backing: Backing) -> AsyncIterator[Service]: + async with serve(server, backing) as found: + yield found + + +async def stream_of(service: Service, topic: str = "out") -> StreamRef: + """A fresh stream, with its owner started.""" + workflow_id = f"streams-service-{uuid.uuid4().hex}" + await service.backing.host(workflow_id) + return StreamRef.for_workflow(workflow_id, topic=topic) + + +async def activity_stream_of(service: Service, topic: str = "out") -> StreamRef: + """A fresh stream of a standalone Activity, with its owner started. + + No Worker polls its task queue, so the Activity run stays open. + """ + activity_id = f"streams-service-{uuid.uuid4().hex}" + owner = await service.backing.client.start_activity( + "never_polled", + id=activity_id, + task_queue=f"unpolled-{uuid.uuid4().hex}", + start_to_close_timeout=timedelta(minutes=5), + ) + return StreamRef.for_activity(activity_id, run_id=owner.run_id, topic=topic) + + +def body(text: str) -> bytes: + """A serialized payload, as a writer's converter and codec leave it.""" + return Payload( + metadata={"encoding": b"binary/test"}, data=text.encode() + ).SerializeToString() + + +def records_of(answer: ReadOutput) -> list[WireRecord]: + return [ + WireRecord.FromString(record.record) + for record in answer.records + if record.superseded is None + ] + + +def texts(answer: ReadOutput) -> list[str]: + return [record.body.data.decode() for record in records_of(answer)] + + +async def append( + service: Service, + ref: StreamRef, + *texts_: str, + producer_id: str = "p", + attempt: int = 1, + sequence: int = 1, +) -> AppendOutput: + return await service.caller.append( + AppendInput( + stream=ref, + producer_id=producer_id, + attempt=attempt, + sequence=sequence, + payloads=[body(text) for text in texts_], + ) + ) + + +async def read( + service: Service, + ref: StreamRef, + after: str = "", + *, + wait_ms: int = 2000, + max_records: int | None = None, +) -> ReadOutput: + return await service.caller.read( + ReadInput( + stream=ref, after_token=after, wait_ms=wait_ms, max_records=max_records + ) + ) + + +def refused_as(error: HTTPStatusError) -> str: + """The stream error class the handler refused the call with. + + It leads the message and is the cause's application error type, and the + two have to agree. + """ + failure = json.loads(error.detail) + cause_type = failure["details"]["cause"]["applicationFailureInfo"]["type"] + assert failure["message"].startswith(f"{cause_type}: "), failure["message"] + return cause_type From 1a45d56ac01ff4a265c818a60a6006698a079dc5 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 18 Oct 2026 01:50:17 -0700 Subject: [PATCH 2/2] Named the Workflow Producer in the stream body notes. --- temporalio/contrib/streams/_body.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/temporalio/contrib/streams/_body.py b/temporalio/contrib/streams/_body.py index e1dfb8c513..bb4c0de47d 100644 --- a/temporalio/contrib/streams/_body.py +++ b/temporalio/contrib/streams/_body.py @@ -4,7 +4,7 @@ other payload the SDK sends then passes through the payload codec and external storage, and a stream body owes the same, or a codec-protected deployment would leak plaintext through its streams. On the direct paths, a -Workflow's writer and the outside handles, the stream runtime never sees a +Workflow Producer and the outside handles, the stream runtime never sees a plaintext body, so both hashes that need one are taken here, before the codec. A codec that encrypts with a fresh nonce makes every retry's bytes differ, so a store that compared encoded bytes would refuse a retry as a divergent write.