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

View Source

Tracks durable versions across shards using gb_sets for O(log N) min extraction.

The Demux needs to track the minimum durable version across all active ShardServers to know when it's safe for the Log to trim WAL segments. This module provides efficient tracking with:

  • O(log N) min extraction via :gb_sets
  • O(log N) version updates
  • Shard activation tracking

Implementation

We maintain both a map of shard_id => durable_version for lookups and a :gb_sets ordered set of {version, shard_id} tuples for efficient min extraction.

When a shard's durable version advances, we:

  1. Remove the old {old_version, shard_id} from gb_sets
  2. Insert the new {new_version, shard_id} into gb_sets
  3. Update the map

Usage

durability = Durability.new()

# Activate a shard with initial version
durability = Durability.activate_shard(durability, 0, 1000)

# Update when shard reports durability
durability = Durability.update_shard(durability, 0, 2000)

# Get minimum durable version
Durability.min_durable_version(durability)
#=> 2000

Summary

Types

shard_id()
t()
version()

Functions

activate_shard(durability, shard_id, initial_version)

Activates a shard with an initial durable version.

active?(durability, shard_id)

Returns true if the given shard is being tracked.

active_shard_count(durability)

Returns the number of active shards being tracked.

active_shards(durability)

Returns all active shard IDs.

deactivate_shard(durability, shard_id)

Deactivates a shard, removing it from durability tracking.

min_durable_version(durability)

Returns the minimum durable version across all tracked shards.

new()

Creates a new empty durability tracker.

shard_version(durability, shard_id)

Returns the durable version for a specific shard.

update_shard(durability, shard_id, new_version)

Updates a shard's durable version.

Types

shard_id()

@type shard_id() :: non_neg_integer()

t()

@type t() :: %Bedrock.DataPlane.Demux.Durability{
  shard_versions: %{required(shard_id()) => version()},
  version_index: :gb_sets.set({version(), shard_id()})
}

version()

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

Functions

activate_shard(durability, shard_id, initial_version)

@spec activate_shard(t(), shard_id(), version()) ::
  {:ok, t()} | {:error, :already_active}

Activates a shard with an initial durable version.

Called when the first transaction touches a shard. The shard starts with initial_version as its durable version (typically the last_seen_version at the time of activation).

Returns {:ok, updated_durability} or {:error, :already_active} if shard is already being tracked.

active?(durability, shard_id)

@spec active?(t(), shard_id()) :: boolean()

Returns true if the given shard is being tracked.

active_shard_count(durability)

@spec active_shard_count(t()) :: non_neg_integer()

Returns the number of active shards being tracked.

active_shards(durability)

@spec active_shards(t()) :: [shard_id()]

Returns all active shard IDs.

deactivate_shard(durability, shard_id)

@spec deactivate_shard(t(), shard_id()) :: t()

Deactivates a shard, removing it from durability tracking.

This is typically used during cleanup or when a shard is no longer needed. Returns the updated durability tracker.

min_durable_version(durability)

@spec min_durable_version(t()) :: version() | nil

Returns the minimum durable version across all tracked shards.

Returns nil if no shards are being tracked.

new()

@spec new() :: t()

Creates a new empty durability tracker.

shard_version(durability, shard_id)

@spec shard_version(t(), shard_id()) :: version() | nil

Returns the durable version for a specific shard.

Returns nil if the shard is not being tracked.

update_shard(durability, shard_id, new_version)

@spec update_shard(t(), shard_id(), version()) ::
  {:ok, t()} | {:error, :not_active | :version_going_backwards}

Updates a shard's durable version.

Called when a ShardServer reports that it has durably written data up to a certain version. The version must be >= the current durable version for that shard.

Returns {:ok, updated_durability} or an error if the shard isn't tracked or the version is going backwards.