ergon_reconciler (ergon v0.5.0)

View Source

Disaster recovery: put the runtime back in step with what the database says.

Run it after a node dies mid-flight, after a failover, or on any restart where something may have been holding work when it stopped. Three moves, in this order, and the order matters:

  1. The host's hydrate callback. Ergon does not know what in-memory state a host keeps, so this is where the host stops suspect processes and rebuilds them. It runs first for a specific reason: releasing messages before stopping the consumers that will receive them just hands redelivered work to processes that are about to be killed.

  2. Release stranded pgmq leases. A consumer that died mid-processing left its messages invisible until their visibility timeout expires, which may be thirty seconds or thirty minutes. ergon_pgmq:release_leases/1 expires them all at once, then metrics are snapshotted so they reflect the recovered state rather than the broken one.

  3. Check pending_parents for drift. Reported by default, repaired only when asked.

ergon_reconciler:run(#{
    pgmq_queues => [~"events"],
    hydrate => fun my_app_state:stop_all_and_rebuild/0
}).

Why the drift check exists

ergon.jobs.pending_parents counts a job's incomplete workflow parents, and it is what jobs_fetch_idx carries in its predicate so that blocked jobs are absent from the index rather than scanned past. That denormalisation is what keeps checkout proportional to the batch size instead of to the blocked backlog, and it is also what makes the counter unverifiable by any normal means: nothing reads job_edges on the hot path any more.

So if the maintaining triggers have a bug, or someone writes the column by hand, a job is left either permanently unrunnable (count too high) or running before its parents finish (count too low). Neither surfaces anywhere. drift/0 recomputes the truth from job_edges and reports only what disagrees.

It doubles as a cross-check on the workflow graph: ergon_graph:ready_children/0 answers the same question through an entirely independent mechanism, a graph match rather than a trigger-maintained counter, so persistent disagreement between them means one of the two is wrong.

Summary

Functions

Jobs whose pending_parents disagrees with ergon.job_edges. Empty when healthy.

Rewrite pending_parents from ergon.job_edges wherever it disagrees. Returns the ids corrected.

Run the recovery flow with defaults: no queues, no hydrate, no repair.

Run the recovery flow.

Types

db_error()

-type db_error() ::
          empty_result | would_create_cycle |
          {job_not_found, ergon_job:job_id()} |
          {pgo_error, map()} |
          term().

drift_row()

-type drift_row() ::
          #{id := ergon_job:job_id(), actual := non_neg_integer(), expected := non_neg_integer()}.

opts()

-type opts() :: #{pgmq_queues => [binary()], hydrate => fun(() -> term()), repair => boolean()}.

pg_null()

-type pg_null() :: null.

queue_stats()

-type queue_stats() ::
          #{released_leases := non_neg_integer(),
            queue_length := non_neg_integer(),
            queue_visible_length := non_neg_integer(),
            oldest_msg_age_sec := number() | pg_null()}.

summary()

-type summary() ::
          #{hydrate := term(),
            pgmq := #{binary() => queue_stats() | {error, db_error()}},
            pending_parents_drift := [drift_row()],
            repaired := [ergon_job:job_id()] | not_attempted}.

Functions

drift()

-spec drift() -> [drift_row()].

Jobs whose pending_parents disagrees with ergon.job_edges. Empty when healthy.

Cheap enough to run on a schedule; the query returns only rows that disagree.

repair_drift()

-spec repair_drift() -> [ergon_job:job_id()].

Rewrite pending_parents from ergon.job_edges wherever it disagrees. Returns the ids corrected.

Writes only the rows that are actually wrong, which matters because each one fires the versioning trigger and accrues a history row.

run()

-spec run() -> summary().

Run the recovery flow with defaults: no queues, no hydrate, no repair.

run(Opts)

-spec run(opts()) -> summary().

Run the recovery flow.

Options:

  • pgmq_queues => [Name], whose leases to release and snapshot. Empty by default, since Ergon cannot know which pgmq queues are the host's.
  • hydrate => fun/0, the host's state rebuild. Defaults to a no-op, which is correct for a host that keeps no in-memory state.
  • repair => true, rewrite pending_parents where it has drifted. Off by default. Repairing silently would hide whatever caused the drift, and the drift itself is the more useful signal.