Bedrock.ControlPlane.Distributor.Recruitment (bedrock v0.7.0)

View Source

On-demand materializer recruitment for the distributor: node selection, worker creation via the node's Foreman, epoch lock, and unlock with the shard's typed pull sources — the same replica set the commit proxies route with, resolved through ShardRouter, exactly as recovery's bootstrap seeds its unlocks.

A worker that was created but never reached service (lock or unlock failed) is removed again before the error returns, so failed recruitment does not leak idle or half-locked workers. Publication of the recruit into the materializers/ family is the CALLER's job (a check-fenced commit); a recruit whose publication aborts is an orphan and is removed the same way — commit abort replaces phase-a's delta-rejection as the orphan-cleanup trigger.

Summary

Types

context()

The recruitment context. logs and log_refs are the epoch's log wiring, handed by the director at recruit time (the same runtime wiring recover_from hands proxies): pull sources derive from them through the single ShardRouter placement site.

Functions

adopt(tag, worker_id, node, context)

Adopts a family-named materializer into the current epoch: recruitment minus creation. The worker already exists (the committed materializers/ family names it — the keyspace is the membership authority) but was never locked into this epoch, typically because its node missed recovery's roll call and rejoined later. It is locked at the epoch and unlocked at the durable version IT reports, so it resumes pulling from exactly where its own store left off.

recruit(tag, context)

Recruits a materializer for the given shard tag. Returns the live pid, the node it was placed on, and the worker id it was created under (so the caller can remove the worker if fencing the recruit into the family fails).

remove_orphaned_worker(worker_id, node, context)

Best-effort removal of a worker left behind by a failed recruitment or an aborted publication. The worker never carried data a client could reach, so removal is safe; any failure to remove it is logged and swallowed — orphan cleanup must never mask the original error.

Types

context()

@type context() :: %{
  :cluster => module(),
  :epoch => Bedrock.epoch(),
  :node_capabilities => %{required(Bedrock.Cluster.capability()) => [node()]},
  :logs => %{required(Bedrock.DataPlane.Log.id()) => [Bedrock.range_tag()]},
  :log_refs => %{
    required(Bedrock.DataPlane.Log.id()) => Bedrock.DataPlane.Log.ref()
  },
  optional(:worker_params) => %{required(String.t()) => term()},
  optional(:create_worker_fn) => fun(),
  optional(:lock_materializer_fn) => fun(),
  optional(:unlock_materializer_fn) => fun(),
  optional(:remove_worker_fn) => fun()
}

The recruitment context. logs and log_refs are the epoch's log wiring, handed by the director at recruit time (the same runtime wiring recover_from hands proxies): pull sources derive from them through the single ShardRouter placement site.

Functions

adopt(tag, worker_id, node, context)

@spec adopt(Bedrock.range_tag(), Bedrock.Service.Worker.id(), node(), context()) ::
  {:ok, pid(), node(), Bedrock.Service.Worker.id()} | {:error, term()}

Adopts a family-named materializer into the current epoch: recruitment minus creation. The worker already exists (the committed materializers/ family names it — the keyspace is the membership authority) but was never locked into this epoch, typically because its node missed recovery's roll call and rejoined later. It is locked at the epoch and unlocked at the durable version IT reports, so it resumes pulling from exactly where its own store left off.

Unlike a failed recruitment, a failed adoption never removes the worker: it pre-exists this attempt and holds real state — enforced structurally by there being no removal call on this path. The caller heals the tag instead, and healing CLEARS this worker's own key — a clear the proxy privatizes onto the shard's stream, so the worker retires in-band, at the version its assignment ends (bedrock-q67.21.6).

recruit(tag, context)

@spec recruit(Bedrock.range_tag(), context()) ::
  {:ok, pid(), node(), Bedrock.Service.Worker.id()} | {:error, term()}

Recruits a materializer for the given shard tag. Returns the live pid, the node it was placed on, and the worker id it was created under (so the caller can remove the worker if fencing the recruit into the family fails).

remove_orphaned_worker(worker_id, node, context)

@spec remove_orphaned_worker(Bedrock.Service.Worker.id(), node(), context()) :: :ok

Best-effort removal of a worker left behind by a failed recruitment or an aborted publication. The worker never carried data a client could reach, so removal is safe; any failure to remove it is logged and swallowed — orphan cleanup must never mask the original error.