From 9f9e3f017150848baa07599dbffe245c768aeb91 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 21 Oct 2026 19:40:28 -0700 Subject: [PATCH 1/5] Pinned the v4 crate and regenerated the Stream bindings. --- .../api/streamservice/v1/stream_service.proto | 25 ++- .../temporal/sdk/streams/v1/extension.proto | 43 ++++++ scripts/_wit/stream-service.wit | 25 ++- .../streamservice/v1/stream_service_pb2.py | 85 +++++----- .../streamservice/v1/stream_service_pb2.pyi | 67 ++++---- temporalio/bridge/Cargo.lock | 2 +- temporalio/bridge/Cargo.toml | 2 +- temporalio/contrib/streams/__init__.py | 3 +- temporalio/contrib/streams/_items.py | 12 -- temporalio/contrib/streams/_outside.py | 9 +- temporalio/contrib/streams/_reader.py | 25 +-- temporalio/contrib/streams/_stream.py | 4 - temporalio/contrib/streams/_workflow.py | 10 +- .../contrib/streams/proto/v1/__init__.py | 8 + .../contrib/streams/proto/v1/extension_pb2.py | 124 +++++++++++---- .../streams/proto/v1/extension_pb2.pyi | 145 ++++++++++++++++++ temporalio/contrib/streams/system/client.py | 16 +- temporalio/contrib/streams/system/models.py | 28 ++-- .../system/operations/append_records.py | 25 +-- .../streams/system/operations/close_stream.py | 22 +-- .../system/operations/create_stream.py | 4 + .../streams/system/operations/read_records.py | 4 - tests/contrib/streams/fake_extension.py | 11 +- tests/contrib/streams/test_bindings.py | 4 +- tests/contrib/streams/test_facade.py | 17 -- tests/contrib/streams/test_redis.py | 41 +---- 26 files changed, 456 insertions(+), 305 deletions(-) diff --git a/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto b/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto index 8b94f6021..a9b835480 100644 --- a/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto +++ b/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto @@ -17,6 +17,10 @@ message StreamRef { // client or Activity ref. The Store keys on id only and never reads these. string workflow_id = 2; string first_execution_run_id = 3; + // The name of the backend that holds the Stream, as the Worker and client config register + // it, so every process routes a ref it receives. Empty is the default backend. The Store + // never reads it. + string backend = 4; } enum ProducerKind { @@ -69,9 +73,9 @@ enum StreamRecordKind { // Lua flags the first batch of a new epoch. The read makes this Record from the flag. STREAM_RECORD_KIND_SUPERSEDED = 2; STREAM_RECORD_KIND_CLOSE = 3; - // The last Record of a read that saw no new Record for idle_timeout. Its offset is the - // newest offset read + 1, the place to resume from. - STREAM_RECORD_KIND_IDLE = 4; + // A Reader that stops after a quiet period sets its own timer and resumes from next_offset. + reserved 4; + reserved "STREAM_RECORD_KIND_IDLE"; } message Supersession { @@ -101,7 +105,8 @@ message StreamRecord { temporal.api.failure.v1.Failure close_failure = 8; } -// Idempotent for one id and one Retention. ALREADY_EXISTS for another Retention. +// Idempotent for one id, one Retention and one auto_close. ALREADY_EXISTS for another +// Retention or another auto_close. message CreateStreamRequest { string namespace = 1; // Required from Workflow code. @@ -110,6 +115,11 @@ message CreateStreamRequest { google.protobuf.Duration retention = 3; // Workflow only. A retry run's first create: the previous run, whose Records it withdraws. string withdraw_run = 4; + // Close the Stream with no result when no Record was appended for this long, counted from + // the create. Empty never closes on its own. It is separate from Retention and the safety + // TTL. The Store applies it on the next access, so no call sees the Stream open after the + // period. UNIMPLEMENTED (detail stream_unsupported) on a Store that can't. + google.protobuf.Duration auto_close = 5; } enum StreamStatus { @@ -180,9 +190,8 @@ message ReadRecordsRequest { // How long to wait for a Stream that nobody created yet. Empty fails at once with // NOT_FOUND. At most 60 s. google.protobuf.Duration create_wait = 8; - // End the read with an IDLE Record when no new Record comes for this long. Empty never - // ends on idle. The facade counts it across reads, since one read waits at most 60 s. - google.protobuf.Duration idle_timeout = 9; + reserved 9; + reserved "idle_timeout"; // Also read the Topics that start with one of these. A Record passes when its Topic is in // topics or starts with a prefix. Both empty reads every Topic. repeated string topic_prefixes = 10; @@ -205,4 +214,6 @@ message DescribeStreamResponse { // Zero when the Stream has no Record. int64 last_offset = 2; google.protobuf.Duration retention = 3; + // The auto_close of the create. Empty when the Stream has none. + google.protobuf.Duration auto_close = 4; } diff --git a/scripts/_proto/temporal/sdk/streams/v1/extension.proto b/scripts/_proto/temporal/sdk/streams/v1/extension.proto index c4c0c005f..5791ae770 100644 --- a/scripts/_proto/temporal/sdk/streams/v1/extension.proto +++ b/scripts/_proto/temporal/sdk/streams/v1/extension.proto @@ -27,6 +27,9 @@ message RedisStreamConfig { int32 blocking_reads_per_node = 8; // Empty is 5 s. google.protobuf.Duration response_timeout = 9; + // The backend name that lang registered this extension under. The crate puts it in each + // StreamRef it builds, such as the one in a create answer. + string backend = 10; } // Method `BufferWorkflowCalls`, synchronous. Lang calls it at the end of each Activation of a @@ -61,6 +64,29 @@ message WorkflowCall { message BufferWorkflowCallsResponse { } +// Method `FlushRunBuffer`, asynchronous. Lang calls it before it runs a Local Activity, also +// before each retry that Core runs in the same task. The crate writes the run's buffered +// creates and appends in the epoch of the Workflow Task that dispatched the Local Activity, so +// the run's Records come before the Local Activity's. Closes still wait for an accepted +// completion. The task's completion writes only what came later, with the numbering going on, +// and sends the task's withdrawal facts to each Stream this call wrote. A Local Activity of a +// task that already completed, or of a task older than an earlier flush, writes nothing. When +// the call fails, the run's next completion fails with the same error, so the retry withdraws +// what was written. Lang then fails the Local Activity attempt. +message FlushRunBufferRequest { + string namespace = 1; + string run_id = 2; + // `ActivityTask.Start.workflow_task` (field 20) of the Local Activity. + temporal.api.streamservice.v1.WorkflowTaskEpoch workflow_task = 3; + // How long the write may take. Empty is 10 s. + google.protobuf.Duration timeout = 4; +} + +message FlushRunBufferResponse { + // The Records the call wrote. Zero when the run had nothing buffered. + int32 records = 1; +} + // Method `OpenWriter`, asynchronous. A buffered writer for an Activity or a client Producer. message OpenWriterRequest { string namespace = 1; @@ -91,6 +117,23 @@ message PublishRequest { message PublishResponse { } +// Method `WriteRecords`, asynchronous. Appends Records at once for an open writer, after the +// Records that `Publish` buffered, and answers when they are in the Store. It keeps the writer's +// Producer epoch and `sequence` order, and a Store failure resends the same batch, so it goes in +// once. Records past the Store's batch limit go as more batches, in order. An awaited publish of +// an Activity or a client uses this call, so it needs no interval flush. A failure after the +// Store retries fails the writer, as a failed flush does. +message WriteRecordsRequest { + uint64 writer_id = 1; + repeated temporal.api.streamservice.v1.AppendRecord records = 2; +} + +message WriteRecordsResponse { + // The offsets of the first and the last Record. Zero when the request has no Record. + int64 first_offset = 1; + int64 last_offset = 2; +} + // Method `FlushWriter`, asynchronous. Waits until the Records published so far are in the Store. message FlushWriterRequest { uint64 writer_id = 1; diff --git a/scripts/_wit/stream-service.wit b/scripts/_wit/stream-service.wit index ed94f21a2..bcf5885be 100644 --- a/scripts/_wit/stream-service.wit +++ b/scripts/_wit/stream-service.wit @@ -26,9 +26,8 @@ interface stream-service { nanos: option, } - // The default Stream of a Workflow has the id :. A Stream that - // the Workflow creates with a name has ::. Each part is - // percent-encoded. So a caller that knows the first run id builds the id with no call. + // A Stream that a Workflow creates has the id ::. Each part + // is percent-encoded. So a caller that knows the first run id builds the id with no call. /// @nexus.doc "Identifies one Stream. The id is opaque to callers." /// @nexus.proto "temporal.api.streamservice.v1.StreamRef" record stream-ref { @@ -38,6 +37,8 @@ interface stream-service { workflow-id: option, /// @nexus.doc "The first run id of the owning run chain. Empty when workflow-id is." first-execution-run-id: option, + /// @nexus.doc "The name of the backend that holds the Stream, as the Worker and client config register it. Empty is the default backend. The Store never reads it." + backend: option, } /// @nexus.proto "temporal.api.streamservice.v1.ProducerKind" @@ -104,7 +105,6 @@ interface stream-service { stream-record-kind-data, stream-record-kind-superseded, stream-record-kind-close, - stream-record-kind-idle, } // For an Activity, Local Activity or client Producer: withdraw the Records of this @@ -139,19 +139,19 @@ 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 CLOSE, IDLE and SUPERSEDED Records." + /// @nexus.doc "Empty on CLOSE and SUPERSEDED Records." topic: option, producer: option, /// @nexus.doc "Set on a data Record." body: option, /// @nexus.doc "Set on a SUPERSEDED Record. Lua flags the batch, and the read makes it." supersession: option, - /// @nexus.doc "Set on a close Record with an outcome. An IDLE Record carries only the offset, the newest read + 1." + /// @nexus.doc "Set on a close Record with an outcome." close-result: option, close-failure: option, } - /// @nexus.doc "Create a Stream. Idempotent for one id and one Retention. ALREADY_EXISTS for another Retention." + /// @nexus.doc "Create a Stream. Idempotent for one id, one Retention and one auto-close. ALREADY_EXISTS for another Retention or another auto-close." /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamRequest" record create-stream-request { /// @nexus.doc "Required from Workflow code. Empty elsewhere asks the service for an id." @@ -160,6 +160,8 @@ interface stream-service { retention: option, /// @nexus.doc "Workflow only. A retry run's first batch or create: the previous run, whose Records it withdraws." withdraw-run: option, + /// @nexus.doc "Close the Stream with no result when no Record was appended for this long. Empty never closes on its own. It is separate from Retention." + auto-close: option, /// @nexus.omit namespace: placeholder, } @@ -255,8 +257,6 @@ interface stream-service { wait: option, /// @nexus.doc "How long to wait for an uncreated Stream. Empty fails at once. At most 60 s." create-wait: option, - /// @nexus.doc "End the read with an IDLE Record when no new Record comes for this long. Empty never ends on idle. The facade counts it across reads." - idle-timeout: option, /// @nexus.omit namespace: placeholder, } @@ -285,18 +285,18 @@ interface stream-service { /// @nexus.offset "stream-offset" last-offset: option, retention: option, + /// @nexus.doc "The auto-close of the create. Empty when the Stream has none." + auto-close: option, } /// @nexus.operation name="CreateStream" create-stream: func(request: create-stream-request) -> create-stream-response; - /// @nexus.doc "In Workflow code, the Brook transport puts this call into a buffer." + /// @nexus.doc "In Workflow code, the Brook transport puts this call into the run's buffer, and the Workflow facade drops the response." /// @nexus.operation name="AppendRecords" - /// @nexus.one-way append-records: func(request: append-records-request) -> append-records-response; /// @nexus.operation name="CloseStream" - /// @nexus.one-way close-stream: func(request: close-stream-request) -> close-stream-response; /// @nexus.doc "Not for Workflow code in Brook. A long poll. NOT_FOUND for an uncreated Stream." @@ -308,7 +308,6 @@ interface stream-service { /// done-field="done" /// offset-field="from-offset" /// next-offset-field="next-offset" - /// idle-timeout-field="idle-timeout" read-records: func(request: read-records-request) -> read-records-response; /// @nexus.operation name="DescribeStream" diff --git a/temporalio/api/streamservice/v1/stream_service_pb2.py b/temporalio/api/streamservice/v1/stream_service_pb2.py index d197f0722..5da897aa4 100644 --- a/temporalio/api/streamservice/v1/stream_service_pb2.py +++ b/temporalio/api/streamservice/v1/stream_service_pb2.py @@ -26,7 +26,7 @@ ) DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( - b'\n2temporal/api/streamservice/v1/stream_service.proto\x12\x1dtemporal.api.streamservice.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a%temporal/api/failure/v1/message.proto"L\n\tStreamRef\x12\n\n\x02id\x18\x01 \x01(\t\x12\x13\n\x0bworkflow_id\x18\x02 \x01(\t\x12\x1e\n\x16\x66irst_execution_run_id\x18\x03 \x01(\t"\xc8\x01\n\rProducerEpoch\x12\x1d\n\x15task_started_event_id\x18\x01 \x01(\x03\x12\x14\n\x0ctask_attempt\x18\x02 \x01(\x05\x12\x35\n\x11task_started_time\x18\x05 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12:\n\x16\x61ttempt_scheduled_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x05"\x9e\x01\n\x08Producer\x12\x39\n\x04kind\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.ProducerKind\x12\n\n\x02id\x18\x02 \x01(\t\x12;\n\x05\x65poch\x18\x03 \x01(\x0b\x32,.temporal.api.streamservice.v1.ProducerEpoch\x12\x0e\n\x06run_id\x18\x04 \x01(\t"p\n\x11WorkflowTaskEpoch\x12\x18\n\x10started_event_id\x18\x01 \x01(\x03\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x05\x12\x30\n\x0cstarted_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp"L\n\x0c\x41ppendRecord\x12\r\n\x05topic\x18\x01 \x01(\t\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload"\xb3\x02\n\x0cSupersession\x12\x42\n\rproducer_kind\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.ProducerKind\x12\x13\n\x0bproducer_id\x18\x02 \x01(\t\x12;\n\x05\x65poch\x18\x03 \x01(\x0b\x32,.temporal.api.streamservice.v1.ProducerEpoch\x12Q\n\x17previous_completed_task\x18\x04 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x05 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x06 \x01(\t\x12\x0e\n\x06run_id\x18\x07 \x01(\t"\x89\x03\n\x0cStreamRecord\x12=\n\x04kind\x18\x01 \x01(\x0e\x32/.temporal.api.streamservice.v1.StreamRecordKind\x12\x0e\n\x06offset\x18\x02 \x01(\x03\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x39\n\x08producer\x18\x04 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12-\n\x04\x62ody\x18\x05 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x41\n\x0csupersession\x18\x06 \x01(\x0b\x32+.temporal.api.streamservice.v1.Supersession\x12\x35\n\x0c\x63lose_result\x18\x07 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x37\n\rclose_failure\x18\x08 \x01(\x0b\x32 .temporal.api.failure.v1.Failure"\x7f\n\x13\x43reateStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x11\n\tstream_id\x18\x02 \x01(\t\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x14\n\x0cwithdraw_run\x18\x04 \x01(\t"\x9e\x01\n\x14\x43reateStreamResponse\x12\x38\n\x06stream\x18\x01 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x0f\n\x07\x63reated\x18\x02 \x01(\x08\x12;\n\x06status\x18\x03 \x01(\x0e\x32+.temporal.api.streamservice.v1.StreamStatus"\xed\x02\n\x14\x41ppendRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12\x10\n\x08sequence\x18\x04 \x01(\x03\x12<\n\x07records\x18\x05 \x03(\x0b\x32+.temporal.api.streamservice.v1.AppendRecord\x12Q\n\x17previous_completed_task\x18\x06 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x07 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x08 \x01(\t"B\n\x15\x41ppendRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03"\xe9\x02\n\x12\x43loseStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12/\n\x06result\x18\x04 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x31\n\x07\x66\x61ilure\x18\x05 \x01(\x0b\x32 .temporal.api.failure.v1.Failure\x12Q\n\x17previous_completed_task\x18\x06 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x07 \x03(\x03"\x15\n\x13\x43loseStreamResponse"\xf6\x02\n\x12ReadRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x13\n\x0b\x66rom_offset\x18\x03 \x01(\x03\x12\x37\n\x05start\x18\x04 \x01(\x0e\x32(.temporal.api.streamservice.v1.ReadStart\x12\x0e\n\x06topics\x18\x05 \x03(\t\x12\x13\n\x0bmax_records\x18\x06 \x01(\x05\x12\'\n\x04wait\x18\x07 \x01(\x0b\x32\x19.google.protobuf.Duration\x12.\n\x0b\x63reate_wait\x18\x08 \x01(\x0b\x32\x19.google.protobuf.Duration\x12/\n\x0cidle_timeout\x18\t \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0etopic_prefixes\x18\n \x03(\t"v\n\x13ReadRecordsResponse\x12<\n\x07records\x18\x01 \x03(\x0b\x32+.temporal.api.streamservice.v1.StreamRecord\x12\x13\n\x0bnext_offset\x18\x02 \x01(\x03\x12\x0c\n\x04\x64one\x18\x03 \x01(\x08"d\n\x15\x44\x65scribeStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef"\x98\x01\n\x16\x44\x65scribeStreamResponse\x12;\n\x06status\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.StreamStatus\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration*\xc8\x01\n\x0cProducerKind\x12\x1d\n\x19PRODUCER_KIND_UNSPECIFIED\x10\x00\x12\x1a\n\x16PRODUCER_KIND_WORKFLOW\x10\x01\x12 \n\x1cPRODUCER_KIND_LOCAL_ACTIVITY\x10\x02\x12\x1a\n\x16PRODUCER_KIND_ACTIVITY\x10\x03\x12\x18\n\x14PRODUCER_KIND_CLIENT\x10\x04\x12%\n!PRODUCER_KIND_STANDALONE_ACTIVITY\x10\x05*\xb1\x01\n\x10StreamRecordKind\x12"\n\x1eSTREAM_RECORD_KIND_UNSPECIFIED\x10\x00\x12\x1b\n\x17STREAM_RECORD_KIND_DATA\x10\x01\x12!\n\x1dSTREAM_RECORD_KIND_SUPERSEDED\x10\x02\x12\x1c\n\x18STREAM_RECORD_KIND_CLOSE\x10\x03\x12\x1b\n\x17STREAM_RECORD_KIND_IDLE\x10\x04*z\n\x0cStreamStatus\x12\x1d\n\x19STREAM_STATUS_UNSPECIFIED\x10\x00\x12\x16\n\x12STREAM_STATUS_OPEN\x10\x01\x12\x18\n\x14STREAM_STATUS_CLOSED\x10\x02\x12\x19\n\x15STREAM_STATUS_EXPIRED\x10\x03*U\n\tReadStart\x12\x1a\n\x16READ_START_UNSPECIFIED\x10\x00\x12\x18\n\x14READ_START_BEGINNING\x10\x01\x12\x12\n\x0eREAD_START_END\x10\x02\x62\x06proto3' + b'\n2temporal/api/streamservice/v1/stream_service.proto\x12\x1dtemporal.api.streamservice.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a%temporal/api/failure/v1/message.proto"]\n\tStreamRef\x12\n\n\x02id\x18\x01 \x01(\t\x12\x13\n\x0bworkflow_id\x18\x02 \x01(\t\x12\x1e\n\x16\x66irst_execution_run_id\x18\x03 \x01(\t\x12\x0f\n\x07\x62\x61\x63kend\x18\x04 \x01(\t"\xc8\x01\n\rProducerEpoch\x12\x1d\n\x15task_started_event_id\x18\x01 \x01(\x03\x12\x14\n\x0ctask_attempt\x18\x02 \x01(\x05\x12\x35\n\x11task_started_time\x18\x05 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12:\n\x16\x61ttempt_scheduled_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x05"\x9e\x01\n\x08Producer\x12\x39\n\x04kind\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.ProducerKind\x12\n\n\x02id\x18\x02 \x01(\t\x12;\n\x05\x65poch\x18\x03 \x01(\x0b\x32,.temporal.api.streamservice.v1.ProducerEpoch\x12\x0e\n\x06run_id\x18\x04 \x01(\t"p\n\x11WorkflowTaskEpoch\x12\x18\n\x10started_event_id\x18\x01 \x01(\x03\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x05\x12\x30\n\x0cstarted_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp"L\n\x0c\x41ppendRecord\x12\r\n\x05topic\x18\x01 \x01(\t\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload"\xb3\x02\n\x0cSupersession\x12\x42\n\rproducer_kind\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.ProducerKind\x12\x13\n\x0bproducer_id\x18\x02 \x01(\t\x12;\n\x05\x65poch\x18\x03 \x01(\x0b\x32,.temporal.api.streamservice.v1.ProducerEpoch\x12Q\n\x17previous_completed_task\x18\x04 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x05 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x06 \x01(\t\x12\x0e\n\x06run_id\x18\x07 \x01(\t"\x89\x03\n\x0cStreamRecord\x12=\n\x04kind\x18\x01 \x01(\x0e\x32/.temporal.api.streamservice.v1.StreamRecordKind\x12\x0e\n\x06offset\x18\x02 \x01(\x03\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x39\n\x08producer\x18\x04 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12-\n\x04\x62ody\x18\x05 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x41\n\x0csupersession\x18\x06 \x01(\x0b\x32+.temporal.api.streamservice.v1.Supersession\x12\x35\n\x0c\x63lose_result\x18\x07 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x37\n\rclose_failure\x18\x08 \x01(\x0b\x32 .temporal.api.failure.v1.Failure"\xae\x01\n\x13\x43reateStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x11\n\tstream_id\x18\x02 \x01(\t\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x14\n\x0cwithdraw_run\x18\x04 \x01(\t\x12-\n\nauto_close\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration"\x9e\x01\n\x14\x43reateStreamResponse\x12\x38\n\x06stream\x18\x01 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x0f\n\x07\x63reated\x18\x02 \x01(\x08\x12;\n\x06status\x18\x03 \x01(\x0e\x32+.temporal.api.streamservice.v1.StreamStatus"\xed\x02\n\x14\x41ppendRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12\x10\n\x08sequence\x18\x04 \x01(\x03\x12<\n\x07records\x18\x05 \x03(\x0b\x32+.temporal.api.streamservice.v1.AppendRecord\x12Q\n\x17previous_completed_task\x18\x06 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x07 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x08 \x01(\t"B\n\x15\x41ppendRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03"\xe9\x02\n\x12\x43loseStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12/\n\x06result\x18\x04 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x31\n\x07\x66\x61ilure\x18\x05 \x01(\x0b\x32 .temporal.api.failure.v1.Failure\x12Q\n\x17previous_completed_task\x18\x06 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x07 \x03(\x03"\x15\n\x13\x43loseStreamResponse"\xd9\x02\n\x12ReadRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x13\n\x0b\x66rom_offset\x18\x03 \x01(\x03\x12\x37\n\x05start\x18\x04 \x01(\x0e\x32(.temporal.api.streamservice.v1.ReadStart\x12\x0e\n\x06topics\x18\x05 \x03(\t\x12\x13\n\x0bmax_records\x18\x06 \x01(\x05\x12\'\n\x04wait\x18\x07 \x01(\x0b\x32\x19.google.protobuf.Duration\x12.\n\x0b\x63reate_wait\x18\x08 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0etopic_prefixes\x18\n \x03(\tJ\x04\x08\t\x10\nR\x0cidle_timeout"v\n\x13ReadRecordsResponse\x12<\n\x07records\x18\x01 \x03(\x0b\x32+.temporal.api.streamservice.v1.StreamRecord\x12\x13\n\x0bnext_offset\x18\x02 \x01(\x03\x12\x0c\n\x04\x64one\x18\x03 \x01(\x08"d\n\x15\x44\x65scribeStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef"\xc7\x01\n\x16\x44\x65scribeStreamResponse\x12;\n\x06status\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.StreamStatus\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12-\n\nauto_close\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration*\xc8\x01\n\x0cProducerKind\x12\x1d\n\x19PRODUCER_KIND_UNSPECIFIED\x10\x00\x12\x1a\n\x16PRODUCER_KIND_WORKFLOW\x10\x01\x12 \n\x1cPRODUCER_KIND_LOCAL_ACTIVITY\x10\x02\x12\x1a\n\x16PRODUCER_KIND_ACTIVITY\x10\x03\x12\x18\n\x14PRODUCER_KIND_CLIENT\x10\x04\x12%\n!PRODUCER_KIND_STANDALONE_ACTIVITY\x10\x05*\xb3\x01\n\x10StreamRecordKind\x12"\n\x1eSTREAM_RECORD_KIND_UNSPECIFIED\x10\x00\x12\x1b\n\x17STREAM_RECORD_KIND_DATA\x10\x01\x12!\n\x1dSTREAM_RECORD_KIND_SUPERSEDED\x10\x02\x12\x1c\n\x18STREAM_RECORD_KIND_CLOSE\x10\x03"\x04\x08\x04\x10\x04*\x17STREAM_RECORD_KIND_IDLE*z\n\x0cStreamStatus\x12\x1d\n\x19STREAM_STATUS_UNSPECIFIED\x10\x00\x12\x16\n\x12STREAM_STATUS_OPEN\x10\x01\x12\x18\n\x14STREAM_STATUS_CLOSED\x10\x02\x12\x19\n\x15STREAM_STATUS_EXPIRED\x10\x03*U\n\tReadStart\x12\x1a\n\x16READ_START_UNSPECIFIED\x10\x00\x12\x18\n\x14READ_START_BEGINNING\x10\x01\x12\x12\n\x0eREAD_START_END\x10\x02\x62\x06proto3' ) _PRODUCERKIND = DESCRIPTOR.enum_types_by_name["ProducerKind"] @@ -47,7 +47,6 @@ STREAM_RECORD_KIND_DATA = 1 STREAM_RECORD_KIND_SUPERSEDED = 2 STREAM_RECORD_KIND_CLOSE = 3 -STREAM_RECORD_KIND_IDLE = 4 STREAM_STATUS_UNSPECIFIED = 0 STREAM_STATUS_OPEN = 1 STREAM_STATUS_CLOSED = 2 @@ -263,46 +262,46 @@ if _descriptor._USE_C_DESCRIPTORS == False: DESCRIPTOR._options = None - _PRODUCERKIND._serialized_start = 3435 - _PRODUCERKIND._serialized_end = 3635 - _STREAMRECORDKIND._serialized_start = 3638 - _STREAMRECORDKIND._serialized_end = 3815 - _STREAMSTATUS._serialized_start = 3817 - _STREAMSTATUS._serialized_end = 3939 - _READSTART._serialized_start = 3941 - _READSTART._serialized_end = 4026 + _PRODUCERKIND._serialized_start = 3518 + _PRODUCERKIND._serialized_end = 3718 + _STREAMRECORDKIND._serialized_start = 3721 + _STREAMRECORDKIND._serialized_end = 3900 + _STREAMSTATUS._serialized_start = 3902 + _STREAMSTATUS._serialized_end = 4024 + _READSTART._serialized_start = 4026 + _READSTART._serialized_end = 4111 _STREAMREF._serialized_start = 227 - _STREAMREF._serialized_end = 303 - _PRODUCEREPOCH._serialized_start = 306 - _PRODUCEREPOCH._serialized_end = 506 - _PRODUCER._serialized_start = 509 - _PRODUCER._serialized_end = 667 - _WORKFLOWTASKEPOCH._serialized_start = 669 - _WORKFLOWTASKEPOCH._serialized_end = 781 - _APPENDRECORD._serialized_start = 783 - _APPENDRECORD._serialized_end = 859 - _SUPERSESSION._serialized_start = 862 - _SUPERSESSION._serialized_end = 1169 - _STREAMRECORD._serialized_start = 1172 - _STREAMRECORD._serialized_end = 1565 - _CREATESTREAMREQUEST._serialized_start = 1567 - _CREATESTREAMREQUEST._serialized_end = 1694 - _CREATESTREAMRESPONSE._serialized_start = 1697 - _CREATESTREAMRESPONSE._serialized_end = 1855 - _APPENDRECORDSREQUEST._serialized_start = 1858 - _APPENDRECORDSREQUEST._serialized_end = 2223 - _APPENDRECORDSRESPONSE._serialized_start = 2225 - _APPENDRECORDSRESPONSE._serialized_end = 2291 - _CLOSESTREAMREQUEST._serialized_start = 2294 - _CLOSESTREAMREQUEST._serialized_end = 2655 - _CLOSESTREAMRESPONSE._serialized_start = 2657 - _CLOSESTREAMRESPONSE._serialized_end = 2678 - _READRECORDSREQUEST._serialized_start = 2681 - _READRECORDSREQUEST._serialized_end = 3055 - _READRECORDSRESPONSE._serialized_start = 3057 - _READRECORDSRESPONSE._serialized_end = 3175 - _DESCRIBESTREAMREQUEST._serialized_start = 3177 - _DESCRIBESTREAMREQUEST._serialized_end = 3277 - _DESCRIBESTREAMRESPONSE._serialized_start = 3280 - _DESCRIBESTREAMRESPONSE._serialized_end = 3432 + _STREAMREF._serialized_end = 320 + _PRODUCEREPOCH._serialized_start = 323 + _PRODUCEREPOCH._serialized_end = 523 + _PRODUCER._serialized_start = 526 + _PRODUCER._serialized_end = 684 + _WORKFLOWTASKEPOCH._serialized_start = 686 + _WORKFLOWTASKEPOCH._serialized_end = 798 + _APPENDRECORD._serialized_start = 800 + _APPENDRECORD._serialized_end = 876 + _SUPERSESSION._serialized_start = 879 + _SUPERSESSION._serialized_end = 1186 + _STREAMRECORD._serialized_start = 1189 + _STREAMRECORD._serialized_end = 1582 + _CREATESTREAMREQUEST._serialized_start = 1585 + _CREATESTREAMREQUEST._serialized_end = 1759 + _CREATESTREAMRESPONSE._serialized_start = 1762 + _CREATESTREAMRESPONSE._serialized_end = 1920 + _APPENDRECORDSREQUEST._serialized_start = 1923 + _APPENDRECORDSREQUEST._serialized_end = 2288 + _APPENDRECORDSRESPONSE._serialized_start = 2290 + _APPENDRECORDSRESPONSE._serialized_end = 2356 + _CLOSESTREAMREQUEST._serialized_start = 2359 + _CLOSESTREAMREQUEST._serialized_end = 2720 + _CLOSESTREAMRESPONSE._serialized_start = 2722 + _CLOSESTREAMRESPONSE._serialized_end = 2743 + _READRECORDSREQUEST._serialized_start = 2746 + _READRECORDSREQUEST._serialized_end = 3091 + _READRECORDSRESPONSE._serialized_start = 3093 + _READRECORDSRESPONSE._serialized_end = 3211 + _DESCRIBESTREAMREQUEST._serialized_start = 3213 + _DESCRIBESTREAMREQUEST._serialized_end = 3313 + _DESCRIBESTREAMRESPONSE._serialized_start = 3316 + _DESCRIBESTREAMRESPONSE._serialized_end = 3515 # @@protoc_insertion_point(module_scope) diff --git a/temporalio/api/streamservice/v1/stream_service_pb2.pyi b/temporalio/api/streamservice/v1/stream_service_pb2.pyi index 5b17a6d1b..6bb44c320 100644 --- a/temporalio/api/streamservice/v1/stream_service_pb2.pyi +++ b/temporalio/api/streamservice/v1/stream_service_pb2.pyi @@ -72,10 +72,6 @@ class _StreamRecordKindEnumTypeWrapper( STREAM_RECORD_KIND_SUPERSEDED: _StreamRecordKind.ValueType # 2 """Lua flags the first batch of a new epoch. The read makes this Record from the flag.""" STREAM_RECORD_KIND_CLOSE: _StreamRecordKind.ValueType # 3 - STREAM_RECORD_KIND_IDLE: _StreamRecordKind.ValueType # 4 - """The last Record of a read that saw no new Record for idle_timeout. Its offset is the - newest offset read + 1, the place to resume from. - """ class StreamRecordKind( _StreamRecordKind, metaclass=_StreamRecordKindEnumTypeWrapper @@ -86,10 +82,6 @@ STREAM_RECORD_KIND_DATA: StreamRecordKind.ValueType # 1 STREAM_RECORD_KIND_SUPERSEDED: StreamRecordKind.ValueType # 2 """Lua flags the first batch of a new epoch. The read makes this Record from the flag.""" STREAM_RECORD_KIND_CLOSE: StreamRecordKind.ValueType # 3 -STREAM_RECORD_KIND_IDLE: StreamRecordKind.ValueType # 4 -"""The last Record of a read that saw no new Record for idle_timeout. Its offset is the -newest offset read + 1, the place to resume from. -""" global___StreamRecordKind = StreamRecordKind class _StreamStatus: @@ -144,22 +136,31 @@ class StreamRef(google.protobuf.message.Message): ID_FIELD_NUMBER: builtins.int WORKFLOW_ID_FIELD_NUMBER: builtins.int FIRST_EXECUTION_RUN_ID_FIELD_NUMBER: builtins.int + BACKEND_FIELD_NUMBER: builtins.int id: builtins.str workflow_id: builtins.str """The Workflow that owns the Stream. Set on a Workflow's ref, empty on a standalone, client or Activity ref. The Store keys on id only and never reads these. """ first_execution_run_id: builtins.str + backend: builtins.str + """The name of the backend that holds the Stream, as the Worker and client config register + it, so every process routes a ref it receives. Empty is the default backend. The Store + never reads it. + """ def __init__( self, *, id: builtins.str = ..., workflow_id: builtins.str = ..., first_execution_run_id: builtins.str = ..., + backend: builtins.str = ..., ) -> None: ... def ClearField( self, field_name: typing_extensions.Literal[ + "backend", + b"backend", "first_execution_run_id", b"first_execution_run_id", "id", @@ -469,7 +470,9 @@ class StreamRecord(google.protobuf.message.Message): global___StreamRecord = StreamRecord class CreateStreamRequest(google.protobuf.message.Message): - """Idempotent for one id and one Retention. ALREADY_EXISTS for another Retention.""" + """Idempotent for one id, one Retention and one auto_close. ALREADY_EXISTS for another + Retention or another auto_close. + """ DESCRIPTOR: google.protobuf.descriptor.Descriptor @@ -477,6 +480,7 @@ class CreateStreamRequest(google.protobuf.message.Message): STREAM_ID_FIELD_NUMBER: builtins.int RETENTION_FIELD_NUMBER: builtins.int WITHDRAW_RUN_FIELD_NUMBER: builtins.int + AUTO_CLOSE_FIELD_NUMBER: builtins.int namespace: builtins.str stream_id: builtins.str """Required from Workflow code.""" @@ -485,6 +489,13 @@ class CreateStreamRequest(google.protobuf.message.Message): """The age after which a Record is dropped. Empty is the Store default.""" withdraw_run: builtins.str """Workflow only. A retry run's first create: the previous run, whose Records it withdraws.""" + @property + def auto_close(self) -> google.protobuf.duration_pb2.Duration: + """Close the Stream with no result when no Record was appended for this long, counted from + the create. Empty never closes on its own. It is separate from Retention and the safety + TTL. The Store applies it on the next access, so no call sees the Stream open after the + period. UNIMPLEMENTED (detail stream_unsupported) on a Store that can't. + """ def __init__( self, *, @@ -492,13 +503,19 @@ class CreateStreamRequest(google.protobuf.message.Message): stream_id: builtins.str = ..., retention: google.protobuf.duration_pb2.Duration | None = ..., withdraw_run: builtins.str = ..., + auto_close: google.protobuf.duration_pb2.Duration | None = ..., ) -> None: ... def HasField( - self, field_name: typing_extensions.Literal["retention", b"retention"] + self, + field_name: typing_extensions.Literal[ + "auto_close", b"auto_close", "retention", b"retention" + ], ) -> builtins.bool: ... def ClearField( self, field_name: typing_extensions.Literal[ + "auto_close", + b"auto_close", "namespace", b"namespace", "retention", @@ -746,7 +763,6 @@ class ReadRecordsRequest(google.protobuf.message.Message): MAX_RECORDS_FIELD_NUMBER: builtins.int WAIT_FIELD_NUMBER: builtins.int CREATE_WAIT_FIELD_NUMBER: builtins.int - IDLE_TIMEOUT_FIELD_NUMBER: builtins.int TOPIC_PREFIXES_FIELD_NUMBER: builtins.int namespace: builtins.str @property @@ -768,11 +784,6 @@ class ReadRecordsRequest(google.protobuf.message.Message): NOT_FOUND. At most 60 s. """ @property - def idle_timeout(self) -> google.protobuf.duration_pb2.Duration: - """End the read with an IDLE Record when no new Record comes for this long. Empty never - ends on idle. The facade counts it across reads, since one read waits at most 60 s. - """ - @property def topic_prefixes( self, ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: @@ -790,20 +801,12 @@ class ReadRecordsRequest(google.protobuf.message.Message): max_records: builtins.int = ..., wait: google.protobuf.duration_pb2.Duration | None = ..., create_wait: google.protobuf.duration_pb2.Duration | None = ..., - idle_timeout: google.protobuf.duration_pb2.Duration | None = ..., topic_prefixes: collections.abc.Iterable[builtins.str] | None = ..., ) -> None: ... def HasField( self, field_name: typing_extensions.Literal[ - "create_wait", - b"create_wait", - "idle_timeout", - b"idle_timeout", - "stream", - b"stream", - "wait", - b"wait", + "create_wait", b"create_wait", "stream", b"stream", "wait", b"wait" ], ) -> builtins.bool: ... def ClearField( @@ -813,8 +816,6 @@ class ReadRecordsRequest(google.protobuf.message.Message): b"create_wait", "from_offset", b"from_offset", - "idle_timeout", - b"idle_timeout", "max_records", b"max_records", "namespace", @@ -897,24 +898,34 @@ class DescribeStreamResponse(google.protobuf.message.Message): STATUS_FIELD_NUMBER: builtins.int LAST_OFFSET_FIELD_NUMBER: builtins.int RETENTION_FIELD_NUMBER: builtins.int + AUTO_CLOSE_FIELD_NUMBER: builtins.int status: global___StreamStatus.ValueType last_offset: builtins.int """Zero when the Stream has no Record.""" @property def retention(self) -> google.protobuf.duration_pb2.Duration: ... + @property + def auto_close(self) -> google.protobuf.duration_pb2.Duration: + """The auto_close of the create. Empty when the Stream has none.""" def __init__( self, *, status: global___StreamStatus.ValueType = ..., last_offset: builtins.int = ..., retention: google.protobuf.duration_pb2.Duration | None = ..., + auto_close: google.protobuf.duration_pb2.Duration | None = ..., ) -> None: ... def HasField( - self, field_name: typing_extensions.Literal["retention", b"retention"] + self, + field_name: typing_extensions.Literal[ + "auto_close", b"auto_close", "retention", b"retention" + ], ) -> builtins.bool: ... def ClearField( self, field_name: typing_extensions.Literal[ + "auto_close", + b"auto_close", "last_offset", b"last_offset", "retention", diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock index 7558e410b..c9268f488 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=3984189fa4d6c809e5489df952b234302e986f76#3984189fa4d6c809e5489df952b234302e986f76" +source = "git+https://github.com/moetemp/sdk-streams?rev=60bf1cee8a522436de4241b33998d7c0e557a700#60bf1cee8a522436de4241b33998d7c0e557a700" dependencies = [ "async-trait", "aws-lc-rs", diff --git a/temporalio/bridge/Cargo.toml b/temporalio/bridge/Cargo.toml index ceab304d6..2388257c1 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 = "3984189fa4d6c809e5489df952b234302e986f76" +rev = "60bf1cee8a522436de4241b33998d7c0e557a700" default-features = false optional = true diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 0d603aff9..acf5cdf94 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -30,7 +30,7 @@ StreamTransportMissingError, StreamUnsupportedError, ) -from temporalio.contrib.streams._items import Closed, Idle, Record, Superseded +from temporalio.contrib.streams._items import Closed, Record, Superseded from temporalio.contrib.streams._outside import OutsideStream from temporalio.contrib.streams._stream import StreamRef, Topic from temporalio.contrib.streams._workflow import ( @@ -42,7 +42,6 @@ __all__ = [ "ClientStream", "Closed", - "Idle", "OutsideStream", "ReadOnlyContextError", "Record", diff --git a/temporalio/contrib/streams/_items.py b/temporalio/contrib/streams/_items.py index 75e3d3797..9cf0f8870 100644 --- a/temporalio/contrib/streams/_items.py +++ b/temporalio/contrib/streams/_items.py @@ -72,18 +72,6 @@ class Closed: """Failure that the close carries. None if it has none.""" -@dataclass(frozen=True) -class Idle: - """The last item of a read that got no new Record for its ``idle_timeout``. - - .. warning:: - This API is experimental. - """ - - offset: int - """The offset to resume from, with ``from_offset``.""" - - # Fields compare in the order of the proto. An empty or negative field compares as zero, as in # the crate's epoch keys. def _n(value: int | None) -> int: diff --git a/temporalio/contrib/streams/_outside.py b/temporalio/contrib/streams/_outside.py index 23dc3aee1..9ebf0f86c 100644 --- a/temporalio/contrib/streams/_outside.py +++ b/temporalio/contrib/streams/_outside.py @@ -110,13 +110,12 @@ def subscribe( topic_prefixes: Sequence[str] = (), from_offset: int | None = None, wait_new_records: bool = True, - idle_timeout: timedelta | None = None, type: Any = None, ) -> AsyncGenerator[Item, None]: """Read the Stream's items in order. - The read ends after a ``Closed`` item, after an ``Idle`` item, or at the newest Record - when ``wait_new_records`` is False. A ``Superseded`` item withdraws earlier Records of + The read ends after a ``Closed`` item, or at the newest Record when + ``wait_new_records`` is False. A ``Superseded`` item withdraws earlier Records of a failed attempt, and it passes every Topic filter. Args: @@ -125,7 +124,6 @@ def subscribe( ``topics`` and no prefixes, the read gives every Topic. from_offset: Start at this offset, inclusive. Defaults to the first Record. wait_new_records: Wait for new Records at the end of the Stream. - idle_timeout: End with an ``Idle`` item when no new Record comes for this long. type: Decode the values of Topics with no type as this type. A Topic from :py:meth:`topic` with a type keeps its own. """ @@ -139,7 +137,6 @@ def subscribe( default_type=type, from_offset=from_offset, wait_new_records=wait_new_records, - idle_timeout=idle_timeout, ) async def _read( @@ -151,7 +148,6 @@ async def _read( default_type: Any, from_offset: int | None, wait_new_records: bool, - idle_timeout: timedelta | None, ) -> AsyncGenerator[Item, None]: items = _reader.read( self._backend.service(self._namespace), @@ -163,7 +159,6 @@ async def _read( default_type=default_type, from_offset=from_offset, wait_new_records=wait_new_records, - idle_timeout=idle_timeout, wait_for_create=self._wait_for_create, run_ended=self._run_ended, ) diff --git a/temporalio/contrib/streams/_reader.py b/temporalio/contrib/streams/_reader.py index 690cffe77..52d8e70f6 100644 --- a/temporalio/contrib/streams/_reader.py +++ b/temporalio/contrib/streams/_reader.py @@ -10,19 +10,17 @@ import temporalio.converter from temporalio.api.streamservice.v1 import ReadStart, StreamRecordKind from temporalio.contrib.streams import _convert, _errors -from temporalio.contrib.streams._items import Closed, Idle, Record, Superseded +from temporalio.contrib.streams._items import Closed, Record, Superseded from temporalio.contrib.streams.system import client as service_client from temporalio.contrib.streams.system import models -Item = Record[Any] | Superseded | Closed | Idle +Item = Record[Any] | Superseded | Closed # The service waits at most 60 s per read, so a longer wait is a loop of reads. _MAX_WAIT = timedelta(seconds=60) # How often a reader that ends with the run checks the run, while it waits for the create # or for new Records. _RUN_CHECK = timedelta(seconds=2) -# The shortest idle time a read asks for, so a read never asks for none. -_MIN_IDLE = timedelta(milliseconds=1) # How long one loop of the generated long poll may run before the facade looks again. _POLL_DEADLINE = timedelta(minutes=5) @@ -38,12 +36,11 @@ async def read( default_type: Any = None, from_offset: int | None, wait_new_records: bool, - idle_timeout: timedelta | None, wait_for_create: timedelta | None, run_ended: Callable[[], Awaitable[bool]] | None = None, ) -> AsyncGenerator[Item, None]: - """Give the items of a Stream in order, until a close, an idle end, or the newest Record - when ``wait_new_records`` is False. + """Give the items of a Stream in order, until a close, or the newest Record when + ``wait_new_records`` is False. """ loop = asyncio.get_running_loop() create_until = ( @@ -54,15 +51,7 @@ async def read( # A reader that ends with the run waits for the create in short steps, so it sees the # run end soon. create_step = _RUN_CHECK if run_ended is not None else _MAX_WAIT - # The idle time counts from the last Record across reads, since a read that ends with the - # run polls in short steps. - idle_since = loop.time() while True: - idle_left = ( - max(idle_timeout - timedelta(seconds=loop.time() - idle_since), _MIN_IDLE) - if idle_timeout is not None - else None - ) create_left = ( timedelta(seconds=max(create_until - loop.time(), 0)) if create_until is not None @@ -74,7 +63,6 @@ async def read( start=None if offset is not None else ReadStart.READ_START_BEGINNING, topics=names, topic_prefixes=list(topic_prefixes), - idle_timeout=idle_left, create_wait=min(create_left, create_step) if create_left else None, ) try: @@ -100,12 +88,11 @@ async def read( if item is None: continue yield item - if isinstance(item, (Closed, Idle)): + if isinstance(item, Closed): return if answer.next_offset is not None: offset = answer.next_offset if answer.records: - idle_since = loop.time() continue if answer.done or not wait_new_records: return @@ -132,6 +119,4 @@ async def _item( if record.kind == StreamRecordKind.STREAM_RECORD_KIND_CLOSE: result = await _convert.decode(converter, record.close_result, None) return Closed(offset=offset, result=result, failure=record.close_failure) - if record.kind == StreamRecordKind.STREAM_RECORD_KIND_IDLE: - return Idle(offset=offset) return None diff --git a/temporalio/contrib/streams/_stream.py b/temporalio/contrib/streams/_stream.py index ee5712975..3c73ce516 100644 --- a/temporalio/contrib/streams/_stream.py +++ b/temporalio/contrib/streams/_stream.py @@ -5,7 +5,6 @@ import abc from collections.abc import AsyncGenerator, Sequence from dataclasses import dataclass -from datetime import timedelta from typing import TYPE_CHECKING, Any, Generic, TypeVar import temporalio.converter @@ -114,7 +113,6 @@ def subscribe( topic_prefixes: Sequence[str] = (), from_offset: int | None = None, wait_new_records: bool = True, - idle_timeout: timedelta | None = None, type: Any = None, ) -> AsyncGenerator[Item, None]: ... @@ -157,7 +155,6 @@ def subscribe( *, from_offset: int | None = None, wait_new_records: bool = True, - idle_timeout: timedelta | None = None, ) -> AsyncGenerator[Item, None]: """Read this Topic only. See the Stream's ``subscribe``.""" if not isinstance(self._stream, _Readable): @@ -166,5 +163,4 @@ def subscribe( topics=[self], from_offset=from_offset, wait_new_records=wait_new_records, - idle_timeout=idle_timeout, ) diff --git a/temporalio/contrib/streams/_workflow.py b/temporalio/contrib/streams/_workflow.py index 7c47b8ce3..bce3e8d0d 100644 --- a/temporalio/contrib/streams/_workflow.py +++ b/temporalio/contrib/streams/_workflow.py @@ -150,9 +150,11 @@ def _publish(self, topic: str, value: Any, force_flush: bool) -> None: self._check("write to") self._ensure_created() self._used = True - append_records( - stream=self._model_ref(), - records=[models.AppendRecord(topic=topic, body=value)], + _now( + append_records( + stream=self._model_ref(), + records=[models.AppendRecord(topic=topic, body=value)], + ) ) def close( @@ -166,7 +168,7 @@ def close( self._check("close") self._ensure_created() self._closed = True - close_stream(stream=self._model_ref(), result=result, failure=failure) + _now(close_stream(stream=self._model_ref(), result=result, failure=failure)) def __enter__(self) -> WorkflowStream: """Use the Stream in a ``with`` block, which closes it at the end.""" diff --git a/temporalio/contrib/streams/proto/v1/__init__.py b/temporalio/contrib/streams/proto/v1/__init__.py index 23717f877..6637363e7 100644 --- a/temporalio/contrib/streams/proto/v1/__init__.py +++ b/temporalio/contrib/streams/proto/v1/__init__.py @@ -7,6 +7,8 @@ CloseWritersResponse, DrainClosesRequest, DrainClosesResponse, + FlushRunBufferRequest, + FlushRunBufferResponse, FlushWriterRequest, FlushWriterResponse, OpenWriterRequest, @@ -16,6 +18,8 @@ RedisStreamConfig, StreamErrorDetails, WorkflowCall, + WriteRecordsRequest, + WriteRecordsResponse, ) __all__ = [ @@ -27,6 +31,8 @@ "CloseWritersResponse", "DrainClosesRequest", "DrainClosesResponse", + "FlushRunBufferRequest", + "FlushRunBufferResponse", "FlushWriterRequest", "FlushWriterResponse", "OpenWriterRequest", @@ -36,4 +42,6 @@ "RedisStreamConfig", "StreamErrorDetails", "WorkflowCall", + "WriteRecordsRequest", + "WriteRecordsResponse", ] diff --git a/temporalio/contrib/streams/proto/v1/extension_pb2.py b/temporalio/contrib/streams/proto/v1/extension_pb2.py index 249d1a422..546cbb679 100644 --- a/temporalio/contrib/streams/proto/v1/extension_pb2.py +++ b/temporalio/contrib/streams/proto/v1/extension_pb2.py @@ -21,7 +21,7 @@ ) DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( - b'\n\'temporal/sdk/streams/v1/extension.proto\x12\x17temporal.sdk.streams.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x32temporal/api/streamservice/v1/stream_service.proto"\xc5\x02\n\x11RedisStreamConfig\x12\x0c\n\x04urls\x18\x01 \x03(\t\x12\x0f\n\x07\x63luster\x18\x02 \x01(\x08\x12\x12\n\nkey_prefix\x18\x03 \x01(\t\x12,\n\tretention\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12-\n\nsafety_ttl\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0e\x64urable_writes\x18\x06 \x01(\x05\x12\x32\n\x0f\x64urable_timeout\x18\x07 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x1f\n\x17\x62locking_reads_per_node\x18\x08 \x01(\x05\x12\x33\n\x10response_timeout\x18\t \x01(\x0b\x32\x19.google.protobuf.Duration"\x91\x01\n\x1a\x42ufferWorkflowCallsRequest\x12\x0e\n\x06run_id\x18\x01 \x01(\t\x12\x14\n\x0cwithdraw_run\x18\x02 \x01(\t\x12\x34\n\x05\x63\x61lls\x18\x03 \x03(\x0b\x32%.temporal.sdk.streams.v1.WorkflowCall\x12\x17\n\x0f\x63hain_continues\x18\x04 \x01(\x08"\x80\x02\n\x0cWorkflowCall\x12\x44\n\x06\x63reate\x18\x01 \x01(\x0b\x32\x32.temporal.api.streamservice.v1.CreateStreamRequestH\x00\x12\x45\n\x06\x61ppend\x18\x02 \x01(\x0b\x32\x33.temporal.api.streamservice.v1.AppendRecordsRequestH\x00\x12\x42\n\x05\x63lose\x18\x03 \x01(\x0b\x32\x31.temporal.api.streamservice.v1.CloseStreamRequestH\x00\x12\x17\n\x0fwhen_run_closes\x18\x04 \x01(\x08\x42\x06\n\x04\x63\x61ll"\x1d\n\x1b\x42ufferWorkflowCallsResponse"\xbb\x02\n\x11OpenWriterRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12\x31\n\x0e\x62\x61tch_interval\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0emax_batch_size\x18\x05 \x01(\x05\x12\x35\n\x12max_retry_duration\x18\x06 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x1c\n\x14max_buffered_records\x18\x07 \x01(\x05"\'\n\x12OpenWriterResponse\x12\x11\n\twriter_id\x18\x01 \x01(\x04"v\n\x0ePublishRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04\x12<\n\x07records\x18\x02 \x03(\x0b\x32+.temporal.api.streamservice.v1.AppendRecord\x12\x13\n\x0b\x66orce_flush\x18\x03 \x01(\x08"\x11\n\x0fPublishResponse"\'\n\x12\x46lushWriterRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04"*\n\x13\x46lushWriterResponse\x12\x13\n\x0blast_offset\x18\x01 \x01(\x03"\'\n\x12\x43loseWriterRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04"*\n\x13\x43loseWriterResponse\x12\x13\n\x0blast_offset\x18\x01 \x01(\x03"\x15\n\x13\x43loseWritersRequest"&\n\x14\x43loseWritersResponse\x12\x0e\n\x06\x66\x61iled\x18\x01 \x01(\x05"@\n\x12\x44rainClosesRequest\x12*\n\x07timeout\x18\x01 \x01(\x0b\x32\x19.google.protobuf.Duration"&\n\x13\x44rainClosesResponse\x12\x0f\n\x07pending\x18\x01 \x01(\x05"$\n\x12StreamErrorDetails\x12\x0e\n\x06reason\x18\x01 \x01(\tb\x06proto3' + b'\n\'temporal/sdk/streams/v1/extension.proto\x12\x17temporal.sdk.streams.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x32temporal/api/streamservice/v1/stream_service.proto"\xd6\x02\n\x11RedisStreamConfig\x12\x0c\n\x04urls\x18\x01 \x03(\t\x12\x0f\n\x07\x63luster\x18\x02 \x01(\x08\x12\x12\n\nkey_prefix\x18\x03 \x01(\t\x12,\n\tretention\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12-\n\nsafety_ttl\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0e\x64urable_writes\x18\x06 \x01(\x05\x12\x32\n\x0f\x64urable_timeout\x18\x07 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x1f\n\x17\x62locking_reads_per_node\x18\x08 \x01(\x05\x12\x33\n\x10response_timeout\x18\t \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x0f\n\x07\x62\x61\x63kend\x18\n \x01(\t"\x91\x01\n\x1a\x42ufferWorkflowCallsRequest\x12\x0e\n\x06run_id\x18\x01 \x01(\t\x12\x14\n\x0cwithdraw_run\x18\x02 \x01(\t\x12\x34\n\x05\x63\x61lls\x18\x03 \x03(\x0b\x32%.temporal.sdk.streams.v1.WorkflowCall\x12\x17\n\x0f\x63hain_continues\x18\x04 \x01(\x08"\x80\x02\n\x0cWorkflowCall\x12\x44\n\x06\x63reate\x18\x01 \x01(\x0b\x32\x32.temporal.api.streamservice.v1.CreateStreamRequestH\x00\x12\x45\n\x06\x61ppend\x18\x02 \x01(\x0b\x32\x33.temporal.api.streamservice.v1.AppendRecordsRequestH\x00\x12\x42\n\x05\x63lose\x18\x03 \x01(\x0b\x32\x31.temporal.api.streamservice.v1.CloseStreamRequestH\x00\x12\x17\n\x0fwhen_run_closes\x18\x04 \x01(\x08\x42\x06\n\x04\x63\x61ll"\x1d\n\x1b\x42ufferWorkflowCallsResponse"\xaf\x01\n\x15\x46lushRunBufferRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x0e\n\x06run_id\x18\x02 \x01(\t\x12G\n\rworkflow_task\x18\x03 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12*\n\x07timeout\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration")\n\x16\x46lushRunBufferResponse\x12\x0f\n\x07records\x18\x01 \x01(\x05"\xbb\x02\n\x11OpenWriterRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12\x31\n\x0e\x62\x61tch_interval\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0emax_batch_size\x18\x05 \x01(\x05\x12\x35\n\x12max_retry_duration\x18\x06 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x1c\n\x14max_buffered_records\x18\x07 \x01(\x05"\'\n\x12OpenWriterResponse\x12\x11\n\twriter_id\x18\x01 \x01(\x04"v\n\x0ePublishRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04\x12<\n\x07records\x18\x02 \x03(\x0b\x32+.temporal.api.streamservice.v1.AppendRecord\x12\x13\n\x0b\x66orce_flush\x18\x03 \x01(\x08"\x11\n\x0fPublishResponse"f\n\x13WriteRecordsRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04\x12<\n\x07records\x18\x02 \x03(\x0b\x32+.temporal.api.streamservice.v1.AppendRecord"A\n\x14WriteRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03"\'\n\x12\x46lushWriterRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04"*\n\x13\x46lushWriterResponse\x12\x13\n\x0blast_offset\x18\x01 \x01(\x03"\'\n\x12\x43loseWriterRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04"*\n\x13\x43loseWriterResponse\x12\x13\n\x0blast_offset\x18\x01 \x01(\x03"\x15\n\x13\x43loseWritersRequest"&\n\x14\x43loseWritersResponse\x12\x0e\n\x06\x66\x61iled\x18\x01 \x01(\x05"@\n\x12\x44rainClosesRequest\x12*\n\x07timeout\x18\x01 \x01(\x0b\x32\x19.google.protobuf.Duration"&\n\x13\x44rainClosesResponse\x12\x0f\n\x07pending\x18\x01 \x01(\x05"$\n\x12StreamErrorDetails\x12\x0e\n\x06reason\x18\x01 \x01(\tb\x06proto3' ) @@ -33,10 +33,14 @@ _BUFFERWORKFLOWCALLSRESPONSE = DESCRIPTOR.message_types_by_name[ "BufferWorkflowCallsResponse" ] +_FLUSHRUNBUFFERREQUEST = DESCRIPTOR.message_types_by_name["FlushRunBufferRequest"] +_FLUSHRUNBUFFERRESPONSE = DESCRIPTOR.message_types_by_name["FlushRunBufferResponse"] _OPENWRITERREQUEST = DESCRIPTOR.message_types_by_name["OpenWriterRequest"] _OPENWRITERRESPONSE = DESCRIPTOR.message_types_by_name["OpenWriterResponse"] _PUBLISHREQUEST = DESCRIPTOR.message_types_by_name["PublishRequest"] _PUBLISHRESPONSE = DESCRIPTOR.message_types_by_name["PublishResponse"] +_WRITERECORDSREQUEST = DESCRIPTOR.message_types_by_name["WriteRecordsRequest"] +_WRITERECORDSRESPONSE = DESCRIPTOR.message_types_by_name["WriteRecordsResponse"] _FLUSHWRITERREQUEST = DESCRIPTOR.message_types_by_name["FlushWriterRequest"] _FLUSHWRITERRESPONSE = DESCRIPTOR.message_types_by_name["FlushWriterResponse"] _CLOSEWRITERREQUEST = DESCRIPTOR.message_types_by_name["CloseWriterRequest"] @@ -90,6 +94,28 @@ ) _sym_db.RegisterMessage(BufferWorkflowCallsResponse) +FlushRunBufferRequest = _reflection.GeneratedProtocolMessageType( + "FlushRunBufferRequest", + (_message.Message,), + { + "DESCRIPTOR": _FLUSHRUNBUFFERREQUEST, + "__module__": "temporalio.contrib.streams.proto.v1.extension_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.FlushRunBufferRequest) + }, +) +_sym_db.RegisterMessage(FlushRunBufferRequest) + +FlushRunBufferResponse = _reflection.GeneratedProtocolMessageType( + "FlushRunBufferResponse", + (_message.Message,), + { + "DESCRIPTOR": _FLUSHRUNBUFFERRESPONSE, + "__module__": "temporalio.contrib.streams.proto.v1.extension_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.FlushRunBufferResponse) + }, +) +_sym_db.RegisterMessage(FlushRunBufferResponse) + OpenWriterRequest = _reflection.GeneratedProtocolMessageType( "OpenWriterRequest", (_message.Message,), @@ -134,6 +160,28 @@ ) _sym_db.RegisterMessage(PublishResponse) +WriteRecordsRequest = _reflection.GeneratedProtocolMessageType( + "WriteRecordsRequest", + (_message.Message,), + { + "DESCRIPTOR": _WRITERECORDSREQUEST, + "__module__": "temporalio.contrib.streams.proto.v1.extension_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.WriteRecordsRequest) + }, +) +_sym_db.RegisterMessage(WriteRecordsRequest) + +WriteRecordsResponse = _reflection.GeneratedProtocolMessageType( + "WriteRecordsResponse", + (_message.Message,), + { + "DESCRIPTOR": _WRITERECORDSRESPONSE, + "__module__": "temporalio.contrib.streams.proto.v1.extension_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.WriteRecordsResponse) + }, +) +_sym_db.RegisterMessage(WriteRecordsResponse) + FlushWriterRequest = _reflection.GeneratedProtocolMessageType( "FlushWriterRequest", (_message.Message,), @@ -236,37 +284,45 @@ if _descriptor._USE_C_DESCRIPTORS == False: DESCRIPTOR._options = None _REDISSTREAMCONFIG._serialized_start = 153 - _REDISSTREAMCONFIG._serialized_end = 478 - _BUFFERWORKFLOWCALLSREQUEST._serialized_start = 481 - _BUFFERWORKFLOWCALLSREQUEST._serialized_end = 626 - _WORKFLOWCALL._serialized_start = 629 - _WORKFLOWCALL._serialized_end = 885 - _BUFFERWORKFLOWCALLSRESPONSE._serialized_start = 887 - _BUFFERWORKFLOWCALLSRESPONSE._serialized_end = 916 - _OPENWRITERREQUEST._serialized_start = 919 - _OPENWRITERREQUEST._serialized_end = 1234 - _OPENWRITERRESPONSE._serialized_start = 1236 - _OPENWRITERRESPONSE._serialized_end = 1275 - _PUBLISHREQUEST._serialized_start = 1277 - _PUBLISHREQUEST._serialized_end = 1395 - _PUBLISHRESPONSE._serialized_start = 1397 - _PUBLISHRESPONSE._serialized_end = 1414 - _FLUSHWRITERREQUEST._serialized_start = 1416 - _FLUSHWRITERREQUEST._serialized_end = 1455 - _FLUSHWRITERRESPONSE._serialized_start = 1457 - _FLUSHWRITERRESPONSE._serialized_end = 1499 - _CLOSEWRITERREQUEST._serialized_start = 1501 - _CLOSEWRITERREQUEST._serialized_end = 1540 - _CLOSEWRITERRESPONSE._serialized_start = 1542 - _CLOSEWRITERRESPONSE._serialized_end = 1584 - _CLOSEWRITERSREQUEST._serialized_start = 1586 - _CLOSEWRITERSREQUEST._serialized_end = 1607 - _CLOSEWRITERSRESPONSE._serialized_start = 1609 - _CLOSEWRITERSRESPONSE._serialized_end = 1647 - _DRAINCLOSESREQUEST._serialized_start = 1649 - _DRAINCLOSESREQUEST._serialized_end = 1713 - _DRAINCLOSESRESPONSE._serialized_start = 1715 - _DRAINCLOSESRESPONSE._serialized_end = 1753 - _STREAMERRORDETAILS._serialized_start = 1755 - _STREAMERRORDETAILS._serialized_end = 1791 + _REDISSTREAMCONFIG._serialized_end = 495 + _BUFFERWORKFLOWCALLSREQUEST._serialized_start = 498 + _BUFFERWORKFLOWCALLSREQUEST._serialized_end = 643 + _WORKFLOWCALL._serialized_start = 646 + _WORKFLOWCALL._serialized_end = 902 + _BUFFERWORKFLOWCALLSRESPONSE._serialized_start = 904 + _BUFFERWORKFLOWCALLSRESPONSE._serialized_end = 933 + _FLUSHRUNBUFFERREQUEST._serialized_start = 936 + _FLUSHRUNBUFFERREQUEST._serialized_end = 1111 + _FLUSHRUNBUFFERRESPONSE._serialized_start = 1113 + _FLUSHRUNBUFFERRESPONSE._serialized_end = 1154 + _OPENWRITERREQUEST._serialized_start = 1157 + _OPENWRITERREQUEST._serialized_end = 1472 + _OPENWRITERRESPONSE._serialized_start = 1474 + _OPENWRITERRESPONSE._serialized_end = 1513 + _PUBLISHREQUEST._serialized_start = 1515 + _PUBLISHREQUEST._serialized_end = 1633 + _PUBLISHRESPONSE._serialized_start = 1635 + _PUBLISHRESPONSE._serialized_end = 1652 + _WRITERECORDSREQUEST._serialized_start = 1654 + _WRITERECORDSREQUEST._serialized_end = 1756 + _WRITERECORDSRESPONSE._serialized_start = 1758 + _WRITERECORDSRESPONSE._serialized_end = 1823 + _FLUSHWRITERREQUEST._serialized_start = 1825 + _FLUSHWRITERREQUEST._serialized_end = 1864 + _FLUSHWRITERRESPONSE._serialized_start = 1866 + _FLUSHWRITERRESPONSE._serialized_end = 1908 + _CLOSEWRITERREQUEST._serialized_start = 1910 + _CLOSEWRITERREQUEST._serialized_end = 1949 + _CLOSEWRITERRESPONSE._serialized_start = 1951 + _CLOSEWRITERRESPONSE._serialized_end = 1993 + _CLOSEWRITERSREQUEST._serialized_start = 1995 + _CLOSEWRITERSREQUEST._serialized_end = 2016 + _CLOSEWRITERSRESPONSE._serialized_start = 2018 + _CLOSEWRITERSRESPONSE._serialized_end = 2056 + _DRAINCLOSESREQUEST._serialized_start = 2058 + _DRAINCLOSESREQUEST._serialized_end = 2122 + _DRAINCLOSESRESPONSE._serialized_start = 2124 + _DRAINCLOSESRESPONSE._serialized_end = 2162 + _STREAMERRORDETAILS._serialized_start = 2164 + _STREAMERRORDETAILS._serialized_end = 2200 # @@protoc_insertion_point(module_scope) diff --git a/temporalio/contrib/streams/proto/v1/extension_pb2.pyi b/temporalio/contrib/streams/proto/v1/extension_pb2.pyi index 798d56528..fad30118e 100644 --- a/temporalio/contrib/streams/proto/v1/extension_pb2.pyi +++ b/temporalio/contrib/streams/proto/v1/extension_pb2.pyi @@ -38,6 +38,7 @@ class RedisStreamConfig(google.protobuf.message.Message): DURABLE_TIMEOUT_FIELD_NUMBER: builtins.int BLOCKING_READS_PER_NODE_FIELD_NUMBER: builtins.int RESPONSE_TIMEOUT_FIELD_NUMBER: builtins.int + BACKEND_FIELD_NUMBER: builtins.int @property def urls( self, @@ -62,6 +63,10 @@ class RedisStreamConfig(google.protobuf.message.Message): @property def response_timeout(self) -> google.protobuf.duration_pb2.Duration: """Empty is 5 s.""" + backend: builtins.str + """The backend name that lang registered this extension under. The crate puts it in each + StreamRef it builds, such as the one in a create answer. + """ def __init__( self, *, @@ -74,6 +79,7 @@ class RedisStreamConfig(google.protobuf.message.Message): durable_timeout: google.protobuf.duration_pb2.Duration | None = ..., blocking_reads_per_node: builtins.int = ..., response_timeout: google.protobuf.duration_pb2.Duration | None = ..., + backend: builtins.str = ..., ) -> None: ... def HasField( self, @@ -91,6 +97,8 @@ class RedisStreamConfig(google.protobuf.message.Message): def ClearField( self, field_name: typing_extensions.Literal[ + "backend", + b"backend", "blocking_reads_per_node", b"blocking_reads_per_node", "cluster", @@ -239,6 +247,82 @@ class BufferWorkflowCallsResponse(google.protobuf.message.Message): global___BufferWorkflowCallsResponse = BufferWorkflowCallsResponse +class FlushRunBufferRequest(google.protobuf.message.Message): + """Method `FlushRunBuffer`, asynchronous. Lang calls it before it runs a Local Activity, also + before each retry that Core runs in the same task. The crate writes the run's buffered + creates and appends in the epoch of the Workflow Task that dispatched the Local Activity, so + the run's Records come before the Local Activity's. Closes still wait for an accepted + completion. The task's completion writes only what came later, with the numbering going on, + and sends the task's withdrawal facts to each Stream this call wrote. A Local Activity of a + task that already completed, or of a task older than an earlier flush, writes nothing. When + the call fails, the run's next completion fails with the same error, so the retry withdraws + what was written. Lang then fails the Local Activity attempt. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + RUN_ID_FIELD_NUMBER: builtins.int + WORKFLOW_TASK_FIELD_NUMBER: builtins.int + TIMEOUT_FIELD_NUMBER: builtins.int + namespace: builtins.str + run_id: builtins.str + @property + def workflow_task( + self, + ) -> temporalio.api.streamservice.v1.stream_service_pb2.WorkflowTaskEpoch: + """`ActivityTask.Start.workflow_task` (field 20) of the Local Activity.""" + @property + def timeout(self) -> google.protobuf.duration_pb2.Duration: + """How long the write may take. Empty is 10 s.""" + def __init__( + self, + *, + namespace: builtins.str = ..., + run_id: builtins.str = ..., + workflow_task: temporalio.api.streamservice.v1.stream_service_pb2.WorkflowTaskEpoch + | None = ..., + timeout: google.protobuf.duration_pb2.Duration | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "timeout", b"timeout", "workflow_task", b"workflow_task" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "namespace", + b"namespace", + "run_id", + b"run_id", + "timeout", + b"timeout", + "workflow_task", + b"workflow_task", + ], + ) -> None: ... + +global___FlushRunBufferRequest = FlushRunBufferRequest + +class FlushRunBufferResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + RECORDS_FIELD_NUMBER: builtins.int + records: builtins.int + """The Records the call wrote. Zero when the run had nothing buffered.""" + def __init__( + self, + *, + records: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["records", b"records"] + ) -> None: ... + +global___FlushRunBufferResponse = FlushRunBufferResponse + class OpenWriterRequest(google.protobuf.message.Message): """Method `OpenWriter`, asynchronous. A buffered writer for an Activity or a client Producer.""" @@ -384,6 +468,67 @@ class PublishResponse(google.protobuf.message.Message): global___PublishResponse = PublishResponse +class WriteRecordsRequest(google.protobuf.message.Message): + """Method `WriteRecords`, asynchronous. Appends Records at once for an open writer, after the + Records that `Publish` buffered, and answers when they are in the Store. It keeps the writer's + Producer epoch and `sequence` order, and a Store failure resends the same batch, so it goes in + once. Records past the Store's batch limit go as more batches, in order. An awaited publish of + an Activity or a client uses this call, so it needs no interval flush. A failure after the + Store retries fails the writer, as a failed flush does. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WRITER_ID_FIELD_NUMBER: builtins.int + RECORDS_FIELD_NUMBER: builtins.int + writer_id: builtins.int + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + temporalio.api.streamservice.v1.stream_service_pb2.AppendRecord + ]: ... + def __init__( + self, + *, + writer_id: builtins.int = ..., + records: collections.abc.Iterable[ + temporalio.api.streamservice.v1.stream_service_pb2.AppendRecord + ] + | None = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "records", b"records", "writer_id", b"writer_id" + ], + ) -> None: ... + +global___WriteRecordsRequest = WriteRecordsRequest + +class WriteRecordsResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FIRST_OFFSET_FIELD_NUMBER: builtins.int + LAST_OFFSET_FIELD_NUMBER: builtins.int + first_offset: builtins.int + """The offsets of the first and the last Record. Zero when the request has no Record.""" + last_offset: builtins.int + def __init__( + self, + *, + first_offset: builtins.int = ..., + last_offset: builtins.int = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "first_offset", b"first_offset", "last_offset", b"last_offset" + ], + ) -> None: ... + +global___WriteRecordsResponse = WriteRecordsResponse + class FlushWriterRequest(google.protobuf.message.Message): """Method `FlushWriter`, asynchronous. Waits until the Records published so far are in the Store.""" diff --git a/temporalio/contrib/streams/system/client.py b/temporalio/contrib/streams/system/client.py index e8164de3a..8ecfc6238 100644 --- a/temporalio/contrib/streams/system/client.py +++ b/temporalio/contrib/streams/system/client.py @@ -169,7 +169,9 @@ async def create_stream(self, request: CreateStreamRequest) -> CreateStreamRespo async def append_records( self, request: AppendRecordsRequest ) -> AppendRecordsResponse: - """In Workflow code, the Brook transport puts this call into a buffer.""" + """In Workflow code, the Brook transport puts this call into the run's buffer, and + the Workflow facade drops the response. + """ payload = _proto_payload( _AppendRecordsRequestTransferTypeConverter().to_transfer_type(request) ) @@ -215,8 +217,6 @@ async def read_records_until_records( Each call sets ``wait`` to what is left of the deadline, at most 60 s. After an answer with nothing, the next call sets ``from_offset`` to the answer's ``next_offset``. - Each call sets ``idle_timeout`` to what is left of the request's, so the idle - time counts across calls. When the deadline passes, the last answer comes back, so the caller keeps its place. A ``TransportError`` with ``retryable`` set is tried again while the deadline @@ -225,11 +225,6 @@ async def read_records_until_records( clock = asyncio.get_running_loop() started = clock.time() end = started + deadline.total_seconds() - idle_timeout = ( - request.idle_timeout.total_seconds() - if request.idle_timeout is not None - else None - ) attempt: ReadRecordsRequest = request pause = _LONG_POLL_FIRST_PAUSE while True: @@ -238,11 +233,6 @@ async def read_records_until_records( attempt = dataclasses.replace( attempt, wait=datetime.timedelta(seconds=wait) ) - if idle_timeout is not None: - idle_left = max(idle_timeout - (clock.time() - started), 0.001) - attempt = dataclasses.replace( - attempt, idle_timeout=datetime.timedelta(seconds=idle_left) - ) asked = clock.time() try: answer = await self.read_records(attempt) diff --git a/temporalio/contrib/streams/system/models.py b/temporalio/contrib/streams/system/models.py index 605fb9855..23b805dc7 100644 --- a/temporalio/contrib/streams/system/models.py +++ b/temporalio/contrib/streams/system/models.py @@ -23,13 +23,14 @@ @dataclasses.dataclass(slots=True) class CreateStreamRequest: - """Create a Stream. Idempotent for one id and one Retention. ALREADY_EXISTS for another - Retention. + """Create a Stream. Idempotent for one id, one Retention and one auto-close. + ALREADY_EXISTS for another Retention or another auto-close. """ stream_id: str retention: datetime.timedelta | None = None withdraw_run: str | None = None + auto_close: datetime.timedelta | None = None class _CreateStreamRequestTransferTypeConverter( @@ -58,6 +59,9 @@ def from_transfer_type( if value.HasField("retention") else None, withdraw_run=value.withdraw_run if bool(value.withdraw_run) else None, + auto_close=_support.duration_from_proto(value.auto_close) + if value.HasField("auto_close") + else None, ) @typing_extensions.override @@ -73,6 +77,8 @@ def to_transfer_type( message.retention.CopyFrom(_support.duration_to_proto(value.retention)) if value.withdraw_run is not None: message.withdraw_run = value.withdraw_run + if value.auto_close is not None: + message.auto_close.CopyFrom(_support.duration_to_proto(value.auto_close)) return message @@ -154,6 +160,7 @@ class StreamRef: id: str workflow_id: str | None = None first_execution_run_id: str | None = None + backend: str | None = None class _StreamRefTransferTypeConverter( @@ -180,6 +187,7 @@ def from_transfer_type( first_execution_run_id=value.first_execution_run_id if bool(value.first_execution_run_id) else None, + backend=value.backend if bool(value.backend) else None, ) @typing_extensions.override @@ -193,6 +201,8 @@ def to_transfer_type( message.workflow_id = value.workflow_id if value.first_execution_run_id is not None: message.first_execution_run_id = value.first_execution_run_id + if value.backend is not None: + message.backend = value.backend return message @@ -792,7 +802,6 @@ class ReadRecordsRequest: max_records: int | None = None wait: datetime.timedelta | None = None create_wait: datetime.timedelta | None = None - idle_timeout: datetime.timedelta | None = None class _ReadRecordsRequestTransferTypeConverter( @@ -833,9 +842,6 @@ def from_transfer_type( create_wait=_support.duration_from_proto(value.create_wait) if value.HasField("create_wait") else None, - idle_timeout=_support.duration_from_proto(value.idle_timeout) - if value.HasField("idle_timeout") - else None, ) @typing_extensions.override @@ -868,10 +874,6 @@ def to_transfer_type( message.wait.CopyFrom(_support.duration_to_proto(value.wait)) if value.create_wait is not None: message.create_wait.CopyFrom(_support.duration_to_proto(value.create_wait)) - if value.idle_timeout is not None: - message.idle_timeout.CopyFrom( - _support.duration_to_proto(value.idle_timeout) - ) return message @@ -1189,6 +1191,7 @@ class DescribeStreamResponse: ) = None last_offset: int | None = None retention: datetime.timedelta | None = None + auto_close: datetime.timedelta | None = None class _DescribeStreamResponseTransferTypeConverter( @@ -1218,6 +1221,9 @@ def from_transfer_type( retention=_support.duration_from_proto(value.retention) if value.HasField("retention") else None, + auto_close=_support.duration_from_proto(value.auto_close) + if value.HasField("auto_close") + else None, ) @typing_extensions.override @@ -1236,6 +1242,8 @@ def to_transfer_type( message.last_offset = value.last_offset if value.retention is not None: message.retention.CopyFrom(_support.duration_to_proto(value.retention)) + if value.auto_close is not None: + message.auto_close.CopyFrom(_support.duration_to_proto(value.auto_close)) return message diff --git a/temporalio/contrib/streams/system/operations/append_records.py b/temporalio/contrib/streams/system/operations/append_records.py index c4c06715f..0d4a1a574 100644 --- a/temporalio/contrib/streams/system/operations/append_records.py +++ b/temporalio/contrib/streams/system/operations/append_records.py @@ -14,31 +14,21 @@ ) -def _append_records( +async def _append_records( request: AppendRecordsRequest, -) -> None: +) -> temporalio.workflow.NexusOperationHandle[AppendRecordsResponse,]: nexus_client = temporalio.workflow.create_nexus_client( service="temporal.api.streamservice.v1.StreamService", endpoint="__temporal_system", ) - start = nexus_client.start_operation( + return await nexus_client.start_operation( operation="AppendRecords", input=request, output_type=AppendRecordsResponse, ) - try: - _ = start.send(None) - except StopIteration: - return - start.close() - raise RuntimeError( - "one-way operation AppendRecords did not complete at once. " - "It needs an outbound interceptor that takes the call, such as one that " - "buffers it." - ) -def append_records( +async def append_records( *, stream: StreamRef | None = None, producer: Producer | None = None, @@ -47,8 +37,9 @@ def append_records( previous_completed_task: WorkflowTaskEpoch | None = None, failed_tasks: list[int] | None = None, withdraw_run: str | None = None, -) -> None: - """In Workflow code, the Brook transport puts this call into a buffer. +) -> temporalio.workflow.NexusOperationHandle[AppendRecordsResponse,]: + """In Workflow code, the Brook transport puts this call into the run's buffer, and the + Workflow facade drops the response. Args: producer: Required outside a Workflow. In a Workflow the transport sets it. @@ -70,4 +61,4 @@ def append_records( failed_tasks=failed_tasks or [], withdraw_run=withdraw_run, ) - _append_records(request) + return await _append_records(request) diff --git a/temporalio/contrib/streams/system/operations/close_stream.py b/temporalio/contrib/streams/system/operations/close_stream.py index 8f686b69f..80b8b5d09 100644 --- a/temporalio/contrib/streams/system/operations/close_stream.py +++ b/temporalio/contrib/streams/system/operations/close_stream.py @@ -15,31 +15,21 @@ ) -def _close_stream( +async def _close_stream( request: CloseStreamRequest, -) -> None: +) -> temporalio.workflow.NexusOperationHandle[CloseStreamResponse,]: nexus_client = temporalio.workflow.create_nexus_client( service="temporal.api.streamservice.v1.StreamService", endpoint="__temporal_system", ) - start = nexus_client.start_operation( + return await nexus_client.start_operation( operation="CloseStream", input=request, output_type=CloseStreamResponse, ) - try: - _ = start.send(None) - except StopIteration: - return - start.close() - raise RuntimeError( - "one-way operation CloseStream did not complete at once. " - "It needs an outbound interceptor that takes the call, such as one that " - "buffers it." - ) -def close_stream( +async def close_stream( *, stream: StreamRef | None = None, producer: Producer | None = None, @@ -47,7 +37,7 @@ def close_stream( failure: BaseException | None = None, previous_completed_task: WorkflowTaskEpoch | None = None, failed_tasks: list[int] | None = None, -) -> None: +) -> temporalio.workflow.NexusOperationHandle[CloseStreamResponse,]: """ Args: previous_completed_task: Workflow only. The previous completed Workflow Task of @@ -63,4 +53,4 @@ def close_stream( previous_completed_task=previous_completed_task, failed_tasks=failed_tasks or [], ) - _close_stream(request) + return await _close_stream(request) diff --git a/temporalio/contrib/streams/system/operations/create_stream.py b/temporalio/contrib/streams/system/operations/create_stream.py index 14c0dc32d..b798ea98e 100644 --- a/temporalio/contrib/streams/system/operations/create_stream.py +++ b/temporalio/contrib/streams/system/operations/create_stream.py @@ -31,6 +31,7 @@ async def create_stream( stream_id: str, retention: datetime.timedelta | None = None, withdraw_run: str | None = None, + auto_close: datetime.timedelta | None = None, ) -> temporalio.workflow.NexusOperationHandle[CreateStreamResponse,]: """ Args: @@ -39,10 +40,13 @@ async def create_stream( retention: The age after which a Record is dropped. Empty is the Store default. withdraw_run: Workflow only. A retry run's first batch or create: the previous run, whose Records it withdraws. + auto_close: Close the Stream with no result when no Record was appended for this + long. Empty never closes on its own. It is separate from Retention. """ request = CreateStreamRequest( stream_id=stream_id, retention=retention, withdraw_run=withdraw_run, + auto_close=auto_close, ) return await _create_stream(request) diff --git a/temporalio/contrib/streams/system/operations/read_records.py b/temporalio/contrib/streams/system/operations/read_records.py index d0cb4e460..8e5675994 100644 --- a/temporalio/contrib/streams/system/operations/read_records.py +++ b/temporalio/contrib/streams/system/operations/read_records.py @@ -39,7 +39,6 @@ async def read_records( max_records: int | None = None, wait: datetime.timedelta | None = None, create_wait: datetime.timedelta | None = None, - idle_timeout: datetime.timedelta | None = None, ) -> temporalio.workflow.NexusOperationHandle[ReadRecordsResponse,]: """Not for Workflow code in Brook. A long poll. NOT_FOUND for an uncreated Stream. @@ -54,8 +53,6 @@ async def read_records( wait: How long to wait for a new Record before an empty answer. At most 60 s. create_wait: How long to wait for an uncreated Stream. Empty fails at once. At most 60 s. - idle_timeout: End the read with an IDLE Record when no new Record comes for this - long. Empty never ends on idle. The facade counts it across reads. """ request = ReadRecordsRequest( stream=stream, @@ -66,6 +63,5 @@ async def read_records( max_records=max_records, wait=wait, create_wait=create_wait, - idle_timeout=idle_timeout, ) return await _read_records(request) diff --git a/tests/contrib/streams/fake_extension.py b/tests/contrib/streams/fake_extension.py index 5c9dc5bde..31eaeeaf4 100644 --- a/tests/contrib/streams/fake_extension.py +++ b/tests/contrib/streams/fake_extension.py @@ -116,8 +116,7 @@ async def _read( stream = self.streams[request.stream.id] offset = request.from_offset or 1 wait = request.wait.ToTimedelta().total_seconds() - idle = request.idle_timeout.ToTimedelta().total_seconds() - until = loop.time() + (min(wait, idle) if idle else wait) + until = loop.time() + wait while True: found = [ record @@ -140,14 +139,6 @@ async def _read( ) left = until - loop.time() if left <= 0: - if idle and wait >= idle: - idle_record = stream_pb.StreamRecord( - kind=stream_pb.StreamRecordKind.STREAM_RECORD_KIND_IDLE, - offset=offset, - ) - return stream_pb.ReadRecordsResponse( - records=[idle_record], next_offset=offset, done=True - ) return stream_pb.ReadRecordsResponse(next_offset=offset) try: await asyncio.wait_for(stream.changed.wait(), left) diff --git a/tests/contrib/streams/test_bindings.py b/tests/contrib/streams/test_bindings.py index cdbdf1701..0bfc603b4 100644 --- a/tests/contrib/streams/test_bindings.py +++ b/tests/contrib/streams/test_bindings.py @@ -112,8 +112,8 @@ def test_extension_protos_reference_the_service_protos(): assert parsed.calls[0].append.stream.id == "wf:run" -def test_one_way_stub_needs_an_interceptor(): +async def test_append_stub_runs_only_in_a_workflow(): from temporalio.contrib.streams.system import append_records with pytest.raises(Exception): - append_records(stream=models.StreamRef(id="wf:run")) + await append_records(stream=models.StreamRef(id="wf:run")) diff --git a/tests/contrib/streams/test_facade.py b/tests/contrib/streams/test_facade.py index 0de124be1..43b38d571 100644 --- a/tests/contrib/streams/test_facade.py +++ b/tests/contrib/streams/test_facade.py @@ -122,23 +122,6 @@ async def test_client_writes_reads_and_closes(client: Client): assert [item.offset for item in resumed] == [second.offset, closed.offset] -async def test_read_ends_idle_and_resumes(client: Client): - client, _ = fake_client(client) - stream = await streams.create_stream(client, "idle-1") - stream.topic("t", type=int).publish(1) - await stream.flush() - items = [ - item async for item in stream.subscribe(idle_timeout=timedelta(milliseconds=50)) - ] - assert isinstance(items[-1], streams.Idle) - assert [ - item - async for item in stream.subscribe( - from_offset=items[-1].offset, wait_new_records=False - ) - ] == [] - - async def test_read_waits_for_create(client: Client): client, _ = fake_client(client) handle = client.get_workflow_handle("wf-1", first_execution_run_id="run-1") diff --git a/tests/contrib/streams/test_redis.py b/tests/contrib/streams/test_redis.py index 7bd9586a2..17bb8bae6 100644 --- a/tests/contrib/streams/test_redis.py +++ b/tests/contrib/streams/test_redis.py @@ -122,7 +122,7 @@ async def run(self, prompt: str) -> str: async def read_all(stream: streams.ClientStream) -> list[streams.Record | object]: - return [item async for item in stream.subscribe(idle_timeout=timedelta(seconds=20))] + return [item async for item in stream.subscribe()] async def test_workflow_and_activity_write_the_default_stream(stream_client: Client): @@ -604,27 +604,6 @@ async def test_records_go_through_the_payload_codec(stream_client: Client): assert first.data.payload.metadata["encoding"] == b"json/plain" -async def test_read_ends_idle_and_resumes(stream_client: Client): - stream = await streams.create_stream(stream_client, f"idle-{uuid.uuid4()}") - stream.topic("t", type=int).publish(1) - await stream.flush() - items = [ - item - async for item in stream.subscribe(idle_timeout=timedelta(milliseconds=200)) - ] - assert isinstance(items[-1], streams.Idle) - stream.topic("t", type=int).publish(2) - await stream.flush() - resumed = [ - item - async for item in stream.subscribe( - from_offset=items[-1].offset, wait_new_records=False - ) - ] - assert [item.data for item in resumed if isinstance(item, streams.Record)] == [2] - assert await stream.get_offset() > items[-1].offset - - @workflow.defn class WaitsForSignal: def __init__(self) -> None: @@ -852,24 +831,6 @@ async def test_for_ref_ends_with_the_owning_run(stream_client: Client): assert asyncio.get_running_loop().time() - start < 30 -async def test_idle_ends_a_read_that_ends_with_the_run(stream_client: Client): - async with new_worker(stream_client, WaitsForever) as worker: - handle = await stream_client.start_workflow( - WaitsForever.run, id=f"wf-{uuid.uuid4()}", task_queue=worker.task_queue - ) - stream = streams.for_handle( - handle, end_with_run=True, wait_for_create=timedelta(seconds=20) - ) - # The read polls in short steps to look at the run, and the idle time still counts - # across them. - items = [ - item async for item in stream.subscribe(idle_timeout=timedelta(seconds=3)) - ] - await handle.terminate() - assert isinstance(items[0], streams.Record) - assert isinstance(items[-1], streams.Idle) - - async def test_workflow_calls_carry_the_owner(stream_client: Client): [plugin] = [ p for p in stream_client.config()["plugins"] if isinstance(p, StreamPlugin) From f7ef0282be6e0327a25b317f04711d035c356ceb Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 21 Oct 2026 20:08:53 -0700 Subject: [PATCH 2/5] Moved the crate pin to the s14 head that names the Stream in errors. --- temporalio/bridge/Cargo.lock | 2 +- temporalio/bridge/Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock index c9268f488..97bed84dd 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=60bf1cee8a522436de4241b33998d7c0e557a700#60bf1cee8a522436de4241b33998d7c0e557a700" +source = "git+https://github.com/moetemp/sdk-streams?rev=6254405544278746d98ac4bbb1f748ce174d1dc2#6254405544278746d98ac4bbb1f748ce174d1dc2" dependencies = [ "async-trait", "aws-lc-rs", diff --git a/temporalio/bridge/Cargo.toml b/temporalio/bridge/Cargo.toml index 2388257c1..b5ab00c00 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 = "60bf1cee8a522436de4241b33998d7c0e557a700" +rev = "6254405544278746d98ac4bbb1f748ce174d1dc2" default-features = false optional = true From 0e24e5a896b04c952b799c7c1f952d6a9326feb8 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 21 Oct 2026 23:23:41 -0700 Subject: [PATCH 3/5] Waited for the Worker's run task with no fixed timeout. --- temporalio/worker/_worker.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/temporalio/worker/_worker.py b/temporalio/worker/_worker.py index 655c00cd0..4a3a1d850 100644 --- a/temporalio/worker/_worker.py +++ b/temporalio/worker/_worker.py @@ -1000,18 +1000,17 @@ async def __aexit__(self, exc_type: type[BaseException] | None, *args: Any) -> N # that code finish, so a plugin can still await calls in its cleanup. self._async_context_exiting = True run_task = self._async_context_run_task - await asyncio.wait([run_task], timeout=_RUN_TASK_EXIT_TIMEOUT.total_seconds()) - if not run_task.done(): + try: + # wait() doesn't raise for the task's own outcome, only when this call is cancelled. + await asyncio.wait([run_task]) + except asyncio.CancelledError: run_task.cancel() + raise # Only re-raise our exception if present and exc_type is cancel if exc_type is asyncio.CancelledError and self._async_context_run_exception: raise self._async_context_run_exception -# How long __aexit__ waits for the plugins' code after the Worker shut down. -_RUN_TASK_EXIT_TIMEOUT = timedelta(seconds=60) - - class WorkerConfig(TypedDict, total=False): """TypedDict of config originally passed to :py:class:`Worker`.""" From a4347550ce6b22007df7530e90dbf2ae2beb4408 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 21 Oct 2026 23:35:46 -0700 Subject: [PATCH 4/5] Made the wait for plugin cleanup a Worker option. --- temporalio/worker/_worker.py | 18 ++++++++++++--- tests/worker/test_worker_exit.py | 38 +++++++++++++++++++++++++++++++- 2 files changed, 52 insertions(+), 4 deletions(-) diff --git a/temporalio/worker/_worker.py b/temporalio/worker/_worker.py index 4a3a1d850..d003f8738 100644 --- a/temporalio/worker/_worker.py +++ b/temporalio/worker/_worker.py @@ -133,6 +133,7 @@ def __init__( max_task_queue_activities_per_second: float | None = None, max_eager_activity_reservations_per_workflow_task: int = 3, graceful_shutdown_timeout: timedelta = timedelta(), + plugin_shutdown_timeout: timedelta | None = timedelta(seconds=60), workflow_failure_exception_types: Sequence[type[BaseException]] = [], shared_state_manager: SharedStateManager | None = None, debug_mode: bool = False, @@ -280,6 +281,11 @@ def __init__( graceful_shutdown_timeout: Amount of time after shutdown is called that activities are given to complete before their tasks are cancelled. + plugin_shutdown_timeout: Amount of time that exiting + ``async with Worker(...)`` waits, after shutdown, for the code + that plugins run after the Worker stops in ``run_worker``. That + code is cancelled after this time. ``None`` waits with no limit. + The default is 60 seconds. workflow_failure_exception_types: The types of exceptions that, if a workflow-thrown exception extends, will cause the workflow/update to fail instead of suspending the workflow via @@ -385,6 +391,7 @@ def __init__( max_task_queue_activities_per_second=max_task_queue_activities_per_second, max_eager_activity_reservations_per_workflow_task=max_eager_activity_reservations_per_workflow_task, graceful_shutdown_timeout=graceful_shutdown_timeout, + plugin_shutdown_timeout=plugin_shutdown_timeout, workflow_failure_exception_types=workflow_failure_exception_types, shared_state_manager=shared_state_manager, debug_mode=debug_mode, @@ -996,16 +1003,20 @@ async def __aexit__(self, exc_type: type[BaseException] | None, *args: Any) -> N if not self._async_context_run_task: raise RuntimeError("Never started") await self.shutdown() - # The run is done after shutdown, except for the code that plugins run after it. Let - # that code finish, so a plugin can still await calls in its cleanup. + # Plugins can run cleanup code after shutdown. Wait for that code instead of cancelling it. self._async_context_exiting = True run_task = self._async_context_run_task + timeout = self._config.get("plugin_shutdown_timeout", timedelta(seconds=60)) try: # wait() doesn't raise for the task's own outcome, only when this call is cancelled. - await asyncio.wait([run_task]) + await asyncio.wait( + [run_task], timeout=timeout.total_seconds() if timeout else None + ) except asyncio.CancelledError: run_task.cancel() raise + if not run_task.done(): + run_task.cancel() # Only re-raise our exception if present and exc_type is cancel if exc_type is asyncio.CancelledError and self._async_context_run_exception: raise self._async_context_run_exception @@ -1045,6 +1056,7 @@ class WorkerConfig(TypedDict, total=False): max_task_queue_activities_per_second: float | None max_eager_activity_reservations_per_workflow_task: int graceful_shutdown_timeout: timedelta + plugin_shutdown_timeout: timedelta | None workflow_failure_exception_types: Sequence[type[BaseException]] shared_state_manager: SharedStateManager | None debug_mode: bool diff --git a/tests/worker/test_worker_exit.py b/tests/worker/test_worker_exit.py index 8950b2e8e..737639f47 100644 --- a/tests/worker/test_worker_exit.py +++ b/tests/worker/test_worker_exit.py @@ -1,5 +1,7 @@ +import asyncio import uuid from collections.abc import Awaitable, Callable +from datetime import timedelta from temporalio import workflow from temporalio.client import Client @@ -40,5 +42,39 @@ async def test_plugin_cleanup_awaits_after_worker_exit(client: Client): ): pass assert plugin.cleanup_result is not None - # The caller's own awaits after the block must not be cancelled. + # The caller can still await after the block. await client.count_workflows() + + +class StuckCleanupPlugin(SimplePlugin): + def __init__(self) -> None: + super().__init__("test-stuck-cleanup") + self.cancelled = asyncio.Event() + + async def run_worker( + self, worker: Worker, next: Callable[[Worker], Awaitable[None]] + ) -> None: + try: + await super().run_worker(worker, next) + finally: + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + self.cancelled.set() + raise + + +async def test_plugin_cleanup_is_cancelled_after_plugin_shutdown_timeout( + client: Client, +): + plugin = StuckCleanupPlugin() + async with Worker( + client, + task_queue=f"tq-{uuid.uuid4()}", + workflows=[NothingWorkflow], + plugins=[plugin], + plugin_shutdown_timeout=timedelta(milliseconds=100), + ): + pass + # The cancel reaches the cleanup code on a later loop iteration, so wait for it. + await asyncio.wait_for(plugin.cancelled.wait(), timeout=10) From cb37d2ed31f9b9d93a63a0c2fe833b64fc789f07 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 21 Oct 2026 23:59:39 -0700 Subject: [PATCH 5/5] Pinned the nexgen revision without the one-way operation. --- scripts/gen_stream_service.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/scripts/gen_stream_service.py b/scripts/gen_stream_service.py index be287b764..d04af620c 100644 --- a/scripts/gen_stream_service.py +++ b/scripts/gen_stream_service.py @@ -2,7 +2,7 @@ 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 and client features yet, so the script installs nexgen at +offset, 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. """ @@ -33,7 +33,7 @@ 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_REVISION = "25b9c1be8f51245618297415aa51df6928c58504" nex_gen_root = base_dir / ".nexgen-streams" / NEX_GEN_REVISION