Docket.Postgres.ClaimPolicy behaviour (docket v0.1.0)

Instance-resolved PostgreSQL admission-plan boundary.

Docket.Postgres.RunStore.claim_due/3 is the sole admission entrypoint and the sole executor of policy plans. A selected implementation receives normalized policy values and pre-quoted identifiers, builds one data-only SQL plan, decodes that statement's rows, and owns its bounded admission observations. A policy may also implement one narrow startup configuration callback. Plan construction never receives a run-store module or query callback.

Summary

Types

claim_batch()

claim_result()

init_context()

plan_context()

runtime_input()

t()

Callbacks

build_plan(plan_context, runtime_input, state)

configure(ctx, state, query)

decode(rows, decoder, state)

init(keyword, init_context)

observe(plan_observation, decoded_observation, claim_result, duration, state)

Types

claim_batch()

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

claim_result()

@type claim_result() :: {:ok, claim_batch()} | {:error, term()}

init_context()

@type init_context() :: %{
  prefix: String.t() | nil,
  identifiers: %{
    runs: String.t(),
    claim_policy: String.t(),
    claim_partitions: String.t()
  }
}

plan_context()

@type plan_context() :: %{
  prefix: String.t() | nil,
  identifiers: %{
    runs: String.t(),
    claim_policy: String.t(),
    claim_partitions: String.t()
  }
}

runtime_input()

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

t()

@opaque t()

Callbacks

build_plan(plan_context, runtime_input, state)

@callback build_plan(plan_context(), runtime_input(), state :: term()) ::
  Docket.Postgres.ClaimPolicy.Plan.t()

configure(ctx, state, query)

(optional)

@callback configure(
  Docket.Backend.ctx(),
  state :: term(),
  query :: (String.t(), [term()] -> {:ok, term()} | {:error, term()})
) :: :ok | {:error, term()}

decode(rows, decoder, state)

@callback decode(rows :: [list()], decoder :: term(), state :: term()) ::
  {:ok, claim_batch(), observation :: map()}
  | {:error, reason :: term(), observation :: map()}

init(keyword, init_context)

@callback init(
  keyword(),
  init_context()
) :: {:ok, state :: term()} | {:error, term()}

observe(plan_observation, decoded_observation, claim_result, duration, state)

@callback observe(
  plan_observation :: map(),
  decoded_observation :: map() | nil,
  claim_result(),
  duration :: integer(),
  state :: term()
) :: :ok