[backfill daemon run retries 2/n] backfill daemon incorporates retries runs when launching new runs

[jamiedemaria]

Summary & Motivation

The backfill daemon doesn't account for run retries. See https://github.com/dagster-io/internal/discussions/12460 for more context. We've decided that we want the daemon to account for automatic and manual retries of runs that occur while the backfill is still in progress. This requires two changes: ensuring the backfill isn't marked completed if there is an in progress run or a failed run that will be automatically retried; and updating the daemon to take the results of retried runs into account when deciding what partitions to materialize in the next iteration.

This PR addresses the second point, updating the backfill daemon to take the results of retried runs into account when deciding what partitions to materialize in the next iteration.

Currently the backfill gets a list of the successfully materialized assets for the backfill by looking at the materialization events for the asset. It determines which assets failed by looking at the failed runs launched by the backfill and pulling the asset partition information from those runs. Any assets downstream of those failed assets will not be launched by the backfill

Now that we want the backfill daemon to account for run retries we need to slightly modify this logic. Since a run can be retried it is possible that an asset can have a successful materialization AND be a failed asset in a failed run. This means that when we determine which assets are failed, we need to cross check with the assets that have been successfully materialized and remove any that are in the materialized list

How I Tested These Changes

new unit tests

manual test

@dg.asset(
    partitions_def=daily_partitions_def,
)
def has_retry_policy(context):
    if context.run.parent_run_id is None:
        raise Exception("I failed")
    else:
        return 1

@dg.asset(partitions_def=weekly_partitions_def)
def another_weekly(context, has_retry_policy):
    return 1

run a backfill of at least 1 week. Can watch the backfill progress and see that once all of the daily assets retry and succeed, the weekly asset will get kicked off. This contrasts with the behavior in the PR down the stack where the daily assets would retry and succeed, but the weekly asset would never get kicked off

Changelog

Manual and automatic retries of runs launched by backfills that occur while the backfill is still in progress are now incorporated into the backfill's status

[jamiedemaria]

• [backfill daemon run retries 3/n] retries of runs in completed backfills should not be considered part of the backfill #25900
• [backfill daemon run retries 2/n] backfill daemon incorporates retries runs when launching new runs #25853 👈 (View in Graphite)
• [backfill daemon run retries 1/n] update how we determine backfill completion to account for retried runs #25771
• [auto run retry tags] 3/n - when a run is retried, add theauto_retry_run_id tag #26054
• [auto run retry tags] 2/n - Add dagster/will_retry tag to runs that fail #25932
• [auto run retry tags] 1/n - make run_retries retry_on_asset_or_op_failure setting available on instance #26046
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.

[graphite-app]

There appears to be a scoping issue with asset_partition_candidates. The code creates these candidates for partition ranges but never uses them because the result.extend(asset_partitions_still_failed) call is outside both branches of the if/else. To fix this:

1. Move the asset_partition_candidates assignment inside the for loop in the first branch
2. Move result.extend(asset_partitions_still_failed) inside each branch of the if/else

This ensures both partition ranges and single partitions are properly filtered against the materialized subset before being added to the result.

Spotted by Graphite Reviewer

Is this helpful? React 👍 or 👎 to let us know.

[jamiedemaria]

Want to confirm that this is a safe change to make. From my reading of _get_failed_asset_partitions it returns the full list of failed partitions, not a list of partitions that failed since the last tick, so in the version of this code before this PR the ORing of the two subsets is a no-op since asset_backfill_data.failed_and_downstream_subset would be a subset of failed_subset

with the change to have _get_failed_asset_partitions account for retries, ORing with asset_backfill_data.failed_and_downstream_subset would result in inaccurate data because a failed partition in asset_backfill_data.failed_and_downstream_subset could have been successfully retried and no longer in failed_subset but would still be included because of the OR operation

[jamiedemaria]

ah ok i forgot that we do the second function to get the downstream of the failed subset, so this isn't a simple replacement since failed_subset doesn't include downstream assets. i will update

[gibsondan]

I think @clairelin135 is more familiar about the ins and outs of this particular codepath than I, I do have one question though

[gibsondan]

i'm not super familiar with this code but how is the 'and downstream' part of 'failed_and_downstrema_subset' affected by the retry case? If we add downstream assets as well to the failed list as soon as we detect a partition fails, do we need to remove them if that run is retried? Or are those added at the very end after we have decided that the backfill is totally done and failed?

[jamiedemaria]

so this specifically is for the case when the backfill has been canceled.

In the normal code path what we do is:

• find which partitions failed
• do a BFS search to find all downstream partitions of the failed partitions
• set that as the failed_and_downstream_subset
• when determining if we can launch a run for a partition, if it is in the failed_and_downstream_subset we know we cannot launch the run

we run this process from scratch on each backfill iteration, so we can handle run retries in the first step by checking if each failed partition has been retried successfully, and if so we remove it from the list of failed partitions. then all the following steps will produce the correct results

for this cancling case it's a bit different. the way it works now is that we take the previous failed_and_downstream_subset and just add any partitions that have failed (effectively this just adds the partitions that failed since the last tick). we don't recalculate the downstream subset for any newly failed partitions (i'm not sure why, maybe @clairelin135 knows).

If we leave the code as is, we could have a case where a failed partition was retried and that retry succeeded since the previous tick. the partition would be in the failed_and_downstream_subset from the previous tick, but is now successful. we want to make sure the right status is reported, so we need to remove it from the final set of failed partitions. and we can do that be removing the materialized_subset from the failed_and_downstream_subset

Another option would be to do the process we do in a normal tick and generate the failed_and_downstream_subset from scratch rather than do this additive approach. Not sure why that wasn't done in the first place though, so i didn't make that change

[clairelin135]

we don't recalculate the downstream subset for any newly failed partitions

I think we added this for performance reasons, if a backfill is canceling we do the cheapest thing possible which is to mark every requested partition as failed. We don't bother doing the graph traversal to mark their downstreams as failed.

[clairelin135]

The changes here in the canceling iteration look good to me

[clairelin135]

The changes here look good to me.

Initially I was wondering if we could avoid moving a partition to the failed subset until all of its retries have concluded, to simplify the state transitions here. But since we're incorporating manual retries we need to be able to move partitions and their downstreams between the failed / materialized subsets regardless.

[clairelin135]

we don't recalculate the downstream subset for any newly failed partitions

I think we added this for performance reasons, if a backfill is canceling we do the cheapest thing possible which is to mark every requested partition as failed. We don't bother doing the graph traversal to mark their downstreams as failed.

[clairelin135]

The changes here in the canceling iteration look good to me

[clairelin135]

We should make sure that if the backfill is in a canceling state, retried runs are not included in the backfill