fix(liveness): count only evidence that can be the node alone (#2260)

AGENTS.md

@@ -211,7 +211,8 @@ All colors MUST use CSS variables. Never hardcode `#hex` values outside of `:roo
 
 ### last_heard vs last_seen
 - `last_seen` = DB timestamp, only updates on adverts/direct upserts
-- `last_heard` = from in-memory packet store, updates on ALL traffic
+- `last_heard` = the newer of `last_seen` and the certain last heard from the in-memory packet store (#2260): the node's own packets plus relay hops whose prefix can only be this node. A 1-byte hop shared with other repeaters is NOT evidence, and neither is `last_relayed` (it counts those shared prefixes too). Liveness — status, Last Heard, the notify offline check — uses only this.
+- Status from `getNodeStatus(n)` is `'active'`, `'stale'` or, for a repeater or room, `'uncertain'`: nothing certain within the silent window, but `last_relayed` (its shared prefix in a path) is. Treat `'uncertain'` as not proven alive (dim it like stale); never as active.
 - Always prefer `n.last_heard || n.last_seen` for display and status calculation
 
 ### Packet Filter (packet-filter.js)

cmd/server/certain_heard.go

@@ -0,0 +1,156 @@
+package main
+
+import (
+	"strings"
+	"time"
+
+	"github.com/meshcore-analyzer/packetpath"
+)
+
+// Certain liveness (#2260). certainHeard holds, per pubkey, the newest
+// FirstSeen of a transmission that can only have come from that node:
+//
+//   - its own packets (decoded pubKey / srcPubKey), and
+//   - flood relay hops whose prefix, at the packet's path hash size, matches
+//     exactly one node that can appear in a path, and no retired one
+//     (inactive_nodes) that may still relay.
+//
+// A 1-byte hop shared by several repeaters is not evidence: the resolver may
+// still attribute it (affinity, geography) for analytics, but a silent node
+// must not stay "Active" on traffic from a namesake prefix. The firmware
+// lets the originator pick the hash size for the whole path
+// (Mesh::sendFlood) and every relay appends its own hash at that size
+// (Mesh::routeRecvPacket), so a multibyte repeater legitimately appears as a
+// 1-byte hop; only its uniqueness decides. Direct-route paths list the hops
+// still to come (Mesh::removeSelfFromPath), and TRACE reuses the path bytes
+// for SNR, so neither proves a hop transmitted.
+//
+// Liveness reads it: Last Heard, the node status, the nodes list and the
+// notify offline check. One entry per node that was ever certainly heard,
+// dropped with the node's byNode list on eviction.
+
+// noteCertainHeard records tx as certain evidence for pubkey. Caller holds
+// s.mu for writing.
+func (s *PacketStore) noteCertainHeard(pubkey string, tx *StoreTx) {
+	if pubkey == "" || tx == nil || tx.FirstSeen == "" {
+		return
+	}
+	if s.certainHeard == nil {
+		s.certainHeard = make(map[string]string)
+	}
+	if tx.FirstSeen > s.certainHeard[pubkey] {
+		s.certainHeard[pubkey] = tx.FirstSeen
+	}
+}
+
+// signatureFailed reports a decoded advert whose signature did not verify
+// (signatureValid false): a corrupted or forged advert is no evidence the
+// node transmitted. Absent means not checked, which is not a failure.
+func (tx *StoreTx) signatureFailed() bool {
+	v, ok := tx.ParsedDecoded()["signatureValid"].(bool)
+	return ok && !v
+}
+
+// relayHopHexLen is the hex length of one relay hop in tx's path, or 0 when
+// the path is no evidence of who transmitted: not a flood route, a TRACE, or
+// an unreadable header.
+func relayHopHexLen(tx *StoreTx) int {
+	if tx.PayloadType != nil && !packetpath.PathBytesAreHops(byte(*tx.PayloadType)) {
+		return 0
+	}
+	routeType, pathByte, ok := packetpath.PathByte(tx.RawHex)
+	if !ok || (routeType != packetpath.RouteFlood && routeType != packetpath.RouteTransportFlood) {
+		return 0
+	}
+	size := int(pathByte>>6) + 1
+	if size > 3 { // reserved (firmware Packet::isValidPathLen)
+		return 0
+	}
+	return 2 * size
+}
+
+// noteCertainRelays records tx for each resolved relay hop pubkey whose hop
+// prefix is unique among path-capable nodes in pm. pm nil records nothing.
+// Caller holds s.mu for writing.
+func (s *PacketStore) noteCertainRelays(tx *StoreTx, pks []string, pm *prefixMap) {
+	if pm == nil || len(pks) == 0 {
+		return
+	}
+	n := relayHopHexLen(tx)
+	if n == 0 {
+		return
+	}
+	for _, pk := range pks {
+		if uniqueHopPrefix(pk, n, pm) {
+			s.noteCertainHeard(pk, tx)
+		}
+	}
+}
+
+// uniqueHopPrefix reports whether pk's first n hex characters, a relay hop
+// at that hash size, match exactly one path-capable node in pm, pk itself,
+// and no retired one. A repeater that never advertised is unknown to both
+// and cannot be ruled out.
+func uniqueHopPrefix(pk string, n int, pm *prefixMap) bool {
+	if pm == nil || n == 0 || len(pk) < n {
+		return false
+	}
+	prefix := strings.ToLower(pk[:n])
+	if _, shared := pm.retired[prefix]; shared {
+		return false
+	}
+	c := pm.relayCandidates(prefix)
+	return len(c) == 1 && strings.EqualFold(c[0].PublicKey, pk)
+}
+
+// markRetired loads the hop prefixes (1 to 3 bytes) of the path-capable
+// nodes in inactive_nodes into pm.retired. Rows of nodes also in nodes are
+// skipped: retention copies a node there and nothing removes it when the
+// node is heard again, and a node is no namesake of itself. A database
+// without the table leaves it empty.
+func (pm *prefixMap) markRetired(db *DB) error {
+	if db == nil || !db.inactiveNodesPresent() {
+		return nil
+	}
+	rows, err := db.conn.Query(`SELECT public_key, COALESCE(role, '') FROM inactive_nodes
+		WHERE lower(public_key) NOT IN (SELECT lower(public_key) FROM nodes)`)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	retired := map[string]struct{}{}
+	for rows.Next() {
+		var pk, role string
+		if err := rows.Scan(&pk, &role); err != nil {
+			return err
+		}
+		if !canAppearInPath(role) {
+			continue
+		}
+		pk = strings.ToLower(pk)
+		for n := 2; n <= 6 && n <= len(pk); n += 2 {
+			retired[pk[:n]] = struct{}{}
+		}
+	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	pm.retired = retired
+	return nil
+}
+
+// lastHeardValue is a node's last_heard in /api/nodes and the node detail:
+// the newer of its DB last_seen (adverts; nil or unparsable counts as
+// never) and heard, its certain last heard from LastHeardMap (zero for
+// none). Before #2260 last_heard was a plain copy of last_seen.
+func lastHeardValue(lastSeen interface{}, heard time.Time) interface{} {
+	if heard.IsZero() {
+		return lastSeen
+	}
+	if s, ok := lastSeen.(string); ok {
+		if t, ok := parseRelayTS(s); ok && !heard.After(t) {
+			return lastSeen
+		}
+	}
+	return heard.UTC().Format("2006-01-02T15:04:05.000Z")
+}

cmd/server/certain_heard_loadpaths_test.go

@@ -0,0 +1,122 @@
+package main
+
+import (
+	"database/sql"
+	"strings"
+	"testing"
+	"time"
+)
+
+// Review (#2261): every load path must feed certainHeard. A nil prefix map
+// at any of the call sites in Load, loadChunk, scanAndMergeChunk, live
+// ingest or late observations would leave the unit tests green; these go
+// through the real paths and assert LastHeardMap.
+
+const lpRelay = "4f2a" + "5555555555555555555555555555555555555555555555555555555555555"
+
+// lpFixture is one flood transmission whose single 2-byte hop is lpRelay,
+// a registered repeater, observed a minute ago. withResolved keeps the
+// observation's resolved_path (the ingestor resolved the hop); otherwise it
+// is NULL and the server re-resolves from path_json. hopInFirstObs false
+// leaves the first observation without a path (for the late-observation
+// case).
+func lpFixture(t *testing.T, withResolved, hopInFirstObs bool) (*DB, *sql.DB, string) {
+	t.Helper()
+	path := createTestDBWithResolvedPath(t, 1, []string{lpRelay})
+	w, err := sql.Open("sqlite3", path+"?_journal_mode=WAL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { w.Close() })
+	at := time.Now().UTC().Add(-time.Minute).Format(time.RFC3339)
+	raw := "1541" + lpRelay[:4] + "00" // GRP_TXT, flood, path byte: 2-byte hashes x1
+	pathJSON, resolved := `["`+strings.ToUpper(lpRelay[:4])+`"]`, interface{}(`["`+lpRelay+`"]`)
+	if !withResolved {
+		resolved = nil
+	}
+	if !hopInFirstObs {
+		pathJSON, resolved = `[]`, nil
+	}
+	for _, q := range []struct {
+		sql  string
+		args []interface{}
+	}{
+		{`ALTER TABLE transmissions ADD COLUMN from_pubkey TEXT`, nil},
+		{`UPDATE transmissions SET raw_hex=?, route_type=1, payload_type=5, first_seen=?`, []interface{}{raw, at}},
+		{`UPDATE observations SET raw_hex=?, timestamp=?, path_json=?, resolved_path=?`, []interface{}{raw, at, pathJSON, resolved}},
+		{`INSERT INTO nodes (public_key, name, role) VALUES (?, 'Relay', 'repeater')`, []interface{}{lpRelay}},
+	} {
+		if _, err := w.Exec(q.sql, q.args...); err != nil {
+			t.Fatalf("%s: %v", q.sql, err)
+		}
+	}
+	db, err := OpenDB(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { db.conn.Close() })
+	return db, w, at
+}
+
+func TestCertainHeardFedByEveryLoadPath(t *testing.T) {
+	type mode struct {
+		name                        string
+		withResolved, hopInFirstObs bool
+		run                         func(t *testing.T, s *PacketStore, db *DB, w *sql.DB, at string)
+	}
+	load := func(t *testing.T, s *PacketStore, _ *DB, _ *sql.DB, _ string) {
+		if err := s.Load(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	chunk := func(t *testing.T, s *PacketStore, _ *DB, _ *sql.DB, _ string) {
+		if err := s.loadChunk(time.Now().Add(-time.Hour), time.Now()); err != nil {
+			t.Fatal(err)
+		}
+	}
+	chunked := func(t *testing.T, s *PacketStore, _ *DB, _ *sql.DB, _ string) {
+		if err := s.LoadChunked(1); err != nil {
+			t.Fatal(err)
+		}
+	}
+	live := func(t *testing.T, s *PacketStore, _ *DB, _ *sql.DB, _ string) {
+		s.IngestNewFromDB(0, 100)
+	}
+	late := func(t *testing.T, s *PacketStore, db *DB, w *sql.DB, at string) {
+		if err := s.Load(); err != nil {
+			t.Fatal(err)
+		}
+		if heard, _ := s.LastHeardMap([]string{lpRelay}); len(heard) != 0 {
+			t.Fatal("setup: the first observation carries no hop")
+		}
+		maxObs := db.GetMaxObservationID()
+		if _, err := w.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, snr, rssi, score, path_json, timestamp, raw_hex, resolved_path)
+			VALUES (2, 1, 'obs2', 'Obs2', 'RX', -9, -79, 5, ?, ?, '', NULL)`, `["`+strings.ToUpper(lpRelay[:4])+`"]`, at); err != nil {
+			t.Fatal(err)
+		}
+		s.IngestNewObservations(maxObs, 100)
+	}
+	for _, m := range []mode{
+		{"Load/resolved_path", true, true, load},
+		{"Load/path_json", false, true, load},
+		{"loadChunk/resolved_path", true, true, chunk},
+		{"loadChunk/path_json", false, true, chunk},
+		{"LoadChunked/resolved_path", true, true, chunked},
+		{"LoadChunked/path_json", false, true, chunked},
+		{"IngestNewFromDB", false, true, live},
+		{"IngestNewObservations", false, false, late},
+	} {
+		t.Run(m.name, func(t *testing.T) {
+			db, w, at := lpFixture(t, m.withResolved, m.hopInFirstObs)
+			s := NewPacketStore(db, &PacketStoreConfig{})
+			s.useResolvedPathIndex = true
+			s.initResolvedPathIndex()
+			m.run(t, s, db, w, at)
+			heard, _ := s.LastHeardMap([]string{lpRelay})
+			want, _ := time.Parse(time.RFC3339, at)
+			if !heard[lpRelay].Equal(want) {
+				t.Fatalf("LastHeardMap[relay] = %v; want %v: this path does not feed certainHeard", heard[lpRelay], want)
+			}
+		})
+	}
+}