@spec checkpoint(name()) :: checkpoint() | nil

Reads name's checkpoint, or nil when it has never written one.

@spec clear_checkpoint(name()) :: :ok

Deletes name's checkpoint row entirely.

The next run starts from the beginning of its scan. That is safe for every operation in this package, because re-processing an item finds it already done, but it is not free: the whole scan runs again.

It also clears a pause, because the pause lives in the same row. An operator who wants to rewind a cursor without resuming a paused operation wants put_checkpoint(name, cursor: nil) instead, which is what the workers themselves use when a scan completes.

@spec list() :: [checkpoint()]

Every checkpoint this installation holds, in name order.

Includes the rows that are not operations ("events_projection", "schema:core"), because an operator listing checkpoints wants to see the table rather than this module's idea of it.

@spec pause(name()) :: :ok

Asks name to stop at its next batch boundary.

Creates the row when the operation has never run, so an operation can be paused before it is first scheduled. Leaves cursor and counts exactly as they are, so a paused operation keeps its position.

@spec paused?(name()) :: boolean()

Whether name is paused. false for an operation that has never run.

@spec put_checkpoint(
  name(),
  keyword()
) :: :ok

Writes name's cursor and counts, leaving any pause alone.

Attributes: :cursor (a map, or nil to say the scan is finished) and :counts (a map). At least one is required. A field that is not given is not written.

A nil cursor is stored as the empty object, because core schema version 7 declares the column NOT NULL, and checkpoint/1 reads an empty cursor back as nil. The two are the same statement: this scan has no position.

This is an operator tool with an edge. Writing a cursor by hand tells a sweep that everything before it is done. Moving one forwards skips work; moving one backwards re-runs work, which is safe only because every operation using this module has idempotent effects. Prefer clear_checkpoint/1 if what you want is "start again".

@spec resume(name()) :: :ok

Clears a pause. The next run starts from the cursor, not from the beginning.

@spec run_batches(name(), keyword(), (map() | nil ->
                                  {:ok, batch()} | {:error, term()})) ::
  {:ok, report()} | {:paused, report()} | {:error, term()}

Runs fun over bounded batches, carrying the cursor in name's checkpoint.

fun is called with the current cursor (a map, or nil at the start of a scan) and returns {:ok, %{cursor: cursor, counts: counts}} or {:error, reason}. A cursor of nil means the scan reached its end.

The loop is the whole of lower-level invariants L05c-1 to L05c-3, in one place rather than once per worker:

1. The pause is read before every batch, including the first, so pausing takes effect within one batch rather than one run.
2. The cursor lives in the checkpoint row and is never a job argument, so two runs of the same operation read the same position and the second one resumes where the first got to rather than restarting from a stale one.
3. A per-item failure is fun's to count and to step over. Only a failure of the batch mechanism itself, the listing query or the checkpoint write, ends the run with {:error, reason}.

Options:

• :max_batches (default 10) - how many batches one call runs before returning with work still waiting. A run that stopped here leaves its cursor in place, so the next tick continues.
• :counts (default %{}) - the counter map batches accumulate into.

Returns {:ok, report}, {:paused, report} or {:error, reason}.

Emits [:aurora_meter, :operations, :batch] per batch with measurements %{items: integer, duration_ms: integer} and metadata %{name: name, result: :ok | :error}. The duration is an in-memory span, so it is measured with AuroraMeter.Clock.monotonic_ms/0 and not with any clock that can step backwards (AuroraMeter.Clock).