From 6bf3d868bc4ef0486fe5996186011e8c87b9daa7 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 25 Oct 2026 14:54:03 +0100 Subject: [PATCH 1/2] fix(reach): do not cache interrupted query results getAllNodes silently truncated the nodes-table scan on a mid-iteration DB error (lock contention, I/O) instead of checking rows.Err() and each Scan's error, unlike every other scan loop in this file. The resulting partial list was published unconditionally into the 30s node cache that buildNodeInfoMap and computeNodeReach's prefixMap depend on, so a Reach response built during that window could cache an incomplete report for its 5-minute TTL. getAllNodes now reports ok=false on any Scan/iteration failure instead of returning a truncated slice; getCachedNodesAndPM keeps serving the prior complete snapshot on a failed refresh rather than replacing it, falling back to an empty-but-valid prefixMap only on a cold start with nothing cached yet. Relates to #593. --- cmd/server/neighbor_api_test.go | 6 +- .../node_cache_scan_interrupted_test.go | 256 ++++++++++++++++++ cmd/server/store.go | 59 +++- 3 files changed, 306 insertions(+), 15 deletions(-) create mode 100644 cmd/server/node_cache_scan_interrupted_test.go 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..9af35b316 --- /dev/null +++ b/cmd/server/node_cache_scan_interrupted_test.go @@ -0,0 +1,256 @@ +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) } + +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 + } + 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 + } + 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() + t.Cleanup(clearNodesScanFailAfterRows) + 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)) + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index 2b4bab51a..8f0ef5d2b 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -7833,7 +7833,13 @@ func (s *PacketStore) logSchemaDegradationOnce(msg string) { } } -func (s *PacketStore) getAllNodes() []nodeInfo { +// getAllNodes returns every row of the nodes table, or ok=false if the scan +// was interrupted partway (a Scan or iteration error) — 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 checking rows.Err() (and each Scan's own error) after +// the loop tells them apart (#593, port of upstream #2277). +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 @@ -7858,7 +7864,7 @@ 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 } } } @@ -7872,14 +7878,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 { + log.Printf("[store] getAllNodes scan failed: %v", scanErr) + return nil, false } n := nodeInfo{PublicKey: pk, Name: nullStrVal(name), Role: nullStrVal(role)} if hasFirstSeen && firstSeen.Valid { @@ -7902,7 +7913,11 @@ func (s *PacketStore) getAllNodes() []nodeInfo { } nodes = append(nodes, n) } - return nodes + if err := rows.Err(); err != nil { + log.Printf("[store] getAllNodes iteration failed: %v", err) + return nil, false + } + return nodes, true } type prefixMap struct { @@ -8005,8 +8020,25 @@ func (s *PacketStore) getCachedNodesAndPM() ([]nodeInfo, *prefixMap) { 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. + if staleNodes == nil && stalePM == nil { + return nodeSnapshot{nil, buildPrefixMap(nil)}, nil + } + return nodeSnapshot{staleNodes, stalePM}, 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 @@ -9803,8 +9835,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 @@ -10467,8 +10501,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 From eaf4d240a13912c51245d3d07e1819a122f8aafa Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 25 Oct 2026 17:53:01 +0100 Subject: [PATCH 2/2] reviewfix(reach): malformed node row is not an interrupted read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-2 review of #603 (#593). F1 (blocking): a per-row Scan failure in getAllNodes made the whole read report ok=false, permanently. nodes.public_key is a bare TEXT PRIMARY KEY, so SQLite admits one NULL row (and a non-numeric lat/lon fails Scan the same way); one such row disabled the shared node cache for every caller for the life of the process — on a cold start the cache is never published, so every call re-runs the full nodes SELECT and Reach 404s a normal repeater. A row that will not Scan is a malformed row, not an interruption: skip, log and count it the way scanReachRows already discards malformed rows. ok=false is kept for what this issue is about — the deepest schema rung's Query error and rows.Err(). F2: a cold-start node-cache failure was never propagated to Reach. With nothing cached yet and the target also an observer, buildNodeInfoMap still found it through the observer enrichment, so a report with zero reliable tokens and zero links was cached for the full 5-minute TTL and kept being served after the DB recovered. getCachedNodesAndPMOK now reports whether the snapshot it returns is the node table or the empty-but-valid fallback, and computeNodeReach turns the latter into an error (500, nothing cached). The ok-dropping getCachedNodesAndPM wrapper keeps its shape for the ~30 callers that cannot act on a failure. F3: buildNodeInfoMap's follow-up "SELECT id, name FROM observers" ignored its Query error and never checked rows.Err() — the local analogue of the upstream reachNodeScopes root cause, a second query in the same build whose error becomes empty strings. buildNodeInfoMapOK now reports that failure and Reach fails the build instead of caching blanked observer names for 5 minutes; a row that will not Scan stays a skipped malformed row. F4: the cold-start buildPrefixMap(nil) fallback now has its own test. Tests: TestGetAllNodes_MalformedRowIsSkippedNotFatal, TestGetCachedNodesAndPM_ColdStartFailureReportsNotOKWithUsableEmptyPM, TestNodeReach_ColdStartNodeCacheFailureIsA500AndNotCached and TestNodeReach_ObserverEnrichmentInterruptedIsA500AndNotCached (query error and mid-scan subcases). All four are red on the previous head. Relates to #593. --- cmd/server/neighbor_api.go | 63 +++-- .../node_cache_scan_interrupted_test.go | 226 ++++++++++++++++++ cmd/server/node_reach.go | 15 +- cmd/server/node_reach_endpoint_test.go | 8 + cmd/server/store.go | 58 +++-- 5 files changed, 338 insertions(+), 32 deletions(-) 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/node_cache_scan_interrupted_test.go b/cmd/server/node_cache_scan_interrupted_test.go index 9af35b316..a0116cc13 100644 --- a/cmd/server/node_cache_scan_interrupted_test.go +++ b/cmd/server/node_cache_scan_interrupted_test.go @@ -51,11 +51,37 @@ 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 @@ -63,6 +89,9 @@ func (c *nodesScanFailConn) QueryContext(ctx context.Context, query string, args 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 } @@ -108,7 +137,9 @@ 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 { @@ -254,3 +285,198 @@ func TestNodeReach_NodeCacheRefreshFailureDoesNotCacheADegradedReport(t *testing 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 8f0ef5d2b..5f9a3ea2f 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -7833,12 +7833,19 @@ func (s *PacketStore) logSchemaDegradationOnce(msg string) { } } -// getAllNodes returns every row of the nodes table, or ok=false if the scan -// was interrupted partway (a Scan or iteration error) — 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 checking rows.Err() (and each Scan's own error) after -// the loop tells them apart (#593, port of upstream #2277). +// 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 @@ -7871,6 +7878,7 @@ func (s *PacketStore) getAllNodes() ([]nodeInfo, bool) { } 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 @@ -7889,8 +7897,8 @@ func (s *PacketStore) getAllNodes() ([]nodeInfo, bool) { scanErr = rows.Scan(&pk, &name, &role, &lat, &lon) } if scanErr != nil { - log.Printf("[store] getAllNodes scan failed: %v", scanErr) - return nil, false + skipped++ + continue } n := nodeInfo{PublicKey: pk, Name: nullStrVal(name), Role: nullStrVal(role)} if hasFirstSeen && firstSeen.Valid { @@ -7913,6 +7921,9 @@ func (s *PacketStore) getAllNodes() ([]nodeInfo, bool) { } nodes = append(nodes, n) } + 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 @@ -8000,22 +8011,36 @@ 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 } @@ -8033,11 +8058,14 @@ func (s *PacketStore) getCachedNodesAndPM() ([]nodeInfo, *prefixMap) { // 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. + // 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)}, nil + return nodeSnapshot{nil, buildPrefixMap(nil), false}, nil } - return nodeSnapshot{staleNodes, stalePM}, nil + return nodeSnapshot{staleNodes, stalePM, true}, nil } pm := buildPrefixMap(nodes) // Issue #1290: exclude observers that advertised `repeat:off` from @@ -8059,10 +8087,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.