diff --git a/docs-site/src/content/docs/operators/monitoring.mdx b/docs-site/src/content/docs/operators/monitoring.mdx index cb116a982..a7af23af5 100644 --- a/docs-site/src/content/docs/operators/monitoring.mdx +++ b/docs-site/src/content/docs/operators/monitoring.mdx @@ -285,3 +285,131 @@ CODEX_POOLER_MEMORY_SAMPLER_LIMIT_BYTES=1073741824 ``` Use a lower threshold or shorter interval only during active investigation, because logs are the only signal likely to capture a worker or scheduler spike that reaches OOM before the next Prometheus scrape. Worker and scheduler pods do not expose the app `/metrics` endpoint and do not start the Prometheus reporter, so combine their sampler logs with Kubernetes cgroup memory, restart, OOM, and memory-limit metrics. + +## Retained account quota evidence + +The authorized native `/metrics` response also exposes account gauges. It reads +all non-deleted identities, including paused, disabled, never-success and +unassigned accounts, in one `REPEATABLE READ, READ ONLY` transaction with three +set-based queries and a five-second total deadline, including connection-pool +checkout. Timeout or request termination cancels the owned collector task, so +queued work cannot run after its scrape has ended. It does not read account +secrets, refresh OAuth, poll providers, enqueue work, call models, switch accounts +or redeem credits. Inactive accounts' retained observations can become stale. +Allow the scrape timeout to exceed the account read deadline plus normal response +processing. HTTP success (`up=1`) alone does not establish collector coverage. + +There are 36 account gauge families. Every name below has prefix `codex_pooler_` and type **gauge**. Label groups are: +`A = account_id,provider`; `P = A,pool_id`; +`M = A,scope,bucket,window,window_kind,meter_id`; +`O = M,source,observation_id`. Account and pool IDs are internal UUIDs. Provider +is `codex_chatgpt_oauth` or `unknown`; meter and observation identities are opaque +SHA-256 digests, never raw model names, provider meter strings, emails or aliases. +Quota groups have no pool label, even when an account belongs to several pools. + +| Metric suffix | Labels | Meaning | +| --- | --- | --- | +| `account_metrics_collection_success` | none | 1 only for a complete read and every complete account projection. | +| `account_metrics_collection_duration_seconds` | none | Actual collection duration. | +| `account_metrics_snapshot_timestamp_seconds` | none | Captured evaluation time; not provider freshness. Absent on read failure. | +| `account_metrics_accounts`, `account_metrics_memberships` | none | Inventory and distinct membership counts. Absent on read failure. | +| `account_metrics_quota_groups`, `account_metrics_observations` | none | Emitted group and observation counts. Absent on read failure. | +| `account_metrics_evidence_ttl_seconds` | none | Actual configured Evidence TTL (default 900 seconds). | +| `account_info` | A | One per non-deleted enrolled identity. | +| `account_state` | A,state | Persisted lifecycle state. | +| `account_disabled`, `account_reauth_required` | A | Explicit disabled flag and persisted reauthentication requirement. Zero does not certify credentials. | +| `account_projection_complete`, `account_projection_overflow` | A | Complete projection and measured cardinality overflow flags. | +| `account_has_quota_observation` | A | At least one time-visible observation; absence is not zero consumption. | +| `account_last_quota_observation_timestamp_seconds` | A | Latest valid non-future observation time, when known. | +| `account_provider_availability_state` | A,state | `available`, `blocked` or `unknown`, using credential-epoch and TTL rules. Same-epoch blocking persists beyond the TTL. | +| `account_provider_availability_observed_timestamp_seconds` | A | Valid non-future current-epoch provider availability time, when known. | +| `account_quota_readiness` | A,state | `ready`, `weekly_only_probe`, `provider_available_no_windows`, `exhausted`, `stale`, `missing_evidence`, `blocked` or `unknown`. This does not establish pool/model eligibility. | +| `account_pool_membership` | P | Membership presence, including paused/disabled memberships. | +| `account_pool_state` | P,status,health,eligibility | Persisted bounded membership state. | +| `account_pool_reconciliation_state` | P,state | Latest valid terminal `succeeded`, `partial`, `failed`, or `unknown`. | +| `account_pool_reconciliation_timestamp_seconds` | P | Valid non-future terminal reconciliation completion time. | +| `account_pool_last_successful_reconciliation_timestamp_seconds` | P | Whole-reconciliation success time; not quota-only poll success. | +| `account_quota_info` | M | Canonical quota group presence. | +| `account_quota_window_duration_seconds` | M | Actual positive stored duration. | +| `account_quota_routing_selection_present` | M | Whether routing selected a known-time observation in this group. | +| `account_quota_observation_info` | O | Retained time-visible source observation. | +| `account_quota_observation_used_percent`, `account_quota_observation_remaining_percent` | O | Valid reported used percentage and `100-used`; missing or invalid values are omitted. | +| `account_quota_observation_reset_timestamp_seconds` | O | Original reported reset time, including elapsed resets; missing reset omitted. | +| `account_quota_observation_timestamp_seconds` | O | Original valid non-future observation time; missing time omitted. | +| `account_quota_observation_freshness` | O,state | Current `fresh`, `stale` or `unknown` Evidence classification. | +| `account_quota_observation_elapsed` | O | 1 when a known reset elapsed; zero proves no current capacity. | +| `account_quota_observation_used_known` | O | 1 only for a finite reported percentage in `[0,100]`. | +| `account_quota_observation_routing_selected` | O | The exact routing-selected known-time observation. | + +Lifecycle and assignment states are `pending`, `active`, `paused`, `refresh_due`, +`refreshing`, `refresh_failed`, `reauth_required`, `disabled`, `errored`, or +`unknown`. Deleted identities and memberships disappear on the next successful +scrape. Health is `unknown`, `active`, `cooldown`, `degraded`, `disabled`, or +`errored`; eligibility is `eligible`, `ineligible`, or `unknown`. Reconciliation +errors, step messages and HTTP response bodies are never metric labels. + +Scope is `account`, `model`, `upstream_model`, `feature`, or `unknown`; bucket is +`account`, `spark`, `reserve`, or `other`. Reserve requires the exact stored +`GPT-Reserve` / `base_model_inference` descriptor pair. Window labels derive from +duration: `5h`, `7d`, `30d`, or `other`; `window_kind` is `primary`, `secondary`, or +`unknown`, with legacy weekly primary normalized only for the group identity. +Sources are `codex_usage_api`, `codex_response_headers`, `codex_rate_limit_event`, +`codex_rate_limit_error`, `local_reconciliation`, or `unknown`. + +Only valid future observation times are excluded. Legacy missing/malformed times +retain source presence, valid percentages and reset evidence, with freshness +forced to `unknown`. These rows are excluded from the +time-dependent routing adapter; when any is present, readiness is conservative +`unknown` unless the existing evaluator already establishes `blocked` or +`exhausted`. No missing time becomes scrape time. The normal routing path is +unchanged, and metric evaluation does not emit routing decision events. + +Each retained source keeps its own reported percentage, reset and observation +age. The routing-selection gauges identify the selected observation; selection +does not certify provider capacity or establish that other observations describe +the same quota cycle. Differences are not automatically classified as conflicts, +and no combined percentage is computed. The admin evidence dialog, selected +percentage and countdown are unchanged. Account Weekly, model, Spark, Reserve +and distinct unknown meters stay separate. Never sum percentages across sources, windows, +meters, memberships or replicas. Windowless provider availability is not unlimited +quota. There are no new account token/cost/request counters: existing storage +cannot establish monotonic totals or consumption through other clients. + +Bounds are 64 groups, 8 observations per group, 512 stored window rows and 64 +memberships per account. Overflow preserves inventory, sets projection incomplete +and overflow, and omits that account's entire quota/membership section. A malformed +account leaves other accounts visible. Database failure exposes operational +metrics and collector failure, without cached or fabricated account samples. +The maximum current sample count is `8 + 11*A + 5*P + 3*M + 9*O`. + +For example, report age for each original observation, retaining its identity: + +```promql +time() - codex_pooler_account_quota_observation_timestamp_seconds +``` + +Display age beside `account_quota_observation_freshness`, routing selection, +`account_projection_complete` and `account_metrics_collection_success`. Missing +age means unknown observation time. Multiple app replicas export the same DB +facts: scrape one logical account target or deduplicate equivalent observation +identities across instances; never sum replicas. Operational counters keep their +existing per-process aggregation. Historical retention belongs to the monitoring +backend; Pooler retains only each source's current stored observation. + +### Metric identity and earlier fork exporters + +`meter_id` is the existing admin source-evidence group digest: SHA-256 of the +Erlang external-term encoding of the canonical logical window key and additional +meter token, with a legacy weekly primary normalized to secondary. It does not +include the observed value, reset, time or source. `observation_id` hashes the +persisted observation index fields with the `quota-persisted-observation-v1:` +prefix; changing a report's values or timestamps does not create a new identity. + +An earlier experimental 38-family fork exporter used a +`quota-observation-group-v1:` prefix for `meter_id` and exposed two automatic +source/reset conflict families. This 36-family upstream contract follows the +current evidence dialog's unprefixed group digest and omits those classifications. +It is not a drop-in replacement for that fork's time series: account and +observation IDs stay compatible, but meter labels and therefore series identities +change. Preserve existing backend history and update dashboards explicitly when +migrating; do not concatenate or sum the two identities as equivalent samples. diff --git a/lib/codex_pooler/admin/upstream_quota_readiness.ex b/lib/codex_pooler/admin/upstream_quota_readiness.ex index afe9fe738..8c8600a94 100644 --- a/lib/codex_pooler/admin/upstream_quota_readiness.ex +++ b/lib/codex_pooler/admin/upstream_quota_readiness.ex @@ -56,17 +56,21 @@ defmodule CodexPooler.Admin.UpstreamQuotaReadiness do end @spec from_snapshot(RoutingQuotaSnapshot.t()) :: t() - def from_snapshot(%RoutingQuotaSnapshot{} = snapshot) do + @spec from_snapshot(RoutingQuotaSnapshot.t(), keyword()) :: t() + def from_snapshot(%RoutingQuotaSnapshot{} = snapshot, opts \\ []) do routing_snapshot = %{ snapshot | raw_windows: Enum.reject(snapshot.raw_windows, &usage_zero_capacity_primary_window?/1) } eligibility = - QuotaWindows.routing_quota_eligibility_from_snapshot(routing_snapshot, account_only: true) + QuotaWindows.routing_quota_eligibility_from_snapshot( + routing_snapshot, + Keyword.put(opts, :account_only, true) + ) snapshot - |> RoutingQuotaSnapshot.effective_windows() + |> RoutingQuotaSnapshot.effective_windows(opts) |> project_readiness(eligibility, snapshot.as_of) end diff --git a/lib/codex_pooler/metrics/account_projection.ex b/lib/codex_pooler/metrics/account_projection.ex new file mode 100644 index 000000000..ddca7d2f0 --- /dev/null +++ b/lib/codex_pooler/metrics/account_projection.ex @@ -0,0 +1,369 @@ +defmodule CodexPooler.Metrics.AccountProjection do + @moduledoc "Pure, bounded native account metric projection. No polling or credentials." + + alias CodexPooler.Admin.UpstreamQuotaReadiness + alias CodexPooler.Metrics.AccountValues, as: V + alias CodexPooler.Quotas.{Evidence, SourceObservations} + + alias CodexPooler.Upstreams.Quota.{ + AccountAvailabilityStore, + AccountQuotaWindow, + RoutingQuotaSnapshot + } + + alias CodexPooler.Upstreams.Quota.WindowSelector + alias CodexPooler.Upstreams.Schemas.UpstreamIdentity + + @statuses ~w(pending active paused refresh_due refreshing refresh_failed reauth_required deleted disabled errored unknown) + @health ~w(unknown active cooldown degraded disabled errored) + @eligibility ~w(eligible ineligible unknown) + @readiness ~w(ready weekly_only_probe provider_available_no_windows exhausted stale missing_evidence blocked unknown) + @sources ~w(codex_usage_api codex_response_headers codex_rate_limit_event codex_rate_limit_error local_reconciliation unknown) + + def project(snapshot) do + identities = + snapshot.identities |> Enum.reject(&(&1.status == "deleted")) |> Enum.uniq_by(& &1.id) + + account_ids = MapSet.new(identities, & &1.id) + + memberships = + snapshot.memberships + |> Enum.filter(&MapSet.member?(account_ids, &1.upstream_identity_id)) + |> Enum.reject(&(&1.status == "deleted")) + |> Enum.uniq_by(&{&1.upstream_identity_id, &1.pool_id}) + + by_account = Enum.group_by(memberships, & &1.upstream_identity_id) + windows = Enum.group_by(snapshot.windows, & &1.upstream_identity_id) + + accounts = + Enum.map(identities, fn identity -> + project_account( + identity, + Map.get(by_account, identity.id, []), + Map.get(windows, identity.id, []), + snapshot.as_of + ) + end) + + %{ + complete: Enum.all?(accounts, & &1.complete), + samples: Enum.flat_map(accounts, & &1.samples), + accounts: length(identities), + memberships: + Enum.reduce(by_account, 0, fn {_id, rows}, total -> total + row_count(rows) end), + groups: Enum.sum(Enum.map(accounts, & &1.groups)), + observations: Enum.sum(Enum.map(accounts, & &1.observations)), + as_of: snapshot.as_of + } + end + + defp project_account(identity, memberships, rows, as_of) do + labels = [ + account_id: V.uuid!(identity.id), + provider: V.closed(identity.credential_provenance, ["codex_chatgpt_oauth"]) + ] + + inventory = [ + sample(:account_info, labels, 1), + sample( + :account_state, + labels ++ [state: V.closed(identity.status, @statuses -- ["deleted"])], + 1 + ), + sample( + :account_disabled, + labels, + V.flag(identity.status == "disabled" or identity.disabled_at != nil) + ), + sample(:account_reauth_required, labels, V.flag(identity.status == "reauth_required")) + ] + + account_sections(identity, memberships, rows, as_of, labels, inventory) + end + + defp account_sections(identity, memberships, rows, as_of, labels, inventory) do + if row_count(rows) > 512 or row_count(memberships) > 64 do + incomplete(inventory, labels, true) + else + bounded_groups(identity, memberships, rows, as_of, labels, inventory) + end + rescue + _exception -> incomplete(inventory, labels, false) + end + + defp bounded_groups(identity, memberships, rows, as_of, labels, inventory) do + groups = rows |> Enum.map(&window/1) |> SourceObservations.groups(as_of) + + if map_size(groups) > 64 or Enum.any?(groups, fn {_key, values} -> length(values) > 8 end) do + incomplete(inventory, labels, true) + else + complete_sections(identity, memberships, rows, groups, as_of, labels, inventory) + end + end + + defp complete_sections(identity, memberships, rows, groups, as_of, labels, inventory) do + visible = groups |> Map.values() |> List.flatten() + snapshot = routing_snapshot(identity, visible, as_of) + + selected = + snapshot + |> RoutingQuotaSnapshot.effective_windows(emit_telemetry: false) + |> MapSet.new(& &1.id) + + identifiers = Map.new(rows, &{&1.id, V.observation_id(&1)}) + + group_samples = + groups + |> Enum.sort() + |> Enum.flat_map(fn {key, values} -> + quota_samples(key, values, selected, identifiers, labels, as_of) + end) + + samples = + inventory ++ + [ + sample(:account_projection_complete, labels, 1), + sample(:account_projection_overflow, labels, 0) + ] ++ + account_evidence(snapshot, visible, labels) ++ + Enum.flat_map(memberships, &membership_samples(&1, labels, as_of)) ++ group_samples + + %{complete: true, samples: samples, groups: map_size(groups), observations: length(visible)} + end + + defp incomplete(inventory, labels, overflow) do + %{ + complete: false, + samples: + inventory ++ + [ + sample(:account_projection_complete, labels, 0), + sample(:account_projection_overflow, labels, V.flag(overflow)) + ], + groups: 0, + observations: 0 + } + end + + defp window(row) do + row + |> Map.drop([:index_model, :index_upstream_model, :row_count]) + |> then(&struct!(AccountQuotaWindow, &1)) + |> Map.update!(:used_percent, &V.percent/1) + |> Map.update!(:observed_at, &V.timestamp/1) + |> Map.update!(:reset_at, &V.timestamp/1) + |> Map.update!(:last_sync_at, &V.timestamp/1) + |> Map.update!(:updated_at, &V.timestamp/1) + |> unknown_freshness() + end + + defp unknown_freshness(%{observed_at: nil} = window), do: %{window | freshness_state: "unknown"} + defp unknown_freshness(window), do: window + + defp routing_snapshot(identity, windows, as_of) do + identity = struct!(UpstreamIdentity, identity) + known = Enum.filter(windows, &match?(%DateTime{}, &1.observed_at)) + snapshot = RoutingQuotaSnapshot.from_identity(identity, known, as_of) + + # Availability's ordinary clock-skew tolerance is intentionally not an + # assertion that a future report existed at this scrape's snapshot time. + if snapshot.availability && + is_nil(V.visible_timestamp(snapshot.availability.observed_at, as_of)), + do: %{snapshot | availability: nil}, + else: snapshot + end + + defp account_evidence(snapshot, windows, labels) do + as_of = snapshot.as_of + availability = snapshot.availability + current = availability && availability.credential_epoch == snapshot.credential_epoch + observed = if current, do: availability.observed_at + + state = + cond do + AccountAvailabilityStore.available?(availability, snapshot.credential_epoch, as_of) -> + "available" + + AccountAvailabilityStore.blocked?(availability, snapshot.credential_epoch, as_of) -> + "blocked" + + true -> + "unknown" + end + + readiness = UpstreamQuotaReadiness.from_snapshot(snapshot, emit_telemetry: false).state + unknown_time = Enum.any?(windows, &is_nil(&1.observed_at)) + + readiness = + if unknown_time and readiness not in ["blocked", "exhausted"], + do: "unknown", + else: readiness + + last = + windows + |> Enum.map(& &1.observed_at) + |> Enum.reject(&is_nil/1) + |> Enum.max(DateTime, fn -> nil end) + + [ + sample(:account_has_quota_observation, labels, V.flag(windows != [])), + sample(:account_last_quota_observation_timestamp_seconds, labels, V.seconds(last)), + sample(:account_provider_availability_state, labels ++ [state: state], 1), + sample( + :account_provider_availability_observed_timestamp_seconds, + labels, + V.seconds(observed) + ), + sample(:account_quota_readiness, labels ++ [state: V.closed(readiness, @readiness)], 1) + ] + end + + defp membership_samples(row, labels, as_of) do + labels = labels ++ [pool_id: V.uuid!(row.pool_id)] + finished = V.visible_timestamp(row.reconciliation_finished_at, as_of) + status = V.closed(row.reconciliation_status, ~w(succeeded partial failed)) + valid = finished != nil and status != "unknown" + + [ + sample(:account_pool_membership, labels, 1), + sample( + :account_pool_state, + labels ++ + [ + status: V.closed(row.status, @statuses), + health: V.closed(row.health_status, @health), + eligibility: V.closed(row.eligibility_status, @eligibility) + ], + 1 + ), + sample( + :account_pool_reconciliation_state, + labels ++ [state: if(valid, do: status, else: "unknown")], + 1 + ), + sample( + :account_pool_reconciliation_timestamp_seconds, + labels, + if(valid, do: V.seconds(finished)) + ), + sample( + :account_pool_last_successful_reconciliation_timestamp_seconds, + labels, + row.last_successful_refresh_at |> V.visible_timestamp(as_of) |> V.seconds() + ) + ] + end + + defp quota_samples(key, windows, selected, identifiers, labels, as_of) do + first = hd(windows) + labels = labels ++ group_labels(windows, key) + + group = [ + sample(:account_quota_info, labels, 1), + sample(:account_quota_window_duration_seconds, labels, duration(first.window_minutes)), + sample( + :account_quota_routing_selection_present, + labels, + V.flag(Enum.any?(windows, &MapSet.member?(selected, &1.id))) + ) + ] + + group ++ + Enum.flat_map(windows, &observation_samples(&1, labels, selected, identifiers, as_of)) + end + + defp group_labels(windows, key) do + labels = Enum.map(windows, &meter_labels(&1, key)) + buckets = labels |> Enum.map(&Keyword.fetch!(&1, :bucket)) |> Enum.uniq() + + bucket = + case buckets do + [single] -> single + _conflicting_descriptors -> "other" + end + + labels |> hd() |> Keyword.put(:bucket, bucket) + end + + defp meter_labels(window, key) do + {scope, _family, _model, _upstream_model, quota_key, kind, minutes} = + WindowSelector.logical_key(window) + + kind = if kind == "primary" and minutes == 10_080, do: "secondary", else: kind + + [ + scope: V.closed(scope, ~w(account model upstream_model feature)), + bucket: bucket(scope, quota_key, window), + window: window_name(minutes), + window_kind: V.closed(kind, ~w(primary secondary)), + meter_id: key + ] + end + + defp bucket("account", "account", _window), do: "account" + defp bucket(scope, "codex_spark", _window) when scope in ~w(model upstream_model), do: "spark" + + defp bucket(_scope, _key, %{ + limit_name: "GPT-Reserve", + raw_metered_feature: "base_model_inference" + }), + do: "reserve" + + defp bucket(_scope, _key, _window), do: "other" + defp window_name(300), do: "5h" + defp window_name(10_080), do: "7d" + defp window_name(43_200), do: "30d" + defp window_name(_minutes), do: "other" + defp duration(minutes) when is_integer(minutes) and minutes > 0, do: minutes * 60 + defp duration(_minutes), do: nil + + defp observation_samples(window, labels, selected, identifiers, as_of) do + labels = + labels ++ + [ + source: V.closed(window.source, @sources), + observation_id: Map.fetch!(identifiers, window.id) + ] + + used = window.used_percent + remaining = if used, do: Decimal.sub(100, used) + + freshness = + if window.observed_at, do: Evidence.current_freshness_state(window, as_of), else: "unknown" + + [ + sample(:account_quota_observation_info, labels, 1), + sample(:account_quota_observation_used_percent, labels, used), + sample(:account_quota_observation_remaining_percent, labels, remaining), + sample( + :account_quota_observation_reset_timestamp_seconds, + labels, + V.seconds(window.reset_at) + ), + sample(:account_quota_observation_timestamp_seconds, labels, V.seconds(window.observed_at)), + sample( + :account_quota_observation_freshness, + labels ++ [state: V.closed(freshness, ~w(fresh stale unknown))], + 1 + ), + sample( + :account_quota_observation_elapsed, + labels, + V.flag(Evidence.expired?(window, as_of)) + ), + sample(:account_quota_observation_used_known, labels, V.flag(used != nil)), + sample( + :account_quota_observation_routing_selected, + labels, + V.flag(MapSet.member?(selected, window.id)) + ) + ] + end + + defp row_count([]), do: 0 + + defp row_count(rows), + do: max(length(rows), Enum.max(Enum.map(rows, &Map.get(&1, :row_count, 0)))) + + defp sample(name, labels, value), do: {name, labels, value} +end diff --git a/lib/codex_pooler/metrics/account_prometheus.ex b/lib/codex_pooler/metrics/account_prometheus.ex new file mode 100644 index 000000000..7ba1711ed --- /dev/null +++ b/lib/codex_pooler/metrics/account_prometheus.ex @@ -0,0 +1,145 @@ +defmodule CodexPooler.Metrics.AccountPrometheus do + @moduledoc "Pure Prometheus 0.0.4 formatter for fixed account gauge families." + + alias CodexPooler.Metrics.AccountValues, as: V + alias CodexPooler.Quotas.Evidence + + @families [ + account_metrics_collection_success: + "One only when the snapshot and every account projection completed.", + account_metrics_collection_duration_seconds: "Elapsed account collection time in seconds.", + account_metrics_snapshot_timestamp_seconds: + "Snapshot evaluation time; not provider freshness.", + account_metrics_accounts: "Non-deleted enrolled identities in the snapshot.", + account_metrics_memberships: "Distinct non-deleted account and pool memberships.", + account_metrics_quota_groups: "Emitted canonical quota groups.", + account_metrics_observations: "Emitted retained quota observations.", + account_metrics_evidence_ttl_seconds: "Configured quota evidence freshness TTL in seconds.", + account_info: "Non-deleted enrolled account inventory.", + account_state: "Current persisted identity lifecycle state.", + account_disabled: "Explicit disabled status or disabled timestamp; not availability.", + account_reauth_required: "Explicit persisted reauthentication requirement.", + account_projection_complete: "One when this account projection is complete.", + account_projection_overflow: "One for a measured account cardinality bound violation.", + account_has_quota_observation: "Presence of time-visible retained quota evidence.", + account_last_quota_observation_timestamp_seconds: + "Latest valid non-future quota observation time.", + account_provider_availability_state: "Current-epoch provider-reported account availability.", + account_provider_availability_observed_timestamp_seconds: + "Valid current-epoch provider availability observation time.", + account_quota_readiness: "Quota readiness; not assignment or model eligibility.", + account_pool_membership: "Current non-deleted account and pool membership.", + account_pool_state: "Persisted membership status, health and eligibility.", + account_pool_reconciliation_state: "Latest valid non-future terminal reconciliation state.", + account_pool_reconciliation_timestamp_seconds: + "Latest valid terminal reconciliation finished time.", + account_pool_last_successful_reconciliation_timestamp_seconds: + "Last successful whole reconciliation, not quota-only polling.", + account_quota_info: "Canonical retained quota meter group.", + account_quota_window_duration_seconds: "Reported positive quota window duration in seconds.", + account_quota_routing_selection_present: + "Presence of the routing selector observation in this group.", + account_quota_observation_info: "Retained time-visible source quota observation.", + account_quota_observation_used_percent: + "Valid reported used percentage; missing values are omitted.", + account_quota_observation_remaining_percent: + "100 minus valid reported used percentage; not verified capacity.", + account_quota_observation_reset_timestamp_seconds: + "Original reported reset time, including elapsed resets.", + account_quota_observation_timestamp_seconds: + "Original valid non-future provider observation time.", + account_quota_observation_freshness: + "Evidence freshness; missing observation time is unknown.", + account_quota_observation_elapsed: + "One when a known reset has elapsed; zero does not establish capacity.", + account_quota_observation_used_known: "One only for a valid reported used percentage.", + account_quota_observation_routing_selected: "One for the exact routing selector observation." + ] + + def families, do: @families + + def render(result, duration) do + result + |> samples(duration) + |> format() + end + + defp samples({:ok, projection}, duration) do + [ + {:account_metrics_collection_success, [], V.flag(projection.complete)}, + {:account_metrics_collection_duration_seconds, [], duration}, + {:account_metrics_snapshot_timestamp_seconds, [], V.seconds(projection.as_of)}, + {:account_metrics_accounts, [], projection.accounts}, + {:account_metrics_memberships, [], projection.memberships}, + {:account_metrics_quota_groups, [], projection.groups}, + {:account_metrics_observations, [], projection.observations}, + {:account_metrics_evidence_ttl_seconds, [], Evidence.freshness_ttl_seconds()} + | projection.samples + ] + end + + defp samples({:error, _reason}, duration) do + [ + {:account_metrics_collection_success, [], 0}, + {:account_metrics_collection_duration_seconds, [], duration}, + {:account_metrics_evidence_ttl_seconds, [], Evidence.freshness_ttl_seconds()} + ] + end + + defp format(samples) do + samples = Enum.reject(samples, fn {_name, _labels, value} -> is_nil(value) end) + keys = Enum.map(samples, fn {name, labels, _value} -> {name, Enum.sort(labels)} end) + + if length(keys) != MapSet.size(MapSet.new(keys)), + do: raise("duplicate account metric identity") + + grouped = Enum.group_by(samples, &elem(&1, 0)) + + @families + |> Enum.map(fn {name, help} -> + metric = "codex_pooler_#{name}" + + [ + "# HELP ", + metric, + " ", + help, + "\n# TYPE ", + metric, + " gauge\n", + grouped + |> Map.get(name, []) + |> Enum.sort() + |> Enum.map(fn {_name, labels, value} -> + [metric, labels(labels), " ", number(value), "\n"] + end) + ] + end) + |> IO.iodata_to_binary() + end + + defp labels([]), do: "" + + defp labels(labels) do + [ + "{", + Enum.map_intersperse(labels, ",", fn {key, value} -> + [Atom.to_string(key), "=\"", escape(value), "\""] + end), + "}" + ] + end + + defp escape(value) when is_binary(value) do + value + |> String.replace("\\", "\\\\") + |> String.replace("\"", "\\\"") + |> String.replace("\n", "\\n") + end + + defp number(%Decimal{coef: coefficient} = value) when is_integer(coefficient), + do: Decimal.to_string(value, :normal) + + defp number(value) when is_integer(value), do: Integer.to_string(value) + defp number(value) when is_float(value), do: Float.to_string(value) +end diff --git a/lib/codex_pooler/metrics/account_snapshot.ex b/lib/codex_pooler/metrics/account_snapshot.ex new file mode 100644 index 000000000..894e03464 --- /dev/null +++ b/lib/codex_pooler/metrics/account_snapshot.ex @@ -0,0 +1,200 @@ +defmodule CodexPooler.Metrics.AccountSnapshot do + @moduledoc "Read-only, bounded account metadata snapshot. Never loads credentials." + + alias CodexPooler.Metrics.{AccountProjection, AccountPrometheus} + alias CodexPooler.Repo + + @deadline_ms 5_000 + @window_fields ~w(id upstream_identity_id quota_key window_kind window_minutes active_limit credits reset_at used_percent limit_name metered_feature source source_precision quota_scope quota_family model upstream_model raw_limit_id raw_limit_name raw_metered_feature freshness_state last_sync_at observed_at merge_precedence updated_at)a + + @column_keys Map.new( + @window_fields ++ + ~w(status disabled_at credential_provenance metadata pool_id health_status eligibility_status last_successful_refresh_at reconciliation_status reconciliation_finished_at row_count index_model index_upstream_model)a, + &{Atom.to_string(&1), &1} + ) + + @identities_sql """ + SELECT id::text, status, disabled_at, credential_provenance, + jsonb_build_object('credential_epoch', metadata->'credential_epoch', + 'quota_account_availability', metadata->'quota_account_availability') AS metadata + FROM upstream_identities WHERE status IS DISTINCT FROM 'deleted' ORDER BY id + """ + + @memberships_sql """ + SELECT upstream_identity_id::text, pool_id::text, status, health_status, + eligibility_status, last_successful_refresh_at, + reconciliation_status, reconciliation_finished_at, row_count + FROM ( + SELECT a.upstream_identity_id, a.pool_id, a.status, a.health_status, + a.eligibility_status, a.last_successful_refresh_at, + CASE WHEN jsonb_typeof(a.metadata->'last_reconciliation'->'status') = 'string' + THEN a.metadata->'last_reconciliation'->>'status' END AS reconciliation_status, + CASE WHEN jsonb_typeof(a.metadata->'last_reconciliation'->'finished_at') = 'string' + THEN a.metadata->'last_reconciliation'->>'finished_at' END AS reconciliation_finished_at, + count(*) OVER (PARTITION BY a.upstream_identity_id) AS row_count, + row_number() OVER (PARTITION BY a.upstream_identity_id ORDER BY a.pool_id) AS position + FROM pool_upstream_assignments a + JOIN upstream_identities i ON i.id = a.upstream_identity_id + WHERE i.status IS DISTINCT FROM 'deleted' AND a.status IS DISTINCT FROM 'deleted' + ) bounded WHERE position <= 65 ORDER BY upstream_identity_id, pool_id + """ + + @doc "Builds each response afresh; failures expose health only and no stale account cache." + def scrape(loader \\ &load/0) do + started = System.monotonic_time(:microsecond) + result = collect(loader) + elapsed = (System.monotonic_time(:microsecond) - started) / 1_000_000 + render(result, elapsed) + end + + defp render(result, elapsed) do + AccountPrometheus.render(result, elapsed) + rescue + _exception -> AccountPrometheus.render({:error, :collection_failed}, elapsed) + end + + defp collect(loader) do + with {:ok, snapshot} <- loader.(), do: {:ok, AccountProjection.project(snapshot)} + rescue + _exception -> {:error, :collection_failed} + catch + :exit, _reason -> {:error, :collection_failed} + end + + @doc "Returns a fresh snapshot or a fixed error; database exception text is never exposed." + def load(repo \\ Repo) do + deadline = System.monotonic_time(:millisecond) + @deadline_ms + + # DBConnection arms its deadline only after checkout succeeds. A single + # owned stream task also bounds a completely occupied connection pool. + # The stream kills and awaits its worker on timeout or caller termination; + # it never leaves a detached collector waiting to query after this return. + [repo] + |> Task.async_stream(&read_snapshot(&1, deadline), + max_concurrency: 1, + timeout: remaining(deadline), + on_timeout: :kill_task + ) + |> Enum.to_list() + |> collection_result() + rescue + _exception -> {:error, :snapshot_unavailable} + catch + :exit, _reason -> {:error, :snapshot_unavailable} + end + + defp collection_result([{:ok, result}]), do: normalize_result(result) + defp collection_result(_result), do: {:error, :snapshot_unavailable} + + defp read_snapshot(repo, deadline) do + repo.transaction( + fn -> + query(repo, "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY", deadline) + as_of = DateTime.utc_now() + identities = query(repo, @identities_sql, deadline) + memberships = query(repo, @memberships_sql, deadline) + windows = query(repo, windows_sql(), deadline) + %{as_of: as_of, identities: identities, memberships: memberships, windows: windows} + end, + timeout: remaining(deadline), + log: false + ) + |> normalize_result() + rescue + _exception -> {:error, :snapshot_unavailable} + catch + :exit, _reason -> {:error, :snapshot_unavailable} + end + + defp normalize_result({:ok, snapshot}), do: {:ok, snapshot} + defp normalize_result(_result), do: {:error, :snapshot_unavailable} + + defp query(repo, sql, deadline) do + result = repo.query!(sql, [], timeout: remaining(deadline), log: false) + + Enum.map(result.rows || [], fn values -> + result.columns + |> Enum.zip(values) + |> Map.new(fn {key, value} -> {Map.fetch!(@column_keys, key), value} end) + end) + end + + # The 513th/65th row is an overflow sentinel. Counts are per identity; no + # global LIMIT can quietly omit accounts later in the ordering. + defp windows_sql do + columns = Enum.map_join(@window_fields, ", ", &"w.#{&1}") + projected = Enum.map_join(@window_fields, ", ", &projected_column/1) + + """ + SELECT #{projected}, metadata, index_model, index_upstream_model, row_count + FROM ( + SELECT #{columns}, + #{window_metadata_sql()} AS metadata, + COALESCE(lower(w.model), '') AS index_model, + COALESCE(lower(w.upstream_model), '') AS index_upstream_model, + count(*) OVER (PARTITION BY w.upstream_identity_id) AS row_count, + row_number() OVER (PARTITION BY w.upstream_identity_id ORDER BY w.id) AS position + FROM account_quota_windows w + JOIN upstream_identities i ON i.id = w.upstream_identity_id + WHERE i.status IS DISTINCT FROM 'deleted' + ) bounded WHERE position <= 513 ORDER BY upstream_identity_id, id + """ + end + + defp window_metadata_sql do + scalars = ~w(rate_limit_allowed rate_limit_reached reset_state) + marker = "w.metadata->'__quota_cycle_confirmation_v1'" + + keys = + ~w(version scope family key kind minutes model upstream_model reset_at provider_observed_at confirmed_at source_class) + + expected = Enum.map_join(keys, ",", &"'#{&1}'") + fields = Enum.map_join(keys, ",", &"'#{&1}', #{marker}->'#{&1}'") + + types = + Enum.map_join(keys, " AND ", fn key -> + allowed = + case key do + numeric when numeric in ~w(version minutes) -> "'number'" + optional when optional in ~w(model upstream_model) -> "'string','null'" + _string -> "'string'" + end + + "jsonb_typeof(#{marker}->'#{key}') IN (#{allowed})" + end) + + ordinary = + Enum.map_join(scalars, " || ", fn key -> + allowed = if key == "reset_state", do: "'string','null'", else: "'boolean','null'" + + """ + CASE WHEN w.metadata ? '#{key}' THEN jsonb_build_object('#{key}', + CASE WHEN jsonb_typeof(w.metadata->'#{key}') IN (#{allowed}) + THEN w.metadata->'#{key}' ELSE 'null'::jsonb END) + ELSE '{}'::jsonb END + """ + end) + + # Check the original marker before reconstruction: stripping unknown keys + # must never make a rejected marker pass valid_marker's exact-size guard. + ordinary <> + """ + || CASE WHEN jsonb_typeof(#{marker}) = 'object' THEN + CASE WHEN (SELECT count(*) FROM jsonb_object_keys(#{marker})) = 12 + AND (#{marker}) ?& ARRAY[#{expected}] AND #{types} + THEN jsonb_build_object('__quota_cycle_confirmation_v1', jsonb_build_object(#{fields})) + ELSE '{}'::jsonb END + ELSE '{}'::jsonb END + """ + end + + defp projected_column(field) when field in [:id, :upstream_identity_id], do: "#{field}::text" + defp projected_column(field), do: to_string(field) + + defp remaining(deadline) do + case deadline - System.monotonic_time(:millisecond) do + milliseconds when milliseconds > 0 -> milliseconds + _elapsed -> raise "account snapshot deadline exceeded" + end + end +end diff --git a/lib/codex_pooler/metrics/account_values.ex b/lib/codex_pooler/metrics/account_values.ex new file mode 100644 index 000000000..154f3175a --- /dev/null +++ b/lib/codex_pooler/metrics/account_values.ex @@ -0,0 +1,64 @@ +defmodule CodexPooler.Metrics.AccountValues do + @moduledoc false + + def timestamp(%DateTime{} = value), do: value + def timestamp(%NaiveDateTime{} = value), do: DateTime.from_naive!(value, "Etc/UTC") + + def timestamp(value) when is_binary(value) do + case DateTime.from_iso8601(value) do + {:ok, parsed, _offset} -> parsed + _invalid -> nil + end + end + + def timestamp(_value), do: nil + + def visible_timestamp(value, as_of) do + case timestamp(value) do + nil -> nil + time -> if DateTime.compare(time, as_of) != :gt, do: time + end + end + + def seconds(nil), do: nil + def seconds(time), do: DateTime.to_unix(time, :microsecond) / 1_000_000 + + def percent(%Decimal{coef: coefficient} = value) when is_integer(coefficient) do + if Decimal.compare(value, 0) != :lt and Decimal.compare(value, 100) != :gt, do: value + end + + def percent(_value), do: nil + + def uuid!(value) do + case Ecto.UUID.cast(value) do + {:ok, uuid} -> uuid + :error -> raise ArgumentError, "invalid internal metric identity" + end + end + + def closed(value, allowed), do: if(value in allowed, do: value, else: "unknown") + def flag(true), do: 1 + def flag(false), do: 0 + + # SQL supplies PostgreSQL lower(model) and lower(upstream_model). Other + # strings deliberately retain their persisted case, spacing and nullability. + def observation_id(row) do + tuple = { + row.upstream_identity_id, + row.quota_scope, + row.quota_family, + row.index_model, + row.index_upstream_model, + row.quota_key, + row.window_kind, + row.window_minutes, + row.source, + row.raw_limit_id || "", + row.raw_limit_name || "", + row.raw_metered_feature || "" + } + + :crypto.hash(:sha256, "quota-persisted-observation-v1:" <> :erlang.term_to_binary(tuple)) + |> Base.encode16(case: :lower) + end +end diff --git a/lib/codex_pooler/quotas/source_observations.ex b/lib/codex_pooler/quotas/source_observations.ex new file mode 100644 index 000000000..a708d3f2b --- /dev/null +++ b/lib/codex_pooler/quotas/source_observations.ex @@ -0,0 +1,29 @@ +defmodule CodexPooler.Quotas.SourceObservations do + @moduledoc "Pure retained quota grouping shared by source evidence dialogs and metrics." + + alias CodexPooler.Quotas.AdditionalMeterIdentity + alias CodexPooler.Upstreams.Quota.AccountQuotaWindow + alias CodexPooler.Upstreams.Quota.WindowSelector + + @spec group_key(AccountQuotaWindow.t()) :: String.t() + def group_key(%AccountQuotaWindow{window_kind: "primary", window_minutes: 10_080} = window), + do: group_key(%{window | window_kind: "secondary"}) + + def group_key(window) do + {WindowSelector.logical_key(window), AdditionalMeterIdentity.token(window)} + |> :erlang.term_to_binary() + |> then(&:crypto.hash(:sha256, &1)) + |> Base.encode16(case: :lower) + end + + def groups(raw_windows, as_of) do + raw_windows + |> Enum.reject(&future_observation?(&1, as_of)) + |> Enum.group_by(&group_key/1) + end + + defp future_observation?(%{observed_at: %DateTime{} = observed_at}, as_of), + do: DateTime.compare(observed_at, as_of) == :gt + + defp future_observation?(_window, _as_of), do: false +end diff --git a/lib/codex_pooler/upstreams/quota/routing_quota_snapshot.ex b/lib/codex_pooler/upstreams/quota/routing_quota_snapshot.ex index 2582f2dca..345ace3f8 100644 --- a/lib/codex_pooler/upstreams/quota/routing_quota_snapshot.ex +++ b/lib/codex_pooler/upstreams/quota/routing_quota_snapshot.ex @@ -86,10 +86,11 @@ defmodule CodexPooler.Upstreams.Quota.RoutingQuotaSnapshot do end @spec effective_windows(t()) :: [AccountQuotaWindow.t()] - def effective_windows(%__MODULE__{as_of: as_of} = snapshot) do + @spec effective_windows(t(), keyword()) :: [AccountQuotaWindow.t()] + def effective_windows(%__MODULE__{as_of: as_of} = snapshot, opts \\ []) do snapshot |> time_visible_raw_windows() - |> Routing.reject_superseded_primary_windows(as_of) + |> Routing.reject_superseded_primary_windows(as_of, opts) |> WindowSelector.logical_windows(as_of) end diff --git a/lib/codex_pooler/upstreams/quota/windows/routing.ex b/lib/codex_pooler/upstreams/quota/windows/routing.ex index 611d0212c..acccfb61f 100644 --- a/lib/codex_pooler/upstreams/quota/windows/routing.ex +++ b/lib/codex_pooler/upstreams/quota/windows/routing.ex @@ -20,7 +20,7 @@ defmodule CodexPooler.Upstreams.Quota.Windows.Routing do routing_windows = windows |> Enum.filter(&window_in_model_scope?(&1, opts)) - |> reject_superseded_primary_windows(timestamp) + |> reject_superseded_primary_windows(timestamp, opts) |> WindowSelector.logical_windows(timestamp) |> select_current_account_primary_variant(timestamp) @@ -173,18 +173,13 @@ defmodule CodexPooler.Upstreams.Quota.Windows.Routing do """ @spec reject_superseded_primary_windows([Quota.AccountQuotaWindow.t()], DateTime.t()) :: [Quota.AccountQuotaWindow.t()] - def reject_superseded_primary_windows(windows, timestamp \\ now()) when is_list(windows) do + @spec reject_superseded_primary_windows([Quota.AccountQuotaWindow.t()], DateTime.t(), keyword()) :: + [Quota.AccountQuotaWindow.t()] + def reject_superseded_primary_windows(windows, timestamp \\ now(), opts \\ []) + when is_list(windows) do Enum.reject(windows, fn window -> if superseded_primary_window?(window, windows, timestamp) do - :telemetry.execute( - [:codex_pooler, :quota, :cycle, :decision], - %{count: 1}, - %{ - scope: quota_scope(window), - decision: :superseded_primary_rejected, - source: source_class(window) - } - ) + record_superseded_rejection(window, opts) true else @@ -193,6 +188,20 @@ defmodule CodexPooler.Upstreams.Quota.Windows.Routing do end) end + defp record_superseded_rejection(window, opts) do + if Keyword.get(opts, :emit_telemetry, true) do + :telemetry.execute( + [:codex_pooler, :quota, :cycle, :decision], + %{count: 1}, + %{ + scope: quota_scope(window), + decision: :superseded_primary_rejected, + source: source_class(window) + } + ) + end + end + defp quota_scope(%Quota.AccountQuotaWindow{quota_scope: scope}) when scope in ["model", "upstream_model"], do: "model" diff --git a/lib/codex_pooler_web/controllers/operations/metrics_controller.ex b/lib/codex_pooler_web/controllers/operations/metrics_controller.ex index 7724f5c56..dec384e3f 100644 --- a/lib/codex_pooler_web/controllers/operations/metrics_controller.ex +++ b/lib/codex_pooler_web/controllers/operations/metrics_controller.ex @@ -2,13 +2,18 @@ defmodule CodexPoolerWeb.Operations.MetricsController do use CodexPoolerWeb, :controller alias CodexPooler.InstanceSettings + alias CodexPooler.Metrics.AccountSnapshot def show(conn, _params) do case authorize_metrics(conn) do :ok -> conn |> put_resp_content_type("text/plain; version=0.0.4") - |> send_resp(200, TelemetryMetricsPrometheus.Core.scrape()) + |> send_resp(200, [ + TelemetryMetricsPrometheus.Core.scrape(), + "\n", + AccountSnapshot.scrape() + ]) {:error, reason} -> conn diff --git a/lib/codex_pooler_web/live/admin/read_models/upstream_accounts_read_model/quota_observations.ex b/lib/codex_pooler_web/live/admin/read_models/upstream_accounts_read_model/quota_observations.ex index dd6aa8cc3..3b0a6c7d0 100644 --- a/lib/codex_pooler_web/live/admin/read_models/upstream_accounts_read_model/quota_observations.ex +++ b/lib/codex_pooler_web/live/admin/read_models/upstream_accounts_read_model/quota_observations.ex @@ -1,9 +1,8 @@ defmodule CodexPoolerWeb.Admin.UpstreamAccountsReadModel.QuotaObservations do @moduledoc false - alias CodexPooler.Quotas.{AdditionalMeterIdentity, Evidence} + alias CodexPooler.Quotas.{Evidence, SourceObservations} alias CodexPooler.Upstreams.Quota.AccountQuotaWindow - alias CodexPooler.Upstreams.Quota.WindowSelector alias CodexPoolerWeb.DateTimeDisplay @sources %{ @@ -29,12 +28,7 @@ defmodule CodexPoolerWeb.Admin.UpstreamAccountsReadModel.QuotaObservations do } @spec group_key(AccountQuotaWindow.t()) :: String.t() - def group_key(%AccountQuotaWindow{window_kind: "primary", window_minutes: 10_080} = window), - do: group_key(%{window | window_kind: "secondary"}) - - def group_key(window) do - fingerprint({WindowSelector.logical_key(window), AdditionalMeterIdentity.token(window)}) - end + defdelegate group_key(window), to: SourceObservations @spec project(AccountQuotaWindow.t(), DateTimeDisplay.preferences(), DateTime.t()) :: observation() diff --git a/test/codex_pooler/metrics/account_projection_test.exs b/test/codex_pooler/metrics/account_projection_test.exs new file mode 100644 index 000000000..fd32b20f4 --- /dev/null +++ b/test/codex_pooler/metrics/account_projection_test.exs @@ -0,0 +1,521 @@ +defmodule CodexPooler.Metrics.AccountProjectionTest do + use ExUnit.Case, async: true + + alias CodexPooler.Metrics.{AccountProjection, AccountPrometheus, AccountSnapshot, AccountValues} + alias CodexPooler.Quotas.{Evidence, SourceObservations} + + alias CodexPooler.Upstreams.Quota.{ + AccountAvailabilityStore, + AccountQuotaWindow, + RoutingQuotaSnapshot + } + + alias CodexPooler.Upstreams.Quota.Windows.Routing + alias CodexPooler.Upstreams.Schemas.UpstreamIdentity + alias CodexPoolerWeb.Admin.UpstreamAccountsReadModel.QuotaObservations + + @now ~U[2026-09-07 06:00:00Z] + @account "11111111-1111-4111-8111-111111111111" + @pool "22222222-2222-4222-8222-222222222222" + + test "all lifecycle states and unassigned identities have one inventory row" do + statuses = + ~w(active paused disabled pending refresh_failed reauth_required refreshing refresh_due errored unexpected deleted) + + identities = Enum.map(statuses, &identity(%{id: Ecto.UUID.generate(), status: &1})) + p = AccountProjection.project(snapshot([], [], identities)) + assert p.complete + assert p.accounts == length(statuses) - 1 + assert length(samples(p, :account_info)) == p.accounts + + assert Enum.any?(samples(p, :account_state), fn {_, labels, _} -> + labels[:state] == "unknown" + end) + + assert samples(p, :account_last_quota_observation_timestamp_seconds) == [] + assert Enum.all?(samples(p, :account_has_quota_observation), &(elem(&1, 2) == 0)) + end + + test "memberships deduplicate independently of account quotas and use closed state domains" do + for status <- + ~w(pending active paused refresh_due refreshing refresh_failed reauth_required disabled errored unexpected) do + member = + membership(%{ + status: status, + health_status: "provider-secret", + eligibility_status: "provider-secret" + }) + + p = + project([window()], [ + member, + member, + membership(%{pool_id: Ecto.UUID.generate()}), + membership(%{status: "deleted"}) + ]) + + assert p.complete + assert p.memberships == 2 + assert p.groups == 1 + assert p.observations == 1 + assert length(samples(p, :account_pool_membership)) == 2 + + refute Enum.any?(samples(p, :account_quota_info), fn {_, labels, _} -> + Keyword.has_key?(labels, :pool_id) + end) + + assert Enum.any?(samples(p, :account_pool_state), fn {_, labels, _} -> + labels[:status] == if(status == "unexpected", do: "unknown", else: status) and + labels[:health] == "unknown" and labels[:eligibility] == "unknown" + end) + end + end + + test "missing, malformed, nonfinite and out-of-range usage never becomes zero" do + for invalid <- [ + nil, + "42", + Decimal.new("NaN"), + Decimal.new("Infinity"), + Decimal.new(-1), + Decimal.new(101) + ] do + p = project([window(%{used_percent: invalid})]) + assert p.complete + assert samples(p, :account_quota_observation_used_percent) == [] + assert samples(p, :account_quota_observation_remaining_percent) == [] + assert [{_, _, 0}] = samples(p, :account_quota_observation_used_known) + end + + for {used, remaining} <- [{0, 100}, {100, 0}] do + p = project([window(%{used_percent: Decimal.new(used)})]) + assert [{_, _, value}] = samples(p, :account_quota_observation_remaining_percent) + assert Decimal.equal?(value, remaining) + end + end + + test "unknown-time evidence retains values and resets without positive readiness" do + for time <- [nil, "invalid-time"] do + missing = + window(%{ + observed_at: time, + used_percent: Decimal.new(90), + source: "codex_response_headers" + }) + + p = project([window(), missing]) + assert p.complete + assert p.observations == 2 + assert length(samples(p, :account_quota_observation_timestamp_seconds)) == 1 + assert length(samples(p, :account_quota_observation_reset_timestamp_seconds)) == 2 + + assert Enum.any?(samples(p, :account_quota_observation_freshness), fn {_, labels, _} -> + labels[:state] == "unknown" + end) + + assert [{_, [account_id: @account, provider: _, state: "unknown"], 1}] = + samples(p, :account_quota_readiness) + + assert [{_, labels, 0}] = + Enum.filter(samples(p, :account_quota_observation_routing_selected), fn {_, labels, + _} -> + labels[:source] == "codex_response_headers" + end) + + assert labels[:observation_id] == AccountValues.observation_id(missing) + legacy = as_window(%{missing | observed_at: nil}) + key = SourceObservations.group_key(legacy) + assert key == QuotaObservations.group_key(legacy) + end + end + + test "TTL, elapsed/resetless and future evidence keep distinct freshness and visibility" do + for {age, state} <- [ + {Evidence.freshness_ttl_seconds(), "fresh"}, + {Evidence.freshness_ttl_seconds() + 1, "stale"} + ] do + p = project([window(%{observed_at: DateTime.add(@now, -age)})]) + assert [{_, labels, 1}] = samples(p, :account_quota_observation_freshness) + assert labels[:state] == state + end + + p = + project([ + window(%{reset_at: DateTime.add(@now, -1)}), + window(%{observed_at: DateTime.add(@now, 1), source: "codex_response_headers"}) + ]) + + assert p.observations == 1 + assert [{_, _, 1}] = samples(p, :account_quota_observation_elapsed) + assert length(samples(p, :account_quota_observation_reset_timestamp_seconds)) == 1 + + assert samples( + project([window(%{reset_at: nil})]), + :account_quota_observation_reset_timestamp_seconds + ) == [] + end + + test "retained differences preserve original values and routing provenance without classification" do + for {older_used, newer_used} <- [{5, 6}, {6, 5}, {6, 6}], gap <- [60, 61] do + old = + window(%{ + source: "codex_response_headers", + observed_at: DateTime.add(@now, -30), + used_percent: Decimal.new(older_used) + }) + + fresh = + window(%{ + used_percent: Decimal.new(newer_used), + reset_at: DateTime.add(old.reset_at, gap) + }) + + p = project([old, fresh]) + assert p.complete + assert p.observations == 2 + + used = + Map.new(samples(p, :account_quota_observation_used_percent), fn {_, labels, value} -> + {labels[:source], value} + end) + + resets = + Map.new(samples(p, :account_quota_observation_reset_timestamp_seconds), fn {_, labels, + value} -> + {labels[:source], value} + end) + + assert used == %{ + "codex_response_headers" => old.used_percent, + "codex_usage_api" => fresh.used_percent + } + + assert resets == %{ + "codex_response_headers" => AccountValues.seconds(old.reset_at), + "codex_usage_api" => AccountValues.seconds(fresh.reset_at) + } + + assert [{_, _, 1}] = samples(p, :account_quota_routing_selection_present) + + selected = + RoutingQuotaSnapshot.from_identity( + struct!(UpstreamIdentity, identity()), + Enum.map([old, fresh], &as_window/1), + @now + ) + |> RoutingQuotaSnapshot.effective_windows() + |> MapSet.new(& &1.source) + + for {_, labels, value} <- samples(p, :account_quota_observation_routing_selected) do + assert value == if(MapSet.member?(selected, labels[:source]), do: 1, else: 0) + end + + body = AccountPrometheus.render({:ok, p}, 0.01) + refute body =~ "conflict" + refute body =~ "uncertain" + end + end + + test "meter identity preserves the upstream evidence dialog digest across report updates" do + # Existing source-dialog identity, captured before extracting its grouping helper. + expected = "2f27296639a30fd4e234f582a14299449afa05c7837250141b3c6bb11c6ef3ea" + + for row <- [ + window(), + window(%{window_kind: "primary", source: "codex_response_headers"}), + window(%{observed_at: nil, reset_at: nil, used_percent: Decimal.new(90)}) + ] do + assert SourceObservations.group_key(as_window(row)) == expected + assert QuotaObservations.group_key(as_window(row)) == expected + assert [{_, labels, 1}] = samples(project([row]), :account_quota_info) + assert labels[:meter_id] == expected + end + end + + test "persistence digest exactly separates indexed raw fields from canonical groups" do + original = window(%{model: "Model", index_model: "model", raw_limit_name: nil}) + + equivalent = %{ + original + | model: "MODEL", + raw_limit_name: "", + id: Ecto.UUID.generate(), + observed_at: nil, + reset_at: nil, + used_percent: nil + } + + assert AccountValues.observation_id(original) == AccountValues.observation_id(equivalent) + + for different <- [ + %{original | raw_limit_name: "X"}, + %{original | raw_limit_name: "x"}, + %{original | source: "Codex_usage_api"}, + %{original | window_kind: "primary"} + ] do + refute AccountValues.observation_id(original) == AccountValues.observation_id(different) + end + + refute AccountValues.observation_id(%{original | raw_limit_name: "X"}) == + AccountValues.observation_id(%{original | raw_limit_name: "x"}) + + assert SourceObservations.group_key(as_window(original)) == + SourceObservations.group_key(as_window(%{original | window_kind: "primary"})) + end + + test "separate durations, model, Spark, Reserve and opaque meters never flatten" do + windows = [ + window(), + window(%{window_minutes: 300, window_kind: "primary"}), + window(%{window_minutes: 43_200, window_kind: "primary"}), + window(%{ + quota_scope: "model", + quota_key: "private-model", + model: "private-model", + index_model: "private-model" + }), + window(%{ + quota_scope: "model", + quota_key: "codex_spark", + quota_family: "codex_spark", + model: "gpt-5.3-codex-spark", + index_model: "gpt-5.3-codex-spark" + }), + window(%{ + quota_scope: "feature", + quota_key: "reserve", + quota_family: "reserve", + limit_name: "GPT-Reserve", + raw_metered_feature: "base_model_inference" + }), + window(%{ + quota_scope: "feature", + quota_key: "unknown", + raw_metered_feature: "private-meter-a" + }), + window(%{ + quota_scope: "feature", + quota_key: "unknown", + raw_metered_feature: "private-meter-b" + }) + ] + + p = project(windows) + assert p.complete + assert p.groups == 8 + + assert Enum.sort( + Enum.uniq( + Enum.map(samples(p, :account_quota_info), fn {_, labels, _} -> labels[:bucket] end) + ) + ) == ~w(account other reserve spark) + + body = AccountPrometheus.render({:ok, p}, 0.01) + + for sentinel <- + ~w(private-model private-meter-a private-meter-b base_model_inference GPT-Reserve) do + refute body =~ sentinel + end + + assert length( + String.split(body, "\n") + |> Enum.filter(&String.starts_with?(&1, "codex_pooler_")) + ) <= 8 + 11 + 3 * 8 + 9 * 8 + end + + test "availability keeps same-epoch blocked state but expires available observations" do + for {state, age, epoch, expected} <- [ + {:available, 0, 1, "available"}, + {:available, 901, 1, "unknown"}, + {:blocked, 90_000, 1, "blocked"}, + {:blocked, 0, 2, "unknown"}, + {:available, -1, 1, "unknown"} + ] do + metadata = %{ + "credential_epoch" => 1, + "quota_account_availability" => + AccountAvailabilityStore.encode!(state, DateTime.add(@now, -age), epoch) + } + + p = AccountProjection.project(snapshot([], [], [identity(%{metadata: metadata})])) + assert p.complete + assert [{_, labels, 1}] = samples(p, :account_provider_availability_state) + assert labels[:state] == expected + + if epoch != 1 or age < 0, + do: assert(samples(p, :account_provider_availability_observed_timestamp_seconds) == []) + end + + metadata = %{ + "credential_epoch" => 1, + "quota_account_availability" => AccountAvailabilityStore.encode!(:available, @now, 1) + } + + p = AccountProjection.project(snapshot([], [], [identity(%{metadata: metadata})])) + assert [{_, labels, 1}] = samples(p, :account_quota_readiness) + assert labels[:state] == "provider_available_no_windows" + end + + test "overflow omits the whole account section and malformed data isolates its account" do + overflows = [ + [window(%{row_count: 513})], + Enum.map(1..65, &window(%{quota_key: "meter-#{&1}", quota_scope: "feature"})), + Enum.map(1..9, &window(%{raw_limit_name: "raw-#{&1}"})) + ] + + for rows <- overflows do + p = project(rows, [membership()]) + refute p.complete + assert [{_, _, 1}] = samples(p, :account_projection_overflow) + assert samples(p, :account_pool_membership) == [] + assert samples(p, :account_quota_info) == [] + end + + refute project([], [membership(%{row_count: 65})]).complete + bad = identity(%{id: Ecto.UUID.generate(), metadata: %{}}) + + p = + AccountProjection.project( + snapshot( + [window(), Map.put(window(%{upstream_identity_id: bad.id}), :unexpected_field, true)], + [], + [identity(), bad] + ) + ) + + assert p.accounts == 2 + assert length(samples(p, :account_info)) == 2 + refute p.complete + assert p.groups == 1 + end + + test "default routing telemetry is preserved and metrics evaluation emits none" do + event = [:codex_pooler, :quota, :cycle, :decision] + ref = make_ref() + + :telemetry.attach( + ref, + event, + fn _, _, metadata, pid -> send(pid, {:decision, metadata}) end, + self() + ) + + on_exit(fn -> :telemetry.detach(ref) end) + + old = + window(%{ + window_kind: "primary", + window_minutes: 300, + observed_at: DateTime.add(@now, -1800), + last_sync_at: DateTime.add(@now, -1800) + }) + + fresh = window() + values = Enum.map([old, fresh], &as_window/1) + normal = Routing.reject_superseded_primary_windows(values, @now) + assert_receive {:decision, %{decision: :superseded_primary_rejected}} + + assert normal == + Routing.reject_superseded_primary_windows(values, @now, emit_telemetry: false) + + refute_receive {:decision, _} + p = project([old, fresh]) + assert p.complete + refute_receive {:decision, _} + + snapshot = + RoutingQuotaSnapshot.from_identity(struct!(UpstreamIdentity, identity()), values, @now) + + assert RoutingQuotaSnapshot.effective_windows(snapshot) == + RoutingQuotaSnapshot.effective_windows(snapshot, emit_telemetry: false) + + assert_receive {:decision, _} + end + + test "next scrape drops removed data and failures expose health without fabricated inventory" do + first = AccountSnapshot.scrape(fn -> {:ok, snapshot([window()], [membership()])} end) + assert first =~ @account + second = AccountSnapshot.scrape(fn -> {:ok, snapshot([], [], [])} end) + refute second =~ @account + failure = AccountSnapshot.scrape(fn -> raise "private-sentinel" end) + assert failure =~ "codex_pooler_account_metrics_collection_success 0" + refute failure =~ "private-sentinel" + refute failure =~ "codex_pooler_account_metrics_accounts 0" + end + + defp samples(projection, metric) do + Enum.filter(projection.samples, fn {name, _labels, value} -> + name == metric and value != nil + end) + end + + defp project(windows, memberships \\ []), + do: AccountProjection.project(snapshot(windows, memberships)) + + defp snapshot(windows, memberships, identities \\ [identity()]), + do: %{as_of: @now, identities: identities, memberships: memberships, windows: windows} + + defp identity(attrs \\ %{}), + do: + Map.merge( + %{ + id: @account, + status: "active", + disabled_at: nil, + credential_provenance: "codex_chatgpt_oauth", + metadata: %{"credential_epoch" => 1} + }, + attrs + ) + + defp membership(attrs \\ %{}), + do: + Map.merge( + %{ + upstream_identity_id: @account, + pool_id: @pool, + status: "active", + health_status: "active", + eligibility_status: "eligible", + reconciliation_status: nil, + reconciliation_finished_at: nil, + last_successful_refresh_at: nil + }, + attrs + ) + + defp as_window(row), + do: + struct!( + AccountQuotaWindow, + Map.drop(row, [:index_model, :index_upstream_model, :row_count]) + ) + + defp window(attrs \\ %{}) do + base = %AccountQuotaWindow{ + id: Ecto.UUID.generate(), + upstream_identity_id: @account, + quota_key: "account", + quota_scope: "account", + quota_family: "account", + window_kind: "secondary", + window_minutes: 10_080, + source: "codex_usage_api", + source_precision: "observed", + freshness_state: "fresh", + observed_at: @now, + last_sync_at: @now, + updated_at: @now, + reset_at: DateTime.add(@now, 86_400), + used_percent: Decimal.new(6), + merge_precedence: 60, + metadata: %{} + } + + base + |> Map.from_struct() + |> Map.drop([:__meta__]) + |> Map.merge(%{index_model: "", index_upstream_model: ""}) + |> Map.merge(attrs) + end +end diff --git a/test/codex_pooler/metrics/account_prometheus_test.exs b/test/codex_pooler/metrics/account_prometheus_test.exs new file mode 100644 index 000000000..59627ada4 --- /dev/null +++ b/test/codex_pooler/metrics/account_prometheus_test.exs @@ -0,0 +1,43 @@ +defmodule CodexPooler.Metrics.AccountPrometheusTest do + use ExUnit.Case, async: true + + alias CodexPooler.Metrics.AccountPrometheus + + test "fixed gauge declarations, exact escaping and unique deterministic sample keys" do + samples = [{:account_info, [account_id: "synthetic\\quote\"\nline", provider: "unknown"], 1}] + body = AccountPrometheus.render({:ok, projection(samples)}, 0.25) + assert body =~ ~S(account_id="synthetic\\quote\"\nline") + assert Enum.count(String.split(body, "\n"), &String.starts_with?(&1, "# TYPE ")) == 36 + + assert Enum.all?( + String.split(body, "\n") |> Enum.filter(&String.starts_with?(&1, "# TYPE ")), + &String.ends_with?(&1, " gauge") + ) + + assert body == AccountPrometheus.render({:ok, projection(Enum.reverse(samples))}, 0.25) + + assert_raise RuntimeError, "duplicate account metric identity", fn -> + AccountPrometheus.render({:ok, projection(samples ++ samples)}, 0.25) + end + end + + test "read failure emits only real health values without fabricated counts or snapshot time" do + body = AccountPrometheus.render({:error, "private-error"}, 0.5) + lines = String.split(body, "\n") |> Enum.filter(&String.starts_with?(&1, "codex_pooler_")) + assert length(lines) == 3 + assert "codex_pooler_account_metrics_collection_success 0" in lines + assert "codex_pooler_account_metrics_collection_duration_seconds 0.5" in lines + refute body =~ "private-error" + end + + defp projection(samples), + do: %{ + complete: true, + as_of: ~U[2026-09-07 06:00:00Z], + accounts: 1, + memberships: 0, + groups: 0, + observations: 0, + samples: samples + } +end diff --git a/test/codex_pooler/metrics/account_snapshot_deadline_test.exs b/test/codex_pooler/metrics/account_snapshot_deadline_test.exs new file mode 100644 index 000000000..c5ef5b610 --- /dev/null +++ b/test/codex_pooler/metrics/account_snapshot_deadline_test.exs @@ -0,0 +1,201 @@ +defmodule CodexPooler.Metrics.AccountSnapshotDeadlineTest do + use ExUnit.Case, async: false + + alias CodexPooler.Metrics.AccountSnapshot + alias CodexPooler.Repo + alias Ecto.Adapters.SQL.Sandbox + + defmodule DeadlineRepo do + use Ecto.Repo, otp_app: :codex_pooler, adapter: Ecto.Adapters.Postgres + end + + defmodule ObservedRepo do + alias CodexPooler.Metrics.AccountSnapshotDeadlineTest.DeadlineRepo + + def transaction(fun, opts) do + [{:config, parent, _delay}] = :ets.lookup(__MODULE__, :config) + {:links, links} = Process.info(self(), :links) + send(parent, {:collector, self(), links, Process.get(:collector_parent_private_state)}) + DeadlineRepo.transaction(fun, opts) + end + + def query!(sql, params, opts) do + [{:config, parent, delay}] = :ets.lookup(__MODULE__, :config) + send(parent, {:collector_query, self(), sql}) + + if delay > 0 and String.contains?(sql, "SELECT id::text") do + [[backend]] = DeadlineRepo.query!("SELECT pg_backend_pid()", [], opts).rows + send(parent, {:collector_backend, self(), backend}) + DeadlineRepo.query!("SELECT pg_sleep($1)", [delay], opts) + end + + DeadlineRepo.query!(sql, params, opts) + end + end + + setup do + :ok = Sandbox.checkout(Repo, sandbox: false) + on_exit(fn -> Sandbox.checkin(Repo) end) + + # A separate one-connection pool against the configured synthetic suite DB + # exercises actual checkout contention, without changing the app's pool. + config = + Keyword.merge(Repo.config(), + pool: DBConnection.ConnectionPool, + pool_size: 1, + queue_target: 10_000, + queue_interval: 10_000, + name: DeadlineRepo + ) + + start_supervised!({DeadlineRepo, config}) + :ets.new(ObservedRepo, [:named_table, :public]) + configure(0) + :ok + end + + @tag timeout: 15_000 + test "fully occupied pool times out near five seconds and its dead collector never queries later" do + holder = hold_connection(:infinity) + Process.put(:collector_parent_private_state, "synthetic-do-not-copy") + started = System.monotonic_time(:millisecond) + body = AccountSnapshot.scrape(fn -> AccountSnapshot.load(ObservedRepo) end) + elapsed = System.monotonic_time(:millisecond) - started + assert elapsed >= 4_500 and elapsed < 5_800 + assert body =~ "codex_pooler_account_metrics_collection_success 0" + refute body =~ "codex_pooler_account_metrics_accounts 0" + assert_receive {:collector, worker, links, nil} + assert_dead(worker, links) + refute_received {:collector_query, ^worker, _} + release(holder) + assert_pool_usable() + refute_receive {:collector_query, ^worker, _}, 100 + end + + @tag timeout: 15_000 + test "slow SQL and mixed checkout-plus-query share the same total deadline" do + for {held_ms, query_seconds} <- [{0, 7}, {2_000, 4}] do + configure(query_seconds) + holder = if held_ms > 0, do: hold_connection(held_ms) + started = System.monotonic_time(:millisecond) + assert {:error, :snapshot_unavailable} = AccountSnapshot.load(ObservedRepo) + elapsed = System.monotonic_time(:millisecond) - started + assert elapsed >= 4_500 and elapsed < 5_800 + assert_receive {:collector, worker, links, nil} + assert_receive {:collector_backend, ^worker, backend} + assert_dead(worker, links) + if holder, do: Task.await(holder) + assert_pool_usable() + assert_backend_idle(backend) + end + end + + test "caller termination cancels queued and executing collectors without detached stream processes" do + for mode <- [:queued, :query] do + configure(if(mode == :query, do: 7, else: 0)) + holder = if mode == :queued, do: hold_connection(:infinity) + {caller, caller_monitor} = spawn_monitor(fn -> AccountSnapshot.load(ObservedRepo) end) + assert_receive {:collector, worker, links, nil} + worker_monitor = Process.monitor(worker) + stream_monitors = Enum.map(links, &{&1, Process.monitor(&1)}) + + backend = + if mode == :query do + assert_receive {:collector_backend, ^worker, backend} + assert_sleeping(backend, 100) + backend + end + + Process.exit(caller, :shutdown) + assert_receive {:DOWN, ^caller_monitor, :process, ^caller, :shutdown} + assert_receive {:DOWN, ^worker_monitor, :process, ^worker, _}, 1_000 + + for {pid, monitor} <- stream_monitors do + assert_receive {:DOWN, ^monitor, :process, ^pid, _}, 1_000 + end + + if holder, do: release(holder) + assert_pool_usable() + if backend, do: assert_backend_idle(backend) + drain_queries(worker) + refute_receive {:collector_query, ^worker, _}, 100 + end + end + + test "successful collection finishes the owned worker and stream before returning" do + assert {:ok, snapshot} = AccountSnapshot.load(ObservedRepo) + assert %DateTime{} = snapshot.as_of + assert_receive {:collector, worker, links, nil} + assert_dead(worker, links) + assert_pool_usable() + end + + defp configure(delay), do: :ets.insert(ObservedRepo, {:config, self(), delay}) + + defp hold_connection(milliseconds) do + parent = self() + + holder = + Task.async(fn -> + DeadlineRepo.transaction( + fn -> + send(parent, :connection_held) + + receive do + :release -> :ok + after + milliseconds -> :ok + end + end, + timeout: 12_000 + ) + end) + + assert_receive :connection_held + holder + end + + defp release(holder) do + send(holder.pid, :release) + assert {:ok, :ok} = Task.await(holder) + end + + defp assert_dead(worker, links) do + for pid <- [worker | links] do + monitor = Process.monitor(pid) + assert_receive {:DOWN, ^monitor, :process, ^pid, _}, 1_000 + end + end + + defp assert_pool_usable do + assert DeadlineRepo.query!("SELECT 1, current_setting('transaction_read_only')", [], + timeout: 1_000 + ).rows == [[1, "off"]] + end + + defp assert_backend_idle(backend) do + assert DeadlineRepo.query!( + "SELECT count(*) FROM pg_stat_activity WHERE pid = $1 AND state = 'active'", + [backend] + ).rows == [[0]] + end + + defp assert_sleeping(_backend, 0), do: flunk("collector query never became active") + + defp assert_sleeping(backend, attempts) do + result = Repo.query!("SELECT state, query FROM pg_stat_activity WHERE pid = $1", [backend]) + + if result.rows != [["active", "SELECT pg_sleep($1)"]] do + Process.sleep(10) + assert_sleeping(backend, attempts - 1) + end + end + + defp drain_queries(worker) do + receive do + {:collector_query, ^worker, _} -> drain_queries(worker) + after + 0 -> :ok + end + end +end diff --git a/test/codex_pooler/metrics/account_snapshot_test.exs b/test/codex_pooler/metrics/account_snapshot_test.exs new file mode 100644 index 000000000..37e91e92c --- /dev/null +++ b/test/codex_pooler/metrics/account_snapshot_test.exs @@ -0,0 +1,456 @@ +defmodule CodexPooler.Metrics.AccountSnapshotTest do + use ExUnit.Case, async: false + + import Ecto.Query + import Phoenix.ConnTest + + alias CodexPooler.Metrics.{AccountProjection, AccountSnapshot, AccountValues} + alias CodexPooler.Repo + alias CodexPooler.Upstreams.Quota.AccountQuotaWindow + alias CodexPooler.Upstreams.Quota.Windows.CycleConfirmation + alias CodexPooler.Upstreams.Schemas.{PoolUpstreamAssignment, UpstreamIdentity} + alias Ecto.Adapters.SQL.Sandbox + + @endpoint CodexPoolerWeb.Endpoint + + defmodule ObservingRepo do + defdelegate transaction(fun, opts), to: CodexPooler.Repo + + def query!(sql, params, opts) do + result = CodexPooler.Repo.query!(sql, params, opts) + [{:observer, observer}] = :ets.lookup(__MODULE__, :observer) + observer.(sql) + result + end + end + + # A real top-level transaction is required to test isolation/read-only mode. + # These committed rows belong only to this isolated test database and are + # removed explicitly; ordinary suite fixtures stay SQL-sandboxed. + setup do + :ok = Sandbox.checkout(Repo, sandbox: false) + on_exit(fn -> Sandbox.checkin(Repo) end) + :ok + end + + test "one bounded repeatable read snapshot omits secret and arbitrary metadata fields" do + with_identity(fn identity -> + _window = + insert_window(identity, %{ + metadata: %{ + "reset_state" => nil, + "rate_limit_allowed" => %{"private-sentinel" => "private-email@example.test"}, + "private-sentinel" => "private-email@example.test" + } + }) + + statements = observe_queries() + assert {:ok, snapshot} = AccountSnapshot.load() + stop_observer(statements) + queries = collected_queries() + assert Enum.count(queries, &String.starts_with?(String.trim(&1), "SELECT")) == 3 + assert Enum.any?(queries, &String.contains?(&1, "REPEATABLE READ, READ ONLY")) + + assert Enum.all?( + queries, + &Regex.match?(~r/\A\s*(SELECT|SET TRANSACTION|begin|commit)/i, &1) + ) + + for forbidden <- + ~w(upstream_secrets oauth access_token refresh_token account_email account_label chatgpt_user_id step_message) do + refute Enum.any?(queries, &String.contains?(&1, forbidden)) + end + + [projected] = Enum.filter(snapshot.identities, &(&1.id == identity.id)) + + assert Enum.sort(Map.keys(projected)) == + ~w(credential_provenance disabled_at id metadata status)a + + assert Map.keys(projected.metadata) |> Enum.sort() == + ~w(credential_epoch quota_account_availability) + + [window] = Enum.filter(snapshot.windows, &(&1.upstream_identity_id == identity.id)) + assert window.metadata == %{"reset_state" => nil, "rate_limit_allowed" => nil} + refute inspect(snapshot) =~ "private-sentinel" + refute inspect(snapshot) =~ "private-email@example.test" + assert AccountProjection.project(snapshot).complete + end) + end + + test "PostgreSQL model normalization and original raw descriptor case drive the digest" do + with_identity(fn identity -> + original = insert_window(identity, %{model: "MODEL", raw_limit_name: "Raw"}) + assert {:ok, snapshot} = AccountSnapshot.load() + first = Enum.find(snapshot.windows, &(&1.id == original.id)) + assert first.index_model == "model" + Repo.update!(Ecto.Changeset.change(original, model: "Model", used_percent: Decimal.new(50))) + assert {:ok, snapshot} = AccountSnapshot.load() + second = Enum.find(snapshot.windows, &(&1.id == original.id)) + assert AccountValues.observation_id(first) == AccountValues.observation_id(second) + alternate = insert_window(identity, %{model: "model", raw_limit_name: "raw"}) + assert {:ok, snapshot} = AccountSnapshot.load() + third = Enum.find(snapshot.windows, &(&1.id == alternate.id)) + refute AccountValues.observation_id(first) == AccountValues.observation_id(third) + end) + end + + test "window sentinel count is per identity and never hides a later account" do + with_identity(fn identity -> + template = insert_window(identity) + + rows = + Enum.map(1..512, fn number -> + template + |> Map.from_struct() + |> Map.drop([:__meta__, :id]) + |> Map.put(:quota_key, "synthetic-#{number}") + end) + + assert {512, nil} = Repo.insert_all(AccountQuotaWindow, rows) + + with_identity(fn later -> + insert_window(later) + assert {:ok, snapshot} = AccountSnapshot.load() + bounded = Enum.filter(snapshot.windows, &(&1.upstream_identity_id == identity.id)) + assert length(bounded) == 513 + assert Enum.all?(bounded, &(&1.row_count == 513)) + assert Enum.any?(snapshot.windows, &(&1.upstream_identity_id == later.id)) + projection = AccountProjection.project(snapshot) + refute projection.complete + assert projection.groups == 1 + end) + end) + end + + test "marker reconstruction preserves validity and rejects extra keys, bad type, stale and future" do + with_identity(fn identity -> + now = DateTime.utc_now() + + window = + insert_window(identity, %{observed_at: now, metadata: %{"reset_state" => "anchored"}}) + + marker = marker(window) + + for {candidate, expected} <- [ + {marker, true}, + {Map.put(marker, "extra", "private-sentinel"), false}, + {[], false}, + {Map.put(marker, "confirmed_at", DateTime.to_iso8601(DateTime.add(now, 600))), false}, + {Map.put( + marker, + "provider_observed_at", + DateTime.to_iso8601(DateTime.add(now, -901)) + ), false} + ] do + original = + Repo.update!( + Ecto.Changeset.change(window, + metadata: %{ + "reset_state" => "anchored", + "__quota_cycle_confirmation_v1" => candidate + } + ) + ) + + assert {:ok, snapshot} = AccountSnapshot.load() + row = Enum.find(snapshot.windows, &(&1.id == window.id)) + + projected = + row + |> Map.drop([:index_model, :index_upstream_model, :row_count]) + |> then(&struct!(AccountQuotaWindow, &1)) + |> Map.update!(:observed_at, &AccountValues.timestamp/1) + |> Map.update!(:reset_at, &AccountValues.timestamp/1) + + assert CycleConfirmation.selector_valid?(original, snapshot.as_of) == expected + assert CycleConfirmation.selector_valid?(projected, snapshot.as_of) == expected + refute inspect(row) =~ "private-sentinel" + end + end) + end + + test "authorized native response appends gauges and account failure keeps operational metrics" do + with_identity(fn identity -> + insert_window(identity) + conn = get(build_conn(), "/metrics") + assert conn.status == 200 + assert conn.resp_body =~ "codex_pooler_account_metrics_collection_success 1" + assert conn.resp_body =~ identity.id + assert conn.resp_body =~ "# TYPE codex_pooler_repo_query_count" + + assert AccountSnapshot.scrape(fn -> {:error, :synthetic_failure} end) =~ + "codex_pooler_account_metrics_collection_success 0" + end) + end + + test "concurrent credential replacement cannot mix epochs inside a read-only snapshot" do + with_identity(fn identity -> + quota = insert_window(identity) + + table = :ets.new(ObservingRepo, [:named_table, :public]) + + observer = fn sql -> + if String.contains?(sql, "SELECT id::text") and + is_nil(Process.get(:account_snapshot_changed)) do + Process.put(:account_snapshot_changed, true) + + assert Repo.query!( + "SELECT current_setting('transaction_isolation'), current_setting('transaction_read_only')" + ).rows == [["repeatable read", "on"]] + + Task.async(fn -> + :ok = Sandbox.checkout(Repo, sandbox: false) + + try do + Repo.update!(Ecto.Changeset.change(identity, metadata: %{"credential_epoch" => 2})) + Repo.update!(Ecto.Changeset.change(quota, used_percent: Decimal.new(80))) + after + Sandbox.checkin(Repo) + end + end) + |> Task.await() + end + end + + :ets.insert(table, {:observer, observer}) + + try do + assert {:ok, snapshot} = AccountSnapshot.load(ObservingRepo) + + assert Enum.find(snapshot.identities, &(&1.id == identity.id)).metadata[ + "credential_epoch" + ] == 1 + + assert Decimal.equal?(Enum.find(snapshot.windows, &(&1.id == quota.id)).used_percent, 6) + assert {:ok, next} = AccountSnapshot.load() + + assert Enum.find(next.identities, &(&1.id == identity.id)).metadata["credential_epoch"] == + 2 + + assert Decimal.equal?(Enum.find(next.windows, &(&1.id == quota.id)).used_percent, 80) + after + :ets.delete(table) + end + end) + end + + test "two collections neither call HTTP, job or account-secret contexts nor mutate rows" do + with_identity(fn identity -> + quota = identity |> insert_window() |> Repo.reload!() + + modules = [ + Req, + Finch, + Oban, + CodexPooler.Upstreams.Secrets, + CodexPooler.Upstreams.OAuth, + CodexPooler.Upstreams.OAuthFlows + ] + + parent = self() + tracer = spawn(fn -> trace_calls(parent) end) + + Enum.each(modules, fn module -> + Code.ensure_loaded!(module) + :erlang.trace_pattern({module, :_, :_}, true, [:local]) + end) + + :erlang.trace(self(), true, [:call, :set_on_spawn, {:tracer, tracer}]) + + try do + for _iteration <- 1..2 do + assert AccountSnapshot.scrape() =~ "codex_pooler_account_metrics_collection_success 1" + end + + :erlang.trace(self(), false, [:all]) + delivered = :erlang.trace_delivered(self()) + assert_receive {:trace_delivered, _, ^delivered} + send(tracer, {:drain, self()}) + assert_receive :trace_drained + refute_received {:forbidden_call, _} + assert Repo.get!(UpstreamIdentity, identity.id) == identity + assert Repo.get!(AccountQuotaWindow, quota.id) == quota + after + :erlang.trace(self(), false, [:all]) + Enum.each(modules, &:erlang.trace_pattern({&1, :_, :_}, false, [:local])) + send(tracer, :stop) + end + end) + end + + defp trace_calls(parent) do + receive do + {:trace, _pid, :call, {module, function, args}} -> + send(parent, {:forbidden_call, {module, function, length(args)}}) + trace_calls(parent) + + {:drain, pid} -> + send(pid, :trace_drained) + trace_calls(parent) + + :stop -> + :ok + end + end + + test "reconciliation SQL projects strings only and never serializes nested JSON payloads" do + with_identity(fn identity -> + pool_id = Ecto.UUID.generate() + + Repo.query!( + "INSERT INTO pools(id,slug,name) VALUES ($1::text::uuid,$2,'synthetic')", + [pool_id, "metrics-scalar-#{pool_id}"] + ) + + Repo.query!( + """ + INSERT INTO pool_upstream_assignments(pool_id,upstream_identity_id,assignment_label,status,health_status,eligibility_status) + VALUES ($1::text::uuid,$2::text::uuid,'synthetic','paused','unknown','eligible') + """, + [pool_id, identity.id] + ) + + try do + for field <- ["status", "finished_at"], + invalid <- [ + %{"message" => "nested-private-sentinel@example.test"}, + ["nested-private-sentinel@example.test"], + true, + 42, + nil + ] do + reconciliation = + Map.put( + %{"status" => "failed", "finished_at" => "2026-09-07T00:00:00Z"}, + field, + invalid + ) + + Repo.query!( + "UPDATE pool_upstream_assignments SET metadata=$1 WHERE upstream_identity_id::text=$2", + [%{"last_reconciliation" => reconciliation}, identity.id] + ) + + assert {:ok, snapshot} = AccountSnapshot.load() + row = Enum.find(snapshot.memberships, &(&1.upstream_identity_id == identity.id)) + + column = + if field == "status", do: :reconciliation_status, else: :reconciliation_finished_at + + assert Map.fetch!(row, column) == nil + assert row.reconciliation_status == reconciliation["status"] or field == "status" + + assert row.reconciliation_finished_at == reconciliation["finished_at"] or + field == "finished_at" + + refute inspect(snapshot) =~ "nested-private-sentinel" + end + after + Repo.delete_all( + from a in PoolUpstreamAssignment, where: a.upstream_identity_id == ^identity.id + ) + + Repo.query!("DELETE FROM pools WHERE id::text=$1", [pool_id]) + end + end) + end + + defp with_identity(fun) do + now = DateTime.utc_now() + + identity = + Repo.insert!(%UpstreamIdentity{ + status: "paused", + account_label: "private-sentinel", + account_email: "private-email@example.test", + onboarding_method: "import", + headers_profile_version: 1, + created_at: now, + updated_at: now, + credential_provenance: "codex_chatgpt_oauth", + metadata: %{"credential_epoch" => 1, "private-sentinel" => "private-email@example.test"} + }) + + try do + fun.(identity) + after + Repo.delete_all(from w in AccountQuotaWindow, where: w.upstream_identity_id == ^identity.id) + + Repo.delete_all( + from a in PoolUpstreamAssignment, where: a.upstream_identity_id == ^identity.id + ) + + Repo.delete!(identity) + end + end + + defp insert_window(identity, attrs \\ %{}) do + now = DateTime.utc_now() + + fields = %{ + upstream_identity_id: identity.id, + quota_key: "account", + quota_scope: "account", + quota_family: "account", + window_kind: "secondary", + window_minutes: 10_080, + source: "codex_usage_api", + source_precision: "observed", + freshness_state: "fresh", + observed_at: now, + last_sync_at: now, + created_at: now, + updated_at: now, + reset_at: DateTime.add(now, 86_400), + used_percent: Decimal.new(6), + merge_precedence: 60, + metadata: %{} + } + + Repo.insert!(struct!(AccountQuotaWindow, Map.merge(fields, attrs))) + end + + defp marker(window) do + %{ + "version" => 1, + "scope" => window.quota_scope, + "family" => window.quota_family, + "key" => window.quota_key, + "kind" => window.window_kind, + "minutes" => window.window_minutes, + "model" => window.model, + "upstream_model" => window.upstream_model, + "reset_at" => DateTime.to_iso8601(window.reset_at), + "provider_observed_at" => DateTime.to_iso8601(window.observed_at), + "confirmed_at" => DateTime.to_iso8601(window.observed_at), + "source_class" => "provider_usage" + } + end + + defp observe_queries do + ref = make_ref() + + :telemetry.attach( + ref, + [:codex_pooler, :repo, :query], + fn _, _, metadata, pid -> + if self() == pid or pid in Process.get(:"$callers", []), + do: send(pid, {:query, metadata.query}) + end, + self() + ) + + on_exit(fn -> :telemetry.detach(ref) end) + ref + end + + defp stop_observer(ref), do: :telemetry.detach(ref) + + defp collected_queries do + receive do + {:query, sql} -> [sql | collected_queries()] + after + 0 -> [] + end + end +end diff --git a/test/codex_pooler_web/controllers/operations/metrics_controller_test.exs b/test/codex_pooler_web/controllers/operations/metrics_controller_test.exs index af128ffdf..f020fba7e 100644 --- a/test/codex_pooler_web/controllers/operations/metrics_controller_test.exs +++ b/test/codex_pooler_web/controllers/operations/metrics_controller_test.exs @@ -78,15 +78,18 @@ defmodule CodexPoolerWeb.Operations.MetricsControllerTest do test "rejects metrics access when configured bearer token is missing", %{conn: conn} do configure_metrics_token!("metrics-secret") + observe_account_collection() conn = get(conn, ~p"/metrics") assert conn.status == 401 assert json_response(conn, 401)["error"]["code"] == "metrics_unauthorized" + refute_received :account_collection end test "rejects metrics access when configured bearer token is wrong", %{conn: conn} do configure_metrics_token!("metrics-secret") + observe_account_collection() conn = conn @@ -95,6 +98,7 @@ defmodule CodexPoolerWeb.Operations.MetricsControllerTest do assert conn.status == 401 assert json_response(conn, 401)["error"]["code"] == "metrics_unauthorized" + refute_received :account_collection end test "allows metrics access with the configured bearer token", %{conn: conn} do @@ -666,17 +670,47 @@ defmodule CodexPoolerWeb.Operations.MetricsControllerTest do end test "fails closed when metrics settings are unavailable", %{conn: conn} do + observe_account_collection() Application.put_env(:codex_pooler, InstanceSettings, repo: FailingRepo) InstanceSettings.reset_cache_for_test() {conn, log} = capture_result_and_log(fn -> get(conn, ~p"/metrics") end) assert log =~ "instance settings db load failed warm_cache=false" + refute_received :account_collection assert conn.status == 401 assert json_response(conn, 401)["error"]["code"] == "metrics_unauthorized" assert json_response(conn, 401)["error"]["message"] == "metrics bearer token is unavailable" end + test "a failed account transaction preserves operational exposition", %{conn: conn} do + # ConnCase has already queried its sandbox transaction. PostgreSQL refuses + # changing its isolation level, providing a real collector failure fixture. + conn = get(conn, ~p"/metrics") + assert conn.status == 200 + assert conn.resp_body =~ "codex_pooler_account_metrics_collection_success 0" + assert conn.resp_body =~ "# TYPE codex_pooler_repo_query_count" + refute conn.resp_body =~ "SET TRANSACTION" + refute conn.resp_body =~ "snapshot_unavailable" + end + + defp observe_account_collection do + ref = make_ref() + + :telemetry.attach( + ref, + [:codex_pooler, :repo, :query], + fn _, _, metadata, pid -> + if (self() == pid or pid in Process.get(:"$callers", [])) and + String.contains?(metadata.query, "SET TRANSACTION"), + do: send(pid, :account_collection) + end, + self() + ) + + on_exit(fn -> :telemetry.detach(ref) end) + end + defp capture_result_and_log(fun) do ref = make_ref()