Short answer: Variables are not transactional. A

Variable.get()

followed by

Variable.set()

is not atomic in Prefect 3.x, so concurrent tasks can race. The simplest way to make the read-modify-write atomic across all workers is to wrap that code in a global concurrency limit with a single slot (a distributed mutex). Step 1 — create the global concurrency limit (one slot):

prefect gcl create variable:var --limit 1

Step 2 — wrap the critical section in your task: Sync task:

from prefect import task
from prefect.concurrency.sync import concurrency
from prefect.variables import Variable

@task
def example_task():
    with concurrency("variable:var", occupy=1, strict=True, timeout_seconds=60, max_retries=3):
        current = Variable.get("var")
        # compute some_new_value here
        Variable.set("var", value=some_new_value, overwrite=True)

Async task:

from prefect import task
from prefect.concurrency.asyncio import concurrency
from prefect.variables import Variable

@task
async def example_task():
    async with concurrency("variable:var", occupy=1, strict=True, timeout_seconds=60, max_retries=3):
        current = await Variable.aget("var")
        # compute some_new_value here
        await Variable.aset("var", value=some_new_value, overwrite=True)

Notes: -

strict=True

will fail fast if the slot cannot be acquired (instead of best-effort). Adjust to your preference. -

overwrite=True

ensures you update the existing variable without create/update ambiguity. - This works across all agents/workers in distributed environments (EKS, multiple pods, etc.). If you only need single-process protection (e.g., threads in one worker), you could also use Prefect’s transactions/locks, but those do not coordinate across multiple workers. For distributed safety, global concurrency limits are the recommended approach. Docs: - Variables concept - Global concurrency limits - How-to: apply global concurrency limits - Concurrency API (async/sync)

Great question — transactions are powerful, but they won’t make a Variable read/modify/write atomic across concurrent tasks. What Prefect transactions are good for - Idempotency and commit/rollback semantics for your own staged results - Mutual exclusion around a transaction key when you use SERIALIZABLE isolation plus a lock manager (memory/filesystem/Redis) What they are not - They do not participate in or wrap Prefect Variables.

Variable.get()

/

Variable.set()

do not run “inside” a transaction and are not coordinated by it. - They do not provide ACID-style atomicity across external systems or Prefect Server state. So if two pods run: - txn A: reads Variable “var” - txn B: reads Variable “var” Both can compute based on the same old value, then write conflicting updates — the transaction lock only protects the transaction key, not the Variable store in Prefect Server. If you still want to use a Prefect transaction for other benefits and add a distributed lock, you can combine them by using a lock manager with SERIALIZABLE isolation. This provides mutual exclusion for the block of code, but it’s the lock doing the protection, not the transaction integrating with Variables. Example with a distributed lock (Redis) + transaction

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

# Use a shared lock key to protect the critical section
LOCK_KEY = "var:update:var"

@task
def example_task():
    with transaction(
        key=LOCK_KEY,
        isolation_level=IsolationLevel.SERIALIZABLE,
        store=ResultStore(lock_manager=RedisLockManager(host="redis", port=6379)),
    ):
        # Only one worker will enter this block at a time
        current = Variable.get("var")
        # compute some_new_value = ...
        Variable.set("var", value=some_new_value, overwrite=True)

But if you’re on Prefect 3.x and running distributed (EKS pods), the recommended and simpler tool for this use case is a global concurrency limit with a single slot — it’s purpose-built for exactly this “distributed mutex” pattern: - Create the limit:

prefect gcl create variable:var --limit 1

- Use it in your task:

from prefect import task
from prefect.concurrency.sync import concurrency
from prefect.variables import Variable

@task
def example_task():
    with concurrency("variable:var", occupy=1, strict=True, timeout_seconds=60, max_retries=3):
        current = Variable.get("var")
        # compute some_new_value
        Variable.set("var", value=some_new_value, overwrite=True)

Summary - Transactions ≠ atomic Variable updates. - For atomic read/modify/write across pods, use a distributed lock — easiest via global concurrency limits (recommended), or via a Redis-backed transaction lock if you already run Redis. - You can still use transactions for idempotency and rollback, but don’t rely on them to serialize Variable access by themselves. Docs: - Transactions - Global concurrency limits - Variables API - Concurrency API

Yes. You can create and manage Global Concurrency Limits via the Python client (and directly via the REST API) — no CLI required. Python client (async)

from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import GlobalConcurrencyLimitCreate, GlobalConcurrencyLimitUpdate

async def ensure_gcl(name: str, limit: int, slot_decay_per_second: float = 0.0):
    async with get_client() as client:
        # Create (will error if it already exists)
        gcl = GlobalConcurrencyLimitCreate(
            name=name,
            limit=limit,
            active=True,
            slot_decay_per_second=slot_decay_per_second,  # >0 to use as a rate limit
        )
        gcl_id = await client.create_global_concurrency_limit(gcl)
        return gcl_id

