mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-10-15 16:50:00 +00:00
fix(store): load the observer IATA in the background chunk loader (#2185)
## 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 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
dc7ebcd6f5
commit
ca62790c81
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
+7
-5
@@ -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),
|
||||
|
||||
Reference in New Issue
Block a user