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

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. Enqueues flush batches for async ObjectStorage persistence
  4. Notifies waiting materializers via WaitingList
  5. Reports durability to Demux only after persistence confirmation

Buffer Management

Transactions are stored as [{version, slice}] with newest at head. Flushing is triggered when:

  • oldest_buffered_version < latest_version - version_gap_threshold

The default version gap threshold is 5,000,000 (~5 seconds in microsecond versions).

Pull API

Materializers call pull/3 to get transactions from a given version. If data is available, it's returned immediately. Otherwise, the materializer is added to a WaitingList and notified when new data arrives.

Summary

Types

shard_id()
slice()
version()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

durable_version(server)

Returns the current durable version (flushed to ObjectStorage).

latest_version(server)

Returns the latest version in the buffer.

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

Pulls transactions starting from the given version.

push(server, version, slice)

Pushes a transaction slice to the ShardServer.

start_link(opts)

Starts a ShardServer for the given shard.

via_tuple(shard_id, registry \\ nil)

Returns the via tuple for a ShardServer.

Types

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 (flushed to ObjectStorage).

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()}]} | {:error, :timeout}

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}]} - Available transactions
  • {:error, :timeout} - No data arrived within timeout

push(server, version, slice)

@spec push(GenServer.server(), version(), slice()) :: :ok

Pushes a transaction slice to the ShardServer.

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 a ShardServer for the given shard.

Options

  • :shard_id - Required. The shard ID this server handles.
  • :demux - Required. PID of the Demux coordinator for durability reporting.
  • :cluster - Required. Cluster name for ObjectStorage paths.
  • :object_storage - Required. ObjectStorage backend.
  • :version_gap - Optional. Version gap threshold for flushing (default: 5_000_000).
  • :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).

via_tuple(shard_id, registry \\ nil)

@spec via_tuple(shard_id(), atom() | nil) :: GenServer.name()

Returns the via tuple for a ShardServer.