async def upsert_gcl(name: str, limit: int):
    async with get_client() as client:
        # Upsert pattern: try create, else update
        try:
            gcl_id = await client.create_global_concurrency_limit(
                GlobalConcurrencyLimitCreate(name=name, limit=limit, active=True)
            )
            return gcl_id
        except Exception:
            await client.update_global_concurrency_limit(
                name,
                GlobalConcurrencyLimitUpdate(limit=limit)
            )

Synchronous wrapper

import asyncio

def ensure_gcl_sync(name: str, limit: int):
    async def _inner():
        return await ensure_gcl(name, limit)
    return asyncio.run(_inner())

Direct REST API - Create:

POST /v2/concurrency_limits/

- Read by name:

GET /v2/concurrency_limits/{id_or_name}

- Update:

PATCH /v2/concurrency_limits/{id_or_name}

- Delete:

DELETE /v2/concurrency_limits/{id_or_name}

- List:

POST /v2/concurrency_limits/filter

Example curl

curl -X POST "$PREFECT_API_URL/api/v2/concurrency_limits/" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "variable:var",
    "limit": 1,
    "active": true,
    "slot_decay_per_second": 0.0
  }'

Docs: - How-to: Global concurrency limits - Concepts: Global concurrency limits

Yes, you can use MemoryLockManager with a Prefect transaction — but only if all contenders run in the same Python process. Key points - MemoryLockManager: in-process only. It protects concurrent tasks/threads within a single flow run process. It will NOT coordinate across pods or across different worker processes in EKS. - FileSystemLockManager: works across processes on the same machine, or across pods only if you mount a shared RWX volume (e.g., EFS) at the same path for every worker. - RedisLockManager: works cluster-wide across pods; use if different flow runs/workers may contend. Example using MemoryLockManager (single process, e.g., within one flow run)

from prefect import task
from prefect.transactions import transaction, IsolationLevel
from prefect.results import ResultStore
from prefect.locking.memory import MemoryLockManager
from prefect.variables import Variable

@task
def example_task():
    with transaction(
        key="var:update:var",  # or per-run: f"var:update:{prefect.runtime.flow_run.id()}"
        isolation_level=IsolationLevel.SERIALIZABLE,
        store=ResultStore(lock_manager=MemoryLockManager()),
        write_on_commit=False,  # optional; if you’re not staging results
    ):
        current = Variable.get("var")
        # compute some_new_value...
        Variable.set("var", value=some_new_value, overwrite=True)

If your contention can happen across pods/workers - MemoryLockManager won’t help. Use one of: - RedisLockManager (distributed):

from prefect_redis.locking import RedisLockManager
    store = ResultStore(lock_manager=RedisLockManager(host="redis", port=6379))

- FileSystemLockManager with a shared RWX volume mounted at the same path on every worker pod:

from prefect.locking.filesystem import FileSystemLockManager
    from pathlib import Path

    store = ResultStore(
        lock_manager=FileSystemLockManager(lock_files_directory=Path("/mnt/prefect-locks"))
    )

Ensure all worker pods mount the same EFS/NFS path at

/mnt/prefect-locks

. One more note on GCLs: you do not need a new global concurrency limit per flow run. Typically you create a single limit (e.g., name “variable:var”, limit=1) once and reuse it across all runs. If your concern was per-run creation overhead, you can avoid that by reusing a single GCL. Docs - Transactions - Global concurrency limits

You nailed it — the issue is that you’re creating a brand new MemoryLockManager (and ResultStore) for each transaction. Each manager instance has its own in-memory lock table, so they don’t coordinate with each other. As a result, both tasks can acquire their “own” lock and run concurrently. Fix: create and reuse a single lock manager (and ResultStore) across your tasks in the same process. Example (single flow run on one worker/pod):

from prefect import task
from prefect.transactions import transaction, IsolationLevel
from prefect.results import ResultStore
from prefect.locking.memory import MemoryLockManager
from prefect.variables import Variable

# Create ONE shared manager/store at import time
MEM_LOCK_MANAGER = MemoryLockManager()
TXN_STORE = ResultStore(lock_manager=MEM_LOCK_MANAGER)

@task
def example_task(var: str):
    with transaction(
        key="test",  # same key across tasks to serialize
        isolation_level=IsolationLevel.SERIALIZABLE,
        store=TXN_STORE,
        write_on_commit=False,  # optional, since you're not staging results
    ):
        print("in transaction")
        import time; time.sleep(5)
        Variable.get(var, False)
        Variable.set(var, value=False, overwrite=True)
        print("exiting transaction")

