Docket.Backend.RunStore behaviour (docket v0.3.0)

Copy Markdown View Source

Persistence contract for the durable run aggregate.

This capability owns every operation that enforces the run row's shared schedule, claim, fence, and poison invariants without publishing a transition: reads, atomic due claims, claim refresh and release, pre-execution claim abandons, and poison recovery. Those concerns cannot be split into independently configured stores because they mutate and fence the same aggregate.

Lifecycle writes use Docket.Backend.TransitionStore. This contract owns focused reads and claim operations only.

Summary

Types

Policy for one pre-execution claim abandon.

Disposition applied by one pre-execution claim abandon.

Results of one claim scan. Poisoned candidates never appear as leases.

Lightweight authority returned for a run that should be launched.

Policy for one atomic due-claim scan.

Stable newest-first run-list cursor: {started_at, run_id}.

Trusted, normalized filters for one run collection read.

Operational result for an exhausted candidate that was not launched.

Callbacks

Hands back a claim whose vehicle could not begin node execution.

Atomically claims a batch of ready or expired runs for system execution.

Reads the last committed graph-run document under an explicit scope.

Reads the committed run with substrate-neutral operational information.

Lists lightweight run summaries under an explicit scope.

Refreshes claim liveness under the exact current token.

Releases the claim under the exact current token.

Recovers a non-terminal poisoned run's backend-owned operational state.

Types

abandon_policy()

@type abandon_policy() :: %{
  :expected_checkpoint_seq => non_neg_integer(),
  :now => DateTime.t(),
  :retry_at => DateTime.t(),
  :max_claim_abandons => pos_integer(),
  optional(:non_poisoning) => boolean(),
  optional(:backoff) => %{base_ms: pos_integer(), cap_ms: pos_integer()}
}

Policy for one pre-execution claim abandon.

expected_checkpoint_seq is the lease's committed sequence, now is the caller's clock reading, retry_at is the future wake the abandoned run receives, and max_claim_abandons bounds consecutive abandons before the run is poisoned instead of rescheduled.

A :non_poisoning abandon reverses the claim-attempt increment, counts one claim abandon, and never poisons regardless of the count. When :backoff is present the store ignores retry_at and computes the wake from its durable abandon count as now + min(base_ms * 2^claim_abandons, cap_ms).

abandon_result()

@type abandon_result() :: {:ok, :rescheduled | :poisoned | :stale}

Disposition applied by one pre-execution claim abandon.

claim_batch()

@type claim_batch() :: %{leases: [claim_lease()], poisoned: [poisoned_claim()]}

Results of one claim scan. Poisoned candidates never appear as leases.

claim_lease()

@type claim_lease() :: %{
  run_id: String.t(),
  owner_scope: owner_scope(),
  graph_id: String.t(),
  graph_hash: String.t(),
  checkpoint_seq: non_neg_integer(),
  claim_token: claim_token(),
  claimed_at: DateTime.t(),
  claim_attempt: pos_integer(),
  orphan_ttl_ms: non_neg_integer()
}

Lightweight authority returned for a run that should be launched.

orphan_ttl_ms echoes the claiming policy's TTL so the holder knows the staleness bound its own claim expires under. It is informational per-claim state, not a promise that the TTL cannot differ on a later claim.

claim_policy()

@type claim_policy() :: %{
  :now => DateTime.t(),
  :limit => pos_integer(),
  :orphan_ttl_ms => non_neg_integer(),
  :max_claim_attempts => pos_integer(),
  optional(:preference) => :ready | :expired | nil
}

Policy for one atomic due-claim scan.

:preference is an optional, advisory demand-1 hint naming the candidate class (:ready or :expired) served first when limit is exactly one. It is operational and local - dispatchers alternate it across consecutive polls and lose the phase on restart - never durable run state. Absent (or at any other demand) selection is class-neutral.

claim_token()

@type claim_token() :: nonempty_binary()

ctx()

@type ctx() :: Docket.Backend.ctx()

list_cursor()

@type list_cursor() :: Docket.RunPage.cursor()

Stable newest-first run-list cursor: {started_at, run_id}.

list_query()

