From df63efa78db5415d5d32516dd71c136a91426435 Mon Sep 17 00:00:00 2001 From: Kpa-clawbot <259247574+Kpa-clawbot@users.noreply.github.com> Date: Fri, 27 Mar 2026 17:26:26 -0700 Subject: [PATCH] fix: poll new observations for existing transmissions (fixes #174) The poller only queried WHERE t.id > sinceID, which missed new observations added to transmissions already in the store. The trace page was correct because it always queries the DB directly. Add IngestNewObservations() that polls observations by o.id watermark, adds them to existing StoreTx entries, re-picks best observation, and invalidates analytics caches. The Poller now tracks both lastTxID and lastObsID watermarks. Includes tests for v3, v2, dedup, best-path re-pick, and GetMaxObservationID. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- cmd/server/coverage_test.go | 122 ++++++++++++++++++++++++++++ cmd/server/db.go | 7 ++ cmd/server/store.go | 153 ++++++++++++++++++++++++++++++++++++ cmd/server/websocket.go | 10 ++- 4 files changed, 290 insertions(+), 2 deletions(-) diff --git a/cmd/server/coverage_test.go b/cmd/server/coverage_test.go index eb2edafa..238ec918 100644 --- a/cmd/server/coverage_test.go +++ b/cmd/server/coverage_test.go @@ -3356,6 +3356,128 @@ func TestIngestNewFromDBDuplicateObs(t *testing.T) { } } +// --- IngestNewObservations (fixes #174) --- + +func TestIngestNewObservations(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + store := NewPacketStore(db) + store.Load() + + // Get initial observation count for transmission 1 (hash abc123def4567890) + initialTx := store.byHash["abc123def4567890"] + if initialTx == nil { + t.Fatal("expected to find transmission abc123def4567890 in store") + } + initialObsCount := initialTx.ObservationCount + if initialObsCount != 2 { + t.Fatalf("expected 2 initial observations, got %d", initialObsCount) + } + + // Record the max obs ID after initial load + maxObsID := db.GetMaxObservationID() + + // Simulate a new observation arriving for the existing transmission AFTER + // the poller has already advanced past its transmission ID + db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) + VALUES (1, 2, 5.0, -100, '["aa","bb","cc"]', ?)`, time.Now().Unix()) + + // Verify IngestNewFromDB does NOT pick up the new observation (tx id hasn't changed) + txMax := store.MaxTransmissionID() + _, newTxMax := store.IngestNewFromDB(txMax, 100) + if initialTx.ObservationCount != initialObsCount { + t.Errorf("IngestNewFromDB should not have changed obs count, was %d now %d", + initialObsCount, initialTx.ObservationCount) + } + _ = newTxMax + + // IngestNewObservations should pick it up + newObsMax := store.IngestNewObservations(maxObsID, 500) + if newObsMax <= maxObsID { + t.Errorf("expected newObsMax > %d, got %d", maxObsID, newObsMax) + } + if initialTx.ObservationCount != initialObsCount+1 { + t.Errorf("expected obs count %d, got %d", initialObsCount+1, initialTx.ObservationCount) + } + if len(initialTx.Observations) != initialObsCount+1 { + t.Errorf("expected %d observations slice len, got %d", initialObsCount+1, len(initialTx.Observations)) + } + + // Best observation should have been re-picked (new obs has longer path) + if initialTx.PathJSON != `["aa","bb","cc"]` { + t.Errorf("expected best path to be updated to longer path, got %s", initialTx.PathJSON) + } + + t.Run("no new observations", func(t *testing.T) { + max := store.IngestNewObservations(newObsMax, 500) + if max != newObsMax { + t.Errorf("expected same max %d, got %d", newObsMax, max) + } + }) + + t.Run("dedup by observer+path", func(t *testing.T) { + // Insert duplicate observation (same observer + path as existing) + db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) + VALUES (1, 1, 12.5, -90, '["aa","bb"]', ?)`, time.Now().Unix()) + prevCount := initialTx.ObservationCount + newMax2 := store.IngestNewObservations(newObsMax, 500) + if initialTx.ObservationCount != prevCount { + t.Errorf("duplicate obs should not increase count, was %d now %d", + prevCount, initialTx.ObservationCount) + } + _ = newMax2 + }) + + t.Run("default limit", func(t *testing.T) { + _ = store.IngestNewObservations(newObsMax, 0) + }) +} + +func TestIngestNewObservationsV2(t *testing.T) { + db := setupTestDBv2(t) + defer db.Close() + seedV2Data(t, db) + store := NewPacketStore(db) + store.Load() + + tx := store.byHash["abc123def4567890"] + if tx == nil { + t.Fatal("expected to find transmission in store") + } + initialCount := tx.ObservationCount + + maxObsID := db.GetMaxObservationID() + + // Add new observation for existing transmission + db.conn.Exec(`INSERT INTO observations (transmission_id, observer_id, observer_name, snr, rssi, path_json, timestamp) + VALUES (1, 'obs2', 'Obs Two', 6.0, -98, '["dd","ee"]', ?)`, time.Now().Unix()) + + newMax := store.IngestNewObservations(maxObsID, 500) + if newMax <= maxObsID { + t.Errorf("expected newMax > %d, got %d", maxObsID, newMax) + } + if tx.ObservationCount != initialCount+1 { + t.Errorf("expected obs count %d, got %d", initialCount+1, tx.ObservationCount) + } +} + +func TestGetMaxObservationID(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + + maxID := db.GetMaxObservationID() + if maxID != 0 { + t.Errorf("expected 0 for empty table, got %d", maxID) + } + + seedTestData(t, db) + maxID = db.GetMaxObservationID() + if maxID <= 0 { + t.Errorf("expected positive max obs ID, got %d", maxID) + } +} + // --- perfMiddleware with endpoint normalization --- func TestPerfMiddlewareEndpointNormalization(t *testing.T) { diff --git a/cmd/server/db.go b/cmd/server/db.go index 9379851f..3f38b21f 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -1488,6 +1488,13 @@ func (db *DB) GetMaxTransmissionID() int { return maxID } +// GetMaxObservationID returns the current max observation ID for polling. +func (db *DB) GetMaxObservationID() int { + var maxID int + db.conn.QueryRow("SELECT COALESCE(MAX(id), 0) FROM observations").Scan(&maxID) + return maxID +} + // GetObserverPacketCounts returns packetsLastHour for all observers (batch query). func (db *DB) GetObserverPacketCounts(sinceEpoch int64) map[string]int { counts := make(map[string]int) diff --git a/cmd/server/store.go b/cmd/server/store.go index f213e5b6..2a108aaa 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -991,6 +991,159 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac return result, newMaxID } +// IngestNewObservations loads new observations for transmissions already in the +// store. This catches observations that arrive after IngestNewFromDB has already +// advanced past the transmission's ID (fixes #174). +func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) int { + if limit <= 0 { + limit = 500 + } + + var querySQL string + if s.db.isV3 { + querySQL = `SELECT o.id, o.transmission_id, obs.id, obs.name, o.direction, + o.snr, o.rssi, o.score, o.path_json, datetime(o.timestamp, 'unixepoch') + FROM observations o + LEFT JOIN observers obs ON obs.rowid = o.observer_idx + WHERE o.id > ? + ORDER BY o.id ASC + LIMIT ?` + } else { + querySQL = `SELECT o.id, o.transmission_id, o.observer_id, o.observer_name, o.direction, + o.snr, o.rssi, o.score, o.path_json, o.timestamp + FROM observations o + WHERE o.id > ? + ORDER BY o.id ASC + LIMIT ?` + } + + rows, err := s.db.conn.Query(querySQL, sinceObsID, limit) + if err != nil { + log.Printf("[store] ingest observations query error: %v", err) + return sinceObsID + } + defer rows.Close() + + type obsRow struct { + obsID int + txID int + observerID string + observerName string + direction string + snr, rssi *float64 + score *int + pathJSON string + timestamp string + } + + var obsRows []obsRow + for rows.Next() { + var oid, txID int + var observerID, observerName, direction, pathJSON, ts sql.NullString + var snr, rssi sql.NullFloat64 + var score sql.NullInt64 + + if err := rows.Scan(&oid, &txID, &observerID, &observerName, &direction, + &snr, &rssi, &score, &pathJSON, &ts); err != nil { + continue + } + + obsRows = append(obsRows, obsRow{ + obsID: oid, + txID: txID, + observerID: nullStrVal(observerID), + observerName: nullStrVal(observerName), + direction: nullStrVal(direction), + snr: nullFloatPtr(snr), + rssi: nullFloatPtr(rssi), + score: nullIntPtr(score), + pathJSON: nullStrVal(pathJSON), + timestamp: nullStrVal(ts), + }) + } + + if len(obsRows) == 0 { + return sinceObsID + } + + s.mu.Lock() + defer s.mu.Unlock() + + newMaxObsID := sinceObsID + updatedTxs := make(map[int]*StoreTx) + + for _, r := range obsRows { + if r.obsID > newMaxObsID { + newMaxObsID = r.obsID + } + + // Already ingested (e.g. by IngestNewFromDB in same cycle) + if _, exists := s.byObsID[r.obsID]; exists { + continue + } + + tx := s.byTxID[r.txID] + if tx == nil { + continue // transmission not yet in store + } + + // Dedup by observer + path + isDupe := false + for _, existing := range tx.Observations { + if existing.ObserverID == r.observerID && existing.PathJSON == r.pathJSON { + isDupe = true + break + } + } + if isDupe { + continue + } + + obs := &StoreObs{ + ID: r.obsID, + TransmissionID: r.txID, + ObserverID: r.observerID, + ObserverName: r.observerName, + Direction: r.direction, + SNR: r.snr, + RSSI: r.rssi, + Score: r.score, + PathJSON: r.pathJSON, + Timestamp: r.timestamp, + } + tx.Observations = append(tx.Observations, obs) + tx.ObservationCount++ + s.byObsID[r.obsID] = obs + if r.observerID != "" { + s.byObserver[r.observerID] = append(s.byObserver[r.observerID], obs) + } + s.totalObs++ + updatedTxs[r.txID] = tx + } + + // Re-pick best observation for updated transmissions + for _, tx := range updatedTxs { + pickBestObservation(tx) + } + + if len(updatedTxs) > 0 { + // Invalidate analytics caches + s.cacheMu.Lock() + s.rfCache = make(map[string]*cachedResult) + s.topoCache = make(map[string]*cachedResult) + s.hashCache = make(map[string]*cachedResult) + s.chanCache = make(map[string]*cachedResult) + s.distCache = make(map[string]*cachedResult) + s.subpathCache = make(map[string]*cachedResult) + s.cacheMu.Unlock() + + log.Printf("[poller] IngestNewObservations: updated %d existing txs, maxObsID %d->%d", + len(updatedTxs), sinceObsID, newMaxObsID) + } + + return newMaxObsID +} + // MaxTransmissionID returns the highest transmission ID in the store. func (s *PacketStore) MaxTransmissionID() int { s.mu.RLock() diff --git a/cmd/server/websocket.go b/cmd/server/websocket.go index 108551b0..d8cb19da 100644 --- a/cmd/server/websocket.go +++ b/cmd/server/websocket.go @@ -165,7 +165,8 @@ func NewPoller(db *DB, hub *Hub, interval time.Duration) *Poller { func (p *Poller) Start() { lastID := p.db.GetMaxTransmissionID() - log.Printf("[poller] starting from transmission ID %d, interval %v", lastID, p.interval) + lastObsID := p.db.GetMaxObservationID() + log.Printf("[poller] starting from transmission ID %d, obs ID %d, interval %v", lastID, lastObsID, p.interval) ticker := time.NewTicker(p.interval) defer ticker.Stop() @@ -174,11 +175,16 @@ func (p *Poller) Start() { select { case <-ticker.C: if p.store != nil { - // Ingest into in-memory store and broadcast + // Ingest new transmissions into in-memory store and broadcast newTxs, newMax := p.store.IngestNewFromDB(lastID, 100) if newMax > lastID { lastID = newMax } + // Ingest new observations for existing transmissions (fixes #174) + newObsMax := p.store.IngestNewObservations(lastObsID, 500) + if newObsMax > lastObsID { + lastObsID = newObsMax + } if len(newTxs) > 0 { log.Printf("[broadcast] sending %d packets to %d clients (lastID now %d)", len(newTxs), p.hub.ClientCount(), lastID) }