Scheduled and Background Work
A scheduled job will run twice sooner or later, so make it idempotent, keep two copies apart, and plan for the overrun.
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) # OSError (EAGAIN/EACCES) if heldBackground work is two jobs at once: something decides when to run, and something does the work. Keep the deciding outside your process where you can — a Kubernetes CronJob, a systemd timer, cron — because an in-process timer dies with the process and nobody is paged. When it must be in-process, sched.scheduler(timefunc=time.monotonic, delayfunc=time.sleep) uses a monotonic clock, so an NTP correction or a leap second cannot schedule your next run in the past.
It will run twice
A retried delivery, an operator pressing the button, a pod rescheduled mid-run: assume at-least-once. Derive an idempotency key from the work itself and make the second run a no-op — insert on a unique constraint, upsert, or check a processed table before acting. "Send the invoice" with no key sends two invoices.
One at a time
On a single host, take an exclusive non-blocking lock on a file descriptor: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) raises OSError with errno set to EAGAIN or EACCES when someone else holds it (check both; which one you get is platform-dependent). Across hosts the lock has to live where both can see it: a database advisory lock, or a row holding a lease with an expiry.
The run that outlasts its interval
If a five-minute job takes six, the ticks pile up. Skip rather than queue — Kubernetes spells that concurrencyPolicy: Forbid — and make the work resumable: process a bounded batch, record a cursor, let the next tick continue. A job that must finish before the next one starts is a job that falls over on its busiest day.
Example
When to use it
- Make a nightly reconciliation safe to re-run after it was killed halfway.
- Stop two replicas from both sending the same reminder email when the scheduler fires twice.
- Drain a million-row backlog across many short runs instead of one job that never finishes.
- Keep a cron job from piling up when one run takes longer than its interval.
More examples
flock, so a crash cannot leave a stale lock
Prefer this to an O_EXCL lock file whenever the job runs on one host: it has no stale-lock failure mode, because there is nothing to clean up. It does not work across hosts or on a network filesystem, which is where you move to a database lock.
import errno
import fcntl
import os
from contextlib import contextmanager
@contextmanager
def flocked(path):
"""Unix only. The lock lives on the open descriptor, not on the file's
existence, so the kernel releases it when this process dies - SIGKILL,
segfault, OOM kill, all of them."""
fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o644) # the file may already exist
try:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError as exc:
# Which errno you get depends on the platform; check both.
if exc.errno in (errno.EAGAIN, errno.EACCES):
raise RuntimeError(f"{path} is held by another run") from None
raise
os.truncate(fd, 0)
os.write(fd, f"{os.getpid()}\n".encode())
yield
finally:
fcntl.flock(fd, fcntl.LOCK_UN)
os.close(fd)
with flocked("/var/run/reconcile.lock"):
reconcile()The idempotency key in the database
An in-memory set works only until the process restarts, which is exactly when a duplicate delivery arrives. Putting the key in the same transaction as the effect is what makes the whole operation idempotent rather than merely usually idempotent.
-- The key is derived from the work, not generated per attempt.
CREATE TABLE processed_event (
event_key text PRIMARY KEY, -- e.g. 'invoice.settled:9f3a1c'
processed_at timestamptz NOT NULL DEFAULT now()
);
-- Claim the work and do it in one transaction. If this row already exists the
-- insert affects no rows, the job sees that, and skips - no second invoice.
BEGIN;
INSERT INTO processed_event (event_key) VALUES ($1)
ON CONFLICT (event_key) DO NOTHING;
-- rowcount 0 here means another run (or an earlier delivery) already owns it
UPDATE invoice SET status = 'settled' WHERE id = $2;
COMMIT;
-- Serialising whole jobs across hosts, with no table at all: an advisory lock,
-- held until the transaction ends and released automatically if the connection
-- drops. pg_try_advisory_xact_lock returns false rather than waiting.
SELECT pg_try_advisory_xact_lock(hashtext('nightly-reconcile'));A bounded, resumable batch
The deadline is the important line: the job gives up voluntarily before the next tick is due, so overlap never happens even under a backlog. Committing the cursor after each batch means a kill mid-run loses one batch of progress rather than all of it.
BATCH = 500
DEADLINE_SECONDS = 240 # under the 5-minute schedule, deliberately
def run_once(db, clock):
"""Process at most one batch, then stop. The next tick continues from the
cursor, so a backlog drains over several runs instead of one run that
overlaps the next."""
started = clock()
cursor = db.get_cursor("invoice_sweep") # last id handled, durable
processed = 0
while clock() - started < DEADLINE_SECONDS:
rows = db.fetch_after(cursor, limit=BATCH)
if not rows:
break # caught up
for row in rows:
handle(row)
cursor = row.id
db.set_cursor("invoice_sweep", cursor) # commit progress per batch
processed += len(rows)
# Emit this: a backlog that never reaches zero is the signal you want an
# alert on, long before the job starts overlapping itself.
return {"processed": processed, "cursor": cursor, "caught_up": not db.fetch_after(cursor, 1)}
Discussion