Detect Nexus operation changes during replay

CHANGELOG.md

@@ -133,6 +133,8 @@ 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.
 * `WorkflowContext::force_task_fail` calls will be respected over a completion if both happen in the same poll
 * Workers no longer advertise a worker control task queue unless the namespace supports worker
   heartbeats and commands and the built-in Nexus command worker is running.

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 <https://github.com/temporalio/features/issues/481>.
     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,
         }
     }

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<ScheduledEventRecorded> {
+        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<HistEventData> for NexusOperationMachineEvents {
     type Error = WFMachinesError;
 
     fn try_from(e: HistEventData) -> Result<Self, Self::Error> {
+        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<Vec<MachineResponse>, 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());
+    }
+}

crates/sdk-core/src/worker/workflow/machines/transition_coverage.rs

@@ -74,8 +74,9 @@ mod machine_coverage_report {
         fail_workflow_state_machine::FailWorkflowMachine,
         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,
+        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,
     };
@@ -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}"),
             }
         }

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(),
                     );