From 40f804d88343bd063ebed2cf539d0d6df5e103f3 Mon Sep 17 00:00:00 2001 From: efiten Date: Wed, 2 Sep 2026 09:11:05 +0200 Subject: [PATCH 1/2] fix(ingestor): a retained status message is not observer liveness (#1885) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem The broker replays every retained `status` message on subscribe, so each ingestor restart pushes all of them through the status path in `handleMessage`. That path stamps `last_seen` with `time.Now()` (deliberately, per #1465). Observed on a live deployment on 2026-08-11: **23 observers all carried `last_seen = 2026-08-06T08:20:09Z`** — 15 seconds after container start — and 18 of them had sent no actual packet in over a month. Their retained publish dates lined up almost 1:1 with `last_packet_at`, i.e. the replay was their only sign of "life": | observer | last real packet | retained status published | |---|---|---| | ON8AR - Observer | never | 2026-03-20 | | BE-BGS-RRY120-RES | never | 2026-04-02 | | A3BEF374 | 2026-04-30 | 2026-04-30 | | BE-BGS-RRY120-RUDY | 2026-05-18 | 2026-05-18 | | BE-JBE-ETG-O1 | 2026-06-10 | 2026-06-10 | That makes dead observers immortal, three ways per restart: 1. `last_seen` jumps forward, so `RemoveStaleObservers` can never age them out as long as a restart happens inside `observerDays`. 2. The unconditional `inactive = 0` reactivation at the end of `UpsertObserverAt` undoes any soft-delete that did land. 3. A metrics sample is filed at ingest time, dating a months-old reading as a present-tense measurement. `UpsertObserverAt`'s docstring already claimed retained replays were a no-op for `last_seen` thanks to the `MAX` guard. That held only while the caller passed the envelope timestamp; #1465 switched it to ingest time, which defeats the guard. ## Fix The retained path now updates metadata only, via a new `UpsertObserverRetained`: - no `last_seen` advance - no `inactive = 0` reactivation - no `packet_count` bump - no metrics sample - **no INSERT** — a retained-only observer the analyzer has never heard from live describes a past that may be months old and does not belong in the list. A live message from the same observer creates the row through the normal path moments later. Live status handling is unchanged. ## Tests Seven tests in `cmd/ingestor/retained_status_test.go`, written before the fix: - 4 that failed on the bug: `last_seen` advance, reactivation of a soft-deleted row, creation of a never-seen observer, metrics-sample insert - 2 regression guards pinning live (non-retained) behaviour: `last_seen` still advances, unknown observer still created - 1 asserting retained metadata is still applied — the snapshot is the observer's last known state, only the liveness signal is suppressed `mockMessage` gained a `retained` field so `Retained()` is controllable. Meta flattening is extracted to `observerMetaColumns` so both write paths bind identical args. Full `cmd/ingestor` suite passes. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5 (1M context) (cherry picked from commit 9bd5f5a3a2647f42091d8cb7a04ac98b7f018d17) Co-Authored-By: Claude Opus 5 --- cmd/ingestor/db.go | 79 ++++++++--- cmd/ingestor/main.go | 16 +++ cmd/ingestor/main_test.go | 7 +- cmd/ingestor/retained_status_test.go | 171 ++++++++++++++++++++++ cmd/ingestor/status_liveness.go | 50 +++++++ cmd/ingestor/status_liveness_test.go | 203 +++++++++++++++++++++++++++ 6 files changed, 507 insertions(+), 19 deletions(-) create mode 100644 cmd/ingestor/retained_status_test.go create mode 100644 cmd/ingestor/status_liveness.go create mode 100644 cmd/ingestor/status_liveness_test.go diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index e37094e00..a6604389d 100644 --- a/cmd/ingestor/db.go +++ b/cmd/ingestor/db.go @@ -1413,8 +1413,68 @@ func (s *Store) UpsertObserverAt(id, name, iata string, meta *ObserverMeta, last } normalizedIATA := strings.TrimSpace(strings.ToUpper(iata)) - var model, firmware, clientVersion, radio interface{} - var batteryMv, uptimeSecs, noiseFloor, canRelay interface{} + model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay := observerMetaColumns(meta) + + _, err := s.stmtUpsertObserver.Exec( + id, name, normalizedIATA, lastSeen, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay, + name, normalizedIATA, ingestNow, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay, + ) + if err != nil { + s.Stats.WriteErrors.Add(1) + return err + } + s.Stats.ObserverUpserts.Add(1) + + // Reactivate if this observer was previously marked inactive + s.db.Exec(`UPDATE observers SET inactive = 0 WHERE id = ? AND inactive = 1`, id) + return nil +} + +// UpsertObserverRetained applies the metadata from a RETAINED status message +// without treating it as a sign of life. The broker replays retained messages +// on every subscribe, so an ingestor restart would otherwise stamp last_seen +// with the restart time for every observer that ever published one — making +// dead observers permanently un-ageable by RemoveStaleObservers, and undoing +// any inactive flag it did manage to set. +// +// Deliberately narrower than UpsertObserverAt: it updates metadata columns on +// an existing row only. It does not advance last_seen, does not clear +// inactive, does not bump packet_count, and does not INSERT — a retained-only +// observer the analyzer has never heard from live describes a past that may be +// months old and does not belong in the list. A live message from the same +// observer arrives moments later and creates the row through the normal path. +func (s *Store) UpsertObserverRetained(id, name, iata string, meta *ObserverMeta) error { + normalizedIATA := strings.TrimSpace(strings.ToUpper(iata)) + model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay := observerMetaColumns(meta) + + _, err := s.db.Exec(` + UPDATE observers SET + name = COALESCE(?, name), + iata = COALESCE(?, iata), + model = COALESCE(?, model), + firmware = COALESCE(?, firmware), + client_version = COALESCE(?, client_version), + radio = COALESCE(?, radio), + battery_mv = COALESCE(?, battery_mv), + uptime_secs = COALESCE(?, uptime_secs), + noise_floor = COALESCE(?, noise_floor), + can_relay = COALESCE(?, can_relay), + can_relay_seen = CASE WHEN ? IS NULL THEN can_relay_seen ELSE 1 END + WHERE id = ?`, + name, normalizedIATA, model, firmware, clientVersion, radio, + batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay, id, + ) + if err != nil { + s.Stats.WriteErrors.Add(1) + return err + } + return nil +} + +// observerMetaColumns flattens an *ObserverMeta into the driver args the +// observer upserts bind. A nil field stays nil so the COALESCE in the SQL +// leaves the existing column untouched (#1290). +func observerMetaColumns(meta *ObserverMeta) (model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay interface{}) { if meta != nil { if meta.Model != nil { model = *meta.Model @@ -1449,20 +1509,7 @@ func (s *Store) UpsertObserverAt(id, name, iata string, meta *ObserverMeta, last } } } - - _, err := s.stmtUpsertObserver.Exec( - id, name, normalizedIATA, lastSeen, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay, - name, normalizedIATA, ingestNow, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay, - ) - if err != nil { - s.Stats.WriteErrors.Add(1) - return err - } - s.Stats.ObserverUpserts.Add(1) - - // Reactivate if this observer was previously marked inactive - s.db.Exec(`UPDATE observers SET inactive = 0 WHERE id = ? AND inactive = 1`, id) - return nil + return model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay } // Close checkpoints the WAL and closes the database. diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index ac102da01..12f96355b 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -632,6 +632,22 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message, name, _ := msg["origin"].(string) iata := parts[1] meta := extractObserverMeta(msg) + // A replayed status message is the broker handing us the observer's + // last published snapshot — it is not evidence the observer is alive + // now. Stamping last_seen from it resurrects dead observers, so the + // replay path only refreshes metadata. + // + // Two ways to recognise one: the retain flag (our own subscribe), and + // the payload itself (a replay that reached us through the mosquitto + // bridge, where the flag does not survive the hop — see + // statusIsLiveness). + if m.Retained() || !statusIsLiveness(msg, time.Now().UTC()) { + if err := store.UpsertObserverRetained(observerID, name, iata, meta); err != nil { + log.Printf("MQTT [%s] retained observer status error: %v", tag, err) + } + log.Print(formatStatusLog(tag, firstNonEmpty(name, observerID), iata)) + return + } // observer.last_seen is "when did the analyzer last hear from this // observer" — fundamentally an ingest-time question. Passing "" makes // UpsertObserverAt use time.Now(), independent of the envelope timestamp diff --git a/cmd/ingestor/main_test.go b/cmd/ingestor/main_test.go index c3a2e0095..817afe7b0 100644 --- a/cmd/ingestor/main_test.go +++ b/cmd/ingestor/main_test.go @@ -105,13 +105,14 @@ func TestUnixTime(t *testing.T) { // mockMessage implements mqtt.Message for testing handleMessage type mockMessage struct { - topic string - payload []byte + topic string + payload []byte + retained bool } func (m *mockMessage) Duplicate() bool { return false } func (m *mockMessage) Qos() byte { return 0 } -func (m *mockMessage) Retained() bool { return false } +func (m *mockMessage) Retained() bool { return m.retained } func (m *mockMessage) Topic() string { return m.topic } func (m *mockMessage) MessageID() uint16 { return 0 } func (m *mockMessage) Payload() []byte { return m.payload } diff --git a/cmd/ingestor/retained_status_test.go b/cmd/ingestor/retained_status_test.go new file mode 100644 index 000000000..64a7ffc47 --- /dev/null +++ b/cmd/ingestor/retained_status_test.go @@ -0,0 +1,171 @@ +package main + +import ( + "testing" + "time" +) + +// A retained MQTT status message is a snapshot the broker replays on every +// subscribe — it says what the observer last published, not that the observer +// is alive now. Treating the replay as liveness makes dead observers immortal: +// last_seen jumps to the ingestor's restart time, RemoveStaleObservers can +// never age them out, and any row it did mark gets reactivated on the next +// restart. Observed on live 2026-08-11: 23 observers, all stamped 15s after +// container start, 18 of them with no real packet in over a month. + +const retainedStatusTopic = "meshcore/LAX/obs-retained/status" + +func retainedStatusMsg(topic string, payload string) *mockMessage { + return &mockMessage{topic: topic, payload: []byte(payload), retained: true} +} + +// seedObserver inserts an observer and forces last_seen to ageDays ago. +func seedObserver(t *testing.T, s *Store, id string, ageDays int) string { + t.Helper() + if err := s.UpsertObserver(id, id, "LAX", nil); err != nil { + t.Fatal(err) + } + old := time.Now().UTC().AddDate(0, 0, -ageDays).Format(time.RFC3339) + if _, err := s.db.Exec(`UPDATE observers SET last_seen = ? WHERE id = ?`, old, id); err != nil { + t.Fatal(err) + } + return old +} + +func observerLastSeen(t *testing.T, s *Store, id string) string { + t.Helper() + var got string + if err := s.db.QueryRow(`SELECT last_seen FROM observers WHERE id = ?`, id).Scan(&got); err != nil { + t.Fatal(err) + } + return got +} + +func TestRetainedStatusDoesNotAdvanceLastSeen(t *testing.T) { + store := newTestStore(t) + want := seedObserver(t, store, "obs-retained", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + retainedStatusMsg(retainedStatusTopic, `{"origin":"obs-retained","noise_floor":-95.5}`), + nil, nil, &Config{}) + + if got := observerLastSeen(t, store, "obs-retained"); got != want { + t.Errorf("last_seen = %s, want %s (retained replay is not liveness)", got, want) + } +} + +func TestRetainedStatusDoesNotReactivateInactiveObserver(t *testing.T) { + store := newTestStore(t) + seedObserver(t, store, "obs-retained", 60) + if _, err := store.db.Exec(`UPDATE observers SET inactive = 1 WHERE id = ?`, "obs-retained"); err != nil { + t.Fatal(err) + } + + handleMessage(store, "test", MQTTSource{Name: "test"}, + retainedStatusMsg(retainedStatusTopic, `{"origin":"obs-retained","noise_floor":-95.5}`), + nil, nil, &Config{}) + + var inactive int + if err := store.db.QueryRow(`SELECT inactive FROM observers WHERE id = ?`, "obs-retained").Scan(&inactive); err != nil { + t.Fatal(err) + } + if inactive != 1 { + t.Errorf("inactive = %d, want 1 (retained replay must not resurrect a soft-deleted observer)", inactive) + } +} + +// A retained snapshot from an observer the analyzer has never heard from live +// describes a past that may be months old. Creating a row for it puts a +// permanently dead observer in the list. +func TestRetainedStatusDoesNotCreateUnknownObserver(t *testing.T) { + store := newTestStore(t) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + retainedStatusMsg("meshcore/LAX/obs-never-seen/status", `{"origin":"Ghost","noise_floor":-95.5}`), + nil, nil, &Config{}) + + var count int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM observers WHERE id = ?`, "obs-never-seen").Scan(&count); err != nil { + t.Fatal(err) + } + if count != 0 { + t.Errorf("observers row count = %d, want 0 (retained-only observer must not be created)", count) + } +} + +// The metrics sample is stamped with ingest time, so a replay would file a +// months-old reading as a present-tense measurement. +func TestRetainedStatusDoesNotInsertMetricsSample(t *testing.T) { + store := newTestStore(t) + seedObserver(t, store, "obs-retained", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + retainedStatusMsg(retainedStatusTopic, `{"origin":"obs-retained","noise_floor":-95.5,"battery_mv":3900}`), + nil, nil, &Config{}) + + var count int + if err := store.db.QueryRow( + `SELECT COUNT(*) FROM observer_metrics WHERE observer_id = ?`, "obs-retained", + ).Scan(&count); err != nil { + t.Fatal(err) + } + if count != 0 { + t.Errorf("observer_metrics rows = %d, want 0 (retained replay is not a new sample)", count) + } +} + +// The retained snapshot is still the observer's last known state, so its +// metadata is worth keeping — only the liveness signal is suppressed. +func TestRetainedStatusStillUpdatesMetadata(t *testing.T) { + store := newTestStore(t) + seedObserver(t, store, "obs-retained", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + retainedStatusMsg(retainedStatusTopic, `{"origin":"obs-retained","firmware":"v1.2.3","noise_floor":-95.5}`), + nil, nil, &Config{}) + + var firmware string + var noiseFloor float64 + if err := store.db.QueryRow( + `SELECT firmware, noise_floor FROM observers WHERE id = ?`, "obs-retained", + ).Scan(&firmware, &noiseFloor); err != nil { + t.Fatal(err) + } + if firmware != "v1.2.3" { + t.Errorf("firmware = %q, want %q", firmware, "v1.2.3") + } + if noiseFloor != -95.5 { + t.Errorf("noise_floor = %v, want -95.5", noiseFloor) + } +} + +// Regression guard for #1465: a live status message must still stamp last_seen +// with ingest time. Only the retained flag changes the behaviour. +func TestLiveStatusStillAdvancesLastSeen(t *testing.T) { + store := newTestStore(t) + before := seedObserver(t, store, "obs-live", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + &mockMessage{topic: "meshcore/LAX/obs-live/status", payload: []byte(`{"origin":"obs-live","noise_floor":-95.5}`)}, + nil, nil, &Config{}) + + if got := observerLastSeen(t, store, "obs-live"); got == before { + t.Errorf("last_seen = %s, want it advanced past %s (live status is liveness)", got, before) + } +} + +func TestLiveStatusStillCreatesUnknownObserver(t *testing.T) { + store := newTestStore(t) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + &mockMessage{topic: "meshcore/LAX/obs-fresh/status", payload: []byte(`{"origin":"Fresh","noise_floor":-95.5}`)}, + nil, nil, &Config{}) + + var count int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM observers WHERE id = ?`, "obs-fresh").Scan(&count); err != nil { + t.Fatal(err) + } + if count != 1 { + t.Errorf("observers row count = %d, want 1 (live status creates the observer)", count) + } +} diff --git a/cmd/ingestor/status_liveness.go b/cmd/ingestor/status_liveness.go new file mode 100644 index 000000000..9b9ef2893 --- /dev/null +++ b/cmd/ingestor/status_liveness.go @@ -0,0 +1,50 @@ +package main + +import ( + "strings" + "time" +) + +// statusLivenessMaxAge is how far a status payload's own timestamp may lag +// before the message stops counting as evidence the observer is alive now. +// +// Deliberately generous: observer clocks are naive (#1478) and can sit a whole +// timezone off, so anything under a day has to be accepted. The replays this +// guards against are months stale, so the margin costs nothing. +const statusLivenessMaxAge = 24 * time.Hour + +// statusIsLiveness reports whether a status message proves the observer was +// alive when we received it. +// +// The retain flag alone cannot answer that. Between EMQX and the ingestor sits +// a mosquitto bridge, and MQTT only sets RETAIN on messages delivered in +// response to a *new* subscription. The bridge runs cleansession, so every +// reconnect resubscribes and pulls EMQX's entire retained set — which it then +// republishes into its own broker as ordinary traffic. The ingestor is already +// subscribed there, so those months-old snapshots arrive with RETAIN cleared, +// indistinguishable from live status. On 2026-08-13 an EMQX restart replayed +// them and recreated 17 long-dead observers in two seconds. +// +// So judge the payload instead of the transport: +// - a status that is not "online" is the observer's own offline notice (its +// LWT, or a replay of it) and is never proof of life; +// - a timestamp older than statusLivenessMaxAge describes a past that may be +// months old. +// +// A payload with no timestamp, or one we cannot parse, is unjudgeable and +// keeps the previous behaviour — this guard removes false liveness, it does +// not invent new grounds to discard observers. +func statusIsLiveness(msg map[string]interface{}, now time.Time) bool { + if s, ok := msg["status"].(string); ok && s != "" && !strings.EqualFold(s, "online") { + return false + } + raw, _ := msg["timestamp"].(string) + if raw == "" { + return true + } + t, _, err := parseEnvelopeTime(raw) + if err != nil { + return true + } + return !t.Before(now.Add(-statusLivenessMaxAge)) +} diff --git a/cmd/ingestor/status_liveness_test.go b/cmd/ingestor/status_liveness_test.go new file mode 100644 index 000000000..6f96b07d1 --- /dev/null +++ b/cmd/ingestor/status_liveness_test.go @@ -0,0 +1,203 @@ +package main + +import ( + "fmt" + "testing" + "time" +) + +// The retain flag alone does not identify a replay. Our live topology puts a +// mosquitto bridge between EMQX and the ingestor, and MQTT only sets RETAIN on +// messages delivered in response to a *new* subscription. When the bridge +// reconnects (it runs cleansession, so every reconnect resubscribes), EMQX +// replays its whole retained set to the bridge, which republishes it into its +// own broker — and the ingestor, already subscribed, receives those months-old +// snapshots with RETAIN cleared, indistinguishable from live traffic. +// +// Observed on live 2026-08-13: EMQX restarted at 13:09 UTC and 17 long-dead +// observers were recreated inside two seconds, among them BE-BGS-RRY120-RES +// whose retained payload was published 2026-04-02 and says "offline". +// +// So judge the payload, not the transport: a status message is evidence of +// life only if it claims to be online and carries a timestamp from roughly now. + +func liveStatusMsg(topic, payload string) *mockMessage { + return &mockMessage{topic: topic, payload: []byte(payload), retained: false} +} + +func statusPayload(origin, status string, ts time.Time) string { + return fmt.Sprintf(`{"origin":%q,"status":%q,"timestamp":%q,"noise_floor":-95.5}`, + origin, status, ts.Format("2006-01-02T15:04:05.000000")) +} + +func TestStaleStatusPayloadDoesNotAdvanceLastSeen(t *testing.T) { + store := newTestStore(t) + want := seedObserver(t, store, "obs-zombie", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + liveStatusMsg("meshcore/LAX/obs-zombie/status", + statusPayload("obs-zombie", "online", time.Now().UTC().AddDate(0, -4, 0))), + nil, nil, &Config{}) + + if got := observerLastSeen(t, store, "obs-zombie"); got != want { + t.Errorf("last_seen = %s, want %s (a 4-month-old payload is a replay, not liveness)", got, want) + } +} + +// The exact live case: an observer purged from the DB must not come back when +// the bridge replays its months-old retained snapshot with RETAIN cleared. +func TestStaleStatusPayloadDoesNotCreateObserver(t *testing.T) { + store := newTestStore(t) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + liveStatusMsg("meshcore/OST/obs-purged/status", + statusPayload("BE-BGS-RRY120-RES", "offline", time.Now().UTC().AddDate(0, -4, 0))), + nil, nil, &Config{}) + + var count int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM observers WHERE id = ?`, "obs-purged").Scan(&count); err != nil { + t.Fatal(err) + } + if count != 0 { + t.Errorf("observers row count = %d, want 0 (replayed snapshot must not resurrect a purged observer)", count) + } +} + +// An offline notice is the observer's own death certificate — the broker +// replaying it, or the LWT firing, is never proof of life. +func TestOfflineStatusIsNotLiveness(t *testing.T) { + store := newTestStore(t) + want := seedObserver(t, store, "obs-offline", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + liveStatusMsg("meshcore/LAX/obs-offline/status", + statusPayload("obs-offline", "offline", time.Now().UTC())), + nil, nil, &Config{}) + + if got := observerLastSeen(t, store, "obs-offline"); got != want { + t.Errorf("last_seen = %s, want %s (an offline status is not liveness)", got, want) + } +} + +func TestStaleStatusPayloadDoesNotReactivateInactiveObserver(t *testing.T) { + store := newTestStore(t) + seedObserver(t, store, "obs-zombie", 60) + if _, err := store.db.Exec(`UPDATE observers SET inactive = 1 WHERE id = ?`, "obs-zombie"); err != nil { + t.Fatal(err) + } + + handleMessage(store, "test", MQTTSource{Name: "test"}, + liveStatusMsg("meshcore/LAX/obs-zombie/status", + statusPayload("obs-zombie", "online", time.Now().UTC().AddDate(0, -4, 0))), + nil, nil, &Config{}) + + var inactive int + if err := store.db.QueryRow(`SELECT inactive FROM observers WHERE id = ?`, "obs-zombie").Scan(&inactive); err != nil { + t.Fatal(err) + } + if inactive != 1 { + t.Errorf("inactive = %d, want 1 (a replay must not undo the soft-delete)", inactive) + } +} + +// ─── The guard must not swallow real observers ────────────────────────────── + +func TestFreshStatusStillCountsAsLiveness(t *testing.T) { + store := newTestStore(t) + stale := seedObserver(t, store, "obs-live", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + liveStatusMsg("meshcore/LAX/obs-live/status", + statusPayload("obs-live", "online", time.Now().UTC())), + nil, nil, &Config{}) + + if got := observerLastSeen(t, store, "obs-live"); got == stale { + t.Errorf("last_seen = %s, want it advanced (a fresh online status is liveness)", got) + } +} + +// Observer clocks are naive and can sit a whole timezone off (#1478). A UTC-12 +// observer must not be mistaken for a replay. +func TestNaiveClockSkewStillCountsAsLiveness(t *testing.T) { + store := newTestStore(t) + stale := seedObserver(t, store, "obs-skewed", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + liveStatusMsg("meshcore/LAX/obs-skewed/status", + statusPayload("obs-skewed", "online", time.Now().UTC().Add(-12*time.Hour))), + nil, nil, &Config{}) + + if got := observerLastSeen(t, store, "obs-skewed"); got == stale { + t.Errorf("last_seen = %s, want it advanced (12h naive skew is a timezone, not a replay)", got) + } +} + +func TestStatusWithoutTimestampStillCountsAsLiveness(t *testing.T) { + store := newTestStore(t) + stale := seedObserver(t, store, "obs-nots", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + liveStatusMsg("meshcore/LAX/obs-nots/status", `{"origin":"obs-nots","noise_floor":-95.5}`), + nil, nil, &Config{}) + + if got := observerLastSeen(t, store, "obs-nots"); got == stale { + t.Errorf("last_seen = %s, want it advanced (no timestamp to judge — keep the old behaviour)", got) + } +} + +func TestStatusWithUnparseableTimestampStillCountsAsLiveness(t *testing.T) { + store := newTestStore(t) + stale := seedObserver(t, store, "obs-badts", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + liveStatusMsg("meshcore/LAX/obs-badts/status", + `{"origin":"obs-badts","status":"online","timestamp":"not-a-time","noise_floor":-95.5}`), + nil, nil, &Config{}) + + if got := observerLastSeen(t, store, "obs-badts"); got == stale { + t.Errorf("last_seen = %s, want it advanced (unjudgeable timestamp — keep the old behaviour)", got) + } +} + +// A replayed snapshot is still the observer's last known state: metadata is +// kept, only the liveness signal is suppressed. +func TestStaleStatusPayloadStillUpdatesMetadata(t *testing.T) { + store := newTestStore(t) + seedObserver(t, store, "obs-zombie", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + liveStatusMsg("meshcore/LAX/obs-zombie/status", + fmt.Sprintf(`{"origin":"obs-zombie","status":"online","timestamp":%q,"firmware":"v1.14.1","noise_floor":-95.5}`, + time.Now().UTC().AddDate(0, -4, 0).Format("2006-01-02T15:04:05.000000"))), + nil, nil, &Config{}) + + var firmware string + if err := store.db.QueryRow(`SELECT firmware FROM observers WHERE id = ?`, "obs-zombie").Scan(&firmware); err != nil { + t.Fatal(err) + } + if firmware != "v1.14.1" { + t.Errorf("firmware = %q, want %q (metadata from a replay is still the last known state)", firmware, "v1.14.1") + } +} + +// No metrics sample either — a months-old reading filed at ingest time reads +// as a present-tense measurement. +func TestStaleStatusPayloadDoesNotInsertMetricsSample(t *testing.T) { + store := newTestStore(t) + seedObserver(t, store, "obs-zombie", 60) + + handleMessage(store, "test", MQTTSource{Name: "test"}, + liveStatusMsg("meshcore/LAX/obs-zombie/status", + statusPayload("obs-zombie", "online", time.Now().UTC().AddDate(0, -4, 0))), + nil, nil, &Config{}) + + var count int + if err := store.db.QueryRow( + `SELECT COUNT(*) FROM observer_metrics WHERE observer_id = ?`, "obs-zombie", + ).Scan(&count); err != nil { + t.Fatal(err) + } + if count != 0 { + t.Errorf("observer_metrics rows = %d, want 0 (a replayed reading is not a new sample)", count) + } +} From ca9d98cb531ff987998b4210931db6645545f53a Mon Sep 17 00:00:00 2001 From: Dennis Jakobsen Date: Mon, 21 Sep 2026 00:33:34 +0200 Subject: [PATCH 2/2] test(ingestor): close the mutation gaps in the retained-status guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An independent mutation run against the port's original seven tests left nine mutants alive. Six of them sit on behaviour the change is actually responsible for; these tests kill all six, each verified by re-applying the mutation and watching the named test fail on a real assertion (not a compile error). packet_count bump re-added to UpsertObserverRetained -> killed can_relay_seen regressed 1 -> 0 on the retained path -> killed strings.EqualFold replaced by a case-sensitive == -> killed the `s != ""` guard dropped -> killed !t.Before(cutoff) flipped to t.After(cutoff) -> killed IATA TrimSpace/ToUpper dropped on the retained path -> killed Two of these matter more than their size suggests. Losing the case-insensitive comparison would silently suppress liveness for any firmware publishing "ONLINE" — the observer would age out while alive, which is worse than the bug this port fixes. And can_relay_seen is the tristate's "we have actually observed this" bit (#1290); a replay carrying no `repeat` field must not erase an answer the live path recorded. The three remaining survivors are left deliberately and are noted in the file: two concern statusLivenessMaxAge's exact value, which is documented as a generous margin rather than a threshold, and one concerns Stats.ObserverUpserts, a counter with no behavioural consequence. No production code and no existing assertion was changed. Co-Authored-By: Claude Opus 5 --- cmd/ingestor/retained_status_gaps_test.go | 207 ++++++++++++++++++++++ 1 file changed, 207 insertions(+) create mode 100644 cmd/ingestor/retained_status_gaps_test.go diff --git a/cmd/ingestor/retained_status_gaps_test.go b/cmd/ingestor/retained_status_gaps_test.go new file mode 100644 index 000000000..cd68cabd7 --- /dev/null +++ b/cmd/ingestor/retained_status_gaps_test.go @@ -0,0 +1,207 @@ +package main + +import ( + "strings" + "testing" + "time" +) + +// Gap-closing tests for #1885. An independent mutation run against the +// original seven tests left nine mutants alive; these kill the five that +// matter. Each test names the mutation it kills so a future reader can tell +// what it is load-bearing for. +// +// The four remaining survivors are deliberate: two concern the 24h constant's +// exact value (any value between "a few hours" and "months" passes, and the +// constant is documented as a generous margin rather than a threshold), and +// two concern Stats.ObserverUpserts, which is a counter with no behavioural +// consequence on either path. + +// Kills: re-adding `packet_count = packet_count + 1` to UpsertObserverRetained. +// +// The count is the observer's real traffic volume. A replay storm that bumped +// it would inflate every dead observer's count by one per ingestor restart, +// and nothing else in the suite looks at the column on this path. +func TestRetainedStatusDoesNotBumpPacketCount(t *testing.T) { + store := newTestStore(t) + seedObserver(t, store, "obs-count", 60) + if _, err := store.db.Exec(`UPDATE observers SET packet_count = 7 WHERE id = ?`, "obs-count"); err != nil { + t.Fatal(err) + } + + handleMessage(store, "test", MQTTSource{Name: "test"}, + retainedStatusMsg("meshcore/LAX/obs-count/status", `{"origin":"obs-count","noise_floor":-95.5}`), + nil, nil, &Config{}) + + var got int + if err := store.db.QueryRow(`SELECT packet_count FROM observers WHERE id = ?`, "obs-count").Scan(&got); err != nil { + t.Fatal(err) + } + if got != 7 { + t.Errorf("packet_count = %d, want 7 (a replay is not a packet)", got) + } + + // The live path must still bump it, or the guard above is just a bug. + handleMessage(store, "test", MQTTSource{Name: "test"}, + liveStatusMsg("meshcore/LAX/obs-count/status", + statusPayload("obs-count", "online", time.Now().UTC())), + nil, nil, &Config{}) + if err := store.db.QueryRow(`SELECT packet_count FROM observers WHERE id = ?`, "obs-count").Scan(&got); err != nil { + t.Fatal(err) + } + if got != 8 { + t.Errorf("packet_count = %d after live status, want 8 (live traffic still counts)", got) + } +} + +// Kills: making can_relay_seen regress 1 -> 0, and never setting it to 1. +// +// can_relay_seen is the tristate's "we have actually observed this" bit +// (#1290). A retained replay carrying no `repeat` field must not erase an +// answer the live path already recorded. +func TestRetainedStatusPreservesCanRelaySeen(t *testing.T) { + store := newTestStore(t) + seedObserver(t, store, "obs-relay", 60) + if _, err := store.db.Exec( + `UPDATE observers SET can_relay = 1, can_relay_seen = 1 WHERE id = ?`, "obs-relay"); err != nil { + t.Fatal(err) + } + + // No `repeat` key: meta.CanRelay stays nil, so the CASE must leave the + // flag alone rather than reset it. + handleMessage(store, "test", MQTTSource{Name: "test"}, + retainedStatusMsg("meshcore/LAX/obs-relay/status", `{"origin":"obs-relay","noise_floor":-95.5}`), + nil, nil, &Config{}) + + var seen, canRelay int + if err := store.db.QueryRow( + `SELECT can_relay_seen, can_relay FROM observers WHERE id = ?`, "obs-relay"). + Scan(&seen, &canRelay); err != nil { + t.Fatal(err) + } + if seen != 1 { + t.Errorf("can_relay_seen = %d, want 1 (a replay without `repeat` must not un-observe it)", seen) + } + if canRelay != 1 { + t.Errorf("can_relay = %d, want 1 (unchanged by a replay that does not report it)", canRelay) + } + + // A replay that DOES carry `repeat` still records the observation: the + // metadata half of the retained path is supposed to keep working. + store2 := newTestStore(t) + seedObserver(t, store2, "obs-relay2", 60) + if _, err := store2.db.Exec( + `UPDATE observers SET can_relay = NULL, can_relay_seen = 0 WHERE id = ?`, "obs-relay2"); err != nil { + t.Fatal(err) + } + handleMessage(store2, "test", MQTTSource{Name: "test"}, + retainedStatusMsg("meshcore/LAX/obs-relay2/status", + `{"origin":"obs-relay2","repeat":"true"}`), + nil, nil, &Config{}) + if err := store2.db.QueryRow( + `SELECT can_relay_seen, can_relay FROM observers WHERE id = ?`, "obs-relay2"). + Scan(&seen, &canRelay); err != nil { + t.Fatal(err) + } + if seen != 1 || canRelay != 1 { + t.Errorf("can_relay_seen/can_relay = %d/%d, want 1/1 (a reported value is still recorded)", seen, canRelay) + } +} + +// Kills: replacing strings.EqualFold with a case-sensitive ==. +// +// The whole guard turns on one string comparison. If it became +// case-sensitive, any firmware publishing "ONLINE" would have its liveness +// silently suppressed — the observer would age out despite being alive, which +// is the opposite of the bug this change fixes and far worse than it. +func TestStatusOnlineComparisonIsCaseInsensitive(t *testing.T) { + now := time.Now().UTC() + for _, s := range []string{"online", "ONLINE", "OnLiNe", "Online"} { + if !statusIsLiveness(map[string]interface{}{"status": s}, now) { + t.Errorf("statusIsLiveness(status=%q) = false, want true", s) + } + } + for _, s := range []string{"offline", "OFFLINE", "OffLine"} { + if statusIsLiveness(map[string]interface{}{"status": s}, now) { + t.Errorf("statusIsLiveness(status=%q) = true, want false", s) + } + } +} + +// Kills: dropping the `s != ""` guard. +// +// An empty status string is not a claim of being offline — it is the absence +// of a claim, and the guard exists so a payload that carries the key but no +// value keeps the previous behaviour instead of being silently discarded. +func TestEmptyStatusStringIsNotAnOfflineClaim(t *testing.T) { + now := time.Now().UTC() + if !statusIsLiveness(map[string]interface{}{"status": ""}, now) { + t.Error(`statusIsLiveness(status="") = false, want true (no claim is not an offline claim)`) + } + // A non-string status is likewise unjudgeable and must not suppress. + if !statusIsLiveness(map[string]interface{}{"status": 1}, now) { + t.Error("statusIsLiveness(status=1) = false, want true (unjudgeable, keep previous behaviour)") + } +} + +// Kills: flipping `!t.Before(cutoff)` to `t.After(cutoff)`. +// +// The two differ only exactly at the cutoff. The existing tests use 4-month +// staleness and "now", so nothing pinned which side of the boundary is +// inclusive. +func TestStatusLivenessAgeBoundaryIsInclusive(t *testing.T) { + // Truncate to whole seconds: the payload timestamp round-trips through + // RFC3339, which carries no sub-second part, so an untruncated `now` + // would put the "exactly at the cutoff" case a few hundred nanoseconds + // on the wrong side of the comparison and test the formatter instead of + // the boundary. + now := time.Now().UTC().Truncate(time.Second) + cutoff := now.Add(-statusLivenessMaxAge) + + cases := []struct { + name string + ts time.Time + want bool + }{ + {"exactly at the cutoff", cutoff, true}, + {"one second inside", cutoff.Add(time.Second), true}, + {"one second outside", cutoff.Add(-time.Second), false}, + } + for _, c := range cases { + msg := map[string]interface{}{ + "status": "online", + "timestamp": c.ts.Format(time.RFC3339), + } + if got := statusIsLiveness(msg, now); got != c.want { + t.Errorf("%s: statusIsLiveness = %v, want %v", c.name, got, c.want) + } + } +} + +// Kills: dropping the TrimSpace/ToUpper normalisation of iata in +// UpsertObserverRetained. +// +// The live path normalises, so a replay writing a raw topic segment would +// make the same observer's region disagree with itself depending on which +// path last touched it. +func TestRetainedStatusNormalizesIATA(t *testing.T) { + store := newTestStore(t) + // Seed through the normal path so the row exists, then blank the column + // so the COALESCE has something to overwrite. + seedObserver(t, store, "obs-iata", 60) + if _, err := store.db.Exec(`UPDATE observers SET iata = 'XXX' WHERE id = ?`, "obs-iata"); err != nil { + t.Fatal(err) + } + + handleMessage(store, "test", MQTTSource{Name: "test"}, + retainedStatusMsg("meshcore/lax/obs-iata/status", `{"origin":"obs-iata"}`), + nil, nil, &Config{}) + + var got string + if err := store.db.QueryRow(`SELECT iata FROM observers WHERE id = ?`, "obs-iata").Scan(&got); err != nil { + t.Fatal(err) + } + if got != strings.ToUpper(got) || got != "LAX" { + t.Errorf("iata = %q, want %q (retained path must normalise like the live path)", got, "LAX") + } +}