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

View Source

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

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. Track durability: receive reports from ShardServers
  6. Report min_durable_version to Log for WAL trimming

Durability Tracking

When a ShardServer flushes data to ObjectStorage, it sends {:durable, shard_id, max_version} to this process. We track the minimum durable version across all active shards using gb_sets.

Shard Activation

When the first transaction touches a shard, we:

  1. Start a ShardServer linked to this process
  2. Add the shard to durability tracking with initial version = last_seen_version

Failure Semantics

ShardServers are linked to this process. If any ShardServer crashes, this process crashes, which should crash the owning Log. Recovery replays from WAL with idempotent ObjectStorage writes.

Summary

Types

shard_id()
version()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

get_shard_server(server, shard_id)

Gets the ShardServer for a given shard.

min_durable_version(server)

Returns the current minimum durable version across all shards.

push(server, version, transaction)

Pushes a committed transaction to the Demux for distribution.

start_link(opts)

Starts the Demux.Server linked to the calling process.

Types

shard_id()

@type shard_id() :: non_neg_integer()

version()

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

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

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.

Returns nil if no shards are active.

push(server, version, transaction)

@spec push(GenServer.server(), version(), binary()) :: :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.