Record every observed route per transmission and show mixed ADVERTs in Relay Airtime Share

cmd/ingestor/db.go

@@ -76,6 +76,8 @@ type Store struct {
 	stmtGetTxByHash            *sql.Stmt
 	stmtInsertTransmission     *sql.Stmt
 	stmtUpdateTxFirstSeen      *sql.Stmt
+	stmtOrTxRouteMask          *sql.Stmt
+	stmtInsertRouteMaskChange  *sql.Stmt
 	stmtBumpTxLastSeen         *sql.Stmt
 	stmtInsertObservation      *sql.Stmt
 	stmtUpsertNode             *sql.Stmt
@@ -849,14 +851,14 @@ func applySchema(db *sql.DB) error {
 func (s *Store) prepareStatements() error {
 	var err error
 
-	s.stmtGetTxByHash, err = s.db.Prepare("SELECT id, first_seen FROM transmissions WHERE hash = ?")
+	s.stmtGetTxByHash, err = s.db.Prepare("SELECT id, first_seen, route_mask FROM transmissions WHERE hash = ?")
 	if err != nil {
 		return err
 	}
 
 	s.stmtInsertTransmission, err = s.db.Prepare(`
-		INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json, channel_hash, scope_name, from_pubkey, last_seen)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+		INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json, channel_hash, scope_name, from_pubkey, last_seen, route_mask)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
 	`)
 	if err != nil {
 		return err
@@ -867,6 +869,22 @@ func (s *Store) prepareStatements() error {
 		return err
 	}
 
+	// #89: OR a newly observed route bit into a known route_mask. The OR runs
+	// in SQL, so it is atomic against any other writer; NULL rows are left to
+	// the backfill (backfillTxRouteMask), which owns un-backfilled legacy rows.
+	s.stmtOrTxRouteMask, err = s.db.Prepare(`UPDATE transmissions SET route_mask = route_mask | ?
+		WHERE id = ? AND route_mask IS NOT NULL AND (route_mask & ?) = 0`)
+	if err != nil {
+		return err
+	}
+	// Change log for running servers (route_mask_changes): the transmission's
+	// full mask as it is inside the current transaction, right after the OR.
+	s.stmtInsertRouteMaskChange, err = s.db.Prepare(`INSERT INTO route_mask_changes (transmission_id, route_mask, created_at)
+		SELECT id, route_mask, ? FROM transmissions WHERE id = ? AND route_mask IS NOT NULL`)
+	if err != nil {
+		return err
+	}
+
 	// Ping-score highscore/leaderboard feature (see ping_triggers.go and
 	// internal/dbschema's ensurePingTriggersTable). INSERT OR IGNORE on
 	// tx_id (PRIMARY KEY) makes this idempotent -- only ever written once
@@ -1126,7 +1144,9 @@ func (s *Store) InsertTransmission(data *PacketData) (bool, error) {
 	// Check for existing transmission
 	var existingID int64
 	var existingFirstSeen string
-	err := s.stmtGetTxByHash.QueryRow(hash).Scan(&existingID, &existingFirstSeen)
+	var existingMask sql.NullInt64
+	routeBit := packetpath.RouteMaskBit(data.RouteType)
+	err := s.stmtGetTxByHash.QueryRow(hash).Scan(&existingID, &existingFirstSeen, &existingMask)
 	if err == nil {
 		// Existing transmission
 		txID = existingID
@@ -1143,6 +1163,7 @@ func (s *Store) InsertTransmission(data *PacketData) (bool, error) {
 			scopeNameForDB(data),
 			nilIfEmpty(data.FromPubkey),
 			epochSecondsForLastSeen(rxTime),
+			routeBit,
 		)
 		if err != nil {
 			s.Stats.WriteErrors.Add(1)
@@ -1203,12 +1224,22 @@ func (s *Store) InsertTransmission(data *PacketData) (bool, error) {
 	)
 	resolvedJSON := marshalResolvedPath(resolved)
 
-	_, err = s.stmtInsertObservation.Exec(
+	obsArgs := []interface{}{
 		txID, observerIdx, data.Direction,
 		data.SNR, data.RSSI, data.Score,
 		data.PathJSON, epochTs, nilIfEmpty(data.RawHex),
 		nilIfEmpty(resolvedJSON),
-	)
+	}
+	// #89: an observation that brings a route bit its transmission does not
+	// have yet is written together with that bit, in one transaction: an
+	// observation must never be visible without its bit, and a known mask is
+	// never backfilled again. Skipped when the bit is already set or the row
+	// is still an un-backfilled legacy row (NULL), which the backfill owns.
+	if !isNew && routeBit != 0 && existingMask.Valid && existingMask.Int64&routeBit == 0 {
+		err = s.insertObservationWithRouteBit(txID, routeBit, obsArgs)
+	} else {
+		_, err = s.stmtInsertObservation.Exec(obsArgs...)
+	}
 	if err != nil {
 		s.Stats.WriteErrors.Add(1)
 		log.Printf("[db] observation insert (non-fatal): %v", err)
@@ -1233,6 +1264,36 @@ func (s *Store) InsertTransmission(data *PacketData) (bool, error) {
 	return isNew, nil
 }
 
+// insertObservationWithRouteBit ORs routeBit into the transmission's
+// route_mask, inserts (or upserts) the observation and, when the mask
+// actually changed, appends a route_mask_changes row with the resulting mask,
+// all in one transaction: either everything becomes visible or nothing does.
+// The change row is what tells a running server about a bit that arrived
+// through an upsert of an existing observation row (same id, so its
+// new-observation poll never sees it). Caller holds writerMu. The pool has a
+// single connection, which this transaction holds until it ends: every
+// statement in between must run on dbTx, never on s.db.
+func (s *Store) insertObservationWithRouteBit(txID, routeBit int64, obsArgs []interface{}) error {
+	dbTx, err := s.db.Begin()
+	if err != nil {
+		return fmt.Errorf("begin route_mask transaction: %w", err)
+	}
+	defer dbTx.Rollback() // no-op after a successful Commit
+	res, err := dbTx.Stmt(s.stmtOrTxRouteMask).Exec(routeBit, txID, routeBit)
+	if err != nil {
+		return fmt.Errorf("route_mask update: %w", err)
+	}
+	if _, err := dbTx.Stmt(s.stmtInsertObservation).Exec(obsArgs...); err != nil {
+		return err
+	}
+	if changed, _ := res.RowsAffected(); changed > 0 {
+		if _, err := dbTx.Stmt(s.stmtInsertRouteMaskChange).Exec(time.Now().Unix(), txID); err != nil {
+			return fmt.Errorf("route_mask change row: %w", err)
+		}
+	}
+	return dbTx.Commit()
+}
+
 // UpsertNode inserts or updates a node.
 func (s *Store) UpsertNode(pubKey, name, role string, lat, lon *float64, lastSeen string) error {
 	ingestNow := time.Now().UTC().Format(time.RFC3339)

cmd/ingestor/decoder.go

@@ -205,7 +205,7 @@ type DecodedPacket struct {
 }
 
 func decodeHeader(b byte) Header {
-	rt := int(b & 0x03)
+	rt := packetpath.RouteTypeFromHeader(b)
 	pt := int((b >> 2) & 0x0F)
 	pv := int((b >> 6) & 0x03)
 

cmd/ingestor/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"crypto/hmac"
 	"crypto/sha256"
 	"crypto/tls"
@@ -281,6 +282,10 @@ func main() {
 			log.Printf("[prune] startup pruned %d transmissions older than %d days", n, packetDays)
 		}
 	}
+	// #89: route_mask_changes rows of transmissions deleted by any path.
+	if _, err := store.PruneOrphanRouteMaskChanges(); err != nil {
+		log.Printf("[prune] route_mask_changes error: %v", err)
+	}
 
 	// Client-RX coverage retention: bound the opt-in coverage tables (#1727).
 	// Independent of the feature flag, so data persists are reaped even after
@@ -309,6 +314,15 @@ func main() {
 		log.Printf("[ingest-buffer] write path ready; draining backlog (0 dropped)")
 	}
 
+	// #89: route_mask backfill runs next to live ingest, after the buffer is
+	// draining, so its one-time pending-index build (~20 s on a staging-sized
+	// DB) is absorbed by IngestBuffer instead of delaying start-up. Cancelled
+	// on shutdown so store.Close() does not wait for it; it resumes on the
+	// next start.
+	routeMaskCtx, stopRouteMaskBackfill := context.WithCancel(context.Background())
+	defer stopRouteMaskBackfill()
+	store.StartRouteMaskBackfill(routeMaskCtx)
+
 	// Daily ticker for node retention
 	retentionTicker := time.NewTicker(1 * time.Hour)
 	go func() {
@@ -352,6 +366,9 @@ func main() {
 				} else if n > 0 {
 					store.RunIncrementalVacuum(vacuumPages)
 				}
+				if _, err := store.PruneOrphanRouteMaskChanges(); err != nil {
+					log.Printf("[prune] route_mask_changes error: %v", err)
+				}
 			}
 		}()
 		log.Printf("[prune] auto-prune enabled: packets older than %d days will be removed daily", packetDays)
@@ -475,6 +492,7 @@ func main() {
 	<-sig
 
 	log.Println("Shutting down...")
+	stopRouteMaskBackfill()
 	retentionTicker.Stop()
 	metricsRetentionTicker.Stop()
 	if packetRetentionTicker != nil {

cmd/ingestor/maintenance.go

@@ -44,11 +44,13 @@ const pruneBatchTransmissions = 250
 // pins the plan.
 const pruneAgedTransmissionIDs = `SELECT id FROM transmissions WHERE first_seen < ? ORDER BY first_seen, id LIMIT ?`
 
-// The two statements of one prune batch. Child observations go first (no
-// CASCADE in SQLite).
+// The statements of one prune batch. Child rows go first (no CASCADE in
+// SQLite): observations, then the batch's route_mask_changes rows (#89; via
+// idx_route_mask_changes_tx), then the transmissions themselves.
 const (
-	pruneObservationsBatch  = `DELETE FROM observations WHERE transmission_id IN (` + pruneAgedTransmissionIDs + `)`
-	pruneTransmissionsBatch = `DELETE FROM transmissions WHERE id IN (` + pruneAgedTransmissionIDs + `)`
+	pruneObservationsBatch     = `DELETE FROM observations WHERE transmission_id IN (` + pruneAgedTransmissionIDs + `)`
+	pruneRouteMaskChangesBatch = `DELETE FROM route_mask_changes WHERE transmission_id IN (` + pruneAgedTransmissionIDs + `)`
+	pruneTransmissionsBatch    = `DELETE FROM transmissions WHERE id IN (` + pruneAgedTransmissionIDs + `)`
 )
 
 // PruneOldPackets deletes transmissions (and their child observations)
@@ -80,6 +82,9 @@ func (s *Store) PruneOldPackets(days int) (int64, error) {
 			if _, err := tx.Exec(pruneObservationsBatch, cutoff, pruneBatchTransmissions); err != nil {
 				return fmt.Errorf("prune observations: %w", err)
 			}
+			if _, err := tx.Exec(pruneRouteMaskChangesBatch, cutoff, pruneBatchTransmissions); err != nil {
+				return fmt.Errorf("prune route_mask_changes: %w", err)
+			}
 			res, err := tx.Exec(pruneTransmissionsBatch, cutoff, pruneBatchTransmissions)
 			if err != nil {
 				return fmt.Errorf("prune transmissions: %w", err)
@@ -104,6 +109,65 @@ func (s *Store) PruneOldPackets(days int) (int64, error) {
 	return total, nil
 }
 
+// routeMaskOrphanPruneBatch bounds one orphan-prune transaction. A var so
+// tests can exercise several batches.
+var routeMaskOrphanPruneBatch = 1000
+
+// PruneOrphanRouteMaskChanges deletes route_mask_changes rows whose
+// transmission no longer exists. PruneOldPackets already deletes a pruned
+// transmission's rows in the same batch; this catches any other deletion path
+// so the log never outlives its transmission. It never deletes a row of an
+// existing transmission: the log carries no time-based expiry, so a slow
+// server can never miss a change of a transmission it still holds.
+//
+// Each writer transaction examines the next routeMaskOrphanPruneBatch ids
+// (a primary-key range, one transmissions primary-key probe per row) and
+// deletes the orphans among them, so its work is bounded however far apart
+// the orphans are. The table is small in practice (22 rows on a
+// 496,798-transmission staging copy; at most three per transmission, one per
+// route bit it can gain). Returns the number of rows deleted.
+func (s *Store) PruneOrphanRouteMaskChanges() (int64, error) {
+	batch := routeMaskOrphanPruneBatch
+	if batch <= 0 {
+		batch = 1000
+	}
+	var total, after int64
+	for {
+		var examined, deleted int64
+		err := s.WriterTx("prune_route_mask_changes", func(tx *sql.Tx) error {
+			var last sql.NullInt64
+			if err := tx.QueryRow(`SELECT MAX(id), COUNT(*) FROM (
+				SELECT id FROM route_mask_changes WHERE id > ? ORDER BY id LIMIT ?)`,
+				after, batch).Scan(&last, &examined); err != nil {
+				return fmt.Errorf("scan route_mask_changes: %w", err)
+			}
+			if examined == 0 {
+				return nil
+			}
+			res, err := tx.Exec(`DELETE FROM route_mask_changes WHERE id > ? AND id <= ?
+				AND NOT EXISTS (SELECT 1 FROM transmissions t WHERE t.id = route_mask_changes.transmission_id)`,
+				after, last.Int64)
+			if err != nil {
+				return fmt.Errorf("prune orphan route_mask_changes: %w", err)
+			}
+			deleted, _ = res.RowsAffected()
+			after = last.Int64
+			return nil
+		})
+		if err != nil {
+			return total, err
+		}
+		total += deleted
+		if examined < int64(batch) {
+			break
+		}
+	}
+	if total > 0 {
+		log.Printf("[prune] deleted %d route_mask_changes rows of deleted transmissions", total)
+	}
+	return total, nil
+}
+
 // PruneOldClientReceptions deletes mobile client-RX coverage rows older than
 // `days` (by rx_at), and client_observers (companion names) whose last_seen has
 // aged out. This bounds the otherwise-unbounded client_receptions table the

cmd/ingestor/prune_chunked_test.go

@@ -248,9 +248,10 @@ func TestPruneAgedTransmissionIDsUsesFirstSeenIndex(t *testing.T) {
 
 	cutoff := time.Now().UTC().AddDate(0, 0, -5).Format(time.RFC3339)
 	for name, q := range map[string]string{
-		"batch subquery":       pruneAgedTransmissionIDs,
-		"observations delete":  pruneObservationsBatch,
-		"transmissions delete": pruneTransmissionsBatch,
+		"batch subquery":            pruneAgedTransmissionIDs,
+		"observations delete":       pruneObservationsBatch,
+		"route_mask_changes delete": pruneRouteMaskChangesBatch,
+		"transmissions delete":      pruneTransmissionsBatch,
 	} {
 		rows, err := store.db.Query("EXPLAIN QUERY PLAN "+q, cutoff, pruneBatchTransmissions)
 		if err != nil {
@@ -282,5 +283,10 @@ func TestPruneAgedTransmissionIDsUsesFirstSeenIndex(t *testing.T) {
 		if strings.Contains(plan, "TEMP B-TREE") {
 			t.Errorf("%s: plan sorts in a temp b-tree instead of walking the index in order: %s", name, plan)
 		}
+		// #89: the change rows of a batch are found through their index,
+		// not by scanning the change log.
+		if name == "route_mask_changes delete" && (!strings.Contains(plan, "idx_route_mask_changes_tx") || strings.Contains(plan, "SCAN route_mask_changes")) {
+			t.Errorf("%s: plan does not delete through idx_route_mask_changes_tx: %s", name, plan)
+		}
 	}
 }