Docket.Backend.RunStore behaviour (docket v0.1.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: insertion and reads, atomic due claims, claim refresh and release, pre-execution claim abandons, advance commits, serialized signal mutations, and poison recovery. Those concerns cannot be split into independently configured stores because they mutate and fence the same aggregate.

Lifecycle code composes run and event writes inside one backend transaction; graph versions are saved separately before a run can reference them. This contract deliberately accepts neutral run proposals and schedule effects, never Docket.Checkpoint or Docket.Runtime.Moment values. A successful outer transaction is the point at which the transition becomes durable.

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.

Neutral proposal for one claim-fenced advance commit.

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

Trusted, normalized filters for one run collection read.

Decision returned by a pure serialized run mutation.

Operational result for an exhausted candidate that was not launched.

Storage effect applied with a committed run transition.

Callbacks

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

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

Commits one neutral runtime proposal under a mandatory token-and-sequence fence.

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

Inserts one initialized, already-durable run.

Reads the committed run with substrate-neutral operational information.

Lists lightweight run summaries under an explicit scope.

Serializes one short read, pure decision, and optional unclaimed run update.

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.

checkpoint_type()

@type checkpoint_type() :: atom()

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()

commit_proposal()

@type commit_proposal() :: %{
  run: Docket.Run.t(),
  expected_checkpoint_seq: non_neg_integer(),
  claim_token: claim_token(),
  checkpoint_type: checkpoint_type(),
  schedule: schedule()
}

Neutral proposal for one claim-fenced advance commit.

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.

mutation()

@type mutation() :: (Docket.Run.t() -> mutation_decision())

mutation_decision()

@type mutation_decision() ::
  {:commit, Docket.Run.t(), checkpoint_type(), schedule(), opaque :: term()}
  | {:no_change, opaque :: term()}
  | {:error, reason :: term()}

Decision returned by a pure serialized run mutation.

mutation_result()

@type mutation_result() ::
  {:ok, {:committed, term()} | {:unchanged, term()}} | {:error, term()}

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.

schedule()

@type schedule() ::
  :retain_claim
  | {:release_claim, :immediate | :external | :terminal | {:at, DateTime.t()}}

Storage effect applied with a committed run transition.

:retain_claim keeps the current token, refreshes its claimed time, and leaves the run without a wake. A release clears the token and claimed time. :immediate records a wake at the backend's current time, {:at, time} records a future or current wake, and :external or :terminal records no wake. The two nil-wake reasons remain distinct here so implementations can validate the proposed run status.

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.

commit(ctx, scope, commit_proposal)

@callback commit(ctx(), scope(), commit_proposal()) ::
  {:ok, Docket.Run.t()} | {:error, :stale_fence | :invalid_commit | :not_found}

Commits one neutral runtime proposal under a mandatory token-and-sequence fence.

The stored checkpoint sequence must equal proposal.expected_checkpoint_seq, the current claim must equal the non-empty proposal.claim_token, and the proposed run's sequence must be exactly expected_checkpoint_seq + 1. The checkpoint type must be a supported Docket checkpoint type. A stored sequence or token mismatch returns {:error, :stale_fence} without changing anything. A nil token, wrong proposed sequence, run identity mismatch, or invalid schedule/status combination returns {:error, :invalid_commit}.

Success replaces the run, records checkpoint_type, resets consecutive claim attempt and abandon counts and poison facts, and applies the schedule atomically. :retain_claim refreshes the current claim for a continuing vehicle; every release schedule clears it. This callback never appends events. Lifecycle code appends the proposal's already-assigned events through Docket.Backend.EventStore in the same outer transaction.

An unknown run or scope mismatch returns {:error, :not_found}. Proposal shape and exact-next-sequence validation happen before any run lookup, so :invalid_commit precedes :not_found: a malformed proposal returns :invalid_commit even when its run id would not be visible under the supplied scope, and only valid proposals return :not_found for an unknown or out-of-scope run.

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.

insert_run(ctx, owner_scope, t, checkpoint_type, wake_at)

@callback insert_run(
  ctx(),
  owner_scope(),
  Docket.Run.t(),
  checkpoint_type(),
  wake_at :: DateTime.t()
) :: {:ok, Docket.Run.t()} | {:error, term()}

Inserts one initialized, already-durable run.

owner_scope determines the stored tenant: :tenantless stores nil and {:tenant, tenant_id} stores that identifier. The initialized run must not have the transient :created status, must already carry its first committed sequence and start time, and requires checkpoint_type == :run_initialized. That type becomes the latest checkpoint metadata, and wake_at is the run's first explicit schedule.

This callback owns the run aggregate and may atomically materialize backend-specific supporting authority needed for that run. Supporting identity must come only from owner_scope; serialized run or payload fields cannot select it. Lifecycle orchestration appends assigned initialization events in the same outer transaction, while graph publication remains a separate, earlier operation.

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.

mutate_run(ctx, scope, run_id, mutation)

@callback mutate_run(ctx(), scope(), run_id :: String.t(), mutation()) ::
  mutation_result()

Serializes one short read, pure decision, and optional unclaimed run update.

The store checks scope before invoking mutation, loads and exclusively serializes the current committed run, and invokes the function while that serialization is held. The function must perform no external I/O.

A commit decision must propose the same run id and exactly the current checkpoint sequence plus one. Serialized mutation is the only unclaimed graph-run write path: it revokes any current claim and applies a release schedule atomically. :retain_claim is therefore invalid for this callback. On success the store resets claim attempt and abandon counts and poison facts and returns the opaque value tagged :committed; lifecycle code can then append events in the same outer transaction.

A no-change decision returns the opaque value tagged :unchanged and must not touch the row, claim, schedule, counters, timestamps, or event sequence. An error decision is returned unchanged and also changes nothing. An unknown run or scope mismatch returns {:error, :not_found} without invoking the function; malformed proposals return {:error, :invalid_mutation}.

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}.