Bedrock.DataPlane.Materializer (bedrock v0.7.0)

Materializer service for the data plane.

Materializer nodes persist key-value data and handle range-based queries.

Summary

Types

fact_name()

id()

key_range()

A shard's replica set for stream pulling: the logs that receive this shard's slices, as {log_id, log_ref} pairs. Resolved once by the director at unlock (ShardRouter.log_ids_for_tag/3); the materializer never re-derives placement and never sees a cluster services map.

recovery_info()

ref()

Functions

get(storage, key_or_selector, version, opts \\ [])

Returns the value for the given key/version, or resolved key-value for KeySelector/version.

get_range(storage, start_key_or_selector, end_key_or_selector, version, opts \\ [])

Returns key-value pairs for keys in the given range at the specified version.

info(storage, fact_names, opts \\ [])

Ask the materializer for various facts about itself.

lock_for_recovery(storage, epoch)

Request that the materializer service lock itself and stop pulling new transactions from the logs. This mechanism is used by a newly elected cluster director to prevent new transactions from being accepted while it is establishing its authority.

lock_for_recovery(storage, epoch, opts)

As lock_for_recovery/2, with a bounded call timeout.

recovery_info()

unlock_after_recovery(storage, durable_version, pull_sources, opts \\ [])

Unlocks the materializer after recovery is complete. This allows the materializer to start accepting new transactions again and continue normal operation.

Types

fact_name()

@type fact_name() ::
  Bedrock.Service.Worker.fact_name()
  | :key_ranges
  | :current_version
  | :durable_version
  | :shard_id
  | :n_objects
  | :path
  | :size_in_bytes
  | :utilization

id()

@type id() :: Bedrock.Service.Worker.id()

key_range()

@type key_range() :: Bedrock.key_range()

pull_sources()

@type pull_sources() :: [{Bedrock.DataPlane.Log.id(), Bedrock.DataPlane.Log.ref()}]

recovery_info()

@type recovery_info() :: %{
  kind: :materializer,
  durable_version: Bedrock.version(),
  oldest_durable_version: Bedrock.version(),
  shard_id: non_neg_integer() | nil
}

ref()

@type ref() :: Bedrock.Service.Worker.ref()

Functions

get(storage, key_or_selector, version, opts \\ [])

@spec get(
  storage_ref :: ref(),
  key :: Bedrock.key(),
  version :: Bedrock.version(),
  opts :: [{:timeout, timeout()}]
) ::
  {:ok, value :: Bedrock.value() | nil}
  | {:error, :version_too_old | :version_too_new}
  | {:failure, :timeout | :unavailable, ref()}

@spec get(
  storage_ref :: ref(),
  key_selector :: Bedrock.KeySelector.t(),
  version :: Bedrock.version(),
  opts :: [{:timeout, timeout()}]
) ::
  {:ok, {resolved_key :: Bedrock.key(), value :: Bedrock.value()} | nil}
  | {:error, :version_too_old | :version_too_new}
  | {:failure, :timeout | :unavailable, ref()}

Returns the value for the given key/version, or resolved key-value for KeySelector/version.

get_range(storage, start_key_or_selector, end_key_or_selector, version, opts \\ [])

@spec get_range(
  storage_ref :: ref(),
  start_key :: Bedrock.key(),
  end_key :: Bedrock.key(),
  version :: Bedrock.version(),
  opts :: [limit: pos_integer(), timeout: timeout()]
) ::
  {:ok, {[{key :: Bedrock.key(), value :: Bedrock.value()}], more :: boolean()}}
  | {:error, :version_too_old | :version_too_new | :unsupported}
  | {:failure, :timeout | :unavailable, ref()}

@spec get_range(
  storage_ref :: ref(),
  start_selector :: Bedrock.KeySelector.t(),
  end_selector :: Bedrock.KeySelector.t(),
  version :: Bedrock.version(),
  opts :: [limit: pos_integer(), timeout: timeout()]
) ::
  {:ok, {[{key :: Bedrock.key(), value :: Bedrock.value()}], more :: boolean()}}
  | {:error,
     :version_too_old
     | :version_too_new
     | :unsupported
     | :not_found
     | :invalid_range}
  | {:failure, :timeout | :unavailable, ref()}

Returns key-value pairs for keys in the given range at the specified version.

Range is [start_key, end_key) - includes start_key, excludes end_key. Supports both binary keys and KeySelectors for range boundaries. Supported by the Olivine materializer engine.

info(storage, fact_names, opts \\ [])

@spec info(
  storage :: ref(),
  [fact_name()],
  opts :: [{:timeout_in_ms, Bedrock.timeout_in_ms()}]
) ::
  {:ok,
   %{
     required(fact_name()) =>
       Bedrock.value()
       | Bedrock.version()
       | [key_range()]
       | non_neg_integer()
       | Path.t()
   }}
  | {:error, :unavailable}

Ask the materializer for various facts about itself.

lock_for_recovery(storage, epoch)

@spec lock_for_recovery(storage_ref :: ref(), recovery_epoch :: Bedrock.epoch()) ::
  {:ok, storage_pid :: pid(),
   recovery_info :: [
     kind: :materializer,
     durable_version: Bedrock.version(),
     oldest_durable_version: Bedrock.version()
   ]}
  | {:error, :newer_epoch_exists}

In order for the lock to succeed, the given epoch needs to be greater than the current epoch.

lock_for_recovery(storage, epoch, opts)

@spec lock_for_recovery(
  ref(),
  Bedrock.epoch(),
  opts :: [{:timeout_in_ms, Bedrock.timeout_in_ms()}]
) ::
  {:ok, pid(), recovery_info :: term()} | {:error, :newer_epoch_exists}

As lock_for_recovery/2, with a bounded call timeout.

recovery_info()

@spec recovery_info() :: [fact_name()]

unlock_after_recovery(storage, durable_version, pull_sources, opts \\ [])

@spec unlock_after_recovery(
  storage :: ref(),
  durable_version :: Bedrock.version(),
  pull_sources(),
  opts :: [{:timeout_in_ms, Bedrock.timeout_in_ms()}]
) :: :ok | {:error, :unavailable} | {:failure, :timeout, ref()}

Unlocks the materializer after recovery is complete. This allows the materializer to start accepting new transactions again and continue normal operation.

The durable version and the shard's pull sources — its replica set of {log_id, log_ref} pairs, resolved once by the director through the same ShardRouter walk the commit proxies route with — must be provided so the materializer resumes at the correct state. The seed is exactly this shard's sources, never a cluster map.