# `Ankusa.Store` [🔗](https://github.com/jamescarr/ankusa/blob/ankusa-v0.7.0/packages/ankusa/lib/ankusa/store.ex#L1) The node's local store: one RocksDB database per instance, under `Config.path(config, "store")`. Holds hooks, one delivery row per hook and sink (the DLQ is the dead rows), the quarantine pen, API-managed sources, rate-limit overrides and the archive catalogue. Column families: * `default` — sequence, the store format version (`m:format_version`), migration markers, sources and their revisions, rate limits, replay jobs, the audit sequence and key, the next tenant ordinal (`m:next_ord`) * `hooks` — `<>` → envelope * `deliveries` — `<>` → delivery row * `index` — due/inflight/dead rows (tenant-ordinal first, the key constructors are in `store/keys.ex`), the by-tenant dead index, active-tenant markers, archive obligations, cleared markers, claim-check refs * `archive` — segment catalogue and imported legacy locations * `quarantine` — quarantined envelopes * `audit` — the audit trail (`<>`) and the rows of mutations still in flight (`<>`), see `Ankusa.Audit` * `tenants` — tenant ordinal ↔ tenant id (`queue/tenants.ex`) * `counters` — per-tenant and node-wide queue counters, written only as `{:merge, :counters, key, delta}` operands (`queue/deliveries/counters.ex`) `Ankusa.Store` is the one process that owns the database handle; every other caller reads or writes through the functions here, which run the RocksDB NIFs (on the dirty schedulers) from the calling process and look the handles up in ETS on every call. Durability is per call: `sync: true` means the write is on disk before the call returns. RocksDB is one WAL per store, so a `sync: false` writer's bytes are durable as soon as any later synced commit lands. Every write is atomic, and the recovery mode tolerates only a torn tail, never earlier damage. `checkpoint/2` is how `Ankusa.Store.Backup` snapshots the store: a consistent copy (hard links where it can) in a directory that must not exist yet. With `backup.enabled`, a store directory that holds no database is restored from the latest backup before it is opened (see `Ankusa.Store.Backup.restore/3`). # `op` ```elixir @type op() :: {:put, atom(), binary(), binary()} | {:delete, atom(), binary()} | {:delete_range, atom(), binary(), binary()} | {:merge, atom(), binary(), integer()} ``` # `checkpoint` ```elixir @spec checkpoint(atom(), Path.t()) :: :ok | {:error, term()} ``` Writes a consistent snapshot of the store to `dir`, which must not exist (RocksDB creates it). Memtables are flushed first, so the snapshot is SST and blob files plus the manifest — hard links when `dir` is on the store's filesystem, copies otherwise. Runs off the store process: readiness keeps answering while it does. One at a time: `{:error, :checkpoint_in_progress}` while another runs; `{:error, :store_unavailable}` when the store is closed or does not answer within 120 s. # `child_spec` Returns a specification to start this module under a supervisor. See `Supervisor`. # `failure_class` ```elixir @spec failure_class(term()) :: :no_space | :io_error | :read_only | :corruption | :busy | :other ``` The class of a RocksDB or file system failure, for the `class` label and the `/ready` `reason`: `:no_space` (ENOSPC), `:io_error` (EIO and the rest of RocksDB's `IO error`s), `:read_only`, `:corruption`, `:busy` (the database lock is held, or a resource is temporarily unavailable) or `:other`. A fixed set, so no label is ever built from an error string. # `first` ```elixir @spec first(atom(), atom(), {binary(), binary()}, keyword()) :: {:ok, {binary(), binary()} | nil} | {:error, term()} ``` The first key and value of `family` in the half-open range `{lower, upper}` (the last, with `reverse: true`), or `nil`: one seek, with `fold/6`'s proof that the scan was not cut short by corruption. # `fold` ```elixir @spec fold( atom(), atom(), {binary(), binary()}, acc, (binary(), binary(), acc -> {:cont, acc} | {:halt, acc}), keyword() ) :: {:ok, acc} | {:error, term()} when acc: term() ``` Folds a scanned key `family` (an internal key family such as `:hooks`, `:due` or `:dead`) over the half-open key range `{lower, upper}`. `fun.(key, value, acc)` returns `{:cont, acc}` or `{:halt, acc}`; `reverse: true` walks newest-first. The RocksDB binding reports corruption met during iteration as a clean end of the scan (`invalid_iterator`), so a scan is only believed when it *proves* it finished: forward it must reach a key at or beyond `upper` (the family's high sentinel always exists to stop on), reverse a key below `lower` or the low sentinel. A scan that goes invalid before that returns `{:error, {:corruption, _}}` instead of a silently shortened result. Stopping early with `{:halt, acc}` is the caller's choice and needs no proof. Sentinels are never passed to `fun`. # `get` ```elixir @spec get(atom(), atom(), binary()) :: {:ok, binary()} | :not_found | {:error, term()} ``` Reads `key` from column family `cf`. Corruption is reported, never hidden. # `multi_get` ```elixir @spec multi_get(atom(), atom(), [binary()]) :: {:ok, [{:ok, binary()} | :not_found]} | {:error, term()} ``` Reads `keys` from column family `cf`, results in key order. Any per-key error turns the whole call into an error. # `prefixes_present` ```elixir @spec prefixes_present(atom(), atom(), [binary()]) :: {:ok, [boolean()]} | {:error, term()} ``` For each key prefix, whether `family` holds any key that starts with it, in the order given. One iterator serves the whole list, so probing a batch costs a seek per prefix instead of an iterator per prefix. It carries `fold/6`'s guarantee: the family's high sentinel always exists, so a seek that finds nothing at all means the iterator went invalid on damage, and the answer is `{:error, {:corruption, _}}`, never a quiet `false`. # `property` ```elixir @spec property(atom(), atom(), String.t()) :: {:ok, non_neg_integer()} | {:error, term()} ``` Reads a RocksDB CF property like `"rocksdb.estimate-num-keys"`, `"rocksdb.total-sst-files-size"` or `"rocksdb.total-blob-file-size"`, parsed to an integer. # `ready` ```elixir @spec ready(atom()) :: :ok | {:error, {:store_unavailable | :write_failed, atom()}} ``` Readiness: whether the store can take a synced write right now. `{:error, {:write_failed, class}}` while any write reported through `report_write_failure/2` failed in the last 5 s: a full disk can refuse an ingest batch while a few bytes still fit. Otherwise writes one meta key with `sync: true` (the same WAL append and fsync an ingest commit needs) and answers `:ok`, or `{:error, {:write_failed, class}}` when the write fails — in which case the store also reopens itself, as for any failed write. `{:error, {:store_unavailable, class}}` when the store is closed (`class` is why its last open failed) or does not answer within 5 s (`:timeout`, or `:not_running` when no store runs). `class` is a `failure_class/1`. The probe's answer is cached for 1 s. # `reopen` ```elixir @spec reopen(atom()) :: :ok | {:error, term()} ``` Closes and reopens the database at the same path, then republishes the handles. RocksDB latches a background error after a failed WAL append (the disk was full) and keeps rejecting writes after space is freed. Reopening replays the WAL, drops a torn tail, and clears the latch. `write/3` asks for it itself through `report_write_failure/2`, at most once every few seconds. A failed reopen leaves the store unavailable, not crashed, and the store then retries by itself, after 5 s and then twice as long each time up to 60 s (each wait jittered down by up to a quarter), until it opens. Every attempt emits `[:ankusa, :store, :reopened]` and `[:ankusa, :store, :unavailable]`. # `report_write_failure` ```elixir @spec report_write_failure(atom(), term()) :: :ok ``` Tell the store a write failed. A failed WAL append (a full disk) can leave RocksDB with a latched background error that outlives the freed space; the store reopens itself to clear it, at most once per `reopen_interval_ms` (5 s), whichever process saw the failure. Emits `[:ankusa, :store, :write_failed]` from the calling process. Never blocks. # `start_link` # `sync_wal` ```elixir @spec sync_wal(atom()) :: :ok | {:error, term()} ``` Makes every write applied so far with `sync: false` durable: one fsync of the WAL (`SyncWAL()`), which does not take RocksDB's write lock, so a writer applying the next batch meanwhile is not held up. `:ok` means every batch whose `write/3` returned before this call began is on disk. A failure is reported like a failed write (`report_write_failure/2`: the store reopens to clear a latched error), and the caller must treat everything it applied since its last successful sync as not durable. # `write` ```elixir @spec write(atom(), [op()], keyword()) :: :ok | {:error, term()} ``` Writes `ops` as one atomic batch. `sync: true` returns only once the batch is durable. `[]` is a no-op success. Keys and values must be binaries: a malformed op raises, it is never reported as an unavailable store. `{:merge, cf, key, delta}` adds `delta` to the integer at `key` without reading it (a RocksDB merge operand, `:erlang_merge_operator`'s `int_add`): concurrent batches from different processes cannot lose each other's counts. Only a column family opened with that operator (`counters`) may be merged into. --- *Consult [api-reference.md](api-reference.md) for complete listing*