Serialization guard for pipeline runs — the abstraction behind "don't let two runs of the same pipeline overlap on the shared database".
Current state: NO-OP
The active implementation is ALLM.Pipeline.Lock.Noop, which runs
the pipeline directly with no locking. The Postgres
session-advisory-lock implementation is preserved in
ALLM.Pipeline.Lock.Advisory and can be restored by
configuring:
config :allm_pipeline, ALLM.Pipeline.Lock,
impl: ALLM.Pipeline.Lock.AdvisoryThis works on a host that declares an ALLM.Pipeline.Registry too — the
registry's lock: supplies the DEFAULT and a config-file impl: outranks it
per environment (see that module's "Precedence"). Restoring it for every
environment is the one-line edit to the host's lock: declaration instead.
Why the advisory lock was dropped
The advisory lock is session-scoped, so it required pinning ONE Postgres
connection (via Repo.checkout/2) for the ENTIRE run. During long LLM-bound
steps (e.g. NarrativeGenerator's ~40s of OpenAI calls) that pinned
connection sits idle at the protocol level, and a checked-out connection is
NOT kept warm by DBConnection's idle ping — so if anything reaps the idle
socket (an RDS idle_session_timeout, a network idle timeout) the run's
lock-holding connection dies mid-run, killing the run with a
DBConnection.ConnectionError at the next query.
Dropping the lock removes the checkout, and with it that entire failure mode.
The overlap guard it provided is, for the pipelines that get their own key,
mostly a cost / duplicate-work guard rather than a correctness invariant; the
genuinely correctness-critical cases are exactly the pairs the host declares
in lock_keys:, each of which carries its own reason on that declaration.
Those can be reintroduced in a form that does NOT hold a connection for the
run's duration — e.g. a lease row claimed with short queries — if overlap
becomes a real problem.
The serialization mapping (which names must share a key) is host domain
knowledge, and this package no longer carries it: batch 1.C moved it off
Advisory's two hardcoded clauses onto the host's ALLM.Pipeline.Registry
(lock_keys:), where ALLM.Pipeline.Config.lock_keys/0 reads it and
Advisory.canonical_lock_name/1 applies it. A host declares it on its own
ALLM.Pipeline.Registry.
Summary
Callbacks
Run fun under whatever serialization the configured implementation
provides, returning fun's result.
Functions
The currently-configured lock implementation module.
Dispatch to the configured ALLM.Pipeline.Lock implementation
(default ALLM.Pipeline.Lock.Noop).
Types
@type name() :: atom()
A pipeline name, e.g. :rich_summary.
Callbacks
@callback with_lock(name(), (-> result)) :: result | {:error, :already_running} when result: var
Run fun under whatever serialization the configured implementation
provides, returning fun's result.
An implementation MAY return {:error, :already_running} instead of running
fun when a concurrent run holds the guard. Noop never does — it always
runs fun.
Functions
@spec impl() :: module()
The currently-configured lock implementation module.
@spec with_lock(name(), (-> result)) :: result | {:error, :already_running} when result: var
Dispatch to the configured ALLM.Pipeline.Lock implementation
(default ALLM.Pipeline.Lock.Noop).