diff --git a/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto b/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto index 1dba23214..931f75c74 100644 --- a/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto +++ b/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto @@ -55,9 +55,31 @@ message WorkflowCommand { ExternalStreamFinalized external_stream_finalized = 26; WorkflowOutputStreamCommit workflow_output_stream_commit = 27; WorkflowOutputStreamBuffered workflow_output_stream_buffered = 28; + SubscribeNotificationChannel subscribe_notification_channel = 29; + UnsubscribeNotificationChannel unsubscribe_notification_channel = 30; } } +// End this workflow's subscription to a notification channel. Notifications +// already recorded on a scheduled event still reach that Workflow Task. +// +// The server records the event whether or not the run held a subscription, so +// every command has an event to match on replay. +message UnsubscribeNotificationChannel { + // Name of the channel, scoped to the namespace. + string channel = 1; +} + +// Subscribe this workflow to a notification channel, so the scheduled event of +// each later Workflow Task carries the notifications folded for it. +// +// The notifications live in History rather than arriving by a side channel, so +// a replay reads the same ones the live run saw. +message SubscribeNotificationChannel { + // Name of the channel, scoped to the namespace. + string channel = 1; +} + // Commits an observation delta for external streams. // // Emitted on *every* completion path where replay-visible stream state changed -- which includes diff --git a/crates/protos/src/protos/mod.rs b/crates/protos/src/protos/mod.rs index 090fb59e8..4281f6631 100644 --- a/crates/protos/src/protos/mod.rs +++ b/crates/protos/src/protos/mod.rs @@ -1805,6 +1805,18 @@ pub mod coresdk { } } + impl Display for SubscribeNotificationChannel { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "SubscribeNotificationChannel({})", self.channel) + } + } + + impl Display for UnsubscribeNotificationChannel { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "UnsubscribeNotificationChannel({})", self.channel) + } + } + impl Display for WorkflowOutputStreamBuffered { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { write!( @@ -2056,6 +2068,26 @@ pub mod temporal { } } + impl From for Attributes { + fn from(s: workflow_commands::SubscribeNotificationChannel) -> Self { + Self::SubscribeNotificationChannelCommandAttributes( + SubscribeNotificationChannelCommandAttributes { + channel: s.channel, + }, + ) + } + } + + impl From for Attributes { + fn from(u: workflow_commands::UnsubscribeNotificationChannel) -> Self { + Self::UnsubscribeNotificationChannelCommandAttributes( + UnsubscribeNotificationChannelCommandAttributes { + channel: u.channel, + }, + ) + } + } + impl From for command::Attributes { fn from(s: workflow_commands::StartTimer) -> Self { Self::StartTimerCommandAttributes(StartTimerCommandAttributes { diff --git a/crates/sdk-core/src/core_tests/external_streams.rs b/crates/sdk-core/src/core_tests/external_streams.rs index 61cc60a87..e2ed62ca4 100644 --- a/crates/sdk-core/src/core_tests/external_streams.rs +++ b/crates/sdk-core/src/core_tests/external_streams.rs @@ -51,6 +51,7 @@ use temporalio_common::{ command::v1::command, common::v1::Payload, enums::v1::{CommandType, EventType}, + notification::v1::Notification, query::v1::WorkflowQuery, workflowservice::v1::RespondWorkflowTaskCompletedResponse, }, @@ -3297,6 +3298,169 @@ async fn an_ordinary_signal_still_reaches_its_handler() { worker.drain_pollers_and_shutdown().await; } +// --- notifications on the scheduled event ------------------------------------ + +fn channel_notification(channel: &str, counter: i64) -> Notification { + Notification { + channel: channel.to_string(), + position: format!("{counter}-0").into_bytes(), + counter, + metadata: HashMap::new(), + ..Default::default() + } +} + +/// The notifications of every `NotificationsReceived` job in the activation, one entry per job. +fn notification_jobs(activation: &WorkflowActivation) -> Vec> { + activation + .jobs + .iter() + .filter_map(|j| match &j.variant { + Some(workflow_activation_job::Variant::NotificationsReceived(n)) => { + Some(n.notifications.clone()) + } + _ => None, + }) + .collect() +} + +/// A history whose second task fails after being scheduled with `n1` and is retried with `n2`. +fn failed_then_retried_history(n1: Vec, n2: Vec) -> TestHistoryBuilder { + let mut t = TestHistoryBuilder::default(); + t.add_wfe_started_with_wft_timeout(Duration::from_secs(300)); + t.add_full_wf_task(); + t.add_workflow_task_scheduled_with_notifications(n1); + t.add_workflow_task_started(); + t.add_workflow_task_failed_with_failure( + temporalio_common::protos::temporal::api::enums::v1::WorkflowTaskFailedCause::Unspecified, + Default::default(), + ); + t.add_workflow_task_scheduled_with_notifications(n2); + t.add_workflow_task_started(); + t +} + +#[rstest::rstest] +#[case::live(false)] +#[case::replay(true)] +#[tokio::test] +async fn a_failed_task_hands_its_notifications_to_the_retry(#[case] cold: bool) { + // The server clears a listener's pending notifications when it puts them on a scheduled + // event, so a retry's event carries only what arrived after the failed attempt. The failed + // attempt's event stays in History, and its notifications reach lang on the retry, folded + // with the retry's own into one job: one per channel, the highest counter kept. + let n1 = vec![channel_notification("tokens", 41)]; + let n2 = vec![ + channel_notification("tokens", 42), + channel_notification("events", 3), + ]; + let t = failed_then_retried_history(n1, n2.clone()); + let batches = if cold { + let resp = hist_to_poll_resp(&t, "fakeid".to_owned(), ResponseType::AllHistory); + vec![ResponseType::Raw(resp.resp)] + } else { + vec![ResponseType::ToTaskNum(1), ResponseType::ToTaskNum(2)] + }; + let mut mock = build_mock_pollers(MockPollCfg::from_resp_batches( + "fakeid", + t, + batches, + mock_worker_client(), + )); + mock.worker_cfg(|w| { + w.task_types = WorkerTaskTypes::workflow_only(); + w.max_cached_workflows = 1; + }); + let worker = mock_worker(mock); + let first = worker.poll_workflow_activation().await.unwrap(); + assert_eq!(first.is_replaying, cold); + assert_eq!( + notification_jobs(&first), + Vec::>::new(), + "the notifications belong to the retried task, got {:?}", + first.jobs + ); + worker + .complete_workflow_activation(WorkflowActivationCompletion::empty(first.run_id)) + .await + .unwrap(); + + let retried = worker.poll_workflow_activation().await.unwrap(); + assert!(!retried.is_replaying, "got jobs {:?}", retried.jobs); + assert_eq!(notification_jobs(&retried), vec![n2]); + worker + .complete_workflow_activation(WorkflowActivationCompletion::from_cmds( + retried.run_id, + vec![CompleteWorkflowExecution::default().into()], + )) + .await + .unwrap(); + worker.drain_pollers_and_shutdown().await; +} + +#[tokio::test] +async fn replaying_a_notified_scheduled_event_yields_the_same_job() { + // History is the record, so a cold worker hands lang the same job in the same activation the + // live run saw it in. The first task issued no command, which is the shape Core would + // otherwise treat as a heartbeat and fold into the next task, moving the job one activation + // early. The Signal keeps the replayed task apart from the live one. + let notifications = vec![channel_notification("tokens", 42)]; + let mut t = TestHistoryBuilder::default(); + t.add_wfe_started_with_wft_timeout(Duration::from_secs(300)); + t.add_full_wf_task(); + t.add_workflow_task_scheduled_with_notifications(notifications.clone()); + t.add_workflow_task_started(); + t.add_workflow_task_completed(); + t.add_we_signaled("a-user-signal", vec![]); + t.add_workflow_task_scheduled_and_started(); + + let cold = hist_to_poll_resp(&t, "fakeid".to_owned(), ResponseType::AllHistory); + let mut mock = build_mock_pollers(MockPollCfg::from_resp_batches( + "fakeid", + t, + [ResponseType::Raw(cold.resp)], + mock_worker_client(), + )); + mock.worker_cfg(|w| { + w.task_types = WorkerTaskTypes::workflow_only(); + w.max_cached_workflows = 1; + }); + let worker = mock_worker(mock); + + let first = worker.poll_workflow_activation().await.unwrap(); + assert!(first.is_replaying); + assert_eq!( + notification_jobs(&first), + Vec::>::new(), + "the notifications belong to the second task, got {:?}", + first.jobs + ); + worker + .complete_workflow_activation(WorkflowActivationCompletion::empty(first.run_id)) + .await + .unwrap(); + + let replayed = worker.poll_workflow_activation().await.unwrap(); + assert!(replayed.is_replaying, "got jobs {:?}", replayed.jobs); + assert_eq!(notification_jobs(&replayed), vec![notifications]); + worker + .complete_workflow_activation(WorkflowActivationCompletion::empty(replayed.run_id)) + .await + .unwrap(); + + let live = worker.poll_workflow_activation().await.unwrap(); + assert!(!live.is_replaying, "got jobs {:?}", live.jobs); + assert_eq!(notification_jobs(&live), Vec::>::new()); + worker + .complete_workflow_activation(WorkflowActivationCompletion::from_cmds( + live.run_id, + vec![CompleteWorkflowExecution::default().into()], + )) + .await + .unwrap(); + worker.drain_pollers_and_shutdown().await; +} + // --- marker emission (C9, C14b) ---------------------------------------------- /// A history for a Workflow Task that records a marker and then starts a timer. diff --git a/crates/sdk-core/src/replay/history_builder.rs b/crates/sdk-core/src/replay/history_builder.rs index 3cf593483..616eaf68d 100644 --- a/crates/sdk-core/src/replay/history_builder.rs +++ b/crates/sdk-core/src/replay/history_builder.rs @@ -114,6 +114,43 @@ impl TestHistoryBuilder { self.workflow_task_scheduled_event_id = self.add_by_type(EventType::WorkflowTaskScheduled); } + /// Add a workflow task scheduled event carrying channel notifications. + pub fn add_workflow_task_scheduled_with_notifications( + &mut self, + notifications: Vec< + temporalio_common::protos::temporal::api::notification::v1::Notification, + >, + ) { + self.workflow_task_scheduled_event_id = self.add(WorkflowTaskScheduledEventAttributes { + notifications, + ..Default::default() + }); + } + + /// Add the event a subscribe-notification-channel command produces. + pub fn add_notification_channel_subscribed(&mut self, channel: &str) -> i64 { + let attrs = WorkflowNotificationChannelSubscribedEventAttributes { + workflow_task_completed_event_id: self.previous_task_completed_id, + channel: channel.to_string(), + }; + self.add(attrs) + } + + /// Add the event an unsubscribe-notification-channel command produces. A zero + /// `subscribed_event_id` is what the server records when the run held no subscription. + pub fn add_notification_channel_unsubscribed( + &mut self, + channel: &str, + subscribed_event_id: i64, + ) -> i64 { + let attrs = WorkflowNotificationChannelUnsubscribedEventAttributes { + workflow_task_completed_event_id: self.previous_task_completed_id, + channel: channel.to_string(), + subscribed_event_id, + }; + self.add(attrs) + } + /// Add a workflow task started event. pub fn add_workflow_task_started(&mut self) { self.final_workflow_task_started_event_id = self.add(WorkflowTaskStartedEventAttributes { diff --git a/crates/sdk-core/src/worker/workflow/history_update.rs b/crates/sdk-core/src/worker/workflow/history_update.rs index 856589946..03325434f 100644 --- a/crates/sdk-core/src/worker/workflow/history_update.rs +++ b/crates/sdk-core/src/worker/workflow/history_update.rs @@ -741,10 +741,14 @@ fn find_end_index_of_next_wft_seq( if let Some(next_next_event) = events.get(ix + 2) { if !saw_command && next_next_event.event_type() == EventType::WorkflowTaskScheduled + && !scheduled_with_notifications(next_next_event) { // If we've never seen an interesting event and the next two events are // a completion followed immediately again by scheduled, then this is a - // WFT heartbeat and also doesn't conclude the sequence. + // WFT heartbeat and also doesn't conclude the sequence. A scheduled + // event with notifications is work lang saw in its own activation + // live, so folding it into this one on replay would hand lang the + // notifications one activation early. continue; } else { // If we see an update accepted command after WFT completed, we want to @@ -801,6 +805,15 @@ fn find_end_index_of_next_wft_seq( NextWFTSeqEndIndex::Incomplete(last_index) } +/// A task scheduled with notifications was handed them as an activation of its own, so it +/// cannot be folded into the heartbeat chain before it the way an empty task is. +fn scheduled_with_notifications(e: &HistoryEvent) -> bool { + matches!( + &e.attributes, + Some(Attributes::WorkflowTaskScheduledEventAttributes(a)) if !a.notifications.is_empty() + ) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/sdk-core/src/worker/workflow/machines/mod.rs b/crates/sdk-core/src/worker/workflow/machines/mod.rs index 53d44aac0..2633bc8fa 100644 --- a/crates/sdk-core/src/worker/workflow/machines/mod.rs +++ b/crates/sdk-core/src/worker/workflow/machines/mod.rs @@ -16,13 +16,19 @@ mod modify_workflow_properties_state_machine; mod nexus_operation_state_machine; mod patch_state_machine; mod signal_external_state_machine; +mod subscribe_notification_channel_state_machine; mod timer_state_machine; +mod unsubscribe_notification_channel_state_machine; mod update_state_machine; mod upsert_search_attributes_state_machine; mod workflow_task_state_machine; +#[cfg(test)] +mod subscribe_notification_channel_state_machine_tests; #[cfg(test)] mod transition_coverage; +#[cfg(test)] +mod unsubscribe_notification_channel_state_machine_tests; pub(crate) use temporalio_common::fsm_trait::MachineError; pub(crate) use workflow_machines::{MachinesWFTResponseContent, WorkflowMachines}; @@ -48,6 +54,7 @@ use std::{ convert::{TryFrom, TryInto}, fmt::{Debug, Display}, }; +use subscribe_notification_channel_state_machine::SubscribeNotificationChannelMachine; use temporalio_common::{ fsm_trait::{StateMachine, TransitionResult}, protos::temporal::api::{ @@ -57,6 +64,7 @@ use temporalio_common::{ }; use temporalio_macros::fsm; use timer_state_machine::TimerMachine; +use unsubscribe_notification_channel_state_machine::UnsubscribeNotificationChannelMachine; use update_state_machine::UpdateMachine; use upsert_search_attributes_state_machine::UpsertSearchAttributesMachine; use workflow_machines::MachineResponse; @@ -85,6 +93,8 @@ enum Machines { ModifyWorkflowPropertiesMachine, UpdateMachine, NexusOperationMachine, + SubscribeNotificationChannelMachine, + UnsubscribeNotificationChannelMachine, } /// Extends [rustfsm::StateMachine] with some functionality specific to the temporal SDK. diff --git a/crates/sdk-core/src/worker/workflow/machines/subscribe_notification_channel_state_machine.rs b/crates/sdk-core/src/worker/workflow/machines/subscribe_notification_channel_state_machine.rs new file mode 100644 index 000000000..a3b2b8794 --- /dev/null +++ b/crates/sdk-core/src/worker/workflow/machines/subscribe_notification_channel_state_machine.rs @@ -0,0 +1,143 @@ +use super::{ + NewMachineWithCommand, StateMachine, TransitionResult, fsm, workflow_machines::MachineResponse, +}; +use crate::worker::workflow::{ + WFMachinesError, fatal, + machines::{EventInfo, HistEventData, WFMachinesAdapter}, + nondeterminism, +}; +use temporalio_common::protos::{ + coresdk::workflow_commands::SubscribeNotificationChannel, + temporal::api::{ + enums::v1::{CommandType, EventType}, + history::v1::{WorkflowNotificationChannelSubscribedEventAttributes, history_event}, + }, +}; + +fsm! { + pub(super) name SubscribeNotificationChannelMachine; + command SubscribeNotificationChannelMachineCommand; + error WFMachinesError; + shared_state SharedState; + + Created --(CommandScheduled) --> CommandIssued; + CommandIssued --(CommandRecorded(WorkflowNotificationChannelSubscribedEventAttributes), + shared on_command_recorded) --> Done; +} + +/// The channel the command named, kept so the recorded event can be held against it. +#[derive(Default, Clone)] +pub(super) struct SharedState { + channel: String, +} + +/// Subscribe this workflow to a notification channel. The command carries only +/// the channel name; the notifications arrive later on the scheduled event of +/// each Workflow Task, so nothing is handed back here. +pub(super) fn subscribe_notification_channel( + lang_cmd: SubscribeNotificationChannel, +) -> NewMachineWithCommand { + let sm = SubscribeNotificationChannelMachine::from_parts( + Created {}.into(), + SharedState { + channel: lang_cmd.channel.clone(), + }, + ); + NewMachineWithCommand { + command: lang_cmd.into(), + machine: sm.into(), + } +} + +#[derive(Debug, derive_more::Display)] +pub(super) enum SubscribeNotificationChannelMachineCommand {} + +#[derive(Debug, Default, Clone, derive_more::Display)] +pub(super) struct Created {} + +#[derive(Debug, Default, Clone, derive_more::Display)] +pub(super) struct CommandIssued {} + +impl CommandIssued { + pub(super) fn on_command_recorded( + self, + dat: &mut SharedState, + attrs: WorkflowNotificationChannelSubscribedEventAttributes, + ) -> SubscribeNotificationChannelMachineTransition { + if dat.channel == attrs.channel { + TransitionResult::default() + } else { + TransitionResult::Err(nondeterminism!( + "Recorded subscription to notification channel {:?} does not match the \ + reissued subscription to channel {:?}", + attrs.channel, + dat.channel + )) + } + } +} + +#[derive(Debug, Default, Clone, derive_more::Display)] +pub(super) struct Done {} + +impl WFMachinesAdapter for SubscribeNotificationChannelMachine { + fn adapt_response( + &self, + _my_command: Self::Command, + _event_info: Option, + ) -> Result, Self::Error> { + Err(fatal!( + "SubscribeNotificationChannel does not use state machine commands" + )) + } +} + +impl TryFrom for SubscribeNotificationChannelMachineEvents { + type Error = WFMachinesError; + + fn try_from(e: HistEventData) -> Result { + let e = e.event; + match e.event_type() { + EventType::WorkflowNotificationChannelSubscribed => { + if let Some( + history_event::Attributes::WorkflowNotificationChannelSubscribedEventAttributes( + attrs, + ), + ) = e.attributes + { + Ok(SubscribeNotificationChannelMachineEvents::CommandRecorded( + attrs, + )) + } else { + Err(fatal!( + "Notification channel subscribed attributes were unset: {e}" + )) + } + } + _ => Err(Self::Error::Nondeterminism(format!( + "SubscribeNotificationChannelMachine does not handle {e}" + ))), + } + } +} + +impl TryFrom for SubscribeNotificationChannelMachineEvents { + type Error = WFMachinesError; + + fn try_from(c: CommandType) -> Result { + match c { + CommandType::SubscribeNotificationChannel => { + Ok(SubscribeNotificationChannelMachineEvents::CommandScheduled) + } + _ => Err(Self::Error::Nondeterminism(format!( + "SubscribeNotificationChannelMachine does not handle command type {c:?}" + ))), + } + } +} + +impl From for CommandIssued { + fn from(_: Created) -> Self { + Self {} + } +} diff --git a/crates/sdk-core/src/worker/workflow/machines/subscribe_notification_channel_state_machine_tests.rs b/crates/sdk-core/src/worker/workflow/machines/subscribe_notification_channel_state_machine_tests.rs new file mode 100644 index 000000000..02c5f9bc7 --- /dev/null +++ b/crates/sdk-core/src/worker/workflow/machines/subscribe_notification_channel_state_machine_tests.rs @@ -0,0 +1,311 @@ +use super::{ + HistEventData, Machines, OnEventWrapper, StateMachine, WFMachinesError, + subscribe_notification_channel_state_machine::*, +}; +use crate::{ + replay::TestHistoryBuilder, + test_help::{ + MockPollCfg, ResponseType, WorkerExt, build_mock_pollers, hist_to_poll_resp, mock_worker, + start_timer_cmd, + }, + worker::client::mocks::mock_worker_client, +}; +use parking_lot::Mutex; +use std::{sync::Arc, time::Duration}; +use temporalio_common::{ + protos::{ + coresdk::{ + workflow_commands::{ + CompleteWorkflowExecution, SubscribeNotificationChannel, workflow_command, + }, + workflow_completion::WorkflowActivationCompletion, + }, + temporal::api::{ + command::v1::command, + enums::v1::{CommandType, EventType, WorkflowTaskFailedCause}, + history::v1::{ + HistoryEvent, WorkflowNotificationChannelSubscribedEventAttributes, history_event, + }, + }, + }, + worker::WorkerTaskTypes, +}; + +fn subscribe(channel: &str) -> workflow_command::Variant { + SubscribeNotificationChannel { + channel: channel.to_string(), + } + .into() +} + +fn recorded(channel: &str) -> SubscribeNotificationChannelMachineEvents { + HistEventData { + event: HistoryEvent { + event_type: EventType::WorkflowNotificationChannelSubscribed as i32, + attributes: Some( + history_event::Attributes::WorkflowNotificationChannelSubscribedEventAttributes( + WorkflowNotificationChannelSubscribedEventAttributes { + workflow_task_completed_event_id: 0, + channel: channel.to_string(), + }, + ), + ), + ..Default::default() + }, + replaying: true, + current_task_is_last_in_history: false, + } + .try_into() + .unwrap() +} + +fn issued(channel: &str) -> SubscribeNotificationChannelMachine { + let Machines::SubscribeNotificationChannelMachine(mut sm) = + subscribe_notification_channel(SubscribeNotificationChannel { + channel: channel.to_string(), + }) + .machine + else { + unreachable!("the constructor builds a SubscribeNotificationChannelMachine"); + }; + OnEventWrapper::on_event_mut( + &mut sm, + CommandType::SubscribeNotificationChannel + .try_into() + .unwrap(), + ) + .expect("CommandScheduled should transition Created -> CommandIssued"); + assert_eq!(CommandIssued {}.to_string(), sm.state().to_string()); + sm +} + +#[test] +fn the_recorded_event_for_the_same_channel_completes_the_machine() { + let mut sm = issued("orders"); + OnEventWrapper::on_event_mut(&mut sm, recorded("orders")) + .expect("CommandRecorded should transition CommandIssued -> Done"); + assert_eq!(Done {}.to_string(), sm.state().to_string()); +} + +#[test] +fn a_recorded_event_for_another_channel_is_nondeterminism() { + let mut sm = issued("orders"); + let err = OnEventWrapper::on_event_mut(&mut sm, recorded("invoices")) + .expect_err("a different channel must not match"); + let message = format!("{err:?}"); + assert!( + message.contains("does not match") && message.contains("invoices"), + "the error must name both channels, got {message}" + ); +} + +#[test] +fn another_event_type_is_nondeterminism() { + let event = HistEventData { + event: HistoryEvent { + event_type: EventType::TimerStarted as i32, + ..Default::default() + }, + replaying: true, + current_task_is_last_in_history: false, + }; + let res: Result = event.try_into(); + assert!(matches!(res, Err(WFMachinesError::Nondeterminism(_)))); +} + +#[tokio::test] +async fn the_command_reaches_the_completion_with_its_channel() { + let mut t = TestHistoryBuilder::default(); + t.add_by_type(EventType::WorkflowExecutionStarted); + t.add_workflow_task_scheduled_and_started(); + + let sent = Arc::new(Mutex::new(vec![])); + let recorder = sent.clone(); + let mut cfg = MockPollCfg::from_resp_batches("fakeid", t, [1], mock_worker_client()); + cfg.completion_mock_fn = Some(Box::new(move |wftc| { + recorder.lock().extend(wftc.commands.iter().cloned()); + Ok(Default::default()) + })); + let mut mock = build_mock_pollers(cfg); + mock.worker_cfg(|w| { + w.task_types = WorkerTaskTypes::workflow_only(); + w.max_cached_workflows = 1; + }); + let worker = mock_worker(mock); + + let first = worker.poll_workflow_activation().await.unwrap(); + assert!(!first.is_replaying); + worker + .complete_workflow_activation(WorkflowActivationCompletion::from_cmd( + first.run_id, + subscribe("orders"), + )) + .await + .unwrap(); + + let commands = sent.lock().clone(); + assert_eq!(commands.len(), 1, "got {commands:?}"); + assert_eq!( + commands[0].command_type(), + CommandType::SubscribeNotificationChannel + ); + assert!( + matches!( + &commands[0].attributes, + Some(command::Attributes::SubscribeNotificationChannelCommandAttributes(a)) + if a.channel == "orders" + ), + "got {:?}", + commands[0] + ); + + worker.drain_pollers_and_shutdown().await; +} + +/// Replays a first task whose recorded command event `record` writes, with lang reissuing +/// `reissued`. Returns the nondeterminism message Core reported, if any, and whether lang was +/// activated for the live task after it. +async fn replay_first_task( + record: impl FnOnce(&mut TestHistoryBuilder), + reissued: workflow_command::Variant, + expect_failure: bool, +) -> (Option, bool) { + let mut t = TestHistoryBuilder::default(); + t.add_by_type(EventType::WorkflowExecutionStarted); + t.add_full_wf_task(); + record(&mut t); + // Gives the live task a job, so a match shows up as an activation rather than as silence. + t.add_we_signaled("go", vec![]); + t.add_workflow_task_scheduled_and_started(); + + let cold = hist_to_poll_resp(&t, "fakeid".to_owned(), ResponseType::AllHistory); + let mut cfg = MockPollCfg::from_resp_batches( + "fakeid", + t, + [ResponseType::Raw(cold.resp)], + mock_worker_client(), + ); + let failure = Arc::new(Mutex::new(None)); + if expect_failure { + cfg.num_expected_fails = 1; + let recorder = failure.clone(); + cfg.expect_fail_wft_matcher = Box::new(move |_, cause, f| { + if matches!(cause, WorkflowTaskFailedCause::NonDeterministicError) { + *recorder.lock() = Some(f.as_ref().map(|f| f.message.clone()).unwrap_or_default()); + } + true + }); + } + let mut mock = build_mock_pollers(cfg); + mock.worker_cfg(|w| { + w.task_types = WorkerTaskTypes::workflow_only(); + w.max_cached_workflows = 1; + }); + let worker = mock_worker(mock); + + let first = worker.poll_workflow_activation().await.unwrap(); + assert!(first.is_replaying); + worker + .complete_workflow_activation(WorkflowActivationCompletion::from_cmd( + first.run_id.clone(), + reissued, + )) + .await + .unwrap(); + + let next = tokio::time::timeout( + Duration::from_millis(500), + worker.poll_workflow_activation(), + ) + .await; + let mut activated = false; + if let Ok(Ok(act)) = next { + if act.is_only_eviction() { + worker + .complete_workflow_activation(WorkflowActivationCompletion::empty(act.run_id)) + .await + .unwrap(); + } else { + assert!( + !expect_failure, + "a mismatch must not activate lang, got {act:?}" + ); + activated = true; + worker + .complete_workflow_activation(WorkflowActivationCompletion::from_cmd( + act.run_id, + CompleteWorkflowExecution::default().into(), + )) + .await + .unwrap(); + } + } + worker.drain_pollers_and_shutdown().await; + let failure = failure.lock().clone(); + (failure, activated) +} + +#[tokio::test] +async fn replay_matches_the_recorded_subscription() { + let (failure, activated) = replay_first_task( + |t| { + t.add_notification_channel_subscribed("orders"); + }, + subscribe("orders"), + false, + ) + .await; + assert_eq!(failure, None); + assert!( + activated, + "the live task after a matched subscription must activate lang" + ); +} + +#[tokio::test] +async fn replay_with_another_channel_is_nondeterminism() { + let failure = replay_first_task( + |t| { + t.add_notification_channel_subscribed("invoices"); + }, + subscribe("orders"), + true, + ) + .await + .0 + .expect("the task must fail as nondeterminism"); + assert!(failure.contains("does not match"), "got {failure}"); +} + +#[tokio::test] +async fn replay_without_the_subscribed_event_is_nondeterminism() { + let failure = replay_first_task( + |t| { + t.add_timer_started("1".to_string()); + }, + subscribe("orders"), + true, + ) + .await + .0 + .expect("the task must fail as nondeterminism"); + assert!( + failure.contains("SubscribeNotificationChannelMachine does not handle"), + "got {failure}" + ); +} + +#[tokio::test] +async fn a_recorded_subscription_lang_does_not_reissue_is_nondeterminism() { + let failure = replay_first_task( + |t| { + t.add_notification_channel_subscribed("orders"); + }, + start_timer_cmd(1, Duration::from_secs(1)), + true, + ) + .await + .0 + .expect("the task must fail as nondeterminism"); + assert!(!failure.is_empty()); +} diff --git a/crates/sdk-core/src/worker/workflow/machines/transition_coverage.rs b/crates/sdk-core/src/worker/workflow/machines/transition_coverage.rs index 0ddd0daf8..daee6afbf 100644 --- a/crates/sdk-core/src/worker/workflow/machines/transition_coverage.rs +++ b/crates/sdk-core/src/worker/workflow/machines/transition_coverage.rs @@ -76,7 +76,10 @@ mod machine_coverage_report { local_activity_state_machine::LocalActivityMachine, modify_workflow_properties_state_machine::ModifyWorkflowPropertiesMachine, patch_state_machine::PatchMachine, signal_external_state_machine::SignalExternalMachine, - timer_state_machine::TimerMachine, update_state_machine::UpdateMachine, + subscribe_notification_channel_state_machine::SubscribeNotificationChannelMachine, + timer_state_machine::TimerMachine, + unsubscribe_notification_channel_state_machine::UnsubscribeNotificationChannelMachine, + update_state_machine::UpdateMachine, upsert_search_attributes_state_machine::UpsertSearchAttributesMachine, workflow_task_state_machine::WorkflowTaskMachine, }; @@ -117,6 +120,9 @@ mod machine_coverage_report { let mut modify_wf_props = ModifyWorkflowPropertiesMachine::visualizer().to_owned(); let mut update = UpdateMachine::visualizer().to_owned(); let mut external_stream = ExternalStreamMachine::visualizer().to_owned(); + let mut subscribe_channel = SubscribeNotificationChannelMachine::visualizer().to_owned(); + let mut unsubscribe_channel = + UnsubscribeNotificationChannelMachine::visualizer().to_owned(); // This isn't at all efficient but doesn't need to be. // Replace transitions in the vizzes with green color if they are covered. @@ -144,6 +150,12 @@ mod machine_coverage_report { } m @ "UpdateMachine" => cover_transitions(m, &mut update, coverage), m @ "ExternalStreamMachine" => cover_transitions(m, &mut external_stream, coverage), + m @ "SubscribeNotificationChannelMachine" => { + cover_transitions(m, &mut subscribe_channel, coverage) + } + m @ "UnsubscribeNotificationChannelMachine" => { + cover_transitions(m, &mut unsubscribe_channel, coverage) + } m => panic!("Unknown machine {m}"), } } diff --git a/crates/sdk-core/src/worker/workflow/machines/unsubscribe_notification_channel_state_machine.rs b/crates/sdk-core/src/worker/workflow/machines/unsubscribe_notification_channel_state_machine.rs new file mode 100644 index 000000000..b7724cb78 --- /dev/null +++ b/crates/sdk-core/src/worker/workflow/machines/unsubscribe_notification_channel_state_machine.rs @@ -0,0 +1,144 @@ +use super::{ + NewMachineWithCommand, StateMachine, TransitionResult, fsm, workflow_machines::MachineResponse, +}; +use crate::worker::workflow::{ + WFMachinesError, fatal, + machines::{EventInfo, HistEventData, WFMachinesAdapter}, + nondeterminism, +}; +use temporalio_common::protos::{ + coresdk::workflow_commands::UnsubscribeNotificationChannel, + temporal::api::{ + enums::v1::{CommandType, EventType}, + history::v1::{WorkflowNotificationChannelUnsubscribedEventAttributes, history_event}, + }, +}; + +fsm! { + pub(super) name UnsubscribeNotificationChannelMachine; + command UnsubscribeNotificationChannelMachineCommand; + error WFMachinesError; + shared_state SharedState; + + Created --(CommandScheduled) --> CommandIssued; + CommandIssued --(CommandRecorded(WorkflowNotificationChannelUnsubscribedEventAttributes), + shared on_command_recorded) --> Done; +} + +/// The channel the command named, kept so the recorded event can be held against it. +#[derive(Default, Clone)] +pub(super) struct SharedState { + channel: String, +} + +/// End this workflow's subscription to a notification channel. The server +/// records the event whether or not the run held a subscription, so the match +/// on replay is by channel alone and the recorded subscribed event id is not +/// part of it. +pub(super) fn unsubscribe_notification_channel( + lang_cmd: UnsubscribeNotificationChannel, +) -> NewMachineWithCommand { + let sm = UnsubscribeNotificationChannelMachine::from_parts( + Created {}.into(), + SharedState { + channel: lang_cmd.channel.clone(), + }, + ); + NewMachineWithCommand { + command: lang_cmd.into(), + machine: sm.into(), + } +} + +#[derive(Debug, derive_more::Display)] +pub(super) enum UnsubscribeNotificationChannelMachineCommand {} + +#[derive(Debug, Default, Clone, derive_more::Display)] +pub(super) struct Created {} + +#[derive(Debug, Default, Clone, derive_more::Display)] +pub(super) struct CommandIssued {} + +impl CommandIssued { + pub(super) fn on_command_recorded( + self, + dat: &mut SharedState, + attrs: WorkflowNotificationChannelUnsubscribedEventAttributes, + ) -> UnsubscribeNotificationChannelMachineTransition { + if dat.channel == attrs.channel { + TransitionResult::default() + } else { + TransitionResult::Err(nondeterminism!( + "Recorded unsubscription from notification channel {:?} does not match the \ + reissued unsubscription from channel {:?}", + attrs.channel, + dat.channel + )) + } + } +} + +#[derive(Debug, Default, Clone, derive_more::Display)] +pub(super) struct Done {} + +impl WFMachinesAdapter for UnsubscribeNotificationChannelMachine { + fn adapt_response( + &self, + _my_command: Self::Command, + _event_info: Option, + ) -> Result, Self::Error> { + Err(fatal!( + "UnsubscribeNotificationChannel does not use state machine commands" + )) + } +} + +impl TryFrom for UnsubscribeNotificationChannelMachineEvents { + type Error = WFMachinesError; + + fn try_from(e: HistEventData) -> Result { + let e = e.event; + match e.event_type() { + EventType::WorkflowNotificationChannelUnsubscribed => { + if let Some( + history_event::Attributes::WorkflowNotificationChannelUnsubscribedEventAttributes( + attrs, + ), + ) = e.attributes + { + Ok(UnsubscribeNotificationChannelMachineEvents::CommandRecorded( + attrs, + )) + } else { + Err(fatal!( + "Notification channel unsubscribed attributes were unset: {e}" + )) + } + } + _ => Err(Self::Error::Nondeterminism(format!( + "UnsubscribeNotificationChannelMachine does not handle {e}" + ))), + } + } +} + +impl TryFrom for UnsubscribeNotificationChannelMachineEvents { + type Error = WFMachinesError; + + fn try_from(c: CommandType) -> Result { + match c { + CommandType::UnsubscribeNotificationChannel => { + Ok(UnsubscribeNotificationChannelMachineEvents::CommandScheduled) + } + _ => Err(Self::Error::Nondeterminism(format!( + "UnsubscribeNotificationChannelMachine does not handle command type {c:?}" + ))), + } + } +} + +impl From for CommandIssued { + fn from(_: Created) -> Self { + Self {} + } +} diff --git a/crates/sdk-core/src/worker/workflow/machines/unsubscribe_notification_channel_state_machine_tests.rs b/crates/sdk-core/src/worker/workflow/machines/unsubscribe_notification_channel_state_machine_tests.rs new file mode 100644 index 000000000..72974e02c --- /dev/null +++ b/crates/sdk-core/src/worker/workflow/machines/unsubscribe_notification_channel_state_machine_tests.rs @@ -0,0 +1,313 @@ +use super::{ + HistEventData, Machines, OnEventWrapper, StateMachine, WFMachinesError, + unsubscribe_notification_channel_state_machine::*, +}; +use crate::{ + replay::TestHistoryBuilder, + test_help::{ + MockPollCfg, ResponseType, WorkerExt, build_mock_pollers, hist_to_poll_resp, mock_worker, + start_timer_cmd, + }, + worker::client::mocks::mock_worker_client, +}; +use parking_lot::Mutex; +use std::{sync::Arc, time::Duration}; +use temporalio_common::{ + protos::{ + coresdk::{ + workflow_commands::{ + CompleteWorkflowExecution, UnsubscribeNotificationChannel, workflow_command, + }, + workflow_completion::WorkflowActivationCompletion, + }, + temporal::api::{ + command::v1::command, + enums::v1::{CommandType, EventType, WorkflowTaskFailedCause}, + history::v1::{ + HistoryEvent, WorkflowNotificationChannelUnsubscribedEventAttributes, history_event, + }, + }, + }, + worker::WorkerTaskTypes, +}; + +fn unsubscribe(channel: &str) -> workflow_command::Variant { + UnsubscribeNotificationChannel { + channel: channel.to_string(), + } + .into() +} + +fn recorded(channel: &str) -> UnsubscribeNotificationChannelMachineEvents { + HistEventData { + event: HistoryEvent { + event_type: EventType::WorkflowNotificationChannelUnsubscribed as i32, + attributes: Some( + history_event::Attributes::WorkflowNotificationChannelUnsubscribedEventAttributes( + WorkflowNotificationChannelUnsubscribedEventAttributes { + workflow_task_completed_event_id: 0, + channel: channel.to_string(), + subscribed_event_id: 0, + }, + ), + ), + ..Default::default() + }, + replaying: true, + current_task_is_last_in_history: false, + } + .try_into() + .unwrap() +} + +fn issued(channel: &str) -> UnsubscribeNotificationChannelMachine { + let Machines::UnsubscribeNotificationChannelMachine(mut sm) = + unsubscribe_notification_channel(UnsubscribeNotificationChannel { + channel: channel.to_string(), + }) + .machine + else { + unreachable!("the constructor builds a UnsubscribeNotificationChannelMachine"); + }; + OnEventWrapper::on_event_mut( + &mut sm, + CommandType::UnsubscribeNotificationChannel + .try_into() + .unwrap(), + ) + .expect("CommandScheduled should transition Created -> CommandIssued"); + assert_eq!(CommandIssued {}.to_string(), sm.state().to_string()); + sm +} + +#[test] +fn the_recorded_event_for_the_same_channel_completes_the_machine() { + let mut sm = issued("orders"); + OnEventWrapper::on_event_mut(&mut sm, recorded("orders")) + .expect("CommandRecorded should transition CommandIssued -> Done"); + assert_eq!(Done {}.to_string(), sm.state().to_string()); +} + +#[test] +fn a_recorded_event_for_another_channel_is_nondeterminism() { + let mut sm = issued("orders"); + let err = OnEventWrapper::on_event_mut(&mut sm, recorded("invoices")) + .expect_err("a different channel must not match"); + let message = format!("{err:?}"); + assert!( + message.contains("does not match") && message.contains("invoices"), + "the error must name both channels, got {message}" + ); +} + +#[test] +fn the_subscribed_event_is_not_an_unsubscription() { + // The two events carry the same channel field, so the type is what tells them apart. + let event = HistEventData { + event: HistoryEvent { + event_type: EventType::WorkflowNotificationChannelSubscribed as i32, + ..Default::default() + }, + replaying: true, + current_task_is_last_in_history: false, + }; + let res: Result = event.try_into(); + assert!(matches!(res, Err(WFMachinesError::Nondeterminism(_)))); +} + +#[tokio::test] +async fn the_command_reaches_the_completion_with_its_channel() { + let mut t = TestHistoryBuilder::default(); + t.add_by_type(EventType::WorkflowExecutionStarted); + t.add_workflow_task_scheduled_and_started(); + + let sent = Arc::new(Mutex::new(vec![])); + let recorder = sent.clone(); + let mut cfg = MockPollCfg::from_resp_batches("fakeid", t, [1], mock_worker_client()); + cfg.completion_mock_fn = Some(Box::new(move |wftc| { + recorder.lock().extend(wftc.commands.iter().cloned()); + Ok(Default::default()) + })); + let mut mock = build_mock_pollers(cfg); + mock.worker_cfg(|w| { + w.task_types = WorkerTaskTypes::workflow_only(); + w.max_cached_workflows = 1; + }); + let worker = mock_worker(mock); + + let first = worker.poll_workflow_activation().await.unwrap(); + assert!(!first.is_replaying); + worker + .complete_workflow_activation(WorkflowActivationCompletion::from_cmd( + first.run_id, + unsubscribe("orders"), + )) + .await + .unwrap(); + + let commands = sent.lock().clone(); + assert_eq!(commands.len(), 1, "got {commands:?}"); + assert_eq!( + commands[0].command_type(), + CommandType::UnsubscribeNotificationChannel + ); + assert!( + matches!( + &commands[0].attributes, + Some(command::Attributes::UnsubscribeNotificationChannelCommandAttributes(a)) + if a.channel == "orders" + ), + "got {:?}", + commands[0] + ); + + worker.drain_pollers_and_shutdown().await; +} + +/// Replays a first task whose recorded command event `record` writes, with lang reissuing +/// `reissued`. Returns the nondeterminism message Core reported, if any, and whether lang was +/// activated for the live task after it. +async fn replay_first_task( + record: impl FnOnce(&mut TestHistoryBuilder), + reissued: workflow_command::Variant, + expect_failure: bool, +) -> (Option, bool) { + let mut t = TestHistoryBuilder::default(); + t.add_by_type(EventType::WorkflowExecutionStarted); + t.add_full_wf_task(); + record(&mut t); + // Gives the live task a job, so a match shows up as an activation rather than as silence. + t.add_we_signaled("go", vec![]); + t.add_workflow_task_scheduled_and_started(); + + let cold = hist_to_poll_resp(&t, "fakeid".to_owned(), ResponseType::AllHistory); + let mut cfg = MockPollCfg::from_resp_batches( + "fakeid", + t, + [ResponseType::Raw(cold.resp)], + mock_worker_client(), + ); + let failure = Arc::new(Mutex::new(None)); + if expect_failure { + cfg.num_expected_fails = 1; + let recorder = failure.clone(); + cfg.expect_fail_wft_matcher = Box::new(move |_, cause, f| { + if matches!(cause, WorkflowTaskFailedCause::NonDeterministicError) { + *recorder.lock() = Some(f.as_ref().map(|f| f.message.clone()).unwrap_or_default()); + } + true + }); + } + let mut mock = build_mock_pollers(cfg); + mock.worker_cfg(|w| { + w.task_types = WorkerTaskTypes::workflow_only(); + w.max_cached_workflows = 1; + }); + let worker = mock_worker(mock); + + let first = worker.poll_workflow_activation().await.unwrap(); + assert!(first.is_replaying); + worker + .complete_workflow_activation(WorkflowActivationCompletion::from_cmd( + first.run_id.clone(), + reissued, + )) + .await + .unwrap(); + + let next = tokio::time::timeout( + Duration::from_millis(500), + worker.poll_workflow_activation(), + ) + .await; + let mut activated = false; + if let Ok(Ok(act)) = next { + if act.is_only_eviction() { + worker + .complete_workflow_activation(WorkflowActivationCompletion::empty(act.run_id)) + .await + .unwrap(); + } else { + assert!( + !expect_failure, + "a mismatch must not activate lang, got {act:?}" + ); + activated = true; + worker + .complete_workflow_activation(WorkflowActivationCompletion::from_cmd( + act.run_id, + CompleteWorkflowExecution::default().into(), + )) + .await + .unwrap(); + } + } + worker.drain_pollers_and_shutdown().await; + let failure = failure.lock().clone(); + (failure, activated) +} + +#[tokio::test] +async fn replay_matches_the_recorded_unsubscription() { + let (failure, activated) = replay_first_task( + |t| { + t.add_notification_channel_unsubscribed("orders", 0); + }, + unsubscribe("orders"), + false, + ) + .await; + assert_eq!(failure, None); + assert!( + activated, + "the live task after a matched unsubscription must activate lang" + ); +} + +#[tokio::test] +async fn replay_with_another_channel_is_nondeterminism() { + let failure = replay_first_task( + |t| { + t.add_notification_channel_unsubscribed("invoices", 0); + }, + unsubscribe("orders"), + true, + ) + .await + .0 + .expect("the task must fail as nondeterminism"); + assert!(failure.contains("does not match"), "got {failure}"); +} + +#[tokio::test] +async fn a_recorded_subscription_does_not_match_an_unsubscription() { + let failure = replay_first_task( + |t| { + t.add_notification_channel_subscribed("orders"); + }, + unsubscribe("orders"), + true, + ) + .await + .0 + .expect("the task must fail as nondeterminism"); + assert!( + failure.contains("UnsubscribeNotificationChannelMachine does not handle"), + "got {failure}" + ); +} + +#[tokio::test] +async fn a_recorded_unsubscription_lang_does_not_reissue_is_nondeterminism() { + let failure = replay_first_task( + |t| { + t.add_notification_channel_unsubscribed("orders", 0); + }, + start_timer_cmd(1, Duration::from_secs(1)), + true, + ) + .await + .0 + .expect("the task must fail as nondeterminism"); + assert!(!failure.is_empty()); +} diff --git a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs index 787076f39..abbc5f95b 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs +++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs @@ -8,7 +8,10 @@ use super::{ continue_as_new_workflow_state_machine::continue_as_new, fail_workflow_state_machine::fail_workflow, local_activity_state_machine::new_local_activity, patch_state_machine::has_change, signal_external_state_machine::new_external_signal, - timer_state_machine::new_timer, upsert_search_attributes_state_machine::upsert_search_attrs, + subscribe_notification_channel_state_machine::subscribe_notification_channel, + timer_state_machine::new_timer, + unsubscribe_notification_channel_state_machine::unsubscribe_notification_channel, + upsert_search_attributes_state_machine::upsert_search_attrs, workflow_machines::local_acts::LocalActivityData, workflow_task_state_machine::WorkflowTaskMachine, }; @@ -63,8 +66,8 @@ use temporalio_common::{ external_data::{ExternalStreamMarkerData, ParkReason}, external_stream, workflow_activation::{ - self, NotifyHasPatch, ReplayExternalStreams, UpdateRandomSeed, WorkflowActivation, - workflow_activation_job, + self, NotificationsReceived, NotifyHasPatch, ReplayExternalStreams, + UpdateRandomSeed, WorkflowActivation, workflow_activation_job, }, workflow_commands::{ContinueAsNewWorkflowExecution, ExternalStreamWait}, }, @@ -75,6 +78,7 @@ use temporalio_common::{ common::v1::SearchAttributes, enums::v1::EventType, history::v1::{HistoryEvent, history_event}, + notification::v1::Notification, protocol::v1::{Message as ProtocolMessage, message::SequencingId}, sdk::v1::{UserMetadata, WorkflowTaskCompletedMetadata}, }, @@ -98,6 +102,11 @@ pub(crate) struct WorkflowMachines { /// Reserved external stream wake Signals seen in history, decoded and suppressed from user /// dispatch, waiting to be classified against the run's wait set. pending_external_stream_wakes: Vec, + /// Channel notifications read from the scheduled events of the task being applied, folded + /// per channel. The server clears what it put on a scheduled event, so a retry's event + /// carries only what arrived since. The failed task's notifications stay in History and + /// reach lang here, folded with the retry's into one job. + pending_notifications: Vec, /// External stream marker machines whose `MarkerRecorded` event has not been reached yet, in /// the order the markers appear in History. /// @@ -307,6 +316,7 @@ impl WorkflowMachines { current_wf_time: None, observed_internal_flags: Rc::new(RefCell::new(observed_internal_flags)), pending_external_stream_wakes: vec![], + pending_notifications: vec![], external_stream_marker_machines: Default::default(), history_size_bytes: 0, continue_as_new_suggested: false, @@ -583,6 +593,23 @@ impl WorkflowMachines { std::mem::take(&mut self.pending_external_stream_wakes) } + /// Hands lang the notifications from this task's scheduled events as one job. + /// + /// Sent while the task's history is applied, so the job sits ahead of any external stream + /// resolve Core queues for the same activation once the history is in. + fn flush_pending_notifications(&mut self) { + if self.pending_notifications.is_empty() { + return; + } + let notifications = std::mem::take(&mut self.pending_notifications); + self.drive_me.send_job( + workflow_activation_job::Variant::NotificationsReceived(NotificationsReceived { + notifications, + }) + .into(), + ); + } + /// Queue a Core-generated job for lang. /// /// Used by the external stream paths, whose activations originate in Core rather than from a @@ -894,6 +921,7 @@ impl WorkflowMachines { } self.last_processed_event = eid; } + self.flush_pending_notifications(); // Needed to delay mutation of self until after we've iterated over peeked events. #[allow(clippy::large_enum_variant)] @@ -1212,6 +1240,14 @@ impl WorkflowMachines { } } Ok(EventType::WorkflowTaskScheduled) => { + if let Some(history_event::Attributes::WorkflowTaskScheduledEventAttributes( + ref attrs, + )) = event_dat.event.attributes + { + for n in &attrs.notifications { + fold_notification(&mut self.pending_notifications, n.clone()); + } + } let wf_task_sm = WorkflowTaskMachine::new(self.next_started_event_id); let key = self.all_machines.insert(wf_task_sm.into()); self.submachine_handle_event(key, event_dat)?; @@ -1713,6 +1749,22 @@ impl WorkflowMachines { CommandIdKind::NeverResolves, ); } + WFCommandVariant::SubscribeNotificationChannel(attrs) => { + // Never resolves: the event it produces records the subscription and hands + // nothing back. The notifications arrive later on a scheduled event. + self.add_cmd_to_wf_task( + subscribe_notification_channel(attrs), + cmd.metadata, + CommandIdKind::NeverResolves, + ); + } + WFCommandVariant::UnsubscribeNotificationChannel(attrs) => { + self.add_cmd_to_wf_task( + unsubscribe_notification_channel(attrs), + cmd.metadata, + CommandIdKind::NeverResolves, + ); + } WFCommandVariant::UpdateResponse(ur) => { let m_key = self.get_machine_by_msg(&ur.protocol_instance_id)?; let m = if let Machines::UpdateMachine(m) = self.machine_mut(m_key) { @@ -2048,3 +2100,16 @@ fn decode_wake_signal( } Some(wake) } + +/// Keep one notification per channel, in order of first appearance. A later one replaces +/// the held one only when its counter is strictly higher, so on a tie the held one stays. +fn fold_notification(folded: &mut Vec, n: Notification) { + match folded.iter_mut().find(|held| held.channel == n.channel) { + Some(held) => { + if n.counter > held.counter { + *held = n; + } + } + None => folded.push(n), + } +} diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs index 5450c6939..820ecc23b 100644 --- a/crates/sdk-core/src/worker/workflow/mod.rs +++ b/crates/sdk-core/src/worker/workflow/mod.rs @@ -1728,6 +1728,8 @@ enum WFCommandVariant { ExternalOutputStreamCommit(WorkflowOutputStreamCommit), /// External output is buffered in lang and needs a run-scoped flush deadline. ExternalOutputStreamBuffered(WorkflowOutputStreamBuffered), + SubscribeNotificationChannel(SubscribeNotificationChannel), + UnsubscribeNotificationChannel(UnsubscribeNotificationChannel), } impl TryFrom for WFCommand { @@ -1807,6 +1809,12 @@ impl TryFrom for WFCommand { workflow_command::Variant::WorkflowOutputStreamBuffered(buffered) => { WFCommandVariant::ExternalOutputStreamBuffered(buffered) } + workflow_command::Variant::SubscribeNotificationChannel(s) => { + WFCommandVariant::SubscribeNotificationChannel(s) + } + workflow_command::Variant::UnsubscribeNotificationChannel(u) => { + WFCommandVariant::UnsubscribeNotificationChannel(u) + } }; Ok(Self { variant, @@ -2026,7 +2034,7 @@ impl LocalActivityRequestSink for LAReqSink { /// 1. init workflow /// 2. patches /// 3. random-seed-updates -/// 4. signals/updates +/// 4. signals/updates/channel notifications /// 5. all others /// 6. local activity resolutions /// 7. queries @@ -2066,6 +2074,7 @@ fn prepare_to_ship_activation(wfa: &mut WorkflowActivation) { workflow_activation_job::Variant::UpdateRandomSeed(_) => 2, workflow_activation_job::Variant::SignalWorkflow(_) => 3, workflow_activation_job::Variant::DoUpdate(_) => 3, + workflow_activation_job::Variant::NotificationsReceived(_) => 3, workflow_activation_job::Variant::ResolveActivity(ra) if ra.is_local => 5, // In principle we should never actually need to sort these with the others, since // queries always get their own activation, but, maintaining the semantic is