diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock index ff49ac751..7558e410b 100644 --- a/temporalio/bridge/Cargo.lock +++ b/temporalio/bridge/Cargo.lock @@ -2879,7 +2879,7 @@ dependencies = [ [[package]] name = "temporalio-streams" version = "0.1.0" -source = "git+https://github.com/moetemp/sdk-streams?rev=813f9d129da2e90f86e8542e9a1ad766e2e7bf4f#813f9d129da2e90f86e8542e9a1ad766e2e7bf4f" +source = "git+https://github.com/moetemp/sdk-streams?rev=3984189fa4d6c809e5489df952b234302e986f76#3984189fa4d6c809e5489df952b234302e986f76" dependencies = [ "async-trait", "aws-lc-rs", diff --git a/temporalio/bridge/Cargo.toml b/temporalio/bridge/Cargo.toml index b57c51d75..ceab304d6 100644 --- a/temporalio/bridge/Cargo.toml +++ b/temporalio/bridge/Cargo.toml @@ -94,7 +94,7 @@ url = "2.2" # crates, so the bridge sees one type of each. [dependencies.temporalio-streams] git = "https://github.com/moetemp/sdk-streams" -rev = "813f9d129da2e90f86e8542e9a1ad766e2e7bf4f" +rev = "3984189fa4d6c809e5489df952b234302e986f76" default-features = false optional = true diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index dc7bb1d52..0d603aff9 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -28,6 +28,7 @@ StreamOffsetError, StreamStorageError, StreamTransportMissingError, + StreamUnsupportedError, ) from temporalio.contrib.streams._items import Closed, Idle, Record, Superseded from temporalio.contrib.streams._outside import OutsideStream @@ -56,6 +57,7 @@ "StreamRef", "StreamStorageError", "StreamTransportMissingError", + "StreamUnsupportedError", "Superseded", "Topic", "WorkflowStream", diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py index 3529f0ec3..be0e2bcce 100644 --- a/temporalio/contrib/streams/_errors.py +++ b/temporalio/contrib/streams/_errors.py @@ -77,6 +77,10 @@ class StreamBufferFullError(StreamError): """The writer's buffer is full. Flush, or publish less often.""" +class StreamUnsupportedError(StreamError): + """The Store lacks a capability that the call needs. The message ends with its name.""" + + class StreamTransportMissingError(StreamError): """No plugin carries Stream calls, such as ``RedisStreamPlugin``.""" @@ -93,6 +97,7 @@ class ReadOnlyContextError(StreamError): "stream_storage": StreamStorageError, "stream_offset": StreamOffsetError, "stream_buffer_full": StreamBufferFullError, + "stream_unsupported": StreamUnsupportedError, } diff --git a/temporalio/contrib/streams/testing.py b/temporalio/contrib/streams/testing.py index 391fb1363..4929dcb42 100644 --- a/temporalio/contrib/streams/testing.py +++ b/temporalio/contrib/streams/testing.py @@ -30,6 +30,7 @@ def __init__( *, retention: timedelta | None = None, safety_ttl: timedelta | None = None, + durable_writes: int = 0, runtime: temporalio.runtime.Runtime | None = None, ) -> None: """Create the plugin. @@ -38,9 +39,12 @@ def __init__( retention: The default Retention of a new Stream. Defaults to 7 days. safety_ttl: How long an open Workflow-owned Stream stays after its last write. Defaults to 90 days. + durable_writes: When above zero, each Workflow flush waits for this many + replicas. The memory Store has none, so each Workflow flush fails with the + missing ``replica_wait`` capability. It is for tests of that failure. runtime: The runtime of the client. Defaults to the default runtime. """ - config = ext_pb.RedisStreamConfig() + config = ext_pb.RedisStreamConfig(durable_writes=durable_writes) if retention is not None: config.retention.FromTimedelta(retention) if safety_ttl is not None: diff --git a/tests/contrib/streams/test_facade.py b/tests/contrib/streams/test_facade.py index b2cf09769..0de124be1 100644 --- a/tests/contrib/streams/test_facade.py +++ b/tests/contrib/streams/test_facade.py @@ -393,3 +393,15 @@ def test_snake_case_refs_still_decode(): assert converter.from_payload(payload, streams.StreamRef) == streams.StreamRef( "old", workflow_id="wf", first_execution_run_id="run" ) + + +def test_unsupported_capability_maps_to_its_error(): + details = ext_pb.StreamErrorDetails(reason="stream_unsupported").SerializeToString() + err = _errors.from_extension_error( + ExtensionError( + "The Store lacks replica_wait", RPCStatusCode.UNIMPLEMENTED, details + ) + ) + assert isinstance(err, streams.StreamUnsupportedError) + assert err.non_retryable and err.code == RPCStatusCode.UNIMPLEMENTED + assert err.message.endswith("replica_wait") diff --git a/tests/contrib/streams/test_redis.py b/tests/contrib/streams/test_redis.py index 6e9c8d9b6..7bd9586a2 100644 --- a/tests/contrib/streams/test_redis.py +++ b/tests/contrib/streams/test_redis.py @@ -37,7 +37,7 @@ SandboxedWorkflowRunner, SandboxRestrictions, ) -from tests.helpers import new_worker +from tests.helpers import assert_task_fail_eventually, new_worker # Ports that other local services own. _RESERVED = {7333, 7861, 7931, 6431, 7101, 7102, 7103} @@ -949,3 +949,17 @@ async def test_final_failure_under_a_retry_policy( 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) + + +async def test_missing_capability_fails_the_workflow_task(client: Client): + # The Workflow flush runs in Core's before_complete, so a Store that lacks a capability + # fails the Workflow Task with the capability's name, not a call in Workflow code. + config = client.config() + config["plugins"] = [MemoryStreamPlugin(durable_writes=1)] + memory_client = Client(**config) + async with new_worker(memory_client, QuickWriter) as worker: + handle = await memory_client.start_workflow( + QuickWriter.run, id=f"wf-{uuid.uuid4()}", task_queue=worker.task_queue + ) + await assert_task_fail_eventually(handle, message_contains="replica_wait") + await handle.terminate()