From ca62790c81eeda5ae6b72c0d4c433ca7160336c9 Mon Sep 17 00:00:00 2001 From: Alex B <142788550+A13xB0@users.noreply.github.com> Date: Mon, 12 Oct 2026 10:13:53 +0100 Subject: [PATCH] fix(store): load the observer IATA in the background chunk loader (#2185) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem With `hotStartupHours` set, startup loads the hot window first. `loadChunk` then fills the rest of the retention window in the background. Its query selected the observer ID and name but not the IATA code, and scanned no IATA, so every packet it loaded had an empty `ObserverIATA`. On a 7-day window with a 24 h hot start, that's six days of packets after every restart that: - have no `observer_iata` in the packets API; - never match the `iata` filter (#1188). `Load`, `scanAndMergeChunk` (the hot window) and both ingest paths already join `observers` to get the IATA code, so only background-loaded packets were affected. **Seen on a live instance** (`hotStartupHours: 24`, restarted about 18 h earlier), sampling `/api/packets?until=…`: | packets | with `observer_iata` | |---|---| | 20–30 h old (hot window) | 200 / 200 | | 46 h old (background-loaded) | 0 / 200 | | 60 h old (background-loaded) | 0 / 200 | With this fix on a staging instance, packets older than its hot window are 300 / 300. ## Change The chunk query selects `COALESCE(obs.iata, '')` for both schemas: the v3 query through the observers join it already had, and the legacy query through a new join on the observer ID, as `Load` does. `loadChunk` scans the column into `ObserverIATA`. ## Tests - `TestLoadChunkKeepsObserverIATA`: red commit first (fails on its assertion), then green. - `cd cmd/server && go test ./...` passes. - SonarQube: quality gate passed, no issues on added lines. **Overlap with #2183:** both edit the same lines of `loadChunk`'s `StoreObs` literal (#2183 interns the observer fields). Whichever lands second needs a trivial rebase; if this lands first, #2183 should intern `ObserverIATA` there too. There's no code dependency. Backend only: no UI, API shape or config change. 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_01KaR4xu8gVykQBkH8yG8m6H --------- Co-authored-by: Claude Opus 5.5 --- cmd/server/loadchunk_observer_iata_test.go | 54 ++++++++++++++++++++++ cmd/server/store.go | 12 +++-- 2 files changed, 61 insertions(+), 5 deletions(-) create mode 100644 cmd/server/loadchunk_observer_iata_test.go diff --git a/cmd/server/loadchunk_observer_iata_test.go b/cmd/server/loadchunk_observer_iata_test.go new file mode 100644 index 00000000..715518ca --- /dev/null +++ b/cmd/server/loadchunk_observer_iata_test.go @@ -0,0 +1,54 @@ +package main + +import ( + "database/sql" + "os" + "path/filepath" + "testing" + "time" +) + +// The background loader (loadChunk) fills in days 2-7 of the retention +// window after the hot startup window. Its query left out the observer's +// IATA code, so packets older than the hot window lost observer_iata in the +// packets API and the iata filter (#1188), while Load and ingest kept it. +func TestLoadChunkKeepsObserverIATA(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "iata.db") + createTestDBAt(t, dbPath, 20) + t.Cleanup(func() { os.RemoveAll(dir) }) + conn, err := sql.Open("sqlite3", dbPath) + if err != nil { + t.Fatal(err) + } + if _, err := conn.Exec(`INSERT INTO observers (id, name, iata) VALUES ('obs1', 'Obs1', 'SJC')`); err != nil { + t.Fatal(err) + } + conn.Close() + + db, err := OpenDB(dbPath) + if err != nil { + t.Fatalf("OpenDB: %v", err) + } + t.Cleanup(func() { db.Close() }) + store := NewPacketStore(db, &PacketStoreConfig{}) + from := time.Date(2025, 12, 31, 0, 0, 0, 0, time.UTC) + if err := store.loadChunk(from, from.Add(72*time.Hour)); err != nil { + t.Fatal(err) + } + store.mu.RLock() + defer store.mu.RUnlock() + if len(store.packets) == 0 { + t.Fatal("loadChunk loaded nothing") + } + for _, tx := range store.packets { + for _, o := range tx.Observations { + if o.ObserverIATA != "SJC" { + t.Fatalf("tx %d: observation IATA = %q, want SJC", tx.ID, o.ObserverIATA) + } + } + if tx.ObserverIATA != "SJC" { + t.Fatalf("tx %d: IATA = %q, want SJC", tx.ID, tx.ObserverIATA) + } + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index b6b3d898..d0f2fae5 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -1203,7 +1203,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { if s.db.isV3 { chunkSQL = `SELECT t.id, t.raw_hex, t.hash, t.first_seen, t.route_type, t.payload_type, t.payload_version, t.decoded_json, - o.id, obs.id, obs.name, o.direction, + o.id, obs.id, obs.name, COALESCE(obs.iata, ''), o.direction, o.snr, o.rssi, o.score, o.path_json, strftime('%Y-%m-%dT%H:%M:%fZ', o.timestamp, 'unixepoch')` + obsRawHexCol + rpCol + scopeNameCol + ` FROM transmissions t LEFT JOIN observations o ON o.transmission_id = t.id @@ -1212,10 +1212,11 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { } else { chunkSQL = `SELECT t.id, t.raw_hex, t.hash, t.first_seen, t.route_type, t.payload_type, t.payload_version, t.decoded_json, - o.id, o.observer_id, o.observer_name, o.direction, + o.id, o.observer_id, o.observer_name, COALESCE(obs.iata, ''), o.direction, o.snr, o.rssi, o.score, o.path_json, o.timestamp` + obsRawHexCol + rpCol + scopeNameCol + ` FROM transmissions t - LEFT JOIN observations o ON o.transmission_id = t.id` + filterClause + ` + LEFT JOIN observations o ON o.transmission_id = t.id + LEFT JOIN observers obs ON obs.id = o.observer_id` + filterClause + ` ORDER BY t.first_seen ASC, o.timestamp DESC` } @@ -1268,7 +1269,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { var rawHex, hash, firstSeen, decodedJSON sql.NullString var routeType, payloadType, payloadVersion sql.NullInt64 var obsID sql.NullInt64 - var observerID, observerName, direction, pathJSON, obsTimestamp sql.NullString + var observerID, observerName, observerIATA, direction, pathJSON, obsTimestamp sql.NullString var snr, rssi sql.NullFloat64 var score sql.NullInt64 var obsRawHex sql.NullString @@ -1277,7 +1278,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { scanArgs := []interface{}{&txID, &rawHex, &hash, &firstSeen, &routeType, &payloadType, &payloadVersion, &decodedJSON, - &obsID, &observerID, &observerName, &direction, + &obsID, &observerID, &observerName, &observerIATA, &direction, &snr, &rssi, &score, &pathJSON, &obsTimestamp} if s.db.hasObsRawHex { scanArgs = append(scanArgs, &obsRawHex) @@ -1330,6 +1331,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { TransmissionID: txID, ObserverID: obsIDStr, ObserverName: nullStrVal(observerName), + ObserverIATA: nullStrVal(observerIATA), Direction: nullStrVal(direction), SNR: nullFloatPtr(snr), RSSI: nullFloatPtr(rssi),