new cache_key from cache_key_fn not invalidating cached results

[N-Demir]

First check

• I added a descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the Prefect documentation for this issue.
• I checked that this issue is related to Prefect and not one of its dependencies.

Bug summary

I'm pulling my hair out because this seems like such trivial functionality to not be working right. I have a custom cache_key_fn to allow me to invalidate the cache by updating an entry in a sql database (ala all the issues that complain about that functionality missing: #10494). The details of my cache_key_fn don't matter but basically I can force the cache_key into being brand new values thereby invalidating the cache.

I am running a flow locally that was run as part of a deployment remotely. When I first start, 10 out of 11 tasks in the flow had run successfully remotely and were cached, so only the single one that didn't runs locally and succeeds. So far so good, I have 11/11 tasks in this flow successfully run and cached. However, I now invalidate the cache via the above method so that I can rerun all the tasks locally and check for their success with some new code changes..... only the 1 that locally was run previously has its cache invalidated. Somehow the other 10/11 that were cached from the remote run earlier are still using cached values? And I've verified that the returned cache_key is brand new!!

So there's obviously some disconnect between the cache_key changing and the cache actually being invalid. How is it even possible for prefect to find the cached values if the cache_key is different?

The particular line where I see it pull the cached value instead of setting the state to running is in prefect/engine.py:propose_state line 2536 where the response in my case is:

OrchestrationResult(state=Cached(message=None, type=COMPLETED, result=PersistedResult(type='reference', artifact_type=None, artifact_description=None, serializer_type='pickle', storage_block_id=UUID('ce1306c0-6e80-402f-9047-17ef7b5f8273'), storage_key='59902298dd55441ba4d53919b856a72d')), status=SetStateStatus.REJECT, details=StateRejectDetails(type='reject_details', reason='Retrieved state from cache'))

What the heck is going on prefect?

Reproduction

-

Error

No response

Versions

2.x

Additional context

No response

[zzstoatzz]

hi @N-Demir - thanks for the issue, we appreciate you expressing your frustration.

We'd like to improve the experience around invalidating caches in general - it would be helpful if you could populate the reproduction section with an MRE and version section with the output of prefect version so its clear what situation you're speaking of. As is, it's not immediately clear to me how this differs significantly from #10494