From f455501cb6a998e165831a7d05eed7880302fdd6 Mon Sep 17 00:00:00 2001 From: James Watkins-Harvey Date: Mon, 24 Aug 2026 18:53:18 -0400 Subject: [PATCH] Add experimental Event Groups support --- CHANGELOG.md | 7 +- crates/sdk-core/Cargo.toml | 1 + .../workflow_tests/event_groups.rs | 560 ++++++++++++++++-- crates/sdk/src/lib.rs | 2 +- crates/workflow/Cargo.toml | 1 + crates/workflow/src/event_groups.rs | 309 ++++++++++ crates/workflow/src/lib.rs | 2 + crates/workflow/src/runtime/instance.rs | 15 +- crates/workflow/src/workflow_context.rs | 235 ++++++-- .../workflow/src/workflow_context/options.rs | 127 ++-- 10 files changed, 1105 insertions(+), 154 deletions(-) create mode 100644 crates/workflow/src/event_groups.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index e54b18e45..5dacf694b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,9 +34,10 @@ relevant information. ## Unreleased ### Added -* `WorkflowContext::all_handlers_finished` and `SyncWorkflowContext::all_handlers_finished` let - Rust workflows wait for active signal and update handler chains before completing or continuing - as new. +* Experimental Event Groups APIs: `EventGroup`, `WorkflowContext::create_event_group`, + `with_event_group` / `with_event_groups`, and `event_groups` on timer, activity, local + activity, child workflow, signal-external, and Nexus operation options. Signal and update + handlers automatically attach an implicit inbound group. This API may change without notice. * `WorkflowStartOptions::memo` attaches a non-indexed memo when starting a workflow, using the same `MemoValues` type already used by continue-as-new and `WorkflowContext::upsert_memo`. Values are serialized with the client's payload converter and codec, matching how `describe` diff --git a/crates/sdk-core/Cargo.toml b/crates/sdk-core/Cargo.toml index 48c8c2f07..4c673b1eb 100644 --- a/crates/sdk-core/Cargo.toml +++ b/crates/sdk-core/Cargo.toml @@ -147,6 +147,7 @@ version = "0.7" [dev-dependencies] assert_matches = "1.5" +sha1 = { version = "0.10", default-features = false } bimap = "0.6.3" bytes = "1.10" clap = { version = "4.5", features = ["derive"] } diff --git a/crates/sdk-core/tests/integ_tests/workflow_tests/event_groups.rs b/crates/sdk-core/tests/integ_tests/workflow_tests/event_groups.rs index 8d2ee63a2..dae6e6509 100644 --- a/crates/sdk-core/tests/integ_tests/workflow_tests/event_groups.rs +++ b/crates/sdk-core/tests/integ_tests/workflow_tests/event_groups.rs @@ -1,28 +1,25 @@ -//! Verify that `EventGroupMarker`s attached to lang-side options propagate all the -//! way down to the server-side `Command`s issued by Core. One mocked test per command -//! kind we currently expose `event_group_markers` on: activity, child workflow, timer, -//! local activity. +//! Event Groups user-facing SDK tests. //! -//! Plus one end-to-end test against a real server, verifying that the markers also -//! land on the resulting `HistoryEvent` (i.e. the server persists what we send). -//! -//! Event Groups are not implemented in the Rust SDK, so these tests build markers as raw -//! protos and set them through the `#[doc(hidden)]` `event_group_markers` option fields, -//! which exist for that purpose only. +//! Mocked tests check that `event_groups` on command options reach Core commands. History tests +//! cover label IDs, scopes, aggregation, implicit handlers, and a few command kinds. -use std::time::Duration; +use std::{collections::HashSet, time::Duration}; use crate::common::{ CoreWfStarter, activity_functions::StdActivities, build_fake_sdk_with_options, mock_sdk_cfg_with_options, }; -use temporalio_client::{UntypedWorkflow, WorkflowStartOptions}; +use sha1::{Digest, Sha1}; +use temporalio_client::{ + UntypedWorkflow, WorkflowExecuteUpdateOptions, WorkflowSignalOptions, WorkflowStartOptions, +}; use temporalio_common::{ data_converters::RawValue, protos::{ coresdk::AsJsonPayloadExt, temporal::api::{ enums::v1::{CommandType, EventType}, + history::v1::{History, history_event}, sdk::v1::{ EventGroupMarker, event_group_marker::{Label, Variant}, @@ -32,8 +29,8 @@ use temporalio_common::{ }; use temporalio_macros::{workflow, workflow_methods}; use temporalio_sdk::{ - ActivityOptions, ChildWorkflowOptions, LocalActivityOptions, TimerOptions, WorkflowContext, - WorkflowResult, + ActivityOptions, ChildWorkflowOptions, EventGroup, LocalActivityOptions, TimerOptions, + WorkflowContext, WorkflowResult, }; use temporalio_sdk_core::{ replay::{DEFAULT_WORKFLOW_TYPE, canned_histories}, @@ -46,6 +43,10 @@ async fn pass_event_group_markers_on_schedule_activity() { let mut mock_cfg = MockPollCfg::from_hist_builder(t); let wf_id = mock_cfg.hists[0].wf_id.clone(); let wf_type = DEFAULT_WORKFLOW_TYPE; + let expected_groups = vec![EventGroup::with_id( + "activity-group-label", + "activity-group", + )]; let expected_markers = vec![label_marker("activity-group", "activity-group-label")]; let expected_for_assert = expected_markers.clone(); @@ -71,19 +72,19 @@ async fn pass_event_group_markers_on_schedule_activity() { #[workflow] struct ActivityWithGroupWorkflow { - event_group_markers: Vec, + event_groups: Vec, } #[workflow_methods(factory_only)] impl ActivityWithGroupWorkflow { #[run(name = DEFAULT_WORKFLOW_TYPE)] async fn run(ctx: &mut WorkflowContext) -> WorkflowResult<()> { - let event_group_markers = ctx.state(|wf| wf.event_group_markers.clone()); + let event_groups = ctx.state(|wf| wf.event_groups.clone()); ctx.execute_activity( StdActivities::default, (), ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5)) - .event_group_markers(event_group_markers) + .event_groups(event_groups) .build(), ) .await?; @@ -97,7 +98,7 @@ async fn pass_event_group_markers_on_schedule_activity() { |options| { options .register_workflow_with_factory(move || ActivityWithGroupWorkflow { - event_group_markers: expected_markers.clone(), + event_groups: expected_groups.clone(), }) .unwrap(); }, @@ -120,6 +121,7 @@ async fn pass_event_group_markers_on_start_child_workflow() { let wf_type = DEFAULT_WORKFLOW_TYPE; let t = canned_histories::single_child_workflow(wf_id); let mut mock_cfg = MockPollCfg::from_hist_builder(t); + let expected_groups = vec![EventGroup::with_id("child-group-label", "child-group")]; let expected_markers = vec![label_marker("child-group", "child-group-label")]; let expected_for_assert = expected_markers.clone(); @@ -146,21 +148,21 @@ async fn pass_event_group_markers_on_start_child_workflow() { #[workflow] struct ChildWithGroupWorkflow { child_wf_id: String, - event_group_markers: Vec, + event_groups: Vec, } #[workflow_methods(factory_only)] impl ChildWithGroupWorkflow { #[run(name = DEFAULT_WORKFLOW_TYPE)] async fn run(ctx: &mut WorkflowContext) -> WorkflowResult<()> { - let (child_wf_id, event_group_markers) = - ctx.state(|wf| (wf.child_wf_id.clone(), wf.event_group_markers.clone())); + let (child_wf_id, event_groups) = + ctx.state(|wf| (wf.child_wf_id.clone(), wf.event_groups.clone())); ctx.start_child_workflow( UntypedWorkflow::new("child"), RawValue::new(vec![]), ChildWorkflowOptions::builder() .workflow_id(child_wf_id) - .event_group_markers(event_group_markers) + .event_groups(event_groups) .build(), ) .await?; @@ -169,7 +171,7 @@ async fn pass_event_group_markers_on_start_child_workflow() { } let child_wf_id = wf_id.to_string(); - let event_group_markers_for_wf = expected_markers.clone(); + let event_groups_for_wf = expected_groups.clone(); let mut worker = mock_sdk_cfg_with_options( mock_cfg, |_| {}, @@ -177,7 +179,7 @@ async fn pass_event_group_markers_on_start_child_workflow() { options .register_workflow_with_factory(move || ChildWithGroupWorkflow { child_wf_id: child_wf_id.clone(), - event_group_markers: event_group_markers_for_wf.clone(), + event_groups: event_groups_for_wf.clone(), }) .unwrap(); }, @@ -200,6 +202,7 @@ async fn pass_event_group_markers_on_start_timer() { let mut mock_cfg = MockPollCfg::from_hist_builder(t); let wf_id = mock_cfg.hists[0].wf_id.clone(); let wf_type = DEFAULT_WORKFLOW_TYPE; + let expected_groups = vec![EventGroup::with_id("timer-group-label", "timer-group")]; let expected_markers = vec![label_marker("timer-group", "timer-group-label")]; let expected_for_assert = expected_markers.clone(); @@ -222,17 +225,17 @@ async fn pass_event_group_markers_on_start_timer() { #[workflow] struct TimerWithGroupWorkflow { - event_group_markers: Vec, + event_groups: Vec, } #[workflow_methods(factory_only)] impl TimerWithGroupWorkflow { #[run(name = DEFAULT_WORKFLOW_TYPE)] async fn run(ctx: &mut WorkflowContext) -> WorkflowResult<()> { - let event_group_markers = ctx.state(|wf| wf.event_group_markers.clone()); + let event_groups = ctx.state(|wf| wf.event_groups.clone()); ctx.timer( TimerOptions::builder(Duration::from_secs(1)) - .event_group_markers(event_group_markers) + .event_groups(event_groups) .build(), ) .await; @@ -240,14 +243,14 @@ async fn pass_event_group_markers_on_start_timer() { } } - let event_group_markers_for_wf = expected_markers.clone(); + let event_groups_for_wf = expected_groups.clone(); let mut worker = mock_sdk_cfg_with_options( mock_cfg, |_| {}, |options| { options .register_workflow_with_factory(move || TimerWithGroupWorkflow { - event_group_markers: event_group_markers_for_wf.clone(), + event_groups: event_groups_for_wf.clone(), }) .unwrap(); }, @@ -266,19 +269,18 @@ async fn pass_event_group_markers_on_start_timer() { /// Local activities pose some particular challenges: the corresponding `RecordMarker` command /// only gets created at a later point, after the local activity completes execution. -/// server, so instead of a command of their own they produce a `RecordMarker` command that -/// Core synthesizes when the activity resolves. Markers attached to the `ScheduleLocalActivity` -/// command have to survive that indirection and end up on the marker command. #[tokio::test] async fn pass_event_group_markers_on_schedule_local_activity() { let t = canned_histories::single_local_activity("1"); let mut mock_cfg = MockPollCfg::from_hist_builder(t); + let expected_groups = vec![EventGroup::with_id( + "local-activity-label", + "local-activity-group", + )]; let expected_markers = vec![label_marker("local-activity-group", "local-activity-label")]; let expected_for_assert = expected_markers.clone(); mock_cfg.completion_asserts_from_expectations(|mut asserts| { - // The activity resolves within the same workflow task that scheduled it, so the marker - // command is flushed together with the workflow completion rather than on its own. asserts.then(move |wft| { assert_eq!(wft.commands.len(), 2); assert_eq!(wft.commands[0].command_type(), CommandType::RecordMarker); @@ -293,19 +295,19 @@ async fn pass_event_group_markers_on_schedule_local_activity() { #[workflow] struct LocalActivityWithGroupWorkflow { - event_group_markers: Vec, + event_groups: Vec, } #[workflow_methods(factory_only)] impl LocalActivityWithGroupWorkflow { #[run(name = DEFAULT_WORKFLOW_TYPE)] async fn run(ctx: &mut WorkflowContext) -> WorkflowResult<()> { - let event_group_markers = ctx.state(|wf| wf.event_group_markers.clone()); + let event_groups = ctx.state(|wf| wf.event_groups.clone()); ctx.execute_local_activity( StdActivities::default, (), LocalActivityOptions::builder() - .event_group_markers(event_group_markers) + .event_groups(event_groups) .build(), ) .await?; @@ -313,13 +315,10 @@ async fn pass_event_group_markers_on_schedule_local_activity() { } } - // Unlike the tests above, this one drives a plain SDK worker off the canned history rather - // than submitting a workflow: the local activity must actually run for a marker to be - // recorded, so the worker needs the activity implementation registered too. let mut worker = build_fake_sdk_with_options(mock_cfg, |options| { options .register_workflow_with_factory(move || LocalActivityWithGroupWorkflow { - event_group_markers: expected_markers.clone(), + event_groups: expected_groups.clone(), }) .unwrap() .register_activities(StdActivities); @@ -327,8 +326,6 @@ async fn pass_event_group_markers_on_schedule_local_activity() { worker.run().await.unwrap(); } -// Constants used by the real-server test below; defining them at module scope so the -// workflow body and the assertion can construct the same marker independently. const PERSIST_TEST_MARKER_ID: &str = "persist-test"; const PERSIST_TEST_MARKER_LABEL: &str = "persist-test-label"; const PERSIST_TEST_LA_MARKER_ID: &str = "persist-test-la"; @@ -346,9 +343,9 @@ impl ActivityEventGroupPersistsWf { StdActivities::default, (), ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5)) - .event_group_markers(vec![label_marker( - PERSIST_TEST_MARKER_ID, + .event_groups(vec![EventGroup::with_id( PERSIST_TEST_MARKER_LABEL, + PERSIST_TEST_MARKER_ID, )]) .build(), ) @@ -358,9 +355,9 @@ impl ActivityEventGroupPersistsWf { (), LocalActivityOptions::builder() .start_to_close_timeout(Duration::from_secs(5)) - .event_group_markers(vec![label_marker( - PERSIST_TEST_LA_MARKER_ID, + .event_groups(vec![EventGroup::with_id( PERSIST_TEST_LA_MARKER_LABEL, + PERSIST_TEST_LA_MARKER_ID, )]) .build(), ) @@ -369,9 +366,6 @@ impl ActivityEventGroupPersistsWf { } } -/// End-to-end: a marker attached to a command must also land on the resulting history event -/// after the server persists it. Covers both an ordinary activity (`ActivityTaskScheduled`) and -/// a local activity, which surfaces as the `MarkerRecorded` event Core writes on resolution. #[tokio::test] async fn event_group_markers_persist_to_history_events() { let wf_name = "event_group_markers_persist_to_history_events"; @@ -416,11 +410,475 @@ async fn event_group_markers_persist_to_history_events() { ); } +#[workflow] +#[derive(Default)] +struct LabelsAndScopesWf; + +#[workflow_methods] +impl LabelsAndScopesWf { + #[run(name = "event_groups_labels_and_scopes")] + async fn run(ctx: &mut WorkflowContext) -> WorkflowResult<()> { + let payment = ctx.create_event_group("payment-processing"); + let same_a = ctx.create_event_group("bbb"); + let same_b = ctx.create_event_group("bbb"); + let customer = EventGroup::with_id("customer-james-watkins", "customer-123456"); + + ctx.execute_activity( + StdActivities::echo, + "activity-a".to_string(), + ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5)) + .event_groups(vec![payment.clone()]) + .build(), + ) + .await?; + ctx.execute_activity( + StdActivities::echo, + "activity-b1".to_string(), + ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5)) + .event_groups(vec![same_a]) + .build(), + ) + .await?; + ctx.execute_activity( + StdActivities::echo, + "activity-b2".to_string(), + ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5)) + .event_groups(vec![same_b]) + .build(), + ) + .await?; + ctx.execute_activity( + StdActivities::echo, + "explicit".to_string(), + ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5)) + .event_groups(vec![payment.clone(), customer.clone()]) + .build(), + ) + .await?; + + let scoped = ctx.with_event_group(payment.clone()); + scoped + .execute_activity( + StdActivities::echo, + "scoped".to_string(), + ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5)).build(), + ) + .await?; + scoped + .execute_activity( + StdActivities::echo, + "scoped-and-explicit".to_string(), + ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5)) + .event_groups(vec![payment.clone()]) + .build(), + ) + .await?; + let nested = scoped.with_event_group(customer); + nested.timer(Duration::from_millis(1)).await; + ctx.execute_activity( + StdActivities::echo, + "unscoped".to_string(), + ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5)).build(), + ) + .await?; + Ok(()) + } +} + +#[tokio::test] +async fn event_groups_labels_scopes_and_aggregation() { + let wf_name = "event_groups_labels_and_scopes"; + let mut starter = CoreWfStarter::new(wf_name); + starter + .sdk_config + .register_activities(StdActivities) + .register_workflow::() + .unwrap(); + let mut worker = starter.worker().await; + starter.start_with_worker(wf_name, &mut worker).await; + worker.run_until_done().await.unwrap(); + + let history = starter.get_history().await; + let original_run_id = original_execution_run_id(&history); + let payment_id = derived_group_id(original_run_id, "payment-processing"); + let bbb_id = derived_group_id(original_run_id, "bbb"); + let scheduled = activities_by_input(&history); + assert_eq!( + label_ids(&scheduled["activity-a"]), + set([payment_id.clone()]) + ); + assert_eq!( + label_ids(&scheduled["activity-b1"]), + label_ids(&scheduled["activity-b2"]) + ); + assert_eq!(label_ids(&scheduled["activity-b1"]), set([bbb_id])); + assert_eq!( + label_ids(&scheduled["explicit"]), + set([payment_id.clone(), "customer-123456".to_string()]) + ); + assert_eq!(label_ids(&scheduled["scoped"]), set([payment_id.clone()])); + assert_eq!( + label_ids(&scheduled["scoped-and-explicit"]), + set([payment_id.clone()]) + ); + assert!(label_ids(&scheduled["unscoped"]).is_empty()); + + let timer = history + .events + .iter() + .find(|e| e.event_type() == EventType::TimerStarted) + .expect("nested-scope timer"); + assert_eq!( + label_ids(timer), + set([payment_id, "customer-123456".to_string()]) + ); + + let a_label = scheduled["activity-a"].event_group_markers[0] + .variant + .as_ref() + .and_then(|variant| match variant { + Variant::Label(label) => label.label.as_ref(), + _ => None, + }) + .expect("label payload"); + assert_eq!( + a_label.metadata.get("encoding").map(Vec::as_slice), + Some(b"json/plain".as_slice()) + ); + assert_eq!(a_label.data, b"\"payment-processing\""); +} + +#[workflow] +#[derive(Default)] +struct CommandsWf; + +#[workflow] +#[derive(Default)] +struct CommandsChildWf; + +#[workflow_methods] +impl CommandsChildWf { + #[run] + async fn run(_ctx: &mut WorkflowContext) -> WorkflowResult<()> { + Ok(()) + } +} + +#[workflow_methods] +impl CommandsWf { + #[run(name = "event_groups_commands")] + async fn run(ctx: &mut WorkflowContext) -> WorkflowResult<()> { + let group = EventGroup::with_id("command-label", "command-id"); + ctx.timer( + TimerOptions::builder(Duration::from_millis(1)) + .event_groups(vec![group.clone()]) + .build(), + ) + .await; + ctx.execute_activity( + StdActivities::default, + (), + ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5)) + .event_groups(vec![group.clone()]) + .build(), + ) + .await?; + ctx.execute_local_activity( + StdActivities::default, + (), + LocalActivityOptions::builder() + .start_to_close_timeout(Duration::from_secs(5)) + .event_groups(vec![group.clone()]) + .build(), + ) + .await?; + let started = ctx + .start_child_workflow( + CommandsChildWf::run, + (), + ChildWorkflowOptions::builder() + .event_groups(vec![group]) + .build(), + ) + .await + .expect("child starts"); + started.result().await?; + Ok(()) + } +} + +#[tokio::test] +async fn event_groups_attach_to_timer_activity_la_and_child() { + let wf_name = "event_groups_commands"; + let mut starter = CoreWfStarter::new(wf_name); + starter + .sdk_config + .register_activities(StdActivities) + .register_workflow::() + .unwrap(); + starter + .sdk_config + .register_workflow::() + .unwrap(); + let mut worker = starter.worker().await; + starter.start_with_worker(wf_name, &mut worker).await; + worker.run_until_done().await.unwrap(); + + let history = starter.get_history().await; + let expected = set(["command-id".to_string()]); + let timer = history + .events + .iter() + .find(|e| e.event_type() == EventType::TimerStarted) + .unwrap(); + assert_eq!(label_ids(timer), expected); + let activity = history + .events + .iter() + .find(|e| e.event_type() == EventType::ActivityTaskScheduled) + .unwrap(); + assert_eq!(label_ids(activity), expected); + let local = history + .events + .iter() + .find(|e| e.event_type() == EventType::MarkerRecorded) + .unwrap(); + assert_eq!(label_ids(local), expected); + let child = history + .events + .iter() + .find(|e| e.event_type() == EventType::StartChildWorkflowExecutionInitiated) + .unwrap(); + assert_eq!(label_ids(child), expected); +} + +#[workflow] +#[derive(Default)] +struct ImplicitHandlersWf { + done: bool, +} + +#[workflow_methods] +impl ImplicitHandlersWf { + #[run(name = "event_groups_implicit_handlers")] + async fn run(ctx: &mut WorkflowContext) -> WorkflowResult<()> { + ctx.wait_condition(|wf| wf.done).await?; + Ok(()) + } + + #[signal] + async fn ping(ctx: &mut WorkflowContext) { + ctx.timer(Duration::from_millis(1)).await; + } + + #[update] + async fn poke( + ctx: &mut WorkflowContext, + ) -> Result<(), Box> { + ctx.timer(Duration::from_millis(1)).await; + Ok(()) + } + + #[signal] + fn done(&mut self, _ctx: &mut temporalio_sdk::SyncWorkflowContext) { + self.done = true; + } +} + +#[tokio::test] +async fn event_groups_implicit_signal_and_update_handlers() { + let wf_name = "event_groups_implicit_handlers"; + let mut starter = CoreWfStarter::new(wf_name); + starter + .sdk_config + .register_workflow::() + .unwrap(); + let mut worker = starter.worker().await; + let task_queue = starter.get_task_queue().to_owned(); + let handle = worker + .submit_workflow( + ImplicitHandlersWf::run, + (), + WorkflowStartOptions::new(task_queue, starter.get_wf_id().to_owned()).build(), + ) + .await + .unwrap(); + let drive = async { + handle + .signal( + ImplicitHandlersWf::ping, + (), + WorkflowSignalOptions::default(), + ) + .await + .unwrap(); + handle + .execute_update( + ImplicitHandlersWf::poke, + (), + WorkflowExecuteUpdateOptions::default(), + ) + .await + .unwrap(); + handle + .signal( + ImplicitHandlersWf::done, + (), + WorkflowSignalOptions::default(), + ) + .await + .unwrap(); + }; + let run = async { + worker.run_until_done().await.unwrap(); + }; + tokio::join!(drive, run); + + let history = starter.get_history().await; + let signal_event = history + .events + .iter() + .find(|e| { + e.event_type() == EventType::WorkflowExecutionSignaled && signal_name(e) == Some("ping") + }) + .expect("ping signal event"); + let update_id = history + .events + .iter() + .find_map(|e| match e.attributes.as_ref()? { + history_event::Attributes::WorkflowExecutionUpdateAcceptedEventAttributes(attrs) => { + Some( + attrs + .accepted_request + .as_ref()? + .meta + .as_ref()? + .update_id + .clone(), + ) + } + _ => None, + }) + .expect("accepted update id"); + + let timers: Vec<_> = history + .events + .iter() + .filter(|e| e.event_type() == EventType::TimerStarted) + .collect(); + assert_eq!(timers.len(), 2); + let signal_timer = timers + .iter() + .find(|e| inbound_event_id(e) == Some(signal_event.event_id)) + .expect("timer inherits inbound signal group"); + assert!(label_ids(signal_timer).is_empty()); + let update_timer = timers + .iter() + .find(|e| inbound_update_id(e).as_deref() == Some(update_id.as_str())) + .expect("timer inherits inbound update group"); + assert!(label_ids(update_timer).is_empty()); +} + fn label_marker(id: &str, label: &str) -> EventGroupMarker { EventGroupMarker { variant: Some(Variant::Label(Label { id: id.to_string(), label: Some(label.as_json_payload().unwrap()), })), - } as EventGroupMarker + } +} + +fn derived_group_id(original_execution_run_id: &str, label: &str) -> String { + let mut hasher = Sha1::new(); + hasher.update(original_execution_run_id.as_bytes()); + hasher.update(label.as_bytes()); + format!("{:x}", hasher.finalize()) +} + +fn original_execution_run_id(history: &History) -> &str { + history + .events + .iter() + .find_map(|event| match event.attributes.as_ref()? { + history_event::Attributes::WorkflowExecutionStartedEventAttributes(attrs) => { + Some(attrs.original_execution_run_id.as_str()) + } + _ => None, + }) + .expect("WorkflowExecutionStarted") +} + +fn activities_by_input( + history: &History, +) -> std::collections::HashMap< + String, + &temporalio_common::protos::temporal::api::history::v1::HistoryEvent, +> { + history + .events + .iter() + .filter(|event| event.event_type() == EventType::ActivityTaskScheduled) + .filter_map(|event| { + let attrs = match event.attributes.as_ref()? { + history_event::Attributes::ActivityTaskScheduledEventAttributes(attrs) => attrs, + _ => return None, + }; + let payload = attrs.input.as_ref()?.payloads.first()?; + let name = String::from_utf8(payload.data.clone()).ok()?; + let name = name.trim_matches('"').to_string(); + Some((name, event)) + }) + .collect() +} + +fn label_ids( + event: &temporalio_common::protos::temporal::api::history::v1::HistoryEvent, +) -> HashSet { + event + .event_group_markers + .iter() + .filter_map(|marker| match marker.variant.as_ref()? { + Variant::Label(label) => Some(label.id.clone()), + _ => None, + }) + .collect() +} + +fn inbound_event_id( + event: &temporalio_common::protos::temporal::api::history::v1::HistoryEvent, +) -> Option { + event + .event_group_markers + .iter() + .find_map(|marker| match marker.variant.as_ref()? { + Variant::InboundEvent(inbound) => Some(inbound.inbound_event_id), + _ => None, + }) +} + +fn inbound_update_id( + event: &temporalio_common::protos::temporal::api::history::v1::HistoryEvent, +) -> Option { + event + .event_group_markers + .iter() + .find_map(|marker| match marker.variant.as_ref()? { + Variant::InboundUpdate(inbound) => Some(inbound.inbound_update_id.clone()), + _ => None, + }) +} + +fn signal_name( + event: &temporalio_common::protos::temporal::api::history::v1::HistoryEvent, +) -> Option<&str> { + match event.attributes.as_ref()? { + history_event::Attributes::WorkflowExecutionSignaledEventAttributes(attrs) => { + Some(attrs.signal_name.as_str()) + } + _ => None, + } +} + +fn set, const N: usize>(ids: [T; N]) -> HashSet { + ids.into_iter().map(Into::into).collect() } diff --git a/crates/sdk/src/lib.rs b/crates/sdk/src/lib.rs index 9a43864cd..db44efd1c 100644 --- a/crates/sdk/src/lib.rs +++ b/crates/sdk/src/lib.rs @@ -96,7 +96,7 @@ pub use temporalio_client::Namespace; pub use temporalio_workflow::{ ActivityCancellationType, ActivityCloseTimeouts, ActivityOptions, BaseWorkflowContext, CancellableFuture, CancellableFutureWithReason, ChildWorkflowCancellationType, - ChildWorkflowOptions, ContinueAsNewOptions, ContinueAsNewVersioningBehavior, + ChildWorkflowOptions, ContinueAsNewOptions, ContinueAsNewVersioningBehavior, EventGroup, ExternalWorkflowHandle, LocalActivityOptions, MemoValue, NexusOperationCancellationType, NexusOperationOptions, ParentClosePolicy, PatchActivationCallback, SignalWorkflowOptions, StartChildWorkflowExecutionFailedCause, StartChildWorkflowOutput, StartedChildWorkflow, diff --git a/crates/workflow/Cargo.toml b/crates/workflow/Cargo.toml index 89c572a17..559c22829 100644 --- a/crates/workflow/Cargo.toml +++ b/crates/workflow/Cargo.toml @@ -27,6 +27,7 @@ prost-types = { workspace = true } rand = { version = "0.10", default-features = false } rand_pcg = "0.10" serde = { version = "1.0", features = ["derive"] } +sha1 = { version = "0.10", default-features = false } thiserror = "2" uuid = { version = "1.18", default-features = false } wit-bindgen = { version = "0.57.1", default-features = false, features = ["macros", "std", "realloc", "bitflags"] } diff --git a/crates/workflow/src/event_groups.rs b/crates/workflow/src/event_groups.rs new file mode 100644 index 000000000..355ddcee3 --- /dev/null +++ b/crates/workflow/src/event_groups.rs @@ -0,0 +1,309 @@ +//! User-facing Event Groups and conversion to protocol markers. + +use sha1::{Digest, Sha1}; +use std::collections::HashMap; +use temporalio_common_wasm::{ + data_converters::{ + GenericPayloadConverter, PayloadConverter, SerializationContext, SerializationContextData, + }, + protos::temporal::api::{ + common::v1::Payload, + sdk::v1::{ + EventGroupMarker, + event_group_marker::{InboundEvent, InboundUpdate, Label, Variant}, + }, + }, +}; + +/// A token that associates workflow commands (and the history events they produce) with a logical +/// group for UI and observability. +/// +/// Attach a group to specific commands via `event_groups` on command options, or to every command +/// produced through a derived context via [`crate::WorkflowContext::with_event_group`]. +/// +/// # Experimental +/// +/// Event Groups is an experimental API and may change without notice. +#[derive(Clone, Debug)] +pub struct EventGroup { + inner: EventGroupInner, +} + +#[derive(Clone, Debug)] +enum EventGroupInner { + Label { id: String, label: String }, + InboundEvent { event_id: i64 }, + InboundUpdate { update_id: String }, +} + +#[derive(Clone, Debug, Eq, PartialEq, Hash)] +enum EventGroupKey { + Label(String), + InboundEvent(i64), + InboundUpdate(String), +} + +/// Ambient Event Groups carried by a workflow context. +/// +/// Implicit groups (signal / update handlers) replace any enclosing explicit scope. Explicit groups +/// nest and compose. +#[derive(Clone, Debug, Default)] +pub(crate) struct ActiveEventGroups { + implicit: Option, + explicit: Vec, +} + +impl EventGroup { + /// Create an Event Group with an explicit opaque identifier. + /// + /// Events are grouped together if and only if their groups have the same `id`, regardless of + /// label. Only the first label seen for a given `id` is used. + /// + /// The identifier is not payload-codec encoded. + /// + /// # Panics + /// + /// Panics if `label` or `id` is empty. + pub fn with_id(label: impl Into, id: impl Into) -> Self { + let label = label.into(); + let id = id.into(); + assert!(!label.is_empty(), "Event group label must not be empty"); + assert!(!id.is_empty(), "Event group id must not be empty"); + Self { + inner: EventGroupInner::Label { id, label }, + } + } + + pub(crate) fn derived(label: impl Into, original_execution_run_id: &str) -> Self { + let label = label.into(); + assert!(!label.is_empty(), "Event group label must not be empty"); + let id = derived_event_group_id(original_execution_run_id, &label); + Self { + inner: EventGroupInner::Label { id, label }, + } + } + + pub(crate) fn inbound_event(event_id: i64) -> Option { + (event_id > 0).then_some(Self { + inner: EventGroupInner::InboundEvent { event_id }, + }) + } + + pub(crate) fn inbound_update(update_id: impl Into) -> Self { + Self { + inner: EventGroupInner::InboundUpdate { + update_id: update_id.into(), + }, + } + } + + fn key(&self) -> EventGroupKey { + match &self.inner { + EventGroupInner::Label { id, .. } => EventGroupKey::Label(id.clone()), + EventGroupInner::InboundEvent { event_id } => EventGroupKey::InboundEvent(*event_id), + EventGroupInner::InboundUpdate { update_id } => { + EventGroupKey::InboundUpdate(update_id.clone()) + } + } + } + + pub(crate) fn to_marker(&self) -> EventGroupMarker { + EventGroupMarker { + variant: Some(match &self.inner { + EventGroupInner::Label { id, label } => Variant::Label(Label { + id: id.clone(), + label: Some(label_payload(label)), + }), + EventGroupInner::InboundEvent { event_id } => Variant::InboundEvent(InboundEvent { + inbound_event_id: *event_id, + }), + EventGroupInner::InboundUpdate { update_id } => { + Variant::InboundUpdate(InboundUpdate { + inbound_update_id: update_id.clone(), + }) + } + }), + } + } + + pub(crate) fn to_markers(groups: impl IntoIterator) -> Vec { + groups.into_iter().map(|group| group.to_marker()).collect() + } +} + +impl ActiveEventGroups { + pub(crate) fn with_explicit(&self, groups: impl IntoIterator) -> Self { + let mut explicit = self.explicit.clone(); + for group in groups { + upsert_group(&mut explicit, group); + } + Self { + implicit: self.implicit.clone(), + explicit, + } + } + + pub(crate) fn with_implicit(&self, implicit: EventGroup) -> Self { + Self { + implicit: Some(implicit), + explicit: Vec::new(), + } + } + + fn to_markers(&self) -> Vec { + let mut groups = Vec::new(); + if let Some(implicit) = &self.implicit { + groups.push(implicit.clone()); + } + groups.extend(self.explicit.iter().cloned()); + EventGroup::to_markers(groups) + } +} + +/// Merge ambient context groups with markers already attached to a command (direct options). +/// Direct markers overwrite ambient ones that share an identity. +pub(crate) fn merge_command_markers( + ambient: &ActiveEventGroups, + extra: Vec, +) -> Vec { + let mut by_key = HashMap::new(); + let mut order = Vec::new(); + for marker in ambient + .to_markers() + .into_iter() + .chain(extra) + .filter(|marker| marker.variant.is_some()) + { + if let Some(key) = marker_key(&marker) + && by_key.insert(key.clone(), marker).is_none() + { + order.push(key); + } + } + order + .into_iter() + .filter_map(|key| by_key.remove(&key)) + .collect() +} + +pub(crate) fn derived_event_group_id(original_execution_run_id: &str, label: &str) -> String { + let mut hasher = Sha1::new(); + hasher.update(original_execution_run_id.as_bytes()); + hasher.update(label.as_bytes()); + format!("{:x}", hasher.finalize()) +} + +fn label_payload(label: &str) -> Payload { + let converter = PayloadConverter::default(); + let context = SerializationContext { + data: &SerializationContextData::Workflow, + converter: &converter, + }; + let label = label.to_owned(); + converter + .to_payload(&context, &label) + .expect("encoding an Event Group label as json/plain is infallible") +} + +fn upsert_group(groups: &mut Vec, group: EventGroup) { + if let Some(existing) = groups + .iter_mut() + .find(|existing| existing.key() == group.key()) + { + *existing = group; + } else { + groups.push(group); + } +} + +fn marker_key(marker: &EventGroupMarker) -> Option { + match marker.variant.as_ref()? { + Variant::Label(label) => Some(EventGroupKey::Label(label.id.clone())), + Variant::InboundEvent(event) => Some(EventGroupKey::InboundEvent(event.inbound_event_id)), + Variant::InboundUpdate(update) => Some(EventGroupKey::InboundUpdate( + update.inbound_update_id.clone(), + )), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn derived_id_matches_sha1_formula() { + let id = derived_event_group_id("run-id", "aaa"); + let mut hasher = Sha1::new(); + hasher.update(b"run-id"); + hasher.update(b"aaa"); + assert_eq!(id, format!("{:x}", hasher.finalize())); + assert_ne!( + derived_event_group_id("run-id", "aaa"), + derived_event_group_id("run-id", "bbb") + ); + assert_ne!( + derived_event_group_id("run-1", "aaa"), + derived_event_group_id("run-2", "aaa") + ); + } + + #[test] + fn same_label_same_run_derives_the_same_id() { + let a = EventGroup::derived("bbb", "run"); + let b = EventGroup::derived("bbb", "run"); + assert_eq!(a.key(), b.key()); + } + + #[test] + fn user_provided_id_is_used_verbatim() { + let group = EventGroup::with_id("ccc", "c-id"); + match group.to_marker().variant { + Some(Variant::Label(label)) => { + assert_eq!(label.id, "c-id"); + assert_eq!( + label + .label + .as_ref() + .unwrap() + .metadata + .get("encoding") + .unwrap(), + b"json/plain" + ); + assert_eq!(label.label.as_ref().unwrap().data, b"\"ccc\""); + } + other => panic!("expected label marker, got {other:?}"), + } + } + + #[test] + fn merge_collapses_duplicate_ids_and_keeps_direct_label() { + let ambient = ActiveEventGroups::default().with_explicit([ + EventGroup::with_id("aaa", "a-id"), + EventGroup::with_id("bbb", "b-id"), + ]); + let extra = vec![EventGroup::with_id("aaa-direct", "a-id").to_marker()]; + let merged = merge_command_markers(&ambient, extra); + assert_eq!(merged.len(), 2); + let a = merged + .iter() + .find_map(|marker| match marker.variant.as_ref()? { + Variant::Label(label) if label.id == "a-id" => Some(label), + _ => None, + }) + .unwrap(); + assert_eq!(a.label.as_ref().unwrap().data, b"\"aaa-direct\""); + } + + #[test] + #[should_panic(expected = "Event group label must not be empty")] + fn empty_label_panics() { + let _ = EventGroup::with_id("", "id"); + } + + #[test] + #[should_panic(expected = "Event group id must not be empty")] + fn empty_id_panics() { + let _ = EventGroup::with_id("label", ""); + } +} diff --git a/crates/workflow/src/lib.rs b/crates/workflow/src/lib.rs index 2949aebf3..3f6c35848 100644 --- a/crates/workflow/src/lib.rs +++ b/crates/workflow/src/lib.rs @@ -17,6 +17,7 @@ pub mod __private { mod cancellation; #[doc(hidden)] pub mod component; +mod event_groups; #[doc(hidden)] pub mod runtime; mod workflow_context; @@ -24,6 +25,7 @@ pub mod workflow_interceptors; pub mod workflows; pub use cancellation::{WorkflowCancellationError, WorkflowCancellationToken}; +pub use event_groups::EventGroup; #[doc(hidden)] pub use runtime::model::{CancellableID, UnblockEvent}; pub use runtime::model::{TimerResult, WorkflowResult, WorkflowTermination}; diff --git a/crates/workflow/src/runtime/instance.rs b/crates/workflow/src/runtime/instance.rs index 205da6630..1ba28d6ad 100644 --- a/crates/workflow/src/runtime/instance.rs +++ b/crates/workflow/src/runtime/instance.rs @@ -541,6 +541,7 @@ where &mut self, signal: SignalWorkflow, ) -> Result { + let originating_event_id = signal.originating_event_id; let name = signal.signal_name; let payloads = Payloads { payloads: signal.input, @@ -551,8 +552,12 @@ where let input = HandleSignalInput::new(name.clone(), input, signal.headers); let handler_execution = self.base_ctx.track_handler(); let mut future = intercepted_signal_future::( - self.ctx.clone(), - self.base_ctx.clone(), + self.ctx + .clone() + .with_implicit_inbound_event(originating_event_id), + self.base_ctx + .clone() + .with_implicit_inbound_event(originating_event_id), self.interceptors.clone(), input, handler_execution, @@ -658,8 +663,10 @@ where let handler_execution = handler_execution.unwrap_or_else(|| self.base_ctx.track_handler()); let mut future = intercepted_update_future::( - self.ctx.clone(), - self.base_ctx.clone(), + self.ctx.clone().with_implicit_inbound_update(id.clone()), + self.base_ctx + .clone() + .with_implicit_inbound_update(id.clone()), self.interceptors.clone(), input, handler_execution, diff --git a/crates/workflow/src/workflow_context.rs b/crates/workflow/src/workflow_context.rs index 85e1d33fc..befa89f4d 100644 --- a/crates/workflow/src/workflow_context.rs +++ b/crates/workflow/src/workflow_context.rs @@ -12,7 +12,8 @@ pub use temporalio_common_wasm::protos::coresdk::child_workflow::StartChildWorkf pub use view::{NamespacedWorkflowInfo, WorkflowContextView}; use crate::{ - MemoValue, WorkflowCancellationError, WorkflowCancellationToken, + EventGroup, MemoValue, WorkflowCancellationError, WorkflowCancellationToken, + event_groups::{ActiveEventGroups, merge_command_markers}, runtime::{ SdkGuardedFuture, SdkWakeGuard, entry::WorkflowImplementation, @@ -87,7 +88,7 @@ use temporalio_common_wasm::{ ModifyWorkflowProperties, RequestCancelActivity, RequestCancelExternalWorkflowExecution, RequestCancelLocalActivity, RequestCancelNexusOperation, SetPatchMarker, UpsertWorkflowSearchAttributes, - signal_external_workflow_execution, workflow_command, + WorkflowCommand, signal_external_workflow_execution, workflow_command, }, }, temporal::api::{ @@ -145,6 +146,7 @@ impl_random_value!(u8, u16, u32, u64, u128, i8, i16, i32, i64, i128, f32, f64); #[derive(Clone)] pub struct BaseWorkflowContext { inner: Rc, + event_groups: ActiveEventGroups, } /// Input provided to a worker's patch activation callback. @@ -643,6 +645,76 @@ impl BaseWorkflowContext { current_waker: RefCell::new(None), workflow_interceptors, }), + event_groups: ActiveEventGroups::default(), + } + } + + fn push_user_command(&self, mut command: WorkflowCommand) { + command.event_group_markers = merge_command_markers( + &self.event_groups, + std::mem::take(&mut command.event_group_markers), + ); + self.inner.runtime.host.push_command(command); + } + + fn original_execution_run_id(&self) -> &str { + &self.inner.initial_information.original_execution_run_id + } + + /// Create an Event Group whose id is derived from this workflow's original execution run id + /// and `label`. + /// + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. + /// + /// # Panics + /// + /// Panics if `label` is empty. + pub fn create_event_group(&self, label: impl Into) -> EventGroup { + EventGroup::derived(label, self.original_execution_run_id()) + } + + /// Return a derived context that attaches `group` to every command issued through it. + /// + /// Nested derivations compose: commands carry the union of enclosing groups. Direct + /// `event_groups` on command options are added to this set. + /// + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. + pub fn with_event_group(&self, group: EventGroup) -> Self { + self.with_event_groups([group]) + } + + /// Return a derived context that attaches each of `groups` to every command issued through it. + /// + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. + pub fn with_event_groups(&self, groups: impl IntoIterator) -> Self { + Self { + inner: self.inner.clone(), + event_groups: self.event_groups.with_explicit(groups), + } + } + + pub(crate) fn with_implicit_inbound_event(&self, event_id: i64) -> Self { + match EventGroup::inbound_event(event_id) { + Some(group) => Self { + inner: self.inner.clone(), + event_groups: self.event_groups.with_implicit(group), + }, + None => self.clone(), + } + } + + pub(crate) fn with_implicit_inbound_update(&self, update_id: impl Into) -> Self { + Self { + inner: self.inner.clone(), + event_groups: self + .event_groups + .with_implicit(EventGroup::inbound_update(update_id)), } } @@ -807,11 +879,7 @@ impl BaseWorkflowContext { .inner .runtime .register_unblocker(PendingCommandId::Timer(seq), unblocker); - base_ctx - .inner - .runtime - .host - .push_command(opts.into_command(seq)); + base_ctx.push_user_command(opts.into_command(seq)); CancellableWorkflowOutboundFuture::new( cmd, base_ctx.cancellation_handle(CancellableID::Timer(seq)), @@ -880,7 +948,7 @@ impl BaseWorkflowContext { if opts.task_queue.is_none() { opts.task_queue = Some(base_ctx.inner.task_queue.clone()); } - base_ctx.inner.runtime.host.push_command(opts.into_command( + base_ctx.push_user_command(opts.into_command( seq, activity_type, payloads, @@ -1082,7 +1150,7 @@ impl BaseWorkflowContext { PendingCommandId::ChildWorkflowComplete(child_seq), unblocker, ); - base_ctx.inner.runtime.host.push_command(opts.into_command( + base_ctx.push_user_command(opts.into_command( child_seq, workflow_type, payloads, @@ -1149,12 +1217,7 @@ impl BaseWorkflowContext { self.inner .runtime .register_unblocker(PendingCommandId::Activity(seq), unblocker); - self.inner.runtime.host.push_command(opts.into_command( - seq, - activity_type, - arguments, - headers, - )); + self.push_user_command(opts.into_command(seq, activity_type, arguments, headers)); cmd } @@ -1234,11 +1297,13 @@ impl BaseWorkflowContext { .inner .runtime .register_unblocker(PendingCommandId::SignalExternal(seq), unblocker); - base_ctx - .inner - .runtime - .host - .push_command(options.into_command(seq, signal_name, payloads, headers, target)); + base_ctx.push_user_command(options.into_command( + seq, + signal_name, + payloads, + headers, + target, + )); cancellable_outbound(SignalChildFut::Running { inner: cmd, data_converter: base_ctx.data_converter().clone(), @@ -1329,11 +1394,7 @@ impl BaseWorkflowContext { .inner .runtime .register_unblocker(PendingCommandId::NexusOpComplete(seq), unblocker); - base_ctx - .inner - .runtime - .host - .push_command(opts.into_command(seq)); + base_ctx.push_user_command(opts.into_command(seq)); let result_future = CancellableWorkflowOutboundFuture::new( result_future, base_ctx.cancellation_handle(CancellableID::NexusOp(seq)), @@ -1521,6 +1582,45 @@ impl SyncWorkflowContext { .fuse() } + /// Create an Event Group whose id is derived from this workflow's original execution run id + /// and `label`. + /// + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. + /// + /// # Panics + /// + /// Panics if `label` is empty. + pub fn create_event_group(&self, label: impl Into) -> EventGroup { + self.base.create_event_group(label) + } + + /// Return a derived context that attaches `group` to every command issued through it. + /// + /// Nested derivations compose: commands carry the union of enclosing groups. Direct + /// `event_groups` on command options are added to this set. + /// + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. + pub fn with_event_group(&self, group: EventGroup) -> Self { + self.with_event_groups([group]) + } + + /// Return a derived context that attaches each of `groups` to every command issued through it. + /// + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. + pub fn with_event_groups(&self, groups: impl IntoIterator) -> Self { + Self { + base: self.base.with_event_groups(groups), + headers: self.headers.clone(), + _phantom: PhantomData, + } + } + /// Signal that this workflow should continue as a new workflow execution with the given input and /// options. /// @@ -1698,7 +1798,7 @@ impl SyncWorkflowContext { }; if res { - self.base.inner.runtime.host.push_command( + self.base.push_user_command( workflow_command::Variant::SetPatchMarker(SetPatchMarker { patch_id: patch_id.to_string(), deprecated, @@ -1751,7 +1851,7 @@ impl SyncWorkflowContext { } let proto = SearchAttributes::updates_to_proto(updates); - self.base.inner.runtime.host.push_command( + self.base.push_user_command( workflow_command::Variant::UpsertWorkflowSearchAttributes( UpsertWorkflowSearchAttributes { search_attributes: Some(proto), @@ -1804,7 +1904,7 @@ impl SyncWorkflowContext { } } } - self.base.inner.runtime.host.push_command( + self.base.push_user_command( workflow_command::Variant::ModifyWorkflowProperties(ModifyWorkflowProperties { upserted_memo: Some(ProtoMemo { fields }), }) @@ -1867,6 +1967,67 @@ impl WorkflowContext { } } + /// Create an Event Group whose id is derived from this workflow's original execution run id + /// and `label`. + /// + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. + /// + /// # Panics + /// + /// Panics if `label` is empty. + pub fn create_event_group(&self, label: impl Into) -> EventGroup { + self.sync.create_event_group(label) + } + + /// Return a derived context that attaches `group` to every command issued through it. + /// + /// Nested derivations compose: commands carry the union of enclosing groups. Direct + /// `event_groups` on command options are added to this set. Clone this derived context into + /// concurrent futures so they keep the attached groups. + /// + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. + pub fn with_event_group(&self, group: EventGroup) -> Self { + self.with_event_groups([group]) + } + + /// Return a derived context that attaches each of `groups` to every command issued through it. + /// + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. + pub fn with_event_groups(&self, groups: impl IntoIterator) -> Self { + Self { + sync: self.sync.with_event_groups(groups), + workflow_state: self.workflow_state.clone(), + } + } + + pub(crate) fn with_implicit_inbound_event(&self, event_id: i64) -> Self { + Self { + sync: SyncWorkflowContext { + base: self.sync.base.with_implicit_inbound_event(event_id), + headers: self.sync.headers.clone(), + _phantom: PhantomData, + }, + workflow_state: self.workflow_state.clone(), + } + } + + pub(crate) fn with_implicit_inbound_update(&self, update_id: impl Into) -> Self { + Self { + sync: SyncWorkflowContext { + base: self.sync.base.with_implicit_inbound_update(update_id), + headers: self.sync.headers.clone(), + _phantom: PhantomData, + }, + workflow_state: self.workflow_state.clone(), + } + } + /// Returns a [`SyncWorkflowContext`] extracted from this context. pub(crate) fn sync_context(&self) -> SyncWorkflowContext { self.sync.clone() @@ -2594,7 +2755,7 @@ impl Future for LATimerBackoffFut { .expect("duration converts ok"), cancellation_token: Some(self.cancellation_token.clone()), summary: None, - event_group_markers: self.la_opts.event_group_markers.clone(), + event_groups: self.la_opts.event_groups.clone(), }); self.timer_fut = Some(Box::pin(timer_f)); self.next_attempt = b.attempt; @@ -3290,7 +3451,6 @@ mod tests { temporal::api::{ common::v1::{Payload, RetryPolicy as ProtoRetryPolicy}, enums::v1::ContinueAsNewVersioningBehavior as ProtoContinueAsNewVersioningBehavior, - sdk::v1::{EventGroupMarker, event_group_marker}, }, }, }; @@ -3520,7 +3680,7 @@ mod tests { duration: Duration::from_secs(1), cancellation_token: Some(token.clone()), summary: None, - event_group_markers: vec![], + event_groups: vec![], }); let mut activity_options = ActivityOptions::start_to_close_timeout(Duration::from_secs(1)); @@ -3735,17 +3895,10 @@ mod tests { Vec::new(), ); let token = WorkflowCancellationToken::new(); - let marker = EventGroupMarker { - variant: Some(event_group_marker::Variant::Label( - event_group_marker::Label { - id: "la-group".to_string(), - label: Some("la-group".as_json_payload().unwrap()), - }, - )), - }; + let group = EventGroup::with_id("la-group", "la-group"); let mut options = LocalActivityOptions { schedule_to_close_timeout: Some(Duration::from_secs(10)), - event_group_markers: vec![marker.clone()], + event_groups: vec![group.clone()], ..Default::default() }; options.cancellation_token = Some(token.clone()); @@ -3783,7 +3936,7 @@ mod tests { ) }) .expect("backoff StartTimer is issued"); - assert_eq!(start_timer.event_group_markers, [marker]); + assert_eq!(start_timer.event_group_markers, [group.to_marker()]); } #[test] diff --git a/crates/workflow/src/workflow_context/options.rs b/crates/workflow/src/workflow_context/options.rs index 16a50722f..dd1eb2dc2 100644 --- a/crates/workflow/src/workflow_context/options.rs +++ b/crates/workflow/src/workflow_context/options.rs @@ -1,6 +1,8 @@ use std::{collections::HashMap, time::Duration}; -use crate::{MemoValues, WorkflowCancellationToken, runtime::types::ContinueAsNewRequest}; +use crate::{ + EventGroup, MemoValues, WorkflowCancellationToken, runtime::types::ContinueAsNewRequest, +}; use temporalio_common_wasm::{ ActivityCloseTimeouts, Priority, RetryPolicy, data_converters::{ @@ -323,13 +325,14 @@ pub struct ActivityOptions { /// If true, disable eager execution for this activity #[builder(default)] pub do_not_eagerly_execute: bool, - /// Event group markers to attach to the resulting `ScheduleActivityTask` command. + /// Event Groups to attach to the resulting schedule-activity command, in addition to any + /// groups from the enclosing Event Group scope. + /// + /// # Experimental /// - /// **Unstable:** Event Groups are not yet implemented in the Rust SDK; this field exists - /// only for internal test purposes. This API *will* change. - #[doc(hidden)] + /// Event Groups is an experimental API and may change without notice. #[builder(default)] - pub event_group_markers: Vec, + pub event_groups: Vec, } impl ActivityOptions { @@ -397,7 +400,7 @@ impl ActivityOptions { }), self.summary, None, - self.event_group_markers, + EventGroup::to_markers(self.event_groups), ) } } @@ -445,13 +448,14 @@ pub struct LocalActivityOptions { pub start_to_close_timeout: Option, /// Single-line summary for this activity that will appear in UI/CLI. pub summary: Option, - /// Event group markers to attach to the resulting `RecordMarker` command. + /// Event Groups to attach to the resulting local-activity command, in addition to any groups + /// from the enclosing Event Group scope. /// - /// **Unstable:** Event Groups are not yet implemented in the Rust SDK; this field exists - /// only for internal test purposes. This API *will* change. - #[doc(hidden)] + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. #[builder(default)] - pub event_group_markers: Vec, + pub event_groups: Vec, } impl Default for LocalActivityOptions { @@ -498,7 +502,7 @@ impl LocalActivityOptions { }), self.summary, None, - self.event_group_markers, + EventGroup::to_markers(self.event_groups), ) } } @@ -540,13 +544,14 @@ pub struct ChildWorkflowOptions { pub search_attributes: Option, /// Priority for the workflow pub priority: Option, - /// Event group markers to attach to the resulting `StartChildWorkflowExecution` command. + /// Event Groups to attach to the resulting start-child-workflow command, in addition to any + /// groups from the enclosing Event Group scope. + /// + /// # Experimental /// - /// **Unstable:** Event Groups are not yet implemented in the Rust SDK; this field exists - /// only for internal test purposes. This API *will* change. - #[doc(hidden)] + /// Event Groups is an experimental API and may change without notice. #[builder(default)] - pub event_group_markers: Vec, + pub event_groups: Vec, } impl ChildWorkflowOptions { @@ -599,7 +604,7 @@ impl ChildWorkflowOptions { }), self.static_summary, self.static_details, - self.event_group_markers, + EventGroup::to_markers(self.event_groups), ) } } @@ -615,13 +620,14 @@ pub struct TimerOptions { pub cancellation_token: Option, /// Summary of the timer pub summary: Option, - /// Event group markers to attach to the resulting `StartTimer` command. + /// Event Groups to attach to the resulting start-timer command, in addition to any groups from + /// the enclosing Event Group scope. /// - /// **Unstable:** Event Groups are not yet implemented in the Rust SDK; this field exists - /// only for internal test purposes. This API *will* change. - #[doc(hidden)] + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. #[builder(default)] - pub event_group_markers: Vec, + pub event_groups: Vec, } impl Default for TimerOptions { @@ -652,7 +658,7 @@ impl TimerOptions { }), self.summary, None, - self.event_group_markers, + EventGroup::to_markers(self.event_groups), ) } } @@ -673,13 +679,14 @@ pub struct SignalWorkflowOptions { pub cancellation_token: Option, /// Single-line summary for this signal that will appear in UI/CLI. pub summary: Option, - /// Event group markers to attach to the resulting `SignalExternalWorkflowExecution` command. + /// Event Groups to attach to the resulting signal-external-workflow command, in addition to any + /// groups from the enclosing Event Group scope. + /// + /// # Experimental /// - /// **Unstable:** Event Groups are not yet implemented in the Rust SDK; this field exists - /// only for internal test purposes. This API *will* change. - #[doc(hidden)] + /// Event Groups is an experimental API and may change without notice. #[builder(default)] - pub event_group_markers: Vec, + pub event_groups: Vec, } impl SignalWorkflowOptions { @@ -703,7 +710,7 @@ impl SignalWorkflowOptions { ), self.summary, None, - self.event_group_markers, + EventGroup::to_markers(self.event_groups), ) } } @@ -753,33 +760,45 @@ pub struct NexusOperationOptions { /// Only applies to asynchronous operations. Synchronous operations ignore this timeout. /// If not set or zero, no start-to-close timeout is enforced. pub start_to_close_timeout: Option, + /// Event Groups to attach to the resulting schedule-nexus-operation command, in addition to + /// any groups from the enclosing Event Group scope. + /// + /// # Experimental + /// + /// Event Groups is an experimental API and may change without notice. + #[builder(default)] + pub event_groups: Vec, } impl NexusOperationOptions { pub(crate) fn into_command(self, seq: u32) -> WorkflowCommand { - workflow_command::Variant::ScheduleNexusOperation(ScheduleNexusOperation { - seq, - endpoint: self.endpoint, - service: self.service, - operation: self.operation, - input: self.input, - schedule_to_close_timeout: self - .schedule_to_close_timeout - .and_then(|duration| duration.try_into().ok()), - schedule_to_start_timeout: self - .schedule_to_start_timeout - .and_then(|duration| duration.try_into().ok()), - start_to_close_timeout: self - .start_to_close_timeout - .and_then(|duration| duration.try_into().ok()), - nexus_header: self.nexus_header, - cancellation_type: ProtoNexusOperationCancellationType::from( - self.cancellation_type - .unwrap_or(NexusOperationCancellationType::WaitCancellationCompleted), - ) - .into(), - }) - .into() + command_with_metadata( + workflow_command::Variant::ScheduleNexusOperation(ScheduleNexusOperation { + seq, + endpoint: self.endpoint, + service: self.service, + operation: self.operation, + input: self.input, + schedule_to_close_timeout: self + .schedule_to_close_timeout + .and_then(|duration| duration.try_into().ok()), + schedule_to_start_timeout: self + .schedule_to_start_timeout + .and_then(|duration| duration.try_into().ok()), + start_to_close_timeout: self + .start_to_close_timeout + .and_then(|duration| duration.try_into().ok()), + nexus_header: self.nexus_header, + cancellation_type: ProtoNexusOperationCancellationType::from( + self.cancellation_type + .unwrap_or(NexusOperationCancellationType::WaitCancellationCompleted), + ) + .into(), + }), + None, + None, + EventGroup::to_markers(self.event_groups), + ) } }