From 9e68cd4438d6464637fcd77a555fec8e3f028cc6 Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 29 Sep 2026 07:48:54 +0000 Subject: [PATCH 1/2] test(ingestor): reproduce silent observerIATAWhitelist drops (#110) handleMessage drops a message whose topic region is not in observerIATAWhitelist without logging anything, so a legitimate region missing from the list loses all its traffic unnoticed. iata_drop_log_test.go drives the real handleMessage and reads the log: one warning per dropped region (normalized code, names the setting), none for repeats, none for allowed traffic or an empty whitelist, one escaped and bounded line for a hostile region segment, and a bounded number of lines, including a shared overflow warning, for thousands of distinct regions. On master the four warning cases fail (0 lines); the allowed/empty control passes. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019TcZHooUiiknVWbECVWzk8 --- cmd/ingestor/iata_drop_log_test.go | 146 +++++++++++++++++++++++++++++ 1 file changed, 146 insertions(+) create mode 100644 cmd/ingestor/iata_drop_log_test.go diff --git a/cmd/ingestor/iata_drop_log_test.go b/cmd/ingestor/iata_drop_log_test.go new file mode 100644 index 000000000..9cf1de7bd --- /dev/null +++ b/cmd/ingestor/iata_drop_log_test.go @@ -0,0 +1,146 @@ +package main + +import ( + "bytes" + "fmt" + "log" + "strings" + "testing" +) + +// #110: when observerIATAWhitelist rejects a region, the drop used to be +// silent, so a legitimate region missing from the list lost all its traffic +// with nothing in the log. These tests drive the real handleMessage and read +// the log it writes. + +// captureLog runs fn with the standard logger writing to a buffer. +func captureLog(t *testing.T, fn func()) string { + t.Helper() + var buf bytes.Buffer + orig, flags := log.Writer(), log.Flags() + log.SetOutput(&buf) + log.SetFlags(0) + defer func() { log.SetOutput(orig); log.SetFlags(flags) }() + fn() + return buf.String() +} + +func regionFilterLines(out string) []string { + var lines []string + for _, l := range strings.Split(out, "\n") { + if strings.Contains(l, "[region-filter]") { + lines = append(lines, l) + } + } + return lines +} + +func statusMsg(region, observer string) *mockMessage { + return &mockMessage{ + topic: "meshcore/" + region + "/" + observer + "/status", + payload: []byte(`{"origin":"n","noise_floor":-110}`), + } +} + +func TestIATAWhitelistDropIsLoggedOncePerRegion(t *testing.T) { + store := newTestStore(t) + cfg := &Config{ObserverIATAWhitelist: []string{"ARN"}} + out := captureLog(t, func() { + for i := 0; i < 5; i++ { + handleMessage(store, "src", MQTTSource{Name: "src"}, statusMsg("got", fmt.Sprintf("obs%d", i)), nil, nil, cfg) + } + }) + lines := regionFilterLines(out) + if len(lines) != 1 { + t.Fatalf("want exactly 1 [region-filter] line for 5 drops of one region, got %d:\n%s", len(lines), out) + } + if !strings.Contains(lines[0], `"GOT"`) || !strings.Contains(lines[0], "observerIATAWhitelist") { + t.Errorf("warning must name the normalized region and the setting: %q", lines[0]) + } + var n int + store.db.QueryRow("SELECT COUNT(*) FROM observers").Scan(&n) + if n != 0 { + t.Errorf("dropped messages must still be dropped, %d observers stored", n) + } +} + +func TestIATAWhitelistDropWarnsPerDistinctRegion(t *testing.T) { + store := newTestStore(t) + cfg := &Config{ObserverIATAWhitelist: []string{"ARN"}} + out := captureLog(t, func() { + for _, r := range []string{"GOT", "MMX", " got ", "CPH", "mmx"} { + handleMessage(store, "src", MQTTSource{Name: "src"}, statusMsg(r, "o"), nil, nil, cfg) + } + }) + lines := regionFilterLines(out) + if len(lines) != 3 { + t.Fatalf("want one line each for GOT, MMX, CPH (codes are normalized), got %d:\n%s", len(lines), out) + } +} + +func TestIATAWhitelistAllowedAndEmptyAreSilent(t *testing.T) { + store := newTestStore(t) + out := captureLog(t, func() { + handleMessage(store, "src", MQTTSource{Name: "src"}, statusMsg("ARN", "a1"), nil, nil, &Config{ObserverIATAWhitelist: []string{"arn"}}) + handleMessage(store, "src", MQTTSource{Name: "src"}, statusMsg("GOT", "a2"), nil, nil, &Config{}) + }) + if lines := regionFilterLines(out); len(lines) != 0 { + t.Fatalf("allowed traffic and an empty whitelist must not warn:\n%s", out) + } + var n int + store.db.QueryRow("SELECT COUNT(*) FROM observers").Scan(&n) + if n != 2 { + t.Errorf("allowed traffic must still be stored, got %d observers", n) + } +} + +// The region is a topic segment the publisher controls: it must not be able +// to forge log lines or make one line arbitrarily long. +func TestIATAWhitelistDropWarningIsOneEscapedBoundedLine(t *testing.T) { + store := newTestStore(t) + cfg := &Config{ObserverIATAWhitelist: []string{"ARN"}} + evil := "XX\nMQTT [src] fake line\r" + strings.Repeat("A", 10000) + out := captureLog(t, func() { + handleMessage(store, "src", MQTTSource{Name: "src"}, statusMsg(evil, "o"), nil, nil, cfg) + }) + lines := regionFilterLines(out) + if len(lines) != 1 { + t.Fatalf("want 1 warning line, got %d:\n%.500s", len(lines), out) + } + if strings.Count(out, "\n") != 1 { + t.Errorf("the warning spans %d lines; control characters must be escaped", strings.Count(out, "\n")) + } + if len(lines[0]) > 300 { + t.Errorf("warning line is %d bytes; the region must be truncated", len(lines[0])) + } +} + +// Many distinct attacker-chosen regions: the warning keeps coming (never +// silent), but through a shared overflow path, so the number of lines stays +// bounded well below the number of regions. +func TestIATAWhitelistDropManyDistinctRegionsStaysBoundedAndVisible(t *testing.T) { + store := newTestStore(t) + cfg := &Config{ObserverIATAWhitelist: []string{"ARN"}} + const distinct = 3000 + out := captureLog(t, func() { + for i := 0; i < distinct; i++ { + handleMessage(store, "src", MQTTSource{Name: "src"}, statusMsg(fmt.Sprintf("Z%05d", i), "o"), nil, nil, cfg) + } + }) + lines := regionFilterLines(out) + if len(lines) == 0 { + t.Fatal("drops beyond the throttle bound must not be silent") + } + if len(lines) >= distinct/2 { + t.Fatalf("%d warning lines for %d regions: the per-region throttle must be bounded", len(lines), distinct) + } + overflow := 0 + for _, l := range lines { + if strings.Contains(l, "throttle table full") { + overflow++ + } + } + if overflow != 1 { + t.Errorf("want exactly one shared overflow warning within the interval, got %d", overflow) + } +} From d9f4c3282d300656be0ac71e1b6d0f22a0fb8ef2 Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 29 Sep 2026 08:03:19 +0000 Subject: [PATCH 2/2] fix(ingestor): warn, throttled and bounded, when observerIATAWhitelist drops a region (#110) A message whose topic region is not in observerIATAWhitelist is now logged once per region and re-logged at most every iataWarnIntervalSec (default 6h) while that region keeps arriving: MQTT [src] [region-filter] dropping region "GOT": not in observerIATAWhitelist; further messages from this region suppressed for 6h0m0s The region is a topic segment the publisher controls, so: - the per-region state is bounded to 512 keys of at most 32 bytes; - beyond that, drops share one overflow warning with the same interval (never silent); - entries older than the interval are reclaimed when the table is full, so the first codes seen cannot own it; - a sweep runs only when an entry can have expired (a lower bound on the oldest entry), so a hostile feed at the cap costs O(1) per drop: about 83 ns/op, against 13-24 us/op with a sweep on every drop; - the code is quoted with %q and truncated, so it cannot forge or stretch log lines. Allowed traffic and an empty whitelist are unchanged. The optional key is documented in config.example.json. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019TcZHooUiiknVWbECVWzk8 --- cmd/ingestor/config.go | 8 ++ cmd/ingestor/iata_drop_warn.go | 133 ++++++++++++++++++ cmd/ingestor/iata_drop_warn_test.go | 209 ++++++++++++++++++++++++++++ cmd/ingestor/main.go | 2 + config.example.json | 4 +- 5 files changed, 355 insertions(+), 1 deletion(-) create mode 100644 cmd/ingestor/iata_drop_warn.go create mode 100644 cmd/ingestor/iata_drop_warn_test.go diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 9b0c08581..3c4082416 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -71,6 +71,14 @@ type Config struct { obsIATAWhitelistCached map[string]bool obsIATAWhitelistOnce sync.Once + // IATAWarnIntervalSec is how often a region dropped by + // ObserverIATAWhitelist is re-logged while it keeps arriving (#110). + // 0 or less means the default, 6 hours. See iata_drop_warn.go. + IATAWarnIntervalSec int `json:"iataWarnIntervalSec,omitempty"` + + // iataDropWarn is the bounded per-region throttle for that warning. + iataDropWarn iataDropThrottle + // 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. diff --git a/cmd/ingestor/iata_drop_warn.go b/cmd/ingestor/iata_drop_warn.go new file mode 100644 index 000000000..6451da0d7 --- /dev/null +++ b/cmd/ingestor/iata_drop_warn.go @@ -0,0 +1,133 @@ +package main + +import ( + "log" + "strings" + "sync" + "time" + "unicode/utf8" +) + +// Throttled warning for observerIATAWhitelist drops (#110). +// +// Dropping in silence fails in the dangerous direction: a legitimate region +// missing from the allow-list loses all its traffic with nothing in the log. +// Logging every drop is not an option either (a foreign feed is thousands of +// messages a day), so each region is logged once and re-logged at most every +// IATAWarnInterval while it keeps arriving; a strict log-once would scroll +// out of any log window and leave an active drop looking healthy. +// +// The region is a topic segment the publisher controls, so the per-region +// state is strictly bounded: at most iataWarnMaxTracked keys of at most +// iataWarnMaxKeyLen bytes. Past the cap the drop is still logged, on one +// shared overflow throttle. Entries older than the interval are reclaimed +// when the table is full, so the first codes ever seen cannot own it. + +const ( + // iataWarnMaxTracked bounds the per-region throttle table. Far above any + // real deployment's region count, small enough to be harmless when a + // hostile publisher sends a new region per message. + iataWarnMaxTracked = 512 + // iataWarnMaxKeyLen caps the stored and logged region code. IATA codes + // are three letters; longer segments are truncated (and may share a key). + iataWarnMaxKeyLen = 32 + // defaultIATAWarnIntervalSec is the default re-log interval (6h). + defaultIATAWarnIntervalSec = 6 * 60 * 60 +) + +// iataDropThrottle is the per-Config throttle state. The zero value is ready. +type iataDropThrottle struct { + mu sync.Mutex + last map[string]time.Time + overflowLast time.Time + // oldest is a lower bound on the oldest time in last: a sweep can free + // nothing until it has expired, so a full table of fresh entries costs + // O(1) per drop, not a full scan. + oldest time.Time + sweeps int // full-table sweeps, for tests +} + +// 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 +} + +// normalizeIATAForWarn is the key and the logged form of a region segment: +// trimmed, upper-cased and cut to iataWarnMaxKeyLen bytes on a rune boundary. +func normalizeIATAForWarn(iata string) string { + code := strings.ToUpper(strings.TrimSpace(iata)) + if len(code) <= iataWarnMaxKeyLen { + return code + } + cut := iataWarnMaxKeyLen + for cut > 0 && !utf8.RuneStart(code[cut]) { + cut-- + } + return code[:cut] +} + +// shouldWarn reports whether a drop of code (already normalized) should be +// logged at now, and whether it goes through the shared overflow throttle. +func (t *iataDropThrottle) shouldWarn(code string, now time.Time, interval time.Duration) (warn, overflow bool) { + t.mu.Lock() + defer t.mu.Unlock() + if last, ok := t.last[code]; ok { + if now.Sub(last) < interval { + return false, false + } + t.last[code] = now + return true, false + } + if t.last == nil { + t.last = make(map[string]time.Time) + } + if len(t.last) >= iataWarnMaxTracked && now.Sub(t.oldest) >= interval { + t.sweeps++ + oldest := now + for k, ts := range t.last { + if now.Sub(ts) >= interval { + delete(t.last, k) + } else if ts.Before(oldest) { + oldest = ts + } + } + t.oldest = oldest + } + if len(t.last) >= iataWarnMaxTracked { + if !t.overflowLast.IsZero() && now.Sub(t.overflowLast) < interval { + return false, true + } + t.overflowLast = now + return true, true + } + if len(t.last) == 0 || now.Before(t.oldest) { + t.oldest = now + } + t.last[code] = now + return true, false +} + +// warnIATADrop logs a throttled warning for a message dropped by +// observerIATAWhitelist. The region is quoted (%q), so control characters in +// the publisher-controlled segment cannot break or forge log lines. +func (c *Config) warnIATADrop(tag, iata string, now time.Time) { + if c == nil { + return + } + code := normalizeIATAForWarn(iata) + interval := c.IATAWarnInterval() + warn, overflow := c.iataDropWarn.shouldWarn(code, now, interval) + if !warn { + return + } + if overflow { + log.Printf("MQTT [%s] [region-filter] dropping region %q: not in observerIATAWhitelist; throttle table full (%d regions), further untracked regions suppressed for %s", + tag, code, iataWarnMaxTracked, interval) + return + } + log.Printf("MQTT [%s] [region-filter] dropping region %q: not in observerIATAWhitelist; further messages from this region suppressed for %s", + tag, code, interval) +} diff --git a/cmd/ingestor/iata_drop_warn_test.go b/cmd/ingestor/iata_drop_warn_test.go new file mode 100644 index 000000000..9ea129312 --- /dev/null +++ b/cmd/ingestor/iata_drop_warn_test.go @@ -0,0 +1,209 @@ +package main + +import ( + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +// Unit tests for the bounded observerIATAWhitelist drop throttle (#110), +// with an explicit clock. iata_drop_log_test.go covers the log output +// through handleMessage. + +var t0IATA = time.Date(2026, 9, 29, 12, 0, 0, 0, time.UTC) + +func TestIATAWarnIntervalDefaultAndConfigured(t *testing.T) { + if got := (&Config{}).IATAWarnInterval(); got != 6*time.Hour { + t.Errorf("default interval %v, want 6h", got) + } + if got := (&Config{IATAWarnIntervalSec: -5}).IATAWarnInterval(); got != 6*time.Hour { + t.Errorf("negative interval %v, want the default", got) + } + if got := (&Config{IATAWarnIntervalSec: 90}).IATAWarnInterval(); got != 90*time.Second { + t.Errorf("configured interval %v, want 90s", got) + } + var nilCfg *Config + if got := nilCfg.IATAWarnInterval(); got != 6*time.Hour { + t.Errorf("nil config interval %v", got) + } +} + +func TestIATAWarnIntervalFromJSON(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.json") + if err := os.WriteFile(path, []byte(`{"observerIATAWhitelist":["ARN"],"iataWarnIntervalSec":600}`), 0o600); err != nil { + t.Fatal(err) + } + cfg, err := LoadConfig(path) + if err != nil { + t.Fatal(err) + } + if cfg.IATAWarnInterval() != 10*time.Minute { + t.Errorf("iataWarnIntervalSec not read: %v", cfg.IATAWarnInterval()) + } +} + +func TestIATADropThrottlePerRegionAndExpiry(t *testing.T) { + var th iataDropThrottle + iv := time.Hour + if w, o := th.shouldWarn("GOT", t0IATA, iv); !w || o { + t.Fatalf("first drop: warn=%v overflow=%v", w, o) + } + if w, _ := th.shouldWarn("GOT", t0IATA.Add(59*time.Minute), iv); w { + t.Error("repeat within the interval must be suppressed") + } + if w, _ := th.shouldWarn("MMX", t0IATA.Add(time.Minute), iv); !w { + t.Error("another region has its own throttle") + } + if w, _ := th.shouldWarn("GOT", t0IATA.Add(time.Hour), iv); !w { + t.Error("re-log once the interval has passed") + } + if w, _ := th.shouldWarn("GOT", t0IATA.Add(time.Hour+time.Second), iv); w { + t.Error("the re-log restarts the interval") + } +} + +func TestNormalizeIATAForWarn(t *testing.T) { + for in, want := range map[string]string{ + " got ": "GOT", + "cph": "CPH", + strings.Repeat("a", 100): strings.Repeat("A", iataWarnMaxKeyLen), + } { + if got := normalizeIATAForWarn(in); got != want { + t.Errorf("normalize(%q) = %q, want %q", in, got, want) + } + } + // a multi-byte rune straddling the cut is dropped, not split + in := strings.Repeat("A", iataWarnMaxKeyLen-1) + "Ø" + got := normalizeIATAForWarn(in) + if len(got) > iataWarnMaxKeyLen || !strings.HasPrefix(got, strings.Repeat("A", iataWarnMaxKeyLen-1)) || strings.ContainsRune(got, '�') { + t.Errorf("normalize(%q) = %q", in, got) + } +} + +// More distinct publisher-chosen codes than the cap: the table never grows +// past it, drops beyond it warn through one shared throttle, and that shared +// warning is re-armed after the interval. +func TestIATADropThrottleIsBoundedWithSharedOverflow(t *testing.T) { + var th iataDropThrottle + iv := time.Hour + now := t0IATA + own, overflowWarns, silent := 0, 0, 0 + for i := 0; i < 20000; i++ { + w, o := th.shouldWarn(fmt.Sprintf("Z%05d", i), now, iv) + switch { + case w && !o: + own++ + case w && o: + overflowWarns++ + default: + silent++ + } + if len(th.last) > iataWarnMaxTracked { + t.Fatalf("throttle table grew to %d entries", len(th.last)) + } + } + if own != iataWarnMaxTracked || overflowWarns != 1 || silent != 20000-iataWarnMaxTracked-1 { + t.Fatalf("own=%d overflow=%d silent=%d", own, overflowWarns, silent) + } + // tracked codes keep their own throttle while the table is full + if w, o := th.shouldWarn("Z00000", now.Add(iv), iv); !w || o { + t.Errorf("a tracked code re-logs on its own slot: warn=%v overflow=%v", w, o) + } +} + +// Once the entries are older than the interval, a new region reclaims a +// slot instead of the first codes ever seen owning the table forever. +func TestIATADropThrottleReclaimsExpiredSlots(t *testing.T) { + var th iataDropThrottle + iv := time.Hour + for i := 0; i < iataWarnMaxTracked; i++ { + th.shouldWarn(fmt.Sprintf("OLD%04d", i), t0IATA, iv) + } + if w, o := th.shouldWarn("EARLY", t0IATA.Add(time.Minute), iv); !w || !o { + t.Fatalf("full table of fresh entries: warn=%v overflow=%v, want the overflow warning", w, o) + } + later := t0IATA.Add(iv) + if w, o := th.shouldWarn("NEW", later, iv); !w || o { + t.Fatalf("expired table: warn=%v overflow=%v, want an own slot", w, o) + } + if _, ok := th.last["NEW"]; !ok || len(th.last) != 1 { + t.Fatalf("after reclaiming: %d entries, NEW tracked=%v", len(th.last), ok) + } + if w, _ := th.shouldWarn("NEW", later.Add(time.Minute), iv); w { + t.Error("the reclaimed slot throttles NEW on its own") + } +} + +// A hostile feed at the cap must not trigger a full-table scan per message: +// sweeps run only when an entry can actually have expired. +func TestIATADropThrottleSweepsAreAmortized(t *testing.T) { + var th iataDropThrottle + iv := time.Hour + now := t0IATA + for i := 0; i < 100000; i++ { + th.shouldWarn(fmt.Sprintf("H%06d", i), now.Add(time.Duration(i)*time.Millisecond), iv) + } + // 100000 drops over 100s, all within one interval: at most one sweep + // (when the table first fills; nothing can have expired after that). + if th.sweeps > 1 { + t.Fatalf("%d full-table sweeps for 100000 drops within one interval", th.sweeps) + } + // after the interval the next new code sweeps once and reclaims + th.shouldWarn("LATE", now.Add(2*iv), iv) + if th.sweeps > 2 || len(th.last) != 1 { + t.Fatalf("sweeps=%d entries=%d after expiry", th.sweeps, len(th.last)) + } +} + +func TestIATADropThrottleConcurrent(t *testing.T) { + var th iataDropThrottle + iv := time.Hour + var wg sync.WaitGroup + var mu sync.Mutex + warns := 0 + for g := 0; g < 16; g++ { + wg.Add(1) + go func(g int) { + defer wg.Done() + for i := 0; i < 500; i++ { + code := "SAME" + if i%2 == 1 { + code = fmt.Sprintf("G%02dI%03d", g, i) + } + if w, _ := th.shouldWarn(code, t0IATA, iv); w && code == "SAME" { + mu.Lock() + warns++ + mu.Unlock() + } + } + }(g) + } + wg.Wait() + if warns != 1 { + t.Fatalf("SAME warned %d times from 16 goroutines within one interval", warns) + } + if len(th.last) > iataWarnMaxTracked { + t.Fatalf("%d entries", len(th.last)) + } +} + +func BenchmarkIATADropThrottleHostileAtCap(b *testing.B) { + var th iataDropThrottle + iv := time.Hour + codes := make([]string, 4096) + for i := range codes { + codes[i] = fmt.Sprintf("B%05d", i) + } + for i := 0; i < iataWarnMaxTracked; i++ { + th.shouldWarn(codes[i], t0IATA, iv) + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + th.shouldWarn(codes[i%len(codes)], t0IATA.Add(time.Duration(i)), iv) + } +} diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 3ab0626b1..db2e816f3 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -683,6 +683,8 @@ 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]) { + // Not silently (#110): a throttled, bounded warning per region. + cfg.warnIATADrop(tag, parts[1], time.Now()) return } diff --git a/config.example.json b/config.example.json index b0282f606..ce4bd1f92 100644 --- a/config.example.json +++ b/config.example.json @@ -6,7 +6,9 @@ "hiddenNamePrefixes": ["🚫"], "_comment_hiddenNamePrefixes": "Node name prefixes that mark a node as hidden from this dashboard (#1181). Mirrors a convention used by other MeshCore map dashboards: an operator who wants their node hidden renames it to start with one of these prefixes and sends an advert; the next advert is dropped from /api/nodes, /api/nodes/search and /api/nodes/{pubkey}. DB rows are preserved so observation history (paths, hops, distances) stays intact for analytics. The node is NOT hidden from the mesh itself — only from this dashboard. Set to [] to disable. Default: [\"🚫\"].", "observerIATAWhitelist": [], - "_comment_observerIATAWhitelist": "Global IATA region whitelist. When non-empty, only observers whose IATA code (from MQTT topic) matches are processed. Case-insensitive. Empty = allow all. Unlike per-source iataFilter, this applies across all MQTT sources.", + "_comment_observerIATAWhitelist": "Global IATA region whitelist. When non-empty, only observers whose IATA code (from MQTT topic) matches are processed. Case-insensitive. Empty = allow all. Unlike per-source iataFilter, this applies across all MQTT sources. A dropped region is logged by the ingestor as one '[region-filter] dropping region \"XYZ\"' line, repeated at most every iataWarnIntervalSec while it keeps arriving.", + "iataWarnIntervalSec": 21600, + "_comment_iataWarnIntervalSec": "Optional (ingestor). Seconds between repeated '[region-filter]' warnings for a region dropped by observerIATAWhitelist. Default 21600 (6h); 0 or omitted = default. At most 512 regions are tracked individually (region codes come from the MQTT topic, so this is bounded); beyond that, drops share one overflow warning with the same interval.", "retention": { "nodeDays": 7, "observerDays": 14,