feat(filter): bounded live-signal store on a shared exposition tokenizer

[hexfusion]

Summary

Add two workspace crates that carry the portable core of Grid live load signals: common, a plane-neutral Prometheus exposition tokenizer with the grid target-label constants, and filter, the data-plane consumer that keeps a bounded per-series window and scores routing candidates on current load. This is Part 1 of moving the grid-only live-signals code from praxis-proxy/ai#1175 into this repo. It lands the store and the hardened parser only; the live poller and pinned praxis client follow in Part 2.

Motivation

Cross-site routing needs to score a candidate on its current load, not a stale or forged value. The exposition text a provider influences crosses a trust boundary when the operator stamps grid_site/grid_provider and republishes it, so the consumer parser must be fail-closed: a quoted comma, an escaped quote, an escaped separator, a future timestamp, or a NaN/Inf value must never forge a target label, inject the store key, or wedge the series. Landing the store and parser first, without the poller, keeps that hardened core reviewable on its own and lets Part 2 add the transport against a settled data model.

What changed

• common: a std-only, plane-neutral crate holding the quote- and escape-aware exposition tokenizer and the grid_site/grid_provider label constants. Both planes may depend on it; a plane-neutral leaf is not a plane-separation violation. The operator producer keeps its own parser for now; migrating it onto common is a Part-2 followup.
• filter: the data-plane consumer. LoadStore keeps a bounded per-series window keyed by site/cluster; samples key on the operator's observation time, so a republished cache value never reads as new. window_worst returns the worst reading in the window so a drained burst stays penalised until it ages out. Extraction is fail-closed: a control char or / in a target value, a duplicated target label, a missing timestamp, or a non-finite value drops the line.
• Bounds against a hostile or misconfigured endpoint: MAX_PROVIDERS (4096), MAX_METRICS_PER_PROVIDER (64), and MAX_SAMPLES_PER_SERIES (1024), each tested. A future timestamp is dropped against the operator's own clock (its Date), tolerating MAX_CLOCK_SKEW_MS of skew.
• Root Cargo.toml: register the two members; pin dashmap to 6.2.1 to match praxis-ai.

Both are plain libs and nothing built depends on them yet, so no image or Containerfile changes. Excludes SPIRE/SPIFFE. No poller, no praxis client, no operator change.

[praxis-bot]

Thorough security model -- fail-closed extraction, slash/control-char rejection, duplicate-label detection, future-timestamp gating, and per-provider/per-series caps are all well-tested. One medium finding below.

[praxis-bot]

Medium -- count cap bypassed on window-conversion failure. The early return here skips the MAX_SAMPLES_PER_SERIES enforcement that follows on line 74. When window.as_millis() exceeds i64::MAX (e.g. a caller passes Duration::MAX), no eviction runs at all -- neither window-based nor count-based -- so the series can grow without bound.

The doc on MAX_SAMPLES_PER_SERIES (line 30-32) promises "Retention past the cap drops the oldest first," but this path violates that invariant. The count cap exists as defense-in-depth and should apply unconditionally.

Suggested restructure so the count cap always runs:

fn push(&mut self, sample: Sample, window: Duration) {
    if self.samples.last().is_some_and(|last| sample.at_ms <= last.at_ms) {
        return;
    }
    self.samples.push(sample);
    let keep_from = i64::try_from(window.as_millis())
        .map(|window_ms| {
            let cutoff = sample.at_ms.saturating_sub(window_ms);
            self.samples.partition_point(|held| held.at_ms < cutoff)
        })
        .unwrap_or(0);
    let over_cap = self.samples.len().saturating_sub(MAX_SAMPLES_PER_SERIES);
    let drop_to = keep_from.max(over_cap);
    if drop_to > 0 {
        self.samples.drain(..drop_to);
    }
}

[shaneutt]

The current caps allow multi-GB allocations, we should lower the per-series cap to ~64–128, trim before pushing, and prune series idle beyond the window then document the resulting worst-case memory.

[shaneutt]

The read path uses the gateway’s now_ms, but samples use the operator’s Date clock. As a result, valid samples that are slightly ahead of the gateway are dropped, and larger skew can make window_worst return None despite ingest accepting the data.

Apply the same tolerance as ingest:

if sample.at_ms < cutoff
    || sample.at_ms > now_ms.saturating_add(MAX_CLOCK_SKEW_MS)

We'll need tests for samples at the cutoff, within and beyond the tolerance, aging out of the window, and unknown keys/metrics returning None.

[shaneutt]

testing gapsd: Add boundary tests at reference_ms + MAX_CLOCK_SKEW_MS and +1 to catch > vs. >=. Also test partial grid_site/grid_provider labels after tokenization.

[shaneutt]

MAX_PROVIDERS is a global, never-released pool, so one peer can silently consume all 4096 slots with distinct grid_provider values and prevent every later provider from routing signals until restart. Relayed data makes this practical: peers control the provider label, and 4096 entries fit within the request limit.

Consider enforcing a per-site cap, evicting keys whose newest sample is older than the window before rejecting new ones, and counting or logging dropped lines (cap hits, parse failures, and future timestamps) so lockouts are diagnosable.

[shaneutt]

cap the metric names and grid_site/grid_provider values (e.g. 128–256 bytes) in parse_sample before building keys and document the resulting byte bound alongside the count caps.

[shaneutt]

The series key drops all labels except grid_site, grid_provider, and the metric name. If a provider emits multiple labeled series for one metric, the first line wins by print order; later lines share its timestamp and are discarded, potentially reporting 0 for a nonzero queue.

Fail closed within each ingest: track (key, metric, at_ms) pairs and, when another line differs only by discarded labels, either drop the metric for that ingest or retain the worst value. Add a test covering two such series.