Short answer: not by default. In Prefect 3.x, task-level caches are in-memory per process unless you configure shared storage—so they won’t automatically persist or be reusable across different work pools or workers. How to make caches reusable across workers/pools - Persist the task result: set

persist_result=True

and point

result_storage

to shared storage (e.g., S3, GCS, Azure blob, NFS). - Persist the cache key/index: provide a

cache_policy

with

key_storage

set to a shared filesystem path (e.g., S3/GCS/Azure/NFS). This is what lets any worker find the existing result by cache key. - Optional (recommended for distributed workers): set a stricter isolation level and a distributed lock to prevent duplicate work. -

isolation_level="SERIALIZABLE"

with a

LockManager

that all workers can reach (e.g., filesystem-based on shared NFS, or a Redis-based lock manager via the prefect-redis integration). Example

from datetime import timedelta
from prefect import task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import RemoteFileSystem
# If you want distributed locking, use a LockManager reachable by all workers

shared_cache_policy = CachePolicy(
    key_storage=RemoteFileSystem(basepath="<s3://my-bucket/prefect/cache>"),
    # isolation_level="SERIALIZABLE",
    # lock_manager=...  # e.g., a Redis-based lock manager
)

shared_results = RemoteFileSystem(basepath="<s3://my-bucket/prefect/results>")

@task(
    persist_result=True,
    result_storage=shared_results,
    cache_policy=shared_cache_policy,
    cache_expiration=timedelta(days=1),
)
def expensive_task(x):
    ...

Key points - Worker pools are just a scheduling boundary; cache reuse depends on shared storage and cache keys, not the pool itself. - The cache key must match. If your cache policy includes the task source, input values, etc., changing code or inputs will produce a different key and skip reuse. - The default cache is an in-process LRU cache, so it’s not reusable across processes, workers, or pools. Docs - Caching in Prefect 3.x: Caching (Prefect 3.x) - Redis integration (for distributed locks/key stores): prefect-redis integration If you share your environment (e.g., S3 vs GCS vs NFS) I can tailor the exact setup for your workers and work pools.

@Marvin my environment is docker self hosted (two worker pools of type docker, with workers polling each pool and creating containers to run the actual flows). I understand that the cache key would need to be unique and inferred as the same for the task in flows ran across both pools. I am more asking about how to set up the persistence from a parameter point of view in the task decorator, as well as what i would assume would be using volume to share data across the flow containers spun up by the workers polling each pool (or if there is another, better solution)

Great context — with Docker workers spinning up flow containers, you’ve got two solid patterns to make task caches reusable across both pools: Option A (recommended): use shared remote storage (S3/GCS/Azure) for both results and cache keys - Pros: no container-to-container volume wiring, works across hosts and pools, easy to scale. - How: - Persist task results to shared storage - Persist the cache key/index to shared storage - Optionally use a distributed lock to avoid duplicate work while a cache entry is being built Example

from datetime import timedelta
from prefect import task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import RemoteFileSystem
# For distributed locking, see the prefect-redis integration docs

shared_results = RemoteFileSystem(basepath="<s3://my-bucket/prefect/results>")
shared_cache = RemoteFileSystem(basepath="<s3://my-bucket/prefect/cache>")

@task(
    persist_result=True,
    result_storage=shared_results,
    cache_policy=CachePolicy(
        key_storage=shared_cache,
        # isolation_level="SERIALIZABLE",
        # lock_manager=...  # e.g., a Redis-based lock manager
    ),
    cache_expiration=timedelta(days=1),
)
def expensive_task(x):
    ...

- Ensure your Docker image/containers have credentials to access the storage (e.g., AWS env vars). - If you want to avoid duplicate “first writer wins” races when the same cache key hits both pools at once, set

isolation_level="SERIALIZABLE"

and supply a distributed

lock_manager

(e.g., Redis). See prefect-redis integration. Option B: use a shared Docker volume (NFS/EFS/host bind) mounted into every flow container - Pros: simple if all containers run on the same host or a shared NFS. - How: - Mount the same host path into every flow container in both pools at the same container path, e.g.,

/mnt/prefect

- Point result storage and cache key storage at directories inside that mount - Use filesystem locks to prevent duplicate work Example

from pathlib import Path
from datetime import timedelta
from prefect import task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import LocalFileSystem
from prefect.locking.filesystem import FileSystemLockManager

shared_results_fs = LocalFileSystem(basepath="/mnt/prefect/results")
shared_cache_fs = LocalFileSystem(basepath="/mnt/prefect/cache")
lock_manager = FileSystemLockManager(lock_files_directory=Path("/mnt/prefect/locks"))

@task(
    persist_result=True,
    result_storage=shared_results_fs,
    cache_policy=CachePolicy(
        key_storage=shared_cache_fs,
        isolation_level="SERIALIZABLE",
        lock_manager=lock_manager,
    ),
    cache_expiration=timedelta(days=1),
)
def expensive_task(x):
    ...

