AuroraMeter.Counter (Aurora Meter v1.0.0)

View Source

Internal. Not part of the supported API (see API inventory). It may change in any release, including a patch.

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

Events-source features

A feature configured feature_sources: %{name => :events} keeps a row here, because AuroraMeter.usage/2 and the quota functions read it, but the row is maintained differently and two of its columns behave in ways that look wrong and are not.

pending_flush stays at zero and the key is never marked dirty: apply_projection/2 is the only writer of value for such a key other than a with_quota reservation, and it writes neither. That is what keeps a durable event out of Store.snapshot_flush_batch/0, out of Storage.flush_batch/3 and therefore out of aurora_meter_counters (I08).

remote grows without ever being cleared. apply_remote/2 adds to it when a peer gossips a projected delta, and only rebase/3 clears it: the flusher rebases the keys in a batch it wrote, and AuroraMeter.Cluster rebases the keys a peer announced in one, and an events-source key is in neither. A large remote on such a key is therefore the expected steady state, not a diagnostic. remote_since_rebase/1 exists for the flusher's reasoning about buffered keys and has no caller that reaches these.

There is no day bucket at all for an events-source feature: every caller of bump_history/4 is either blocked by a guard (incr/4 from AuroraMeter.track/4, the non-deferred branch of reserve/6 from AuroraMeter.reserve/2,3) or not reached (commit_work/5, which AuroraMeter.Entitlements.with_quota/4 does not call for these features). So AuroraMeter.history/3 reports zeros; the durable series is AuroraMeter.Events.stream/1.

Summary

Types

day_key()

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

key()

Any counter key.

pending()

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

period_key()

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

Functions

all_for(tenant_key, period_start)

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

apply_remote(key, delta)

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)

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

clear_dirty(key)

Removes a single key from the dirty set.

clear_touched(key)

Removes a single key from the touched set.

day_value(tenant_key, feature, date)

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

dirty_keys()

Snapshots the current set of dirty keys.

history_key?(arg1)

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

incr(tenant_key, feature, qty, period_start)

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

mark_dirty(key)

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

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

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.

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

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

remote_since_rebase(key)

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

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

Atomically reserves qty against an optional hard limit.

take_pending(key, kind)

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

Snapshots the current set of touched keys.

value(tenant_key, feature, period_start)

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

warm_day_values(tenant_key, feature)

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.

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