Bedrock.ObjectStorage behaviour (bedrock v0.5.2)

View Source

Behaviour and API for object storage backends.

ObjectStorage provides a simple key-value interface for storing and retrieving binary data objects. It supports multiple backends (local filesystem, S3, GCS) through a common behaviour.

Operations

Canonical Error Reasons

Backends may return storage-specific errors. The public API normalizes known errors to canonical reasons:

  • :not_found
  • :already_exists
  • :version_mismatch
  • :access_denied

Path Structure

Object keys follow a hierarchical structure:

  • Cluster state: /{cluster}/state
  • Chunks: /{cluster}/shards/{tag}/chunks/{inverted_version}
  • Snapshots: /{cluster}/shards/{tag}/snapshots/{inverted_version}

Inverted Version Keys

Object stores list in ascending order only. To get newest objects first, use inverted version numbers: (2^64 - 1) - version

See Bedrock.ObjectStorage.Keys for key formatting helpers.

Summary

Types

backend()
content_type()
data()
error()
error_reason()
key()
opts()
version_token()

Callbacks

delete(backend, key)

Delete an object by key.

get(backend, key)

Retrieve an object by key.

get_with_version(backend, key)

Retrieve an object with its version token for conditional updates.

list(backend, prefix, opts)

List objects with the given prefix.

put(backend, key, data, opts)

Store an object at the given key.

put_if_not_exists(backend, key, data, opts)

Store an object only if it doesn't already exist (conditional write).

put_if_version_matches(backend, key, version_token, data, opts)

Store an object only if its version matches the expected token.

Functions

backend(module, config \\ [])

Creates an opaque backend reference for the given module and config.

delete(arg, key)

Delete an object by key.

get(arg, key)

Retrieve an object by key.

get_with_version(arg, key)

Retrieve an object with its version token.

list(arg, prefix, opts \\ [])

List objects with the given prefix.

normalize_error(arg)

Normalizes backend-specific error reasons to canonical ObjectStorage reasons.

put(arg, key, data, opts \\ [])

Store an object at the given key.

put_if_not_exists(arg, key, data, opts \\ [])

Store an object only if it doesn't already exist.

put_if_version_matches(arg, key, version_token, data, opts \\ [])

Store an object only if its version matches.

Types

backend()

@type backend() :: {module(), keyword()}

content_type()

@type content_type() :: String.t()

data()

@type data() :: iodata()

error()

@type error() :: {:error, error_reason()}

error_reason()

@type error_reason() ::
  :not_found | :already_exists | :version_mismatch | :access_denied | term()

key()

@type key() :: String.t()

opts()

@type opts() :: keyword()

version_token()

@type version_token() :: String.t()

Callbacks

delete(backend, key)

@callback delete(backend :: term(), key :: key()) :: :ok | error()

Delete an object by key.

Deletion is idempotent - deleting a non-existent object succeeds.

Returns

  • :ok - Object deleted (or didn't exist)
  • {:error, reason} - Deletion failed

get(backend, key)

@callback get(backend :: term(), key :: key()) :: {:ok, data()} | error()

Retrieve an object by key.

Returns

  • {:ok, data} - Object data
  • {:error, :not_found} - Object does not exist
  • {:error, reason} - Retrieval failed

get_with_version(backend, key)

@callback get_with_version(backend :: term(), key :: key()) ::
  {:ok, data(), version_token()} | error()

Retrieve an object with its version token for conditional updates.

Returns the object data along with an opaque version token that can be passed to put_if_version_matches/5 to implement optimistic concurrency.

Returns

  • {:ok, data, version_token} - Object data and version token
  • {:error, :not_found} - Object does not exist
  • {:error, reason} - Retrieval failed

list(backend, prefix, opts)

@callback list(backend :: term(), prefix :: String.t(), opts :: opts()) :: Enumerable.t()

List objects with the given prefix.

Returns a lazy stream that fetches pages as needed. Objects are returned in ascending lexicographic order by key.

Options

  • :limit - Maximum number of keys to return (default: unlimited)

Returns

A Stream of key strings.

put(backend, key, data, opts)

@callback put(backend :: term(), key :: key(), data :: data(), opts :: opts()) ::
  :ok | error()

Store an object at the given key.

Options

  • :content_type - MIME type of the data (default: "application/octet-stream")

Returns

  • :ok - Object stored successfully
  • {:error, reason} - Storage failed

put_if_not_exists(backend, key, data, opts)

@callback put_if_not_exists(
  backend :: term(),
  key :: key(),
  data :: data(),
  opts :: opts()
) ::
  :ok | error()

Store an object only if it doesn't already exist (conditional write).

This provides atomicity for write operations where concurrent writers might attempt to create the same object. Only one writer will succeed.

Options

Same as put/4.

Returns

  • :ok - Object stored successfully (was new)
  • {:error, :already_exists} - Object already exists
  • {:error, reason} - Storage failed

put_if_version_matches(backend, key, version_token, data, opts)

@callback put_if_version_matches(
  backend :: term(),
  key :: key(),
  version_token :: version_token(),
  data :: data(),
  opts :: opts()
) :: :ok | error()

Store an object only if its version matches the expected token.

This implements optimistic locking (compare-and-swap) semantics. The operation succeeds only if the current version of the object matches the provided version_token (obtained from get_with_version/2).

Returns

  • :ok - Object updated successfully
  • {:error, :version_mismatch} - Object was modified since version_token was obtained
  • {:error, :not_found} - Object does not exist
  • {:error, reason} - Update failed

Functions

backend(module, config \\ [])

@spec backend(module :: module(), config :: keyword()) :: {module(), keyword()}

Creates an opaque backend reference for the given module and config.

Examples

backend = ObjectStorage.backend(ObjectStorage.LocalFilesystem, root: "/tmp/objects")
ObjectStorage.put(backend, "test/key", "data")

delete(arg, key)

@spec delete(backend :: {module(), keyword()}, key :: key()) :: :ok | error()

Delete an object by key.

get(arg, key)

@spec get(backend :: {module(), keyword()}, key :: key()) :: {:ok, data()} | error()

Retrieve an object by key.

get_with_version(arg, key)

@spec get_with_version(backend :: {module(), keyword()}, key :: key()) ::
  {:ok, data(), version_token()} | error()

Retrieve an object with its version token.

list(arg, prefix, opts \\ [])

@spec list(backend :: {module(), keyword()}, prefix :: String.t(), opts :: opts()) ::
  Enumerable.t()

List objects with the given prefix.

normalize_error(arg)

@spec normalize_error({:error, term()}) :: error()

Normalizes backend-specific error reasons to canonical ObjectStorage reasons.

Canonical reasons:

  • :not_found
  • :already_exists
  • :version_mismatch
  • :access_denied

put(arg, key, data, opts \\ [])

@spec put(
  backend :: {module(), keyword()},
  key :: key(),
  data :: data(),
  opts :: opts()
) ::
  :ok | error()

Store an object at the given key.

put_if_not_exists(arg, key, data, opts \\ [])

@spec put_if_not_exists(
  backend :: {module(), keyword()},
  key :: key(),
  data :: data(),
  opts :: opts()
) :: :ok | error()

Store an object only if it doesn't already exist.

put_if_version_matches(arg, key, version_token, data, opts \\ [])

@spec put_if_version_matches(
  backend :: {module(), keyword()},
  key :: key(),
  version_token :: version_token(),
  data :: data(),
  opts :: opts()
) :: :ok | error()

Store an object only if its version matches.