AuroraMeter.Counter (Aurora Meter v0.4.0)

View Source

The hot path: atomic ETS counter operations.

Increments use a single :ets.update_counter/3 call (lock-free, concurrency-safe). A cold key is seeded once from the last flushed database value so the in-ETS value is always absolute; after that first touch the path is pure ETS. Every mutation marks the key dirty (for the flusher) and touched (for the broadcaster).

Each row is {key, value, pending_flush, pending_gossip, remote, reserved}:

  • value is this node's view of the cluster-wide total
  • pending_flush is what this node has added since its last database flush
  • remote is how much of value came from other nodes since the last rebase, so anything reasoning about what the database holds can tell that this node's view has moved for a reason the database has not
  • pending_gossip is what this node has added since its last PubSub tick
  • reserved occupies quota for unfinished with_quota work, but is not flushed or gossiped as completed usage

The flusher writes pending_flush as a delta (value = value + Δ), so nodes add up instead of overwriting one another, then re-bases value on the total the database returns. The broadcaster ships pending_gossip to the other nodes so their views converge within one tick. See AuroraMeter.Cluster and the clustering guide.

Two key shapes share the table:

  • {tenant_key, feature, period_start} — the billing-period counter that entitlements and reporting read
  • {tenant_key, feature, {:day, date}} — a UTC day bucket, maintained alongside the period counter when :history is enabled, feeding AuroraMeter.history/3

Summary

Types

A history counter key: {tenant_key, feature, {:day, date}}.

Any counter key.

Which pending column to take: the database flush or the PubSub gossip.

A period counter key: {tenant_key, feature, period_start}.

Functions

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

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).

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

Removes a single key from the dirty set.

Removes a single key from the touched set.

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

Snapshots the current set of dirty keys.

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

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

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

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.

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

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

Atomically reserves qty against an optional hard limit.

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

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.

Snapshots the current set of touched keys.

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

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

Types

day_key()

@type day_key() :: {String.t(), atom(), {:day, Date.t()}}

A history counter key: {tenant_key, feature, {:day, date}}.

key()

@type key() :: period_key() | day_key()

Any counter key.

pending()

@type pending() :: :flush | :gossip

Which pending column to take: the database flush or the PubSub gossip.

period_key()

@type period_key() :: {String.t(), atom(), DateTime.t()}

A period counter key: {tenant_key, feature, period_start}.

Functions

all_for(tenant_key, period_start)

@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.

apply_remote(key, delta)

@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).

base(key)

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

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

clear_dirty(key)

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

Removes a single key from the dirty set.

clear_touched(key)

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

Removes a single key from the touched set.

day_value(tenant_key, feature, date)

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

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

dirty_keys()

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

Snapshots the current set of dirty keys.

history_key?(arg1)

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

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

incr(tenant_key, feature, qty, period_start)

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

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

mark_dirty(key)

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

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

rebase(key, total, source \\ :flush)

@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.

release(tenant_key, feature, qty, period_start, on \\ nil)

@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.

remote_since_rebase(key)

@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.

reserve(tenant_key, feature, qty, period_start, limit, deferred \\ false)

@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.

restore_pending(key, delta)

@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.

take_pending(key, kind)

@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.

touched_keys()

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

Snapshots the current set of touched keys.

value(tenant_key, feature, period_start)

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

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

warm_day_values(tenant_key, feature)

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

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