Bedrock.DataPlane.Materializer.Olivine.Streaming (bedrock v0.6.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
Shard-to-log placement is deterministic arithmetic shared with the commit
proxies: the index map is the sorted log ids and the golden-ratio walk is
ShardRouter.get_log_indices/3 (see topology_phase.build_routing_data/1
— the two sites must agree). 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
Functions
The replica set for a shard: the log ids that receive this shard's slices, in golden-ratio order over the sorted log-id list.
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(), logs: %{ required(Bedrock.DataPlane.Log.id()) => Bedrock.ControlPlane.Config.LogDescriptor.t() }, services: %{ required(String.t()) => Bedrock.ControlPlane.Config.ServiceDescriptor.t() }, failed_logs: %{required(Bedrock.DataPlane.Log.id()) => integer()}, shard_server: pid() | nil, current_log_id: Bedrock.DataPlane.Log.id() | nil, ingest_fn: ingest_fn() }
Functions
@spec call_timeout() :: pos_integer()
@spec candidate_log_ids(non_neg_integer(), %{ required(Bedrock.DataPlane.Log.id()) => Bedrock.ControlPlane.Config.LogDescriptor.t() }) :: [Bedrock.DataPlane.Log.id()]
The replica set for a shard: the log ids that receive this shard's slices, in golden-ratio order over the sorted log-id list.
Mirrors the commit proxies' routing exactly (the index map built in
topology_phase.build_routing_data/1 is the sorted log ids, and the
replication factor currently spans all logs).
@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, {Bedrock.DataPlane.Log.id(), Bedrock.ControlPlane.Config.ServiceDescriptor.t()}} | :no_available_logs
@spec start_pulling( shard_num :: non_neg_integer(), start_after :: Bedrock.version(), logs :: %{ required(Bedrock.DataPlane.Log.id()) => Bedrock.ControlPlane.Config.LogDescriptor.t() }, services :: %{ required(String.t()) => Bedrock.ControlPlane.Config.ServiceDescriptor.t() }, ingest_fn() ) :: Task.t()
@spec stop(Task.t()) :: :ok
@spec stream_loop(puller_state()) :: no_return()