[Brook] Mapped a Store's missing capability to StreamUnsupportedError.

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
 

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",

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,
 }
 
 

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:

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")

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()