Mounting the volume into flow containers - Prefect’s Docker work pool uses a base job template that defines container settings (image, env, volumes, etc.). You can: 1) Inspect the current template:

prefect work-pool inspect <your-docker-pool>

2) Save and edit the template JSON to include a “volumes” variable and wire it to the Docker run configuration if it isn’t there already. 3) Update the pool:

prefect work-pool create "<your-docker-pool>" --type docker --base-job-template ./docker-template.json --overwrite

4) Set the volumes via job variables on your deployments (or pool defaults). Typical format is Docker-style bind mounts (e.g.,

"/srv/prefect:/mnt/prefect:rw"

), but confirm the exact field name/shape in your pool’s template schema shown by

inspect

. Notes and gotchas - Worker pools are just scheduling boundaries; cache reuse depends on shared storage and consistent cache keys. - The default cache is in-memory per process — not shared — so you must configure storage as shown above. - If code or inputs change, the cache key will change; only equal keys reuse results. Customize

cache_key_fn

if you need tighter control.

- If your two Docker pools run on different hosts, Option A (remote object storage) is usually simpler than maintaining a cross-host shared volume. Docs for deeper reference - Caching in Prefect 3.x: Caching (Prefect 3.x) - Redis integration (optional distributed locks): prefect-redis integration If you share: - whether the two pools run on the same host or multiple hosts - your preferred storage (S3/GCS/Azure vs NFS/host bind) - the output of

prefect work-pool inspect <pool>

…I can give you exact

@task

settings and a precise base job template/job_variables snippet for your setup.

Thanks — with two Docker pools on the same host and a shared volume, you’re in great shape. What you need to persist for cross-container reuse - Two things matter for cross-process caching: 1) The cache index (mapping from cache key -> stored result reference) 2) The result itself - If you only persist the cache index but don’t persist the result, another container can find the key but won’t be able to load the value. So set both: -

cache_policy.key_storage

to your shared volume -

persist_result=True

and

result_storage

to your shared volume Do I need multiple filesystems? - Not strictly. You can: - Use one LocalFileSystem for both key storage and results (they’ll mix in one directory), or - Use two LocalFileSystem instances pointing at different subdirectories under the same mount. This is cleaner and lets you clear cache without touching results. - Functionally, both work. Operationally, separate subdirs is nicer. Is /mnt required? - No. It’s just a common Linux convention. Use any consistent container path you’ve mounted, e.g.

/data/prefect

or

/srv/prefect

. The key is: every flow container in both pools should mount the same host path to the same container path. Minimal, clean setup on a single shared mount - Assume you mounted the host path to

/srv/prefect

inside each flow container.

from datetime import timedelta
from pathlib import Path

from prefect import task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import LocalFileSystem
from prefect.locking.filesystem import FileSystemLockManager

# Separate subdirs under the same shared mount
results_fs = LocalFileSystem(basepath="/srv/prefect/results")
cache_fs = LocalFileSystem(basepath="/srv/prefect/cache")
locks_dir = Path("/srv/prefect/locks")

@task(
    persist_result=True,                 # store results so other containers can read them
    result_storage=results_fs,           # write results to shared volume
    cache_policy=CachePolicy(
        key_storage=cache_fs,            # write cache index to shared volume
        isolation_level="SERIALIZABLE",  # optional but recommended to avoid duplicate work
        lock_manager=FileSystemLockManager(lock_files_directory=locks_dir),
    ),
    cache_expiration=timedelta(days=1),
)
def expensive_task(x):
    # your work here
    return ...

Notes and tips - You can point both

result_storage

and

key_storage

at the same LocalFileSystem (same basepath) if you want to keep it ultra-simple; I still recommend separate subfolders to avoid clutter and simplify maintenance. - The lock manager + SERIALIZABLE helps when both pools receive the same cacheable task at the same time — only one will compute; the other will wait and reuse. -

cache_result_in_memory=True

(default) only helps within a single process; it doesn’t replace persisted storage. - Make sure your pool’s base job template or job variables mount the shared host directory into the container at the exact path you use above; you’ve said that’s already done. Docs for reference - Caching in Prefect 3.x: Caching (Prefect 3.x) - Redis option (not required in your single-host setup, but useful for distributed locks): prefect-redis integration If you share the container path you’ve mounted (e.g.,

/srv/prefect

vs something else), I can tweak the snippet exactly to your tree.

Yes. The easiest and safest way to “refresh” a specific cached task is to call that task again with refresh_cache=True for the same inputs. This bypasses the cache for that run, recomputes the value, and then updates the persisted cache so future runs will use the fresh result. Key points - Use the same inputs and the same cache key function so you hit the same cache key. - Do this at the end of your flow (or in a maintenance flow) to warm the cache for next runs. - You can do it selectively for just the items you want to refresh. Pattern - Keep your normal cached task as-is. - At the end of the flow, call a “refresh” variant of the same task with refresh_cache=True. Example (continuing your local filesystem setup)

