Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 142 additions & 5 deletions crates/workflow/src/engine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -956,11 +956,23 @@ impl<D: WorkflowData, S: StateStore<D> + 'static> WorkflowEngine<D, S> {

let step_duration = step_start.elapsed();

self.state_store
.update(instance_id, |s| {
s.context = context.clone();
})
.await?;
// Persist the step's context mutations — but NOT when the step opted
// out via `StepResult::Skip`. Sibling steps that become ready at the
// same time run in parallel, and each takes its own `get_context`
// snapshot above; the write-back below replaces the *entire* shared
// context (last-writer-wins). A skipped step did no work, so its
// snapshot is stale relative to a concurrent sibling's write, and
// persisting it silently clobbers that sibling's mutations. This was
// the worker-registration race: one branch's no-op `Skip` steps wiped
// the other branch's `actual_workers`, surfacing downstream as
// `register_workers: Context value not found: workers`.
if !matches!(result, Ok(Ok(StepResult::Skip))) {
self.state_store
.update(instance_id, |s| {
s.context = context.clone();
})
.await?;
}
Comment on lines +969 to +975

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

When a step returns StepResult::Skip, its status in the state store is never updated to StepStatus::Skipped and remains StepStatus::Running (or StepStatus::Retrying) indefinitely.

At the start of execute_step_with_retry (line 931), the step status is set to StepStatus::Running or StepStatus::Retrying. When the step returns StepResult::Skip, execute_step_with_retry does not update the step's status in the state store (unlike the Success case). Furthermore, in the parallel execution loop (lines 755-758), needs_update is set to false for Ok(StepResult::Skip), meaning the caller also skips updating the state store.

To fix this, we should update the step status to StepStatus::Skipped and set completed_at when StepResult::Skip is returned.

            if !matches!(result, Ok(Ok(StepResult::Skip))) {
                self.state_store
                    .update(instance_id, |s| {
                        s.context = context.clone();
                    })
                    .await?;
            } else {
                self.state_store
                    .update(instance_id, |s| {
                        if let Some(step_state) = s.step_states.get_mut(&step.id) {
                            step_state.status = StepStatus::Skipped;
                            step_state.completed_at = Some(Utc::now());
                        }
                    })
                    .await?;
            }


match result {
Ok(Ok(StepResult::Success)) => {
Expand Down Expand Up @@ -1254,3 +1266,128 @@ impl<D: WorkflowData, S: StateStore<D> + 'static> std::fmt::Debug for WorkflowEn
.finish()
}
}

#[cfg(test)]
mod skip_clobber_tests {
use std::{sync::Arc, time::Duration};

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use tokio::sync::Notify;

use crate::{
StepDefinition, StepExecutor, StepResult, WorkflowContext, WorkflowData,
WorkflowDefinition, WorkflowEngine, WorkflowResult, WorkflowStatus,
};

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
struct Data {
value: String,
}

impl WorkflowData for Data {
fn workflow_type() -> &'static str {
"skip_clobber_test"
}
}

/// Writes `value = "written"`, but only after the skipper has taken its
/// snapshot — so the write is guaranteed to land after that snapshot.
struct Writer {
skipper_started: Arc<Notify>,
writer_done: Arc<Notify>,
}

#[async_trait]
impl StepExecutor<Data> for Writer {
async fn execute(&self, ctx: &mut WorkflowContext<Data>) -> WorkflowResult<StepResult> {
self.skipper_started.notified().await;
ctx.data.value = "written".to_string();
self.writer_done.notify_one();
Ok(StepResult::Success)
}
}

/// Returns `Skip` without mutating, but only after the writer has written and
/// its write-back has landed. A buggy unconditional write-back of this step's
/// stale snapshot would then clobber the writer's value.
struct SlowSkipper {
skipper_started: Arc<Notify>,
writer_done: Arc<Notify>,
}

#[async_trait]
impl StepExecutor<Data> for SlowSkipper {
async fn execute(&self, _ctx: &mut WorkflowContext<Data>) -> WorkflowResult<StepResult> {
// Reaching here implies the engine already snapshotted our context.
self.skipper_started.notify_one();
self.writer_done.notified().await;
// Cushion so the writer's async write-back is fully applied first.
tokio::time::sleep(Duration::from_millis(50)).await;
Ok(StepResult::Skip)
Comment on lines +1324 to +1327

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick | 🔵 Trivial | 💤 Low value

Minor: timing-based synchronization could be fragile under load.

The 50ms sleep ensures Writer's context persistence completes before SlowSkipper returns, but writer_done is notified inside Writer's execute(), before the persistence at lines 969-975 actually runs. Under heavy CI load, 50ms might occasionally be insufficient.

Consider bumping to 100-200ms for extra margin, or adding a comment acknowledging this is a best-effort timing cushion rather than a deterministic guarantee.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@crates/workflow/src/engine.rs` around lines 1324 - 1327, The timing cushion
after awaiting self.writer_done is fragile because writer_done is signaled
inside Writer::execute() before the actual persistence step runs; increase the
sleep from 50ms to a larger margin (e.g., 150–200ms) and add a clarifying
comment that this is a best-effort timing cushion (not a deterministic sync),
referencing the writer_done notification and Writer::execute() persistence
sequence so future readers understand why the extra delay is required.

}
}

/// A `Skip` step must not persist its (stale) context snapshot, otherwise it
/// clobbers a concurrently-running sibling step's writes. Regression test for
/// the worker-registration race (`register_workers: Context value not found:
/// workers`).
#[tokio::test]
async fn skipped_step_does_not_clobber_sibling_context() {
let skipper_started = Arc::new(Notify::new());
let writer_done = Arc::new(Notify::new());

// Both steps have no dependencies, so the engine launches them in
// parallel — reproducing the sibling-branch race.
let def = WorkflowDefinition::<Data>::new("clobber", "Clobber Repro")
.add_step(StepDefinition::new(
"writer",
"Writer",
Arc::new(Writer {
skipper_started: skipper_started.clone(),
writer_done: writer_done.clone(),
}),
))
.add_step(StepDefinition::new(
"skipper",
"Skipper",
Arc::new(SlowSkipper {
skipper_started: skipper_started.clone(),
writer_done: writer_done.clone(),
}),
));

let engine = WorkflowEngine::<Data>::new();
let def_id = def.id.clone();
engine.register_workflow(def).expect("register workflow");

let instance = engine
.start_workflow(def_id, Data::default())
.await
.expect("start workflow");

// Poll for terminal status (wait_for_completion cleans up the state we
// need to inspect).
let mut state = engine.get_status(instance).await.expect("status");
for _ in 0..200 {
if matches!(
state.status,
WorkflowStatus::Completed | WorkflowStatus::Failed
) {
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
state = engine.get_status(instance).await.expect("status");
}

assert_eq!(
state.status,
WorkflowStatus::Completed,
"workflow should complete"
);
assert_eq!(
state.context.data.value, "written",
"skipped sibling clobbered the writer's context mutation"
);
Comment on lines +1388 to +1391

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Add an assertion to verify that the skipped step's status is correctly updated to StepStatus::Skipped in the state store, ensuring that skipped steps do not remain in the Running state.

        assert_eq!(
            state.context.data.value, "written",
            "skipped sibling clobbered the writer's context mutation"
        );
        assert_eq!(
            state.step_states.get("skipper").unwrap().status,
            StepStatus::Skipped,
            "skipped step should have Skipped status in the state store"
        );

}
}
Loading