Bedrock.DataPlane.Demux.Server (bedrock v0.6.0)

View Source

Demux coordinator that receives transactions from Log, slices by shard, routes to ShardServers, and commands deterministic chunk cuts.

Responsibilities

  1. Receive transactions pushed from Log via GenServer.cast
  2. Walk transaction, slice mutations per shard using MutationSlicer
  3. Send {:txn, version, slice} to each touched ShardServer
  4. Spin up ShardServer on first touch (lazy)
  5. Compute deterministic cut versions and command flushes
  6. Track durability: receive confirmations from ShardServers
  7. 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:

  1. Start an anonymous ShardServer linked to this process
  2. 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

shard_id()

@type shard_id() :: non_neg_integer()

version()

@type version() :: Bedrock.version()

Functions

advance_known_committed_version(server, version)

@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.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

default_cut_interval_us()

@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.

get_shard_server(server, shard_id)

@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.

min_durable_version(server)

@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.

push(server, version, transaction, known_committed_version \\ nil)

@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 name
  • version - Commit version (8-byte binary)
  • transaction - Encoded transaction binary with SHARD_INDEX

start_link(opts)

@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).