ALLM.Pipeline.Lock behaviour (allm_pipeline v0.1.0)

Copy Markdown View Source

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.Advisory

This 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

Types

A pipeline name, e.g. :rich_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

name()

@type name() :: atom()

A pipeline name, e.g. :rich_summary.

Callbacks

with_lock(name, function)

@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

impl()

@spec impl() :: module()

The currently-configured lock implementation module.

with_lock(name, fun)

@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).