Skip to content

fix(cache): recompute tasks with missing local outputs - #3469

Open
1fanwang wants to merge 1 commit into
flyteorg:masterfrom
1fanwang:1fannnw/refresh-missing-local-cache
Open

1fanwang wants to merge 1 commit into
flyteorg:masterfrom
1fanwang:1fannnw/refresh-missing-local-cache

Conversation

@1fanwang

Copy link
Copy Markdown
Contributor

Tracking issue

Related to flyteorg/flyte#6393.

Why are the changes needed?

A local workflow can fail on its second run after a consumer renames a cached output file. The cache returns the old path, and input conversion raises TypeTransformerFailedError: Error encountered while converting inputs of 'repro_6393.process_file':.

What changes were proposed in this pull request?

Treat missing local artifact references as cache misses so the producer runs again. Valid cached results and remote references keep their existing behavior. Other filesystem errors still propagate.

This does not isolate in-place edits or inspect paths embedded in opaque serialized values.

How was this patch tested?

The same workflow and SQLite cache produced these results:

Checkout Result
Upstream, first run The producer ran and the consumer renamed its file.
Upstream, second run Input conversion failed before the consumer ran.
Patched, same cache The producer ran again and the workflow completed.

Setup process

E2E

I used Python 3.12.9 with the Flytekit development environment. Save this as repro_6393.py in both checkouts:

import os
import tempfile
from flytekit import Cache, FlyteFile, task, workflow

@task(cache=Cache(version="cache-artifact-probe-1"))
def create_file() -> FlyteFile:
    with tempfile.NamedTemporaryFile(mode="wb", delete=False, dir=os.environ["HOME"]) as file:
        file.write(b"Hello, world!")
        print("PRODUCER_EXECUTED")
        return FlyteFile(path=file.name)

@task
def process_file(file: FlyteFile) -> None:
    path = file.download()
    with open(path, "rb") as source:
        assert source.read() == b"Hello, world!"
    os.rename(src=path, dst=path + ".processed")
    print("CONSUMER_RENAMED_FILE")

@workflow
def wf() -> None:
    process_file(file=create_file())

Keep the same HOME across all three runs:

export HOME="$(mktemp -d /tmp/flyte-cache-XXXXXX)"
export PYTHONPATH=.
export FLYTE_SDK_LOGGING_LEVEL=30
pyflyte run repro_6393.py wf > result.log 2>&1
rc=$?
grep -A 1 -E '^(PRODUCER_EXECUTED|CONSUMER_RENAMED_FILE|TypeTransformerFailedError)' result.log
printf 'exit_code=%s\n' "$rc"

Run twice on upstream, then once on the patched checkout without clearing the cache.

Raw output

Upstream, first run:

PRODUCER_EXECUTED
CONSUMER_RENAMED_FILE
exit_code=0

Upstream, second run:

TypeTransformerFailedError: Error encountered while converting inputs of
'repro_6393.process_file':
exit_code=1

Patched, same cache:

PRODUCER_EXECUTED
CONSUMER_RENAMED_FILE
exit_code=0

Screenshots

Not applicable.

Check all the applicable boxes

  • I updated the documentation accordingly.
  • All new and existing tests passed.
  • All commits are signed-off.

Related PRs

None.

Docs link

The cache lookup docstring describes the missing-artifact behavior.

Signed-off-by: 1fanwang <1fannnw@gmail.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant