feat(ingestor): log a throttled warning when the IATA whitelist drops a region (#2067)

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 <nullrouten@gmail.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

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) {

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")
+	}
+}

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
 	}