Ankusa.BlobStore behaviour (ankusa v0.7.0)

Copy Markdown View Source

Object storage: PUT whole immutable objects, range GET one record, DELETE for retention, LIST by prefix. The default is the local filesystem; S3, GCS, Azure Blob and OCI are drop-in adapters.

This module is both the behaviour and the instance-scoped facade. Every call names a scope:

  • :segments — compacted hook segments (Ankusa.Storage): the storage.blob_store, with every new key under storage.key_prefix. Nodes that share one bucket give themselves distinct prefixes ("node-a/"), because a segment key (seg/<first>-<last>) is only unique per node.
  • :backup — store backups (Ankusa.Store.Backup): the backup.blob_store when set, else the segment store, with storage.key_prefix. A backup belongs to one node's store, like its segments.

Claim packs do not go through this facade: Ankusa.ClaimCheck.Stores writes and reads them across claim_check.blob_stores, never prefixed.

The prefix is applied once, when a new object is named (object_key/3); the full key is what the archive catalogue records, and put/get/ get_range/delete take full keys verbatim. Changing storage.key_prefix therefore moves only new segments, never strands written ones.

Encryption at rest

The object-store adapters take an :sse option (Ankusa.BlobStore.SSE) that asks the provider to encrypt every object a put/4 writes, with a KMS key or the provider's own. It is per store, so it covers the segment store, the backup store and every claim store as each is configured.

Transport

The HTTP adapters share their transport options (Ankusa.BlobStore.Options): one deadline per operation, jittered retries that honour Retry-After (Ankusa.BlobStore.Retry), a breaker per bucket (Ankusa.BlobStore.Breaker), multipart uploads above a size, integrity checksums (Ankusa.BlobStore.Checksum) and a connection pool per endpoint (Ankusa.BlobStore.Pool).

Summary

Types

scope()

Callbacks

delete(instance, key, opts)
get(instance, key, opts)
get_range(instance, key, offset, length, opts)
list(instance, prefix, opts)

Every key under prefix, sorted, across all of the store's result pages. A page that fails fails the call: a partial listing is never presented as the whole.

put(instance, key, data, opts)
put_if_absent(instance, key, data, opts)

put/4, but only if no object exists under key: {:error, :already_exists} when one does, with that object left exactly as it was. The adapters use the store's own precondition (S3 and Azure If-None-Match: *, GCS ifGenerationMatch=0, OCI if-none-match: *), so the check and the write are one atomic request. Ankusa.ClaimCheck writes every pack with it: a reused or colliding pack id then refuses instead of overwriting claims that are already referenced.

Functions

delete(instance, scope, key)
get(instance, scope, key)
get_range(instance, scope, key, offset, length)
list(instance, scope, prefix)

Every key of this node's objects in scope under prefix (behind storage.key_prefix for :segments and :backup), as full keys.

object_key(instance, scope, key)

The full key a new object in scope is written under: key behind storage.key_prefix. The compactor names each segment with it and records the result in the archive catalogue, so a later prefix change only moves new segments; the ones already written are read back under the key they were written with.

put(instance, scope, key, data)
put_if_absent(instance, scope, key, data)
same_object(mod, instance, key, data, opts)

Settle an :already_exists: :ok when key already holds exactly data (a conditional PUT that landed, whose answer was lost and which was then sent again, finds its own object), else {:error, :already_exists}. A store that cannot be read back is {:error, :already_exists} too: the caller mints a new key rather than assume.

secondary_config(config)

The second object store backups are also written to (backup.secondary), under the same storage.key_prefix; nil when none is configured.

store(instance, scope)

The {module, opts} a scope resolves to on instance.

validate_config!(config)

Check the :sse option and the transport options (Ankusa.BlobStore.Options) of every blob store the config names (storage.blob_store, backup.blob_store, backup.secondary, claim_check.blob_stores).

Types

scope()

@type scope() :: :segments | :backup

Callbacks

delete(instance, key, opts)

@callback delete(instance :: atom(), key :: String.t(), opts :: keyword()) :: :ok

