@spec all_for(String.t(), DateTime.t()) :: %{required(atom()) => integer()}

Returns a map of feature => value for a tenant's warm period counters in a period.

@spec apply_remote(key(), integer()) :: :ok | :cold

Applies a delta received from another node to this node's view. Only the value moves: the delta is not ours to flush or re-gossip. The key is marked touched so local LiveViews see the change. Cold keys are skipped (they seed from the database on first read, which already contains every flushed delta).

@spec base(key()) :: integer() | nil

This node's base for a key: what it believes the database holds (value - pending_flush).

@spec clear_dirty(key()) :: :ok

Removes a single key from the dirty set.

@spec clear_touched(key()) :: :ok

Removes a single key from the touched set.

@spec day_value(String.t(), atom(), Date.t()) :: integer()

Returns the current value of a UTC day bucket (rehydrating from the database if cold).

@spec dirty_keys() :: [key()]

Snapshots the current set of dirty keys.

@spec history_key?(key()) :: boolean()

Whether a key is a history (day bucket) key.

@spec incr(String.t(), atom(), integer(), DateTime.t()) :: integer()

Increments a period counter by qty and returns the new value.

@spec mark_dirty(key()) :: :ok

Marks a counter key dirty (pending flush) and touched (pending broadcast).

@spec rebase(key(), integer(), :flush | :gossip) :: :ok | :cold

Re-bases this node's view on an authoritative database total: value becomes total + pending_flush. Applied as a delta against a snapshot, so a bump that lands mid-rebase is kept exactly. Cold keys are skipped.

source says where the total came from, and only :flush — this node's own write, which read the row back — clears remote. A total announced by another node is a database total that node saw, and this node may have applied gossiped deltas since; clearing remote for one of those told remote_since_rebase/1 the view had not moved when it had, which is the one question the flusher asks before deciding whether a failed write landed.

@spec release(String.t(), atom(), integer(), DateTime.t(), Date.t() | nil) :: :ok

Releases a previously reserved qty (rollback on a failed function).

on is the day the reservation was counted against. Without it the day history is decremented from the clock — so work that started at 23:59:59 and gave up a second later took its release out of the next day, leaving one day permanently over-counted and the other under. The period counter was taught this; the day bucket beside it was not.

@spec remote_since_rebase(key()) :: non_neg_integer()

How much of this key's value came from other nodes since the last rebase.

Zero means base/1 really is what this node believes the database holds; anything else means it is that plus deltas whose own nodes may not have written them yet.

@spec reserve(
  String.t(),
  atom(),
  integer(),
  DateTime.t(),
  non_neg_integer() | nil,
  boolean()
) ::
  :ok | {:error, :limit_exceeded}

Atomically reserves qty against an optional hard limit.

Increments first; if the new value exceeds limit it rolls the increment back and returns {:error, :limit_exceeded}. A nil limit always succeeds.

@spec restore_pending(key(), integer()) :: :ok

Puts a taken flush delta back (the database write failed) and re-marks the key dirty so the next flush retries it.

@spec take_pending(key(), pending()) :: integer()

Atomically takes and zeroes a pending column, returning the delta accumulated since the last take. Concurrent bumps between the read and the zeroing are preserved (the column is decremented by the amount read, not set to zero). Cold keys yield 0.

@spec touched_keys() :: [key()]

Snapshots the current set of touched keys.

@spec value(String.t(), atom(), DateTime.t()) :: integer()

Returns the current value for a period counter (rehydrating from the database if cold).

@spec warm_day_values(String.t(), atom()) :: %{required(Date.t()) => integer()}

Returns date => value for a feature's warm day buckets (no database access).