Distributed Locking
TTL-bounded, owner-scoped locks backed by the queue's storage — mutual exclusion across workers and processes.
TTL-bounded, owner-scoped locks backed by the queue's storage — mutual exclusion across workers and processes.
Coordinate across processes without a separate lock server. flexiq's distributed locks are TTL-bounded and owner-scoped, backed by the same storage as the task queue — locks work across multiple worker processes on one machine, and across multiple machines when the backend is a shared server (Postgres or Redis) rather than a local SQLite file.
Use a lock when multiple workers or processes must not run the same critical section at the same time — refreshing a shared cache, running a singleton periodic task, or calling an external API with a single-writer constraint.
Celery has no built-in lock — teams reach for
celery-onceor a Redis lock. flexiq'squeue.lock()uses the queue database directly, so there's nothing extra to run.
Coming from BullMQ? BullMQ's locking is internal — a
lockDurationper job that stops two workers double-processing it, not something you can call from application code. flexiq'squeue.lock()/withLock()is a general-purpose distributed lock you can use for any critical section, job processing or not.
The Rust SDK wraps no lock API — no guard type, no scoped form, no background renewal.
queue.storage()reaches theStoragelock methods directly, and everything below is written against those four calls.
The scoped form acquires, runs, and releases in one call — the shortest path for a critical section:
with queue.lock("cache-refresh"):
refresh_cache()The lock is automatically released when the with block exits, even if an
exception is raised.
await queue.withLock("report:2026-06", async () => {
await rebuildReport();
});withLock acquires, runs, and releases — throwing LockNotAcquiredError
if the lock is held elsewhere.
boolean ran = flexiq.withLock("report:2026-06", 60_000, () -> rebuildReport());
if (!ran) {
log.info("another worker holds the lock");
}withLock acquires, runs, and releases — returning whether the body ran.
It never throws for a lock that's already held; it returns false instead.
The Rust SDK does not wrap locks in a scoped guard. queue.storage() reaches
the Storage lock methods, and the release is yours to make:
use flexiq::Storage;
let owner = format!("worker-{}", std::process::id());
if queue.storage().acquire_lock("cache-refresh", &owner, 30_000)? {
refresh_cache();
queue.storage().release_lock("cache-refresh", &owner)?;
}A panic between the two calls leaks the lock until its TTL expires — which is what the TTL is for, but it is the only thing releasing it.
async with queue.alock("cache-refresh"):
await refresh_cache()alock() accepts the same parameters as lock() and is safe to use inside
async functions and FastAPI/Django async views.
The Storage lock methods are synchronous and there is no async variant of any
of them. Inside an async function, run them on a blocking thread rather than on
the executor:
use flexiq::Storage;
let storage = queue.storage().clone();
let held = tokio::task::spawn_blocking(move || {
storage.acquire_lock("cache-refresh", "worker-1", 30_000)
})
.await
.expect("the blocking task did not panic")?;For cases that don't map to a single block — acquiring in one place and releasing in another — drive the lock explicitly:
lock = queue.lock("resource", ttl=30)
if lock.acquire():
try:
... # critical section
finally:
lock.release()
else:
... # another process holds the locklock.extend(ttl=None) pushes the expiry out on demand, and lock.info()
returns the current holder. Calling acquire() directly like this — instead
of entering with — does not start the auto-extend background thread
even when auto_extend=True; only the context manager does that. Prefer
with queue.lock(...) for anything that maps to a single block.
using lock = queue.lock("resource", { ttlMs: 30_000 });
if (lock.acquire()) {
// ... critical section
} // released at block exit (via `using`)lock.extend(ms), lock.info(), and lock.ownerId round out the API.
With using, the lock releases automatically when the block exits;
otherwise call lock.release().
Lock is AutoCloseable, so try-with-resources releases it at block exit:
try (Lock lock = flexiq.lock("resource", 30_000)) {
if (lock.acquire()) {
// ... critical section
}
} // released at block exit| Method | Description |
|---|---|
acquire() | Try once; false if another owner holds a live lock. |
tryAcquire(Duration timeout) | Retry every 50ms until obtained or the timeout elapses. |
extend(long ttlMs) | Push the expiry out if still held; false means the lock was lost. |
release() / close() | Give the lock up (no-op if not held). |
flexiq.lock(name) without a TTL defaults to 30 seconds.
There is no handle type — the manual form is the only form:
use flexiq::Storage;
let storage = queue.storage();
let owner = format!("worker-{}", std::process::id());
if storage.acquire_lock("resource", &owner, 30_000)? {
// ... critical section
storage.extend_lock("resource", &owner, 30_000)?; // push the expiry out
storage.release_lock("resource", &owner)?;
} else {
// ... another process holds the lock
}| Method | Description |
|---|---|
acquire_lock(name, owner, ttl_ms) | Try once; false if another owner holds a live lock. |
extend_lock(name, owner, ttl_ms) | Push the expiry out; false means this owner no longer holds it. |
get_lock_info(name) | The current holder and expiry, as Option<LockInfo>. |
release_lock(name, owner) | Give the lock up; false if this owner did not hold it. |
acquire_lock refuses a live lock even for the owner that already holds it —
renew with extend_lock rather than calling acquire_lock twice.
Whether a held lock renews itself before its TTL expires differs by SDK:
When auto_extend=True (the default), a background thread extends the
lock's TTL at ttl / 3 intervals. This prevents the lock from expiring
during a long-running operation without requiring an artificially large
TTL:
# This lock stays alive for as long as the block runs, even if it takes
# several minutes — the background thread re-extends it every ttl/3.
with queue.lock("long-job", ttl=30, auto_extend=True):
run_slow_operation()autoExtend defaults to true — while held, the lock renews itself at
ttlMs / 3. If a renewal ever fails (the lock expired or was stolen),
auto-extend stops rather than fighting for a lock this handle no longer
holds. Pass autoExtend: false to manage extension yourself with
lock.extend():
using lock = queue.lock("long-job", { ttlMs: 30_000, autoExtend: false });Locks do not auto-extend. Choose a TTL comfortably longer than the
protected work's worst case, or call extend() at checkpoints — a failed
extend means the lock expired and another owner may now hold it, so stop
the critical section. Expiry is what keeps a crashed owner from holding a
lock forever.
Locks do not auto-extend, and nothing renews one in the background. Choose a
TTL comfortably longer than the protected work's worst case, or call
extend_lock at checkpoints — false there means this owner no longer holds
the lock, so stop the critical section rather than finish it unprotected.
Expiry is what keeps a crashed owner from holding a lock forever.
By default, a failed acquisition doesn't wait — the caller finds out immediately:
lock() raises LockNotAcquiredError immediately if the lock is held by
another process. Pass timeout to retry every retry_interval seconds
until it succeeds or the timeout elapses:
from flexiq.locks import LockNotAcquiredError
try:
with queue.lock("resource", timeout=5.0):
do_work()
except LockNotAcquiredError:
print("Could not acquire lock within 5 seconds")There's no built-in retry-with-timeout on Lock or withLock — both fail
fast. If you need to wait, poll acquire() in your own loop with a short
delay between attempts.
tryAcquire(Duration) retries every 50ms until it succeeds or the timeout
elapses:
try (Lock lock = flexiq.lock("resource", 30_000)) {
if (!lock.tryAcquire(Duration.ofSeconds(5))) {
log.info("could not acquire lock within 5 seconds");
return;
}
doWork();
}acquire_lock tries exactly once. Nothing on the Storage surface retries with
a timeout, so waiting is a loop you write:
use flexiq::Storage;
use std::time::{Duration, Instant};
let deadline = Instant::now() + Duration::from_secs(5);
let mut held = false;
while !held && Instant::now() < deadline {
held = queue.storage().acquire_lock("resource", &owner, 30_000)?;
if !held {
std::thread::sleep(Duration::from_millis(50));
}
}Check who currently holds a lock — useful for a dashboard, a health check, or deciding whether to skip a queued retry:
info = queue.lock("my-lock").info()
# {"lock_name": "my-lock", "owner_id": "...", "acquired_at": ..., "expires_at": ...}const info = queue.lock("my-lock").info();
// { lockName, ownerId, acquiredAt, expiresAt } | undefinedOptional<LockInfo> info = flexiq.lockInfo("my-lock");
info.ifPresent(i -> log.info("held by {} until {}", i.ownerId, i.expiresAt));use flexiq::Storage;
if let Some(info) = queue.storage().get_lock_info("my-lock")? {
// LockInfo { lock_name, owner_id, acquired_at, expires_at } — times are Unix ms.
println!("held by {} until {}", info.owner_id, info.expires_at);
}owner_id comes back raw: it is a bearer token, and whoever knows it can
release or extend the lock. Mask it yourself before showing it to anyone who
is not the holder.
owner_id is a bearer token — whoever knows it can release or extend the
lock. The high-level queue.lock(name).info() masks it to "***" for
any caller that isn't the current holder, so prefer it over the
low-level queue._inner.get_lock_info() (which returns the raw token)
when surfacing lock status to other parties, e.g. a dashboard or another
tenant.
owner_id comes back raw, and there is nothing that masks it — get_lock_info
is the storage read, with no handle above it that could know whether the
caller is the current holder. Whoever knows the token can release or extend the
lock, so mask it yourself before surfacing lock status to another party: a
dashboard, another tenant, a log line.
from flexiq.locks import LockNotAcquiredError
try:
with queue.lock("my-lock", timeout=2.0):
critical_section()
except LockNotAcquiredError:
# Another process holds the lock; handle gracefully
log.warning("skipping — another process is running the critical section")import { LockNotAcquiredError } from "@byteveda/flexiq";
try {
await queue.withLock("my-lock", async () => {
await criticalSection();
});
} catch (error) {
if (error instanceof LockNotAcquiredError) {
// Another process holds the lock; handle gracefully
log.warn("skipping — another process is running the critical section");
} else {
throw error;
}
}boolean ran = flexiq.withLock("my-lock", 30_000, () -> criticalSection());
if (!ran) {
// Another process holds the lock; withLock doesn't throw for this case
log.warn("skipping — another process is running the critical section");
}use flexiq::Storage;
let owner = format!("worker-{}", std::process::id());
if !queue.storage().acquire_lock("my-lock", &owner, 30_000)? {
// Another process holds the lock; a held lock is `false`, not an error.
eprintln!("skipping — another process is running the critical section");
return Ok(());
}
critical_section();
queue.storage().release_lock("my-lock", &owner)?;The Err arm means the backend could not be reached, never that the lock was
taken — those are two different answers and they arrive on different channels.
withLock throws a different error, LockLostError, if the lease was
lost while the guarded function was still running (the TTL expired
faster than auto-extend could renew it) — the critical section may have
run unprotected, so the result is discarded either way.
There is no lost-lease signal. Nothing watches the lock while the critical
section runs, so the only way to learn the TTL lapsed — or that another owner
took the lock over — is extend_lock returning false at a checkpoint you
placed. Everything between the lapse and that checkpoint ran unprotected, which
is the argument for checking rather than trusting the TTL.
Because lock state lives in the queue's storage, locks are effective across multiple worker processes on the same machine. Locking across different machines works too, but only when they share a networked backend — SQLite is a local file, so cross-machine locking requires the Postgres or Redis backend:
# Process A (machine 1)
with queue.lock("billing-run"):
run_billing()
# Process B (machine 2) — waits or raises while process A holds the lock
with queue.lock("billing-run", timeout=30.0):
run_billing()// Process A (machine 1)
await queue.withLock("billing-run", async () => {
await runBilling();
});
// Process B (machine 2) — throws LockNotAcquiredError while A holds it
await queue.withLock("billing-run", async () => {
await runBilling();
});// Process A (machine 1)
flexiq.withLock("billing-run", 30_000, () -> runBilling());
// Process B (machine 2) — returns false while A holds it; retry-wait instead:
try (Lock lock = flexiq.lock("billing-run", 30_000)) {
if (lock.tryAcquire(Duration.ofSeconds(30))) {
runBilling();
}
}use flexiq::Storage;
// Process A (machine 1) — `FlexiQ::from_storage` over a shared Postgres or
// Redis backend, so both machines see the same lock table.
let owner = format!("machine-1-{}", std::process::id());
if queue.storage().acquire_lock("billing-run", &owner, 30_000)? {
run_billing();
queue.storage().release_lock("billing-run", &owner)?;
}
// Process B (machine 2) — acquire_lock returns false while A holds it.There is no retry-with-timeout: poll acquire_lock in your own loop with a
short delay if process B needs to wait rather than skip.
On SQLite, cross-process locking works via WAL mode and exclusive
transactions, but SQLite file locking is machine-local — it only works
across processes on the same machine. For multi-machine deployments, use
the PostgreSQL backend, where SELECT FOR UPDATE SKIP LOCKED guards lock
acquisition too, or the Redis backend, which supports the same
acquire/release/extend/info operations via Lua scripts.
For advanced use cases, drive a lock without a DistributedLock handle,
using the same primitives it's built on:
import uuid
owner_id = uuid.uuid4().hex
# Acquire (ttl_ms, not seconds)
acquired = queue._inner.acquire_lock("my-lock", owner_id, ttl_ms=30_000)
# Extend
queue._inner.extend_lock("my-lock", owner_id, ttl_ms=30_000)
# Inspect
info = queue._inner.get_lock_info("my-lock")
# {"lock_name": "my-lock", "owner_id": "...", "acquired_at": ..., "expires_at": ...}
# Release
queue._inner.release_lock("my-lock", owner_id)The low-level API skips auto-extension and does not release on
exception. Prefer queue.lock() / queue.alock() for production code.
An expired lock stops blocking an acquisition the moment its TTL passes —
acquire_lock only refuses an unexpired one — but the row itself stays until
something removes it. reap_expired_locks is that something: it deletes every
lock that expired before the timestamp you hand it and returns how many it
removed.
use flexiq::Storage;
let removed = queue.storage().reap_expired_locks(flexiq::now_millis())?;You rarely need to: the scheduler a worker spawns already calls it on its stale sweep, so a deployment that runs any worker at all is reaping locks. Calling it by hand is for a process that only ever produces.