Bedrock.DataPlane.Materializer.Olivine.Streaming (bedrock v0.7.0)
View SourceThe 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
@type ingest_fn() :: ([Bedrock.DataPlane.Transaction.encoded()], kcv :: Bedrock.version() | nil -> :ok)
@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() }
@type source() :: {Bedrock.DataPlane.Log.id(), Bedrock.DataPlane.Log.ref()}
Functions
@spec call_timeout() :: pos_integer()
@spec circuit_breaker_timeout() :: pos_integer()
@spec ensure_shard_server(puller_state()) :: {:ok, puller_state()} | {:wait, puller_state()}
@spec heartbeat_at(Bedrock.version()) :: Bedrock.DataPlane.Transaction.encoded()
@spec mark_log_as_failed(puller_state(), Bedrock.DataPlane.Log.id()) :: puller_state()
@spec pull_limit() :: pos_integer()
@spec pull_once(puller_state()) :: puller_state()
@spec reset_failed_logs(puller_state()) :: puller_state()
@spec retry_delay() :: pos_integer()
@spec select_log(puller_state()) :: {:ok, source()} | :no_available_logs
@spec start_pulling( shard_num :: non_neg_integer(), start_after :: Bedrock.version(), sources :: [source()], ingest_fn() ) :: Task.t()
@spec stop(Task.t()) :: :ok
@spec stream_loop(puller_state()) :: no_return()