From c10bf2982211325da83e04da55c1b8a07e4b9118 Mon Sep 17 00:00:00 2001 From: JSup Date: Thu, 16 Jul 2026 04:36:37 +0900 Subject: [PATCH 1/3] fix: detect nexus operation changes during replay --- CHANGELOG.md | 2 + crates/sdk-core/src/internal_flags.rs | 3 + .../machines/nexus_operation_state_machine.rs | 147 ++++++++++++++++-- .../workflow/machines/workflow_machines.rs | 5 +- 4 files changed, 147 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 966ee13fa..b96297498 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -99,3 +99,5 @@ to docs, or any other relevant information. metric reader reports an export error. * Worker heartbeat now samples host CPU/memory at the heartbeat interval (only when enabled) rather than every 100ms. +* Workflow replay now reports nondeterminism when a scheduled Nexus operation's service or operation + differs from the command that produced it. diff --git a/crates/sdk-core/src/internal_flags.rs b/crates/sdk-core/src/internal_flags.rs index 8155a7f67..60c22877b 100644 --- a/crates/sdk-core/src/internal_flags.rs +++ b/crates/sdk-core/src/internal_flags.rs @@ -33,6 +33,8 @@ pub enum CoreInternalFlags { /// if in the sequence delivered by lang they came after a terminal command. /// See . MoveTerminalCommands = 3, + /// Detects service and operation changes when replaying Nexus operation schedules. + NexusOperationDeterminismChecks = 4, /// We received a value higher than this code can understand. TooHigh = u32::MAX, } @@ -175,6 +177,7 @@ impl CoreInternalFlags { 1 => Self::IdAndTypeDeterminismChecks, 2 => Self::UpsertSearchAttributeOnPatch, 3 => Self::MoveTerminalCommands, + 4 => Self::NexusOperationDeterminismChecks, _ => Self::TooHigh, } } diff --git a/crates/sdk-core/src/worker/workflow/machines/nexus_operation_state_machine.rs b/crates/sdk-core/src/worker/workflow/machines/nexus_operation_state_machine.rs index b8c744982..87271c45d 100644 --- a/crates/sdk-core/src/worker/workflow/machines/nexus_operation_state_machine.rs +++ b/crates/sdk-core/src/worker/workflow/machines/nexus_operation_state_machine.rs @@ -1,11 +1,14 @@ use super::{MachineError, StateMachine, TransitionResult, fsm}; -use crate::worker::workflow::{ - WFMachinesError, - machines::{ - EventInfo, HistEventData, NewMachineWithCommand, OnEventWrapper, WFMachinesAdapter, - workflow_machines::MachineResponse, +use crate::{ + internal_flags::CoreInternalFlags, + worker::workflow::{ + InternalFlagsRef, WFMachinesError, + machines::{ + EventInfo, HistEventData, NewMachineWithCommand, OnEventWrapper, WFMachinesAdapter, + workflow_machines::MachineResponse, + }, + nondeterminism, }, - nondeterminism, }; use itertools::Itertools; use temporalio_common::protos::{ @@ -124,6 +127,7 @@ pub(super) struct SharedState { endpoint: String, service: String, operation: String, + internal_flags: InternalFlagsRef, cancelled_before_sent: bool, cancel_sent: bool, @@ -132,7 +136,10 @@ pub(super) struct SharedState { } impl NexusOperationMachine { - pub(super) fn new_scheduled(attribs: ScheduleNexusOperation) -> NewMachineWithCommand { + pub(super) fn new_scheduled( + attribs: ScheduleNexusOperation, + internal_flags: InternalFlagsRef, + ) -> NewMachineWithCommand { let s = Self::from_parts( ScheduleCommandCreated.into(), SharedState { @@ -141,6 +148,7 @@ impl NexusOperationMachine { endpoint: attribs.endpoint.clone(), service: attribs.service.clone(), operation: attribs.operation.clone(), + internal_flags, cancelled_before_sent: false, cancel_sent: false, cancel_type: attribs.cancellation_type(), @@ -174,6 +182,9 @@ pub(super) struct ScheduleCommandCreated; pub(super) struct NexusOpScheduledData { event_id: i64, + service: String, + operation: String, + last_task_in_history: bool, } impl ScheduleCommandCreated { @@ -182,6 +193,25 @@ impl ScheduleCommandCreated { state: &mut SharedState, event_dat: NexusOpScheduledData, ) -> NexusOperationMachineTransition { + if state.internal_flags.borrow_mut().try_use( + CoreInternalFlags::NexusOperationDeterminismChecks, + event_dat.last_task_in_history, + ) { + if event_dat.service != state.service { + return TransitionResult::Err(nondeterminism!( + "Nexus operation service of scheduled event '{}' does not match service of command '{}'", + event_dat.service, + state.service + )); + } + if event_dat.operation != state.operation { + return TransitionResult::Err(nondeterminism!( + "Nexus operation of scheduled event '{}' does not match operation of command '{}'", + event_dat.operation, + state.operation + )); + } + } state.scheduled_event_id = event_dat.event_id; NexusOperationMachineTransition::default() } @@ -445,14 +475,19 @@ impl TryFrom for NexusOperationMachineEvents { type Error = WFMachinesError; fn try_from(e: HistEventData) -> Result { + let last_task_in_history = e.current_task_is_last_in_history; let e = e.event; Ok(match EventType::try_from(e.event_type) { Ok(EventType::NexusOperationScheduled) => { - if let Some(history_event::Attributes::NexusOperationScheduledEventAttributes(_)) = - e.attributes + if let Some(history_event::Attributes::NexusOperationScheduledEventAttributes( + attrs, + )) = e.attributes { Self::NexusOperationScheduled(NexusOpScheduledData { event_id: e.event_id, + service: attrs.service, + operation: attrs.operation, + last_task_in_history, }) } else { return Err(nondeterminism!( @@ -716,3 +751,97 @@ impl SharedState { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + internal_flags::InternalFlags, + worker::workflow::machines::{Machines, TemporalStateMachine}, + }; + use std::{cell::RefCell, rc::Rc}; + use temporalio_common::protos::temporal::api::{ + enums::v1::EventType, + history::v1::{HistoryEvent, NexusOperationScheduledEventAttributes, history_event}, + workflowservice::v1::get_system_info_response::Capabilities, + }; + + fn apply_scheduled_event( + service: &str, + operation: &str, + current_task_is_last_in_history: bool, + ) -> Result, WFMachinesError> { + let internal_flags = Rc::new(RefCell::new(InternalFlags::new( + &Capabilities { + sdk_metadata: true, + ..Default::default() + }, + "sdk".to_owned(), + "version".to_owned(), + ))); + let new_machine = NexusOperationMachine::new_scheduled( + ScheduleNexusOperation { + service: "service".to_owned(), + operation: "operation".to_owned(), + ..Default::default() + }, + internal_flags, + ); + let mut machine = match new_machine.machine { + Machines::NexusOperationMachine(machine) => machine, + _ => panic!("wrong machine type"), + }; + + machine.handle_event(HistEventData { + event: HistoryEvent { + event_id: 1, + event_type: EventType::NexusOperationScheduled as i32, + attributes: Some( + history_event::Attributes::NexusOperationScheduledEventAttributes( + NexusOperationScheduledEventAttributes { + service: service.to_owned(), + operation: operation.to_owned(), + ..Default::default() + }, + ), + ), + ..Default::default() + }, + replaying: true, + current_task_is_last_in_history, + }) + } + + #[test] + fn matching_service_and_operation_are_deterministic() { + assert!(apply_scheduled_event("service", "operation", true).is_ok()); + } + + #[test] + fn service_mismatch_is_nondeterministic() { + let result = apply_scheduled_event("other-service", "operation", true); + + assert!(matches!( + result, + Err(WFMachinesError::Nondeterminism(message)) + if message.contains("Nexus operation service") && message.contains("does not match") + )); + } + + #[test] + fn operation_mismatch_is_nondeterministic() { + let result = apply_scheduled_event("service", "other-operation", true); + + assert!(matches!( + result, + Err(WFMachinesError::Nondeterminism(message)) + if message.contains("Nexus operation of scheduled event") + && message.contains("does not match") + )); + } + + #[test] + fn mismatch_is_allowed_before_determinism_flag_is_recorded() { + assert!(apply_scheduled_event("other-service", "other-operation", false).is_ok()); + } +} 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 e0f13a23e..d0c21b9df 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs +++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs @@ -1546,7 +1546,10 @@ impl WorkflowMachines { WFCommandVariant::ScheduleNexusOperation(attrs) => { let seq = attrs.seq; self.add_cmd_to_wf_task( - NexusOperationMachine::new_scheduled(attrs), + NexusOperationMachine::new_scheduled( + attrs, + self.observed_internal_flags.clone(), + ), cmd.metadata, CommandID::NexusOperation(seq).into(), ); From 4887408816bd18953fb3b7b9a5789fd79276cf84 Mon Sep 17 00:00:00 2001 From: JSup Date: Thu, 16 Jul 2026 05:10:24 +0900 Subject: [PATCH 2/3] fix: register NexusOperationMachine in transition coverage reporter --- .../src/worker/workflow/machines/transition_coverage.rs | 3 +++ 1 file changed, 3 insertions(+) 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 b461d8dd6..2cf9f310d 100644 --- a/crates/sdk-core/src/worker/workflow/machines/transition_coverage.rs +++ b/crates/sdk-core/src/worker/workflow/machines/transition_coverage.rs @@ -74,6 +74,7 @@ mod machine_coverage_report { fail_workflow_state_machine::FailWorkflowMachine, local_activity_state_machine::LocalActivityMachine, modify_workflow_properties_state_machine::ModifyWorkflowPropertiesMachine, + nexus_operation_state_machine::NexusOperationMachine, patch_state_machine::PatchMachine, signal_external_state_machine::SignalExternalMachine, timer_state_machine::TimerMachine, update_state_machine::UpdateMachine, upsert_search_attributes_state_machine::UpsertSearchAttributesMachine, @@ -115,6 +116,7 @@ mod machine_coverage_report { let mut upsert_search_attr = UpsertSearchAttributesMachine::visualizer().to_owned(); let mut modify_wf_props = ModifyWorkflowPropertiesMachine::visualizer().to_owned(); let mut update = UpdateMachine::visualizer().to_owned(); + let mut nexus_op = NexusOperationMachine::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. @@ -141,6 +143,7 @@ mod machine_coverage_report { cover_transitions(m, &mut modify_wf_props, coverage) } m @ "UpdateMachine" => cover_transitions(m, &mut update, coverage), + m @ "NexusOperationMachine" => cover_transitions(m, &mut nexus_op, coverage), m => panic!("Unknown machine {m}"), } } From 37f932ec3d1d0b323c1ab6e58c59a0c7311a61d2 Mon Sep 17 00:00:00 2001 From: JSup Date: Thu, 23 Jul 2026 08:47:24 +0900 Subject: [PATCH 3/3] style: format Nexus transition coverage imports Signed-off-by: JSup --- .../src/worker/workflow/machines/transition_coverage.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 2cf9f310d..48d0793b2 100644 --- a/crates/sdk-core/src/worker/workflow/machines/transition_coverage.rs +++ b/crates/sdk-core/src/worker/workflow/machines/transition_coverage.rs @@ -74,9 +74,9 @@ mod machine_coverage_report { fail_workflow_state_machine::FailWorkflowMachine, local_activity_state_machine::LocalActivityMachine, modify_workflow_properties_state_machine::ModifyWorkflowPropertiesMachine, - nexus_operation_state_machine::NexusOperationMachine, - patch_state_machine::PatchMachine, signal_external_state_machine::SignalExternalMachine, - timer_state_machine::TimerMachine, update_state_machine::UpdateMachine, + nexus_operation_state_machine::NexusOperationMachine, patch_state_machine::PatchMachine, + signal_external_state_machine::SignalExternalMachine, timer_state_machine::TimerMachine, + update_state_machine::UpdateMachine, upsert_search_attributes_state_machine::UpsertSearchAttributesMachine, workflow_task_state_machine::WorkflowTaskMachine, };