Notes - MemoryLockManager works only within the same Python process. In your case (single flow run, same worker/pod, default thread-based task runner), that’s fine. - If you ever run tasks in separate processes or across pods (Dask/Ray, multiple workers, etc.), MemoryLockManager won’t help. Use: - FileSystemLockManager with a shared RWX volume mounted to the same path in all pods, or - RedisLockManager for cluster-wide locking. - The transaction key must match for all contenders (you used "test", which is fine). If you want per-flow-run exclusivity (not global), use something like

f"test:{prefect.runtime.flow_run.id()}"

. Docs: - Transactions (Prefect 3.x)

Short answer: FileSystemLockManager needs a POSIX filesystem path. S3 is object storage and does not provide the required filesystem/atomic semantics. You can’t point it at

s3://...

, and using s3fs-fuse is not recommended for locking (non-POSIX semantics, eventual consistency, non-atomic renames/creates). Recommended options for distributed transactions - Use a real network filesystem (EFS/FSx) mounted RWX into all worker pods, then point FileSystemLockManager at that shared path. - Or use RedisLockManager (ElastiCache Redis works great) for cluster-wide locking. Example: FileSystemLockManager with EFS 1) Mount the same EFS path in every worker/flow pod at, e.g.,

/mnt/prefect-locks

(via AWS EFS CSI driver with a RWX PVC). 2) Use that path in your transaction store:

from pathlib import Path
from prefect.transactions import transaction, IsolationLevel
from prefect.results import ResultStore
from prefect.locking.filesystem import FileSystemLockManager
from prefect.variables import Variable

LOCK_DIR = Path("/mnt/prefect-locks")  # EFS mount shared across pods
STORE = ResultStore(lock_manager=FileSystemLockManager(lock_files_directory=LOCK_DIR))

def update_var_safely(var: str, new_value):
    with transaction(
        key="test",  # or f"test:{prefect.runtime.flow_run.id()}" for per-run exclusivity
        isolation_level=IsolationLevel.SERIALIZABLE,
        store=STORE,
        write_on_commit=False,
    ):
        current = Variable.get(var, False)
        Variable.set(var, value=new_value, overwrite=True)

Example: RedisLockManager (distributed, no shared filesystem needed)

from prefect.transactions import transaction, IsolationLevel
from prefect.results import ResultStore
from prefect_redis.locking import RedisLockManager
from prefect.variables import Variable

STORE = ResultStore(lock_manager=RedisLockManager(
    host="your-redis-host", port=6379, username=None, password=None, ssl=True
))

def update_var_safely(var: str, new_value):
    with transaction(
        key="test",
        isolation_level=IsolationLevel.SERIALIZABLE,
        store=STORE,
        write_on_commit=False,
    ):
        current = Variable.get(var, False)
        Variable.set(var, value=new_value, overwrite=True)

Why not S3/s3fs for locks? - S3 is eventually consistent in some operations and lacks POSIX guarantees needed for reliable mutual exclusion. Lock acquisition in FileSystemLockManager depends on atomic file operations, which S3 does not provide. If you want to proceed with a “remote” store without running stateful storage yourself, Redis (managed ElastiCache) is the simplest and most reliable choice for distributed locks on EKS. References - Prefect transactions - AWS EFS CSI driver: EFS CSI on EKS Quick check: do you have EFS handy, or would ElastiCache Redis be easier for your environment?

Yes. RedisLockManager works anywhere your code can reach the Redis server — locally or in EKS. It’s just a network-backed lock; it doesn’t depend on where the flow runs or where Prefect Server is. When to use which: - Single process (one Python interpreter): MemoryLockManager is enough. - Multiple processes/machines (your laptop + EKS worker pods, or multiple pods): use RedisLockManager. Quick local setup - Start Redis locally (Docker example):

docker run --rm -p 6379:6379 redis:7

- Install the integration:

pip install prefect-redis

- Use it in your transaction:

from prefect.transactions import transaction, IsolationLevel
from prefect.results import ResultStore
from prefect_redis.locking import RedisLockManager
from prefect.variables import Variable

# Reuse ONE store/manager instance in your process
STORE = ResultStore(lock_manager=RedisLockManager(host="localhost", port=6379))

def update_var_safely(var: str, new_value):
    with transaction(
        key="var:update:var",           # same key across contenders
        isolation_level=IsolationLevel.SERIALIZABLE,
        store=STORE,
        write_on_commit=False,          # optional if not staging results
    ):
        current = Variable.get(var, False)
        Variable.set(var, value=new_value, overwrite=True)

Connecting to a remote/managed Redis (e.g., ElastiCache)

STORE = ResultStore(
    lock_manager=RedisLockManager(
        host="your-redis-host",
        port=6379,
        username="...",                # if needed
        password="...",                # if needed
        ssl=True                       # enable TLS if required
    )
)

As long as your local machine can reach that Redis (VPN/VPC access, security groups, etc.), the same lock will coordinate between your local runs and your EKS workers. Reference: - prefect-redis (GitHub) - Transactions (Prefect 3.x)