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

View Source

Per-shard GenServer that buffers transactions and writes to ObjectStorage.

Each ShardServer handles a single shard's transaction stream:

  1. Receives transaction slices from Demux
  2. Buffers in memory with newest at head
  3. Obeys {:flush, cut_version} commands from Demux, persisting everything at or below the cut as one chunk via the async persistence queue
  4. Notifies waiting materializers via WaitingList
  5. 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

currency()

@type currency() :: %{high_water: version() | nil, kcv: version() | nil}

shard_id()

@type shard_id() :: non_neg_integer()

slice()

@type slice() :: binary()

version()

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

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

durable_version(server)

@spec durable_version(GenServer.server()) :: version() | nil

Returns the current durable version (the highest confirmed cut).

flush(server, cut_version)

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

latest_version(server)

@spec latest_version(GenServer.server()) :: version() | nil

Returns the latest version in the buffer.

pull(server, from_version, opts \\ [])

@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 past from_version with nothing to deliver), with currency = %{high_water: v, kcv: k}
  • {:error, :timeout} - Nothing learned within timeout

push(server, version, slice, kcv \\ nil)

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

start_link(opts)

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