One Writer per LangGraph Graph: Non-Blocking Postgres Advisory Locks

Take a session-level lock with pg_try_advisory_lock on a dedicated, unpooled connection, keyed by a stable hash of the graph name. The winner runs the tick; every other replica gets false at once and returns.

LangGraph in production, a 10-part series. ← Part 6 · Part 8 →

A blocking lock serialises the duplicate; it does not prevent it

Check-then-invoke is correct inside one process and a race across two. Both replicas read the same run key, find no thread, and start the graph. The checkpointer then "saves a snapshot of graph state at each super-step" (LangGraph checkpointers) onto that one thread. Interleaved state, one plan attempted twice. Postgres will not police it, since "it is up to the application to use them correctly" (PostgreSQL).

Replica two blocks on pg_advisory_lock, the winner finishes, the lock frees, and the loser walks into the same tick the winner already checkpointed, holding a connection the whole wait. You get the duplicate twice, spaced out.

pg_try_advisory_lock answers at once instead: it "will either obtain the lock immediately and return true, or return false without waiting" (PostgreSQL). The loser logs who it deferred to and exits. pg_try_advisory_xact_lock is the wrong knob: it releases at the first commit (PostgreSQL), and a graph run spans several.

The lock lives on its own connection, and it dies with the session

Session-level locks are "held until explicitly released or the session ends" (PostgreSQL). Open one direct connection for the lock, hold it for exactly the run, close it — never pooled. From a pool, the session returns still holding the lock, so every later tick skips silently; or a transaction-mode pooler unlocks on a different session: "If the lock was not held, false is returned, and in addition, an SQL warning will be reported" (PostgreSQL).

Crash recovery is free: kill the process, the session ends, the lock goes. The one gap is a host that vanishes while Postgres believes the session alive, so set keepalives.

hash() is not a lock key, and one integer needs a registry

Python's hash() is salted per process, so replicas derive different integers, both acquire, and a correctly installed lock protects nothing. Use zlib.crc32. The single-bigint form needs a key registry; the pair drops it by namespacing per application. Namespace the first integer regardless — the advisory key space is shared by every client on the database.

The whole mechanism is one context manager

import sys, threading, time, zlib
from contextlib import contextmanager
from typing import TypedDict

import psycopg
from langgraph.graph import END, START, StateGraph

dsn = sys.argv[1] if len(sys.argv) > 1 else "postgresql://warehouse@localhost:55432/scratch"
lock_space = 7301 # one number per application; the advisory key space is shared


class RestockState(TypedDict, total=False):
    feed_down: bool
    planned: bool


def lock_key(graph: str) -> tuple[int, int]:
    return lock_space, zlib.crc32(graph.encode()) - 2**31    # stable across processes; fits int4


@contextmanager
def sole_writer(graph: str, log: list[str]):
    conn = psycopg.connect(dsn, autocommit=True)      # direct, never pooled
    try:
        got = conn.execute("SELECT pg_try_advisory_lock(%s, %s)",
                           lock_key(graph)).fetchone()[0]
        try:
            yield got
        finally:
            if got:
                freed = conn.execute("SELECT pg_advisory_unlock(%s, %s)",
                                     lock_key(graph)).fetchone()[0]
                if not freed:
                    log.append(f"unlock of {graph} returned false")
    finally:
        conn.close()


def tick(graph, work, who, log):
    with sole_writer(graph, log) as mine:
        if not mine:
            log.append(f"{who}: skipped, another worker holds {graph}")
            return
        try:
            work()
            log.append(f"{who}: ran {graph}")
        except Exception as exc:
            log.append(f"{who}: {graph} failed with {type(exc).__name__}; lock still released")


def plan(state: RestockState) -> RestockState:
    time.sleep(0.5)        # widens the race window; not a tuning value
    if state.get("feed_down"):
        raise RuntimeError("supplier feed down")
    return {"planned": True}


builder = StateGraph(RestockState)
builder.add_node("plan", plan)
builder.add_edge(START, "plan")
builder.add_edge("plan", END)
restock = builder.compile()      # with part 02's checkpointer in production

if __name__ == "__main__":
    log: list[str] = []
    barrier = threading.Barrier(2)

    def replica(name):
        barrier.wait()
        tick("restock_plan", lambda: restock.invoke({"feed_down": False}), name, log)

    workers = [threading.Thread(target=replica, args=(n,)) for n in ("replica-a", "replica-b")]
    [w.start() for w in workers]
    [w.join() for w in workers]
    tick("restock_plan", lambda: restock.invoke({"feed_down": True}), "replica-a", log)
    tick("restock_plan", lambda: restock.invoke({"feed_down": False}), "replica-b", log)
    tick("pack_audit", lambda: restock.invoke({"feed_down": False}), "replica-a", log)
    print("\n".join(sorted(log[:2])) + "\n" + "\n".join(log[2:]))

replica-a: skipped, another worker holds restock_plan
replica-b: ran restock_plan
replica-a: restock_plan failed with RuntimeError; lock still released
replica-b: ran restock_plan
replica-a: ran pack_audit

Line three is the release; line four is the proof, a fresh tick running straight after. The tests to write: two threads at a barrier producing one ran and one skipped, pg_terminate_backend leaving the lock reclaimable, two graph names never blocking each other. A blocking version in the contention test should fail with two runs.

A skip is work that did not happen, and blocking does not pay for it

A skip is data loss with good manners: if the winner dies mid-run and the window passes, nothing retries. A blocking lock produces a second execution. A durable work row claimed with SELECT … FOR UPDATE SKIP LOCKED leaves the unit visible and claimable, at the cost of a lease and an expiry rule. A Redis or etcd lease covers non-Postgres writers at the cost of another system. A scheduler enqueueing to consumers needs at-most-once semantics before it protects anything.

Practical takeaways

• Skip-lock when the tick is periodic and re-derives its work from current state.
• Durable claim when a unit must survive a crash or show up as a backlog.
• External lease when some writers are not Postgres clients.

Postgres enforces the exclusion; it does not enforce that anyone asks for it. A new entry point that writes the thread without entering sole_writer will produce exactly the duplicate the lock exists to stop.