diff --git a/temporalio/bridge/proto/streams/__init__.py b/temporalio/bridge/proto/streams/__init__.py index 0873b3cad3..fe24ec0a56 100644 --- a/temporalio/bridge/proto/streams/__init__.py +++ b/temporalio/bridge/proto/streams/__init__.py @@ -8,6 +8,8 @@ CloseResponse, DeleteOwnerRequest, DeleteOwnerResponse, + DescribeStoreRequest, + DescribeStoreResponse, LatestRequest, LatestResponse, MemoryStoreConfig, @@ -20,6 +22,7 @@ RetireProducerResponse, StoreAppendRequest, StoreAppendResponse, + StoreCapabilities, StoredRecord, StoreLatestRequest, StoreLatestResponse, @@ -43,6 +46,8 @@ "CloseResponse", "DeleteOwnerRequest", "DeleteOwnerResponse", + "DescribeStoreRequest", + "DescribeStoreResponse", "LatestRequest", "LatestResponse", "MemoryStoreConfig", @@ -55,6 +60,7 @@ "RetireProducerResponse", "StoreAppendRequest", "StoreAppendResponse", + "StoreCapabilities", "StoreLatestRequest", "StoreLatestResponse", "StoreReadRequest", diff --git a/temporalio/bridge/proto/streams/streams_pb2.py b/temporalio/bridge/proto/streams/streams_pb2.py index 2f40a0ea14..66242f0964 100644 --- a/temporalio/bridge/proto/streams/streams_pb2.py +++ b/temporalio/bridge/proto/streams/streams_pb2.py @@ -25,7 +25,7 @@ ) DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( - b'\n\'temporal/sdk/core/streams/streams.proto\x12\x0f\x63oresdk.streams\x1a\x1egoogle/protobuf/duration.proto\x1a$temporal/api/common/v1/message.proto\x1a%temporal/sdk/streams/v1/message.proto"\xbc\x01\n\x11StreamStoreConfig\x12\x32\n\x05redis\x18\x01 \x01(\x0b\x32!.coresdk.streams.RedisStoreConfigH\x00\x12\x34\n\x06memory\x18\x02 \x01(\x0b\x32".coresdk.streams.MemoryStoreConfigH\x00\x12\x34\n\x11owner_start_grace\x18\x06 \x01(\x0b\x32\x19.google.protobuf.DurationB\x07\n\x05store"\xc9\x01\n\x10RedisStoreConfig\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\x1f\n\x17\x62locking_reads_per_node\x18\x05 \x01(\r\x12\x33\n\x10response_timeout\x18\x06 \x01(\x0b\x32\x19.google.protobuf.Duration"\x13\n\x11MemoryStoreConfig"\x8c\x01\n\rStreamAddress\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x34\n\nowner_kind\x18\x02 \x01(\x0e\x32 .coresdk.streams.StreamOwnerKind\x12\x13\n\x0bworkflow_id\x18\x03 \x01(\t\x12\x0e\n\x06run_id\x18\x04 \x01(\t\x12\r\n\x05topic\x18\x05 \x01(\t"G\n\x07\x43hainId\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x13\n\x0bworkflow_id\x18\x02 \x01(\t\x12\x14\n\x0c\x66irst_run_id\x18\x03 \x01(\t"5\n\rNamedProducer\x12\x13\n\x0bproducer_id\x18\x01 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x03"]\n\x10\x41\x63tivityProducer\x12\x13\n\x0bworkflow_id\x18\x01 \x01(\t\x12\x0e\n\x06run_id\x18\x02 \x01(\t\x12\x13\n\x0b\x61\x63tivity_id\x18\x03 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x05"\x8c\x01\n\x0c\x41ppendRecord\x12\x37\n\x04kind\x18\x01 \x01(\x0e\x32).temporal.sdk.streams.v1.StreamRecordKind\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x14\n\x0c\x63ontent_hash\x18\x03 \x01(\x0c"\x85\x02\n\rAppendRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress\x12/\n\x05named\x18\x02 \x01(\x0b\x32\x1e.coresdk.streams.NamedProducerH\x00\x12\x35\n\x08\x61\x63tivity\x18\x03 \x01(\x0b\x32!.coresdk.streams.ActivityProducerH\x00\x12\x10\n\x08sequence\x18\x04 \x01(\x03\x12.\n\x07records\x18\x05 \x03(\x0b\x32\x1d.coresdk.streams.AppendRecord\x12\x0e\n\x06\x64igest\x18\x06 \x01(\x0c\x42\n\n\x08producer";\n\x0e\x41ppendResponse\x12\x14\n\x0c\x66irst_cursor\x18\x01 \x01(\t\x12\x13\n\x0blast_cursor\x18\x02 \x01(\t"\x99\x01\n\x0bReadRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress\x12\r\n\x05\x61\x66ter\x18\x02 \x01(\t\x12\'\n\x04wait\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x13\n\x0bmax_records\x18\x04 \x01(\r\x12\r\n\x05state\x18\x05 \x01(\x0c"]\n\x0cSupersession\x12\r\n\x05topic\x18\x01 \x01(\t\x12\x13\n\x0bproducer_id\x18\x02 \x01(\t\x12\x18\n\x10previous_attempt\x18\x03 \x01(\x03\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x03"\xa3\x01\n\nReadRecord\x12\x0e\n\x06\x63ursor\x18\x01 \x01(\t\x12\x37\n\x06stored\x18\x02 \x01(\x0b\x32%.temporal.sdk.streams.v1.StreamRecordH\x00\x12\x33\n\nsuperseded\x18\x03 \x01(\x0b\x32\x1d.coresdk.streams.SupersessionH\x00\x12\r\n\x05stale\x18\x04 \x01(\x08\x42\x08\n\x06record"i\n\x0cReadResponse\x12,\n\x07records\x18\x01 \x03(\x0b\x32\x1b.coresdk.streams.ReadRecord\x12\x0e\n\x06\x63ursor\x18\x02 \x01(\t\x12\r\n\x05state\x18\x03 \x01(\x0c\x12\x0c\n\x04\x64one\x18\x04 \x01(\x08"?\n\rLatestRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress" \n\x0eLatestResponse\x12\x0e\n\x06\x63ursor\x18\x01 \x01(\t">\n\x0c\x43loseRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress"\x0f\n\rCloseResponse"r\n\x12\x44\x65leteOwnerRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x34\n\nowner_kind\x18\x02 \x01(\x0e\x32 .coresdk.streams.StreamOwnerKind\x12\x13\n\x0bworkflow_id\x18\x03 \x01(\t"&\n\x13\x44\x65leteOwnerResponse\x12\x0f\n\x07\x64\x65leted\x18\x01 \x01(\x04"\xbb\x01\n\x15RetireProducerRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress\x12/\n\x05named\x18\x02 \x01(\x0b\x32\x1e.coresdk.streams.NamedProducerH\x00\x12\x35\n\x08\x61\x63tivity\x18\x03 \x01(\x0b\x32!.coresdk.streams.ActivityProducerH\x00\x42\n\n\x08producer"\x18\n\x16RetireProducerResponse"b\n\rStreamFailure\x12\x30\n\x04kind\x18\x01 \x01(\x0e\x32".coresdk.streams.StreamFailureKind\x12\x0f\n\x07message\x18\x02 \x01(\t\x12\x0e\n\x06\x63ursor\x18\x03 \x01(\t"\xa5\x01\n\x12StoreAppendRequest\x12\'\n\x05\x63hain\x18\x01 \x01(\x0b\x32\x18.coresdk.streams.ChainId\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x13\n\x0bproducer_id\x18\x03 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x03\x12\x10\n\x08sequence\x18\x05 \x01(\x03\x12\x0e\n\x06\x64igest\x18\x06 \x01(\x0c\x12\x0f\n\x07records\x18\x07 \x03(\x0c"D\n\x13StoreAppendResponse\x12\x16\n\x0e\x66irst_position\x18\x01 \x01(\t\x12\x15\n\rlast_position\x18\x02 \x01(\t"\xa0\x01\n\x10StoreReadRequest\x12\'\n\x05\x63hain\x18\x01 \x01(\x0b\x32\x18.coresdk.streams.ChainId\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x16\n\x0e\x61\x66ter_position\x18\x03 \x01(\t\x12\'\n\x04wait\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x13\n\x0bmax_records\x18\x05 \x01(\r"0\n\x0cStoredRecord\x12\x10\n\x08position\x18\x01 \x01(\t\x12\x0e\n\x06record\x18\x02 \x01(\x0c"S\n\x11StoreReadResponse\x12.\n\x07records\x18\x01 \x03(\x0b\x32\x1d.coresdk.streams.StoredRecord\x12\x0e\n\x06\x63losed\x18\x02 \x01(\x08"L\n\x12StoreLatestRequest\x12\'\n\x05\x63hain\x18\x01 \x01(\x0b\x32\x18.coresdk.streams.ChainId\x12\r\n\x05topic\x18\x02 \x01(\t"\'\n\x13StoreLatestResponse\x12\x10\n\x08position\x18\x01 \x01(\t*T\n\x0fStreamOwnerKind\x12!\n\x1dSTREAM_OWNER_KIND_UNSPECIFIED\x10\x00\x12\x1e\n\x1aSTREAM_OWNER_KIND_WORKFLOW\x10\x01*\xea\x03\n\x11StreamFailureKind\x12#\n\x1fSTREAM_FAILURE_KIND_UNSPECIFIED\x10\x00\x12*\n&STREAM_FAILURE_KIND_PRODUCER_DIVERGENT\x10\x01\x12&\n"STREAM_FAILURE_KIND_PRODUCER_STALE\x10\x02\x12\'\n#STREAM_FAILURE_KIND_OUTCOME_UNKNOWN\x10\x03\x12\x1f\n\x1bSTREAM_FAILURE_KIND_REFUSED\x10\x04\x12\x1e\n\x1aSTREAM_FAILURE_KIND_CLOSED\x10\x05\x12\x1e\n\x1aSTREAM_FAILURE_KIND_CURSOR\x10\x06\x12\x1f\n\x1bSTREAM_FAILURE_KIND_EXPIRED\x10\x07\x12!\n\x1dSTREAM_FAILURE_KIND_NOT_FOUND\x10\x08\x12\x1e\n\x1aSTREAM_FAILURE_KIND_RECORD\x10\t\x12\x1f\n\x1bSTREAM_FAILURE_KIND_STORAGE\x10\n\x12#\n\x1fSTREAM_FAILURE_KIND_UNSUPPORTED\x10\x0b\x12(\n$STREAM_FAILURE_KIND_INVALID_ARGUMENT\x10\x0c\x32\xef\x03\n\rStreamService\x12I\n\x06\x41ppend\x12\x1e.coresdk.streams.AppendRequest\x1a\x1f.coresdk.streams.AppendResponse\x12\x43\n\x04Read\x12\x1c.coresdk.streams.ReadRequest\x1a\x1d.coresdk.streams.ReadResponse\x12I\n\x06Latest\x12\x1e.coresdk.streams.LatestRequest\x1a\x1f.coresdk.streams.LatestResponse\x12\x46\n\x05\x43lose\x12\x1d.coresdk.streams.CloseRequest\x1a\x1e.coresdk.streams.CloseResponse\x12X\n\x0b\x44\x65leteOwner\x12#.coresdk.streams.DeleteOwnerRequest\x1a$.coresdk.streams.DeleteOwnerResponse\x12\x61\n\x0eRetireProducer\x12&.coresdk.streams.RetireProducerRequest\x1a\'.coresdk.streams.RetireProducerResponseB-\xea\x02*Temporalio::Internal::Bridge::Api::Streamsb\x06proto3' + b'\n\'temporal/sdk/core/streams/streams.proto\x12\x0f\x63oresdk.streams\x1a\x1egoogle/protobuf/duration.proto\x1a$temporal/api/common/v1/message.proto\x1a%temporal/sdk/streams/v1/message.proto"\xbc\x01\n\x11StreamStoreConfig\x12\x32\n\x05redis\x18\x01 \x01(\x0b\x32!.coresdk.streams.RedisStoreConfigH\x00\x12\x34\n\x06memory\x18\x02 \x01(\x0b\x32".coresdk.streams.MemoryStoreConfigH\x00\x12\x34\n\x11owner_start_grace\x18\x06 \x01(\x0b\x32\x19.google.protobuf.DurationB\x07\n\x05store"\xd8\x01\n\x10RedisStoreConfig\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\x1f\n\x17\x62locking_reads_per_node\x18\x05 \x01(\r\x12\x33\n\x10response_timeout\x18\x06 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\r\n\x05plain\x18\n \x01(\x08"M\n\x11MemoryStoreConfig\x12\x38\n\x0c\x63\x61pabilities\x18\x01 \x01(\x0b\x32".coresdk.streams.StoreCapabilities"\x8c\x01\n\rStreamAddress\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x34\n\nowner_kind\x18\x02 \x01(\x0e\x32 .coresdk.streams.StreamOwnerKind\x12\x13\n\x0bworkflow_id\x18\x03 \x01(\t\x12\x0e\n\x06run_id\x18\x04 \x01(\t\x12\r\n\x05topic\x18\x05 \x01(\t"G\n\x07\x43hainId\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x13\n\x0bworkflow_id\x18\x02 \x01(\t\x12\x14\n\x0c\x66irst_run_id\x18\x03 \x01(\t"5\n\rNamedProducer\x12\x13\n\x0bproducer_id\x18\x01 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x03"]\n\x10\x41\x63tivityProducer\x12\x13\n\x0bworkflow_id\x18\x01 \x01(\t\x12\x0e\n\x06run_id\x18\x02 \x01(\t\x12\x13\n\x0b\x61\x63tivity_id\x18\x03 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x05"\x8c\x01\n\x0c\x41ppendRecord\x12\x37\n\x04kind\x18\x01 \x01(\x0e\x32).temporal.sdk.streams.v1.StreamRecordKind\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x14\n\x0c\x63ontent_hash\x18\x03 \x01(\x0c"\x85\x02\n\rAppendRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress\x12/\n\x05named\x18\x02 \x01(\x0b\x32\x1e.coresdk.streams.NamedProducerH\x00\x12\x35\n\x08\x61\x63tivity\x18\x03 \x01(\x0b\x32!.coresdk.streams.ActivityProducerH\x00\x12\x10\n\x08sequence\x18\x04 \x01(\x03\x12.\n\x07records\x18\x05 \x03(\x0b\x32\x1d.coresdk.streams.AppendRecord\x12\x0e\n\x06\x64igest\x18\x06 \x01(\x0c\x42\n\n\x08producer";\n\x0e\x41ppendResponse\x12\x14\n\x0c\x66irst_cursor\x18\x01 \x01(\t\x12\x13\n\x0blast_cursor\x18\x02 \x01(\t"\x99\x01\n\x0bReadRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress\x12\r\n\x05\x61\x66ter\x18\x02 \x01(\t\x12\'\n\x04wait\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x13\n\x0bmax_records\x18\x04 \x01(\r\x12\r\n\x05state\x18\x05 \x01(\x0c"]\n\x0cSupersession\x12\r\n\x05topic\x18\x01 \x01(\t\x12\x13\n\x0bproducer_id\x18\x02 \x01(\t\x12\x18\n\x10previous_attempt\x18\x03 \x01(\x03\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x03"\xa3\x01\n\nReadRecord\x12\x0e\n\x06\x63ursor\x18\x01 \x01(\t\x12\x37\n\x06stored\x18\x02 \x01(\x0b\x32%.temporal.sdk.streams.v1.StreamRecordH\x00\x12\x33\n\nsuperseded\x18\x03 \x01(\x0b\x32\x1d.coresdk.streams.SupersessionH\x00\x12\r\n\x05stale\x18\x04 \x01(\x08\x42\x08\n\x06record"i\n\x0cReadResponse\x12,\n\x07records\x18\x01 \x03(\x0b\x32\x1b.coresdk.streams.ReadRecord\x12\x0e\n\x06\x63ursor\x18\x02 \x01(\t\x12\r\n\x05state\x18\x03 \x01(\x0c\x12\x0c\n\x04\x64one\x18\x04 \x01(\x08"?\n\rLatestRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress" \n\x0eLatestResponse\x12\x0e\n\x06\x63ursor\x18\x01 \x01(\t">\n\x0c\x43loseRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress"\x0f\n\rCloseResponse"r\n\x12\x44\x65leteOwnerRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x34\n\nowner_kind\x18\x02 \x01(\x0e\x32 .coresdk.streams.StreamOwnerKind\x12\x13\n\x0bworkflow_id\x18\x03 \x01(\t"&\n\x13\x44\x65leteOwnerResponse\x12\x0f\n\x07\x64\x65leted\x18\x01 \x01(\x04"\xbb\x01\n\x15RetireProducerRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress\x12/\n\x05named\x18\x02 \x01(\x0b\x32\x1e.coresdk.streams.NamedProducerH\x00\x12\x35\n\x08\x61\x63tivity\x18\x03 \x01(\x0b\x32!.coresdk.streams.ActivityProducerH\x00\x42\n\n\x08producer"\x18\n\x16RetireProducerResponse"\x16\n\x14\x44\x65scribeStoreRequest"_\n\x15\x44\x65scribeStoreResponse\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x38\n\x0c\x63\x61pabilities\x18\x02 \x01(\x0b\x32".coresdk.streams.StoreCapabilities"\xea\x02\n\x11StoreCapabilities\x12\x16\n\x0eproducer_state\x18\x01 \x01(\x08\x12\x18\n\x10\x63onditional_tail\x18\x02 \x01(\x08\x12\x19\n\x11\x63onditional_fence\x18\x03 \x01(\x08\x12\x11\n\tkept_last\x18\x04 \x01(\x08\x12\r\n\x05holds\x18\x05 \x01(\x08\x12\x13\n\x0b\x63hange_feed\x18\x06 \x01(\x08\x12\x12\n\ncheap_logs\x18\x07 \x01(\x08\x12\x14\n\x0c\x64\x65lete_owner\x18\x08 \x01(\x08\x12\x15\n\rtopic_listing\x18\t \x01(\x08\x12\x11\n\tlong_poll\x18\n \x01(\x08\x12\x17\n\x0f\x64\x65nse_positions\x18\x0b \x01(\x08\x12\x31\n\x0ereplay_horizon\x18\x0c \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x18\n\x10max_record_bytes\x18\r \x01(\x04\x12\x17\n\x0fmax_batch_bytes\x18\x0e \x01(\x04"b\n\rStreamFailure\x12\x30\n\x04kind\x18\x01 \x01(\x0e\x32".coresdk.streams.StreamFailureKind\x12\x0f\n\x07message\x18\x02 \x01(\t\x12\x0e\n\x06\x63ursor\x18\x03 \x01(\t"\xa5\x01\n\x12StoreAppendRequest\x12\'\n\x05\x63hain\x18\x01 \x01(\x0b\x32\x18.coresdk.streams.ChainId\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x13\n\x0bproducer_id\x18\x03 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x03\x12\x10\n\x08sequence\x18\x05 \x01(\x03\x12\x0e\n\x06\x64igest\x18\x06 \x01(\x0c\x12\x0f\n\x07records\x18\x07 \x03(\x0c"D\n\x13StoreAppendResponse\x12\x16\n\x0e\x66irst_position\x18\x01 \x01(\t\x12\x15\n\rlast_position\x18\x02 \x01(\t"\xa0\x01\n\x10StoreReadRequest\x12\'\n\x05\x63hain\x18\x01 \x01(\x0b\x32\x18.coresdk.streams.ChainId\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x16\n\x0e\x61\x66ter_position\x18\x03 \x01(\t\x12\'\n\x04wait\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x13\n\x0bmax_records\x18\x05 \x01(\r"0\n\x0cStoredRecord\x12\x10\n\x08position\x18\x01 \x01(\t\x12\x0e\n\x06record\x18\x02 \x01(\x0c"S\n\x11StoreReadResponse\x12.\n\x07records\x18\x01 \x03(\x0b\x32\x1d.coresdk.streams.StoredRecord\x12\x0e\n\x06\x63losed\x18\x02 \x01(\x08"L\n\x12StoreLatestRequest\x12\'\n\x05\x63hain\x18\x01 \x01(\x0b\x32\x18.coresdk.streams.ChainId\x12\r\n\x05topic\x18\x02 \x01(\t"\'\n\x13StoreLatestResponse\x12\x10\n\x08position\x18\x01 \x01(\t*T\n\x0fStreamOwnerKind\x12!\n\x1dSTREAM_OWNER_KIND_UNSPECIFIED\x10\x00\x12\x1e\n\x1aSTREAM_OWNER_KIND_WORKFLOW\x10\x01*\xea\x03\n\x11StreamFailureKind\x12#\n\x1fSTREAM_FAILURE_KIND_UNSPECIFIED\x10\x00\x12*\n&STREAM_FAILURE_KIND_PRODUCER_DIVERGENT\x10\x01\x12&\n"STREAM_FAILURE_KIND_PRODUCER_STALE\x10\x02\x12\'\n#STREAM_FAILURE_KIND_OUTCOME_UNKNOWN\x10\x03\x12\x1f\n\x1bSTREAM_FAILURE_KIND_REFUSED\x10\x04\x12\x1e\n\x1aSTREAM_FAILURE_KIND_CLOSED\x10\x05\x12\x1e\n\x1aSTREAM_FAILURE_KIND_CURSOR\x10\x06\x12\x1f\n\x1bSTREAM_FAILURE_KIND_EXPIRED\x10\x07\x12!\n\x1dSTREAM_FAILURE_KIND_NOT_FOUND\x10\x08\x12\x1e\n\x1aSTREAM_FAILURE_KIND_RECORD\x10\t\x12\x1f\n\x1bSTREAM_FAILURE_KIND_STORAGE\x10\n\x12#\n\x1fSTREAM_FAILURE_KIND_UNSUPPORTED\x10\x0b\x12(\n$STREAM_FAILURE_KIND_INVALID_ARGUMENT\x10\x0c\x32\xcf\x04\n\rStreamService\x12I\n\x06\x41ppend\x12\x1e.coresdk.streams.AppendRequest\x1a\x1f.coresdk.streams.AppendResponse\x12\x43\n\x04Read\x12\x1c.coresdk.streams.ReadRequest\x1a\x1d.coresdk.streams.ReadResponse\x12I\n\x06Latest\x12\x1e.coresdk.streams.LatestRequest\x1a\x1f.coresdk.streams.LatestResponse\x12\x46\n\x05\x43lose\x12\x1d.coresdk.streams.CloseRequest\x1a\x1e.coresdk.streams.CloseResponse\x12X\n\x0b\x44\x65leteOwner\x12#.coresdk.streams.DeleteOwnerRequest\x1a$.coresdk.streams.DeleteOwnerResponse\x12\x61\n\x0eRetireProducer\x12&.coresdk.streams.RetireProducerRequest\x1a\'.coresdk.streams.RetireProducerResponse\x12^\n\rDescribeStore\x12%.coresdk.streams.DescribeStoreRequest\x1a&.coresdk.streams.DescribeStoreResponseB-\xea\x02*Temporalio::Internal::Bridge::Api::Streamsb\x06proto3' ) _STREAMOWNERKIND = DESCRIPTOR.enum_types_by_name["StreamOwnerKind"] @@ -71,6 +71,9 @@ _DELETEOWNERRESPONSE = DESCRIPTOR.message_types_by_name["DeleteOwnerResponse"] _RETIREPRODUCERREQUEST = DESCRIPTOR.message_types_by_name["RetireProducerRequest"] _RETIREPRODUCERRESPONSE = DESCRIPTOR.message_types_by_name["RetireProducerResponse"] +_DESCRIBESTOREREQUEST = DESCRIPTOR.message_types_by_name["DescribeStoreRequest"] +_DESCRIBESTORERESPONSE = DESCRIPTOR.message_types_by_name["DescribeStoreResponse"] +_STORECAPABILITIES = DESCRIPTOR.message_types_by_name["StoreCapabilities"] _STREAMFAILURE = DESCRIPTOR.message_types_by_name["StreamFailure"] _STOREAPPENDREQUEST = DESCRIPTOR.message_types_by_name["StoreAppendRequest"] _STOREAPPENDRESPONSE = DESCRIPTOR.message_types_by_name["StoreAppendResponse"] @@ -321,6 +324,39 @@ ) _sym_db.RegisterMessage(RetireProducerResponse) +DescribeStoreRequest = _reflection.GeneratedProtocolMessageType( + "DescribeStoreRequest", + (_message.Message,), + { + "DESCRIPTOR": _DESCRIBESTOREREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.DescribeStoreRequest) + }, +) +_sym_db.RegisterMessage(DescribeStoreRequest) + +DescribeStoreResponse = _reflection.GeneratedProtocolMessageType( + "DescribeStoreResponse", + (_message.Message,), + { + "DESCRIPTOR": _DESCRIBESTORERESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.DescribeStoreResponse) + }, +) +_sym_db.RegisterMessage(DescribeStoreResponse) + +StoreCapabilities = _reflection.GeneratedProtocolMessageType( + "StoreCapabilities", + (_message.Message,), + { + "DESCRIPTOR": _STORECAPABILITIES, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.StoreCapabilities) + }, +) +_sym_db.RegisterMessage(StoreCapabilities) + StreamFailure = _reflection.GeneratedProtocolMessageType( "StreamFailure", (_message.Message,), @@ -415,70 +451,76 @@ DESCRIPTOR._serialized_options = ( b"\352\002*Temporalio::Internal::Bridge::Api::Streams" ) - _STREAMOWNERKIND._serialized_start = 3250 - _STREAMOWNERKIND._serialized_end = 3334 - _STREAMFAILUREKIND._serialized_start = 3337 - _STREAMFAILUREKIND._serialized_end = 3827 + _STREAMOWNERKIND._serialized_start = 3809 + _STREAMOWNERKIND._serialized_end = 3893 + _STREAMFAILUREKIND._serialized_start = 3896 + _STREAMFAILUREKIND._serialized_end = 4386 _STREAMSTORECONFIG._serialized_start = 170 _STREAMSTORECONFIG._serialized_end = 358 _REDISSTORECONFIG._serialized_start = 361 - _REDISSTORECONFIG._serialized_end = 562 - _MEMORYSTORECONFIG._serialized_start = 564 - _MEMORYSTORECONFIG._serialized_end = 583 - _STREAMADDRESS._serialized_start = 586 - _STREAMADDRESS._serialized_end = 726 - _CHAINID._serialized_start = 728 - _CHAINID._serialized_end = 799 - _NAMEDPRODUCER._serialized_start = 801 - _NAMEDPRODUCER._serialized_end = 854 - _ACTIVITYPRODUCER._serialized_start = 856 - _ACTIVITYPRODUCER._serialized_end = 949 - _APPENDRECORD._serialized_start = 952 - _APPENDRECORD._serialized_end = 1092 - _APPENDREQUEST._serialized_start = 1095 - _APPENDREQUEST._serialized_end = 1356 - _APPENDRESPONSE._serialized_start = 1358 - _APPENDRESPONSE._serialized_end = 1417 - _READREQUEST._serialized_start = 1420 - _READREQUEST._serialized_end = 1573 - _SUPERSESSION._serialized_start = 1575 - _SUPERSESSION._serialized_end = 1668 - _READRECORD._serialized_start = 1671 - _READRECORD._serialized_end = 1834 - _READRESPONSE._serialized_start = 1836 - _READRESPONSE._serialized_end = 1941 - _LATESTREQUEST._serialized_start = 1943 - _LATESTREQUEST._serialized_end = 2006 - _LATESTRESPONSE._serialized_start = 2008 - _LATESTRESPONSE._serialized_end = 2040 - _CLOSEREQUEST._serialized_start = 2042 - _CLOSEREQUEST._serialized_end = 2104 - _CLOSERESPONSE._serialized_start = 2106 - _CLOSERESPONSE._serialized_end = 2121 - _DELETEOWNERREQUEST._serialized_start = 2123 - _DELETEOWNERREQUEST._serialized_end = 2237 - _DELETEOWNERRESPONSE._serialized_start = 2239 - _DELETEOWNERRESPONSE._serialized_end = 2277 - _RETIREPRODUCERREQUEST._serialized_start = 2280 - _RETIREPRODUCERREQUEST._serialized_end = 2467 - _RETIREPRODUCERRESPONSE._serialized_start = 2469 - _RETIREPRODUCERRESPONSE._serialized_end = 2493 - _STREAMFAILURE._serialized_start = 2495 - _STREAMFAILURE._serialized_end = 2593 - _STOREAPPENDREQUEST._serialized_start = 2596 - _STOREAPPENDREQUEST._serialized_end = 2761 - _STOREAPPENDRESPONSE._serialized_start = 2763 - _STOREAPPENDRESPONSE._serialized_end = 2831 - _STOREREADREQUEST._serialized_start = 2834 - _STOREREADREQUEST._serialized_end = 2994 - _STOREDRECORD._serialized_start = 2996 - _STOREDRECORD._serialized_end = 3044 - _STOREREADRESPONSE._serialized_start = 3046 - _STOREREADRESPONSE._serialized_end = 3129 - _STORELATESTREQUEST._serialized_start = 3131 - _STORELATESTREQUEST._serialized_end = 3207 - _STORELATESTRESPONSE._serialized_start = 3209 - _STORELATESTRESPONSE._serialized_end = 3248 - _STREAMSERVICE._serialized_start = 3830 - _STREAMSERVICE._serialized_end = 4325 + _REDISSTORECONFIG._serialized_end = 577 + _MEMORYSTORECONFIG._serialized_start = 579 + _MEMORYSTORECONFIG._serialized_end = 656 + _STREAMADDRESS._serialized_start = 659 + _STREAMADDRESS._serialized_end = 799 + _CHAINID._serialized_start = 801 + _CHAINID._serialized_end = 872 + _NAMEDPRODUCER._serialized_start = 874 + _NAMEDPRODUCER._serialized_end = 927 + _ACTIVITYPRODUCER._serialized_start = 929 + _ACTIVITYPRODUCER._serialized_end = 1022 + _APPENDRECORD._serialized_start = 1025 + _APPENDRECORD._serialized_end = 1165 + _APPENDREQUEST._serialized_start = 1168 + _APPENDREQUEST._serialized_end = 1429 + _APPENDRESPONSE._serialized_start = 1431 + _APPENDRESPONSE._serialized_end = 1490 + _READREQUEST._serialized_start = 1493 + _READREQUEST._serialized_end = 1646 + _SUPERSESSION._serialized_start = 1648 + _SUPERSESSION._serialized_end = 1741 + _READRECORD._serialized_start = 1744 + _READRECORD._serialized_end = 1907 + _READRESPONSE._serialized_start = 1909 + _READRESPONSE._serialized_end = 2014 + _LATESTREQUEST._serialized_start = 2016 + _LATESTREQUEST._serialized_end = 2079 + _LATESTRESPONSE._serialized_start = 2081 + _LATESTRESPONSE._serialized_end = 2113 + _CLOSEREQUEST._serialized_start = 2115 + _CLOSEREQUEST._serialized_end = 2177 + _CLOSERESPONSE._serialized_start = 2179 + _CLOSERESPONSE._serialized_end = 2194 + _DELETEOWNERREQUEST._serialized_start = 2196 + _DELETEOWNERREQUEST._serialized_end = 2310 + _DELETEOWNERRESPONSE._serialized_start = 2312 + _DELETEOWNERRESPONSE._serialized_end = 2350 + _RETIREPRODUCERREQUEST._serialized_start = 2353 + _RETIREPRODUCERREQUEST._serialized_end = 2540 + _RETIREPRODUCERRESPONSE._serialized_start = 2542 + _RETIREPRODUCERRESPONSE._serialized_end = 2566 + _DESCRIBESTOREREQUEST._serialized_start = 2568 + _DESCRIBESTOREREQUEST._serialized_end = 2590 + _DESCRIBESTORERESPONSE._serialized_start = 2592 + _DESCRIBESTORERESPONSE._serialized_end = 2687 + _STORECAPABILITIES._serialized_start = 2690 + _STORECAPABILITIES._serialized_end = 3052 + _STREAMFAILURE._serialized_start = 3054 + _STREAMFAILURE._serialized_end = 3152 + _STOREAPPENDREQUEST._serialized_start = 3155 + _STOREAPPENDREQUEST._serialized_end = 3320 + _STOREAPPENDRESPONSE._serialized_start = 3322 + _STOREAPPENDRESPONSE._serialized_end = 3390 + _STOREREADREQUEST._serialized_start = 3393 + _STOREREADREQUEST._serialized_end = 3553 + _STOREDRECORD._serialized_start = 3555 + _STOREDRECORD._serialized_end = 3603 + _STOREREADRESPONSE._serialized_start = 3605 + _STOREREADRESPONSE._serialized_end = 3688 + _STORELATESTREQUEST._serialized_start = 3690 + _STORELATESTREQUEST._serialized_end = 3766 + _STORELATESTRESPONSE._serialized_start = 3768 + _STORELATESTRESPONSE._serialized_end = 3807 + _STREAMSERVICE._serialized_start = 4389 + _STREAMSERVICE._serialized_end = 4980 # @@protoc_insertion_point(module_scope) diff --git a/temporalio/bridge/proto/streams/streams_pb2.pyi b/temporalio/bridge/proto/streams/streams_pb2.pyi index 5f50d47527..3c799418a3 100644 --- a/temporalio/bridge/proto/streams/streams_pb2.pyi +++ b/temporalio/bridge/proto/streams/streams_pb2.pyi @@ -193,6 +193,7 @@ class RedisStoreConfig(google.protobuf.message.Message): RETENTION_FIELD_NUMBER: builtins.int BLOCKING_READS_PER_NODE_FIELD_NUMBER: builtins.int RESPONSE_TIMEOUT_FIELD_NUMBER: builtins.int + PLAIN_FIELD_NUMBER: builtins.int @property def urls( self, @@ -215,6 +216,10 @@ class RedisStoreConfig(google.protobuf.message.Message): @property def response_timeout(self) -> google.protobuf.duration_pb2.Duration: """How long a short call waits for its reply. Unset uses the default.""" + plain: builtins.bool + """Runs with no Lua, so the store keeps no producer state and no last position, and Core takes + their fallbacks. For tests of the fallbacks on a real store. + """ def __init__( self, *, @@ -224,6 +229,7 @@ class RedisStoreConfig(google.protobuf.message.Message): retention: google.protobuf.duration_pb2.Duration | None = ..., blocking_reads_per_node: builtins.int = ..., response_timeout: google.protobuf.duration_pb2.Duration | None = ..., + plain: builtins.bool = ..., ) -> None: ... def HasField( self, @@ -240,6 +246,8 @@ class RedisStoreConfig(google.protobuf.message.Message): b"cluster", "key_prefix", b"key_prefix", + "plain", + b"plain", "response_timeout", b"response_timeout", "retention", @@ -258,8 +266,22 @@ class MemoryStoreConfig(google.protobuf.message.Message): DESCRIPTOR: google.protobuf.descriptor.Descriptor + CAPABILITIES_FIELD_NUMBER: builtins.int + @property + def capabilities(self) -> global___StoreCapabilities: + """A mask. Only the capabilities it sets stay, and its nonzero limits replace the store's. + Unset keeps every capability, so tests and users can try the weaker guarantees. + """ def __init__( self, + *, + capabilities: global___StoreCapabilities | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["capabilities", b"capabilities"] + ) -> builtins.bool: ... + def ClearField( + self, field_name: typing_extensions.Literal["capabilities", b"capabilities"] ) -> None: ... global___MemoryStoreConfig = MemoryStoreConfig @@ -921,6 +943,149 @@ class RetireProducerResponse(google.protobuf.message.Message): global___RetireProducerResponse = RetireProducerResponse +class DescribeStoreRequest(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___DescribeStoreRequest = DescribeStoreRequest + +class DescribeStoreResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAME_FIELD_NUMBER: builtins.int + CAPABILITIES_FIELD_NUMBER: builtins.int + name: builtins.str + """The short name the store mints into its cursors, such as `redis` or `memory`.""" + @property + def capabilities(self) -> global___StoreCapabilities: ... + def __init__( + self, + *, + name: builtins.str = ..., + capabilities: global___StoreCapabilities | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["capabilities", b"capabilities"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "capabilities", b"capabilities", "name", b"name" + ], + ) -> None: ... + +global___DescribeStoreResponse = DescribeStoreResponse + +class StoreCapabilities(google.protobuf.message.Message): + """What a store can do beyond the log every store gives. Core takes a fallback for each + capability the store lacks, with the weaker guarantee the fallback names. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + PRODUCER_STATE_FIELD_NUMBER: builtins.int + CONDITIONAL_TAIL_FIELD_NUMBER: builtins.int + CONDITIONAL_FENCE_FIELD_NUMBER: builtins.int + KEPT_LAST_FIELD_NUMBER: builtins.int + HOLDS_FIELD_NUMBER: builtins.int + CHANGE_FEED_FIELD_NUMBER: builtins.int + CHEAP_LOGS_FIELD_NUMBER: builtins.int + DELETE_OWNER_FIELD_NUMBER: builtins.int + TOPIC_LISTING_FIELD_NUMBER: builtins.int + LONG_POLL_FIELD_NUMBER: builtins.int + DENSE_POSITIONS_FIELD_NUMBER: builtins.int + REPLAY_HORIZON_FIELD_NUMBER: builtins.int + MAX_RECORD_BYTES_FIELD_NUMBER: builtins.int + MAX_BATCH_BYTES_FIELD_NUMBER: builtins.int + producer_state: builtins.bool + """The append checks the producer attempt's dedupe entry and the close flags in the same step.""" + conditional_tail: builtins.bool + """The append can fail unless the log's next position is the expected one.""" + conditional_fence: builtins.bool + """The append can fail unless the stream's fence token is the expected one.""" + kept_last: builtins.bool + """The store knows a topic's newest position after retention dropped every record.""" + holds: builtins.bool + """A holder can keep records from retention.""" + change_feed: builtins.bool + """One consumer can read every append in order.""" + cheap_logs: builtins.bool + """A log per topic of each chain costs nothing much.""" + delete_owner: builtins.bool + """The store deletes every stream of one owner.""" + topic_listing: builtins.bool + """The store lists the topics of one chain.""" + long_poll: builtins.bool + """A read can wait for the next record.""" + dense_positions: builtins.bool + """The next record's position is always the last one plus one.""" + @property + def replay_horizon(self) -> google.protobuf.duration_pb2.Duration: + """How long the store keeps records with no hold, when a positions-only reader can count on it.""" + max_record_bytes: builtins.int + """The largest record the store takes, with its envelope.""" + max_batch_bytes: builtins.int + """The most record bytes one append takes.""" + def __init__( + self, + *, + producer_state: builtins.bool = ..., + conditional_tail: builtins.bool = ..., + conditional_fence: builtins.bool = ..., + kept_last: builtins.bool = ..., + holds: builtins.bool = ..., + change_feed: builtins.bool = ..., + cheap_logs: builtins.bool = ..., + delete_owner: builtins.bool = ..., + topic_listing: builtins.bool = ..., + long_poll: builtins.bool = ..., + dense_positions: builtins.bool = ..., + replay_horizon: google.protobuf.duration_pb2.Duration | None = ..., + max_record_bytes: builtins.int = ..., + max_batch_bytes: builtins.int = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["replay_horizon", b"replay_horizon"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "change_feed", + b"change_feed", + "cheap_logs", + b"cheap_logs", + "conditional_fence", + b"conditional_fence", + "conditional_tail", + b"conditional_tail", + "delete_owner", + b"delete_owner", + "dense_positions", + b"dense_positions", + "holds", + b"holds", + "kept_last", + b"kept_last", + "long_poll", + b"long_poll", + "max_batch_bytes", + b"max_batch_bytes", + "max_record_bytes", + b"max_record_bytes", + "producer_state", + b"producer_state", + "replay_horizon", + b"replay_horizon", + "topic_listing", + b"topic_listing", + ], + ) -> None: ... + +global___StoreCapabilities = StoreCapabilities + class StreamFailure(google.protobuf.message.Message): DESCRIPTOR: google.protobuf.descriptor.Descriptor diff --git a/temporalio/bridge/proto/streams/v1/message_pb2.py b/temporalio/bridge/proto/streams/v1/message_pb2.py index 5222b2682d..747a2452ed 100644 --- a/temporalio/bridge/proto/streams/v1/message_pb2.py +++ b/temporalio/bridge/proto/streams/v1/message_pb2.py @@ -20,7 +20,7 @@ ) DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( - b'\n%temporal/sdk/streams/v1/message.proto\x12\x17temporal.sdk.streams.v1\x1a$temporal/api/common/v1/message.proto"\x8e\x03\n\x0cStreamRecord\x12-\n\x04\x62ody\x18\x01 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x45\n\x08metadata\x18\x02 \x03(\x0b\x32\x33.temporal.sdk.streams.v1.StreamRecord.MetadataEntry\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x37\n\x04kind\x18\x04 \x01(\x0e\x32).temporal.sdk.streams.v1.StreamRecordKind\x12\x13\n\x0bproducer_id\x18\x05 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x06 \x01(\x03\x12\x10\n\x08sequence\x18\x07 \x01(\x03\x12\x1d\n\x15task_started_event_id\x18\t \x01(\x03\x12\x17\n\x0f\x63ompleted_tasks\x18\n \x03(\x03\x1aP\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12.\n\x05value\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload:\x02\x38\x01*x\n\x10StreamRecordKind\x12"\n\x1eSTREAM_RECORD_KIND_UNSPECIFIED\x10\x00\x12\x1b\n\x17STREAM_RECORD_KIND_DATA\x10\x01\x12\x1d\n\x19STREAM_RECORD_KIND_FINISH\x10\x02"\x04\x08\x03\x10\x03\x42\x32\xea\x02/Temporalio::Internal::Bridge::Api::StreamRecordb\x06proto3' + b'\n%temporal/sdk/streams/v1/message.proto\x12\x17temporal.sdk.streams.v1\x1a$temporal/api/common/v1/message.proto"\x8e\x03\n\x0cStreamRecord\x12-\n\x04\x62ody\x18\x01 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x45\n\x08metadata\x18\x02 \x03(\x0b\x32\x33.temporal.sdk.streams.v1.StreamRecord.MetadataEntry\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x37\n\x04kind\x18\x04 \x01(\x0e\x32).temporal.sdk.streams.v1.StreamRecordKind\x12\x13\n\x0bproducer_id\x18\x05 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x06 \x01(\x03\x12\x10\n\x08sequence\x18\x07 \x01(\x03\x12\x1d\n\x15task_started_event_id\x18\t \x01(\x03\x12\x17\n\x0f\x63ompleted_tasks\x18\n \x03(\x03\x1aP\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12.\n\x05value\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload:\x02\x38\x01*\xba\x01\n\x10StreamRecordKind\x12"\n\x1eSTREAM_RECORD_KIND_UNSPECIFIED\x10\x00\x12\x1b\n\x17STREAM_RECORD_KIND_DATA\x10\x01\x12\x1d\n\x19STREAM_RECORD_KIND_FINISH\x10\x02\x12\x1c\n\x18STREAM_RECORD_KIND_CLOSE\x10\x05\x12\x1c\n\x18STREAM_RECORD_KIND_TOPIC\x10\x06"\x04\x08\x03\x10\x03"\x04\x08\x04\x10\x04\x42\x32\xea\x02/Temporalio::Internal::Bridge::Api::StreamRecordb\x06proto3' ) _STREAMRECORDKIND = DESCRIPTOR.enum_types_by_name["StreamRecordKind"] @@ -28,6 +28,8 @@ STREAM_RECORD_KIND_UNSPECIFIED = 0 STREAM_RECORD_KIND_DATA = 1 STREAM_RECORD_KIND_FINISH = 2 +STREAM_RECORD_KIND_CLOSE = 5 +STREAM_RECORD_KIND_TOPIC = 6 _STREAMRECORD = DESCRIPTOR.message_types_by_name["StreamRecord"] @@ -60,8 +62,8 @@ ) _STREAMRECORD_METADATAENTRY._options = None _STREAMRECORD_METADATAENTRY._serialized_options = b"8\001" - _STREAMRECORDKIND._serialized_start = 505 - _STREAMRECORDKIND._serialized_end = 625 + _STREAMRECORDKIND._serialized_start = 506 + _STREAMRECORDKIND._serialized_end = 692 _STREAMRECORD._serialized_start = 105 _STREAMRECORD._serialized_end = 503 _STREAMRECORD_METADATAENTRY._serialized_start = 423 diff --git a/temporalio/bridge/proto/streams/v1/message_pb2.pyi b/temporalio/bridge/proto/streams/v1/message_pb2.pyi index 49fb52f7e5..dc4042aa4c 100644 --- a/temporalio/bridge/proto/streams/v1/message_pb2.pyi +++ b/temporalio/bridge/proto/streams/v1/message_pb2.pyi @@ -41,6 +41,15 @@ class _StreamRecordKindEnumTypeWrapper( """The producer named by `producer_id` writes nothing more on `topic`. Says nothing about that producer's outcome and does not end the stream. """ + STREAM_RECORD_KIND_CLOSE: _StreamRecordKind.ValueType # 5 + """Ends `topic` for every reader, or every topic of the chain when `topic` + is empty. Readers drop every record after the first one. On a store + without close flags, producers and the owner's Core write it to close. + """ + STREAM_RECORD_KIND_TOPIC: _StreamRecordKind.ValueType # 6 + """Names `topic` in the chain's control log `_chain`, so a closer finds the + chain's topics on a store that can't list them. Topic readers skip it. + """ class StreamRecordKind(_StreamRecordKind, metaclass=_StreamRecordKindEnumTypeWrapper): """What a record means to a reader.""" @@ -53,6 +62,15 @@ STREAM_RECORD_KIND_FINISH: StreamRecordKind.ValueType # 2 """The producer named by `producer_id` writes nothing more on `topic`. Says nothing about that producer's outcome and does not end the stream. """ +STREAM_RECORD_KIND_CLOSE: StreamRecordKind.ValueType # 5 +"""Ends `topic` for every reader, or every topic of the chain when `topic` +is empty. Readers drop every record after the first one. On a store +without close flags, producers and the owner's Core write it to close. +""" +STREAM_RECORD_KIND_TOPIC: StreamRecordKind.ValueType # 6 +"""Names `topic` in the chain's control log `_chain`, so a closer finds the +chain's topics on a store that can't list them. Topic readers skip it. +""" global___StreamRecordKind = StreamRecordKind class StreamRecord(google.protobuf.message.Message): diff --git a/temporalio/bridge/sdk-core b/temporalio/bridge/sdk-core index a8cba03275..725bee46f8 160000 --- a/temporalio/bridge/sdk-core +++ b/temporalio/bridge/sdk-core @@ -1 +1 @@ -Subproject commit a8cba032753b19cfc339afd839ae9e5976754393 +Subproject commit 725bee46f85a6498f96317ba8a55004088cfab81 diff --git a/temporalio/bridge/streams_generated.py b/temporalio/bridge/streams_generated.py index 9c6b6971ce..d3896a9bbb 100644 --- a/temporalio/bridge/streams_generated.py +++ b/temporalio/bridge/streams_generated.py @@ -42,6 +42,14 @@ async def delete_owner( "DeleteOwner", req, temporalio.bridge.proto.streams.DeleteOwnerResponse ) + async def describe_store( + self, req: temporalio.bridge.proto.streams.DescribeStoreRequest + ) -> temporalio.bridge.proto.streams.DescribeStoreResponse: + """Invokes the StreamService.describe_store call.""" + return await self._store.call( + "DescribeStore", req, temporalio.bridge.proto.streams.DescribeStoreResponse + ) + async def latest( self, req: temporalio.bridge.proto.streams.LatestRequest ) -> temporalio.bridge.proto.streams.LatestResponse: diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index f288e2f5c9..0a707db485 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -11,12 +11,15 @@ The contract: 1. **Producers write with an identity.** A producer has an id, an attempt - and a sequence that starts at one for each attempt. A retry of the - newest batch with the same content is written once and returns the - original position. A retry with different content, or with a lower - sequence, is refused with :class:`StreamProducerError`. Content is - compared over the payloads the converter produced, before the codec, so - a codec with a fresh nonce per call still deduplicates a retry. + and a sequence that starts at one for each attempt. Every reader drops a + second copy of a record by that identity, and ends at the first + :attr:`RecordKind.CLOSE`. On a store with producer state (see + :class:`StoreCapabilities`), a retry of the newest batch with the same + content is written once and returns the original position. A retry with + different content, or with a lower sequence, is refused with + :class:`StreamProducerError`. Content is compared over the payloads the + converter produced, before the codec, so a codec with a fresh nonce per + call still deduplicates a retry. 2. **A new attempt supersedes the old one.** When a reader sees the first record of a producer's newer attempt, it yields a :attr:`RecordKind.SUPERSEDED` record first. It withdraws only what that @@ -61,6 +64,7 @@ as an ordinary payload. """ +from temporalio.contrib.streams._capabilities import StoreCapabilities from temporalio.contrib.streams._cursor import BEGINNING, END from temporalio.contrib.streams._errors import ( StreamClosedError, @@ -108,6 +112,7 @@ "DEFAULT_TOPIC", "END", "RecordKind", + "StoreCapabilities", "StreamClosedError", "StreamCursorError", "StreamError", diff --git a/temporalio/contrib/streams/_capabilities.py b/temporalio/contrib/streams/_capabilities.py new file mode 100644 index 0000000000..23acbf6edd --- /dev/null +++ b/temporalio/contrib/streams/_capabilities.py @@ -0,0 +1,147 @@ +"""What a store can do beyond the ordered log that every store keeps.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import timedelta +from typing import Any + +import temporalio.bridge.proto.streams +from temporalio.bridge.proto.streams import StreamFailureKind +from temporalio.bridge.streams import StreamCallFailure, StreamStore + +__all__ = ["StoreCapabilities"] + + +@dataclass(frozen=True) +class StoreCapabilities: + """The capabilities of a store, as Core reports them. + + Every store keeps an ordered log per topic, and Core applies the same + reader rules on every store. A capability changes what a producer is + told and how much the store keeps. Without one, Core takes a fallback, + with a weaker guarantee. + + As a mask for :class:`temporalio.contrib.streams.memory.MemoryStreams`, + a field set to ``False`` turns that capability off. A mask can't turn + on what the store doesn't have. A nonzero limit in a mask replaces the + store's own, so a test can split batches early. + """ + + producer_state: bool = True + """The store dedupes a producer's retry and refuses appends after a close. + + Without it, a retry can land a second copy, which readers drop, and an + append that races a close can answer success, which no reader delivers. + """ + conditional_tail: bool = True + """An append can require the log's next position.""" + conditional_fence: bool = True + """An append can require the stream's fence token.""" + kept_last: bool = True + """The store knows a log's last position after retention dropped it all. + + Without it, a cursor whose record retention dropped reads as expired, + also when nothing after it was lost. + """ + holds: bool = True + """A reader can hold records against retention.""" + change_feed: bool = True + """One consumer can read every append across all logs.""" + cheap_logs: bool = True + """The store keeps one log per topic of each run chain.""" + delete_owner: bool = True + """The store deletes every log of one owner. + + Without it, ``delete_workflow_streams`` raises + :class:`temporalio.contrib.streams.StreamUnsupportedError`. + """ + topic_listing: bool = True + """The store lists the topics of one run chain.""" + long_poll: bool = True + """A read can wait in the store for the next record.""" + dense_positions: bool = True + """Positions in a log have no gaps.""" + replay_horizon: timedelta | None = None + """How long the store keeps every record for sure, or ``None`` if it doesn't say.""" + max_record_bytes: int = 0 + """The largest record the store takes, with its envelope. + + A larger record fails the append with ``ValueError``. + """ + max_batch_bytes: int = 0 + """The most record bytes one append takes.""" + + @staticmethod + def _from_proto(proto: Any) -> StoreCapabilities: + return StoreCapabilities( + producer_state=proto.producer_state, + conditional_tail=proto.conditional_tail, + conditional_fence=proto.conditional_fence, + kept_last=proto.kept_last, + holds=proto.holds, + change_feed=proto.change_feed, + cheap_logs=proto.cheap_logs, + delete_owner=proto.delete_owner, + topic_listing=proto.topic_listing, + long_poll=proto.long_poll, + dense_positions=proto.dense_positions, + replay_horizon=proto.replay_horizon.ToTimedelta() + if proto.HasField("replay_horizon") + else None, + max_record_bytes=proto.max_record_bytes, + max_batch_bytes=proto.max_batch_bytes, + ) + + def _to_proto(self) -> Any: + """The mask as Core takes it. + + Raises: + ValueError: The Core in this SDK can't mask a store. + """ + proto_type = getattr(temporalio.bridge.proto.streams, "StoreCapabilities", None) + if proto_type is None: + raise ValueError("this SDK's Core can't mask a store's capabilities") + proto = proto_type( + producer_state=self.producer_state, + conditional_tail=self.conditional_tail, + conditional_fence=self.conditional_fence, + kept_last=self.kept_last, + holds=self.holds, + change_feed=self.change_feed, + cheap_logs=self.cheap_logs, + delete_owner=self.delete_owner, + topic_listing=self.topic_listing, + long_poll=self.long_poll, + dense_positions=self.dense_positions, + max_record_bytes=self.max_record_bytes, + max_batch_bytes=self.max_batch_bytes, + ) + if self.replay_horizon is not None: + proto.replay_horizon.FromTimedelta(self.replay_horizon) + return proto + + +# What a Core from before DescribeStore serves. Its Redis and memory stores have +# every capability that Core knew of, and none had a conditional append. +_BEFORE_DESCRIBE = StoreCapabilities(conditional_tail=False, conditional_fence=False) + + +async def describe(store: StreamStore) -> StoreCapabilities: + """What Core reports for ``store``. + + Raises: + StreamCallFailure: Core failed the call. + """ + proto = temporalio.bridge.proto.streams + request = getattr(proto, "DescribeStoreRequest", None) + response = getattr(proto, "DescribeStoreResponse", None) + if request is None or response is None: + return _BEFORE_DESCRIBE + try: + described = await store.call("DescribeStore", request(), response) + except StreamCallFailure as failure: + if failure.failure.kind == StreamFailureKind.STREAM_FAILURE_KIND_UNSUPPORTED: + return _BEFORE_DESCRIBE + raise + return StoreCapabilities._from_proto(described.capabilities) diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py index 5099b454b7..1771294026 100644 --- a/temporalio/contrib/streams/_errors.py +++ b/temporalio/contrib/streams/_errors.py @@ -87,7 +87,9 @@ class StreamClosedError(StreamRefusedError): """The stream is closed and refuses appends. The store refused the write, so nothing was written. Records the stream - retains stay readable. + retains stay readable. Only a store with producer state refuses. On + another store, an append that races a close can succeed, and no reader + delivers it. """ @@ -101,7 +103,8 @@ class StreamProducerError(StreamRefusedError): ``divergent`` is true when the sequence was already written with different content, so the caller changed the batch on a retry. Otherwise the sequence is stale, most often from a second producer object for the - same session. + same session. Only a store with producer state refuses. On another + store, the batch lands, and readers keep the first copy. """ def __init__(self, message: str, *, divergent: bool = False) -> None: @@ -116,7 +119,9 @@ class StreamOutcomeUnknownError(StreamStorageError): The connection failed or timed out after the request left. Retrying the same values on the same producer is safe: the producer keeps its sequence until an append succeeds, and the store returns the original - position for a retry that repeats a batch it already holds. + position for a retry that repeats a batch it already holds. A store + without producer state can keep a second copy, which readers drop, and + then it returns the copy's position. """ diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index 3285cb578a..d031c4b28e 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -125,7 +125,8 @@ async def append(self, *values: T) -> Cursor: :class:`temporalio.contrib.streams.StreamOutcomeUnknownError` can be repeated with the same values: the store answers a repeat of the batch it holds last for this producer attempt with the original - position and writes nothing. + position and writes nothing. A store without producer state writes + the repeat and answers its position, and readers drop the copy. An empty call writes nothing and returns the cursor of this producer's last record, or ``BEGINNING`` when it has written none. diff --git a/temporalio/contrib/streams/_plugin.py b/temporalio/contrib/streams/_plugin.py index 4dfa5125c8..c681849757 100644 --- a/temporalio/contrib/streams/_plugin.py +++ b/temporalio/contrib/streams/_plugin.py @@ -28,6 +28,7 @@ ) from temporalio.bridge.streams import StreamCallFailure, StreamStore from temporalio.bridge.streams_generated import StreamService +from temporalio.contrib.streams._capabilities import StoreCapabilities, describe from temporalio.contrib.streams._errors import error_from_failure from temporalio.plugin import SimplePlugin @@ -60,6 +61,7 @@ def __init__(self, name: str, config: StreamStoreConfig) -> None: self._config = config self._service: StreamService | None = None self._store: StreamStore | None = None + self._capabilities: StoreCapabilities | None = None self._connecting: asyncio.Lock | None = None self._service_client: ServiceClient | None = None # Whether Client.connect connected the store, which a Worker needs. @@ -149,6 +151,17 @@ def _temporal_stream_store(self) -> StreamStoreRef | None: # The Worker looks for this name on its plugins and gives the store to Core. return self._store.ref if self._store is not None else None + @property + def capabilities(self) -> StoreCapabilities: + """What the store can do, as Core found when a client connected it. + + Raises: + ValueError: No client connected this store yet. + """ + if self._capabilities is None: + raise ValueError(f"no client has connected stream store {self.name()!r}") + return self._capabilities + async def delete_workflow_streams(self, namespace: str, workflow_id: str) -> int: """Delete every stream of ``workflow_id`` in ``namespace``. @@ -167,6 +180,9 @@ async def delete_workflow_streams(self, namespace: str, workflow_id: str) -> int Raises: ValueError: No client connected this store yet. + temporalio.contrib.streams.StreamUnsupportedError: The store + can't delete by owner (see :attr:`capabilities`). Records go + at retention. temporalio.contrib.streams.StreamStorageError: The store failed; some keys may be deleted. """ @@ -206,8 +222,9 @@ async def _connected(self, service_client: ServiceClient) -> None: bridge: temporalio.bridge.client.Client = await cast( Any, service_client )._connected_client() - self._store = await call(StreamStore.connect(bridge, self._config)) - self._service = StreamService(self._store) + store = await call(StreamStore.connect(bridge, self._config)) + self._capabilities = await call(describe(store)) + self._store, self._service = store, StreamService(store) self._service_client = service_client diff --git a/temporalio/contrib/streams/_record.py b/temporalio/contrib/streams/_record.py index b1624b63bd..002ce419a8 100644 --- a/temporalio/contrib/streams/_record.py +++ b/temporalio/contrib/streams/_record.py @@ -64,6 +64,26 @@ class RecordKind(enum.IntEnum): earlier attempt's records twice is harmless and missing it is not. """ + # 4 is kept for a kind a later release stores. + + CLOSE = 5 + """The topic in ``topic`` is closed, or every topic of the run chain when empty. + + A producer writes it when it closes the topic, or when it learns that the + owner ended. The owner's Core writes it when the run chain ends through + a Workflow Task. Core's read ends at the first one and doesn't deliver + it, and it drops what an append that raced the close wrote after it. So + a reader sees one only in records that come from somewhere else. + """ + + TOPIC = 6 + """A producer started writing on the topic in ``topic``. + + It sits in the run chain's control log, on a store that can't list a + chain's topics, so a closer finds every topic to close. A read never + delivers it. + """ + @dataclass(frozen=True) class Cursor: diff --git a/temporalio/contrib/streams/_supersession.py b/temporalio/contrib/streams/_supersession.py index a7c4ebd2e3..2821f2f144 100644 --- a/temporalio/contrib/streams/_supersession.py +++ b/temporalio/contrib/streams/_supersession.py @@ -1,4 +1,4 @@ -"""Applying ``SUPERSEDED`` for a reader. +"""Applying ``SUPERSEDED`` and the other reader rules for a reader. A ``SUPERSEDED`` record withdraws only what one producer wrote in its earlier attempts. Records of other producers stay valid, wherever they sit @@ -6,6 +6,13 @@ find the withdrawn ones by producer and attempt, not by position. This module holds the records a read delivered under their producer, and names the ones each ``SUPERSEDED`` withdraws. + +It also applies the rules Core's reader applies on every store. It drops a +second copy of a record, which a store without producer state can hold, and +everything after the first ``CLOSE``. Core already dropped them from its own +reads, so on those the rules change nothing. They matter for records that +come from somewhere else, for example a copy of a read kept by the +application. """ from __future__ import annotations @@ -52,6 +59,9 @@ class SupersessionTracker(Generic[T]): it holds only the records of the newest Workflow Task prefix, since a rerun can't withdraw what a completed task wrote. + It drops a record as Core's reader does (see :meth:`drops`). Such a + record withdraws nothing and is not held. + A record with an empty ``producer_id`` or no attempt can't be withdrawn, so it is not held. Of one producer, it holds the newest ``max_records_per_producer`` records and drops the oldest past that, so @@ -83,6 +93,42 @@ def __init__( self._held: OrderedDict[str, list[tuple[int, StreamRecord[T]]]] = OrderedDict() self._bytes = 0 self._count = 0 + # Per producer, the newest attempt seen and its highest sequence. + self._marks: OrderedDict[str, tuple[int, int]] = OrderedDict() + self._mark_bytes = 0 + self._closed = False + # The read's topic, from its first record, so a CLOSE of another topic + # is told apart. + self._topic: str | None = None + + @property + def closed(self) -> bool: + """Whether the tracker saw a ``CLOSE``, so the stream ended.""" + return self._closed + + def drops(self, record: StreamRecord[T]) -> bool: + """Whether Core's reader rules drop ``record``, so a reader doesn't show it. + + ``CLOSE`` and ``TOPIC`` records are dropped, and so is every record + after the first ``CLOSE`` of the topic or of the run chain. So is a + copy of a record already seen, with the same producer and attempt + and a sequence at or below the highest seen for that attempt. A + store without producer state can hold such a copy after a retry of + an append whose answer was lost. A record with sequence 0 is never + a copy. Neither is a ``stale`` one, which an older attempt wrote. + Pass a dropped record to :meth:`observe` too, since a ``CLOSE`` + ends the stream there. + """ + if self._closed or record.kind in (RecordKind.CLOSE, RecordKind.TOPIC): + return True + if record.kind is RecordKind.SUPERSEDED or record.stale: + return False + mark = self._marks.get(record.producer_id) + return ( + mark is not None + and record.attempt == mark[0] + and 0 < record.sequence <= mark[1] + ) def observe(self, record: StreamRecord[T]) -> list[StreamRecord[T]]: """Take the next record of the read, and return the records it withdraws. @@ -94,9 +140,19 @@ def observe(self, record: StreamRecord[T]) -> list[StreamRecord[T]]: other tasks completed. That task is the new attempt's own on a rerun, or an earlier one whose rerun wrote nothing to the topic. For a record marked ``stale``, it is the record itself, since an older - attempt wrote it after the read delivered a newer one. For any other - record, nothing. + attempt wrote it after the read delivered a newer one. For a record + that :meth:`drops`, and for any other record, nothing. """ + if record.kind is RecordKind.CLOSE and not self._closed: + if not record.topic or self._topic in (None, record.topic): + self._closed = True + return [] + if self.drops(record): + return [] + if self._topic is None: + self._topic = record.topic + if record.kind is not RecordKind.SUPERSEDED: + self._mark(record) if record.kind is RecordKind.SUPERSEDED: supersession = record.supersession if supersession is None: @@ -137,6 +193,25 @@ def observe(self, record: StreamRecord[T]) -> list[StreamRecord[T]]: self.forget(next(iter(self._held))) return [] + def _mark(self, record: StreamRecord[T]) -> None: + if not record.producer_id: + return + mark = self._marks.get(record.producer_id) + if mark is None: + self._mark_bytes += _size(record.producer_id) + self._marks[record.producer_id] = (record.attempt, record.sequence) + elif record.attempt > mark[0]: + self._marks[record.producer_id] = (record.attempt, record.sequence) + elif record.attempt == mark[0] and record.sequence > mark[1]: + self._marks[record.producer_id] = (mark[0], record.sequence) + self._marks.move_to_end(record.producer_id) + while len(self._marks) > 1 and ( + len(self._marks) > self._max_producers + or self._mark_bytes > _MAX_STATE_BYTES + ): + forgotten, _ = self._marks.popitem(last=False) + self._mark_bytes -= _size(forgotten) + def standing(self) -> list[StreamRecord[T]]: """The held records no ``SUPERSEDED`` withdrew yet, in stream order.""" held = [entry for entries in self._held.values() for entry in entries] @@ -177,7 +252,9 @@ async def track_supersession( ) -> AsyncGenerator[tuple[StreamRecord[T], list[StreamRecord[T]]], None]: """Yield each record of ``records`` with the earlier records it withdraws. - A wrapper over :class:`SupersessionTracker` for a read:: + It skips the records that :meth:`SupersessionTracker.drops`, and stops + at the first ``CLOSE``. A wrapper over :class:`SupersessionTracker` + for a read:: async for record, withdrawn in track_supersession(handle.read()): for old in withdrawn: @@ -195,6 +272,11 @@ async def track_supersession( ) try: async for record in records: + if tracker.drops(record): + tracker.observe(record) + if tracker.closed: + return + continue yield record, tracker.observe(record) finally: close = getattr(records, "aclose", None) diff --git a/temporalio/contrib/streams/memory.py b/temporalio/contrib/streams/memory.py index 904edea881..08467c9cb4 100644 --- a/temporalio/contrib/streams/memory.py +++ b/temporalio/contrib/streams/memory.py @@ -3,8 +3,10 @@ from __future__ import annotations from datetime import timedelta +from typing import Any from temporalio.bridge.proto.streams import MemoryStoreConfig, StreamStoreConfig +from temporalio.contrib.streams._capabilities import StoreCapabilities from temporalio.contrib.streams._plugin import ( StreamStorePlugin, with_owner_start_grace, @@ -21,7 +23,12 @@ class MemoryStreams(StreamStorePlugin): none of it. """ - def __init__(self, *, owner_start_grace: timedelta | None = None) -> None: + def __init__( + self, + *, + owner_start_grace: timedelta | None = None, + capabilities: StoreCapabilities | None = None, + ) -> None: """Create the store's configuration. Args: @@ -30,13 +37,21 @@ def __init__(self, *, owner_start_grace: timedelta | None = None) -> None: :class:`temporalio.contrib.streams.StreamNotFoundError`, so a reader can start before its owner. 60 seconds when unset. Zero fails at once. + capabilities: A mask over the store's capabilities. A field set to + ``False`` turns that capability off, and Core takes its + fallback, as on a store that lacks it. For tests, and to try + the weaker guarantees. Every capability when unset. Raises: - ValueError: ``owner_start_grace`` is negative. + ValueError: ``owner_start_grace`` is negative, or a mask was + given to an SDK whose Core can't mask a store. """ + # Set only when asked for, so the config also fits a Core without masks. + mask: dict[str, Any] = {} + if capabilities is not None: + mask["capabilities"] = capabilities._to_proto() + memory = MemoryStoreConfig(**mask) super().__init__( "temporalio.contrib.streams.MemoryStreams", - with_owner_start_grace( - StreamStoreConfig(memory=MemoryStoreConfig()), owner_start_grace - ), + with_owner_start_grace(StreamStoreConfig(memory=memory), owner_start_grace), ) diff --git a/temporalio/contrib/streams/redis.py b/temporalio/contrib/streams/redis.py index 486e2a5d0e..f69c8ecf0e 100644 --- a/temporalio/contrib/streams/redis.py +++ b/temporalio/contrib/streams/redis.py @@ -4,6 +4,7 @@ from collections.abc import Sequence from datetime import timedelta +from typing import Any from temporalio.bridge.proto.streams import RedisStoreConfig, StreamStoreConfig from temporalio.contrib.streams._plugin import ( @@ -33,6 +34,7 @@ def __init__( blocking_reads_per_node: int | None = None, response_timeout: timedelta | None = None, owner_start_grace: timedelta | None = None, + plain: bool = False, ) -> None: """Create the store's configuration. @@ -59,6 +61,10 @@ def __init__( :class:`temporalio.contrib.streams.StreamNotFoundError`, so a reader can start before its owner. 60 seconds when unset. Zero fails at once. + plain: Run no Lua scripts, for a Redis that refuses them. The + store then has no producer state, no kept last position, no + holds and no change feed, and Core takes the fallback of + each (see :attr:`capabilities`). Raises: ValueError: No URL was given, ``key_prefix`` is empty, @@ -74,11 +80,14 @@ def __init__( raise ValueError( f"retention must be at least a millisecond, got {retention}" ) + # Set only when asked for, so the config also fits a Core without it. + extra: dict[str, Any] = {"plain": True} if plain else {} config = RedisStoreConfig( urls=urls, cluster=cluster, key_prefix=key_prefix, blocking_reads_per_node=blocking_reads_per_node or 0, + **extra, ) config.retention.FromTimedelta(retention) if response_timeout is not None: diff --git a/tests/contrib/streams/test_store_capabilities.py b/tests/contrib/streams/test_store_capabilities.py new file mode 100644 index 0000000000..144d109a66 --- /dev/null +++ b/tests/contrib/streams/test_store_capabilities.py @@ -0,0 +1,163 @@ +"""A store reports its capabilities, and Core's fallbacks keep reads whole. + +The masked memory store and Redis with ``plain=True`` lack producer state, +so a retry or a second producer object can land a copy. Readers must not +see it. Redis runs when ``STREAMS_REDIS_URL`` is set. +""" + +from __future__ import annotations + +import dataclasses +import os +import uuid + +import pytest + +import temporalio.bridge.proto.streams +from temporalio.bridge.proto.streams import StreamFailure, StreamFailureKind +from temporalio.bridge.streams import StreamCallFailure +from temporalio.client import Client +from temporalio.contrib.streams import ( + StoreCapabilities, + StreamProducerError, + StreamStorePlugin, + StreamUnsupportedError, + get_stream_handle, + topic, +) +from temporalio.contrib.streams._capabilities import describe +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.contrib.streams.redis import RedisStreams +from tests.contrib.streams._support import connect_with, read_all +from tests.contrib.streams.test_outside_handles import Waits, running_owner +from tests.helpers import new_worker + +TOKENS = topic("tokens", str) +# What a store without producer state looks like, on the memory store. +NO_PRODUCER_STATE = StoreCapabilities(producer_state=False, kept_last=False) +# A Core from before DescribeStore can't mask a store or run Redis plain. +HAS_DESCRIBE = hasattr(temporalio.bridge.proto.streams, "DescribeStoreRequest") +NO_DESCRIBE = "this Core can't describe or mask a store" +needs_describe = pytest.mark.skipif(not HAS_DESCRIBE, reason=NO_DESCRIBE) + + +def redis_url() -> str: + url = os.environ.get("STREAMS_REDIS_URL", "") + if not url: + pytest.skip("set STREAMS_REDIS_URL to run on Redis") + return url + + +def store(kind: str) -> StreamStorePlugin: + if kind in ("masked", "plain") and not HAS_DESCRIBE: + pytest.skip(NO_DESCRIBE) + if kind == "memory": + return MemoryStreams() + if kind == "masked": + return MemoryStreams(capabilities=NO_PRODUCER_STATE) + prefix = f"capabilities-{uuid.uuid4().hex}" + return RedisStreams(redis_url(), key_prefix=prefix, plain=kind == "plain") + + +@pytest.fixture(params=["memory", "masked", "redis", "plain"]) +async def streams(request: pytest.FixtureRequest, client: Client) -> Client: + return await connect_with(client, store(request.param)) + + +def capabilities_of(client: Client) -> StoreCapabilities: + (plugin,) = client.config()["plugins"] + assert isinstance(plugin, StreamStorePlugin) + return plugin.capabilities + + +def test_capabilities_need_a_connected_store(): + with pytest.raises(ValueError, match="no client has connected"): + MemoryStreams().capabilities + + +@needs_describe +async def test_a_mask_turns_off_only_what_it_names(client: Client): + full = capabilities_of(await connect_with(client, MemoryStreams())) + assert full.producer_state and full.kept_last and full.delete_owner + masked = capabilities_of(await connect_with(client, store("masked"))) + assert masked == dataclasses.replace(full, producer_state=False, kept_last=False) + + +@needs_describe +async def test_plain_redis_has_no_store_side_state(client: Client): + plain = capabilities_of(await connect_with(client, store("plain"))) + assert not (plain.producer_state or plain.kept_last) + assert not (plain.holds or plain.change_feed) + full = capabilities_of(await connect_with(client, store("redis"))) + assert full.producer_state and full.kept_last + + +async def test_a_reader_sees_one_copy_of_each_batch(streams: Client): + has_state = capabilities_of(streams).producer_state + async with new_worker(streams, Waits) as worker: + workflow_id = await running_owner(streams, worker.task_queue) + stream = get_stream_handle(streams, workflow_id, topic=TOKENS) + first = await stream.producer(producer_id="p", attempt=1).append("a", "b") + # A second object for the same session repeats the batch, as a retry + # after a lost answer does. + again = stream.producer(producer_id="p", attempt=1) + assert (await again.append("a", "b") == first) == has_state + divergent = stream.producer(producer_id="p", attempt=1) + if has_state: + with pytest.raises(StreamProducerError): + await divergent.append("x", "y") + else: + await divergent.append("x", "y") + await again.append("c") + owner = streams.get_workflow_handle(workflow_id) + await owner.signal(Waits.finish) + await owner.result() + records = await read_all(stream.read()) + assert [r.value for r in records] == ["a", "b", "c"] + + +@needs_describe +async def test_a_store_that_cannot_delete_by_owner_says_so(client: Client): + masked = MemoryStreams(capabilities=StoreCapabilities(delete_owner=False)) + await connect_with(client, masked) + assert not masked.capabilities.delete_owner + with pytest.raises(StreamUnsupportedError): + await masked.delete_workflow_streams(client.namespace, "any") + + +class _CoreWithoutDescribe: + """A store of a Core that knows no DescribeStore call.""" + + def __init__(self, kind: StreamFailureKind.ValueType) -> None: + self.kind = kind + + async def call(self, rpc: str, _req: object, _resp_type: type) -> object: + raise StreamCallFailure(StreamFailure(kind=self.kind, message=f"no {rpc}")) + + +async def test_a_core_without_describe_serves_every_capability( + monkeypatch: pytest.MonkeyPatch, +): + # The fake store never parses, so any message types stand in for the call's. + proto = temporalio.bridge.proto.streams + monkeypatch.setattr(proto, "DescribeStoreRequest", StreamFailure, raising=False) + monkeypatch.setattr(proto, "DescribeStoreResponse", StreamFailure, raising=False) + unknown = _CoreWithoutDescribe(StreamFailureKind.STREAM_FAILURE_KIND_UNSUPPORTED) + capabilities = await describe(unknown) # type: ignore[arg-type] + assert capabilities.producer_state and capabilities.kept_last + assert capabilities.delete_owner and capabilities.topic_listing + assert not (capabilities.conditional_tail or capabilities.conditional_fence) + # Any other failure is the store's, not a missing call. + failing = _CoreWithoutDescribe(StreamFailureKind.STREAM_FAILURE_KIND_STORAGE) + with pytest.raises(StreamCallFailure): + await describe(failing) # type: ignore[arg-type] + + +async def test_an_sdk_without_the_describe_protos_serves_every_capability( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.delattr( + temporalio.bridge.proto.streams, "DescribeStoreRequest", raising=False + ) + store = _CoreWithoutDescribe(StreamFailureKind.STREAM_FAILURE_KIND_STORAGE) + assert (await describe(store)).producer_state # type: ignore[arg-type] diff --git a/tests/contrib/streams/test_supersession.py b/tests/contrib/streams/test_supersession.py index 5f2889ee3e..516de7bd66 100644 --- a/tests/contrib/streams/test_supersession.py +++ b/tests/contrib/streams/test_supersession.py @@ -2,6 +2,7 @@ from __future__ import annotations +import dataclasses import json from collections.abc import AsyncIterator from pathlib import Path @@ -31,18 +32,31 @@ from tests.helpers import new_worker -def data(producer: str, attempt: int, value: str, stale: bool = False) -> StreamRecord: +def data( + producer: str, attempt: int, value: str, stale: bool = False, sequence: int = 0 +) -> StreamRecord: return StreamRecord( kind=RecordKind.DATA, cursor=Cursor(value), topic="t", producer_id=producer, attempt=attempt, + sequence=sequence, value=value, stale=stale, ) +def close(producer: str) -> StreamRecord: + return StreamRecord( + kind=RecordKind.CLOSE, + cursor=Cursor(f"close-{producer}"), + topic="t", + producer_id=producer, + attempt=1, + ) + + def superseded(producer: str, previous: int, attempt: int) -> StreamRecord: return StreamRecord( kind=RecordKind.SUPERSEDED, @@ -145,6 +159,86 @@ async def records() -> AsyncIterator[StreamRecord[str]]: assert closed == [True] +def test_a_second_copy_of_a_batch_is_dropped(): + tracker: SupersessionTracker[str] = SupersessionTracker() + first = [data("a", 1, "a1", sequence=1), data("a", 1, "a2", sequence=2)] + for record in [*first, data("b", 1, "b1", sequence=1)]: + assert not tracker.drops(record) + tracker.observe(record) + # A retry after a lost answer landed both records again. + copies = [data("a", 1, "a1", sequence=1), data("a", 1, "a2", sequence=2)] + assert all(tracker.drops(copy) for copy in copies) + assert all(tracker.observe(copy) == [] for copy in copies) + assert not tracker.drops(data("a", 1, "a3", sequence=3)) + # Unnumbered records are never copies. + assert not tracker.drops(data("c", 1, "c", sequence=0)) + tracker.observe(data("c", 1, "c", sequence=0)) + assert not tracker.drops(data("c", 1, "c", sequence=0)) + assert [r.value for r in tracker.standing()] == ["a1", "a2", "b1", "c"] + + +def test_a_copy_from_an_older_attempt_stays_stale(): + tracker: SupersessionTracker[str] = SupersessionTracker() + tracker.observe(data("a", 1, "a1", sequence=1)) + tracker.observe(superseded("a", 1, 2)) + tracker.observe(data("a", 2, "b1", sequence=1)) + late = data("a", 1, "a1", stale=True, sequence=1) + assert not tracker.drops(late) + assert tracker.observe(late) == [late] + assert tracker.drops(data("a", 2, "b1", sequence=1)) + + +def test_the_first_close_ends_the_stream(): + tracker: SupersessionTracker[str] = SupersessionTracker() + tracker.observe(data("a", 1, "a1", sequence=1)) + # A read never delivers a close, but the tracker must see it. + assert tracker.drops(close("a")) + assert tracker.observe(close("a")) == [] + assert tracker.closed + # An append that raced the close, and a second closer. + for late in (data("b", 1, "b1", sequence=1), close("b")): + assert tracker.drops(late) + assert tracker.observe(late) == [] + assert [r.value for r in tracker.standing()] == ["a1"] + + +async def test_track_supersession_skips_copies_and_stops_at_the_close(): + closed = [] + + async def records() -> AsyncIterator[StreamRecord[str]]: + try: + yield data("a", 1, "a1", sequence=1) + yield data("a", 1, "a1", sequence=1) + yield data("a", 1, "a2", sequence=2) + yield close("a") + yield data("a", 1, "never read", sequence=3) + finally: + closed.append(True) + + seen = [record async for record, _ in track_supersession(records())] + assert [r.value for r in seen] == ["a1", "a2"] + assert closed == [True] + + +def test_a_close_of_another_topic_does_not_end_the_stream(): + tracker: SupersessionTracker[str] = SupersessionTracker() + tracker.observe(data("a", 1, "a1", sequence=1)) + other = dataclasses.replace(close("c"), topic="other") + assert tracker.drops(other) + tracker.observe(other) + assert not tracker.closed + tracker.observe(dataclasses.replace(close("c"), topic="")) + assert tracker.closed + + +def test_the_tracker_forgets_the_copies_of_producers_it_saw_least_recently(): + tracker: SupersessionTracker[str] = SupersessionTracker(max_producers=2) + for producer in ("a", "b", "c"): + tracker.observe(data(producer, 1, producer, sequence=1)) + assert not tracker.drops(data("a", 1, "a", sequence=1)) + assert tracker.drops(data("c", 1, "c", sequence=1)) + + def test_a_workflow_rerun_withdraws_only_its_own_tasks_records(): tracker: SupersessionTracker[str] = SupersessionTracker() task_one, task_two = 1 << 16, 7 << 16 @@ -193,6 +287,44 @@ def test_a_workflow_supersession_withdraws_what_cores_vectors_say( assert [r.value for r in withdrawn] == (["previous"] if case["superseded"] else []) +_READER_RULES = _VECTORS.with_name("reader_rules.json") + + +@pytest.mark.parametrize( + "case", + # A Core from before the reader rules has no vectors for them. + json.loads(_READER_RULES.read_text())["cases"] if _READER_RULES.exists() else [], + ids=lambda case: case["label"], +) +def test_the_tracker_drops_what_cores_reader_rules_drop(case: dict[str, Any]): + records: list[StreamRecord[str]] = [ + StreamRecord( + kind=RecordKind(stored.get("kind", 1)), + cursor=Cursor(str(index)), + topic=stored.get("topic", "out"), + producer_id=stored["producer"], + attempt=stored["attempt"], + sequence=stored["sequence"], + stale=index in case.get("stale", []), + ) + for index, stored in enumerate(case["records"]) + ] + tracker: SupersessionTracker[str] = SupersessionTracker() + start = case.get("resume_after") + if start is not None: + # A resumed read knows only the record at its cursor. + tracker.observe(records[start]) + delivered = [] + for index, record in enumerate(records): + if start is not None and index <= start: + continue + if not tracker.drops(record): + delivered.append(index) + tracker.observe(record) + assert delivered == case["delivered"] + assert tracker.closed == case.get("closed", False) + + def test_a_workflow_producer_holds_only_its_newest_tasks_records(): tracker: SupersessionTracker[str] = SupersessionTracker() tracker.observe(data("@run", 1 << 16 | 1, "x"))