Docket.Postgres.Pruner (docket v0.1.0-dev)

Copy Markdown View Source

Periodic, bounded retention for Postgres events, terminal runs, and graph versions.

A pass deletes persisted events before their run when event retention is shorter, then deletes terminal runs and relies on the event foreign key to cascade any remaining events. It deletes graph versions only when no run references them and they rank older than the newest ten publications for their owner scope and graph_id. Revision order is immutable (inserted_at, graph_hash), matching the public graph-version API. The newest ten revisions per owner therefore survive even when they have no runs, and one tenant's publications cannot evict another tenant's history.

Passes are serialized per database schema with a transaction-scoped advisory lock. A competing node reports a skipped pass instead of waiting. Candidate selection is also bounded and uses SKIP LOCKED, so application row locks do not stall retention.

The batch size bounds explicitly selected event and run rows. A run may own more than one batch of events, so foreign-key cascade work is deliberately not claimed to be bounded by the batch size.

Each pass emits [:docket, :postgres, :pruner, :pass]. Its numeric measurements are events_deleted, runs_deleted, cascade_events_deleted, graphs_deleted, and duration; duration is in native System.monotonic_time/0 units, following the :telemetry span convention. Metadata contains only the bounded result dimension.

Summary

Types

counts()
ctx()
option()
policy()
result()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

prune(ctx, policy)

Runs one bounded retention pass and emits committed count telemetry.

start_link(opts)

Types

counts()

@type counts() :: %{
  events_deleted: non_neg_integer(),
  runs_deleted: non_neg_integer(),
  cascade_events_deleted: non_neg_integer(),
  graphs_deleted: non_neg_integer()
}

ctx()

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

option()

@type option() ::
  {:name, GenServer.name()}
  | {:context, ctx()}
  | {:interval_ms, pos_integer()}
  | {:event_retention_ms, non_neg_integer()}
  | {:run_retention_ms, non_neg_integer()}
  | {:batch_size, pos_integer()}
  | {:clock, (-> DateTime.t() | :database)}

policy()

@type policy() :: %{
  now: DateTime.t() | :database,
  event_retention_ms: non_neg_integer(),
  run_retention_ms: non_neg_integer(),
  batch_size: pos_integer()
}

result()

@type result() :: {:ok, counts()} | {:skipped, :locked} | {:error, term()}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

prune(ctx, policy)

@spec prune(ctx(), policy()) :: result()

Runs one bounded retention pass and emits committed count telemetry.

start_link(opts)

@spec start_link([option()]) :: GenServer.on_start()