[BUG] Incorrect output data when recovering partially completed dynamic workflows

[hamersaw]

Describe the bug

Dynamic workflows work by compiling a workflow closure defining subnodes which is then subsequently evaluated. In this process, Flyte assigns each subnode an incremental unique node ID which is used to track status, etc. Since these node IDs are incremental values, in scenarios where the dynamic task has some non-deterministic ordering a node ID in one execution of the dynamic workflow may not represent the same subnode as another execution.

This is problematic in partially completed dynamic workflows because the recovery mechanisms rely on node IDs to reuse previously completed executions. Under certain circumstances recovery of dynamic tasks will corrupt the result data.

Expected behavior

Recovery should work in all scenarios for dynamic workflows.

One solution is to ensure that the recovered dynamic workflow reuses the previously computed dynamic workflow closure rather than recomputing it. This means that node IDs would be the same for both the original execution and the recovered execution.

Additional context to reproduce

To reproduce run the following workflow with an input dict:

import time

from flytekit import Resources, dynamic, task, workflow
from typing import Dict

@task(requests=Resources(cpu="500m", mem="400Mi"), limits=Resources(cpu="1", mem="600Mi"))
def echo(a: int) -> int:
    time.sleep(10)
    return a

@task(requests=Resources(cpu="500m", mem="400Mi"), limits=Resources(cpu="1", mem="600Mi"))
def sleep(a: int, b: Dict[str, int]) -> Dict[str, int]:
    time.sleep(a)
    return b

@dynamic(cache=True, cache_version="1.0", requests=Resources(cpu="500m", mem="400Mi"), limits=Resources(cpu="1", mem="600Mi"))
def dynamic_dict(a: Dict[str, int]) -> Dict[str, int]:
    b = {}
    for x, y in a.items():
        b[x] = echo(a=y)

    return b

@workflow
def dynamic_dict_wf(a: Dict[str, int]) -> Dict[str, int]:
    b = dynamic_dict(a=a)
    c = sleep(a=10, b=b)
    return c

Then manually fail of the of the dynamic subnodes by repeatedly deleting the pod in k8s before it completes until Flyte exhausts the number of completions. Then use the 'Recover' button in the UI to start a workflow recovery. Analyze output values of the dynamic task to see incorrect data.

Screenshots

An example of the reproduce context. Launched the aforementioned workflow with inputs:

{
   "foo": 1,
   "bar": 2,
   "baz": 4,
}

Manually failed the n0-0-dn1 node to fail the workflow, the result is below:

In this execution nodes n0-0-dn0 and n0-0-dn2 correspond to the input values 4 and 2.


So this means the n0-0-dn1 should be responsible for 1 because the python iteration over this Dict was orderd 4,1,2.

Then we recover the workflow and allow it to succeed and see that the output values are:

{
   "bar": 2,
   "baz": 2,
   "foo": 4,
}

This is because the dynamic task recompiled the workflow closure which resulted in a different ordering of the Dict iteration. The results from n0-0-dn0 and n0-0-dn2 were reused - albeit applied to different keys (ie. bar & foo rather than baz and bar) and n0-0-dn1 was executed.

Are you sure this issue hasn't been raised already?

• Yes

Have you read the Code of Conduct?

• Yes

[hamersaw]

prematurely closed from PR merge

[hamersaw]

completed now