From fa579ac767fefd1287ef7110477ed4fd2b9b9dab Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 23:11:55 -0700 Subject: [PATCH] Vendored the Activity owner kind and added it to the stream protos. A standalone Activity owns streams in Jetty. The API names it by kind and Activity id, and the stream protos carry it with the attempt start time that orders Activity attempts. --- .../temporal/api/enums/v1/stream.proto | 6 +++++ .../temporal/api/stream/v1/message.proto | 12 ++++++---- .../temporal/sdk/core/streams/streams.proto | 24 +++++++++++++++++-- crates/sdk-core-c-bridge/src/tests/mod.rs | 2 ++ crates/sdk-core/CHANGELOG.md | 5 ++++ .../src/core_tests/external_streams.rs | 1 + crates/sdk-core/src/streams.rs | 9 +++++++ .../src/worker/workflow/stream_output.rs | 1 + crates/streams/src/driver.rs | 2 ++ crates/streams/src/handle.rs | 5 ++++ crates/streams/src/memory.rs | 2 ++ crates/streams/src/notify.rs | 2 ++ crates/streams/src/reader.rs | 1 + crates/streams/src/record.rs | 1 + crates/streams/src/redis/keys.rs | 4 ++++ crates/streams/src/redis/mod.rs | 1 + crates/streams/src/service.rs | 3 +++ crates/streams/src/subscribe.rs | 4 ++++ crates/streams/tests/conformance.rs | 3 +++ crates/streams/tests/cross_impl.rs | 2 ++ crates/streams/tests/driver.rs | 2 ++ crates/streams/tests/redis_store.rs | 8 +++++++ 22 files changed, 94 insertions(+), 6 deletions(-) diff --git a/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto index 4a079844ce..6c2dfa80c5 100644 --- a/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto +++ b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto @@ -16,4 +16,10 @@ enum StreamOwnerKind { // failure. If the Workflow's run chain ends with no close and callbacks are attached, the // stream notifier closes the stream with a failure within `streamnotifier.ownerCheckInterval`. STREAM_OWNER_KIND_WORKFLOW = 1; + // A standalone Activity owns the stream. The Activity closes the stream itself, with a result + // or a failure. If the Activity run reaches a terminal status with no close and callbacks + // are attached, the stream notifier closes the stream within + // `streamnotifier.ownerCheckInterval`. A completed run gives no result. A failed, canceled, + // terminated or timed-out run gives the run's own failure. + STREAM_OWNER_KIND_ACTIVITY = 2; } diff --git a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto index 2576cdb42f..755fa7c7bb 100644 --- a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto @@ -18,14 +18,18 @@ import "temporal/api/failure/v1/message.proto"; // Names one stream within a namespace: its owner and its topic. message StreamReference { temporal.api.enums.v1.StreamOwnerKind owner_kind = 1; - // The owning Workflow, for `STREAM_OWNER_KIND_WORKFLOW`. + // The owning Workflow, for `STREAM_OWNER_KIND_WORKFLOW`. Empty for other kinds. string workflow_id = 2; - // The id of the first run in the owning Workflow's run chain. The stream notifier is keyed by - // it, so one notifier serves the stream across the chain's runs, and a stream that reuses the - // Workflow id in a new chain gets a notifier of its own. An empty run id is refused. + // For `STREAM_OWNER_KIND_WORKFLOW`, the id of the first run in the owning Workflow's run + // chain. The stream notifier is keyed by it, so one notifier serves the stream across the + // chain's runs, and a stream that reuses the Workflow id in a new chain gets a notifier of its + // own. For `STREAM_OWNER_KIND_ACTIVITY`, the id of the owning Activity run. An empty run id is + // refused. string run_id = 3; // The stream's topic within its owner. string topic = 4; + // The owning standalone Activity, for `STREAM_OWNER_KIND_ACTIVITY`. Empty for other kinds. + string activity_id = 5; } // A callback attached to a stream notifier, as DescribeStreamNotifier reports it. diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto index df86d7b891..2aacf8c76d 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -4,6 +4,7 @@ package coresdk.streams; option ruby_package = "Temporalio::Internal::Bridge::Api::Streams"; import "google/protobuf/duration.proto"; +import "google/protobuf/timestamp.proto"; import "temporal/api/common/v1/message.proto"; import "temporal/api/failure/v1/message.proto"; import "temporal/sdk/streams/v1/message.proto"; @@ -120,25 +121,37 @@ message MemoryStoreConfig {} enum StreamOwnerKind { STREAM_OWNER_KIND_UNSPECIFIED = 0; STREAM_OWNER_KIND_WORKFLOW = 1; + // A standalone Activity run. It is a chain of one run, so its streams end with the run. + STREAM_OWNER_KIND_ACTIVITY = 2; } // One stream, a topic of an owner in a namespace. message StreamAddress { string namespace = 1; StreamOwnerKind owner_kind = 2; + // The owning Workflow. Empty for an Activity owner. string workflow_id = 3; // Pins the stream to the run chain that holds this run. Empty follows the chain that holds the - // Workflow id when the call starts. + // Workflow id when the call starts. For an Activity owner, the Activity run, and empty names + // the newest run of the Activity id when the call starts. string run_id = 4; string topic = 5; + // The owning standalone Activity. Empty for a Workflow owner. + string activity_id = 6; } // A Workflow's run chain. Its streams are keyed by its first run, so they follow -// Continue-as-New, and a new chain on the same Workflow id gets streams of its own. +// Continue-as-New, and a new chain on the same Workflow id gets streams of its own. A standalone +// Activity run is a chain of one run, with the run as `first_run_id`. message ChainId { string namespace = 1; + // Empty for an Activity owner. string workflow_id = 2; string first_run_id = 3; + // `UNSPECIFIED` is a Workflow owner, so a chain from before the Activity kind stays valid. + StreamOwnerKind owner_kind = 4; + // Empty for a Workflow owner. + string activity_id = 5; } // A producer that names itself. @@ -156,6 +169,10 @@ message ActivityProducer { string run_id = 2; string activity_id = 3; int32 attempt = 4; + // When the attempt started, as the Activity task gives it. With it, Core writes the attempt as + // `started_unix_ms << 16 | min(attempt, 65535)`, so a later attempt and a reset attempt both + // supersede an earlier one. Unset writes `attempt` as it is. + google.protobuf.Timestamp attempt_started_time = 5; } // One record lang appends. @@ -418,7 +435,10 @@ message RunStreamDriverResponse {} message DeleteOwnerRequest { string namespace = 1; StreamOwnerKind owner_kind = 2; + // The Workflow whose streams go, for a Workflow owner. string workflow_id = 3; + // The standalone Activity whose streams go, across all its runs, for an Activity owner. + string activity_id = 4; } message DeleteOwnerResponse { diff --git a/crates/sdk-core-c-bridge/src/tests/mod.rs b/crates/sdk-core-c-bridge/src/tests/mod.rs index 95c3cabf12..f96ecf862d 100644 --- a/crates/sdk-core-c-bridge/src/tests/mod.rs +++ b/crates/sdk-core-c-bridge/src/tests/mod.rs @@ -537,6 +537,7 @@ fn test_stream_store_appends_and_reads_through_the_c_bridge() { workflow_id: "wf".to_string(), run_id: String::new(), topic: "out".to_string(), + ..Default::default() }; let appended = context @@ -632,6 +633,7 @@ fn test_a_cancelled_stream_read_says_it_only_stopped_reading() { workflow_id: "wf".to_string(), run_id: String::new(), topic: "out".to_string(), + ..Default::default() }), after: "$end".to_string(), wait: Some(std::time::Duration::from_secs(30).try_into().unwrap()), diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index a78ed897a8..b53a33df68 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -144,6 +144,11 @@ relevant information. retry would be refused the same way, also when the run closed the topic itself before its task reran. Any other store failure still fails the task. Local Activities that a completion with stream output schedules start only once that output is in the store. +* Experimental: the stream protos name a standalone Activity owner. `StreamOwnerKind` gains + `ACTIVITY`, `StreamAddress` and `DeleteOwnerRequest` gain `activity_id`, and `ChainId` gains + `owner_kind` and `activity_id`. `ActivityProducer` gains `attempt_started_time`. The vendored + API's `StreamReference` gains `activity_id` and its owner kind gains `ACTIVITY`. Core still + refuses the Activity kind as `UNSUPPORTED`. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting diff --git a/crates/sdk-core/src/core_tests/external_streams.rs b/crates/sdk-core/src/core_tests/external_streams.rs index d239cc3da4..added0df89 100644 --- a/crates/sdk-core/src/core_tests/external_streams.rs +++ b/crates/sdk-core/src/core_tests/external_streams.rs @@ -1153,6 +1153,7 @@ async fn a_topic_closed_before_the_publish_drops_its_records_and_the_task_goes_o namespace: crate::test_help::NAMESPACE.to_string(), workflow_id: "fakeid".to_string(), first_run_id: String::new(), + ..Default::default() }; let appended_at_completion: Arc>> = Default::default(); let history = timer_history(); diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs index 809e031fae..9472b81590 100644 --- a/crates/sdk-core/src/streams.rs +++ b/crates/sdk-core/src/streams.rs @@ -146,6 +146,7 @@ fn chain_reference(chain: &ChainId, topic: String) -> StreamReference { workflow_id: chain.workflow_id.clone(), run_id: chain.first_run_id.clone(), topic, + ..Default::default() } } @@ -372,6 +373,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: "run-1".to_string(), + ..Default::default() }, topic: "tokens".to_string(), position: "cursor".to_string(), @@ -386,6 +388,7 @@ mod tests { workflow_id: "wf".to_string(), run_id: "run-1".to_string(), topic: "tokens".to_string(), + ..Default::default() }) ); assert_eq!((request.position.as_str(), request.counter), ("cursor", 7)); @@ -400,6 +403,7 @@ mod tests { namespace: namespace.to_string(), workflow_id: workflow_id.to_string(), first_run_id: "run-1".to_string(), + ..Default::default() }, topic: "tokens".to_string(), position: format!("cursor-{counter}"), @@ -444,6 +448,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "owner".to_string(), first_run_id: "chain-1".to_string(), + ..Default::default() }, topic: "tokens".to_string(), request_id: "jetty/reader-run/1".to_string(), @@ -459,6 +464,7 @@ mod tests { workflow_id: "owner".to_string(), run_id: "chain-1".to_string(), topic: "tokens".to_string(), + ..Default::default() }) ); assert_eq!(request.request_id, "jetty/reader-run/1"); @@ -682,6 +688,7 @@ mod tests { workflow_id: workflow_id.clone(), run_id: started.run_id.clone(), topic: "tokens".to_string(), + ..Default::default() }), request_id: "caller".to_string(), callback: Some( @@ -714,6 +721,7 @@ mod tests { workflow_id: workflow_id.clone(), run_id: String::new(), topic: "tokens".to_string(), + ..Default::default() }; Some(Live { connection, @@ -855,6 +863,7 @@ mod tests { workflow_id: live.workflow_id.clone(), run_id: live.run_id.clone(), topic: "tokens".to_string(), + ..Default::default() }), } .into_request(), diff --git a/crates/sdk-core/src/worker/workflow/stream_output.rs b/crates/sdk-core/src/worker/workflow/stream_output.rs index faa6afb1da..0d995f0850 100644 --- a/crates/sdk-core/src/worker/workflow/stream_output.rs +++ b/crates/sdk-core/src/worker/workflow/stream_output.rs @@ -137,6 +137,7 @@ impl OutputStore { namespace: self.namespace.clone(), workflow_id: workflow_id.to_string(), first_run_id: first_run_id.to_string(), + ..Default::default() } } diff --git a/crates/streams/src/driver.rs b/crates/streams/src/driver.rs index 4a8a364230..d396b69a5e 100644 --- a/crates/streams/src/driver.rs +++ b/crates/streams/src/driver.rs @@ -269,6 +269,7 @@ impl ChainKey { namespace: self.0, workflow_id: self.1, first_run_id: self.2, + ..Default::default() } } } @@ -326,6 +327,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: workflow_id.to_string(), first_run_id: "run-1".to_string(), + ..Default::default() } } diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 0a74385d63..9bd025d33b 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -944,6 +944,7 @@ fn chain_of_key(owner: &OwnerKey, first_run_id: &str) -> ChainId { namespace: owner.namespace.clone(), workflow_id: owner.workflow_id.clone(), first_run_id: first_run_id.to_string(), + ..Default::default() } } @@ -1088,6 +1089,7 @@ mod tests { workflow_id: "wf".to_string(), run_id: String::new(), topic: topic.to_string(), + ..Default::default() } } @@ -1120,6 +1122,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: first_run_id.to_string(), + ..Default::default() } } @@ -1183,6 +1186,7 @@ mod tests { run_id: "run-1".to_string(), activity_id: "fetch".to_string(), attempt: 2, + ..Default::default() })); setup.streams.append(request).await.unwrap(); let read = setup @@ -2261,6 +2265,7 @@ mod tests { namespace: "ns".to_string(), owner_kind: StreamOwnerKind::Workflow as i32, workflow_id: "wf".to_string(), + ..Default::default() }; assert!( setup diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index 9877027401..85d46b9867 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -620,6 +620,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: "run-1".to_string(), + ..Default::default() } } @@ -680,6 +681,7 @@ mod tests { namespace: "ns".to_string(), owner_kind: StreamOwnerKind::Unspecified as i32, workflow_id: "wf".to_string(), + ..Default::default() }) .await .unwrap_err(); diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index 4b253c9144..982d6b6fa4 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -217,6 +217,7 @@ impl StreamKey { namespace: self.namespace.clone(), workflow_id: self.workflow_id.clone(), first_run_id: self.first_run_id.clone(), + ..Default::default() } } } @@ -905,6 +906,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: first_run_id.to_string(), + ..Default::default() } } diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index 1649c2a182..7d0396a9b4 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -546,6 +546,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: "run-1".to_string(), + ..Default::default() }, topic: "out".to_string(), stream_hash: stream_hash("ns", "workflow", "wf", "out"), diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs index 1d01d0b958..94d975ce90 100644 --- a/crates/streams/src/record.rs +++ b/crates/streams/src/record.rs @@ -188,6 +188,7 @@ mod tests { run_id: "run-1".to_string(), activity_id: "fetch".to_string(), attempt: 3, + ..Default::default() }; assert_eq!(activity_producer_id(&activity), "fetch@run-1"); } diff --git a/crates/streams/src/redis/keys.rs b/crates/streams/src/redis/keys.rs index 5928da8488..16328c9534 100644 --- a/crates/streams/src/redis/keys.rs +++ b/crates/streams/src/redis/keys.rs @@ -106,6 +106,7 @@ pub(crate) fn chain_of_log(prefix: &str, key: &str) -> Option<(ChainId, String)> namespace: unpart(namespace)?, workflow_id: unpart(workflow_id)?, first_run_id: unpart(first_run_id)?, + ..Default::default() }, unpart(topic)?, )) @@ -166,6 +167,7 @@ mod tests { namespace: text(case, "namespace").to_string(), workflow_id: text(case, "workflow_id").to_string(), first_run_id: text(case, "first_run_id").to_string(), + ..Default::default() }, ); let topic = text(case, "topic"); @@ -184,6 +186,7 @@ mod tests { namespace: text(case, "namespace").to_string(), workflow_id: text(case, "workflow_id").to_string(), first_run_id: text(case, "first_run_id").to_string(), + ..Default::default() }; let prefix = text(case, "prefix"); let keys = ChainKeys::new(prefix, &chain); @@ -218,6 +221,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "wf".to_string(), first_run_id: "run-1".to_string(), + ..Default::default() }; let pattern = owner_pattern("p", "ns", "wf"); assert_eq!(pattern, "p:{ns:wf:*"); diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index ca6b6ffcc6..c710af5017 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -873,6 +873,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: workflow_id.to_string(), first_run_id: "run-1".to_string(), + ..Default::default() } } diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 18aa0de077..40a6210ef0 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -375,6 +375,7 @@ mod tests { workflow_id: "wf".to_string(), run_id: String::new(), topic: "out".to_string(), + ..Default::default() } } @@ -578,6 +579,7 @@ mod tests { namespace: "ns".to_string(), owner_kind: StreamOwnerKind::Unspecified as i32, workflow_id: "wf".to_string(), + ..Default::default() }; let failure = service .call("DeleteOwner", &other_owner.encode_to_vec()) @@ -588,6 +590,7 @@ mod tests { namespace: "ns".to_string(), owner_kind: StreamOwnerKind::Workflow as i32, workflow_id: String::new(), + ..Default::default() }; let failure = service .call("DeleteOwner", &no_id.encode_to_vec()) diff --git a/crates/streams/src/subscribe.rs b/crates/streams/src/subscribe.rs index 8326d9b001..da2745c28f 100644 --- a/crates/streams/src/subscribe.rs +++ b/crates/streams/src/subscribe.rs @@ -183,6 +183,7 @@ impl Subscriptions { namespace: stream.namespace.clone(), workflow_id: stream.workflow_id.clone(), first_run_id: attached.first_run_id, + ..Default::default() }, None => streams.chain_now(stream).await?, }; @@ -363,6 +364,7 @@ mod tests { workflow_id: "owner".to_string(), run_id: String::new(), topic: "tokens".to_string(), + ..Default::default() } } @@ -391,6 +393,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "owner".to_string(), first_run_id: first_run_id.to_string(), + ..Default::default() } } @@ -641,6 +644,7 @@ mod tests { namespace: "ns".to_string(), workflow_id: "owner".to_string(), first_run_id, + ..Default::default() }), topic: "tokens".to_string(), producer_id: "p".to_string(), diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 0bcd9789c3..54618b4f75 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -58,6 +58,7 @@ impl Case for Memory { namespace: "default".to_string(), workflow_id: unique("conformance"), first_run_id: unique("run"), + ..Default::default() } } @@ -383,6 +384,7 @@ mod on_redis { namespace: "default".to_string(), workflow_id: unique("conformance"), first_run_id: unique("run"), + ..Default::default() } } @@ -1445,6 +1447,7 @@ mod cases { namespace: chain.namespace.clone(), owner_kind: StreamOwnerKind::Workflow as i32, workflow_id: chain.workflow_id.clone(), + ..Default::default() }; assert!(store.delete_owner(request.clone()).await.unwrap().deleted > 0); assert_eq!(store.delete_owner(request).await.unwrap().deleted, 0); diff --git a/crates/streams/tests/cross_impl.rs b/crates/streams/tests/cross_impl.rs index 70585ff96f..eef5e36e84 100644 --- a/crates/streams/tests/cross_impl.rs +++ b/crates/streams/tests/cross_impl.rs @@ -104,6 +104,7 @@ impl Both { namespace: NAMESPACE.to_string(), workflow_id: self.workflow_id.clone(), first_run_id: FIRST_RUN.to_string(), + ..Default::default() } } @@ -303,6 +304,7 @@ impl Both { workflow_id: self.workflow_id.clone(), run_id: String::new(), topic: "events".to_string(), + ..Default::default() }), producer: Some(Producer::Named(NamedProducer { producer_id: "p".to_string(), diff --git a/crates/streams/tests/driver.rs b/crates/streams/tests/driver.rs index c6d06d070c..46ae80cc67 100644 --- a/crates/streams/tests/driver.rs +++ b/crates/streams/tests/driver.rs @@ -131,6 +131,7 @@ async fn write( namespace: "ns".to_string(), workflow_id: workflow_id.clone(), first_run_id: "run-1".to_string(), + ..Default::default() }), topic: "out".to_string(), producer_id: "p".to_string(), @@ -422,6 +423,7 @@ async fn traffic( namespace: "ns".to_string(), workflow_id: workflow_id.clone(), first_run_id: "run-1".to_string(), + ..Default::default() }), topic: "out".to_string(), producer_id: "p".to_string(), diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index 7694cbfa04..cd28ce58f7 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -87,6 +87,7 @@ fn chain() -> ChainId { namespace: "ns".to_string(), workflow_id: unique("wf"), first_run_id: "run-1".to_string(), + ..Default::default() } } @@ -319,6 +320,7 @@ async fn ids_with_separators_and_braces_keep_their_own_keys() { namespace: "ns".to_string(), workflow_id: format!("{}:{{b}}", unique("a")), first_run_id: "c".to_string(), + ..Default::default() }; let (head, tail) = one.workflow_id.split_once(':').unwrap(); let other = ChainId { @@ -806,6 +808,7 @@ async fn a_new_producer_marks_an_ended_chain_closed_until_it_expires() { workflow_id: chain.workflow_id.clone(), run_id: String::new(), topic: "events".to_string(), + ..Default::default() }), producer: Some(Producer::Named(NamedProducer { producer_id: producer.to_string(), @@ -1274,6 +1277,7 @@ fn delete_request(workflow_id: &str) -> temporalio_streams::proto::DeleteOwnerRe namespace: "ns".to_string(), owner_kind: temporalio_streams::proto::StreamOwnerKind::Workflow as i32, workflow_id: workflow_id.to_string(), + ..Default::default() } } @@ -1425,6 +1429,7 @@ async fn a_delete_reaches_every_primary_of_a_cluster() { namespace: "ns".to_string(), workflow_id: workflow_id.clone(), first_run_id: format!("run-{run}"), + ..Default::default() }; store .append(append(&chain, "p", 1, 1, b"\x01")) @@ -1731,6 +1736,7 @@ async fn a_replay_read_on_redis_repeats_the_read_and_survives_retention_under_a_ workflow_id: chain.workflow_id.clone(), run_id: String::new(), topic: "events".to_string(), + ..Default::default() }; let append = |attempt: i64, sequence: i64, value: &str| AppendRequest { stream: Some(stream.clone()), @@ -1817,6 +1823,7 @@ async fn a_replay_read_from_a_trimmed_start_on_redis_is_rec_01() { workflow_id: chain.workflow_id.clone(), run_id: String::new(), topic: "events".to_string(), + ..Default::default() }; let append = |sequence: i64| AppendRequest { stream: Some(stream.clone()), @@ -1897,6 +1904,7 @@ async fn a_chains_notification_entry_lands_in_the_chains_own_slot() { namespace: unique("ns:with{braces}"), workflow_id: unique("wf"), first_run_id: unique("run"), + ..Default::default() }; let landed = store .append(append(&chain, "p", 1, 1, b"\x01"))