get(instance, key, opts)

@callback get(instance :: atom(), key :: String.t(), opts :: keyword()) ::
  {:ok, binary()} | {:error, term()}

get_range(instance, key, offset, length, opts)

@callback get_range(
  instance :: atom(),
  key :: String.t(),
  offset :: non_neg_integer(),
  length :: pos_integer(),
  opts :: keyword()
) :: {:ok, binary()} | {:error, term()}

list(instance, prefix, opts)

@callback list(instance :: atom(), prefix :: String.t(), opts :: keyword()) ::
  {:ok, [String.t()]} | {:error, term()}

Every key under prefix, sorted, across all of the store's result pages. A page that fails fails the call: a partial listing is never presented as the whole.

put(instance, key, data, opts)

@callback put(instance :: atom(), key :: String.t(), data :: iodata(), opts :: keyword()) ::
  :ok | {:error, term()}

put_if_absent(instance, key, data, opts)

@callback put_if_absent(
  instance :: atom(),
  key :: String.t(),
  data :: iodata(),
  opts :: keyword()
) ::
  :ok | {:error, :already_exists} | {:error, term()}

put/4, but only if no object exists under key: {:error, :already_exists} when one does, with that object left exactly as it was. The adapters use the store's own precondition (S3 and Azure If-None-Match: *, GCS ifGenerationMatch=0, OCI if-none-match: *), so the check and the write are one atomic request. Ankusa.ClaimCheck writes every pack with it: a reused or colliding pack id then refuses instead of overwriting claims that are already referenced.

An object that exists with exactly the bytes being written is :ok (same_object/5): a retried request whose first attempt landed is not a collision with another writer.

Functions

delete(instance, scope, key)

@spec delete(atom(), scope(), String.t()) :: :ok

get(instance, scope, key)

@spec get(atom(), scope(), String.t()) :: {:ok, binary()} | {:error, term()}

get_range(instance, scope, key, offset, length)

@spec get_range(atom(), scope(), String.t(), non_neg_integer(), pos_integer()) ::
  {:ok, binary()} | {:error, term()}

list(instance, scope, prefix)

@spec list(atom(), scope(), String.t()) :: {:ok, [String.t()]} | {:error, term()}

Every key of this node's objects in scope under prefix (behind storage.key_prefix for :segments and :backup), as full keys.

object_key(instance, scope, key)

@spec object_key(atom(), scope(), String.t()) :: String.t()

The full key a new object in scope is written under: key behind storage.key_prefix. The compactor names each segment with it and records the result in the archive catalogue, so a later prefix change only moves new segments; the ones already written are read back under the key they were written with.

put(instance, scope, key, data)

@spec put(atom(), scope(), String.t(), iodata()) :: :ok | {:error, term()}

put_if_absent(instance, scope, key, data)

@spec put_if_absent(atom(), scope(), String.t(), iodata()) ::
  :ok | {:error, :already_exists} | {:error, term()}

same_object(mod, instance, key, data, opts)

@spec same_object(module(), atom(), String.t(), iodata(), keyword()) ::
  :ok | {:error, :already_exists}

Settle an :already_exists: :ok when key already holds exactly data (a conditional PUT that landed, whose answer was lost and which was then sent again, finds its own object), else {:error, :already_exists}. A store that cannot be read back is {:error, :already_exists} too: the caller mints a new key rather than assume.

secondary_config(config)

@spec secondary_config(Ankusa.Config.t()) :: {module(), keyword(), String.t()} | nil

The second object store backups are also written to (backup.secondary), under the same storage.key_prefix; nil when none is configured.

store(instance, scope)

@spec store(atom(), scope()) :: {module(), keyword()}

The {module, opts} a scope resolves to on instance.

validate_config!(config)

@spec validate_config!(Ankusa.Config.t()) :: :ok

Check the :sse option and the transport options (Ankusa.BlobStore.Options) of every blob store the config names (storage.blob_store, backup.blob_store, backup.secondary, claim_check.blob_stores).