Bedrock.DataPlane.Demux.Server (bedrock v0.6.0)
View SourceDemux coordinator that receives transactions from Log, slices by shard, routes to ShardServers, and commands deterministic chunk cuts.
Responsibilities
- Receive transactions pushed from Log via
GenServer.cast - Walk transaction, slice mutations per shard using MutationSlicer
- Send
{:txn, version, slice}to each touched ShardServer - Spin up ShardServer on first touch (lazy)
- Compute deterministic cut versions and command flushes
- Track durability: receive confirmations from ShardServers
- Report min_durable_version to Log for WAL trimming
Deterministic cuts
Versions are microsecond timestamps, so all flush timing is pure version
arithmetic: versions fall into fixed buckets of cut_interval_us
(bucket = div(version, cut_interval_us)). When a pushed version — data or
heartbeat — crosses into a new bucket, the previous bucket closes and every
ShardServer is commanded {:flush, cut_version} with the last version of
the closed bucket. The same versions produce the same cuts on every
replica and every replay, which makes chunks byte-identical and lets
put_if_not_exists/:already_exists count as confirmation.
Known-committed gating
A cut fires only once the known-committed version (KCV) has reached it.
KCV is accumulated monotonically with max; it usually arrives on a push,
but can advance independently while Shale is waiting for a transaction's
predecessor. Such an advance can release a pending cut but never changes
transaction high-water. Nothing not known-committed ever becomes durable,
so chunks can never contain versions a recovery would discard, and recovery
needs no chunk cleanup: the uncommitted tail lives only in ShardServer
buffers, which die with the Demux tree.
Durability Tracking
A ShardServer confirms a cut with its pid, shard id, and cut version. The pid must match this Demux's current shard map, so a confirmation can only advance the replica and child incarnation that produced it. The cut is the shard's floor contribution: everything it has ever seen at or below that version is durable (an empty buffer confirms immediately, so idle shards never pin the floor). We track the minimum across all active shards using gb_sets. With no shards at all, the last completed cut is the floor, so a heartbeat-only log still trims.
Shard Activation
When the first transaction touches a shard, we:
- Start an anonymous ShardServer linked to this process
- Add the shard to durability tracking with the last completed cut as its initial floor contribution — never a merely-buffered version
Failure Semantics
ShardServers are linked to this process. A slicing error, failure to start a required ShardServer, or child crash stops this process before the push can be mistaken for an empty route. That stops the owning Log, and recovery replays from WAL with idempotent ObjectStorage writes.
Summary
Functions
Advances the known-committed watermark independently of transaction delivery.
Returns a specification to start this module under a supervisor.
The default cut-interval in microseconds of version-time.
Gets the ShardServer for a given shard.
Returns the current minimum durable version across all shards, or the last completed cut when no shards are active.
Pushes a committed transaction to the Demux for distribution.
Starts the Demux.Server linked to the calling process.
Types
@type shard_id() :: non_neg_integer()
@type version() :: Bedrock.version()
Functions
@spec advance_known_committed_version(GenServer.server(), version()) :: :ok
Advances the known-committed watermark independently of transaction delivery.
The update is monotonic and does not change transaction high-water. It can release a deterministic cut that was already waiting for commitment.
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec default_cut_interval_us() :: pos_integer()
The default cut-interval in microseconds of version-time.
The WAL rolls its active segment on the same boundaries (see
Shale.Pushing), so that trimming — which can never touch the active
segment — physically drops history at the cut cadence.
@spec get_shard_server(GenServer.server(), shard_id()) :: {:ok, pid()} | {:error, term()}
Gets the ShardServer for a given shard.
Called by materializers to discover their ShardServer for pulling. Creates the ShardServer if it doesn't exist yet.
@spec min_durable_version(GenServer.server()) :: version() | nil
Returns the current minimum durable version across all shards, or the last completed cut when no shards are active.
Returns nil if no cut has completed and no shards are active.
@spec push( GenServer.server(), version(), binary(), known_committed_version :: version() | nil ) :: :ok
Pushes a committed transaction to the Demux for distribution.
Called by Log after a transaction is committed. This is async (cast).
Parameters
server- Demux.Server pid or nameversion- Commit version (8-byte binary)transaction- Encoded transaction binary with SHARD_INDEX
@spec start_link(keyword()) :: GenServer.on_start()
Starts the Demux.Server linked to the calling process.
Options
:cluster- Required. Cluster name.:object_storage- Required. ObjectStorage backend.:log- Required. PID of the owning Log.:cut_interval_us- Optional. Version-time bucket width for deterministic cuts (default: 5_000_000, ~5s).