from datetime import timedelta
from pathlib import Path

from prefect import flow, task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import LocalFileSystem
from prefect.locking.filesystem import FileSystemLockManager

results_fs = LocalFileSystem(basepath="/srv/prefect/results")
cache_fs = LocalFileSystem(basepath="/srv/prefect/cache")
locks_dir = Path("/srv/prefect/locks")

@task(
    persist_result=True,
    result_storage=results_fs,
    cache_policy=CachePolicy(
        key_storage=cache_fs,
        isolation_level="SERIALIZABLE",
        lock_manager=FileSystemLockManager(lock_files_directory=locks_dir),
    ),
    cache_expiration=timedelta(days=1),
)
def expensive_task(x):
    # heavy work
    return f"result for {x}"

# A “refresh” view of the same task
refresh_expensive_task = expensive_task.with_options(refresh_cache=True)

@flow
def my_flow(items):
    # normal cached calls
    results = [expensive_task.submit(i) for i in items]
    # ... do the rest of your flow

    # At the end, refresh selectively for next runs
    to_refresh = [items[0]]  # pick whatever keys you want to refresh
    for i in to_refresh:
        # submit refresh; this recomputes and updates the persisted cache
        refresh_expensive_task.submit(i)

Notes - You can refresh multiple keys by looping and calling the refresh variant with each input. - If you don’t need an immediate refresh, you can rely on

cache_expiration

so keys expire naturally. - If you rarely need to refresh, you can place the refresh logic in a separate “maintenance” flow that only runs when you want to force updates. Less recommended alternatives - Manually deleting cache files from the shared volume is possible but brittle; the file layout is an internal detail and easy to get wrong. - Permanently setting

refresh_cache=True

on the decorator forces recompute every time (usually not what you want). Docs - Caching in Prefect 3.x: Caching (Prefect 3.x) If you share how you choose which items to refresh (e.g., all, a filtered subset, time-based), I can sketch the exact refresh loop or a small maintenance flow for your case.

Great question. Here’s how

refresh_cache=True

behaves for a task in Prefect 3.x: Short answers to your 1/2/3: 1) Does it empty the current cache? No. It bypasses the cache read, recomputes the task, and then overwrites/updates the persisted cache entry for that key. It does not “delete” the cache first; it replaces it. 2) Does it just refresh the expiry on the same data? Not as a “touch.” The expiration gets reset because a new record is written after recomputation. There’s no “extend TTL without recompute” operation. 3) Does it recompute and persist but not use the result in this run? By default, no — the call returns the newly computed result like any task call. If you don’t want to use it in the current flow, call the refresh at the end (or in a maintenance flow) and ignore the returned future so it doesn’t affect earlier downstream logic. A practical “refresh at end” pattern - Use the cached task as usual early in the flow. - After your downstream work is done, call a “refresh” view of the same task to rebuild the cache for next time, and discard the result.

from datetime import timedelta
from pathlib import Path

from prefect import flow, task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import LocalFileSystem
from prefect.locking.filesystem import FileSystemLockManager

results_fs = LocalFileSystem(basepath="/srv/prefect/results")
cache_fs = LocalFileSystem(basepath="/srv/prefect/cache")
locks_dir = Path("/srv/prefect/locks")

@task(
    persist_result=True,
    result_storage=results_fs,
    cache_policy=CachePolicy(
        key_storage=cache_fs,
        isolation_level="SERIALIZABLE",  # optional but helps avoid duplicate work
        lock_manager=FileSystemLockManager(lock_files_directory=locks_dir),
    ),
    cache_expiration=timedelta(days=1),
)
def expensive_task(x):
    # heavy work
    return f"result for {x}"

# A “refresh” view that bypasses read and recomputes
refresh_expensive_task = expensive_task.with_options(refresh_cache=True)

@flow
def my_flow(items):
    # Use the cached task for normal work
    results = [expensive_task.submit(i) for i in items]
    # ... downstream work using `results` ...

    # At the end, rebuild cache for a subset or all items for next runs
    for i in items[:2]:  # choose which keys to refresh
        _ = refresh_expensive_task.submit(i)  # recompute + update cache; ignore output

Notes - If concurrent runs might hit the same key while you refresh, using

isolation_level="SERIALIZABLE"

with

FileSystemLockManager

ensures one writer and consistent reads. - There isn’t a first-class “invalidate this one cache entry without recompute” API. Manual deletion of specific cache files is possible but brittle; recomputing with

refresh_cache=True

is the safer route. - If you need to force new keys (e.g., schema change) you can add a salt/version into your

cache_key_fn

so old and new entries don’t collide. Docs for reference - Caching in Prefect 3.x: Caching (Prefect 3.x) If you want, I can show how to target a specific cache key (via a custom

cache_key_fn

) so you can refresh exactly the entries you want.