Orkestra.Projector.GenServer (orkestra v0.2.0)

Copy Markdown View Source

Runtime GenServer for Orkestra projection subsystem.

Subscribes to the event store from the persisted checkpoint position, applies pushed events strictly sequentially (one per handle_info — OTP mailbox guarantees in-order, single-consumer processing — PROJ-04), commits the read-model write and the checkpoint upsert atomically in a single Ecto.Multi transaction (STORE-03), retries failed events with exponential backoff via Process.send_after, and parks exhausted events to dead-letter while staying alive in an idle halted state (ERR-04).

Start Configuration

Pass a map with the following keys when starting the GenServer:

%{
  repo:             MyApp.OrderProjection.Repo,   # per-projection Ecto.Repo
  projector_name:   "MyApp.OrderProjection",      # unique string identifier
  storage_adapter:  Orkestra.Projection.Storage.Postgres,  # Storage behaviour
  event_store:      Orkestra.EventStore.InMemory,  # EventStore behaviour
  lifecycle_config: %{max_retries: 5, backoff_base_ms: 500, backoff_cap_ms: 30_000},
  adapter_opts:     [handler: &my_handler/3]       # forwarded to storage_adapter.write/4
}

An optional :name key sets the GenServer registered name.

Deferred Init

init/1 does not call the Repo directly. Instead it enqueues a :load_checkpoint message so callers (e.g., tests) can call Ecto.Adapters.SQL.Sandbox.allow/3 after start_supervised!/1 returns but before the GenServer processes its first mailbox message (RESEARCH Pitfall 1).

Halt Behaviour

A halted projector discards incoming events and stays alive — it never returns a stop tuple from handle_info. This avoids supervisor restart loops and keeps the halted status observable via the persisted projection_checkpoints row.

Step Name Convention

The GenServer's checkpoint Multi steps are named :checkpoint, :halted_checkpoint, and :dead_letter. The injected storage adapter must use :read_model_-prefixed step names to prevent Ecto.Multi.append/2 name clashes (RESEARCH Pitfall 2).

Summary

Types

Internal GenServer state.

Functions

Returns a specification to start this module under a supervisor.

Starts the Projector GenServer linked to the calling process.

Types

state()

@type state() :: %{
  repo: module(),
  projector_name: String.t(),
  storage_adapter: module(),
  event_store: module(),
  lifecycle_config: Orkestra.Projector.Lifecycle.config(),
  adapter_opts: keyword(),
  subscription_ref: reference() | nil,
  attempts: non_neg_integer(),
  halted: boolean(),
  writes_paused: boolean(),
  last_seen_position: non_neg_integer() | nil,
  rebuild_total: non_neg_integer() | nil,
  rebuild_events_replayed: non_neg_integer(),
  es_buffer: list(),
  es_batch_size: non_neg_integer(),
  es_mode: :live | :catching_up
}

Internal GenServer state.

All fields except the runtime fields (subscription_ref, attempts, halted) are populated from the start config and are immutable for the lifetime of the process.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

start_link(config)

@spec start_link(map()) :: GenServer.on_start()

Starts the Projector GenServer linked to the calling process.

config must be a map with the keys listed in the module doc. Pass an optional :name key to register the process under a name.