diff --git a/cmd/server/neighbor_api.go b/cmd/server/neighbor_api.go index 8934989b7..0b953a254 100644 --- a/cmd/server/neighbor_api.go +++ b/cmd/server/neighbor_api.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "log" "net/http" "sort" "strconv" @@ -501,13 +502,31 @@ func observerList(m map[string]bool) []string { // buildNodeInfoMap returns a map of lowercase pubkey → nodeInfo for name/role lookups. func (s *Server) buildNodeInfoMap() map[string]nodeInfo { + m, _ := s.buildNodeInfoMapOK() + return m +} + +// buildNodeInfoMapOK is buildNodeInfoMap with the completeness of the map +// reported. ok=false means part of it could not be read: the shared node +// cache had nothing complete to serve, or the observers enrichment query +// failed or was interrupted mid-iteration. That second case is the local +// analogue of the upstream reachNodeScopes root cause (#593, upstream 2277): +// a follow-up query in the same build whose error silently becomes empty +// strings, so a Reach report with blanked observer names would be cached for +// the full 5-minute TTL and keep being served after the DB recovered +// (round-2 review F3). +// +// Callers that cache what they derive from the map must use this variant and +// fail the build; the endpoints that render it live keep using +// buildNodeInfoMap, where a degraded name is preferable to a 500. +func (s *Server) buildNodeInfoMapOK() (map[string]nodeInfo, bool) { if s.store == nil { - return nil + return nil, false } // FirstSeen is folded into getAllNodes (and therefore into the 30s // node cache) so callers like /api/nodes/{pk}/reach get the field // without a per-request SELECT — fixes #1627 r3 regression. - nodes, _ := s.store.getCachedNodesAndPM() + nodes, _, ok := s.store.getCachedNodesAndPMOK() m := make(map[string]nodeInfo, len(nodes)) for _, n := range nodes { m[strings.ToLower(n.PublicKey)] = n @@ -515,24 +534,38 @@ func (s *Server) buildNodeInfoMap() map[string]nodeInfo { // Enrich observer-only nodes: if an observer pubkey isn't already in the // map (i.e. it's not also a repeater/companion), add it with role "observer". - if s.db != nil { + if s.db != nil && s.db.conn != nil { rows, err := s.db.conn.Query("SELECT id, name FROM observers") - if err == nil { - defer rows.Close() - for rows.Next() { - var id, name string - if rows.Scan(&id, &name) != nil { - continue - } - key := strings.ToLower(id) - if _, exists := m[key]; !exists { - m[key] = nodeInfo{PublicKey: id, Name: name, Role: "observer"} - } + if err != nil { + log.Printf("[neighbors] buildNodeInfoMap observers query failed: %v", err) + return m, false + } + defer rows.Close() + // A row that won't Scan is malformed, not an interruption: the + // observers table admits a NULL name, and reach_rank.go's roster + // scan is pinned to skipping those too. Counted so it isn't silent. + var skipped int + for rows.Next() { + var id, name string + if rows.Scan(&id, &name) != nil { + skipped++ + continue + } + key := strings.ToLower(id) + if _, exists := m[key]; !exists { + m[key] = nodeInfo{PublicKey: id, Name: name, Role: "observer"} } } + if skipped > 0 { + log.Printf("[neighbors] buildNodeInfoMap discarded %d malformed observers rows", skipped) + } + if err := rows.Err(); err != nil { + log.Printf("[neighbors] buildNodeInfoMap observers iteration failed: %v", err) + return m, false + } } - return m + return m, ok } // dedupPrefixEntries merges unresolved prefix entries with resolved pubkey entries diff --git a/cmd/server/neighbor_api_test.go b/cmd/server/neighbor_api_test.go index d79349dfe..3917ab9cd 100644 --- a/cmd/server/neighbor_api_test.go +++ b/cmd/server/neighbor_api_test.go @@ -630,9 +630,9 @@ func TestGetAllNodes_FirstSeenSchemaFallback(t *testing.T) { defer db.conn.Close() store := NewPacketStore(db, nil) - nodes := store.getAllNodes() - if len(nodes) != 1 { - t.Fatalf("expected 1 row from fallback rung, got %d", len(nodes)) + nodes, ok := store.getAllNodes() + if !ok || len(nodes) != 1 { + t.Fatalf("expected 1 row from fallback rung, got %d (ok=%v)", len(nodes), ok) } n := nodes[0] if n.PublicKey != "BBBB2222" { diff --git a/cmd/server/node_cache_scan_interrupted_test.go b/cmd/server/node_cache_scan_interrupted_test.go new file mode 100644 index 000000000..a0116cc13 --- /dev/null +++ b/cmd/server/node_cache_scan_interrupted_test.go @@ -0,0 +1,482 @@ +package main + +import ( + "context" + "database/sql" + "database/sql/driver" + "encoding/json" + "fmt" + "net/http" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + _ "modernc.org/sqlite" +) + +// Issue #593 (port of upstream PR 2277): a query interrupted partway through +// (lock contention, I/O, a cancelled context) must never be mistaken for "no +// more rows" and cached as if it were a complete, successful read. This file +// covers getAllNodes/getCachedNodesAndPM — the node cache that +// buildNodeInfoMap and computeNodeReach's prefixMap both depend on — which, +// unlike reach_rank.go's scanRankRows/loadDegreeSnapshot, did not check +// rows.Err() after its scan loop and discarded each row's Scan error. +// +// nodesScanFailRows/nodesScanFailConn wrap the real modernc.org/sqlite driver +// (same technique as ping_score_bulk_test.go's countingConn) so the "nodes" +// table query runs for real, but its driver.Rows.Next fails after a chosen +// number of rows — a deterministic stand-in for a connection that drops mid +// scan, rather than one that fails before the query even starts (already +// handled by the existing err != nil checks around s.db.conn.Query). + +type nodesScanFailDriver struct{ inner driver.Driver } + +func (d *nodesScanFailDriver) Open(name string) (driver.Conn, error) { + c, err := d.inner.Open(name) + if err != nil { + return nil, err + } + return &nodesScanFailConn{Conn: c}, nil +} + +type nodesScanFailConn struct{ driver.Conn } + +// nodesScanFailAfterRows: -1 disables fault injection; >= 0 makes the +// (n+1)th Next() call on the nodes-table query fail instead of returning a +// row, however many real rows already came back successfully. +var nodesScanFailAfterRows int64 = -1 + +func setNodesScanFailAfterRows(n int64) { atomic.StoreInt64(&nodesScanFailAfterRows, n) } +func clearNodesScanFailAfterRows() { atomic.StoreInt64(&nodesScanFailAfterRows, -1) } + +// The same two faults for buildNodeInfoMap's observer-enrichment query +// (#593 F3). The match is the exact query text so the fault hits only that +// read: scanReachRows joins the table as "LEFT JOIN observers obs" and +// reach_rank's own roster scan adds a WHERE clause, so both stay healthy +// and the test isolates the enrichment query. +// +// observersScanFailAfterRows: -1 disables; >= 0 makes the (n+1)th Next() fail. +// observersQueryFails: 1 makes the query itself fail, before any row. +var ( + observersScanFailAfterRows int64 = -1 + observersQueryFails int64 +) + +// observersEnrichQuery must stay identical to buildNodeInfoMap's query text. +const observersEnrichQuery = "SELECT id, name FROM observers" + +func setObserversScanFailAfterRows(n int64) { atomic.StoreInt64(&observersScanFailAfterRows, n) } +func setObserversQueryFails() { atomic.StoreInt64(&observersQueryFails, 1) } +func clearObserversFaults() { + atomic.StoreInt64(&observersScanFailAfterRows, -1) + atomic.StoreInt64(&observersQueryFails, 0) +} + +func (c *nodesScanFailConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) { + qc, ok := c.Conn.(driver.QueryerContext) + if !ok { + return nil, driver.ErrSkip + } + if atomic.LoadInt64(&observersQueryFails) == 1 && query == observersEnrichQuery { + return nil, fmt.Errorf("injected observers query failure") + } + rows, err := qc.QueryContext(ctx, query, args) + if err != nil { + return nil, err + } + if failAfter := atomic.LoadInt64(&nodesScanFailAfterRows); failAfter >= 0 && strings.Contains(query, "FROM nodes") { + return &nodesScanFailRows{Rows: rows, failAfter: failAfter}, nil + } + if failAfter := atomic.LoadInt64(&observersScanFailAfterRows); failAfter >= 0 && query == observersEnrichQuery { + return &nodesScanFailRows{Rows: rows, failAfter: failAfter}, nil + } + return rows, nil +} + +type nodesScanFailRows struct { + driver.Rows + failAfter int64 + n int64 +} + +func (r *nodesScanFailRows) Next(dest []driver.Value) error { + if r.n >= r.failAfter { + return fmt.Errorf("injected mid-scan failure after %d rows", r.failAfter) + } + err := r.Rows.Next(dest) + if err == nil { + r.n++ + } + return err +} + +var registerNodesScanFailDriverOnce sync.Once + +func registerNodesScanFailDriver() { + registerNodesScanFailDriverOnce.Do(func() { + tmp, err := sql.Open("sqlite", ":memory:") + if err != nil { + panic(err) + } + realDriver := tmp.Driver() + tmp.Close() + sql.Register("sqlite-nodes-scan-fail", &nodesScanFailDriver{inner: realDriver}) + }) +} + +var nodesScanFailDBNameCounter int64 + +type nodesScanFailRow struct{ pk, name string } + +// newNodesScanFailDB builds an isolated DB (own driver registration, like +// setupPacketPathCountingDB) seeded with the given node rows, on the +// fault-injecting driver above. rows defaults to 3 arbitrary nodes when nil. +func newNodesScanFailDB(t *testing.T, rows []nodesScanFailRow) *DB { + t.Helper() + registerNodesScanFailDriver() + clearNodesScanFailAfterRows() + clearObserversFaults() + t.Cleanup(clearNodesScanFailAfterRows) + t.Cleanup(clearObserversFaults) + name := fmt.Sprintf("nodesfail%d", atomic.AddInt64(&nodesScanFailDBNameCounter, 1)) + conn, err := sql.Open("sqlite-nodes-scan-fail", "file:"+name+"?mode=memory&cache=shared") + if err != nil { + t.Fatal(err) + } + conn.SetMaxOpenConns(1) + t.Cleanup(func() { conn.Close() }) + stmts := []string{ + `CREATE TABLE nodes (public_key TEXT, name TEXT, role TEXT, lat REAL, lon REAL, last_seen TEXT, first_seen TEXT, advert_count INTEGER)`, + } + for _, s := range stmts { + if _, err := conn.Exec(s); err != nil { + t.Fatal(err) + } + } + if rows == nil { + rows = []nodesScanFailRow{{pk64("aa"), "A"}, {pk64("bb"), "B"}, {pk64("cc"), "C"}} + } + for _, r := range rows { + if _, err := conn.Exec(`INSERT INTO nodes (public_key, name, role, lat, lon, last_seen, first_seen, advert_count) VALUES (?, ?, 'repeater', 50.0, 5.0, '2026-06-01T00:00:00Z', '2026-06-01T00:00:00Z', 1)`, r.pk, r.name); err != nil { + t.Fatal(err) + } + } + db := &DB{conn: conn} + db.isV3Flag.forceTrue() + return db +} + +// TestGetAllNodes_MidScanFailureDoesNotTruncate is the direct, store-level +// reproduction: a connection that fails on its 2nd row (of 3) must make +// getAllNodes report failure, never a silently truncated 1-row list. +func TestGetAllNodes_MidScanFailureDoesNotTruncate(t *testing.T) { + db := newNodesScanFailDB(t, nil) + store := &PacketStore{db: db} + + // Baseline: no injected failure, all 3 rows come back. + nodes, ok := store.getAllNodes() + if !ok || len(nodes) != 3 { + t.Fatalf("baseline: ok=%v len=%d, want ok=true len=3", ok, len(nodes)) + } + + setNodesScanFailAfterRows(1) // 1 good row, then the connection drops + nodes, ok = store.getAllNodes() + if ok { + t.Fatalf("mid-scan failure after 1/3 rows: ok=true with %d nodes, want ok=false (a truncated read must not be reported as a complete one)", len(nodes)) + } + if nodes != nil { + t.Fatalf("mid-scan failure: got %d nodes, want nil on failure", len(nodes)) + } +} + +// TestGetCachedNodesAndPM_MidScanFailureKeepsStaleCompleteSnapshot is the +// caching-layer reproduction: once a complete snapshot is cached, a refresh +// that fails partway through must not replace it with a partial or empty +// one — every consumer (buildNodeInfoMap, Reach's reliableTokens) would +// otherwise silently lose nodes for the rest of the 30s TTL. +func TestGetCachedNodesAndPM_MidScanFailureKeepsStaleCompleteSnapshot(t *testing.T) { + db := newNodesScanFailDB(t, nil) + store := &PacketStore{db: db} + + nodes, pm := store.getCachedNodesAndPM() + if len(nodes) != 3 || pm == nil { + t.Fatalf("warm-up: got %d nodes, pm=%v, want 3 nodes and a non-nil pm", len(nodes), pm) + } + + // Expire the cache's TTL and make the next scan fail after 1 of 3 rows. + store.cacheMu.Lock() + store.nodeCacheTime = time.Now().Add(-time.Minute) + store.cacheMu.Unlock() + setNodesScanFailAfterRows(1) + + nodes2, pm2 := store.getCachedNodesAndPM() + if len(nodes2) != 3 || pm2 == nil { + t.Fatalf("after a failed refresh: got %d nodes, pm=%v, want the stale complete snapshot (3 nodes) preserved", len(nodes2), pm2) + } + + // Recovery: once the connection is healthy again, the next refresh + // (after the TTL, which a failed refresh deliberately does not bump) + // replaces the snapshot normally. + clearNodesScanFailAfterRows() + store.cacheMu.Lock() + store.nodeCacheTime = time.Now().Add(-time.Minute) + store.cacheMu.Unlock() + nodes3, pm3 := store.getCachedNodesAndPM() + if len(nodes3) != 3 || pm3 == nil { + t.Fatalf("after recovery: got %d nodes, pm=%v, want 3", len(nodes3), pm3) + } +} + +// TestNodeReach_NodeCacheRefreshFailureDoesNotCacheADegradedReport is the +// end-to-end reproduction through the Reach HTTP handler: a node-cache +// refresh that fails mid-scan while a Reach request is building its (5 +// minute TTL) cached response must not leave that response's attribution +// silently empty — the fix in getCachedNodesAndPM keeps serving the prior +// complete node snapshot, so Reach's own scan (unaffected by this failure) +// still resolves reliable tokens and links correctly. +func TestNodeReach_NodeCacheRefreshFailureDoesNotCacheADegradedReport(t *testing.T) { + reachDB, target := newReachIntegrationDB(t, `["AABB","01FA","CCDD"]`) + defer reachDB.conn.Close() + + // A second, fault-injectable connection backs the *shared* node cache + // (getAllNodes/getCachedNodesAndPM) that computeNodeReach's prefixMap + // comes from; its pubkeys must match newReachIntegrationDB's target (N) + // and its two neighbours (A, B) so reliableTokens/attributeDirections + // can resolve the hop tokens the Reach scan reads from reachDB. + nodesDB := newNodesScanFailDB(t, []nodesScanFailRow{ + {target, "N"}, {pk64("aabb"), "A"}, {pk64("ccdd"), "B"}, + }) + cfg := &Config{} + store := newTestStoreWithDB(t, reachDB, cfg) + store.db = nodesDB // getAllNodes/getCachedNodesAndPM read from the fault-injectable DB + srv := &Server{store: store, db: reachDB, cfg: cfg, perfStats: NewPerfStats()} + resetReachState(t, srv) + + // Warm both caches with a healthy request. + warm := serveReach(srv, "/api/nodes/"+target+"/reach?days=30") + if warm.Code != http.StatusOK { + t.Fatalf("warm-up: status=%d body=%s", warm.Code, warm.Body.String()) + } + + // Force the shared node-cache's next refresh to fail mid-scan, expire + // its TTL, and clear the Reach response cache so the next request is a + // genuinely cold compute that must consult the (now-failing) node cache. + store.cacheMu.Lock() + store.nodeCacheTime = time.Now().Add(-time.Minute) + store.cacheMu.Unlock() + setNodesScanFailAfterRows(1) + clearReachCache(t, srv) + + rr := serveReach(srv, "/api/nodes/"+target+"/reach?days=30") + if rr.Code != http.StatusOK { + t.Fatalf("after node-cache refresh failure: status=%d body=%s", rr.Code, rr.Body.String()) + } + var resp NodeReachResponse + if err := json.Unmarshal(rr.Body.Bytes(), &resp); err != nil { + t.Fatalf("bad json: %v (%s)", err, rr.Body.String()) + } + // The target's own reliable path-hash token still resolves and both + // neighbours are still attributed (they come from the stale-but-complete + // node cache preserved by the fix) — the report is not silently + // degraded to zero links/observers by the node-cache refresh failure. + if !contains(resp.ReliableTokens, "01FA") || len(resp.Links) != 2 { + t.Fatalf("reach report degraded after a node-cache refresh failure: tokens=%v links=%d, want 01FA token and 2 links", resp.ReliableTokens, len(resp.Links)) + } +} + +// newReachFailDriverDB builds the full newReachIntegrationDB schema on the +// fault-injecting driver, so one connection backs both the shared node cache +// (getAllNodes) and the Reach request path (scanReachRows, the observer +// enrichment). #593 F2/F3. +func newReachFailDriverDB(t *testing.T, obsPath string) (*DB, string) { + t.Helper() + registerNodesScanFailDriver() + clearNodesScanFailAfterRows() + clearObserversFaults() + t.Cleanup(clearNodesScanFailAfterRows) + t.Cleanup(clearObserversFaults) + name := fmt.Sprintf("reachfail%d", atomic.AddInt64(&nodesScanFailDBNameCounter, 1)) + conn, err := sql.Open("sqlite-nodes-scan-fail", "file:"+name+"?mode=memory&cache=shared") + if err != nil { + t.Fatal(err) + } + conn.SetMaxOpenConns(1) + t.Cleanup(func() { conn.Close() }) + return seedReachIntegrationDB(t, conn, obsPath) +} + +// reachCacheLen reports how many entries the Reach response cache holds, so a +// test can assert a failed build cached nothing at all (rather than only that +// this one request returned an error). +func reachCacheLen(srv *Server) int { + srv.reach.cacheMu.Lock() + defer srv.reach.cacheMu.Unlock() + return len(srv.reach.cache) +} + +// TestGetAllNodes_MalformedRowIsSkippedNotFatal — review round 2, F1. A row +// the scan cannot decode (nodes.public_key is a bare TEXT PRIMARY KEY, so +// SQLite permits one NULL; a non-numeric lat/lon is the same class) is a +// malformed row, not an interrupted read. It must be skipped, logged and +// counted the way scanReachRows already discards malformed rows — never turn +// into a permanent ok=false that disables the shared node cache for every +// caller for the lifetime of the process. +func TestGetAllNodes_MalformedRowIsSkippedNotFatal(t *testing.T) { + db := newNodesScanFailDB(t, nil) + if _, err := db.conn.Exec(`INSERT INTO nodes (public_key, name, role, lat, lon, last_seen, first_seen, advert_count) VALUES (NULL, 'bad', 'repeater', 50.0, 5.0, '2026-06-01T00:00:00Z', '2026-06-01T00:00:00Z', 1)`); err != nil { + t.Fatal(err) + } + store := &PacketStore{db: db} + + nodes, ok := store.getAllNodes() + if !ok { + t.Fatalf("one NULL public_key row made getAllNodes report ok=false; a malformed row must be skipped, not fail the whole read") + } + if len(nodes) != 3 { + t.Fatalf("got %d nodes, want the 3 well-formed rows (the NULL-pubkey row skipped)", len(nodes)) + } + + // ...and the cache is actually published, so later callers are cache + // hits instead of re-running the full nodes SELECT on every request. + cached, pm := store.getCachedNodesAndPM() + if len(cached) != 3 || pm == nil { + t.Fatalf("getCachedNodesAndPM: got %d nodes pm=%v, want 3 nodes and a non-nil pm", len(cached), pm) + } + store.cacheMu.Lock() + published := store.nodeCache != nil && store.nodePM != nil + store.cacheMu.Unlock() + if !published { + t.Fatalf("node cache was not published after a load containing one malformed row — every caller would re-query forever") + } + // The well-formed repeaters are resolvable through the published prefix + // map, i.e. the snapshot is usable and not just non-empty. + if got := uniqueResolve(pm, "aa"); got != pk64("aa") { + t.Fatalf("uniqueResolve(\"aa\")=%q, want %q", got, pk64("aa")) + } +} + +// TestGetCachedNodesAndPM_ColdStartFailureReportsNotOKWithUsableEmptyPM — +// review round 2, F2/F4. On a cold start (nothing cached yet) a failed scan +// has no complete snapshot to fall back on. Two things must hold: the +// fallback *prefixMap is non-nil and usable (every caller dereferences it), +// and the failure is reported to callers that would otherwise cache a result +// built on an empty map. +func TestGetCachedNodesAndPM_ColdStartFailureReportsNotOKWithUsableEmptyPM(t *testing.T) { + db := newNodesScanFailDB(t, nil) + store := &PacketStore{db: db} + setNodesScanFailAfterRows(0) // fails before any row, with nothing cached yet + + nodes, pm, ok := store.getCachedNodesAndPMOK() + if ok { + t.Fatalf("cold-start scan failure: ok=true, want false (there is no complete snapshot to serve)") + } + if len(nodes) != 0 { + t.Fatalf("cold-start scan failure: got %d nodes, want none", len(nodes)) + } + if pm == nil { + t.Fatalf("cold-start scan failure returned a nil *prefixMap; callers dereference it unconditionally") + } + // Usable, not merely non-nil: these are the two calls Reach makes. + if got := uniqueResolve(pm, "aa"); got != "" { + t.Fatalf("uniqueResolve on the empty fallback map = %q, want \"\"", got) + } + if got := pm.without(map[string]bool{pk64("aa"): true}); got == nil { + t.Fatalf("pm.without returned nil on the empty fallback map") + } + + // The ok-dropping wrapper keeps its old shape for the many callers that + // cannot act on a failure. + nodes2, pm2 := store.getCachedNodesAndPM() + if len(nodes2) != 0 || pm2 == nil { + t.Fatalf("getCachedNodesAndPM wrapper: got %d nodes pm=%v, want 0 nodes and a non-nil pm", len(nodes2), pm2) + } +} + +// TestNodeReach_ColdStartNodeCacheFailureIsA500AndNotCached — review round 2, +// F2. With nothing cached yet, a failed nodes scan leaves Reach with an empty +// prefix map. If the target is *also* an observer, buildNodeInfoMap still +// finds it through the observer enrichment, so the build "succeeds" and a +// report with zero reliable tokens and zero links gets cached for the full +// 5-minute TTL — surviving the DB's recovery. The failure must reach the +// handler instead: 500, nothing cached. +func TestNodeReach_ColdStartNodeCacheFailureIsA500AndNotCached(t *testing.T) { + reachDB, target := newReachFailDriverDB(t, `["AABB","01FA","CCDD"]`) + // The target is an observer as well as a node — the case where the + // empty-node-cache failure stays invisible to the Reach build. + if _, err := reachDB.conn.Exec(`INSERT INTO observers (id, name) VALUES (?, 'N-obs')`, target); err != nil { + t.Fatal(err) + } + cfg := &Config{} + store := newTestStoreWithDB(t, reachDB, cfg) + srv := &Server{store: store, db: reachDB, cfg: cfg, perfStats: NewPerfStats()} + resetReachState(t, srv) + + setNodesScanFailAfterRows(0) // cold start: the very first nodes scan fails + + rr := serveReach(srv, "/api/nodes/"+target+"/reach?days=30") + if rr.Code != http.StatusInternalServerError { + t.Fatalf("cold-start node-cache failure: status=%d body=%s, want 500 (a report built on an empty node cache must not be served or cached)", rr.Code, rr.Body.String()) + } + if n := reachCacheLen(srv); n != 0 { + t.Fatalf("reach cache holds %d entries after a failed build, want 0", n) + } + + // Recovery on the very next request, with no TTL to wait out. + clearNodesScanFailAfterRows() + ok := serveReach(srv, "/api/nodes/"+target+"/reach?days=30") + if ok.Code != http.StatusOK { + t.Fatalf("after recovery: status=%d body=%s, want 200", ok.Code, ok.Body.String()) + } + var resp NodeReachResponse + if err := json.Unmarshal(ok.Body.Bytes(), &resp); err != nil { + t.Fatalf("bad json: %v (%s)", err, ok.Body.String()) + } + if !contains(resp.ReliableTokens, "01FA") || len(resp.Links) != 2 { + t.Fatalf("after recovery: tokens=%v links=%d, want the 01FA token and 2 links", resp.ReliableTokens, len(resp.Links)) + } +} + +// TestNodeReach_ObserverEnrichmentInterruptedIsA500AndNotCached — review +// round 2, F3. buildNodeInfoMap's follow-up "SELECT id, name FROM observers" +// is the closest local analogue of the upstream reachNodeScopes root cause: a +// second query in the same build whose error turns into empty strings. Both +// an outright Query failure and a mid-iteration failure must stop the Reach +// build, not cache a report with observer names silently blanked for 5 +// minutes. +func TestNodeReach_ObserverEnrichmentInterruptedIsA500AndNotCached(t *testing.T) { + for _, tc := range []struct { + name string + inject func() + }{ + {"query fails", setObserversQueryFails}, + {"fails mid-scan", func() { setObserversScanFailAfterRows(0) }}, + } { + t.Run(tc.name, func(t *testing.T) { + reachDB, target := newReachFailDriverDB(t, `["AABB","01FA","CCDD"]`) + if _, err := reachDB.conn.Exec(`INSERT INTO observers (id, name) VALUES (?, 'OBS-NAMED')`, pk64("eeff")); err != nil { + t.Fatal(err) + } + cfg := &Config{} + store := newTestStoreWithDB(t, reachDB, cfg) + srv := &Server{store: store, db: reachDB, cfg: cfg, perfStats: NewPerfStats()} + resetReachState(t, srv) + + tc.inject() + rr := serveReach(srv, "/api/nodes/"+target+"/reach?days=30") + if rr.Code != http.StatusInternalServerError { + t.Fatalf("interrupted observer enrichment: status=%d body=%s, want 500", rr.Code, rr.Body.String()) + } + if n := reachCacheLen(srv); n != 0 { + t.Fatalf("reach cache holds %d entries after a failed build, want 0", n) + } + + clearObserversFaults() + ok := serveReach(srv, "/api/nodes/"+target+"/reach?days=30") + if ok.Code != http.StatusOK { + t.Fatalf("after recovery: status=%d body=%s, want 200", ok.Code, ok.Body.String()) + } + }) + } +} diff --git a/cmd/server/node_reach.go b/cmd/server/node_reach.go index 9b4061a0b..2bb5f895a 100644 --- a/cmd/server/node_reach.go +++ b/cmd/server/node_reach.go @@ -708,12 +708,23 @@ func (s *Server) computeNodeReach(ctx context.Context, pubkey string, days int) if s.store == nil || s.db == nil || s.db.conn == nil { return NodeReachResponse{}, false, nil } - nodeMap := s.buildNodeInfoMap() + // An incomplete lookup table (interrupted node-cache load, or an + // interrupted observers enrichment) is a backend failure, not a node + // with no neighbours: returning it as ok would cache a report with + // missing links or blanked observer names for the full response TTL, + // outliving the DB's recovery (#593 round-2 review F2/F3). + nodeMap, mapOK := s.buildNodeInfoMapOK() + if !mapOK { + return NodeReachResponse{}, false, errors.New("reach: node lookup table incomplete (interrupted read)") + } self, found := nodeMap[pubkey] if !found { return NodeReachResponse{}, false, nil } - _, pm := s.store.getCachedNodesAndPM() + _, pm, pmOK := s.store.getCachedNodesAndPMOK() + if !pmOK { + return NodeReachResponse{}, false, errors.New("reach: node snapshot unavailable (interrupted read)") + } tokens := reliableTokens(pubkey, pm) since := time.Now().UTC().Add(-time.Duration(days) * 24 * time.Hour) diff --git a/cmd/server/node_reach_endpoint_test.go b/cmd/server/node_reach_endpoint_test.go index 9fbf8690e..2f59476c2 100644 --- a/cmd/server/node_reach_endpoint_test.go +++ b/cmd/server/node_reach_endpoint_test.go @@ -63,6 +63,14 @@ func newReachIntegrationDB(t *testing.T, obsPath string) (*DB, string) { if err != nil { t.Fatal(err) } + return seedReachIntegrationDB(t, conn, obsPath) +} + +// seedReachIntegrationDB creates and fills the schema above on an +// already-open connection, so a test can supply one on a fault-injecting +// driver instead of the plain one (#593 F3). +func seedReachIntegrationDB(t *testing.T, conn *sql.DB, obsPath string) (*DB, string) { + t.Helper() n := pk64("01fa") // target — unique 2-byte token "01fa" a := pk64("aabb") // predecessor → we hear A b := pk64("ccdd") // successor → B hears us diff --git a/cmd/server/store.go b/cmd/server/store.go index 48f9575be..07dda2b96 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -7818,7 +7818,20 @@ func (s *PacketStore) logSchemaDegradationOnce(msg string) { } } -func (s *PacketStore) getAllNodes() []nodeInfo { +// getAllNodes returns every row of the nodes table, or ok=false if the read +// was interrupted — never a silently truncated slice. A genuine mid-scan DB +// error (lock contention, I/O) must not be mistaken for "no more rows": +// rows.Next() ends the loop the same way either way, so only rows.Err() +// after the loop tells them apart (#593, port of upstream #2277). +// +// ok=false means the read failed: the deepest schema rung's Query errored, +// or rows.Err() reported an interruption. A row that fails to Scan is a +// different thing — a malformed row (nodes.public_key is a bare TEXT +// PRIMARY KEY, so SQLite admits one NULL, and a non-numeric lat/lon fails +// the same way). Those are skipped, logged and counted the way scanReachRows +// already discards malformed rows, so one bad row cannot permanently disable +// the shared node cache for every caller (round-2 review F1). +func (s *PacketStore) getAllNodes() ([]nodeInfo, bool) { // Schema probe: try richest → leanest. Logs a one-shot warning at each // rung when we fall back to a thinner schema so operators see that a // column is missing and the new tiebreak features are degraded. See @@ -7843,13 +7856,14 @@ func (s *PacketStore) getAllNodes() []nodeInfo { rows, err = s.db.conn.Query("SELECT public_key, name, role, lat, lon FROM nodes") hasLastSeen = false if err != nil { - return nil + return nil, false } } } } defer rows.Close() var nodes []nodeInfo + var skipped int // malformed rows discarded — surfaced below so schema/ingest bugs aren't silent for rows.Next() { var pk string var name, role sql.NullString @@ -7857,14 +7871,19 @@ func (s *PacketStore) getAllNodes() []nodeInfo { var lastSeen sql.NullString var advertCount sql.NullInt64 var firstSeen sql.NullString + var scanErr error if hasFirstSeen { - rows.Scan(&pk, &name, &role, &lat, &lon, &lastSeen, &advertCount, &firstSeen) + scanErr = rows.Scan(&pk, &name, &role, &lat, &lon, &lastSeen, &advertCount, &firstSeen) } else if hasAdvertCount { - rows.Scan(&pk, &name, &role, &lat, &lon, &lastSeen, &advertCount) + scanErr = rows.Scan(&pk, &name, &role, &lat, &lon, &lastSeen, &advertCount) } else if hasLastSeen { - rows.Scan(&pk, &name, &role, &lat, &lon, &lastSeen) + scanErr = rows.Scan(&pk, &name, &role, &lat, &lon, &lastSeen) } else { - rows.Scan(&pk, &name, &role, &lat, &lon) + scanErr = rows.Scan(&pk, &name, &role, &lat, &lon) + } + if scanErr != nil { + skipped++ + continue } n := nodeInfo{PublicKey: pk, Name: nullStrVal(name), Role: nullStrVal(role)} if hasFirstSeen && firstSeen.Valid { @@ -7887,7 +7906,14 @@ func (s *PacketStore) getAllNodes() []nodeInfo { } nodes = append(nodes, n) } - return nodes + if skipped > 0 { + log.Printf("[store] getAllNodes discarded %d malformed nodes rows (kept %d)", skipped, len(nodes)) + } + if err := rows.Err(); err != nil { + log.Printf("[store] getAllNodes iteration failed: %v", err) + return nil, false + } + return nodes, true } type prefixMap struct { @@ -7970,28 +7996,62 @@ func buildPrefixMap(nodes []nodeInfo) *prefixMap { // Call WITHOUT s.mu: cold/expired reads wait for SQL. A request keeps its // snapshot even if invalidated concurrently, but never republishes it then. func (s *PacketStore) getCachedNodesAndPM() ([]nodeInfo, *prefixMap) { + nodes, pm, _ := s.getCachedNodesAndPMOK() + return nodes, pm +} + +// getCachedNodesAndPMOK is getCachedNodesAndPM with the completeness of the +// returned snapshot reported. ok=false means the refresh failed and there was +// no complete snapshot to fall back on (a cold start), so what comes back is +// the empty-but-valid fallback rather than the whole nodes table. Callers +// that cache what they derive from it — Reach, whose response cache has a +// 5-minute TTL — must treat that as a backend failure instead of publishing +// a report with no links in it (round-2 review F2). The many callers that +// cannot act on a failure keep using getCachedNodesAndPM. +func (s *PacketStore) getCachedNodesAndPMOK() ([]nodeInfo, *prefixMap, bool) { s.cacheMu.Lock() if s.nodeCache != nil && time.Since(s.nodeCacheTime) < 30*time.Second { nodes, pm := s.nodeCache, s.nodePM s.cacheMu.Unlock() - return nodes, pm + return nodes, pm, true } s.cacheMu.Unlock() type nodeSnapshot struct { nodes []nodeInfo pm *prefixMap + ok bool } v, _, _ := s.nodeCacheLoad.Do("nodes", func() (interface{}, error) { s.cacheMu.Lock() if s.nodeCache != nil && time.Since(s.nodeCacheTime) < 30*time.Second { - out := nodeSnapshot{s.nodeCache, s.nodePM} + out := nodeSnapshot{s.nodeCache, s.nodePM, true} s.cacheMu.Unlock() return out, nil } gen := s.nodeCacheGen + staleNodes, stalePM := s.nodeCache, s.nodePM s.cacheMu.Unlock() - nodes := s.getAllNodes() + nodes, ok := s.getAllNodes() + if !ok { + // A mid-scan failure (lock contention, I/O, interrupted read) + // must not replace a complete cached snapshot with a partial or + // empty one — every consumer (buildNodeInfoMap, Reach's path-hash + // resolver) would then silently lose nodes until the next + // successful refresh. Keep serving whatever was cached, stale or + // not, and let the next call retry (#593, port of upstream #2277). + // On a cold start (nothing cached yet) stalePM is nil: every + // caller of getCachedNodesAndPM relies on a non-nil *prefixMap, + // so fall back to buildPrefixMap's empty-but-valid map, same as + // the pre-fix behaviour for a genuinely empty nodes table — and + // report ok=false, since that map is a placeholder, not the node + // table (round-2 review F2/F4). Nothing is published either way, + // so the next call retries. + if staleNodes == nil && stalePM == nil { + return nodeSnapshot{nil, buildPrefixMap(nil), false}, nil + } + return nodeSnapshot{staleNodes, stalePM, true}, nil + } pm := buildPrefixMap(nodes) // Issue #1290: exclude observers that advertised `repeat:off` from // the path-hop candidate set. Failure is non-fatal — we log via the @@ -8012,10 +8072,10 @@ func (s *PacketStore) getCachedNodesAndPM() ([]nodeInfo, *prefixMap) { s.nodeCacheTime = time.Now() } s.cacheMu.Unlock() - return nodeSnapshot{nodes, pm}, nil + return nodeSnapshot{nodes, pm, true}, nil }) out := v.(nodeSnapshot) - return out.nodes, out.pm + return out.nodes, out.pm, out.ok } // InvalidateNodeCache forces the next getCachedNodesAndPM call to rebuild. @@ -9770,8 +9830,10 @@ type twoByteCellInfo struct { } func (s *PacketStore) computeHashCollisions(region, area string) map[string]interface{} { - // Get all nodes from DB - nodes := s.getAllNodes() + // Get all nodes from DB. A mid-scan failure degrades to an empty list + // here (best-effort analytics, not cached), same as the old deepest + // schema-failure case. + nodes, _ := s.getAllNodes() hashInfo := s.GetNodeHashSizeInfo() // If region is specified, filter to only nodes seen by regional observers @@ -10419,8 +10481,9 @@ func (s *PacketStore) computeMultiByteCapability(adopterHashSizes map[string]int // Get hash size info from adverts (has its own locking) hashInfo := s.GetNodeHashSizeInfo() - // Get all nodes for name/role lookup - allNodes := s.getAllNodes() + // Get all nodes for name/role lookup. A mid-scan failure degrades to an + // empty list here (best-effort analytics, not cached). + allNodes, _ := s.getAllNodes() nodeByPK := make(map[string]nodeInfo, len(allNodes)) for _, n := range allNodes { nodeByPK[n.PublicKey] = n