Bedrock.DataPlane.Demux.ShardServer (bedrock v0.6.0)
View SourcePer-shard GenServer that buffers transactions and writes to ObjectStorage.
Each ShardServer handles a single shard's transaction stream:
- Receives transaction slices from Demux
- Buffers in memory with newest at head
- Obeys
{:flush, cut_version}commands from Demux, persisting everything at or below the cut as one chunk via the async persistence queue - Notifies waiting materializers via WaitingList
- Reports durability to Demux only after persistence confirmation
Commanded cuts
ShardServers never decide when to flush. Demux computes deterministic cut
versions (pure version arithmetic — versions are microsecond timestamps)
and commands flush/2. Because every replica of a shard sees the same
slices and the same cuts, the resulting chunks are byte-identical, which is
why put_if_not_exists returning {:error, :already_exists} counts as a
successful confirmation.
On a cut the server reports its floor contribution — the cut version itself, meaning "everything I have ever seen at or below this version is durable" — not the max flushed data version. An empty buffer confirms the cut immediately, so idle shards never pin the WAL trim floor.
Chunks are named for the last commit they contain
(c/{shard}/{inverted(max_contained)}), preserving the cheap
next-key-after seek contract for readers.
Pull API
Materializers call pull/3 to get transactions from a given version. Where
the reply comes from is decided by one comparison against the durable
version (the last confirmed cut): at or below it, the chunk range in object
storage; above it, the buffer. The two regions always meet — buffered
entries are only evicted once their chunk write confirms — so the stream is
continuous from any starting position.
Every reply carries currency — %{high_water: v, kcv: k} — so an empty
reply still means something: "nothing for you, but you are current through
v" (FoundationDB's empty tag peek shape). Busy shards learn currency from
slice pushes. Idle shards learn it by subscription, never by timer:
parking a puller sends the Demux a one-shot currency request carrying what
this shard has already seen; the Demux replies immediately if it knows
more, and otherwise holds the interest and answers on its next push. The
whole chain is event-driven — a reader is never waiting on a clock, only
on messages that are already in flight.
Ownership
A ShardServer is anonymous and private to the process that starts it. In
production that caller is one log replica's Demux, whose shard map is the
only registry. start_link/1 records the actual caller as the owner, so
durability and currency messages cannot be redirected through an option.
Summary
Functions
Returns a specification to start this module under a supervisor.
Returns the current durable version (the highest confirmed cut).
Commands the ShardServer to make everything at or below cut_version
durable and confirm.
Returns the latest version in the buffer.
Pulls transactions starting from the given version.
Pushes a transaction slice to the ShardServer, along with the known-committed version piggybacked from the commit proxies.
Starts an anonymous ShardServer for the given shard, owned by the caller.
Types
@type shard_id() :: non_neg_integer()
@type slice() :: binary()
@type version() :: Bedrock.version()
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec durable_version(GenServer.server()) :: version() | nil
Returns the current durable version (the highest confirmed cut).
@spec flush(GenServer.server(), version()) :: :ok
Commands the ShardServer to make everything at or below cut_version
durable and confirm.
Called by Demux on deterministic bucket boundaries. This is a cast; the confirmation arrives at the Demux tagged with this server's pid once persistence is confirmed (immediately, when nothing is buffered at or below the cut).
@spec latest_version(GenServer.server()) :: version() | nil
Returns the latest version in the buffer.
@spec pull(GenServer.server(), version(), keyword()) :: {:ok, [{version(), slice()}], currency()} | {:error, :timeout | term()}
Pulls transactions starting from the given version.
Returns transactions from the buffer and/or ObjectStorage.
If no data is available at the requested version, waits up to timeout
for new data to arrive.
Options
:timeout- How long to wait for data (default: 30_000ms):limit- Maximum transactions to return (default: 100)
Returns
{:ok, [{version, slice}], currency}- Available transactions (possibly[]when the shard is known current pastfrom_versionwith nothing to deliver), withcurrency = %{high_water: v, kcv: k}{:error, :timeout}- Nothing learned within timeout
@spec push(GenServer.server(), version(), slice(), kcv :: version() | nil) :: :ok
Pushes a transaction slice to the ShardServer, along with the known-committed version piggybacked from the commit proxies.
Called by Demux when a transaction touches this shard. This is a cast (async, non-blocking).
@spec start_link(keyword()) :: GenServer.on_start()
Starts an anonymous ShardServer for the given shard, owned by the caller.
Options
:shard_id- Required. The shard ID this server handles.:cluster- Required. Cluster name for ObjectStorage paths.:object_storage- Required. ObjectStorage backend.:persistence_queue_capacity- Optional. Max queued flush batches (default: 1024).:persistence_max_retries- Optional. Retry limit for flush failures (default: 5).:persistence_retry_backoff_ms- Optional. Base retry backoff for flush retries (default: 25).:persistence_retry_tick_ms- Optional. Retry polling tick for flush retries (default: 25).