Short answer: don’t rely on Prefect alone for this exact requirement. Because your subflows are running in Kubernetes, they may be executing in different processes / different pods. In Prefect 3.6.12, there is no built-in per-parent-flow-run mutex for “only one subflow from this specific flow run may enter this critical section at a time”. What is built in: - Tag-based concurrency limits — useful, but generally global/shared unless you create dynamic limits - Global concurrency limits via

prefect.concurrency.*

— also global by name - Lock managers in `prefect.locking`: -

MemoryLockManager

-

FileSystemLockManager

Those built-in lock managers are not a good fit for Kubernetes-distributed mutual exclusion unless all writers share the same filesystem. So for your case, the safest pattern is: Use an external distributed lock keyed by the parent flow run id + S3 object key. That gives you exactly this scope: - subflows from the same run of flow A serialize access - subflows from different runs of flow A do not block each other A lock backend could be Redis, DynamoDB, Postgres advisory locks, etc. Example pattern

from contextlib import contextmanager
from prefect import flow, task
from prefect.context import get_run_context
import time

# Pseudocode: replace with Redis / DynamoDB / etc.
@contextmanager
def distributed_lock(lock_name: str, timeout: int = 60):
    acquired = False
    try:
        # acquire lock here
        # e.g. SETNX in Redis with TTL
        acquired = True
        if not acquired:
            raise RuntimeError(f"Could not acquire lock: {lock_name}")
        yield
    finally:
        if acquired:
            # release lock here
            pass

@task
def write_shared_s3_file(parent_flow_run_id: str, s3_key: str, payload: str):
    lock_name = f"s3-write:{parent_flow_run_id}:{s3_key}"

    with distributed_lock(lock_name, timeout=120):
        # read/modify/write critical section
        # boto3 put/get here
        time.sleep(5)

@flow
def child_flow(parent_flow_run_id: str, s3_key: str, payload: str):
    write_shared_s3_file(parent_flow_run_id, s3_key, payload)

@flow
def flow_a():
    ctx = get_run_context()
    parent_flow_run_id = str(ctx.flow_run.id)
    s3_key = f"some/path/{parent_flow_run_id}.json"

    child_flow.submit(parent_flow_run_id, s3_key, "a")
    child_flow.submit(parent_flow_run_id, s3_key, "b")
    child_flow.submit(parent_flow_run_id, s3_key, "c")

A few important notes: 1. Pass the parent flow run ID explicitly Inside a subflow,

get_run_context().flow_run.id

is the subflow run’s ID, not the parent’s. So if you want all sibling subflows to coordinate, pass the parent ID down explicitly. 2. Scope the lock to the parent run Use a lock key like:

s3-write:{parent_flow_run_id}:{s3_key}

That way: - same parent run + same file => serialized - different parent runs => independent 3. S3 itself is not a lock manager If multiple writers do read-modify-write on the same object, you can still get lost updates unless you protect that section with a real distributed lock or redesign the write pattern. 4. Prefer “single writer” if possible If you can change the design, an even better pattern is: - subflows write their outputs to separate S3 objects - parent flow does one final aggregation write That avoids distributed locking entirely. About Prefect concurrency limits You could try dynamic concurrency limits, for example one limit name per parent flow run, but that’s usually clunky operationally because you’d need to create/manage those limits dynamically. Also, concurrency limits are generally better for throttling than for implementing a business-critical distributed mutex around a shared file write. If you want to explore those APIs anyway, relevant areas are: -

prefect.concurrency.sync.concurrency

-

prefect.concurrency.asyncio.concurrency

-

prefect.context.tags

-

prefect.locking.*

Docs:

- Tag-based concurrency limits - Global concurrency limits - Flows and subflows Recommendation For Kubernetes + parallel subflows + shared S3 object: - Best practical solution: external distributed lock keyed by parent flow run ID - Best architectural solution: avoid shared writes; have subflows write separate objects and merge once If you want, I can give you a concrete implementation using either: - Redis - DynamoDB (nice if you’re already on AWS) - Postgres advisory locks

Yes — if you have Redis available, using a Prefect transaction with a Redis lock manager is a reasonable option. A couple of important specifics for Prefect 3.6.12: -

transaction(...)

does not take a

lock_manager=

argument directly - instead, the lock manager is attached to a

ResultStore

- then you pass that

ResultStore

as

store=...

to

transaction(...)

- for distributed locking with Redis, you’d use

RedisLockManager

from the

prefect-redis

integration, not core

prefect

So the shape is:

transaction(..., store=ResultStore(lock_manager=RedisLockManager(...)))

Relevant APIs: -

prefect.transactions.transaction

-

prefect.transactions.atransaction

-

prefect.locking.protocol.LockManager

-

prefect_redis.RedisLockManager

A sketch:

from prefect import flow, task
from prefect.transactions import transaction, IsolationLevel
from prefect.results import ResultStore
from prefect_redis import RedisLockManager

lock_manager = RedisLockManager(host="redis", port=6379)
store = ResultStore(lock_manager=lock_manager)

@task
def write_shared_file(parent_flow_run_id: str, s3_key: str, payload: str):
    txn_key = f"s3-write:{parent_flow_run_id}:{s3_key}"

    with transaction(
        key=txn_key,
        store=store,
        isolation_level=IsolationLevel.SERIALIZABLE,
    ):
        # critical section
        # read/modify/write your S3 object here
        pass

But there’s a subtle caveat: Prefect transactions are primarily designed around result storage / staged commit semantics, not as a general-purpose “mutex block” feature. So this can work, but whether it’s the best tool depends on what you mean by “write to the same S3 file”: Good fit - you want a distributed lock around a critical section - you’re okay using the transaction key as the mutex key - you want sibling subflows in Kubernetes to serialize on that key Less ideal fit - you just need a simple distributed mutex and don’t care about transaction/result semantics - your S3 update is custom read/modify/write logic unrelated to Prefect result persistence In that second case, a plain Redis lock is often simpler and clearer than wrapping the code in a Prefect transaction. For your specific requirement, I’d recommend: Option A — simplest and clearest: use a plain Redis lock

lock_key = f"s3-write:{parent_flow_run_id}:{s3_key}"

Option B — if you want to stay inside Prefect’s transaction model: use

transaction(
    key=lock_key,
    store=ResultStore(lock_manager=RedisLockManager(...)),
    isolation_level=IsolationLevel.SERIALIZABLE,
)

Two more gotchas: 1. Use the parent flow run id, not the subflow run id If the lock should be shared across sibling subflows, pass the parent flow run ID explicitly to each subflow/task. 2. Make the transaction key specific enough Use something like:

s3-write:{parent_flow_run_id}:{s3_key}

Otherwise unrelated writes may block each other. If you want, I can give you a full working example for Prefect 3.6.12 using: -

prefect-redis

+

transaction(...)

, or -

prefect-redis

+ a direct Redis mutex without transactions Also, if you want to install the integration, see: - prefect-redis docs - Prefect transactions