fix(ingestor): warn, throttled and bounded, when observerIATAWhitelist drops a region

[dborup]

Relates to #110

Summary

Until now, a message whose MQTT topic region was not in observerIATAWhitelist was dropped without a trace. A legitimate region that was left off the list lost all its traffic and nothing appeared in the log.

The ingestor now writes one grep-friendly line per dropped region:

MQTT [src] [region-filter] dropping region "GOT": not in observerIATAWhitelist; further messages from this region suppressed for 6h0m0s

• Throttling. While the region keeps arriving, the line is repeated at most every iataWarnIntervalSec (default 6 h).
• Bounded state. The region code is a topic segment the publisher controls, so the throttle state is strictly bounded. This is also where the fork deliberately differs from upstream (see below).

Upstream reference (read only): Kpa-clawbot/CoreScope#2067.

Plan and design

Autonomous run, so the plan is written here instead of waiting for approval (AGENTS.md rule 5).

1. Test first (commit 9e68cd44). iata_drop_log_test.go drives the real handleMessage and reads the log it writes. It is red on master.
2. Fix (commit d9f4c328).

cmd/ingestor/iata_drop_warn.go (new)

• Normalization. The code is trimmed, upper-cased and cut to 32 bytes on a rune boundary. The result is both the key and the logged form.
• Per-region throttle. The first drop of a region warns, and later drops within the interval are silent. After the interval the region warns again, and that restarts the interval.
• Bounded table. At most 512 regions are tracked.
  • Beyond that, drops share one overflow warning with the same interval, so they are never silent. The overflow line reads "throttle table full".
  • When the table is full, entries older than the interval are reclaimed, so the first codes ever seen do not own it forever.
• Amortized sweep. A lower bound on the oldest entry means a sweep only runs when some entry can actually have expired. A hostile feed at the cap therefore costs O(1) per drop, not a 512-entry scan.
• Log injection. The region is logged with %q, so control characters cannot break or forge log lines.

Wiring and configuration

• config.go. New optional iataWarnIntervalSec. A value of 0 or less means the 6 h default. The throttle state is unexported.
• main.go. On the existing reject branch only, cfg.warnIATADrop(tag, parts[1], time.Now()) is called before return. Allowed traffic and an empty whitelist take exactly the same path as before.
• config.example.json. Documents the key and the log line.

Where the fork differs from upstream

• Key length. Upstream bounds the number of keys but not their length, so 512 keys of about 64 KB each would be accepted. Here keys are capped at 32 bytes.
• Log quoting. Upstream logs the raw code with %s. Here it is quoted with %q.
• Sweep cost. Upstream sweeps all 512 entries on every drop once the table is full. Here the sweep is amortized: benchmark below.
• Interval format. Upstream formats the interval as whole hours (%.0fh), which prints "0h" for a 90-second interval. Here it is printed as a duration.

Config and customizer (AGENTS.md rule 8). iataWarnIntervalSec is an ingestor setting, not a UI value, so it has no customizer counterpart.

Acceptance criteria

Criterion	Status	Evidence
Fixed, grep-friendly warning with the normalized region code	Met	TestIATAWhitelistDropIsLoggedOncePerRegion ("GOT", observerIATAWhitelist, [region-filter])
Throttled per code, configurable interval, safe default	Met	TestIATADropThrottlePerRegionAndExpiry, TestIATAWarnIntervalDefaultAndConfigured, TestIATAWarnIntervalFromJSON
Per-code state strictly bounded	Met	512 keys of at most 32 bytes: TestIATADropThrottleIsBoundedWithSharedOverflow (20,000 distinct codes), TestNormalizeIATAForWarn
Traffic beyond the bound is not silent: bounded/shared overflow warning	Met	Same test: exactly 1 overflow warning per interval; TestIATAWhitelistDropManyDistinctRegionsStaysBoundedAndVisible (3,000 regions through handleMessage)
Expired entries reclaimed	Met	TestIATADropThrottleReclaimsExpiredSlots; sweeps amortized: TestIATADropThrottleSweepsAreAmortized
Allowed traffic and an empty whitelist unchanged	Met	TestIATAWhitelistAllowedAndEmptyAreSilent; the existing TestHandleMessageObserverIATAWhitelist and IATA filter tests still pass
Tests: throttling, expiry, more attacker values than the cap, overflow, reclamation	Met	As above, plus TestIATADropThrottleConcurrent (16 goroutines)
Ingestor suite with -race	Met	See Tests
Document the optional key	Met	config.example.json

Tests

Reproduction

• go test -run TestIATAWhitelist ./cmd/ingestor on commit 9e68cd44: the 4 warning cases fail with "got 0" lines. The allowed/empty control passes.
• On this branch, all IATA tests pass (21 including the existing ones).

Full suite with the race detector

• Command: cd cmd/ingestor && go test -race -count=1 -timeout 60m ./....
• Result: ok github.com/corescope/ingestor 1030.112s, no data races.
• The default 10-minute timeout is too short for -race on this 4-core sandbox: TestNeighborEdgesBuilderDeltaScan alone ran for 4 minutes. CI runs the ingestor without -race, with -timeout 20m.

