diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7fafc44..6a5d754 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -113,11 +113,9 @@ jobs: uses: actions-rust-lang/audit@v1 with: # RUSTSEC-2023-0071: rsa crate (from sqlx-mysql, not used - we use postgres only) - # RUSTSEC-2025-0111: tokio-tar (from testcontainers, dev dependency only) - # RUSTSEC-2026-0066: astral-tokio-tar (renamed tokio-tar, same issue, testcontainers dev dep) # RUSTSEC-2025-0134: rustls-pemfile unmaintained (transitive dep, awaiting upstream fix) # RUSTSEC-2026-0097: rand unsoundness with custom logger (no fix available; we don't use custom logger) - ignore: RUSTSEC-2023-0071,RUSTSEC-2025-0111,RUSTSEC-2026-0066,RUSTSEC-2025-0134,RUSTSEC-2026-0097 + ignore: RUSTSEC-2023-0071,RUSTSEC-2025-0134,RUSTSEC-2026-0097 denyWarnings: false createIssues: false diff --git a/AGENTS.md b/AGENTS.md index b3266cc..2de7c1b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -167,6 +167,7 @@ See `docs/adr/` for all 19 Architecture Decision Records. - `POST /v1/signals/webhook/{name}` - Generic webhook adapter (configured in `correlation.yaml`; JSONPath field mapping; HMAC/bearer/none auth) - `POST /v1/signals/corroborator` - Corroborating signal adapter (ADR 021). Sources configured with `mode: corroborating` post dimension-tagged signals that strengthen open signal groups without ever triggering mitigations on their own. Declared `match_dimensions` are authoritative: only declared dimensions are consulted during matching. Rejected with 400 if the source is unknown, `mode: primary`, or no declared dimension is populated. Correlation engine must be enabled. - `GET /v1/signals/corroborator/activity?minutes=N` - Per-source corroborator activity summary aggregated across the live cache and attached signal-group rows. Used by the Signals dashboard so `mode: corroborating` sources surface realistic `last_seen`/`count` instead of always reading as "never seen". +- `GET /v1/signals/corroborator/cache?source=X&limit=N` - **Admin only.** Lists corroborating signals currently cached and unattached + unexpired (i.e. waiting for a matching primary event to arrive within `window_seconds`). Used by the Correlation dashboard's Cache tab; pair with the `prefixd_corroborator_cache_size{source}` gauge for alerting on runaway caches. - `GET /v1/config/correlation` - Correlation config (admin, secrets redacted) - `PUT /v1/config/correlation` - Update correlation config (admin only, writes YAML + hot-reload) diff --git a/CHANGELOG.md b/CHANGELOG.md index f65a116..fa7bccd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **Corroborating signals v2 (PR B)** — Follow-up to ADR 021's initial ship that addresses the four review-deferred items as a coordinated set: + - **Playbook-aware late finalization.** Migration `012_signal_groups_playbook.sql` adds nullable `signal_groups.playbook_name`, populated by the daemon on the next primary event for each group via `COALESCE`. The corroborator-side aggregate recompute now re-resolves the playbook by name from live state and is allowed to flip `corroboration_met=true` using the override min_sources/threshold. Conservative fallback is preserved: a NULL or stale `playbook_name` still keeps the v0.16.0 no-flip behavior (the next primary event picks up the flag). + - **Per-source attribution on `prefixd_corroborator_expired_total`.** The counter regains its `{source}` label set, with `delete_expired_corroborating_signals` collecting attribution in the same `DELETE … RETURNING` query that performs the delete. **Operator note:** this is a label change. PromQL queries written against the v0.16.0 unlabelled counter must add a `sum()` to recover the previous shape. + - **New gauge `prefixd_corroborator_cache_size{source}`** updated by the reconcile loop after each sweep. Operators can alert on caches growing without bound (e.g. a source posting heavily while no matching primary event ever lands). Stale labels are explicitly zeroed when a source's cache drains between ticks. + - **Cached-corroborators admin endpoint and dashboard panel.** `GET /v1/signals/corroborator/cache` (admin-only) returns `{ now, total, by_source[], signals[] }` filtered to unattached + unexpired rows, with optional `?source=` and `?limit=` (clamped to 1..1000). New "Cache" tab on the Correlation page renders per-source counts plus a dense table of cached signals with relative ingested/expires timestamps and dimension chips. Backed by a new `useCachedCorroborators` SWR hook (30s refresh). + - **`CorroboratorResponse.cached` removed** in favor of the existing `status ∈ {attached, cached}` discriminator. The boolean was always `true` and added no information; status fully describes the outcome. Coordinated minor breaking change — bump to v0.17.0 — since the endpoint is new in this release line. + ## [0.16.0] - 2026-04-19 ### Added diff --git a/Cargo.lock b/Cargo.lock index 975d2ee..199e399 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -77,7 +77,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -88,7 +88,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -120,9 +120,9 @@ dependencies = [ [[package]] name = "astral-tokio-tar" -version = "0.5.6" +version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ec179a06c1769b1e42e1e2cbe74c7dcdb3d6383c838454d063eaac5bbb7ebbe5" +checksum = "4ce73b17c62717c4b6a9af10b43e87c578b0cac27e00666d48304d3b7d2c0693" dependencies = [ "filetime", "futures-core", @@ -212,9 +212,9 @@ dependencies = [ [[package]] name = "axum" -version = "0.8.8" +version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b52af3cb4058c895d37317bb27508dccc8e5f2d39454016b297bf4a400597b8" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", "axum-macros", @@ -287,9 +287,9 @@ dependencies = [ [[package]] name = "axum-macros" -version = "0.5.0" +version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "604fde5e028fea851ce1d8570bbdc034bec850d157f7569d10f347d06808c05c" +checksum = "7aa268c23bfbbd2c4363b9cd302a4f504fb2a9dfe7e3451d66f35dd392e20aca" dependencies = [ "proc-macro2", "quote", @@ -380,9 +380,9 @@ dependencies = [ [[package]] name = "bollard" -version = "0.19.4" +version = "0.20.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "87a52479c9237eb04047ddb94788c41ca0d26eaff8b697ecfbb4c32f7fdc3b1b" +checksum = "ee04c4c84f1f811b017f2fbb7dd8815c976e7ca98593de9c1e2afad0f636bff4" dependencies = [ "async-stream", "base64 0.22.1", @@ -390,7 +390,6 @@ dependencies = [ "bollard-buildkit-proto", "bollard-stubs", "bytes", - "chrono", "futures-core", "futures-util", "hex", @@ -408,14 +407,13 @@ dependencies = [ "rand 0.9.2", "rustls", "rustls-native-certs", - "rustls-pemfile", "rustls-pki-types", "serde", "serde_derive", "serde_json", - "serde_repr", "serde_urlencoded", "thiserror 2.0.18", + "time", "tokio", "tokio-stream", "tokio-util", @@ -440,19 +438,18 @@ dependencies = [ [[package]] name = "bollard-stubs" -version = "1.49.1-rc.28.4.0" +version = "1.52.1-rc.29.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5731fe885755e92beff1950774068e0cae67ea6ec7587381536fca84f1779623" +checksum = "0f0a8ca8799131c1837d1282c3f81f31e76ceb0ce426e04a7fe1ccee3287c066" dependencies = [ "base64 0.22.1", "bollard-buildkit-proto", "bytes", - "chrono", "prost", "serde", "serde_json", "serde_repr", - "serde_with", + "time", ] [[package]] @@ -942,7 +939,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -985,12 +982,12 @@ checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" [[package]] name = "ferroid" -version = "0.8.9" +version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bb330bbd4cb7a5b9f559427f06f98a4f853a137c8298f3bd3f8ca57663e21986" +checksum = "ee93edf3c501f0035bbeffeccfed0b79e14c311f12195ec0e661e114a0f60da4" dependencies = [ "portable-atomic", - "rand 0.9.2", + "rand 0.10.0", "web-time", ] @@ -1888,7 +1885,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -2790,7 +2787,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2858,7 +2855,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2869,9 +2866,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" [[package]] name = "rustls-webpki" -version = "0.103.12" +version = "0.103.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8279bb85272c9f10811ae6a6c547ff594d6a7f3c6c6b02ee9726d1d0dcfcdd06" +checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" dependencies = [ "aws-lc-rs", "ring", @@ -3230,7 +3227,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -3566,14 +3563,14 @@ dependencies = [ "getrandom 0.4.1", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] name = "testcontainers" -version = "0.26.3" +version = "0.27.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a81ec0158db5fbb9831e09d1813fe5ea9023a2b5e6e8e0a5fe67e2a820733629" +checksum = "bfd5785b5483672915ed5fe3cddf9f546802779fc1eceff0a6fb7321fac81c1e" dependencies = [ "astral-tokio-tar", "async-trait", @@ -3584,6 +3581,7 @@ dependencies = [ "etcetera 0.11.0", "ferroid", "futures", + "http", "itertools 0.14.0", "log", "memchr", @@ -3601,9 +3599,9 @@ dependencies = [ [[package]] name = "testcontainers-modules" -version = "0.14.0" +version = "0.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e75e78ff453128a2c7da9a5d5a3325ea34ea214d4bf51eab3417de23a4e5147" +checksum = "e5985fde5befe4ffa77a052e035e16c2da86e8bae301baa9f9904ad3c494d357" dependencies = [ "testcontainers", ] @@ -3725,9 +3723,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.51.0" +version = "1.52.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2bd1c4c0fc4a7ab90fc15ef6daaa3ec3b893f004f915f2392557ed23237820cd" +checksum = "b67dee974fe86fd92cc45b7a95fdd2f99a36a6d7b0d431a231178d3d670bbcc6" dependencies = [ "bytes", "libc", @@ -3785,9 +3783,9 @@ dependencies = [ [[package]] name = "tokio-tungstenite" -version = "0.28.0" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857" +checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c" dependencies = [ "futures-util", "log", @@ -4091,9 +4089,9 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "tungstenite" -version = "0.28.0" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" +checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" dependencies = [ "bytes", "data-encoding", @@ -4103,7 +4101,6 @@ dependencies = [ "rand 0.9.2", "sha1", "thiserror 2.0.18", - "utf-8", ] [[package]] @@ -4261,9 +4258,9 @@ dependencies = [ [[package]] name = "uuid" -version = "1.22.0" +version = "1.23.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a68d3c8f01c0cfa54a75291d83601161799e4a89a39e0929f4b0354d88757a37" +checksum = "ddd74a9687298c6858e9b88ec8935ec45d22e8fd5e6394fa1bd4e99a87789c76" dependencies = [ "getrandom 0.4.1", "js-sys", @@ -4510,7 +4507,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index d44faf6..b01f074 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -84,8 +84,8 @@ prost-build = "0.14" tokio-test = "0.4" tempfile = "3" criterion = { version = "0.8", features = ["async_tokio"] } -testcontainers = "0.26" -testcontainers-modules = { version = "0.14", features = ["postgres"] } +testcontainers = "0.27" +testcontainers-modules = { version = "0.15", features = ["postgres"] } proptest = "1" [[bench]] diff --git a/ROADMAP.md b/ROADMAP.md index 8300b6b..5817610 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -289,12 +289,12 @@ Example: FastNetMon says UDP flood at 0.6 confidence + router CPU spiking + host - [x] Corroboration requirements ("require 2+ sources") - [x] Correlation explainability (`why` details in API/UI for each mitigation decision) - [x] Corroborating-only signals v1 — coarse telemetry (router CPU, PoP interface, per-customer NetFlow) can strengthen groups without ever triggering mitigations on its own (ADR 021, PR #109) -- [ ] **Corroborating signals v2 (PR B)** — follow-ups from the ADR 021 review: - - [ ] Playbook-override-aware corroborator finalization: let a late corroborator promote `corroboration_met` → `true` on its own path, using the resolved override from the group's most recent primary event. Eliminates the current dependency on "another primary event fires within the window" for late corroborators to take effect. - - [ ] Per-source attribution on `prefixd_corroborator_expired_total`: select expiring rows grouped by source before delete so the metric can be labelled without a full rewrite of the sweep path. - - [ ] API cleanup: drop the redundant `cached: true` field from `CorroboratorResponse`; `status ∈ {attached, cached}` already fully describes the outcome. Coordinate with a minor API version bump since the endpoint is new in this release. - - [ ] Dashboard "cached corroborators" panel: small widget on the Correlation dashboard showing live count of unattached-but-unexpired corroborators per source, sourced from a new `/v1/signals/corroborator/cache` listing endpoint (admin-only). - - [ ] Gauge metric `prefixd_corroborator_cache_size{source}` updated by the reconcile loop for Prometheus alerting on runaway caches. +- [x] **Corroborating signals v2 (PR B)** — follow-ups from the ADR 021 review (shipped in v0.17.0): + - [x] Playbook-override-aware corroborator finalization: late corroborators now promote `corroboration_met` → `true` on their own path using the override resolved from `signal_groups.playbook_name` (migration 012). Conservative fallback preserved when the stored playbook is missing from live config. + - [x] Per-source attribution on `prefixd_corroborator_expired_total{source}`: counter regains `&["source"]` label set; sweep collects attribution via `DELETE … RETURNING` grouped by source. + - [x] API cleanup: dropped the redundant `cached: true` field from `CorroboratorResponse`; `status ∈ {attached, cached}` is the discriminator. + - [x] Dashboard "cached corroborators" panel: new Cache tab on the Correlation page backed by `GET /v1/signals/corroborator/cache` (admin-only) showing per-source counts and a dense signal table. + - [x] Gauge metric `prefixd_corroborator_cache_size{source}` updated by the reconcile loop after each sweep; stale labels zeroed on drain. - [ ] Router telemetry adapter (JTI, gNMI) as the first production consumer of corroborator mode (already listed under Signal Adapters) - [ ] Replay mode for tuning (simulate historical incidents without announcing FlowSpec rules) diff --git a/docs/adr/021-corroborating-signals.md b/docs/adr/021-corroborating-signals.md index 40dd6dd..90d1ee6 100644 --- a/docs/adr/021-corroborating-signals.md +++ b/docs/adr/021-corroborating-signals.md @@ -230,17 +230,59 @@ initial implementation. All were fixed before merge. Notable ones: ## Known limits / deferred to PR B -- A corroborator that lands late in the window and pushes a group over - its threshold does not immediately fire the mitigation; it waits for - the next primary-path event (if any) to re-evaluate. This is a - product choice, not a correctness gap — implementing playbook- - override-aware finalization on the corroborator path is scheduled for - PR B (see ROADMAP). -- `prefixd_corroborator_expired_total` has no source label. PR B will - restore per-source attribution by collecting rows before delete. -- `CorroboratorResponse.cached` is always `true`; the field is - redundant given `status ∈ {attached, cached}` and is flagged for - removal in PR B. +(All five PR B items shipped in v0.17.0 — see CHANGELOG and the +"PR B addenda" section below for design notes on each.) + +## PR B addenda (v0.17.0) + +- **Late corroborator finalization is now playbook-aware.** Migration + `012_signal_groups_playbook.sql` adds nullable + `signal_groups.playbook_name`. The daemon writes it on group create + and `COALESCE`-backfills it on the next primary event for any + pre-upgrade group. The corroborator-side aggregate recompute + re-resolves the playbook by name from live state; if the playbook + exists, it applies the override min_sources/threshold and is now + allowed to flip `corroboration_met` from false → true. If the stored + name is NULL or no longer resolves (admin removed the playbook), the + conservative v0.16.0 behavior is preserved: aggregates update but the + flag is not flipped — the next primary event picks it up. This + decouples late corroborator finalization from "another primary event + must fire within the window" without weakening the + primary-required-once invariant. + + Mitigation actuation deliberately stays single-sourced through + `handle_ban`; corroborator-path recompute updates state but does not + fire FlowSpec. The next primary event reads the now-true flag and + triggers normally. This keeps the actuation surface narrow and + preserves the existing test/audit surface for mitigations. + +- **Per-source attribution on the expired counter.** + `delete_expired_corroborating_signals` now uses a single + `DELETE … RETURNING source` with `GROUP BY source` so attribution is + collected in the same query that performs the delete (no + read-then-delete race). `CORROBORATOR_EXPIRED_TOTAL` regains its + `&["source"]` label set. Operator note: this is a label change. + PromQL queries written against the v0.16.0 unlabelled counter must + add `sum()` to recover the previous shape. + +- **`prefixd_corroborator_cache_size{source}` gauge.** Updated by the + reconcile loop after each sweep using the new + `count_cached_corroborators_by_source(now)` repository method. The + scheduler keeps an in-process `last_cache_sources` set so labels + whose source drained between ticks are explicitly zeroed (Prometheus + would otherwise keep stale non-zero values forever). + +- **`/v1/signals/corroborator/cache` admin endpoint.** Returns + `{ now, total, by_source[], signals[] }` filtered to + unattached + unexpired rows. Optional `?source=` and `?limit=` + (clamped to 1..1000). Dashboard-side, the new Cache tab on the + Correlation page renders per-source badges and a dense table of + cached signals. + +- **`CorroboratorResponse.cached` removed.** Always-true booleans add + no information. `status ∈ {attached, cached}` is the canonical + discriminator. v0.17.0 minor breaking change since the endpoint is + new in this release line. ## References diff --git a/docs/api.md b/docs/api.md index 26e2806..ccb63c1 100644 --- a/docs/api.md +++ b/docs/api.md @@ -946,8 +946,7 @@ Content-Type: application/json { "signal_id": "a4f1b2c3-…", "status": "attached", - "attached_group_ids": ["e2b9…-1f3c"], - "cached": true + "attached_group_ids": ["e2b9…-1f3c"] } ``` @@ -957,10 +956,11 @@ Content-Type: application/json primary event arrival. - `attached_group_ids` — UUIDs of signal groups this signal contributed to. -- `cached` — currently always `true` (signals are retained in the cache - after attach for late fan-out). This field is flagged for removal in - a follow-up release in favor of `status` alone; new integrations - should rely on `status` / `attached_group_ids` instead. + +> **v0.17.0 breaking change:** the always-true `cached` field on this +> response was dropped. `status ∈ {attached, cached}` is the canonical +> discriminator. Update integrations that read `cached` to read +> `status === "cached"` instead. **Error responses:** @@ -1008,6 +1008,61 @@ group and kept its live cache row for late fan-out; the intent is --- +### List Cached Corroborators (admin) + +```http +GET /v1/signals/corroborator/cache?source=router-cpu&limit=200 +``` + +**Admin-only.** Lists corroborating signals currently in the cache that +are unattached and unexpired — i.e., signals that posted before any +matching primary event landed and are still waiting inside the +`window_seconds` TTL. Use this to debug a `mode: corroborating` source +that ingests heavily but never seems to attach. + +**Query parameters:** + +- `limit` *(optional, default 100, range 1–1000)* — Page size. +- `source` *(optional)* — Filter by signal source name. + +**Response:** + +```json +{ + "now": "2026-04-29T18:23:00Z", + "total": 17, + "by_source": [ + {"source": "router-cpu", "count": 11}, + {"source": "pop-utilization", "count": 6} + ], + "signals": [ + { + "signal_id": "...", + "source": "router-cpu", + "vector": "udp_flood", + "customer_id": null, + "pop": "iad1", + "service_id": null, + "interface": null, + "confidence": 0.7, + "weight": 0.6, + "ingested_at": "2026-04-29T18:22:55Z", + "expires_at": "2026-04-29T18:27:55Z", + "raw_details": {...}, + "attached_group_ids": [] + } + ] +} +``` + +- `total` and `by_source[]` summarize **unattached, unexpired** rows + globally (not just the page). +- `signals[]` is paginated by `limit` and ordered by `ingested_at` desc. +- Pair this with the new `prefixd_corroborator_cache_size{source}` + gauge for alerting on caches growing without bound. + +--- + ## Safelist ### List Safelist diff --git a/docs/detectors/corroborating-signals.md b/docs/detectors/corroborating-signals.md index ea5ae83..78dc5b9 100644 --- a/docs/detectors/corroborating-signals.md +++ b/docs/detectors/corroborating-signals.md @@ -82,16 +82,18 @@ Response: { "signal_id": "…", "status": "attached", - "attached_group_ids": ["…"], - "cached": true + "attached_group_ids": ["…"] } ``` -- `attached` — at least one open signal group matched and was +- `status: attached` — at least one open signal group matched and was strengthened. -- `cached` — no matching group yet; the signal is held for up to - `correlation.window_seconds` and drained if a matching primary event - arrives in that window. +- `status: cached` — no matching group yet; the signal is held for up + to `correlation.window_seconds` and drained if a matching primary + event arrives in that window. + +> The v0.16.0 `cached: true` field on this response was always-true +> and was removed in v0.17.0. Branch on `status` instead. ## 3. From the CLI @@ -166,18 +168,27 @@ def post_cpu_alert(pop: str, utilization: float): ## 7. Known limits -See the ADR 021 "Known limits / deferred to PR B" section for the -authoritative list. Most operator-visible one right now: - -- A corroborator that arrives *after* a primary event and pushes - aggregates past the threshold does **not** immediately trigger the - mitigation. Finalization happens on the next primary-ingest path. In - practice this means: if two primary events fire inside the window, - the corroborator's contribution is picked up. If only one primary - fires and a corroborator arrives later, the mitigation waits on - another primary signal (or on the cache drain when the next primary - for the same dimensions lands). Tracked as PR B work item - *Playbook-override-aware corroborator finalization*. +All five PR B follow-ups from the ADR 021 review shipped in v0.17.0. +A few residual operator notes: + +- **Mitigation still actuates from the primary path.** A late + corroborator can flip `corroboration_met` to true on its own (using + the playbook override stored on the group), but the FlowSpec + announcement is still triggered by the next primary event for that + group. This keeps actuation single-sourced. In practice: the flag + promotion is observable on `GET /v1/signal-groups/{id}` and on the + dashboard immediately; the mitigation lands when a matching primary + ingest happens (which, given the corroborator just confirmed the + group, is usually within seconds). +- **Stale playbook names are conservative.** If an operator removed a + playbook between group creation and a late corroborator landing, + recompute will keep the v0.16.0 no-flip behavior. Re-add the + playbook (or wait for the next primary event with a fresh playbook + name) to recover. +- **Cache-size gauge is per-source.** Alert on + `prefixd_corroborator_cache_size{source="..."}` rather than a sum + to detect a single misconfigured source filling the cache without + affecting the global view. ## Troubleshooting diff --git a/frontend/app/(dashboard)/correlation/page.tsx b/frontend/app/(dashboard)/correlation/page.tsx index 92f9dae..d566a0c 100644 --- a/frontend/app/(dashboard)/correlation/page.tsx +++ b/frontend/app/(dashboard)/correlation/page.tsx @@ -4,11 +4,12 @@ import { useState } from "react" import { DashboardLayout } from "@/components/dashboard/dashboard-layout" import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs" import { Badge } from "@/components/ui/badge" -import { Radio, Layers, Settings } from "lucide-react" +import { Radio, Layers, Settings, Database } from "lucide-react" import { useOpenSignalGroupCount } from "@/hooks/use-api" import { SignalsTab } from "@/components/dashboard/correlation/signals-tab" import { GroupsTab } from "@/components/dashboard/correlation/groups-tab" import { ConfigTab } from "@/components/dashboard/correlation/config-tab" +import { CacheTab } from "@/components/dashboard/correlation/cache-tab" export default function CorrelationPage() { const [activeTab, setActiveTab] = useState("signals") @@ -40,6 +41,10 @@ export default function CorrelationPage() { )} + + + Cache + Config @@ -54,6 +59,10 @@ export default function CorrelationPage() { + + + + diff --git a/frontend/components/dashboard/correlation/cache-tab.tsx b/frontend/components/dashboard/correlation/cache-tab.tsx new file mode 100644 index 0000000..4878579 --- /dev/null +++ b/frontend/components/dashboard/correlation/cache-tab.tsx @@ -0,0 +1,148 @@ +"use client" + +import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card" +import { Badge } from "@/components/ui/badge" +import { Skeleton } from "@/components/ui/skeleton" +import { Database, AlertCircle } from "lucide-react" +import { useCachedCorroborators } from "@/hooks/use-api" +import { formatDistanceToNow } from "date-fns" + +function dimsLabel(s: { + customer_id: string | null + pop: string | null + service_id: string | null + interface: string | null +}): string { + const parts: string[] = [] + if (s.customer_id) parts.push(`customer=${s.customer_id}`) + if (s.pop) parts.push(`pop=${s.pop}`) + if (s.service_id) parts.push(`service=${s.service_id}`) + if (s.interface) parts.push(`iface=${s.interface}`) + return parts.length > 0 ? parts.join(" ") : "—" +} + +export function CacheTab() { + const { data, error, isLoading } = useCachedCorroborators({ limit: 200 }) + + if (isLoading) { + return ( +
+ + +
+ ) + } + + if (error) { + return ( + + + + Failed to load corroborator cache + + + ) + } + + if (!data || data.total === 0) { + return ( + + + +

+ No corroborator signals are currently cached. +

+

+ Cached signals appear here when a corroborating-only source posts before + any matching primary event lands. +

+ + + ) + } + + return ( +
+ + + + + Cache size by source + + + +
+ {data.by_source.map((b) => ( + + {b.source}: {b.count} + + ))} +
+

+ Total cached, unattached, unexpired: {data.total}. Watch the + prefixd_corroborator_cache_size + gauge to alert on caches growing without bound. +

+ + + + + + Cached signals + + +
+ + + + + + + + + + + + + {data.signals.map((s) => ( + + + + + + + + + ))} + +
SourceVectorDimensionsConfidenceIngestedExpires
{s.source} + {s.vector ?? "—"} + + {dimsLabel(s)} + + {s.confidence != null ? s.confidence.toFixed(2) : "—"} + + {formatDistanceToNow(new Date(s.ingested_at), { + addSuffix: true, + })} + + {formatDistanceToNow(new Date(s.expires_at), { + addSuffix: true, + })} +
+
+ + +
+ ) +} diff --git a/frontend/hooks/use-api.ts b/frontend/hooks/use-api.ts index ac5e568..7c960e5 100644 --- a/frontend/hooks/use-api.ts +++ b/frontend/hooks/use-api.ts @@ -363,3 +363,25 @@ export function useOpenSignalGroupCount() { const { data } = useSignalGroups({ status: "open", limit: 1 }) return data?.count ?? 0 } + +export function useCachedCorroborators(params: { + source?: string + limit?: number +} = {}) { + const key = `cached-corroborators:${params.source ?? ""}:${params.limit ?? 100}` + return useSWR( + key, + MOCK_MODE + ? async () => ({ + now: new Date().toISOString(), + total: 0, + by_source: [], + signals: [], + } as Awaited>) + : () => api.getCachedCorroborators(params), + { + refreshInterval: 30_000, + revalidateOnFocus: !MOCK_MODE, + }, + ) +} diff --git a/frontend/lib/api.ts b/frontend/lib/api.ts index 1a3caa3..4accd96 100644 --- a/frontend/lib/api.ts +++ b/frontend/lib/api.ts @@ -760,7 +760,6 @@ export interface CorroboratorResponse { signal_id: string status: "attached" | "cached" attached_group_ids: string[] - cached: boolean } export async function sendCorroborator( @@ -853,6 +852,46 @@ interface CorroboratorActivityResponse { sources: CorroboratorActivityEntry[] } +export interface CachedCorroboratorSignal { + signal_id: string + source: string + vector: string | null + customer_id: string | null + pop: string | null + service_id: string | null + interface: string | null + confidence: number | null + weight: number + ingested_at: string + expires_at: string + raw_details: unknown + attached_group_ids: string[] +} + +export interface CachedCorroboratorBySource { + source: string + count: number +} + +export interface CachedCorroboratorsResponse { + now: string + total: number + by_source: CachedCorroboratorBySource[] + signals: CachedCorroboratorSignal[] +} + +export async function getCachedCorroborators( + params: { source?: string; limit?: number } = {}, +): Promise { + const qs = new URLSearchParams() + if (params.source) qs.set("source", params.source) + if (params.limit) qs.set("limit", String(params.limit)) + const suffix = qs.toString() ? `?${qs}` : "" + return fetchApi( + `/v1/signals/corroborator/cache${suffix}`, + ) +} + export async function getSignalSources(): Promise { // Signal source status is derived from correlation config + recent // events + corroborator activity. Corroborating-only sources never hit diff --git a/migrations/012_signal_groups_playbook.sql b/migrations/012_signal_groups_playbook.sql new file mode 100644 index 0000000..0d0309d --- /dev/null +++ b/migrations/012_signal_groups_playbook.sql @@ -0,0 +1,32 @@ +-- Migration 012: Remember the resolved playbook on signal_groups so the +-- corroborator path can re-resolve playbook-specific correlation +-- overrides (min_sources / confidence_threshold) without needing the +-- full primary-event context. +-- +-- Background: PR A (ADR 021) deliberately left `recompute_group_aggregates` +-- conservative: a corroborator-only recompute would never flip +-- `corroboration_met` from false -> true, because it didn't know which +-- playbook governed the group and thus couldn't resolve the override. +-- This migration adds a nullable `playbook_name`. Primary-event ingest +-- writes it; the corroborator path looks it up and resolves the override +-- against the live playbook config. +-- +-- Backfill is intentionally NOT done in SQL. `mitigations.match_json` +-- doesn't carry the resolved playbook name, and re-running the matcher +-- against a playbook YAML snapshot at migration time is brittle (the +-- live playbook list may have changed between when the group was +-- created and when this migration runs). +-- +-- Instead, the daemon backfills `playbook_name` at runtime using +-- `COALESCE(playbook_name, $resolved)` on the next primary-event ingest +-- path for each group that still has it NULL (see +-- `handle_ban` in `src/api/handlers.rs`). Until that next primary event +-- arrives, the corroborator-side recompute path falls back to the +-- v0.16.0 conservative behavior (no flip of `corroboration_met`). + +ALTER TABLE signal_groups + ADD COLUMN IF NOT EXISTS playbook_name TEXT; + +CREATE INDEX IF NOT EXISTS idx_signal_groups_playbook + ON signal_groups (playbook_name) + WHERE playbook_name IS NOT NULL; diff --git a/src/api/handlers.rs b/src/api/handlers.rs index 447c01a..a3e667b 100644 --- a/src/api/handlers.rs +++ b/src/api/handlers.rs @@ -715,6 +715,9 @@ async fn handle_ban( .add_interface(interface.clone()); } } + // Remember the resolved playbook on the group so the corroborator + // path (PR B) can re-resolve the override at recompute time. + new_group.playbook_name = matching_playbook.map(|p| p.name.clone()); let group = state .repo .insert_signal_group(&new_group) @@ -740,7 +743,14 @@ async fn handle_ban( group.primary_dimensions.add_pop(state.settings.pop.clone()); dims_changed = before != group.primary_dimensions; } - if dims_changed { + // Backfill playbook_name on existing groups created before PR B (or + // before any primary event resolved a playbook). COALESCE in + // update_signal_group keeps an existing non-NULL value stable. + let playbook_changed = group.playbook_name.is_none() && matching_playbook.is_some(); + if playbook_changed { + group.playbook_name = matching_playbook.map(|p| p.name.clone()); + } + if dims_changed || playbook_changed { state .repo .update_signal_group(&group) @@ -5152,9 +5162,13 @@ pub struct CorroboratorInput { #[derive(Debug, Serialize, ToSchema)] pub struct CorroboratorResponse { pub signal_id: Uuid, + /// One of "attached" (one or more groups matched) or "cached" + /// (no matching group; held in the corroborator cache until a + /// matching primary event arrives or TTL expires). Use this + /// instead of the v0.16.0 `cached` boolean field, which is + /// removed in v0.17.0. pub status: String, pub attached_group_ids: Vec, - pub cached: bool, } /// Ingest a corroborating signal. @@ -5347,7 +5361,6 @@ async fn ingest_corroborator_inner( "attached".to_string() }, attached_group_ids, - cached: true, }), )) } @@ -5422,8 +5435,105 @@ pub async fn get_corroborator_activity( })) } +/// Cached-corroborators listing endpoint (PR B). Admin-only. Lists +/// signals currently in the corroborator cache that are unattached and +/// unexpired — i.e. waiting for a matching primary event to drain. +/// Useful for L1 ops to spot a source that's posting heavily but never +/// landing on a real incident. +#[derive(Debug, serde::Deserialize)] +pub struct CachedCorroboratorsQuery { + /// Page size; clamped to [1, 1000]. + #[serde(default)] + pub limit: Option, + /// Filter by signal source. Optional. + #[serde(default)] + pub source: Option, +} + +#[derive(Debug, serde::Serialize, utoipa::ToSchema)] +pub struct CachedCorroboratorsResponse { + pub now: chrono::DateTime, + pub total: u64, + pub by_source: Vec, + pub signals: Vec, +} + +#[derive(Debug, serde::Serialize, utoipa::ToSchema)] +pub struct CachedCorroboratorBySource { + pub source: String, + pub count: u64, +} + +#[utoipa::path( + get, + path = "/v1/signals/corroborator/cache", + tag = "signals", + params( + ("limit" = Option, Query, description = "Page size, default 100, max 1000"), + ("source" = Option, Query, description = "Filter by signal source"), + ), + responses( + (status = 200, description = "Cached corroborator listing", body = CachedCorroboratorsResponse), + (status = 401, description = "Authentication required"), + (status = 403, description = "Admin role required"), + ) +)] +pub async fn list_cached_corroborators_handler( + State(state): State>, + auth_session: AuthSession, + headers: HeaderMap, + axum::extract::Query(query): axum::extract::Query, +) -> Result, StatusCode> { + use crate::domain::OperatorRole; + let auth_header = headers.get(AUTHORIZATION).and_then(|h| h.to_str().ok()); + require_role(&state, &auth_session, auth_header, OperatorRole::Admin)?; + + let limit = query.limit.unwrap_or(100).clamp(1, 1000) as i64; + let now = chrono::Utc::now(); + let signals = state + .repo + .list_cached_corroborators(now, limit, query.source.as_deref()) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + // When ?source= is provided, scope total + by_source to that + // source as well so the response is internally consistent + // (otherwise clients see a `total` that doesn't match the rows + // they were just handed). + let by_source_rows = state + .repo + .count_cached_corroborators_by_source(now) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + let by_source_rows: Vec<(String, u64)> = match query.source.as_deref() { + Some(filter) => by_source_rows + .into_iter() + .filter(|(s, _)| s == filter) + .collect(), + None => by_source_rows, + }; + let total = by_source_rows.iter().map(|(_, n)| *n).sum(); + let by_source = by_source_rows + .into_iter() + .map(|(source, count)| CachedCorroboratorBySource { source, count }) + .collect(); + Ok(Json(CachedCorroboratorsResponse { + now, + total, + by_source, + signals, + })) +} + /// Recompute a signal group's derived_confidence, source_count and /// corroboration_met flag from its events (including corroborators). +/// +/// PR B: this path now re-resolves the playbook-specific correlation +/// override using the stored `playbook_name` on the group, and is +/// allowed to flip `corroboration_met` from false→true even when the +/// triggering ingest was a corroborator. We deliberately do NOT +/// actuate a mitigation from here — the next primary-path event will +/// pick up the flipped flag and trigger normally. This keeps mitigation +/// actuation single-sourced through `handle_ban`. async fn recompute_group_aggregates(state: &Arc, group_id: Uuid) -> Result<(), AppError> { use crate::correlation::CorrelationEngine; @@ -5453,17 +5563,68 @@ async fn recompute_group_aggregates(state: &Arc, group_id: Uuid) -> Re let has_primary = events.iter().any(|e| !e.is_corroborating); - let mut updated = group; - updated.derived_confidence = derived; - updated.source_count = count; - // Only primary ingest has enough context to evaluate playbook-specific - // overrides safely. Corroborator-only recomputes update aggregates but do - // not promote a group from false->true on their own. - updated.corroboration_met = if has_primary { - updated.corroboration_met + // Resolve the playbook override using the group's stored playbook_name. + // If the group was created before PR B (playbook_name is NULL) or the + // playbook has since been removed, fall back to the conservative + // pre-PR-B behavior: aggregates update, but we don't flip + // corroboration_met → true on the corroborator path. + let was_met = group.corroboration_met; + let mut newly_met = false; + let new_met = if has_primary { + let correlation_config = state.correlation_config.read().await.clone(); + let playbooks = state.playbooks.read().await.clone(); + let resolved_playbook = group + .playbook_name + .as_deref() + .and_then(|name| playbooks.playbooks.iter().find(|p| p.name == name)); + let override_ = resolved_playbook.and_then(|p| p.correlation.as_ref()); + match (resolved_playbook, group.playbook_name.as_deref()) { + (Some(_), _) => { + let met = CorrelationEngine::check_corroboration_with_primary( + count, + derived, + has_primary, + &correlation_config, + override_, + ); + if met && !was_met { + newly_met = true; + } + met + } + (None, Some(missing)) => { + tracing::debug!( + group_id = %group_id, + playbook = %missing, + "stored playbook_name no longer resolves; keeping previous corroboration_met" + ); + was_met + } + (None, None) => { + // Pre-PR-B group with no resolved playbook yet — preserve + // previous behavior. + was_met + } + } } else { + // No primary event yet → invariant: corroboration cannot be met. false }; + + if newly_met { + tracing::info!( + group_id = %group_id, + derived_confidence = derived, + source_count = count, + playbook = ?group.playbook_name, + "signal group reached corroboration threshold via corroborator path; awaiting next primary event to actuate mitigation" + ); + } + + let mut updated = group; + updated.derived_confidence = derived; + updated.source_count = count; + updated.corroboration_met = new_met; state .repo .update_signal_group(&updated) diff --git a/src/api/openapi.rs b/src/api/openapi.rs index 0da833b..29f9e05 100644 --- a/src/api/openapi.rs +++ b/src/api/openapi.rs @@ -61,6 +61,7 @@ use crate::db::{GlobalStats, NotificationPreferences, PopInfo, PopStats, Safelis super::handlers::ingest_webhook, super::handlers::ingest_corroborator, super::handlers::get_corroborator_activity, + super::handlers::list_cached_corroborators_handler, super::handlers::get_correlation_config, super::handlers::update_correlation_config, ), @@ -117,6 +118,8 @@ use crate::db::{GlobalStats, NotificationPreferences, PopInfo, PopStats, Safelis super::handlers::CorroboratorResponse, super::handlers::CorroboratorActivityResponse, super::handlers::CorroboratorActivityEntry, + super::handlers::CachedCorroboratorsResponse, + super::handlers::CachedCorroboratorBySource, crate::correlation::PrimaryDimensions, crate::correlation::MatchDimension, crate::correlation::SourceMode, diff --git a/src/api/routes.rs b/src/api/routes.rs index 4e63c88..6ca4336 100644 --- a/src/api/routes.rs +++ b/src/api/routes.rs @@ -126,6 +126,10 @@ fn api_routes() -> Router> { "/v1/signals/corroborator/activity", axum::routing::get(handlers::get_corroborator_activity), ) + .route( + "/v1/signals/corroborator/cache", + axum::routing::get(handlers::list_cached_corroborators_handler), + ) } /// Common layers applied to both production and test routers diff --git a/src/bin/prefixdctl.rs b/src/bin/prefixdctl.rs index 4dcae07..61d231d 100644 --- a/src/bin/prefixdctl.rs +++ b/src/bin/prefixdctl.rs @@ -649,8 +649,6 @@ struct CorroboratorResponse { signal_id: String, status: String, attached_group_ids: Vec, - #[serde(default)] - cached: bool, } #[allow(clippy::too_many_arguments)] diff --git a/src/correlation/engine.rs b/src/correlation/engine.rs index 93e0f4b..a893b5f 100644 --- a/src/correlation/engine.rs +++ b/src/correlation/engine.rs @@ -21,6 +21,15 @@ pub struct SignalGroup { /// Used by the corroborator matching flow (ADR 021). #[serde(default)] pub primary_dimensions: PrimaryDimensions, + /// Name of the playbook that matched the primary event(s) for this + /// group. Used by the corroborator path (PR B) to re-resolve the + /// playbook-specific correlation override (`min_sources`, + /// `confidence_threshold`) without needing the full primary-event + /// context. NULL means: no primary event resolved a playbook yet + /// (or we're upgrading from a pre-PR-B build); corroborator-only + /// recompute falls back to the conservative no-flip behavior. + #[serde(default)] + pub playbook_name: Option, } /// Serializable form of aggregated primary-event dimensions. Stored in the @@ -271,6 +280,7 @@ impl CorrelationEngine { status: SignalGroupStatus::Open, corroboration_met: false, primary_dimensions: PrimaryDimensions::default(), + playbook_name: None, } } @@ -700,6 +710,7 @@ mod tests { status: SignalGroupStatus::Open, corroboration_met: true, primary_dimensions: PrimaryDimensions::default(), + playbook_name: None, }; let contributions = vec![ @@ -741,6 +752,7 @@ mod tests { status: SignalGroupStatus::Open, corroboration_met: false, primary_dimensions: PrimaryDimensions::default(), + playbook_name: None, }; let contributions = vec![SourceContribution { @@ -803,6 +815,7 @@ mod tests { status: SignalGroupStatus::Open, corroboration_met: false, primary_dimensions: PrimaryDimensions::default(), + playbook_name: None, } } diff --git a/src/db/mock.rs b/src/db/mock.rs index e236eea..c721263 100644 --- a/src/db/mock.rs +++ b/src/db/mock.rs @@ -791,14 +791,15 @@ impl RepositoryTrait for MockRepository { now: chrono::DateTime, ) -> Result { let mut cache = self.corroborating_signals.lock().unwrap(); - let mut unattached = 0u64; + let mut unattached: std::collections::HashMap = + std::collections::HashMap::new(); let mut attached = 0u64; cache.retain(|s| { if s.expires_at > now { return true; } if s.attached_group_ids.is_empty() { - unattached += 1; + *unattached.entry(s.source.clone()).or_insert(0) += 1; } else { attached += 1; } @@ -822,16 +823,33 @@ impl RepositoryTrait for MockRepository { &self, now: chrono::DateTime, limit: i64, + source: Option<&str>, ) -> Result> { let cache = self.corroborating_signals.lock().unwrap(); Ok(cache .iter() .filter(|s| s.expires_at > now && s.attached_group_ids.is_empty()) + .filter(|s| source.is_none_or(|src| s.source == src)) .take(limit.max(0) as usize) .cloned() .collect()) } + async fn count_cached_corroborators_by_source( + &self, + now: chrono::DateTime, + ) -> Result> { + use std::collections::HashMap; + let cache = self.corroborating_signals.lock().unwrap(); + let mut counts: HashMap = HashMap::new(); + for s in cache.iter() { + if s.expires_at > now && s.attached_group_ids.is_empty() { + *counts.entry(s.source.clone()).or_insert(0) += 1; + } + } + Ok(counts.into_iter().collect()) + } + async fn corroborator_source_activity( &self, since: chrono::DateTime, diff --git a/src/db/mod.rs b/src/db/mod.rs index 67bc489..ee23994 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -81,6 +81,11 @@ async fn run_migrations(pool: &PgPool) -> Result<()> { "backfill_primary_dimensions", include_str!("../../migrations/011_backfill_primary_dimensions.sql"), ), + ( + 12, + "signal_groups_playbook", + include_str!("../../migrations/012_signal_groups_playbook.sql"), + ), ]; // Bootstrap: run all migrations first (they use IF NOT EXISTS) diff --git a/src/db/repository.rs b/src/db/repository.rs index b4439f5..0051acc 100644 --- a/src/db/repository.rs +++ b/src/db/repository.rs @@ -911,18 +911,20 @@ impl RepositoryTrait for Repository { WITH existing AS ( SELECT group_id, victim_ip, vector, created_at, window_expires_at, derived_confidence, source_count, status, corroboration_met, - primary_dimensions + primary_dimensions, playbook_name FROM signal_groups WHERE victim_ip = $2 AND vector = $3 AND status = 'open' AND window_expires_at > NOW() LIMIT 1 ), inserted AS ( INSERT INTO signal_groups (group_id, victim_ip, vector, created_at, window_expires_at, - derived_confidence, source_count, status, corroboration_met, primary_dimensions) - SELECT $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 + derived_confidence, source_count, status, corroboration_met, primary_dimensions, + playbook_name) + SELECT $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11 WHERE NOT EXISTS (SELECT 1 FROM existing) RETURNING group_id, victim_ip, vector, created_at, window_expires_at, - derived_confidence, source_count, status, corroboration_met, primary_dimensions + derived_confidence, source_count, status, corroboration_met, primary_dimensions, + playbook_name ) SELECT * FROM existing UNION ALL @@ -940,6 +942,7 @@ impl RepositoryTrait for Repository { .bind(group.status.as_str()) .bind(group.corroboration_met) .bind(serde_json::to_value(&group.primary_dimensions).unwrap_or(serde_json::json!({}))) + .bind(group.playbook_name.as_deref()) .fetch_one(&self.pool) .await; @@ -957,7 +960,7 @@ impl RepositoryTrait for Repository { r#" SELECT group_id, victim_ip, vector, created_at, window_expires_at, derived_confidence, source_count, status, corroboration_met, - primary_dimensions + primary_dimensions, playbook_name FROM signal_groups WHERE victim_ip = $1 AND vector = $2 AND status = 'open' AND window_expires_at > NOW() @@ -982,7 +985,8 @@ impl RepositoryTrait for Repository { source_count = $3, status = $4, corroboration_met = $5, - primary_dimensions = $6 + primary_dimensions = $6, + playbook_name = COALESCE($7, playbook_name) WHERE group_id = $1 "#, ) @@ -992,6 +996,7 @@ impl RepositoryTrait for Repository { .bind(group.status.as_str()) .bind(group.corroboration_met) .bind(serde_json::to_value(&group.primary_dimensions).unwrap_or(serde_json::json!({}))) + .bind(group.playbook_name.as_deref()) .execute(&self.pool) .await?; Ok(()) @@ -1002,7 +1007,7 @@ impl RepositoryTrait for Repository { r#" SELECT group_id, victim_ip, vector, created_at, window_expires_at, derived_confidence, source_count, status, corroboration_met, - primary_dimensions + primary_dimensions, playbook_name FROM signal_groups WHERE group_id = $1 "#, ) @@ -1017,7 +1022,7 @@ impl RepositoryTrait for Repository { r#" SELECT group_id, victim_ip, vector, created_at, window_expires_at, derived_confidence, source_count, status, corroboration_met, - primary_dimensions + primary_dimensions, playbook_name FROM signal_groups WHERE victim_ip = $1 AND vector = $2 AND status = 'open' AND window_expires_at > NOW() @@ -1097,7 +1102,7 @@ impl RepositoryTrait for Repository { r#" SELECT group_id, victim_ip, vector, created_at, window_expires_at, derived_confidence, source_count, status, corroboration_met, - primary_dimensions + primary_dimensions, playbook_name FROM signal_groups WHERE ($1::text IS NULL OR status = $1) AND ($2::text IS NULL OR vector = $2) @@ -1132,7 +1137,7 @@ impl RepositoryTrait for Repository { r#" SELECT group_id, victim_ip, vector, created_at, window_expires_at, derived_confidence, source_count, status, corroboration_met, - primary_dimensions + primary_dimensions, playbook_name FROM signal_groups WHERE status = 'open' AND window_expires_at <= NOW() "#, @@ -1161,7 +1166,7 @@ impl RepositoryTrait for Repository { r#" SELECT group_id, victim_ip, vector, created_at, window_expires_at, derived_confidence, source_count, status, corroboration_met, - primary_dimensions + primary_dimensions, playbook_name FROM signal_groups WHERE status = 'open' AND window_expires_at > $1 @@ -1332,24 +1337,22 @@ impl RepositoryTrait for Repository { &self, now: chrono::DateTime, ) -> Result { - // Two statements so the scheduler can attribute the expired metric - // to truly-unattached signals (cache misses) while still cleaning - // attached audit rows. Both run inside the same request round-trip - // order but we don't need a transaction: the counters are - // monotonic and the delete predicate is narrow. - let unattached: (i64,) = sqlx::query_as( + // Two statements: one DELETE-RETURNING grouped by source for + // per-source attribution on the expired metric; another for + // attached rows (no source attribution — they aren't cache misses). + let unattached_rows: Vec<(String, i64)> = sqlx::query_as( r#" WITH deleted AS ( DELETE FROM corroborating_signals WHERE expires_at <= $1 AND cardinality(attached_group_ids) = 0 - RETURNING signal_id + RETURNING source ) - SELECT COUNT(*) FROM deleted + SELECT source, COUNT(*)::BIGINT AS n FROM deleted GROUP BY source "#, ) .bind(now) - .fetch_one(&self.pool) + .fetch_all(&self.pool) .await?; let attached: (i64,) = sqlx::query_as( r#" @@ -1365,8 +1368,12 @@ impl RepositoryTrait for Repository { .bind(now) .fetch_one(&self.pool) .await?; + let unattached_expired = unattached_rows + .into_iter() + .map(|(source, n)| (source, n.max(0) as u64)) + .collect(); Ok(crate::db::traits::CorroboratorSweepStats { - unattached_expired: unattached.0.max(0) as u64, + unattached_expired, attached_expired: attached.0.max(0) as u64, }) } @@ -1385,6 +1392,7 @@ impl RepositoryTrait for Repository { &self, now: chrono::DateTime, limit: i64, + source: Option<&str>, ) -> Result> { let rows: Vec = sqlx::query_as( r#" @@ -1393,17 +1401,41 @@ impl RepositoryTrait for Repository { FROM corroborating_signals WHERE expires_at > $1 AND cardinality(attached_group_ids) = 0 + AND ($3::text IS NULL OR source = $3) ORDER BY ingested_at DESC LIMIT $2 "#, ) .bind(now) .bind(limit.max(0)) + .bind(source) .fetch_all(&self.pool) .await?; Ok(rows.into_iter().map(Into::into).collect()) } + async fn count_cached_corroborators_by_source( + &self, + now: DateTime, + ) -> Result> { + let rows: Vec<(String, i64)> = sqlx::query_as( + r#" + SELECT source, COUNT(*)::BIGINT AS n + FROM corroborating_signals + WHERE expires_at > $1 + AND cardinality(attached_group_ids) = 0 + GROUP BY source + "#, + ) + .bind(now) + .fetch_all(&self.pool) + .await?; + Ok(rows + .into_iter() + .map(|(source, n)| (source, n.max(0) as u64)) + .collect()) + } + async fn corroborator_source_activity( &self, since: DateTime, @@ -1502,6 +1534,7 @@ struct SignalGroupRow { status: String, corroboration_met: bool, primary_dimensions: serde_json::Value, + playbook_name: Option, } impl From for SignalGroup { @@ -1517,6 +1550,7 @@ impl From for SignalGroup { status: row.status.parse().unwrap_or(SignalGroupStatus::Open), corroboration_met: row.corroboration_met, primary_dimensions: serde_json::from_value(row.primary_dimensions).unwrap_or_default(), + playbook_name: row.playbook_name, } } } diff --git a/src/db/traits.rs b/src/db/traits.rs index 4b4ab79..1b64a0e 100644 --- a/src/db/traits.rs +++ b/src/db/traits.rs @@ -12,17 +12,25 @@ use super::{GlobalStats, PopInfo, SafelistEntry, TimeseriesBucket}; /// Return value for `delete_expired_corroborating_signals`. /// -/// `unattached_expired` is the count of signals the scheduler should -/// increment `CORROBORATOR_EXPIRED_TOTAL` by: signals that were cached -/// because no primary group matched at ingest and then timed out without -/// ever attaching. `attached_expired` rows are the audit copies retained -/// for late fan-out; their deletion is bookkeeping, not a cache miss. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +/// `unattached_expired` is per-source counts of signals the scheduler +/// should increment `CORROBORATOR_EXPIRED_TOTAL{source}` by: signals +/// that were cached because no primary group matched at ingest and +/// then timed out without ever attaching. `attached_expired` rows are +/// the audit copies retained for late fan-out; their deletion is +/// bookkeeping, not a cache miss, so we don't attribute them by +/// source. +#[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct CorroboratorSweepStats { - pub unattached_expired: u64, + pub unattached_expired: std::collections::HashMap, pub attached_expired: u64, } +impl CorroboratorSweepStats { + pub fn unattached_total(&self) -> u64 { + self.unattached_expired.values().sum() + } +} + /// Per-source activity summary for the Signals dashboard, covering both /// primary-event sources (via the `events` table) and corroborator-only /// sources (via `corroborating_signals` + `signal_group_events`). @@ -246,11 +254,14 @@ pub trait RepositoryTrait: Send + Sync { /// Count currently-cached (unattached, unexpired) corroborating signals. /// For operator dashboards / metrics. async fn count_cached_corroborators(&self, now: chrono::DateTime) -> Result; - /// List currently-cached corroborating signals for admin UI / debugging. + /// List currently-cached corroborating signals for admin UI / + /// debugging. When `source` is `Some`, results are filtered to + /// only that source. async fn list_cached_corroborators( &self, now: chrono::DateTime, limit: i64, + source: Option<&str>, ) -> Result>; /// Aggregate corroborator activity per source since `since`, across /// both the live cache (`corroborating_signals`) and attached rows @@ -261,4 +272,13 @@ pub trait RepositoryTrait: Send + Sync { &self, since: DateTime, ) -> Result>; + /// Per-source counts of currently-cached unattached, unexpired + /// corroborating signals. Used by the reconcile loop to set the + /// `prefixd_corroborator_cache_size{source}` gauge on each tick so + /// operators can alert on caches that grow without bound (e.g. + /// when a source posts but no matching primary event ever lands). + async fn count_cached_corroborators_by_source( + &self, + now: DateTime, + ) -> Result>; } diff --git a/src/observability/metrics.rs b/src/observability/metrics.rs index 4f7dd23..44c24e1 100644 --- a/src/observability/metrics.rs +++ b/src/observability/metrics.rs @@ -269,7 +269,20 @@ pub static CORROBORATOR_EXPIRED_TOTAL: Lazy = Lazy::new(|| { register_counter_vec!( "prefixd_corroborator_expired_total", "Corroborating signals removed by cache sweep without ever attaching", - &[] + &["source"] + ) + .unwrap() +}); + +/// Currently-cached, unattached, unexpired corroborating signals by source. +/// Updated by the reconcile loop every tick after the sweep. Operators can +/// alert on this gauge growing without bound (e.g. a source posting heavily +/// while no matching primary event ever lands). +pub static CORROBORATOR_CACHE_SIZE: Lazy = Lazy::new(|| { + register_gauge_vec!( + "prefixd_corroborator_cache_size", + "Currently-cached unattached corroborating signals, by source", + &["source"] ) .unwrap() }); @@ -315,6 +328,7 @@ pub fn init_metrics() { Lazy::force(&CORROBORATOR_INGESTED_TOTAL); Lazy::force(&CORROBORATOR_ATTACHED_TOTAL); Lazy::force(&CORROBORATOR_EXPIRED_TOTAL); + Lazy::force(&CORROBORATOR_CACHE_SIZE); } /// Update database pool metrics from sqlx pool stats diff --git a/src/scheduler/reconcile.rs b/src/scheduler/reconcile.rs index a9496c1..6609070 100644 --- a/src/scheduler/reconcile.rs +++ b/src/scheduler/reconcile.rs @@ -17,6 +17,11 @@ pub struct ReconciliationLoop { dry_run: bool, ws_broadcast: Option>, alerting: Option>>>, + /// Set of source labels we last set on + /// `CORROBORATOR_CACHE_SIZE`. Used to zero-out gauges when a + /// source's cache drains to empty between ticks (Prometheus would + /// otherwise keep the last non-zero value forever). + last_cache_sources: tokio::sync::Mutex>, } impl ReconciliationLoop { @@ -33,6 +38,7 @@ impl ReconciliationLoop { dry_run, ws_broadcast: None, alerting: None, + last_cache_sources: tokio::sync::Mutex::new(std::collections::HashSet::new()), } } @@ -203,30 +209,48 @@ impl ReconciliationLoop { async fn sweep_corroborator_cache(&self) -> anyhow::Result<()> { // Clean expired rows from the corroborating_signals cache. The - // repository splits the delete into unattached vs attached so we - // only charge `CORROBORATOR_EXPIRED_TOTAL` for true cache misses - // (ingested, never matched any group, timed out). Attached rows - // still get GC'd but don't inflate the expired counter — their - // audit trail already lives on signal_group_events. + // repository splits the delete into unattached (per-source) vs + // attached: only true cache misses (ingested, never matched any + // group, timed out) charge `CORROBORATOR_EXPIRED_TOTAL{source}`. + // Attached rows are GC'd silently — their audit trail already + // lives on signal_group_events. // - // Per-source attribution for the expired counter is deferred to - // PR B (see ROADMAP -> Correlation Engine -> Corroborating - // signals v2). + // After the sweep we also refresh `CORROBORATOR_CACHE_SIZE{source}` + // so operators can alert on caches growing without bound. let now = chrono::Utc::now(); let stats = self.repo.delete_expired_corroborating_signals(now).await?; - let total = stats.unattached_expired + stats.attached_expired; + let unattached_total = stats.unattached_total(); + let total = unattached_total + stats.attached_expired; if total > 0 { tracing::info!( - unattached_expired = stats.unattached_expired, + unattached_expired = unattached_total, attached_expired = stats.attached_expired, "swept expired corroborating signals from cache" ); - if stats.unattached_expired > 0 { + for (source, count) in &stats.unattached_expired { crate::observability::metrics::CORROBORATOR_EXPIRED_TOTAL - .with_label_values(&[] as &[&str]) - .inc_by(stats.unattached_expired as f64); + .with_label_values(&[source.as_str()]) + .inc_by(*count as f64); } } + + // Refresh the cache_size gauge: zero out previous values for + // sources that no longer have rows, then set per-source counts. + let by_source = self.repo.count_cached_corroborators_by_source(now).await?; + let live_sources: std::collections::HashSet = + by_source.iter().map(|(s, _)| s.clone()).collect(); + let mut last = self.last_cache_sources.lock().await; + for stale in last.difference(&live_sources) { + crate::observability::metrics::CORROBORATOR_CACHE_SIZE + .with_label_values(&[stale.as_str()]) + .set(0.0); + } + for (source, count) in &by_source { + crate::observability::metrics::CORROBORATOR_CACHE_SIZE + .with_label_values(&[source.as_str()]) + .set(*count as f64); + } + *last = live_sources; Ok(()) } diff --git a/tests/integration.rs b/tests/integration.rs index 7641a02..10d8fcf 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -5515,11 +5515,110 @@ async fn test_expired_sweep_splits_attached_vs_unattached() { .delete_expired_corroborating_signals(now) .await .unwrap(); - assert_eq!(stats.unattached_expired, 1); + assert_eq!(stats.unattached_total(), 1); + assert_eq!(stats.unattached_expired.get("router-cpu"), Some(&1)); assert_eq!(stats.attached_expired, 1); assert_eq!(repo.count_cached_corroborators(now).await.unwrap(), 1); } +#[tokio::test] +async fn test_expired_sweep_attributes_per_source() { + use chrono::{Duration, Utc}; + use prefixd::correlation::CorroboratingSignal; + use uuid::Uuid; + + let repo = MockRepository::new(); + let now = Utc::now(); + + // 2 expired unattached from router-cpu, 1 from pop-utilization + for src in ["router-cpu", "router-cpu", "pop-utilization"] { + repo.insert_corroborating_signal(&CorroboratingSignal { + signal_id: Uuid::new_v4(), + source: src.to_string(), + vector: None, + customer_id: None, + pop: Some("iad1".to_string()), + service_id: None, + interface: None, + confidence: Some(0.5), + weight: 0.5, + ingested_at: now - Duration::seconds(600), + expires_at: now - Duration::seconds(60), + raw_details: None, + attached_group_ids: vec![], + }) + .await + .unwrap(); + } + + let stats = repo + .delete_expired_corroborating_signals(now) + .await + .unwrap(); + assert_eq!(stats.unattached_expired.len(), 2); + assert_eq!(stats.unattached_expired.get("router-cpu"), Some(&2)); + assert_eq!(stats.unattached_expired.get("pop-utilization"), Some(&1)); + assert_eq!(stats.attached_expired, 0); +} + +#[tokio::test] +async fn test_count_cached_corroborators_by_source() { + use chrono::{Duration, Utc}; + use prefixd::correlation::CorroboratingSignal; + use uuid::Uuid; + + let repo = MockRepository::new(); + let now = Utc::now(); + + // 3 unattached unexpired from a, 1 unattached unexpired from b, + // 1 attached unexpired from a (excluded), 1 expired unattached from a + // (excluded). + let cases = [ + ("a", false, false, 0i64), // unexpired, unattached + ("a", false, false, 0), // unexpired, unattached + ("a", false, false, 0), // unexpired, unattached + ("b", false, false, 0), // unexpired, unattached + ("a", true, false, 0), // attached -> excluded + ("a", false, true, -3600i64), // expired -> excluded (insert with past expires_at) + ]; + for (src, attached, expired, ingest_offset) in cases { + repo.insert_corroborating_signal(&CorroboratingSignal { + signal_id: Uuid::new_v4(), + source: src.to_string(), + vector: None, + customer_id: None, + pop: Some("iad1".to_string()), + service_id: None, + interface: None, + confidence: Some(0.5), + weight: 0.5, + ingested_at: now + Duration::seconds(ingest_offset), + expires_at: if expired { + now - Duration::seconds(60) + } else { + now + Duration::seconds(300) + }, + raw_details: None, + attached_group_ids: if attached { + vec![Uuid::new_v4()] + } else { + vec![] + }, + }) + .await + .unwrap(); + } + + let counts: std::collections::HashMap = repo + .count_cached_corroborators_by_source(now) + .await + .unwrap() + .into_iter() + .collect(); + assert_eq!(counts.get("a"), Some(&3)); + assert_eq!(counts.get("b"), Some(&1)); +} + #[tokio::test] async fn test_corroborator_source_activity_merges_cache_rows() { use chrono::{Duration, Utc}; @@ -5567,3 +5666,439 @@ async fn test_corroborator_source_activity_merges_cache_rows() { assert!(cpu.last_seen.is_some()); assert_eq!(by_source.remove("pop-utilization").unwrap().count, 1); } + +#[tokio::test] +async fn test_late_corroborator_finalizes_with_playbook_override() { + // PR B: a late corroborator can flip corroboration_met=true on its + // own path, using the override resolved from the group's stored + // playbook_name. We set up a group whose primary event lands below + // the global threshold but above the playbook override threshold, + // and confirm a corroborator promotes the flag. + use prefixd::correlation::{ + MatchDimension, PlaybookCorrelationOverride, SourceConfig, SourceMode, + }; + use prefixd::domain::AttackVector; + + let repo: Arc = Arc::new(MockRepository::new()); + let announcer = Arc::new(MockAnnouncer::new()); + + // Global config: min_sources=3, threshold=0.9 — primary alone is far + // from meeting it. The playbook override drops both to 2 / 0.5, so a + // single corroborator (count → 2) should finalize. + let mut settings = test_settings_with_correlation(true, 3, 0.9); + settings.correlation.sources.insert( + "router-cpu".to_string(), + SourceConfig { + weight: 0.6, + r#type: "telemetry".to_string(), + confidence_mapping: std::collections::HashMap::new(), + mode: SourceMode::Corroborating, + match_dimensions: vec![MatchDimension::Pop], + }, + ); + + let playbooks = Playbooks { + playbooks: vec![Playbook { + name: "udp_flood_test".to_string(), + match_criteria: PlaybookMatch { + vector: AttackVector::UdpFlood, + require_top_ports: false, + }, + correlation: Some(PlaybookCorrelationOverride { + min_sources: Some(2), + confidence_threshold: Some(0.5), + }), + steps: vec![PlaybookStep { + action: PlaybookAction::Police, + rate_bps: Some(5_000_000), + ttl_seconds: 120, + require_confidence_at_least: None, + require_persistence_seconds: None, + }], + }], + }; + + let state = AppState::new( + settings, + test_inventory(), + playbooks, + repo.clone(), + announcer, + std::path::PathBuf::from("."), + ) + .expect("state"); + let app = create_test_router(state); + + // Step 1: primary event — group exists, playbook_name is set, + // corroboration_met=false (single source against override min=2). + let event_body = serde_json::json!({ + "source": "fastnetmon", + "vector": "udp_flood", + "victim_ip": "203.0.113.10", + "timestamp": chrono::Utc::now().to_rfc3339(), + "confidence": 0.7, + "action": "ban" + }); + let resp = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/events") + .header("content-type", "application/json") + .body(Body::from(event_body.to_string())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::ACCEPTED); + + let groups_before = repo + .list_signal_groups( + &prefixd::correlation::engine::SignalGroupFilter::default(), + &prefixd::db::ListParams { + limit: 10, + ..Default::default() + }, + ) + .await + .unwrap(); + assert_eq!(groups_before.len(), 1); + let group = &groups_before[0]; + assert_eq!(group.playbook_name.as_deref(), Some("udp_flood_test")); + assert!( + !group.corroboration_met, + "single primary event should not yet meet override threshold" + ); + + // Step 2: corroborator with matching pop → group now has 2 distinct + // sources, derived_confidence above 0.5, override allows promotion. + let sig_body = serde_json::json!({ + "source": "router-cpu", + "pop": "test1", + "vector": "udp_flood", + "confidence": 0.6 + }); + let resp2 = app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/signals/corroborator") + .header("content-type", "application/json") + .body(Body::from(sig_body.to_string())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp2.status(), StatusCode::OK); + + let groups_after = repo + .list_signal_groups( + &prefixd::correlation::engine::SignalGroupFilter::default(), + &prefixd::db::ListParams { + limit: 10, + ..Default::default() + }, + ) + .await + .unwrap(); + let group_after = &groups_after[0]; + assert!( + group_after.corroboration_met, + "late corroborator should finalize via playbook override" + ); + assert_eq!(group_after.source_count, 2); +} + +#[tokio::test] +async fn test_late_corroborator_skips_when_playbook_name_is_stale() { + // If a group's stored playbook_name no longer resolves (admin removed + // it from the live config), the corroborator path falls back to the + // conservative v0.16.0 behavior: aggregates update on each ingest + // but corroboration_met is *not* flipped from false -> true. We + // exercise this end-to-end by: + // 1) seeding an open group whose stored playbook name doesn't + // exist in the live playbook list, + // 2) posting a corroborator that *would* satisfy the override if + // the playbook resolved, and + // 3) asserting the flag stayed false even though source_count and + // derived_confidence advanced. + use chrono::{Duration, Utc}; + use prefixd::correlation::engine::{PrimaryDimensions, SignalGroup, SignalGroupStatus}; + use prefixd::correlation::{MatchDimension, SourceConfig, SourceMode}; + use uuid::Uuid; + + let repo: Arc = Arc::new(MockRepository::new()); + let announcer = Arc::new(MockAnnouncer::new()); + + // Global config that would never trigger on its own (min=3, 0.9). + // No playbook named `no_such_playbook` is configured, so the stored + // value won't resolve; conservative fallback should fire. + // + // We register two corroborating sources on the same dimension so a + // pair of POSTs gets us source_count=2 — enough to *would-be* + // satisfy a 2/0.5 override if the playbook resolved. + let mut settings = test_settings_with_correlation(true, 3, 0.9); + for src in ["router-cpu", "pop-utilization"] { + settings.correlation.sources.insert( + src.to_string(), + SourceConfig { + weight: 0.6, + r#type: "telemetry".to_string(), + confidence_mapping: std::collections::HashMap::new(), + mode: SourceMode::Corroborating, + match_dimensions: vec![MatchDimension::Pop], + }, + ); + } + + // Seed a group as if a primary event already created it pointing at + // a playbook that has since been removed from config. + let now = Utc::now(); + let group_id = Uuid::new_v4(); + repo.insert_signal_group(&SignalGroup { + group_id, + victim_ip: "203.0.113.10".to_string(), + vector: "udp_flood".to_string(), + created_at: now, + window_expires_at: now + Duration::seconds(300), + derived_confidence: 0.7, + source_count: 1, + status: SignalGroupStatus::Open, + corroboration_met: false, + primary_dimensions: { + let mut d = PrimaryDimensions::default(); + d.add_pop("test1".to_string()); + d + }, + playbook_name: Some("no_such_playbook".to_string()), + }) + .await + .unwrap(); + // Make has_primary=true so the corroborator path is even allowed + // to consider the group. + repo.add_event_to_group(group_id, Uuid::new_v4(), 1.0) + .await + .unwrap(); + + let state = AppState::new( + settings, + test_inventory(), + test_playbooks(), // does NOT contain "no_such_playbook" + repo.clone(), + announcer, + std::path::PathBuf::from("."), + ) + .expect("state"); + let app = create_test_router(state); + + // POST two corroborators from distinct sources matching on pop. + // Without the stale-playbook guard, recompute_group_aggregates + // would resolve the override and flip corroboration_met to true + // (source_count reaches 2, derived_confidence above 0.5 override). + // With the guard, both ingests advance aggregates but the flag + // stays false. + for src in ["router-cpu", "pop-utilization"] { + let body = serde_json::json!({ + "source": src, + "pop": "test1", + "vector": "udp_flood", + "confidence": 0.9 + }); + let resp = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/signals/corroborator") + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + } + + let group = repo.get_signal_group(group_id).await.unwrap().unwrap(); + assert_eq!( + group.source_count, 2, + "aggregates should still update on the corroborator path" + ); + assert!( + !group.corroboration_met, + "stale playbook_name must keep the conservative no-flip behavior" + ); +} + +#[tokio::test] +async fn test_cache_listing_endpoint_returns_unattached_signals() { + use chrono::{Duration, Utc}; + use prefixd::correlation::{CorroboratingSignal, MatchDimension, SourceConfig, SourceMode}; + use uuid::Uuid; + + let repo: Arc = Arc::new(MockRepository::new()); + let announcer = Arc::new(MockAnnouncer::new()); + let mut settings = test_settings_with_correlation(true, 1, 0.5); + settings.correlation.sources.insert( + "router-cpu".to_string(), + SourceConfig { + weight: 0.5, + r#type: "telemetry".to_string(), + confidence_mapping: std::collections::HashMap::new(), + mode: SourceMode::Corroborating, + match_dimensions: vec![MatchDimension::Pop], + }, + ); + + let now = Utc::now(); + repo.insert_corroborating_signal(&CorroboratingSignal { + signal_id: Uuid::new_v4(), + source: "router-cpu".to_string(), + vector: Some("udp_flood".to_string()), + customer_id: None, + pop: Some("iad1".to_string()), + service_id: None, + interface: None, + confidence: Some(0.5), + weight: 0.5, + ingested_at: now, + expires_at: now + Duration::seconds(300), + raw_details: None, + attached_group_ids: vec![], + }) + .await + .unwrap(); + // Attached row that should NOT show up in the listing. + repo.insert_corroborating_signal(&CorroboratingSignal { + signal_id: Uuid::new_v4(), + source: "router-cpu".to_string(), + vector: Some("udp_flood".to_string()), + customer_id: None, + pop: Some("iad1".to_string()), + service_id: None, + interface: None, + confidence: Some(0.5), + weight: 0.5, + ingested_at: now, + expires_at: now + Duration::seconds(300), + raw_details: None, + attached_group_ids: vec![Uuid::new_v4()], + }) + .await + .unwrap(); + + let state = AppState::new( + settings, + test_inventory(), + test_playbooks(), + repo, + announcer, + std::path::PathBuf::from("."), + ) + .expect("state"); + let app = create_test_router(state); + + let resp = app + .oneshot( + Request::builder() + .method("GET") + .uri("/v1/signals/corroborator/cache?limit=10") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(resp.into_body(), 1024 * 1024) + .await + .unwrap(); + let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + let signals = json["signals"].as_array().unwrap(); + assert_eq!(signals.len(), 1, "attached signal should be excluded"); + assert_eq!(json["total"], 1); + assert_eq!(json["by_source"][0]["source"], "router-cpu"); + assert_eq!(json["by_source"][0]["count"], 1); +} + +#[tokio::test] +async fn test_cache_listing_endpoint_source_filter_scopes_aggregates() { + // Regression: when `?source=` is passed, total + by_source must + // also be filtered to that source. Otherwise clients see a `total` + // that doesn't match the rows they were just handed. + use chrono::{Duration, Utc}; + use prefixd::correlation::{CorroboratingSignal, MatchDimension, SourceConfig, SourceMode}; + use uuid::Uuid; + + let repo: Arc = Arc::new(MockRepository::new()); + let announcer = Arc::new(MockAnnouncer::new()); + let mut settings = test_settings_with_correlation(true, 1, 0.5); + for src in ["router-cpu", "pop-utilization"] { + settings.correlation.sources.insert( + src.to_string(), + SourceConfig { + weight: 0.5, + r#type: "telemetry".to_string(), + confidence_mapping: std::collections::HashMap::new(), + mode: SourceMode::Corroborating, + match_dimensions: vec![MatchDimension::Pop], + }, + ); + } + + // Three rows: 2 router-cpu, 1 pop-utilization. All unattached + unexpired. + let now = Utc::now(); + for src in ["router-cpu", "router-cpu", "pop-utilization"] { + repo.insert_corroborating_signal(&CorroboratingSignal { + signal_id: Uuid::new_v4(), + source: src.to_string(), + vector: Some("udp_flood".to_string()), + customer_id: None, + pop: Some("iad1".to_string()), + service_id: None, + interface: None, + confidence: Some(0.5), + weight: 0.5, + ingested_at: now, + expires_at: now + Duration::seconds(300), + raw_details: None, + attached_group_ids: vec![], + }) + .await + .unwrap(); + } + + let state = AppState::new( + settings, + test_inventory(), + test_playbooks(), + repo, + announcer, + std::path::PathBuf::from("."), + ) + .expect("state"); + let app = create_test_router(state); + + let resp = app + .oneshot( + Request::builder() + .method("GET") + .uri("/v1/signals/corroborator/cache?source=pop-utilization&limit=10") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(resp.into_body(), 1024 * 1024) + .await + .unwrap(); + let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + let signals = json["signals"].as_array().unwrap(); + assert_eq!(signals.len(), 1); + assert_eq!(json["total"], 1, "total must be scoped to ?source="); + let by_source = json["by_source"].as_array().unwrap(); + assert_eq!(by_source.len(), 1); + assert_eq!(by_source[0]["source"], "pop-utilization"); + assert_eq!(by_source[0]["count"], 1); +}