Postgres session-advisory-lock implementation of ALLM.Pipeline.Lock.
PRESERVED, NOT ACTIVE. The active implementation is
ALLM.Pipeline.Lock.Noop — see ALLM.Pipeline.Lock for
why the lock was dropped and how to restore this one. This module is kept
intact so the guarantee can be brought back deliberately rather than
reconstructed from scratch. The serialization mapping it applies is no longer
its own — batch 1.C moved it onto the host's ALLM.Pipeline.Registry
(lock_keys:); the host-side membership guard is a consumer repo's
registry-declared-values test, with its runner test pinning the derived
lock keys.
Mechanism
Scheduled cron runs of the same pipeline can overlap if a previous run is slow (e.g. an ordinance pipeline still grinding through PDFs when the next weekly trigger fires). To prevent overlap on the shared Postgres database, each invocation acquires a session-scoped advisory lock keyed on the pipeline name.
- We use
pg_try_advisory_lock(key)(NOT the_xact_variant) so the lock survives across the many transactions inside a long pipeline. - The lock is held by a single dedicated connection, checked out via
Ecto.Repo.checkout/2.pg_try_advisory_lockis bound to the connection that ran it; releasing on a different connection is silently a no-op, so we MUST run lock + work + unlock on the same checked-out connection. This pinned connection is exactly the fragility that motivated the switch toNoop— it sits idle during long LLM-bound steps and can be reaped. - The key is derived from the pipeline name (
:erlang.phash2/1), stable within a process lifetime and small enough for Postgres'sbigint. Different pipelines get different keys, so unrelated pipelines run concurrently. EXCEPTIONS collapse to one key viacanonical_lock_name/1. - On contention
pg_try_advisory_lockreturnsfalseand we return{:error, :already_running}immediately, without invoking the pipeline. try/afterreleases the lock on every exit path, including raises. An advisory lock is also session-bound, so even a hard crash releases it when the connection dies.
Summary
Functions
Map pipelines that must serialize against each other to one canonical name.
The Postgres advisory-lock key a pipeline holds for the duration of a run.
Functions
@spec canonical_lock_name(ALLM.Pipeline.Lock.name()) :: atom()
Map pipelines that must serialize against each other to one canonical name.
WHICH pipelines must serialize is host domain knowledge — it depends on what
external session they share and which rows they replace — so the mapping is
declared as lock_keys: on the host's ALLM.Pipeline.Registry (batch 1.C
moved it off two hardcoded clauses here) and resolved at runtime by
ALLM.Pipeline.Config.lock_keys/0. The per-pair reasons travel WITH the
values, on that declaration.
An undeclared pipeline maps to itself, so it gets its own key and runs concurrently with everything else.
@spec lock_key_for(ALLM.Pipeline.Lock.name()) :: non_neg_integer()
The Postgres advisory-lock key a pipeline holds for the duration of a run.
Pipelines that must serialize against each other collapse to the SAME key via
canonical_lock_name/1. Exposed publicly so runner_test.exs can assert the
real derivation rather than a hand-mirrored copy that could drift.