Bedrock.DataPlane.Materializer.Olivine.Streaming (bedrock v0.7.0)

View Source

The materializer's stream puller: one loop, from snapshot to live.

Pulls this materializer's shard stream from a Demux ShardServer — chunks for history, buffer for recent data, seamlessly — instead of drinking the whole WAL from the log. The materializer's only contact with a log is discovery: Log.get_shard_server/2, once per session (and again on failover).

Discovery

The replica set arrives in the unlock seed: [{log_id, log_ref}], resolved once by the director through the same ShardRouter walk the commit proxies route with (ShardRouter.log_ids_for_tag/3) — the puller never re-derives placement. Any replica works: every replica sees the same slices and the same cuts, so their chunks are byte-identical and their buffers hold the same suffix. Failover walks the replica set with the same circuit-breaker pattern the old log puller used.

Currency

An empty pull reply that carries a high-water — "nothing for you, but you are current through v" — is rematerialized as a heartbeat: an empty transaction at v fed through the normal apply path. Version advancement, read wake-ups, and window math all ride existing code, unchanged.

Backpressure

Batches are handed to the owner through a synchronous ingest function whose reply the owner withholds while its intake queue is over high-water. The puller cannot outrun the applier because the applier holds the reply.

Summary

Types

ingest_fn()

@type ingest_fn() :: ([Bedrock.DataPlane.Transaction.encoded()],
                kcv :: Bedrock.version() | nil ->
                  :ok)

puller_state()

@type puller_state() :: %{
  shard_num: non_neg_integer(),
  next_version: Bedrock.version(),
  sources: [source()],
  failed_logs: %{required(Bedrock.DataPlane.Log.id()) => integer()},
  shard_server: pid() | nil,
  current_log_id: Bedrock.DataPlane.Log.id() | nil,
  ingest_fn: ingest_fn()
}

source()

Functions

call_timeout()

@spec call_timeout() :: pos_integer()

circuit_breaker_timeout()

@spec circuit_breaker_timeout() :: pos_integer()

ensure_shard_server(state)

@spec ensure_shard_server(puller_state()) ::
  {:ok, puller_state()} | {:wait, puller_state()}

heartbeat_at(version)

mark_log_as_failed(state, log_id)

@spec mark_log_as_failed(puller_state(), Bedrock.DataPlane.Log.id()) :: puller_state()

pull_limit()

@spec pull_limit() :: pos_integer()

pull_once(state)

@spec pull_once(puller_state()) :: puller_state()

reset_failed_logs(state)

@spec reset_failed_logs(puller_state()) :: puller_state()

retry_delay()

@spec retry_delay() :: pos_integer()

select_log(map)

@spec select_log(puller_state()) :: {:ok, source()} | :no_available_logs

start_pulling(shard_num, start_after, sources, ingest_fn)

@spec start_pulling(
  shard_num :: non_neg_integer(),
  start_after :: Bedrock.version(),
  sources :: [source()],
  ingest_fn()
) :: Task.t()

stop(puller)

@spec stop(Task.t()) :: :ok

stream_loop(state)

@spec stream_loop(puller_state()) :: no_return()