@type list_query() :: %{
  limit: pos_integer(),
  before: list_cursor() | nil,
  graph_id: String.t() | nil,
  graph_hash: String.t() | nil,
  statuses: [Docket.Run.durable_status()] | nil
}

Trusted, normalized filters for one run collection read.

Public callers are normalized into this shape before reaching storage. before is exclusive, and statuses is either nil or a non-empty list containing only durable run statuses.

owner_scope()

@type owner_scope() :: Docket.Backend.owner_scope()

poisoned_claim()

@type poisoned_claim() :: %{
  run_id: String.t(),
  poisoned_at: DateTime.t(),
  poison_reason: String.t()
}

Operational result for an exhausted candidate that was not launched.

scope()

@type scope() :: Docket.Backend.scope()

Callbacks

abandon_claim(ctx, atom, run_id, claim_token, abandon_policy)

@callback abandon_claim(
  ctx(),
  :system,
  run_id :: String.t(),
  claim_token(),
  abandon_policy()
) :: abandon_result()

Hands back a claim whose vehicle could not begin node execution.

This is the disposition for deterministic pre-execution failure — above all graph compilation incompatibility, where the executing node cannot compile the stored effective graph against its locally installed node contracts. That is a deployment-compatibility condition, not node execution failure, so it must consume neither the run's execution-attempt budget nor be reported through the run's failure machinery. Transient infrastructure errors (for example a failed graph fetch) are not abandons; they use ordinary release_claim/5 or crash into claim expiry.

The operation is fenced on the exact current token and the lease's committed checkpoint sequence, so it applies only before the holder's first lifecycle commit. A matched abandon atomically:

  • clears the claim token and claimed time;
  • hands the acquisition increment back by decrementing the claim-attempt count (floored at zero), keeping poison-by-attempts exclusively about claims that reached execution;
  • increments the consecutive claim-abandon count; and
  • either records policy.retry_at as the next wake and returns {:ok, :rescheduled}, or — when the abandon count had already reached policy.max_claim_abandons — records paired poison facts at policy.now with reason "max_claim_abandons_exceeded" and returns {:ok, :poisoned} so persistent incompatibility becomes an explicit operator concern instead of unbounded retry.

It never touches the committed run document, checkpoint_seq, or the run's updated_at. A stale token, an advanced sequence, or an unknown run changes nothing and returns {:ok, :stale}; after a steal or a serialized signal commit the abandon can never disturb the winning claim or schedule. Every return is success-shaped because the caller's next action is identical in all cases — stop without committing; the atom is for telemetry. The abandon count resets on any committed run mutation and on poison recovery.

policy.retry_at must not precede policy.now: the future wake is what keeps an incompatible node from immediately re-claiming the same run. Callers should add jitter to their retry backoff so runs abandoned together do not become due together. A rolling deployment therefore self-heals — a compatible node claims the run at or after retry_at — while a fleet that can never compile the graph poisons it after a bounded number of abandons. Both timestamps are normalized like every other operational timestamp.

claim_due(ctx, atom, claim_policy)

@callback claim_due(ctx(), :system, claim_policy()) ::
  {:ok, claim_batch()} | {:error, term()}

Atomically claims a batch of ready or expired runs for system execution.

Ready candidates are non-poisoned running runs with no claim and a wake at or before policy.now. Expired candidates are non-poisoned running runs whose current claim is older than policy.orphan_ttl_ms relative to that same timestamp.

For each selected candidate whose current claim-attempt count is below max_claim_attempts, the store assigns a fresh token, sets claimed_at to policy.now, clears wake_at, increments the count, and returns a lease. If the count has already reached the maximum, the store clears claim and wake, records paired poison facts at policy.now, and returns a poison result without a lease. A maximum of three therefore permits exactly three claims that proceed into node execution; a fourth recovery need poisons the run. Claims handed back through abandon_claim/5 before execution began do not count toward that maximum — they are bounded separately by the abandon policy.

Selection, attempt accounting, claim steal, and poison mutation occur in the same atomic operation. The total number of returned outcomes is at most policy.limit.

