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

@@ -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) {

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)

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()

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)
 				}