AuroraMeter.
View Source
The resumable cursor every bounded task in Aurora Meter keeps.
One row of aurora_meter_checkpoints per task, named by the task:
"events_backfill", "events_projection", "events_replay:<generation>",
"schema:core". A row carries a cursor (where the task got to), counts
(what it has done so far), a state and an updated_at. A task that is
killed and restarted reads its row and carries on; an operator reads it to
see where a stalled task stopped.
This is repo-direct rather than a AuroraMeter.Storage callback, for the
reason AuroraMeter.Credits is (ADR 0005): it is operational bookkeeping for
core's own tooling, not adapter-visible state, and a third-party adapter
would have no reason to implement six callbacks that exist only to serve a
backfill.
The state column
state is the task lifecycle, a free string so a later task can add its own
vocabulary without a migration. The values this release writes are "idle",
"running", "paused", "active" (the projection row, which is not a task)
and "applied" (the "schema:core" marker).
pause/1 and resume/1 set "paused" and "idle". A running task reads
paused?/1 at a batch boundary and stops cleanly; it never interrupts a
batch. AuroraMeter.Operations is the documented operator surface over the
same rows, and is what a host and Aurora Meter Pro use; this module is the
table's own, and is what core's backfill, replay and migration tasks use
because their names predate the "<operation>:<scope>" shape Operations
enforces.
The state column is not a lock
A state of "running" says a task believed it was running when it last
wrote. It is a report, and nothing in this package decides a correctness
question from it or from the age of updated_at. A clock cannot decide
exclusion here: clock_timestamp() is the database host's OS clock and it is
corrected, so a short lease compared against it can invert. Exclusion between
two runners is taken with a Postgres advisory lock instead (claim/3, and
AuroraMeter.Events.Backfill), which has no clock in it at all.
updated_at is stamped by the database, with clock_timestamp() in the
statement, so any comparison against it is a comparison of two readings of
one clock. Read the other side with AuroraMeter.Clock.db_now/0.
The heartbeat is not a lease
heartbeat/3 stamps cursor["heartbeat_at"] and cursor["runner"] so an
operator can tell a stalled task from a finished one. Nothing in this
package decides an exclusion question from its age. open-findings.md X100
measured clock_timestamp() stepping backwards nine times in 300 seconds,
worst 439 ms, on a 32.5 second cadence, so "the heartbeat is older than N
seconds" is not a sound test for "the runner is gone" at any N a person would
pick. The sound test is claim/3: a session advisory lock dies with the
connection that held it, so a runner that was killed has already released it
and a runner that is alive has not, with no duration anywhere in the decision.
The rule is about exclusion at seconds scale, and that boundary is worth
stating because one caller is on the other side of it.
AuroraMeter.Retention does compare a "flush:<node>" row's age against a
cutoff, and it is sound there for the reason architecture-map.md section 3
gives: unsafe at seconds, safe at minutes and hours. The smallest window it
will act on is one day, refused below that at boot, which is about 200,000
times the measured backwards step; both sides of that comparison are stamped
by clock_timestamp(); and the decision it feeds is a veto on deleting
data rather than a licence to take another runner's work. Its moduledoc
carries the arithmetic. A decision at seconds scale still has no business
reading this column.
The missing table
get/1 returns nil when aurora_meter_checkpoints does not exist, because
that is the state of every database below core schema version 7 and
get("schema:core") == nil is exactly how a caller detects one. Every write
raises on a missing table: writing a checkpoint to a pre-V7 database is a
bug, not a state to tolerate.
Summary
Functions
Lists every checkpoint, name order. [] when the table is absent.
Runs fun while holding a Postgres session advisory lock derived from
name, on one pinned connection, and returns {:ok, fun.()}.
Removes a checkpoint. Succeeds whether or not the row was there.
Reads one checkpoint, or nil when there is no such row.
Stamps cursor["heartbeat_at"] and cursor["runner"] and sets the state to
"running", leaving every other cursor key and the counts alone.
Asks a task to stop at its next batch boundary.
Whether name is paused. false when there is no row and when the table is
absent.
Writes cursor, counts and state for name, creating the row if needed.
Writes cursor and counts for name without touching state, creating
the row as "idle" when it does not exist.
Clears a pause. The next run of the task starts from its checkpoint.
The runner identity heartbeat/2 stamps: this node and this process.
Merges the given fields into an existing row, leaving the others alone.
Types
Functions
Lists every checkpoint, name order. [] when the table is absent.
Options: :repo.
@spec claim(String.t(), (-> result), keyword()) :: {:ok, result} | {:error, :already_running} when result: term()
Runs fun while holding a Postgres session advisory lock derived from
name, on one pinned connection, and returns {:ok, fun.()}.
Returns {:error, :already_running} without calling fun when another
connection holds it. This is the exclusion every bounded task in this package
uses, and it has no clock in it: the lock is granted or it is not, and it is
released by pg_advisory_unlock, by the connection closing or by the process
that held it dying. A runner killed with kill -9 therefore leaves nothing
behind to time out (open-findings.md X100).
The lock key is {0x4155524E, :erlang.phash2(name)}. The namespace is
distinct from the fixed-key namespace 0x4155524F that
AuroraMeter.Events.Backfill uses, so a name can never collide with a task
that took a hand-picked key. Two different names that hash alike would
exclude each other, which is over-exclusion rather than a correctness
failure, and there are four names in this package.
Options: :repo, :timeout (per statement, default 15_000).
Removes a checkpoint. Succeeds whether or not the row was there.
Options: :repo.
Reads one checkpoint, or nil when there is no such row.
Also nil when the table itself is absent, which is every database below
core schema version 7.
Options: :repo (defaults to the configured repo), for a caller that is
inside a migration and must use the migrator's repo.
Examples
iex> AuroraMeter.Checkpoints.get("no_such_task")
nilStamps cursor["heartbeat_at"] and cursor["runner"] and sets the state to
"running", leaving every other cursor key and the counts alone.
The instant is clock_timestamp(), written by the database in the same
statement, so it is never a node's reading of the time. Call it inside the
transaction that commits a batch, so a heartbeat can never be fresher than
the work it claims to be reporting.
It is a report for a human, not a lease: see the module documentation.
Returns {:error, :not_found} when there is no such row.
Options: :repo, and :runner (default "<node>/<pid>").
Asks a task to stop at its next batch boundary.
Creates the row when the task has not run yet, so a task can be paused before it is started. Never interrupts a batch in flight.
Options: :repo.
Whether name is paused. false when there is no row and when the table is
absent.
Options: :repo.
Examples
iex> AuroraMeter.Checkpoints.paused?("no_such_task")
falseWrites cursor, counts and state for name, creating the row if needed.
updated_at is stamped by the database. Call this inside the same
transaction as the work it records, so the cursor can never run ahead of the
work.
Options: :repo.
Writes cursor and counts for name without touching state, creating
the row as "idle" when it does not exist.
Either field may be :absent, which leaves it as it is. A cursor of nil
is a value rather than an omission: it is how a run says its scan reached the
end, and it is stored as the empty object because the column is NOT NULL.
This is what a batch loop advancing a cursor uses, so that an operator pausing
the operation mid run cannot be clobbered by the batch that was already in
flight. Options: :repo.
Clears a pause. The next run of the task starts from its checkpoint.
Options: :repo.
@spec runner() :: String.t()
The runner identity heartbeat/2 stamps: this node and this process.
Informational. Nothing decides anything from it; it is what an operator reads to find the machine a stalled task is on.
Merges the given fields into an existing row, leaving the others alone.
Returns {:error, :not_found} when there is no such row; it never creates
one, because a partial update of a checkpoint that does not exist is a
caller mistake rather than a state worth inventing.
Options: :repo.
Examples
iex> AuroraMeter.Checkpoints.update("no_such_task", state: "paused")
{:error, :not_found}