Backfills can complete before a failed run’s retry decision is published #34204

Codex (GPT-6):

What is the issue?

An asset backfill can be considered complete after a run's FAILURE status is stored but before its dagster/will_retry decision is published. This is an ordinary publication race; no process crash is required.

handle_new_event updates run storage before calculating and storing WILL_RETRY_TAG. Meanwhile, backfill_is_complete only queries failed runs already tagged WILL_RETRY_TAG=true. If completion runs between those operations, it misses a pending automatic retry. If the backfill becomes terminal, the automatic retry handler subsequently rejects that backfill.

Sources: failure publication, backfill completion, and retry eligibility. The protection introduced in PR #25771 appears intended to prevent this premature completion, but the publication window remains.

Expected behavior

With automatic run retries enabled, completion should wait while a failed run's retry decision is still being published. An explicit false decision and disabled automatic retries should retain their existing completion behavior.

Reproduction

Save the script below as /tmp/dagster_retry_publication_repro.py and run:

uv run --no-project --isolated --python 3.14 --with 'dagster==1.13.13' python /tmp/dagster_retry_publication_repro.py

It intercepts the native policy calculation after the real failure status has been persisted, evaluates completion in that window, then lets Dagster finish publishing the decision. No retry tag is manually written. The target asset is marked failed in the backfill data so the other completion condition is satisfied.

Observed output on the unmodified PyPI package:

During publication: status=FAILURE, will_retry=None, complete=True
After publication: status=FAILURE, will_retry=true, complete=False

Standalone reproduction

import logging
import time
from unittest.mock import patch

import dagster as dg
from dagster._core.definitions.assets.graph.asset_graph_subset import AssetGraphSubset
from dagster._core.execution.asset_backfill import AssetBackfillData, backfill_is_complete
from dagster._core.storage.tags import BACKFILL_ID_TAG, WILL_RETRY_TAG


def main():
    backfill_id = "retry-publication-example"
    target = AssetGraphSubset(non_partitioned_asset_keys={dg.AssetKey("example")})
    with dg.instance_for_test(
        overrides={"run_retries": {"enabled": True, "max_retries": 1}}
    ) as instance:
        data = AssetBackfillData.empty(target, time.time(), instance)
        data = data._replace(failed_and_downstream_subset=target)
        run = instance.add_run(
            dg.DagsterRun(job_name="example", tags={BACKFILL_ID_TAG: backfill_id})
        )
        should_retry = instance._should_retry_run
        observations = []

        def inspect_before_retry_decision(failed_run, failure_reason):
            stored = instance.get_run_by_id(run.run_id)
            assert stored is not None
            assert stored.status == dg.DagsterRunStatus.FAILURE
            assert WILL_RETRY_TAG not in stored.tags
            complete = backfill_is_complete(
                backfill_id, data, instance, logging.getLogger("repro")
            )
            observations.append(complete)
            print(f"During publication: status={stored.status.value}, "
                  f"will_retry={stored.tags.get(WILL_RETRY_TAG)}, complete={complete}")
            return should_retry(failed_run, failure_reason)

        with patch.object(instance, "_should_retry_run", inspect_before_retry_decision):
            instance.report_run_failed(run)

        stored = instance.get_run_by_id(run.run_id)
        assert stored is not None
        assert stored.tags[WILL_RETRY_TAG] == "true"
        complete = backfill_is_complete(
            backfill_id, data, instance, logging.getLogger("repro")
        )
        print(f"After publication: status={stored.status.value}, "
              f"will_retry={stored.tags[WILL_RETRY_TAG]}, complete={complete}")
        assert observations == [True]
        assert complete is False


if __name__ == "__main__":
    main()

A separate four-case regression on unmodified upstream master c741afaeacf960dc58558a4c216fd8f50a6a8207 returned 1 failed, 3 passed: only retries-enabled plus a missing decision incorrectly completed. Existing explicit true, explicit false, and retries-disabled cases passed.

Version and deployment

Dagster 1.13.13, Python 3.14, local SQLite test instance. Also reproduced at the upstream master commit above. This report establishes the completion predicate/publication interleaving; it does not claim an observed production incident.

Proposed scope

A minimal completion guard can treat a missing decision as pending while automatic retries are enabled, without calculating retry policy or changing storage. If publication never finishes, that conservative guard can require manual recovery; automatic crash recovery and event replay are separate concerns. I am preparing a small PR with the guard and regression coverage.