Locks, Leases, and Recovery

1. Distributed Lock Isn't a Lock

When multiple workers can perform the same task, the hard part is deciding which one actually should. The instinctive answer is a lock to make duplicate ownership impossible in the first place, so there's nothing to coordinate.

Look again at that last index, the one scoped to status = 'running'. It does double duty: it's a correctness invariant and a distributed lock.

Building the pipeline, we had to guarantee that an invoice delivered twice: first with a Pub/Sub redelivery and a retry after a timeout. These two messages racing never runs through the pipeline in parallel. The classic instinct is SELECT ... FOR UPDATE on the invoice row, instead we let the schema refuse the second worker. Here, the lock would just an INSERT, and the loser bounces off the index:

CREATE OR REPLACE FUNCTION acquire_processing_lock(...)
RETURNS UUID AS $$
BEGIN
  INSERT INTO processing_jobs (invoice_id, attempt, status, started_at)
  VALUES (p_invoice_id, p_attempt, 'running', NOW())
  RETURNING id INTO v_job_id;
  RETURN v_job_id;
EXCEPTION WHEN unique_violation THEN
  RETURN NULL;  -- someone else is already running this invoice
END; $$;

const jobId = await this.jobRepo.acquireLock(invoiceId, attempt);
if (!jobId) return;

We chose an index because this lock has to survive a process death, where this transaction is held for the entire multi-minute pipeline run. A FOR UPDATE lock vanishes the instant the connection drops and a processing_jobs row with status = 'running' can be queried.

2. Match the Lock to Its Lifetime

Next, we asked what should happen when the process holding a lock disappears. Different kinds of work need different failure behavior, and the wrong lock can appear to work perfectly until the first crash or race.

Our pipeline ended up using two different locking mechanisms because it had two different lock lifetimes. The running-job lock guards a long-lived workflow. Duplicate detection needs something shorter-lived.

Some customers had separate invoice rows that looked distinct at first, but after review, turned out to be the same invoice re-sent by a vendor. The partial-index trick does not help here because both rows legitimately exist. The race is that two workers can ask “does a matching invoice already exist?” at the same time, both see nothing, and both mark themselves as original.

So we needed mutual exclusion around a computed business key, not around a specific row. That is what a Postgres advisory lock gives us: a lock on an arbitrary integer we choose, independent of any table.

const lockKey = computeLockKey(invoice.org_id, vendorName, invoiceNumber);
await this.invoiceRepo.advisoryLock(lockKey);
try {
  const matches = await this.invoiceRepo.findDuplicates(/* … */);
  if (matches.length === 0) {
    await this.invoiceRepo.updateDuplicateStatus(invoice.id, "none");
  } else {
    // join/create a duplicate group
  }
} finally {
  await this.invoiceRepo.advisoryUnlock(lockKey);
}

Now, two invoices with the same business key hash to the same lockKey and serialize. Invoices with different keys never contend. The two locks are not interchangeable, and using the wrong one here would not necessarily fail loudly. It could show up weeks later as the same invoice being approved twice.

3. When the Lock Holder Dies

Every distributed lock eventually meets the same problem where the worker that acquired it stops existing. How the system recovers afterward is most important especially in long-running jobs.

A durable lock has a durable failure mode. If a worker claims the lock and its instance is then killed mid-invoice, that running row sits there forever and the invoice is stuck where no live worker holds it, but the index says one does.

So the first thing the pipeline does, before acquiring, is sweep out failed ones:

await this.jobRepo.releaseStaleLocks(invoiceId, 15); // flip 'running' > 15min → 'failed'
const jobId = await this.jobRepo.acquireLock(invoiceId, attempt);
if (!jobId) return;

The second half is making the retry cheap. At the start of each step the orchestrator writes a breadcrumb, and each step's shouldRun() is idempotent.

await this.invoiceRepo.updateProcessingStage(invoiceId, step.stageName);

A restarted worker doesn't begin at step zero; it fast-forwards through completed work instead of re-running an expensive OCR call.