diff --git a/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto b/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto index cf338ca55..8b94f6021 100644 --- a/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto +++ b/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto @@ -13,6 +13,10 @@ import "temporal/api/failure/v1/message.proto"; // Identifies one Stream. The id is opaque to callers. message StreamRef { string id = 1; + // 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. + string workflow_id = 2; + string first_execution_run_id = 3; } enum ProducerKind { diff --git a/scripts/_wit/stream-service.wit b/scripts/_wit/stream-service.wit index cb29651b3..50fed2c50 100644 --- a/scripts/_wit/stream-service.wit +++ b/scripts/_wit/stream-service.wit @@ -36,6 +36,10 @@ interface stream-service { record stream-ref { /// @nexus.doc "The Stream id. 1 to 4096 bytes after encoding." id: string, + /// @nexus.doc "The Workflow that owns the Stream. Empty on a standalone, client or Activity ref." + 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.proto "temporal.api.streamservice.v1.ProducerKind" diff --git a/temporalio/api/streamservice/v1/stream_service_pb2.py b/temporalio/api/streamservice/v1/stream_service_pb2.py index cd9592760..d197f0722 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"\x17\n\tStreamRef\x12\n\n\x02id\x18\x01 \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"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' ) _PRODUCERKIND = DESCRIPTOR.enum_types_by_name["ProducerKind"] @@ -263,46 +263,46 @@ if _descriptor._USE_C_DESCRIPTORS == False: DESCRIPTOR._options = None - _PRODUCERKIND._serialized_start = 3382 - _PRODUCERKIND._serialized_end = 3582 - _STREAMRECORDKIND._serialized_start = 3585 - _STREAMRECORDKIND._serialized_end = 3762 - _STREAMSTATUS._serialized_start = 3764 - _STREAMSTATUS._serialized_end = 3886 - _READSTART._serialized_start = 3888 - _READSTART._serialized_end = 3973 + _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 _STREAMREF._serialized_start = 227 - _STREAMREF._serialized_end = 250 - _PRODUCEREPOCH._serialized_start = 253 - _PRODUCEREPOCH._serialized_end = 453 - _PRODUCER._serialized_start = 456 - _PRODUCER._serialized_end = 614 - _WORKFLOWTASKEPOCH._serialized_start = 616 - _WORKFLOWTASKEPOCH._serialized_end = 728 - _APPENDRECORD._serialized_start = 730 - _APPENDRECORD._serialized_end = 806 - _SUPERSESSION._serialized_start = 809 - _SUPERSESSION._serialized_end = 1116 - _STREAMRECORD._serialized_start = 1119 - _STREAMRECORD._serialized_end = 1512 - _CREATESTREAMREQUEST._serialized_start = 1514 - _CREATESTREAMREQUEST._serialized_end = 1641 - _CREATESTREAMRESPONSE._serialized_start = 1644 - _CREATESTREAMRESPONSE._serialized_end = 1802 - _APPENDRECORDSREQUEST._serialized_start = 1805 - _APPENDRECORDSREQUEST._serialized_end = 2170 - _APPENDRECORDSRESPONSE._serialized_start = 2172 - _APPENDRECORDSRESPONSE._serialized_end = 2238 - _CLOSESTREAMREQUEST._serialized_start = 2241 - _CLOSESTREAMREQUEST._serialized_end = 2602 - _CLOSESTREAMRESPONSE._serialized_start = 2604 - _CLOSESTREAMRESPONSE._serialized_end = 2625 - _READRECORDSREQUEST._serialized_start = 2628 - _READRECORDSREQUEST._serialized_end = 3002 - _READRECORDSRESPONSE._serialized_start = 3004 - _READRECORDSRESPONSE._serialized_end = 3122 - _DESCRIBESTREAMREQUEST._serialized_start = 3124 - _DESCRIBESTREAMREQUEST._serialized_end = 3224 - _DESCRIBESTREAMRESPONSE._serialized_start = 3227 - _DESCRIBESTREAMRESPONSE._serialized_end = 3379 + _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 # @@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 e1f4a62bc..5b17a6d1b 100644 --- a/temporalio/api/streamservice/v1/stream_service_pb2.pyi +++ b/temporalio/api/streamservice/v1/stream_service_pb2.pyi @@ -142,14 +142,31 @@ class StreamRef(google.protobuf.message.Message): DESCRIPTOR: google.protobuf.descriptor.Descriptor ID_FIELD_NUMBER: builtins.int + WORKFLOW_ID_FIELD_NUMBER: builtins.int + FIRST_EXECUTION_RUN_ID_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 def __init__( self, *, id: builtins.str = ..., + workflow_id: builtins.str = ..., + first_execution_run_id: builtins.str = ..., ) -> None: ... def ClearField( - self, field_name: typing_extensions.Literal["id", b"id"] + self, + field_name: typing_extensions.Literal[ + "first_execution_run_id", + b"first_execution_run_id", + "id", + b"id", + "workflow_id", + b"workflow_id", + ], ) -> None: ... global___StreamRef = StreamRef diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock index 347599f2b..ff49ac751 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=1c8c01f5ce87ca4ebb3fba155430920ff0a95c5c#1c8c01f5ce87ca4ebb3fba155430920ff0a95c5c" +source = "git+https://github.com/moetemp/sdk-streams?rev=813f9d129da2e90f86e8542e9a1ad766e2e7bf4f#813f9d129da2e90f86e8542e9a1ad766e2e7bf4f" dependencies = [ "async-trait", "aws-lc-rs", diff --git a/temporalio/bridge/Cargo.toml b/temporalio/bridge/Cargo.toml index ad770a3f8..b57c51d75 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 = "1c8c01f5ce87ca4ebb3fba155430920ff0a95c5c" +rev = "813f9d129da2e90f86e8542e9a1ad766e2e7bf4f" default-features = false optional = true diff --git a/temporalio/contrib/streams/_client.py b/temporalio/contrib/streams/_client.py index a6b6ff249..94e89748a 100644 --- a/temporalio/contrib/streams/_client.py +++ b/temporalio/contrib/streams/_client.py @@ -138,6 +138,7 @@ async def find_id() -> str: raise ValueError( "The server gave no first run id, so the Stream id is unknown" ) + stream._owner = (handle.id, first) return stream_id_of(first) def stream_id_of(first_run_id: str) -> str: @@ -150,7 +151,7 @@ def stream_id_of(first_run_id: str) -> str: if handle.first_execution_run_id else find_id ) - return ClientStream( + stream = ClientStream( handle._client, backend_of_client(handle._client), stream_id, @@ -158,6 +159,9 @@ def stream_id_of(first_run_id: str) -> str: wait_for_create=wait_for_create, handle=handle if end_with_run else None, ) + if handle.first_execution_run_id: + stream._owner = (handle.id, handle.first_execution_run_id) + return stream def for_ref( @@ -165,6 +169,7 @@ def for_ref( ref: StreamRef, *, wait_for_create: timedelta | None = None, + end_with_run: bool = False, batch_interval: timedelta | None = None, max_batch_size: int | None = None, max_retry_duration: timedelta | None = None, @@ -178,17 +183,33 @@ def for_ref( ref: The reference of the Stream. wait_for_create: How long a read waits for a Stream that doesn't exist yet. Defaults to no wait. + end_with_run: End a read when the owning Workflow's run is no longer running, as + :py:func:`for_handle` does. Only a Workflow's Stream has an owner. batch_interval: How often this client's writes flush. Defaults to 100 ms. max_batch_size: At most this many Records in one batch. max_retry_duration: How long a flush retries a Store failure. + + Raises: + ValueError: ``end_with_run`` is set, and the ref has no owning Workflow. """ - return ClientStream( + handle = None + if end_with_run: + if not ref.workflow_id or not ref.first_execution_run_id: + raise ValueError("end_with_run needs a ref of a Workflow's Stream") + handle = client.get_workflow_handle( + ref.workflow_id, first_execution_run_id=ref.first_execution_run_id + ) + stream = ClientStream( client, backend_of_client(client), ref.id, batch_options(batch_interval, max_batch_size, max_retry_duration), wait_for_create=wait_for_create, + handle=handle, ) + if ref.workflow_id and ref.first_execution_run_id: + stream._owner = (ref.workflow_id, ref.first_execution_run_id) + return stream async def create_stream( diff --git a/temporalio/contrib/streams/_outside.py b/temporalio/contrib/streams/_outside.py index 9d663ae5d..23dc3aee1 100644 --- a/temporalio/contrib/streams/_outside.py +++ b/temporalio/contrib/streams/_outside.py @@ -41,6 +41,8 @@ def __init__( self._namespace = namespace # A client that has only a Workflow id learns the first run id on its first call. self._id = stream_id if isinstance(stream_id, str) else "" + # The owning Workflow of a Workflow's Stream, which its ref carries. + self._owner: tuple[str, str] | None = None self._find_id = None if isinstance(stream_id, str) else stream_id self._writer_factory = writer self._writer: StreamWriter | None = None @@ -61,7 +63,13 @@ def ref(self) -> StreamRef: raise RuntimeError( "The Stream id is not known yet. Call get_offset() first." ) - return StreamRef(self._id) + if self._owner is None: + return StreamRef(self._id) + return StreamRef( + self._id, + workflow_id=self._owner[0], + first_execution_run_id=self._owner[1], + ) async def _stream_id(self) -> str: if not self._id: diff --git a/temporalio/contrib/streams/_reader.py b/temporalio/contrib/streams/_reader.py index 2be35ce74..690cffe77 100644 --- a/temporalio/contrib/streams/_reader.py +++ b/temporalio/contrib/streams/_reader.py @@ -18,8 +18,11 @@ # 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. +# 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) @@ -51,7 +54,15 @@ 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 @@ -63,14 +74,18 @@ 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_timeout, + idle_timeout=idle_left, create_wait=min(create_left, create_step) if create_left else None, ) try: with _convert.raw_bodies(converter.failure_converter): if wait_new_records: + # A reader that ends with the run looks at the run between short polls. answer = await service.read_records_until_records( - request, deadline=_POLL_DEADLINE + request, + deadline=_RUN_CHECK + if run_ended is not None + else _POLL_DEADLINE, ) else: answer = await service.read_records(request) @@ -90,6 +105,7 @@ async def read( 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 diff --git a/temporalio/contrib/streams/_stream.py b/temporalio/contrib/streams/_stream.py index 16a275792..cc87a8d43 100644 --- a/temporalio/contrib/streams/_stream.py +++ b/temporalio/contrib/streams/_stream.py @@ -8,12 +8,42 @@ from datetime import timedelta from typing import TYPE_CHECKING, Any, Generic, TypeVar +import temporalio.converter + if TYPE_CHECKING: from temporalio.contrib.streams._reader import Item T = TypeVar("T") +class _StreamRefJson( + temporalio.converter.TransferTypeConverter["StreamRef", dict[str, str]] +): + # The JSON form is the proto JSON of temporal.api.streamservice.v1.StreamRef, so every SDK + # reads a ref that another wrote. Snake case keys from older refs still decode. + transfer_type: type[dict[str, str]] | None = dict + + def to_transfer_type(self, value: StreamRef) -> dict[str, str]: + json = {"id": value.id} + if value.workflow_id: + json["workflowId"] = value.workflow_id + if value.first_execution_run_id: + json["firstExecutionRunId"] = value.first_execution_run_id + return json + + def from_transfer_type( + self, value: dict[str, str], type_hint: type[StreamRef] + ) -> StreamRef: + return StreamRef( + value["id"], + workflow_id=value.get("workflowId") or value.get("workflow_id") or None, + first_execution_run_id=value.get("firstExecutionRunId") + or value.get("first_execution_run_id") + or None, + ) + + +@temporalio.converter.transfer_type_convertible(_StreamRefJson) @dataclass(frozen=True) class StreamRef: """Names one Stream. Pass it to Workflows and Activities as an argument. @@ -25,6 +55,14 @@ class StreamRef: id: str """The Stream id. It is opaque to callers.""" + workflow_id: str | None = None + """For a Workflow's Stream, the id of the Workflow that owns it. None for a standalone + Stream.""" + + first_execution_run_id: str | None = None + """For a Workflow's Stream, the first run id of the owning run chain. None for a + standalone Stream.""" + class _Writable(abc.ABC): # A Stream kind that Topics publish to. diff --git a/temporalio/contrib/streams/_workflow.py b/temporalio/contrib/streams/_workflow.py index 89be9c336..7c47b8ce3 100644 --- a/temporalio/contrib/streams/_workflow.py +++ b/temporalio/contrib/streams/_workflow.py @@ -80,7 +80,7 @@ class WorkflowStream(_Writable): def __init__( self, - stream_id: str, + ref: StreamRef, *, create: bool, retention: timedelta | None, @@ -89,7 +89,8 @@ def __init__( """Create a Stream handle. Use :py:func:`workflow_stream` or :py:func:`create_workflow_stream` instead. """ - self._id = stream_id + self._ref = ref + self._id = ref.id self._create = create self._retention = retention self._auto_close = auto_close @@ -100,7 +101,14 @@ def __init__( @property def ref(self) -> StreamRef: """Reference to pass to Activities, child Workflows and clients.""" - return StreamRef(self._id) + return self._ref + + def _model_ref(self) -> models.StreamRef: + return models.StreamRef( + id=self._id, + workflow_id=self._ref.workflow_id, + first_execution_run_id=self._ref.first_execution_run_id, + ) @property def auto_close(self) -> bool: @@ -143,7 +151,7 @@ def _publish(self, topic: str, value: Any, force_flush: bool) -> None: self._ensure_created() self._used = True append_records( - stream=models.StreamRef(id=self._id), + stream=self._model_ref(), records=[models.AppendRecord(topic=topic, body=value)], ) @@ -158,9 +166,7 @@ def close( self._check("close") self._ensure_created() self._closed = True - close_stream( - stream=models.StreamRef(id=self._id), result=result, failure=failure - ) + 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.""" @@ -202,18 +208,18 @@ def workflow_stream( state = _state() if ref is None: info = temporalio.workflow.info() - stream_id = _ids.default_stream_id( - info.workflow_id, info.first_execution_run_id + ref = StreamRef( + _ids.default_stream_id(info.workflow_id, info.first_execution_run_id), + workflow_id=info.workflow_id, + first_execution_run_id=info.first_execution_run_id, ) create, close = True, True if auto_close is None else auto_close else: - stream_id = ref.id create, close = False, bool(auto_close) + stream_id = ref.id stream = state.streams.get(stream_id) if stream is None: - stream = WorkflowStream( - stream_id, create=create, retention=None, auto_close=close - ) + stream = WorkflowStream(ref, create=create, retention=None, auto_close=close) state.streams[stream_id] = stream elif auto_close is not None: stream._auto_close = auto_close @@ -237,9 +243,12 @@ def create_workflow_stream( stream_id = _ids.scoped_stream_id( info.workflow_id, info.first_execution_run_id, name ) - stream = WorkflowStream( - stream_id, create=True, retention=retention, auto_close=False + ref = StreamRef( + stream_id, + workflow_id=info.workflow_id, + first_execution_run_id=info.first_execution_run_id, ) + stream = WorkflowStream(ref, create=True, retention=retention, auto_close=False) stream._check("create") stream._ensure_created() _state().streams[stream_id] = stream diff --git a/temporalio/contrib/streams/redis.py b/temporalio/contrib/streams/redis.py index 06f35d7b8..1782a19de 100644 --- a/temporalio/contrib/streams/redis.py +++ b/temporalio/contrib/streams/redis.py @@ -260,7 +260,12 @@ async def _encode_call( if request.retention is not None: create.retention.FromTimedelta(request.retention) return ext_pb.WorkflowCall(create=create) - stream = stream_pb.StreamRef(id=request.stream.id if request.stream else "") + ref = request.stream or models.StreamRef(id="") + stream = stream_pb.StreamRef( + id=ref.id, + workflow_id=ref.workflow_id or "", + first_execution_run_id=ref.first_execution_run_id or "", + ) if call.operation == "AppendRecords": records = request.records or [] bodies = await converter.encode([record.body for record in records]) @@ -433,7 +438,10 @@ def _close_with_run( "CloseStream", _convert_in_sandbox( models.CloseStreamRequest( - stream=models.StreamRef(id=stream_id), + # A received ref keeps its own owner. + stream=state.streams[stream_id]._model_ref() + if stream_id in state.streams + else _owned_ref(stream_id), result=result, failure=failure, ) @@ -470,6 +478,16 @@ def _convert_in_sandbox(request: Any) -> Any: return request +def _owned_ref(stream_id: str) -> models.StreamRef: + # A Stream that the run creates, or its default Stream, is the run chain's own. + info = temporalio.workflow.info() + return models.StreamRef( + id=stream_id, + workflow_id=info.workflow_id, + first_execution_run_id=info.first_execution_run_id, + ) + + class _WorkflowOutbound(temporalio.worker.WorkflowOutboundInterceptor): _temporal_stream_transport = True @@ -500,7 +518,7 @@ async def start_system_nexus_operation( operation = input.operation_name if operation == "CreateStream": result: Any = models.CreateStreamResponse( - stream=models.StreamRef(id=input.input.stream_id) + stream=_owned_ref(input.input.stream_id) ) elif operation == "AppendRecords": result = models.AppendRecordsResponse() diff --git a/temporalio/contrib/streams/system/models.py b/temporalio/contrib/streams/system/models.py index 54bcc87ae..605fb9855 100644 --- a/temporalio/contrib/streams/system/models.py +++ b/temporalio/contrib/streams/system/models.py @@ -152,6 +152,8 @@ class StreamRef: """Identifies one Stream. The id is opaque to callers.""" id: str + workflow_id: str | None = None + first_execution_run_id: str | None = None class _StreamRefTransferTypeConverter( @@ -174,6 +176,10 @@ def from_transfer_type( id = value.id return StreamRef( id=id, + workflow_id=value.workflow_id if bool(value.workflow_id) else None, + first_execution_run_id=value.first_execution_run_id + if bool(value.first_execution_run_id) + else None, ) @typing_extensions.override @@ -183,6 +189,10 @@ def to_transfer_type( ) -> temporalio.api.streamservice.v1.stream_service_pb2.StreamRef: message = temporalio.api.streamservice.v1.stream_service_pb2.StreamRef() message.id = value.id + if value.workflow_id is not None: + 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 return message diff --git a/tests/contrib/streams/test_facade.py b/tests/contrib/streams/test_facade.py index d0e0c6416..b2cf09769 100644 --- a/tests/contrib/streams/test_facade.py +++ b/tests/contrib/streams/test_facade.py @@ -1,12 +1,16 @@ import asyncio +import json import uuid from dataclasses import dataclass from datetime import timedelta from typing import Any +import google.protobuf.json_format import pytest +import temporalio.api.common.v1 import temporalio.api.streamservice.v1 as stream_pb +import temporalio.converter from temporalio import workflow from temporalio.client import Client from temporalio.common import RawValue @@ -339,3 +343,53 @@ async def test_read_by_topic_prefix(client: Client): if isinstance(item, streams.Record) ] assert read == ["agent/a", "agent/b", "exact"] + + +def test_ref_without_owner_cannot_end_with_run(client: Client): + client, _ = fake_client(client) + with pytest.raises(ValueError): + streams.for_ref(client, streams.StreamRef("standalone"), end_with_run=True) + + +def test_old_refs_still_decode(): + converter = temporalio.converter.DataConverter.default.payload_converter + payload = converter.to_payload({"id": "old"}) + ref = converter.from_payload(payload, streams.StreamRef) + assert ref == streams.StreamRef("old") + assert ref.workflow_id is None and ref.first_execution_run_id is None + + +@pytest.mark.parametrize( + "ref", + [ + streams.StreamRef("only-id"), + streams.StreamRef("wf:run", workflow_id="wf", first_execution_run_id="run"), + ], +) +def test_ref_json_is_the_proto_json(ref: streams.StreamRef): + converter = temporalio.converter.DataConverter.default.payload_converter + payload = converter.to_payload(ref) + proto = stream_pb.StreamRef( + id=ref.id, + workflow_id=ref.workflow_id or "", + first_execution_run_id=ref.first_execution_run_id or "", + ) + assert json.loads(payload.data) == json.loads( + google.protobuf.json_format.MessageToJson(proto) + ) + # Another SDK's proto JSON decodes to the same ref. + from_proto = temporalio.api.common.v1.Payload( + metadata={"encoding": b"json/plain"}, + data=google.protobuf.json_format.MessageToJson(proto).encode(), + ) + assert converter.from_payload(from_proto, streams.StreamRef) == ref + + +def test_snake_case_refs_still_decode(): + converter = temporalio.converter.DataConverter.default.payload_converter + payload = converter.to_payload( + {"id": "old", "workflow_id": "wf", "first_execution_run_id": "run"} + ) + assert converter.from_payload(payload, streams.StreamRef) == streams.StreamRef( + "old", workflow_id="wf", first_execution_run_id="run" + ) diff --git a/tests/contrib/streams/test_redis.py b/tests/contrib/streams/test_redis.py index 4b351f30a..bf52ed02a 100644 --- a/tests/contrib/streams/test_redis.py +++ b/tests/contrib/streams/test_redis.py @@ -18,6 +18,7 @@ from temporalio.client import Client from temporalio.common import RawValue, RetryPolicy from temporalio.contrib import streams +from temporalio.contrib.streams.proto import v1 as ext_pb from temporalio.contrib.streams.redis import ( RedisStreamPlugin, StreamPlugin, @@ -813,3 +814,89 @@ async def test_read_by_topic_prefix_on_redis(stream_client: Client): if isinstance(item, streams.Record) ] assert read == ["agent/a", "agent/b", "exact"] + + +@workflow.defn +class LeavesStreamOpen: + @workflow.run + async def run(self) -> streams.StreamRef: + # Not in a with block, so nothing closes it when the run ends. + open_stream = streams.create_workflow_stream("open") + open_stream.topic("t", type=str).publish("only") + return open_stream.ref + + +async def test_for_ref_ends_with_the_owning_run(stream_client: Client): + async with new_worker(stream_client, LeavesStreamOpen) as worker: + handle = await stream_client.start_workflow( + LeavesStreamOpen.run, id=f"wf-{uuid.uuid4()}", task_queue=worker.task_queue + ) + ref = await handle.result() + assert ref.workflow_id == handle.id + assert ref.first_execution_run_id == handle.first_execution_run_id + stream = streams.for_ref(stream_client, ref, end_with_run=True) + items = [item async for item in stream.subscribe(type=str)] + assert [i.data for i in items if isinstance(i, streams.Record)] == ["only"] + # A ref to a Stream that the finished run never created ends at once, not after the + # whole wait. + missing = streams.StreamRef( + ref.id + "-missing", + workflow_id=ref.workflow_id, + first_execution_run_id=ref.first_execution_run_id, + ) + waiting = streams.for_ref( + stream_client, missing, end_with_run=True, wait_for_create=timedelta(minutes=5) + ) + start = asyncio.get_running_loop().time() + assert [item async for item in waiting.subscribe()] == [] + 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) + ] + sent: list[bytes] = [] + call = plugin._temporal_stream_backend.call + + def record(method: str, request: Any) -> bytes: + if method == "BufferWorkflowCalls": + sent.append(request.SerializeToString()) + return call(method, request) + + plugin._temporal_stream_backend.call = record # type: ignore[method-assign] + async with new_worker(stream_client, QuickWriter) as worker: + handle = await stream_client.start_workflow( + QuickWriter.run, id=f"wf-{uuid.uuid4()}", task_queue=worker.task_queue + ) + await handle.result() + refs = [] + for data in sent: + request = ext_pb.BufferWorkflowCallsRequest.FromString(data) + for workflow_call in request.calls: + if workflow_call.HasField("append"): + refs.append(workflow_call.append.stream) + if workflow_call.HasField("close"): + refs.append(workflow_call.close.stream) + assert refs + for ref in refs: + assert ref.workflow_id == handle.id + assert ref.first_execution_run_id == handle.first_execution_run_id