Mutation checks (each run through the IATA tests, then restored)

Mutant	Result
No cap	Killed by 4 tests
Overflow silent	Killed by 3 tests
No reclamation	Killed by 2 tests
%s instead of %q	Killed by 2 tests
No truncation	Killed by 2 tests
Sweep on every drop at the cap	Killed: 99,488 sweeps for 100,000 drops

Performance (BenchmarkIATADropThrottleHostileAtCap, a new region on every drop with the table full, 4 cores)

Variant	ns/op	allocs/op
This branch	83.25 / 81.22 / 85.85	0
Sweep on every drop (upstream's shape)	13,574 / 14,966 / 23,765	0

Allowed traffic pays nothing: the warning runs only on the reject branch.

Static checks. go vet ./... is clean, and gofmt -l is clean on the touched files. Other ingestor files that are not gofmt-clean were so already.

Not verified

• No live broker or staging run. No production configuration change is part of this.
• What the warning volume looks like with a real foreign feed has not been observed.

Overlap with my other open PRs

• PR fix(packets): empty observer/type selections on Clear Filters #132 and PR fix(analytics): treat the distance index's 202 as a transient building state #133 are frontend-only.
• This PR is the only one touching cmd/ingestor/config.go, cmd/ingestor/main.go and config.example.json.
• My planned PR for ingestor: assign explicit, collision-resistant MQTT client IDs #118 (MQTT client IDs) will also touch cmd/ingestor/config.go and config.example.json, in different places.

🤖 Generated with Claude Code

https://claude.ai/code/session_019TcZHooUiiknVWbECVWzk8

---

Generated by Claude Code

[dborup]

Independent review of d9f4c328

Verdict: APPROVE with nits. This is a recommendation only; merging is the owner's call.

Reviewed head: d9f4c3282d300656be0ac71e1b6d0f22a0fb8ef2 (unchanged before and after the review). Work done on a git archive of the PR head, commit A (9e68cd44) and origin/master.

Labels: [F] freshly verified by me · [T] taken from the PR text · [A] assumption · [K] known limitation.

Findings

1. P3, test gap. The refresh of the oldest lower bound after a sweep is not covered. Removing t.oldest = oldest (cmd/ingestor/iata_drop_warn.go:97) keeps every IATA test green [F]. That mutant brings back the per-drop full-table scan that the PR sets out to avoid.
   • Reviewer probe: 30,000 new codes, 1 ms apart, over three 10 s intervals. The head does 1,024 sweeps. The mutant does 20,000, one per drop at the cap after the first interval [F].
   • TestIATADropThrottleSweepsAreAmortized and BenchmarkIATADropThrottleHostileAtCap both stay inside one interval: the benchmark advances 1 ns per op against a 1 h interval. They cannot see this.
   • Suggestion: extend the amortization test across two or more intervals and assert on the sweep count.
2. P3, test gap. The overflow line's escaping is not covered. Changing %q to %s in the overflow log.Printf (iata_drop_warn.go:127) survives [F]. TestIATAWhitelistDropWarningIsOneEscapedBoundedLine only exercises the per-region line. The overflow path is exactly where a hostile publisher sits, because it is only reached after 512 distinct codes. The code is correct today (%q).
3. P3, test gap. Re-arming the overflow throttle after the interval is not covered. A mutant that makes the overflow warning fire once and never again (drop the now.Sub(t.overflowLast) < interval part, iata_drop_warn.go:100) survives [F].
   • That would make beyond-the-cap traffic silent after the first line, which is the acceptance criterion "Traffic beyond the bound is not silently ignored".
   • The code is correct today.
4. nit. The throttle is shared by all MQTT sources (one per Config). The log line's [tag] names only the first source that dropped a region. Drops of the same region from another source inside the interval are not logged. That matches "per code" in the issue, but an operator reading the log could attribute the drop to the wrong feed [F] (code reading).
5. nit. normalizeIATAForWarn upper-cases the whole segment before cutting it to 32 bytes (iata_drop_warn.go:61). A long hostile topic segment therefore costs O(len) and an allocation per drop. IsObserverIATAAllowed already does the same ToUpper(TrimSpace(...)) on every message, so this doubles an existing cost rather than adding a new kind. Truncating first would remove it. Not measured [A].
6. Observation. Test-first holds for the integration test (iata_drop_log_test.go in commit A). The unit tests in iata_drop_warn_test.go arrive with the fix, which is unavoidable because they test new, unexported symbols.

Metadata

• [F] Both commits (9e68cd44, d9f4c328) have author and committer dborup <kontakt@meshview.dk>.
• [F] Base is 85bfee49. origin/master has since moved to d264716c (feat(channels): show observed path hash sizes #131). merge-tree against the current master has no conflicts.
• [F] No closing keywords ("Relates to ingestor: warn when observerIATAWhitelist drops a region without unbounded throttle state #110"); closingIssuesReferences = 0; no @mentions; no full upstream URLs (only Kpa-clawbot/CoreScope#2067 in code format).
• [F] .github/workflows/deploy.yml is not changed. The new Go tests run under the existing go test ./....
• [F] Overlap with the other batch PRs (fix(packets): empty observer/type selections on Clear Filters #132, fix(analytics): treat the distance index's 202 as a transient building state #133, fix(live): wire every persisted view toggle before Live init awaits #135): no shared files. A stacked merge-tree master → fix(packets): empty observer/type selections on Clear Filters #132 → fix(analytics): treat the distance index's 202 as a transient building state #133 → fix(ingestor): warn, throttled and bounded, when observerIATAWhitelist drops a region #134 → fix(live): wire every persisted view toggle before Live init awaits #135 has no conflicts. The PR text notes that the planned ingestor: assign explicit, collision-resistant MQTT client IDs #118 PR will also touch cmd/ingestor/config.go and config.example.json [T].
• [F] CI on d9f4c328 is complete: Go, Playwright and Docker are SUCCESS; the rest are SKIPPED.

Acceptance criteria (issue #110)

Criterion	Result
Fixed, grep-friendly warning with the normalized code	Met [F]: [region-filter] dropping region "GOT"; whitelist check and warning use the same ToUpper(TrimSpace) normalization
Throttled per code, configurable interval, safe default	Met [F]: tests, and mutants M7 and M9 are caught
Per-code state strictly bounded	Met [F]: 512 keys of at most 32 bytes; mutants M1 and M5 are caught
Beyond the cap: bounded, shared overflow warning	Met in code [F]. Tests cover the first overflow line, but not its escaping (finding 2) or re-arming (finding 3)
Expired entries reclaimed	Met [F]: mutant M3 is caught. The amortization bound is not covered across intervals (finding 1)
Allowed traffic and an empty whitelist unchanged	Met [F]: the warning only runs on the existing reject branch (main.go:687); TestIATAWhitelistAllowedAndEmptyAreSilent and the existing whitelist tests pass
Tests for throttling, expiry, more attacker values than the cap, overflow, reclamation	Met, with the gaps above
Ingestor suite with -race	Met [F]: go test -race -count=1 ./... gives ok github.com/corescope/ingestor 420.080s, 0 data races
Optional key documented	Met [F]: config.example.json

Test-first and mutants

• [F] Commit A: 4 of the 5 new integration tests fail ("got 0" lines); the allowed/empty control passes. Head: all 21 IATA tests pass.
• [F] Twelve mutants, nine caught:

Mutant	Result
M1 no cap	4 fail
M2 overflow silent	3 fail
M3 no reclamation	2 fail
M4 %s in the per-region line	2 fail
M4b %s in the overflow line	survives (finding 2)
M5 no truncation	2 fail
M6 main.go does not call the warning	4 fail
M7 no per-region throttle	5 fail
M8 oldest not refreshed after a sweep	survives (finding 1)
M9 interval ignores config	2 fail
M10 overflow never re-arms	survives (finding 3)

Performance

• [F] BenchmarkIATADropThrottleHostileAtCap: 27.3 / 27.5 / 28.0 ns/op, 0 B/op, 0 allocs/op (12-core arm64, so faster than the PR's 4-core numbers). This confirms O(1) per drop within one interval.
• [F] Across intervals the sweep is amortized, not O(1): 1,024 sweeps per 30,000 hostile drops in the probe above. Each sweep removes at least the expired oldest entry, so the worst case is on the order of 512 sweeps × 512 entries per interval. That is harmless at a 6 h default.
• [F] Allowed traffic pays nothing: time.Now() and the throttle run only on the reject branch.

Security and invariants

• [F] Log injection: both lines use %q. The per-region line is tested; the overflow line is not (finding 2).
• [F] Bounded structures: map[string]time.Time with at most 512 keys of at most 32 bytes. No new map[string]interface{}.
• [F] The change is only in cmd/ingestor/. cmd/server is untouched, so the mode=ro invariant holds.
• [F] go vet is clean; gofmt -l on the touched files is clean.

Not verified

• A live broker, a real foreign feed and staging. The PR's "Not verified" list says the same, and it is honest.
• The effect of config hot-reload (if any) on the throttle state. I did not look for a reload path [A].

[dborup]

I found one reproducible issue that should be fixed before merge:

[P2] A large positive iataWarnIntervalSec can disable throttling through time.Duration overflow.

IATAWarnInterval() currently returns time.Duration(c.IATAWarnIntervalSec) * time.Second. On 64-bit builds, a valid JSON integer above 9,223,372,036 seconds can overflow to a negative duration. I reproduced this with 10_000_000_000, which becomes -2346317h47m53.709551616s. With a negative interval, now.Sub(last) < interval is always false, so every rejected MQTT packet can emit a warning. That defeats the PR's bounded/throttled logging guarantee and can cause a log flood after a configuration typo.

Please validate before multiplication and either reject the value, fall back to the default, or clamp it to the maximum representable duration. A regression test should load the large value through LoadConfig and verify that a second drop is still suppressed.

Everything else I checked was clean: the current merge result is conflict-free; the focused IATA tests pass normally and 5x under -race; go vet ./... and git diff --check pass. I found no other merge blockers.