Selection keeps both continuously eligible classes making progress:

  • With limit >= 2 and both classes non-empty, at least one outcome (lease or poison - the reservation is per outcome, not per lease) goes to each class. The remaining demand follows the implementation's admission order with a stable (eligible_at, id) tie-break inside equal scheduling priority. Legacy uses global age-first order; a partition-fair implementation may put bounded partition rotation ahead of global age.
  • With limit == 1, the optional policy.preference class is served first, falling through to the other class when the preferred one is empty so preference never wastes demand. Without a preference, stable age/ID order decides among rows with equal implementation scheduling priority; a partition-fair implementation may put its partition order ahead of global age.

Eligibility-time ordering provides aging, not fairness. Unless a selected implementation publishes a narrower conditional contract, implementations promise no strict FIFO, bounded queue wait, tenant or workload-class fairness, or starvation freedom under sustained arrivals or persistent row locks. Concurrent claimants may skip locked rows, so per-class progress holds only up to rows another transaction currently holds.

fetch_run(ctx, scope, run_id)

@callback fetch_run(ctx(), scope(), run_id :: String.t()) ::
  {:ok, Docket.Run.t()} | {:error, :not_found}

Reads the last committed graph-run document under an explicit scope.

An unknown run or scope mismatch returns {:error, :not_found}. Backend/infrastructure failure and corrupt persisted run state are not collapsed into absence; implementations raise in those cases.

inspect_run(ctx, scope, run_id)

@callback inspect_run(ctx(), scope(), run_id :: String.t()) ::
  {:ok, Docket.RunInfo.t()} | {:error, :not_found}

Reads the committed run with substrate-neutral operational information.

The Docket.RunInfo projection exposes the wake, claimed time, claim-attempt count, and poison facts, but never the current claim token. An unknown run or scope mismatch returns {:error, :not_found}. Backend/infrastructure failure and corrupt persisted run state raise rather than being reported as :not_found.

list_runs(ctx, scope, list_query)

@callback list_runs(ctx(), scope(), list_query()) :: {:ok, Docket.RunPage.t()}

Lists lightweight run summaries under an explicit scope.

Results are ordered newest first by immutable (started_at, run_id) and use an exclusive keyset cursor. Scope and every requested filter must be enforced by the backing query; implementations must not load out-of-scope rows and filter them in application memory. The full durable run state is deliberately absent from Docket.RunSummary and must not be decoded for a collection read.

No matching rows is successful and returns an empty Docket.RunPage whose next_before preserves the supplied cursor and whose has_more? is false. Page construction trusts implementations to uphold ordering and cursor exclusivity rather than scanning backend output again.

refresh_claim(ctx, atom, run_id, claim_token, now)

@callback refresh_claim(
  ctx(),
  :system,
  run_id :: String.t(),
  claim_token(),
  now :: DateTime.t()
) :: :ok | {:error, :claim_lost}

Refreshes claim liveness under the exact current token.

Success advances claimed_at, and never backward: the stored time only moves toward the backend's authoritative current time, so a refresh that was delayed in flight cannot regress a fresher value written by a :retain_claim commit and re-expose the run to steal. now is the caller's clock reading; backends with an authoritative clock of their own may prefer that clock over now. A stale token, missing run, or lost claim returns {:error, :claim_lost} and changes nothing.

release_claim(ctx, atom, run_id, claim_token, now)

@callback release_claim(
  ctx(),
  :system,
  run_id :: String.t(),
  claim_token(),
  now :: DateTime.t()
) :: :ok

Releases the claim under the exact current token.

Release is idempotent. A stale token or missing run returns :ok without disturbing a newer claim or the run's schedule. A matching release clears the claim token and claimed time and records an immediate wake at now, so it cannot strand a non-poisoned running row without a claim or schedule.

retry_poisoned_run(ctx, scope, run_id, now)

@callback retry_poisoned_run(ctx(), scope(), run_id :: String.t(), now :: DateTime.t()) ::
  {:ok, Docket.Run.t()} | {:error, :not_found | :inactive_run}

Recovers a non-terminal poisoned run's backend-owned operational state.

Terminal status is checked first and returns {:error, :inactive_run}. Otherwise a poisoned run has its poison facts and claim cleared, its claim attempt and abandon counts reset, and an immediate wake recorded at now. Calling this for an already-unpoisoned non-terminal run is idempotent success and changes nothing. The command consumes neither commit nor event sequence.

An unknown run or scope mismatch returns {:error, :not_found}.