From 291393dcc0ad996bc510581ffd98ca488adf98b0 Mon Sep 17 00:00:00 2001 From: efiten Date: Wed, 23 Sep 2026 10:58:27 +0200 Subject: [PATCH] feat(ingestor): log a throttled warning when the IATA whitelist drops a region (#2067) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Takes over #2008 by @nullrouten0, as offered there on 2026-09-16 and 2026-09-17. **Their commit is the first of the two here, unchanged and under their authorship**; the second is only the fix for the one blocker. Closing #2008 in favour of this so the rebase and the fix travel together, not to reassign the work. ## The feature, unchanged `observerIATAWhitelist` dropped non-whitelisted regions silently. An allow-list fails in the dangerous direction: a legitimate but unlisted region vanishes with nothing to show for it. One line per dropped region now, re-logged at most every `iataWarnIntervalSec` (new optional key, default 6h) for as long as that region keeps arriving. The periodic re-log rather than a strict log-once is the author's call and it is the right one: a single edge event rolls out of any scrape window, leaving an actively-dropping region indistinguishable from a healthy one. ## The blocker, now fixed `ShouldWarnIATADrop` keyed its throttle map on a topic segment the **publisher** controls and never evicted it — a remote memory sink. Measured on the original branch: 200,000 distinct codes retained 200,000 entries and 15.1 MB of heap. My review offered two shapes. This takes the cap rather than shape-validation, and the reason matters: **nothing in this codebase constrains an IATA code's shape.** It is uppercased and trimmed in `config.go` and `db.go` and never validated. Rejecting by shape would invent a rule operators have not agreed to, and would silently drop the warning for anyone whose code does not fit it — the same failure mode, one level down. So `iataWarnMaxTracked = 512`: far above any real deployment (the reference instance runs 43 observers across a handful of regions) and small enough that a hostile feed gains nothing. **Past the cap the drop is still logged**, throttled on one shared timestamp instead of a per-code one. Swallowing it there would reintroduce exactly the silent failure this feature exists to fix. ## Tests The author's `iata_drop_warn_test.go` plus three: - the map stops growing when fed 2048 distinct codes - a new code past the cap still warns once, is then throttled, and speaks again after the interval elapses - an already-tracked code's throttling is unchanged, so the cap does not alter the normal path ## Verification `gofmt` clean, cherry-picked cleanly onto current master (`cmd/ingestor/main.go` auto-merged). Go tests not run locally: no cgo toolchain here since #1992, and per AGENTS.md `CGO_ENABLED=0` links a stub that proves nothing. CI is their first run. ## Not done The `iataWarnIntervalSec` key is undocumented outside the struct comment. If there is a config reference that should list it, say where and I will add it. --------- Co-authored-by: nullrouten Co-authored-by: Claude Opus 5 (1M context) --- cmd/ingestor/config.go | 98 ++++++++++++++ cmd/ingestor/iata_drop_warn_test.go | 196 ++++++++++++++++++++++++++++ cmd/ingestor/main.go | 8 ++ 3 files changed, 302 insertions(+) create mode 100644 cmd/ingestor/iata_drop_warn_test.go diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 439c306f..e892a14e 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -8,6 +8,7 @@ import ( "os" "strings" "sync" + "time" "github.com/meshcore-analyzer/dbconfig" "github.com/meshcore-analyzer/geofilter" @@ -80,6 +81,17 @@ type Config struct { obsIATAWhitelistCached map[string]bool obsIATAWhitelistOnce sync.Once + // IATAWarnIntervalSec throttles the one-line-per-region warning emitted when + // ObserverIATAWhitelist rejects a region. 0 => defaultIATAWarnIntervalSec. + IATAWarnIntervalSec int `json:"iataWarnIntervalSec,omitempty"` + + // iataWarnLast tracks when each dropped region was last logged. + iataWarnMu sync.Mutex + iataWarnLast map[string]time.Time + // iataWarnOverflowLast throttles the shared warning used once + // iataWarnLast has reached iataWarnMaxTracked. + iataWarnOverflowLast time.Time + // ObserverBlacklist is a list of observer public keys to drop at ingest. // Messages from blacklisted observers are silently discarded — no DB writes, // no UpsertObserver, no observations, no metrics. @@ -385,6 +397,92 @@ func (c *Config) IsObserverIATAAllowed(iata string) bool { return c.obsIATAWhitelistCached[strings.ToUpper(strings.TrimSpace(iata))] } +// iataWarnMaxTracked bounds the per-region throttle map. The key is +// publisher-controlled, so it cannot be allowed to grow with traffic. 512 is +// far above any real deployment's region count (the reference instance has 43 +// observers across a handful of regions) and small enough that a hostile feed +// buys nothing: the map stops growing and the warning keeps coming. +const iataWarnMaxTracked = 512 + +// defaultIATAWarnIntervalSec is the re-log interval for whitelist drops (6h). +const defaultIATAWarnIntervalSec = 21600 + +// IATAWarnInterval returns how often a dropped region is re-logged. +func (c *Config) IATAWarnInterval() time.Duration { + if c == nil || c.IATAWarnIntervalSec <= 0 { + return defaultIATAWarnIntervalSec * time.Second + } + return time.Duration(c.IATAWarnIntervalSec) * time.Second +} + +// ShouldWarnIATADrop reports whether a whitelist drop for this region should be +// logged now, recording the decision when it returns true. +// +// Logging every dropped message is not an option: a foreign feed runs to +// thousands of messages a day and would flood the container log. But dropping +// in silence is worse — an allow-list fails in the dangerous direction, where a +// legitimate but unlisted region simply vanishes with nothing to show for it. +// So: one line per region, re-logged at most every IATAWarnInterval for as long +// as that region keeps arriving. The re-log is deliberate — a strict log-once +// would emit a single edge event that any scrape window eventually rolls past, +// leaving an actively-dropping region looking identical to a healthy one. +func (c *Config) ShouldWarnIATADrop(iata string) bool { + if c == nil { + return false + } + code := strings.ToUpper(strings.TrimSpace(iata)) + if code == "" { + return false + } + interval := c.IATAWarnInterval() + now := time.Now() + + c.iataWarnMu.Lock() + defer c.iataWarnMu.Unlock() + if last, ok := c.iataWarnLast[code]; ok && now.Sub(last) < interval { + return false + } + if c.iataWarnLast == nil { + c.iataWarnLast = make(map[string]time.Time) + } + // The key comes from a topic segment the publisher controls, so an + // unbounded map here is a remote memory sink: 200k distinct codes retained + // 200k entries and 15.1 MB of heap when this was measured. Nothing in the + // codebase constrains an IATA code's shape (it is uppercased and trimmed, + // never validated), so rejecting by shape would invent a rule operators + // have not agreed to. Bound the map instead. + // + // Past the cap the drop is still logged, throttled on one shared timestamp + // rather than a per-code one. That keeps the flood-protection the feature + // exists for while making the overflow itself visible: silently dropping + // the warning would reintroduce the bug this PR fixes, one level up. + if len(c.iataWarnLast) >= iataWarnMaxTracked { + // Sweep first. An entry older than the interval holds nothing back: + // the code would be re-logged on its next sighting anyway, so dropping + // it changes no behaviour and frees the slot. Without this the first + // iataWarnMaxTracked codes ever seen own the map forever, and a + // legitimate region that starts arriving later is stuck sharing the + // overflow throttle with whatever transient junk got there first. + // Only runs at the cap, so the normal path pays nothing. + for k, t := range c.iataWarnLast { + if now.Sub(t) >= interval { + delete(c.iataWarnLast, k) + } + } + } + if len(c.iataWarnLast) >= iataWarnMaxTracked { + if _, known := c.iataWarnLast[code]; !known { + if now.Sub(c.iataWarnOverflowLast) < interval { + return false + } + c.iataWarnOverflowLast = now + return true + } + } + c.iataWarnLast[code] = now + return true +} + // LoadConfig reads configuration from a JSON file, with env var overrides. // If the config file does not exist, sensible defaults are used (zero-config startup). func LoadConfig(path string) (*Config, error) { diff --git a/cmd/ingestor/iata_drop_warn_test.go b/cmd/ingestor/iata_drop_warn_test.go new file mode 100644 index 00000000..fa0b0bb5 --- /dev/null +++ b/cmd/ingestor/iata_drop_warn_test.go @@ -0,0 +1,196 @@ +package main + +import ( + "fmt" + "testing" + "time" +) + +func TestIngestorIATAWhitelistGate(t *testing.T) { + // Empty whitelist: the facility stays inert, everything passes. + open := &Config{} + for _, code := range []string{"SJC", "PHL", "", "zzz"} { + if !open.IsObserverIATAAllowed(code) { + t.Errorf("empty whitelist should allow %q", code) + } + } + + cfg := &Config{ObserverIATAWhitelist: []string{"SJC", "oak", " MRY "}} + tests := []struct { + iata string + want bool + }{ + {"SJC", true}, + {"sjc", true}, + {"OAK", true}, + {"MRY", true}, + {" mry ", true}, + {"PHL", false}, + {"MCO", false}, + {"", false}, + } + for _, tt := range tests { + if got := cfg.IsObserverIATAAllowed(tt.iata); got != tt.want { + t.Errorf("IsObserverIATAAllowed(%q) = %v, want %v", tt.iata, got, tt.want) + } + } +} + +func TestIngestorIATAWarnInterval(t *testing.T) { + if got := (&Config{}).IATAWarnInterval(); got != 6*time.Hour { + t.Errorf("default interval = %v, want 6h", got) + } + if got := (&Config{IATAWarnIntervalSec: 90}).IATAWarnInterval(); got != 90*time.Second { + t.Errorf("configured interval = %v, want 90s", got) + } + // Negative/zero fall back to the default rather than logging every message. + if got := (&Config{IATAWarnIntervalSec: -5}).IATAWarnInterval(); got != 6*time.Hour { + t.Errorf("negative interval = %v, want 6h", got) + } +} + +func TestIngestorShouldWarnIATADropThrottles(t *testing.T) { + cfg := &Config{ObserverIATAWhitelist: []string{"SJC"}, IATAWarnIntervalSec: 3600} + + if !cfg.ShouldWarnIATADrop("PHL") { + t.Fatal("first drop for a region should warn") + } + if cfg.ShouldWarnIATADrop("PHL") { + t.Error("second drop inside the interval should be suppressed") + } + if cfg.ShouldWarnIATADrop("phl") { + t.Error("case variant should hit the same throttle bucket") + } + // A different region is tracked independently. + if !cfg.ShouldWarnIATADrop("MCO") { + t.Error("a distinct region should warn on its first drop") + } + + // Once the interval elapses the region re-logs, so an ongoing drop stays + // visible to a scraper instead of decaying into silence. + cfg.iataWarnMu.Lock() + cfg.iataWarnLast["PHL"] = time.Now().Add(-2 * time.Hour) + cfg.iataWarnMu.Unlock() + if !cfg.ShouldWarnIATADrop("PHL") { + t.Error("region should re-log after the interval elapses") + } +} + +func TestIngestorShouldWarnIATADropEdges(t *testing.T) { + var nilCfg *Config + if nilCfg.ShouldWarnIATADrop("PHL") { + t.Error("nil config should not warn") + } + cfg := &Config{} + if cfg.ShouldWarnIATADrop(" ") { + t.Error("blank region should not warn") + } +} + +// TestShouldWarnIATADropBoundsItsMap pins the fix for the review blocker on +// #2008: the throttle key is a topic segment the publisher controls, so an +// unbounded map is a remote memory sink. Measured on the original branch, +// 200,000 distinct codes retained 200,000 entries and 15.1 MB of heap. +func TestShouldWarnIATADropBoundsItsMap(t *testing.T) { + c := &Config{} + + // Far more distinct codes than the cap, as a hostile or broken feed would. + for i := 0; i < iataWarnMaxTracked*4; i++ { + c.ShouldWarnIATADrop(fmt.Sprintf("X%05d", i)) + } + + c.iataWarnMu.Lock() + tracked := len(c.iataWarnLast) + c.iataWarnMu.Unlock() + + if tracked > iataWarnMaxTracked { + t.Errorf("tracked %d codes, cap is %d — the map grows with publisher-supplied input", + tracked, iataWarnMaxTracked) + } +} + +// TestShouldWarnIATADropKeepsWarningPastTheCap is the other half: bounding the +// map must not silence the warning, which would reintroduce the silent-drop bug +// this feature exists to fix, one level up. Past the cap the warning is +// throttled on a shared timestamp rather than dropped. +func TestShouldWarnIATADropKeepsWarningPastTheCap(t *testing.T) { + c := &Config{} + for i := 0; i < iataWarnMaxTracked; i++ { + c.ShouldWarnIATADrop(fmt.Sprintf("F%05d", i)) + } + + // A brand new code, with the map already full. + if !c.ShouldWarnIATADrop("ZZZ") { + t.Fatal("a drop past the cap must still warn once, not be swallowed") + } + // ...and the next one is throttled rather than flooding. + if c.ShouldWarnIATADrop("YYY") { + t.Error("a second overflow drop inside the interval must be throttled") + } + + // After the interval elapses, it speaks again. Rewind the shared timestamp + // rather than sleeping. + c.iataWarnMu.Lock() + c.iataWarnOverflowLast = time.Now().Add(-2 * c.IATAWarnInterval()) + c.iataWarnMu.Unlock() + if !c.ShouldWarnIATADrop("WWW") { + t.Error("once the interval has passed the overflow warning must return") + } +} + +// TestShouldWarnIATADropStillThrottlesKnownCodes guards the normal path: the +// cap must not change behaviour for a code already being tracked. +func TestShouldWarnIATADropStillThrottlesKnownCodes(t *testing.T) { + c := &Config{} + if !c.ShouldWarnIATADrop("BRU") { + t.Fatal("first sighting must warn") + } + if c.ShouldWarnIATADrop("BRU") { + t.Error("second sighting inside the interval must be throttled") + } +} + +// TestShouldWarnIATADropReclaimsExpiredSlots covers the flaw the cap had on its +// own: the map never shrank, so the first iataWarnMaxTracked codes ever seen +// owned it forever and a legitimate region arriving later was permanently +// demoted to the shared overflow throttle behind whatever transient junk filled +// it first. +// +// Sweeping expired entries at the cap is semantically free — an entry past the +// interval would be re-logged on its next sighting regardless — so the map ends +// up holding only regions that are actively arriving. +func TestShouldWarnIATADropReclaimsExpiredSlots(t *testing.T) { + c := &Config{} + + // Fill the map with junk, then age all of it past the interval. + for i := 0; i < iataWarnMaxTracked; i++ { + c.ShouldWarnIATADrop(fmt.Sprintf("J%05d", i)) + } + stale := time.Now().Add(-2 * c.IATAWarnInterval()) + c.iataWarnMu.Lock() + for k := range c.iataWarnLast { + c.iataWarnLast[k] = stale + } + c.iataWarnMu.Unlock() + + // A legitimate region shows up now. It must get its own slot, not the + // shared overflow throttle. + if !c.ShouldWarnIATADrop("BRU") { + t.Fatal("a new region must warn once") + } + c.iataWarnMu.Lock() + _, tracked := c.iataWarnLast["BRU"] + size := len(c.iataWarnLast) + c.iataWarnMu.Unlock() + + if !tracked { + t.Error("BRU must occupy its own slot once the stale entries were reclaimed") + } + if size > iataWarnMaxTracked { + t.Errorf("map grew to %d past the cap of %d", size, iataWarnMaxTracked) + } + // And its own throttle applies, rather than the shared one. + if c.ShouldWarnIATADrop("BRU") { + t.Error("a tracked region must be throttled on its own timestamp") + } +} diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 73bf73eb..ed6da8d8 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -823,6 +823,14 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message, // Global observer IATA whitelist: if configured, drop messages from observers // in non-whitelisted IATA regions. Applies to ALL message types (status + packets). if len(parts) > 1 && !cfg.IsObserverIATAAllowed(parts[1]) { + // Throttled to one line per region per cfg.IATAWarnInterval — see + // ShouldWarnIATADrop. Format matches the fleet's Python region filter so + // one scraper regex covers both. + if cfg.ShouldWarnIATADrop(parts[1]) { + code := strings.ToUpper(strings.TrimSpace(parts[1])) + log.Printf("MQTT [%s] [region-filter] dropping unknown region '%s' (not in observerIATAWhitelist) -- further messages from %s suppressed for %.0fh", + tag, code, code, cfg.IATAWarnInterval().Hours()) + } return }