fix(k8s): skip health check failure if STEP_SUCCESS already in event log

[futurewasfree]

Summary & Motivation

Fixes a race condition in the K8s executor (#22565) where a step could be marked as failed even though it had already emitted STEP_SUCCESS.

The StepDelegatingExecutor runs two concurrent concerns in the same polling loop: tailing the event log via _pop_events (every ~1s) and running a K8s job health check via check_step_health (every 20s). A step is only removed from running_steps once _pop_events returns a STEP_SUCCESS event. This creates a window where the health check fires, finds the K8s job gone or failed, and emits STEP_FAILURE — even though STEP_SUCCESS is already in the database, just not yet consumed by the tailer.

Two concrete triggers:

1. TTL race — ttlSecondsAfterFinished is short enough that Kubernetes deletes the job between the pod writing STEP_SUCCESS and the executor polling it. get_job_status returns 404, health check returns unhealthy.

2. OOMKill race — the pod is killed during post-success cleanup (IO manager teardown, connection closing, etc.) after writing STEP_SUCCESS but before the process exits. Kubernetes marks the job as failed (status.failed > 0), health check returns unhealthy.

The fix is in StepDelegatingExecutor.execute: when a health check returns unhealthy, query the event log directly for a STEP_SUCCESS event matching the step key. If one exists, skip emitting the failure — the next _pop_events iteration will consume the success event and cleanly remove the step from running_steps.

How I Tested These Changes

Manually traced the race condition through the executor polling loop. The query uses of_type=DagsterEventType.STEP_SUCCESS, which is indexed in the event log schema, so the overhead on the genuine-failure path is one narrow DB query before the failure is emitted.

Changelog

Fixed an intermittent false STEP_FAILURE on the K8s executor where a step that had already succeeded could be marked as failed if the Kubernetes job was deleted (via ttlSecondsAfterFinished) or killed (OOMKill during cleanup) before the executor's event tailer processed the STEP_SUCCESS event.

[greptile-apps]

Greptile Summary

This PR fixes a race condition in StepDelegatingExecutor where a K8s step could be marked failed even after writing STEP_SUCCESS, if the health check fired before _pop_events consumed that event. The fix queries the event log for unconsumed STEP_SUCCESS or STEP_UP_FOR_RETRY records (filtered via seen_storage_ids) before emitting a health-check failure.

• Executor fix (step_delegating_executor.py): When check_step_health returns unhealthy, the executor now queries the DB for any unconsumed terminal step events matching the step key; if found, it skips emitting STEP_FAILURE and lets the next _pop_events call cleanly handle the event. The seen_storage_ids guard correctly prevents a consumed STEP_UP_FOR_RETRY from a prior retry attempt from masking a genuine failure on the current attempt.
• Tests (test_step_delegating_executor.py): Two new tests are added — one asserting that an unconsumed STEP_SUCCESS suppresses a false failure, and one asserting that a consumed STEP_UP_FOR_RETRY does not suppress a genuine failure on a later attempt. The first test has a design issue: it injects a mock STEP_SUCCESS only for slow_op_0 while running a 3-op parallel job, leaving the other two ops unprotected.

Confidence Score: 4/5

The executor change itself is logically correct and handles the race condition well; the test added to validate it has a design flaw that would cause it to fail in practice.

The production fix in step_delegating_executor.py is sound — the seen_storage_ids guard cleanly distinguishes unconsumed events from consumed ones, handling both the TTL and OOMKill races. The concern is in the new test: it runs a 3-op parallel job but only injects a mock unconsumed STEP_SUCCESS for slow_op_0. Health checks for slow_op_1 and slow_op_2 will find no matching unconsumed STEP_SUCCESS during their 2-second sleep window, so false failures will be emitted for them — breaking the test's own assertions of zero failures and a successful run.

The new test in test_step_delegating_executor.py needs attention: the mock injecting an unconsumed STEP_SUCCESS must cover all steps in the job, not just slow_op_0.

Important Files Changed

Filename	Overview
python_modules/dagster/dagster/_core/executor/step_delegating/step_delegating_executor.py	Adds a DB guard before emitting STEP_FAILURE: checks for unconsumed STEP_SUCCESS or STEP_UP_FOR_RETRY events in the event log before declaring a health-check failure, correctly filtering by seen_storage_ids to avoid suppressing genuine failures on retry attempts. Logic is sound; typo fix in comment also included.
python_modules/dagster/dagster_tests/execution_tests/engine_tests/test_step_delegating_executor.py	Adds two new tests for the health-check race fix and a consumed-STEP_UP_FOR_RETRY regression guard. The first new test has a design flaw: it only injects a mock STEP_SUCCESS for slow_op_0 while running a 3-op parallel job, leaving slow_op_1 and slow_op_2 unprotected from false failures.

Sequence Diagram

sequenceDiagram
    participant Pod as K8s Pod
    participant DB as Event Log DB
    participant Tailer as _pop_events
    participant HealthCheck as check_step_health
    participant Executor as StepDelegatingExecutor

    Pod->>DB: Write STEP_SUCCESS
    Note over Pod: TTL deletes job / OOMKill

    HealthCheck->>HealthCheck: fires (interval elapsed)
    HealthCheck->>Executor: returns unhealthy (job gone/failed)

    Note over Executor: NEW: guard before emitting failure
    Executor->>DB: "get_records_for_run(STEP_SUCCESS | STEP_UP_FOR_RETRY)"
    DB-->>Executor: [STEP_SUCCESS record, storage_id not in seen_storage_ids]
    Executor->>Executor: "step_already_handled = True, continue (skip failure)"

    Tailer->>DB: get_records_for_run (next poll)
    DB-->>Tailer: STEP_SUCCESS record
    Tailer->>Executor: yield STEP_SUCCESS event
    Executor->>Executor: del running_steps[step.key]

Loading

_{Reviews (3): Last reviewed commit: "test(step_delegating_executor): add heal..." | Re-trigger Greptile}

[futurewasfree]

trying you again @dpeng817 to get attention. maybe you could add relevent person from the team?

[smackesey]

Thanks for the fix @futurewasfree — the diagnosis is right, and the guard is in the correct place to close this race. Apologies for the slow turnaround.

I pushed 4ecfb5c to your branch (it has "allow edits by maintainers" enabled) to scope the guard to records the event tailer hasn't consumed yet (record.storage_id not in seen_storage_ids). Without that scoping, including STEP_UP_FOR_RETRY introduces a hang: a step that failed attempt 1 with a retry policy leaves a consumed STEP_UP_FOR_RETRY record in the event log, and if attempt 2's k8s job then genuinely dies without writing an event, the stale record would suppress the health-check failure on every polling round — nothing would ever remove the step from running_steps, so the run would hang indefinitely. Restricting the check to unconsumed records makes it match exactly the "already in the DB, not yet tailed" window this PR targets, for both event types.

The remaining gap before this can merge is test coverage. The right spot is python_modules/dagster/dagster_tests/execution_tests/engine_tests/test_step_delegating_executor.py, with two cases: (1) health check unhealthy + unconsumed STEP_SUCCESS in the log → no STEP_FAILURE emitted; (2) health check unhealthy on a retried step whose only matching record is the consumed STEP_UP_FOR_RETRY from the prior attempt → failure still emitted. If you don't have bandwidth for that, let us know.

[futurewasfree]

The remaining gap before this can merge is test coverage. The right spot is python_modules/dagster/dagster_tests/execution_tests/engine_tests/test_step_delegating_executor.py, with two cases: (1) health check unhealthy + unconsumed STEP_SUCCESS in the log → no STEP_FAILURE emitted; (2) health check unhealthy on a retried step whose only matching record is the consumed STEP_UP_FOR_RETRY from the prior attempt → failure still emitted. If you don't have bandwidth for that, let us know.

Could you take a look again? :) @smackesey