fix(server): set hasResolvedPath before the startup load reads it (#184 part 2)

cmd/server/db.go

@@ -112,6 +112,10 @@ type DB struct {
 	// channelsRowsHook wraps the result rows of each real query, so a test
 	// can fail the iteration part-way.
 	channelsRowsHook func(kind string, rows channelRows) channelRows
+	// schemaProbeHook (#184), nil in production, runs at the start of every
+	// detectSchema pass. A non-nil error makes the pass give up the way a
+	// failed PRAGMA table_info does: no flag is set.
+	schemaProbeHook func() error
 }
 
 // channelRows is the part of *sql.Rows the channel list scans use.
@@ -231,6 +235,11 @@ func (db *DB) Close() error {
 
 // detectSchema checks if the observations table uses v3 schema (observer_idx).
 func (db *DB) detectSchema() {
+	if db.schemaProbeHook != nil {
+		if err := db.schemaProbeHook(); err != nil {
+			return
+		}
+	}
 	rows, err := db.conn.Query("PRAGMA table_info(observations)")
 	if err != nil {
 		return

cmd/server/main.go

@@ -285,11 +285,6 @@ func main() {
 	// all live in the ingestor; the server only reads the snapshot and
 	// then refreshes it via the recompNeighborGraph slot every 60s.
 	dbPath = database.path
-	// Optimization only, not required for correctness: hasResolvedPath()
-	// self-heals on its own via a PRAGMA re-probe if this weren't here --
-	// AssertReady above already guarantees the column exists, so skip
-	// even that first probe.
-	database.hasResolvedPathFlag.forceTrue()
 
 	// WaitGroup for background init steps that gate /api/healthz readiness.
 	var initWg sync.WaitGroup

cmd/server/schema_wait.go

@@ -137,8 +137,17 @@ func waitForServerSchema(ctx context.Context, db *sql.DB, p schemaWaitPolicy, cl
 // then re-reads the optional-column flags OpenDB detected before the wait,
 // because on a pre-#93 database the route_mask column only appears while
 // the server waits (detectSchema only ever sets flags).
+//
+// #184: AssertReady requires observations.resolved_path, so once it passes
+// the flag is set here, before main.go starts the start-up load. The re-probe
+// alone is not enough: detectSchema gives up silently when its PRAGMA fails,
+// and a false flag makes the load index the hot window without byPathHop.
 func waitForDBSchema(ctx context.Context, db *DB, p schemaWaitPolicy, clk schemaWaitClock, logf func(string, ...interface{})) (int, error) {
-	return waitForDBSchemaWith(ctx, db, func() error { return dbschema.AssertReady(db.conn) }, p, clk, logf)
+	attempts, err := waitForDBSchemaWith(ctx, db, func() error { return dbschema.AssertReady(db.conn) }, p, clk, logf)
+	if err == nil {
+		db.hasResolvedPathFlag.forceTrue()
+	}
+	return attempts, err
 }
 
 func waitForDBSchemaWith(ctx context.Context, db *DB, check func() error, p schemaWaitPolicy, clk schemaWaitClock, logf func(string, ...interface{})) (int, error) {

cmd/server/startup_resolved_path_flag_test.go

@@ -0,0 +1,107 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/meshcore-analyzer/dbschema"
+)
+
+// Issue #184 part 2: the start-up load must select observations.resolved_path
+// once the schema gate has passed, because the gate (dbschema.AssertReady)
+// requires that column. If it does not, the hot window is indexed through the
+// NULL branch (byNode only) and its relays get no byPathHop credit until the
+// next restart.
+//
+// Both cases start the way OpenDB does when the server probes before the
+// ingestor has added the column: the flag is false. The ingestor then adds it,
+// and the server runs its schema gate and the start-up load, in main.go's
+// order. The DB is built without OpenDB's 2 s healer so the window is
+// deterministic; in production the healer does not bound it either, because
+// main.go only forced the flag after the first chunk was ready.
+func TestStartupLoadIndexesResolvedPathAfterSchemaGate_184(t *testing.T) {
+	cases := []struct {
+		name     string
+		probeErr error // returned by every detectSchema pass after the gate
+	}{
+		// The scenario in the issue: waitForDBSchema re-probes after the
+		// gate, so the column the ingestor added in the meantime is seen.
+		{name: "column added after the first probe", probeErr: nil},
+		// detectSchema swallows a failed PRAGMA. The gate itself has proved
+		// the column, so the flag must not depend on that probe.
+		{name: "post-gate probe fails", probeErr: errors.New("simulated SQLITE_BUSY on PRAGMA table_info")},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			path, rw := fixtureSchemaDB(t)
+			for _, q := range []string{`DROP INDEX IF EXISTS idx_observations_resolved_path`, `ALTER TABLE observations DROP COLUMN resolved_path`} {
+				if _, err := rw.Exec(q); err != nil {
+					t.Fatal(err)
+				}
+			}
+
+			conn, err := sql.Open("sqlite", "file:"+path+"?mode=ro&_journal_mode=WAL&_busy_timeout=5000")
+			if err != nil {
+				t.Fatal(err)
+			}
+			db := &DB{conn: conn, path: path}
+			defer db.Close()
+			db.detectSchema() // OpenDB's probe, before the ingestor's migration
+			if db.hasResolvedPath() {
+				t.Fatal("setup: resolved_path detected before the ingestor added it")
+			}
+
+			// The ingestor migrates (adds the column) and writes a resolved
+			// hop for one observation that carries a path.
+			if err := dbschema.Apply(rw, func(string, ...interface{}) {}); err != nil {
+				t.Fatal(err)
+			}
+			var obsID, txID int
+			var pathJSON string
+			if err := rw.QueryRow(`SELECT id, transmission_id, path_json FROM observations
+				WHERE path_json IS NOT NULL AND path_json NOT IN ('', '[]') ORDER BY id LIMIT 1`).Scan(&obsID, &txID, &pathJSON); err != nil {
+				t.Fatal(err)
+			}
+			relay := strings.Repeat("18", 31) + "4a" // matches no fixture node or hop
+			rp := make([]*string, len(parsePathJSON(pathJSON)))
+			rp[0] = &relay
+			if _, err := rw.Exec(`UPDATE observations SET resolved_path = ? WHERE id = ?`, marshalResolvedPath(rp), obsID); err != nil {
+				t.Fatal(err)
+			}
+
+			db.schemaProbeHook = func() error { return tc.probeErr }
+			clk := &fakeClock{now: time.Unix(0, 0)}
+			var log logSink
+			if _, err := waitForDBSchema(context.Background(), db, testSchemaPolicy, clk, log.logf); err != nil {
+				t.Fatal(err)
+			}
+			if !db.hasResolvedPath() {
+				t.Error("hasResolvedPath() is false after the schema gate, which requires observations.resolved_path")
+			}
+
+			// main.go: graph before packets, then RunStartupLoad.
+			s := NewPacketStore(db, nil)
+			if neighborEdgesTableExists(db.conn) {
+				s.graph.Store(loadNeighborEdgesFromDB(db.conn))
+			}
+			if err := s.RunStartupLoad(1000); err != nil {
+				t.Fatal(err)
+			}
+			s.mu.RLock()
+			defer s.mu.RUnlock()
+			found := false
+			for _, tx := range s.byPathHop[relay] {
+				if tx.ID == txID {
+					found = true
+				}
+			}
+			if !found {
+				t.Fatalf("byPathHop[%s…] does not hold tx %d: the start-up load did not read resolved_path (indexed %d tx)", relay[:8], txID, len(s.byPathHop[relay]))
+			}
+		})
+	}
+}