From 7f7126da62ee2e82dc5625eca9d6375ce1fdb9a2 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 25 Oct 2026 09:46:29 +0100 Subject: [PATCH 1/2] fix(ingestor): repair resolved_path written with the transmission's first route type (#289 M2) Relates to #289 (M2 of the plan on that issue; M1 shipped in #308). Until #308 the resolved-path backfill read t.route_type -- the route type of a transmission's first reception -- for every observation under that row, so a reception whose own frame carried the other route family was resolved with the wrong rule: a DIRECT reception anchored on the observer stored a relay it never heard the packet from, and a FLOOD reception treated as DIRECT stayed NULL with the backfill watermark already past it. The repair is an async ingestor migration (resolved_path_route_type_repair_v1). It walks observations.id from 0 in bounded windows, reads only the rows where the #308 helper and t.route_type disagree about flood-ness, re-resolves them outside the write transaction and rewrites the ones the route type changes. Every UPDATE is guarded on the value the batch read, and the watermark commits with the rewrites, so the pass is idempotent and resumable. Also adds scripts/resolved-path-route-repair-estimate.sh, a read-only sizing script for a snapshot copy, with unit tests and ShellCheck in CI. Co-Authored-By: Claude Opus 5 --- .github/workflows/deploy.yml | 6 + cmd/ingestor/main.go | 8 + cmd/ingestor/resolved_path_route_repair.go | 459 ++++++++++++++++++ .../resolved_path_route_repair_289_test.go | 427 ++++++++++++++++ ...solved_path_route_repair_bench_289_test.go | 203 ++++++++ .../resolved-path-route-repair-estimate.sh | 147 ++++++ ...est-resolved-path-route-repair-estimate.sh | 131 +++++ 7 files changed, 1381 insertions(+) create mode 100644 cmd/ingestor/resolved_path_route_repair.go create mode 100644 cmd/ingestor/resolved_path_route_repair_289_test.go create mode 100644 cmd/ingestor/resolved_path_route_repair_bench_289_test.go create mode 100755 scripts/resolved-path-route-repair-estimate.sh create mode 100755 scripts/test-resolved-path-route-repair-estimate.sh diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 376acf7e6..c56b3dc35 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -364,6 +364,12 @@ jobs: - name: ShellCheck decode-backfill measurement scripts (issue #380) run: shellcheck scripts/decode-backfill-estimate.sh scripts/test-decode-backfill-estimate.sh + - name: Resolved-path route-repair measurement script unit tests (issue #289) + run: bash scripts/test-resolved-path-route-repair-estimate.sh + + - name: ShellCheck resolved-path route-repair measurement scripts (issue #289) + run: shellcheck scripts/resolved-path-route-repair-estimate.sh scripts/test-resolved-path-route-repair-estimate.sh + - name: QA blacklist script unit tests (issues #1977, #83) run: bash qa/scripts/test-blacklist-sql.sh diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index af2a86d4f..a24c86281 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -532,6 +532,14 @@ func main() { defer stopResolvedPathBackfill() } + // #289 M2: repair resolved_path rows the pre-#308 backfill wrote with the + // transmission's first route type, once per database. Same placement as + // the backfill above (it needs the same primed index and graph) and the + // same shutdown: cancelled here, resumed from its watermark next start. + routeRepairCtx, stopResolvedPathRouteRepair := context.WithCancel(context.Background()) + defer stopResolvedPathRouteRepair() + store.StartResolvedPathRouteRepair(routeRepairCtx) + // #380: re-decode historical enc_ GRP_TXT rows and unknown-scoped // transport rows whenever the effective key set grows (config/SIGHUP, // approved shared channels, approved region scopes). Same placement as diff --git a/cmd/ingestor/resolved_path_route_repair.go b/cmd/ingestor/resolved_path_route_repair.go new file mode 100644 index 000000000..cec06a470 --- /dev/null +++ b/cmd/ingestor/resolved_path_route_repair.go @@ -0,0 +1,459 @@ +package main + +import ( + "context" + "database/sql" + "errors" + "fmt" + "log" + "strings" + "time" + + "github.com/meshcore-analyzer/packetpath" +) + +// Issue #289 M2 — repair observations.resolved_path rows written with the +// transmission's first route type. +// +// Why those rows exist. DIRECT and FLOOD receptions of the same payload share +// one transmissions row, because the content hash does not cover the header +// route bits (MeshCore src/Packet.cpp calculatePacketHash). Until #308 the +// resolved-path backfill read t.route_type — the route type of the *first* +// reception — for every observation under that row, so a reception whose own +// frame carries the other route family was resolved with the wrong rule: +// +// - a DIRECT reception under a FLOOD transmission was anchored on the +// observer, which stored a relay the observer never heard the packet from; +// - a FLOOD reception under a DIRECT transmission got no observer anchor, so +// its row stayed NULL. The #188 backfill moves its watermark past every row +// it scans, whether or not the row resolved, so the corrected backfill +// never revisits it. Those NULL rows are candidates here too, which is why +// this pass scans by id from 0 rather than from the backfill watermark. +// +// #308 fixed the readers. Rows they had already written stay wrong until this +// pass rewrites them, which is what M2 of the plan on #289 asks for. +// +// What a candidate is. A row is a candidate when the #308 helper +// (observationRouteType: the observation's own frame header first, then the +// route_mask guard, then t.route_type) disagrees with t.route_type about +// flood-ness — flood-ness is the only thing resolveObservationPath reads. That +// covers both bug directions above and the rows whose frame is gone but whose +// route_mask holds both families (the #308 guard now reports those as unknown, +// where the pre-#308 backfill used t.route_type). Rows the two agree on are +// never read into a batch. +// +// What it writes. For each candidate the pass resolves the hops twice with the +// current graph and index: once with the observation's own route type (`want`) +// and once with t.route_type (`had`, what the pre-#308 reader produced). It +// writes only when +// +// 1. the two differ — so a row whose answer does not depend on the route type +// is left alone even if its stored value is stale for some other reason +// (that is not #289, and rewriting it would turn this pass into an +// unbounded re-resolve of history); and +// 2. the stored value is not already `want` — so a row live ingest resolved +// correctly, and a row an earlier run of this pass already repaired, is +// not written again. +// +// `want` may be empty, and then the row is set back to NULL: that is the +// repair for a DIRECT reception stored with a wrong relay. Every UPDATE is +// guarded on the value the batch read, so a row live ingest rewrote in the +// meantime keeps ingest's value and a repeated batch is a no-op. +// +// Bounds and resume (the #89 / #380 shape): +// - the pass needs the primed prefix index and a post-build neighbour graph +// with at least one edge (resolvedPathBackfillReady) — resolving against a +// cold graph would overwrite correct rows with NULL — and waits for the +// next post-build graph when that is not yet true; +// - each batch scans a bounded observation-id window and reads at most +// batchSize candidates, so a window with no candidates cannot become one +// unbounded range scan on the single SQLite connection; +// - resolving happens in Go outside any transaction (#267); one WriterTx per +// batch holds writerMu, so a live insert queues behind a batch and not +// behind the run; +// - the row rewrites and the new watermark commit in one transaction, so a +// crash or a SIGTERM repeats at most the batch that did not commit. The +// ceiling is pinned on the first batch and persisted: rows above it are +// written by live ingest, which has always used the frame's own route type. +// +// It runs once per database, recorded in _async_migrations. Server visibility +// is restart-only: the server indexes an observation's resolved_path when it +// first polls or loads the row, and a later rewrite reaches its indexes on the +// next Load. cmd/server is not touched. +const ( + resolvedPathRouteRepairComponent = "resolved_path_route_repair" + resolvedPathRouteRepairMigration = "resolved_path_route_type_repair_v1" +) + +// Bounds. Package vars so tests can drive the multi-batch loop with a few +// rows; customizer candidates later (AGENTS.md rule 8), like the #188 +// backfill's batchSize/pauseMs. +var ( + resolvedPathRouteRepairBatchSize = 500 + resolvedPathRouteRepairIDWindow = int64(20000) + resolvedPathRouteRepairPause = 250 * time.Millisecond + + // resolvedPathRouteRepairResolve is the resolver a batch calls. A + // variable so the #267 test can check that no row is resolved while the + // write transaction holds writerMu. + resolvedPathRouteRepairResolve = resolveObservationPath + + // resolvedPathRouteRepairPreWriteHook is a test seam, nil in production: + // called after a batch has resolved its rows and before it writes them. + resolvedPathRouteRepairPreWriteHook func() +) + +// resolvedPathRouteRepairProgressEvery logs progress every N batches. +const resolvedPathRouteRepairProgressEvery = 50 + +// StartResolvedPathRouteRepair schedules the repair as an async migration. +// Call it after the neighbour-edges builder is running, next to the #188 +// backfill: the pass needs that builder's graph and must not delay boot. +// Cancelling ctx stops it at the next statement; it is then recorded as +// unfinished and resumes from its watermark on the next start. +func (s *Store) StartResolvedPathRouteRepair(ctx context.Context) { + // PREFLIGHT: async=true reason="#289 M2 one-off repair: windowed scan of observations joined to transmissions with batched rewrites; must not block ingestor boot" + if err := s.RunAsyncMigration(ctx, resolvedPathRouteRepairMigration, s.repairResolvedPathRouteTypes); err != nil { + log.Printf("[%s] scheduling %s failed: %v", resolvedPathRouteRepairComponent, resolvedPathRouteRepairMigration, err) + } +} + +func ensureResolvedPathRouteRepairState(db *sql.DB) error { + // PREFLIGHT: async=true reason="CREATE TABLE IF NOT EXISTS for a one-row state table; constant cost at any scale" + _, err := db.Exec(`CREATE TABLE IF NOT EXISTS resolved_path_route_repair_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + watermark INTEGER NOT NULL, + ceiling INTEGER NOT NULL, + scanned INTEGER NOT NULL DEFAULT 0, + candidates INTEGER NOT NULL DEFAULT 0, + repaired INTEGER NOT NULL DEFAULT 0, + cleared INTEGER NOT NULL DEFAULT 0, + started_at TEXT NOT NULL, + updated_at TEXT NOT NULL + )`) + return err +} + +// ResolvedPathRouteRepairState is the persisted progress of the pass. +type ResolvedPathRouteRepairState struct { + Watermark int64 + Ceiling int64 + Scanned int64 // candidate rows the SQL prefilter returned + Candidates int64 // of those, the rows #308 and t.route_type really disagree on + Repaired int64 // rows rewritten to a resolved path + Cleared int64 // rows set back to NULL (a wrong relay removed) + StartedAt string + UpdatedAt string +} + +// ResolvedPathRouteRepairState reads the persisted state, or nil when the +// pass has never run. +func (s *Store) ResolvedPathRouteRepairState() (*ResolvedPathRouteRepairState, error) { + if err := ensureResolvedPathRouteRepairState(s.db); err != nil { + return nil, err + } + var st ResolvedPathRouteRepairState + err := s.db.QueryRow(`SELECT watermark, ceiling, scanned, candidates, repaired, cleared, started_at, updated_at + FROM resolved_path_route_repair_state WHERE id = 1`).Scan(&st.Watermark, &st.Ceiling, + &st.Scanned, &st.Candidates, &st.Repaired, &st.Cleared, &st.StartedAt, &st.UpdatedAt) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, err + } + return &st, nil +} + +// repairResolvedPathRouteTypes is the resolved_path_route_type_repair_v1 +// migration body: it runs batches from the persisted watermark to the pinned +// ceiling. +func (s *Store) repairResolvedPathRouteTypes(ctx context.Context, _ *sql.DB) error { + if err := ensureResolvedPathRouteRepairState(s.db); err != nil { + return fmt.Errorf("ensure state table: %w", err) + } + st, err := s.resolvedPathRouteRepairStart() + if err != nil { + return err + } + if st.Watermark >= st.Ceiling { + log.Printf("[%s] nothing to do (watermark %d, ceiling %d)", resolvedPathRouteRepairComponent, st.Watermark, st.Ceiling) + return nil + } + start := time.Now() + log.Printf("[%s] starting: ids (%d, %d], batch %d, window %d, pause %s", + resolvedPathRouteRepairComponent, st.Watermark, st.Ceiling, resolvedPathRouteRepairBatchSize, + resolvedPathRouteRepairIDWindow, resolvedPathRouteRepairPause) + var maxHold time.Duration + for batches := 0; st.Watermark < st.Ceiling; { + if err := ctx.Err(); err != nil { + return err + } + // Resolving against a cold index or graph would rewrite correct rows + // as NULL, so a pass that is not ready waits instead of writing. + if err := s.waitResolvedPathRouteRepairReady(ctx); err != nil { + return err + } + b, err := s.resolvedPathRouteRepairBatch(ctx, st) + if errors.Is(err, errResolvedPathBackfillNotReady) { + // The snapshot stopped being usable between the gate above and + // the batch. Nothing was written: wait for the next post-build + // graph and retry the same watermark. + continue + } + if err != nil { + return err + } + batches++ + if b.Hold > maxHold { + maxHold = b.Hold + } + if batches%resolvedPathRouteRepairProgressEvery == 0 { + log.Printf("[%s] progress: watermark %d of %d, %d candidates, %d repaired, %d cleared", + resolvedPathRouteRepairComponent, st.Watermark, st.Ceiling, st.Candidates, st.Repaired, st.Cleared) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(resolvedPathRouteRepairPause): + } + } + log.Printf("[%s] done: %d prefilter rows read, %d candidates, %d repaired, %d cleared, max write hold %s, in %s", + resolvedPathRouteRepairComponent, st.Scanned, st.Candidates, st.Repaired, st.Cleared, + maxHold.Round(time.Microsecond), time.Since(start).Round(time.Millisecond)) + return nil +} + +// resolvedPathRouteRepairStart reads the persisted state, or pins a new +// ceiling at the current MAX(observations.id) and records it. Rows above the +// ceiling were written by live ingest, which has always used the frame's own +// route type. +func (s *Store) resolvedPathRouteRepairStart() (*ResolvedPathRouteRepairState, error) { + st, err := s.ResolvedPathRouteRepairState() + if err != nil { + return nil, fmt.Errorf("read state: %w", err) + } + if st != nil { + return st, nil + } + var ceiling int64 + if err := s.db.QueryRow(`SELECT COALESCE(MAX(id), 0) FROM observations`).Scan(&ceiling); err != nil { + return nil, fmt.Errorf("read ceiling: %w", err) + } + now := time.Now().UTC().Format(time.RFC3339) + st = &ResolvedPathRouteRepairState{Ceiling: ceiling, StartedAt: now, UpdatedAt: now} + if _, err := s.WriterExec(resolvedPathRouteRepairComponent, + `INSERT OR IGNORE INTO resolved_path_route_repair_state (id, watermark, ceiling, started_at, updated_at) + VALUES (1, 0, ?, ?, ?)`, ceiling, now, now); err != nil { + return nil, fmt.Errorf("record state: %w", err) + } + // A concurrent start may have won the INSERT; re-read so both agree. + st2, err := s.ResolvedPathRouteRepairState() + if err != nil || st2 == nil { + return st, err + } + return st2, nil +} + +// waitResolvedPathRouteRepairReady blocks until the prefix index and a +// post-build neighbour graph with at least one edge are available, the same +// readiness gate the #188 backfill uses. +func (s *Store) waitResolvedPathRouteRepairReady(ctx context.Context) error { + for logged := false; ; { + // Take the wake-up channel before the check, so a graph published in + // between is not missed. + _, next := s.neighborGraph.buildState() + if s.resolvedPathBackfillReady(s.prefixIdx.load(), s.neighborGraph.load()) { + return nil + } + if !logged { + log.Printf("[%s] waiting: %v", resolvedPathRouteRepairComponent, errResolvedPathBackfillNotReady) + logged = true + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-next: + } + } +} + +// resolvedPathRouteRepairBatchResult reports one batch. +type resolvedPathRouteRepairBatchResult struct { + Scanned int // candidate rows read (the SQL prefilter's rows) + Candidates int // rows the #308 helper really disagrees on + Repaired int // rows rewritten to a resolved path + Cleared int // rows set back to NULL + Hold time.Duration // duration of the write transaction +} + +type resolvedPathRouteRepairRow struct { + id int64 + pathJSON string + stored string // the stored resolved_path, "" for NULL + header string + txRoute int + mask sql.NullInt64 + payloadType int + fromPubkey string + observerID string +} + +// resolvedPathRouteRepairCandidateSQL is the prefilter: a superset of the +// candidates, cheap enough to run over a bounded id window. The precise test +// is observationRouteType in Go (the SQL cannot call the #308 helper), so this +// only has to avoid missing a candidate. +// +// - first branch: the row has a frame, and the route family in its header +// (bit 1 of the low nibble of the header byte, i.e. of raw_hex's second +// character) differs from t.route_type's family. A character that is not +// hex makes instr() return 0 and the expression -1, whose bit 1 is set, so +// an unparseable header is never dropped here when t.route_type is a flood +// type — Go then falls through to the route_mask rule, as #308 does. +// - second branch: the row has no usable frame, its route_mask holds both +// families (so #308 reports the route type as unknown: no observer anchor) +// and t.route_type is a flood type, which is what the pre-#308 backfill +// used instead. +// +// Rows with no hops are skipped: resolveObservationPath returns nothing for +// them under either route type. +var resolvedPathRouteRepairCandidateSQL = fmt.Sprintf(`( + (length(o.raw_hex) >= 2 + AND ((instr('0123456789abcdef', lower(substr(o.raw_hex, 2, 1))) - 1) & 2) != (COALESCE(t.route_type, -1) & 2)) + OR (COALESCE(length(o.raw_hex), 0) < 2 + AND t.route_mask IS NOT NULL + AND (t.route_mask & %d) != 0 + AND (t.route_mask & %d) != 0 + AND COALESCE(t.route_type, -1) IN (%d, %d)) + )`, packetpath.RouteMaskFlood, packetpath.RouteMaskDirect, packetpath.RouteTransportFlood, packetpath.RouteFlood) + +// resolvedPathRouteRepairBatch reads one bounded window of candidates, +// re-resolves them outside any lock and rewrites the rows that the route type +// changes. It advances st in place after the batch commits. +func (s *Store) resolvedPathRouteRepairBatch(ctx context.Context, st *ResolvedPathRouteRepairState) (resolvedPathRouteRepairBatchResult, error) { + var b resolvedPathRouteRepairBatchResult + graph, idx := s.neighborGraph.load(), s.prefixIdx.load() + if !s.resolvedPathBackfillReady(idx, graph) { + return b, errResolvedPathBackfillNotReady + } + windowEnd := st.Watermark + resolvedPathRouteRepairIDWindow + if windowEnd > st.Ceiling { + windowEnd = st.Ceiling + } + rows, err := s.db.QueryContext(ctx, `SELECT o.id, COALESCE(o.path_json, ''), COALESCE(o.resolved_path, ''), + COALESCE(substr(o.raw_hex, 1, 2), ''), COALESCE(t.route_type, -1), t.route_mask, + COALESCE(t.payload_type, -1), COALESCE(t.from_pubkey, ''), COALESCE(obs.id, '') + FROM observations o + JOIN transmissions t ON t.id = o.transmission_id + LEFT JOIN observers obs ON obs.rowid = o.observer_idx + WHERE o.id > ? AND o.id <= ? + AND COALESCE(o.path_json, '') NOT IN ('', '[]') + AND `+resolvedPathRouteRepairCandidateSQL+` + ORDER BY o.id + LIMIT ?`, st.Watermark, windowEnd, resolvedPathRouteRepairBatchSize) + if err != nil { + return b, fmt.Errorf("select batch: %w", err) + } + batch := make([]resolvedPathRouteRepairRow, 0, resolvedPathRouteRepairBatchSize) + for rows.Next() { + var r resolvedPathRouteRepairRow + if err := rows.Scan(&r.id, &r.pathJSON, &r.stored, &r.header, &r.txRoute, &r.mask, + &r.payloadType, &r.fromPubkey, &r.observerID); err != nil { + rows.Close() + return b, fmt.Errorf("scan batch: %w", err) + } + batch = append(batch, r) + } + rows.Close() + if err := rows.Err(); err != nil { + return b, fmt.Errorf("read batch: %w", err) + } + b.Scanned = len(batch) + + // The LIMIT truncated the window: continue from the last row read. Only a + // window read to its end moves the watermark to the window end. + next := windowEnd + if len(batch) == resolvedPathRouteRepairBatchSize { + next = batch[len(batch)-1].id + } + + // Resolve outside any transaction (#267), with the same inputs + // InsertTransmission uses (fromPubkey only for ADVERTs). + type update struct { + id int64 + want string // "" means set the row back to NULL + had string // the value the batch read, the UPDATE's guard + } + var updates []update + for _, r := range batch { + if err := ctx.Err(); err != nil { + return b, err + } + rt := observationRouteType(r.header, r.txRoute, r.mask) + if isFloodRoute(rt) == isFloodRoute(r.txRoute) { + continue // the prefilter's superset: #308 and t.route_type agree + } + hops := parsePathArray(r.pathJSON) + if len(hops) == 0 { + continue + } + b.Candidates++ + from := "" + if r.payloadType == int(payloadADVERT) { + from = strings.ToLower(r.fromPubkey) + } + want := marshalResolvedPath(resolvedPathRouteRepairResolve(hops, from, r.observerID, rt, graph, idx)) + // What the pre-#308 reader produced for this row with today's graph. + // When the route type does not change the answer, the stored value is + // not a #289 defect and this pass leaves it alone. + had := marshalResolvedPath(resolvedPathRouteRepairResolve(hops, from, r.observerID, r.txRoute, graph, idx)) + if want == had || r.stored == want { + continue + } + updates = append(updates, update{r.id, want, r.stored}) + } + + if resolvedPathRouteRepairPreWriteHook != nil { + resolvedPathRouteRepairPreWriteHook() + } + scanned, candidates := st.Scanned+int64(b.Scanned), st.Candidates+int64(b.Candidates) + holdStart := time.Now() + err = s.WriterTx(resolvedPathRouteRepairComponent, func(tx *sql.Tx) error { + b.Repaired, b.Cleared = 0, 0 + if len(updates) > 0 { + // The guard keeps the value live ingest wrote after this batch + // read the row, and makes a repeated batch a no-op. + stmt, err := tx.Prepare(`UPDATE observations SET resolved_path = ? + WHERE id = ? AND COALESCE(resolved_path, '') = ?`) + if err != nil { + return err + } + defer stmt.Close() + for _, u := range updates { + res, err := stmt.Exec(nilIfEmpty(u.want), u.id, u.had) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n > 0 { + if u.want == "" { + b.Cleared++ + } else { + b.Repaired++ + } + } + } + } + _, err := tx.Exec(`UPDATE resolved_path_route_repair_state + SET watermark = ?, scanned = ?, candidates = ?, repaired = repaired + ?, cleared = cleared + ?, updated_at = ? + WHERE id = 1`, next, scanned, candidates, b.Repaired, b.Cleared, time.Now().UTC().Format(time.RFC3339)) + return err + }) + b.Hold = time.Since(holdStart) + if err != nil { + return b, fmt.Errorf("write batch: %w", err) + } + st.Watermark, st.Scanned, st.Candidates = next, scanned, candidates + st.Repaired += int64(b.Repaired) + st.Cleared += int64(b.Cleared) + return b, nil +} diff --git a/cmd/ingestor/resolved_path_route_repair_289_test.go b/cmd/ingestor/resolved_path_route_repair_289_test.go new file mode 100644 index 000000000..219c72bb0 --- /dev/null +++ b/cmd/ingestor/resolved_path_route_repair_289_test.go @@ -0,0 +1,427 @@ +package main + +import ( + "context" + "database/sql" + "errors" + "path/filepath" + "testing" + "time" +) + +// Issue #289 M2: repair observations.resolved_path rows the pre-#308 +// resolved-path backfill wrote with the transmission's first route type +// (t.route_type) instead of each observation's own. +// +// The fixture is the #188 one (backfillFixture188): two relays sharing the +// 1-byte prefix "c3", one observer edge obs188 <-> c3a. So a flood ["c3"] +// resolves to c3a through the observer anchor and a DIRECT ["c3"] resolves to +// nothing. ["c355"] names c3a uniquely and resolves under either route type. + +// rp289 is the stored form of a resolved path over the given pubkeys. +func rp289(pks ...string) string { + out := make([]*string, len(pks)) + for i := range pks { + pk := pks[i] + if pk != "" { + out[i] = &pk + } + } + return marshalResolvedPath(out) +} + +// setResolvedPath289 writes a row's resolved_path directly ("" = NULL): the +// value the pre-#308 backfill, or live ingest, had left there. +func setResolvedPath289(t *testing.T, store *Store, id int64, rp string) { + t.Helper() + if _, err := store.db.Exec(`UPDATE observations SET resolved_path = ? WHERE id = ?`, nilIfEmpty(rp), id); err != nil { + t.Fatal(err) + } +} + +// smallRepairBatches289 makes the pass run many small batches so the +// multi-batch loop, the watermark and a resume are exercised on a few rows. +func smallRepairBatches289(t *testing.T, window int64, batch int) { + t.Helper() + oldWindow, oldBatch, oldPause := resolvedPathRouteRepairIDWindow, resolvedPathRouteRepairBatchSize, resolvedPathRouteRepairPause + resolvedPathRouteRepairIDWindow, resolvedPathRouteRepairBatchSize, resolvedPathRouteRepairPause = window, batch, 0 + t.Cleanup(func() { + resolvedPathRouteRepairIDWindow, resolvedPathRouteRepairBatchSize, resolvedPathRouteRepairPause = oldWindow, oldBatch, oldPause + }) +} + +func repairState289(t *testing.T, store *Store) *ResolvedPathRouteRepairState { + t.Helper() + st, err := store.ResolvedPathRouteRepairState() + if err != nil { + t.Fatal(err) + } + if st == nil { + t.Fatal("no resolved_path_route_repair_state row") + } + return st +} + +// resetRepairWatermark289 rewinds the pass so the next run rescans everything +// with the same ceiling: used to show that the second scan writes nothing. +func resetRepairWatermark289(t *testing.T, store *Store) { + t.Helper() + if _, err := store.db.Exec(`UPDATE resolved_path_route_repair_state SET watermark = 0 WHERE id = 1`); err != nil { + t.Fatal(err) + } +} + +type repairCase289 struct { + name string + txRoute int + mask sql.NullInt64 + rawHex string + path string + stored string // resolved_path before the pass ("" = NULL) + want string // resolved_path after the pass ("" = NULL) + id int64 +} + +// Both bug directions, the route_mask fallback, the rows that must not change, +// and the mismatch whose answer does not depend on the route type. +func repairCases289() []repairCase289 { + return []repairCase289{ + // #289 direction 1: a DIRECT reception anchored on the observer stored + // a relay it never heard the packet from. The repair removes it. + {name: "DIRECT frame under a FLOOD transmission", txRoute: routeFlood188, mask: mask289(mixedMask289), + rawHex: directFrame289, path: `["c3"]`, stored: rp289(c3a), want: ""}, + // #289 direction 2: a FLOOD reception treated as DIRECT got no observer + // anchor and stayed NULL. The repair resolves it. + {name: "FLOOD frame under a DIRECT transmission", txRoute: routeDirect188, mask: mask289(mixedMask289), + rawHex: floodFrame289, path: `["c3"]`, stored: "", want: rp289(c3a)}, + // No frame and a mask with both families: #308 reports the route type + // as unknown, the pre-#308 backfill used t.route_type (FLOOD). + {name: "no frame, mixed mask, FLOOD transmission", txRoute: routeFlood188, mask: mask289(mixedMask289), + rawHex: "", path: `["c3"]`, stored: rp289(c3a), want: ""}, + + // Not candidates: the frame header and t.route_type agree. + {name: "FLOOD frame under a FLOOD transmission", txRoute: routeFlood188, mask: mask289(floodMask289), + rawHex: floodFrame289, path: `["c3"]`, stored: rp289(c3a), want: rp289(c3a)}, + {name: "DIRECT frame under a DIRECT transmission", txRoute: routeDirect188, mask: mask289(0b0100), + rawHex: directFrame289, path: `["c3"]`, stored: "", want: ""}, + // No frame: #308 and t.route_type agree on flood-ness. + {name: "no frame, mixed mask, DIRECT transmission", txRoute: routeDirect188, mask: mask289(mixedMask289), + rawHex: "", path: `["c3"]`, stored: "", want: ""}, + {name: "no frame, flood-only mask", txRoute: routeFlood188, mask: mask289(floodMask289), + rawHex: "", path: `["c3"]`, stored: rp289(c3a), want: rp289(c3a)}, + {name: "no frame, NULL mask", txRoute: routeFlood188, mask: sql.NullInt64{}, + rawHex: "", path: `["c3"]`, stored: rp289(c3a), want: rp289(c3a)}, + + // A candidate whose answer does not depend on the route type: ["c355"] + // names c3a uniquely either way. Its stored value is stale for some + // other reason, which is not #289, so the pass leaves it alone. + {name: "DIRECT frame under a FLOOD transmission, unique 2-byte hop", txRoute: routeFlood188, + mask: mask289(mixedMask289), rawHex: directFrame289, path: `["c355"]`, stored: rp289(c3b), want: rp289(c3b)}, + } +} + +// seedRepairCases289 seeds every case into one store, so the candidate +// selection runs over a mixed table. +func seedRepairCases289(t *testing.T, store *Store, cases []repairCase289) { + t.Helper() + for i := range cases { + c := &cases[i] + c.id = seedObservation289(t, store, c.txRoute, c.mask, c.rawHex, c.path, ts190+int64(i)) + setResolvedPath289(t, store, c.id, c.stored) + } +} + +func assertRepairCases289(t *testing.T, store *Store, cases []repairCase289) { + t.Helper() + for _, c := range cases { + got := resolvedPathOf188(t, store, c.id) + if c.want == "" { + if got.Valid { + t.Errorf("%s: resolved_path = %s, want NULL", c.name, got.String) + } + continue + } + if !got.Valid { + t.Errorf("%s: resolved_path is NULL, want %s", c.name, c.want) + continue + } + if got.String != c.want { + t.Errorf("%s: resolved_path = %s, want %s", c.name, got.String, c.want) + } + } +} + +func runRepair289(t *testing.T, store *Store) { + t.Helper() + if err := store.repairResolvedPathRouteTypes(context.Background(), store.db); err != nil { + t.Fatal(err) + } +} + +// Both bug directions are repaired, and a row whose route type the frame and +// the transmission agree on is not touched. +func TestResolvedPathRouteRepair_RepairsBothDirections_289(t *testing.T) { + smallRepairBatches289(t, 3, 2) + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + cases := repairCases289() + seedRepairCases289(t, store, cases) + primeIndexAndGraph188(t, store) + + runRepair289(t, store) + assertRepairCases289(t, store, cases) + + st := repairState289(t, store) + if st.Watermark < st.Ceiling { + t.Fatalf("watermark %d did not reach the ceiling %d", st.Watermark, st.Ceiling) + } + if st.Repaired != 1 || st.Cleared != 2 { + t.Fatalf("state = %d repaired, %d cleared; want 1 repaired (the NULL flood row) and 2 cleared (the two wrong relays)", st.Repaired, st.Cleared) + } + if st.Candidates != 4 { + t.Fatalf("candidates = %d, want 4 (three defects plus the route-type-independent row)", st.Candidates) + } +} + +// A FLOOD reception treated as DIRECT was left NULL by the pre-#308 backfill, +// which then moved its watermark past the row. The corrected backfill never +// revisits it; this pass scans from id 0 and does. +func TestResolvedPathRouteRepair_RepairsNullRowBelowBackfillWatermark_289(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + id := seedObservation289(t, store, routeDirect188, mask289(mixedMask289), floodFrame289, `["c3"]`, ts190) + primeIndexAndGraph188(t, store) + + // The pre-#308 backfill scanned the row, resolved nothing (it used the + // transmission's DIRECT route type) and persisted a watermark past it. + if err := ensureResolvedPathBackfillState(store.db); err != nil { + t.Fatal(err) + } + if _, err := store.db.Exec(`INSERT INTO resolved_path_backfill_state (id, watermark, updated_at) VALUES (1, ?, '2026-06-01T00:00:00Z') + ON CONFLICT(id) DO UPDATE SET watermark = excluded.watermark`, id); err != nil { + t.Fatal(err) + } + res, err := store.RunResolvedPathBackfill(context.Background(), 100, 0) + if err != nil { + t.Fatal(err) + } + if res.Scanned != 0 { + t.Fatalf("the backfill scanned %d rows below its watermark, want 0", res.Scanned) + } + if rp := resolvedPathOf188(t, store, id); rp.Valid { + t.Fatalf("setup: resolved_path = %s, want the NULL row the pre-#308 backfill left", rp.String) + } + + runRepair289(t, store) + + rp := resolvedPathOf188(t, store, id) + if !rp.Valid || rp.String != rp289(c3a) { + t.Fatalf("resolved_path = %v, want %s", rp, rp289(c3a)) + } +} + +// A second pass writes nothing: the watermark has reached the ceiling, and a +// rescan from 0 finds every candidate already carrying the value it wants. +func TestResolvedPathRouteRepair_Idempotent_289(t *testing.T) { + smallRepairBatches289(t, 3, 2) + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + cases := repairCases289() + seedRepairCases289(t, store, cases) + primeIndexAndGraph188(t, store) + + runRepair289(t, store) + first := repairState289(t, store) + + runRepair289(t, store) // watermark already at the ceiling + if st := repairState289(t, store); st.Repaired != first.Repaired || st.Cleared != first.Cleared || st.Scanned != first.Scanned { + t.Fatalf("a completed pass ran again: %+v, want %+v", st, first) + } + + resetRepairWatermark289(t, store) // rescan the same ids + runRepair289(t, store) + st := repairState289(t, store) + if st.Repaired != first.Repaired || st.Cleared != first.Cleared { + t.Fatalf("the rescan wrote %d repaired / %d cleared more than the first pass (%d/%d)", + st.Repaired-first.Repaired, st.Cleared-first.Cleared, first.Repaired, first.Cleared) + } + if st.Candidates <= first.Candidates { + t.Fatalf("the rescan found %d candidates in all, want more than the first pass's %d (it must have read the rows again)", st.Candidates, first.Candidates) + } + assertRepairCases289(t, store, cases) +} + +// A cancelled pass keeps the watermark of its last committed batch and the +// next run finishes the rest. +func TestResolvedPathRouteRepair_ResumesAfterStop_289(t *testing.T) { + smallRepairBatches289(t, 1, 1) // one id per batch + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + cases := repairCases289() + seedRepairCases289(t, store, cases) + primeIndexAndGraph188(t, store) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + batches := 0 + oldHook := resolvedPathRouteRepairPreWriteHook + resolvedPathRouteRepairPreWriteHook = func() { + batches++ + if batches == 2 { + cancel() + } + } + t.Cleanup(func() { resolvedPathRouteRepairPreWriteHook = oldHook }) + + err := store.repairResolvedPathRouteTypes(ctx, store.db) + if !errors.Is(err, context.Canceled) { + t.Fatalf("stopped pass returned %v, want context.Canceled", err) + } + stopped := repairState289(t, store) + if stopped.Watermark == 0 || stopped.Watermark >= stopped.Ceiling { + t.Fatalf("watermark after the stop = %d, want a committed batch below the ceiling %d", stopped.Watermark, stopped.Ceiling) + } + // The first case (id of cases[0]) is in the first window and is repaired; + // the last case is not reached yet. + if rp := resolvedPathOf188(t, store, cases[0].id); rp.Valid { + t.Fatalf("the first batch did not commit its repair: resolved_path = %s, want NULL", rp.String) + } + last := cases[len(cases)-2] // the NULL-mask row that must stay as seeded + if rp := resolvedPathOf188(t, store, last.id); !rp.Valid { + t.Fatalf("a row above the watermark was already written: %s", last.name) + } + + resolvedPathRouteRepairPreWriteHook = oldHook + runRepair289(t, store) + assertRepairCases289(t, store, cases) + if st := repairState289(t, store); st.Watermark < st.Ceiling { + t.Fatalf("watermark %d did not reach the ceiling %d after the resume", st.Watermark, st.Ceiling) + } + if st := repairState289(t, store); st.Scanned <= stopped.Scanned { + t.Fatalf("the resume scanned nothing new (%d, stopped at %d)", st.Scanned, stopped.Scanned) + } +} + +// The pass resolves outside its write transaction (#267) and opens one write +// transaction per batch, so a live insert queues behind a batch and not behind +// the run. +func TestResolvedPathRouteRepair_ResolvesOutsideWriteTransaction_289(t *testing.T) { + smallRepairBatches289(t, 2, 1) + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + cases := repairCases289() + seedRepairCases289(t, store, cases) + primeIndexAndGraph188(t, store) + + var resolved, underLock int + old := resolvedPathRouteRepairResolve + resolvedPathRouteRepairResolve = func(hops []string, fromPubkey, observerID string, routeType int, graph *NeighborGraph, idx prefixIndex) []*string { + resolved++ + if writerMuHeld() { + underLock++ + } + return old(hops, fromPubkey, observerID, routeType, graph, idx) + } + t.Cleanup(func() { resolvedPathRouteRepairResolve = old }) + + batches := 0 + oldHook := resolvedPathRouteRepairPreWriteHook + resolvedPathRouteRepairPreWriteHook = func() { batches++ } + t.Cleanup(func() { resolvedPathRouteRepairPreWriteHook = oldHook }) + + txBefore := writerStatsAgg.get(resolvedPathRouteRepairComponent).snapshot().Count + runRepair289(t, store) + + if resolved == 0 { + t.Fatal("the pass resolved nothing") + } + if underLock != 0 { + t.Fatalf("%d of %d resolutions ran while the write transaction held writerMu", underLock, resolved) + } + if batches < 2 { + t.Fatalf("%d batches, want several so the per-batch bound is exercised", batches) + } + // One write transaction per batch, plus the one that recorded the state row. + if n := writerStatsAgg.get(resolvedPathRouteRepairComponent).snapshot().Count - txBefore; n != int64(batches)+1 { + t.Fatalf("%d write transactions for %d batches, want one per batch plus the state row", n, batches) + } +} + +// Live ingest resolves with the frame's own route type and is authoritative. +// A row it rewrote between the batch's read and its write keeps ingest's +// value: every UPDATE is guarded on the value the batch read. +func TestResolvedPathRouteRepair_DoesNotOverwriteAConcurrentWrite_289(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + id := seedObservation289(t, store, routeFlood188, mask289(mixedMask289), directFrame289, `["c3"]`, ts190) + setResolvedPath289(t, store, id, rp289(c3a)) // the pre-#308 backfill's wrong relay + primeIndexAndGraph188(t, store) + + old := resolvedPathRouteRepairPreWriteHook + resolvedPathRouteRepairPreWriteHook = func() { setResolvedPath289(t, store, id, rp289(c3b)) } + t.Cleanup(func() { resolvedPathRouteRepairPreWriteHook = old }) + + runRepair289(t, store) + + rp := resolvedPathOf188(t, store, id) + if !rp.Valid || rp.String != rp289(c3b) { + t.Fatalf("resolved_path = %v, want the concurrently written %s", rp, rp289(c3b)) + } + if st := repairState289(t, store); st.Cleared != 0 || st.Repaired != 0 { + t.Fatalf("state = %d repaired, %d cleared; want 0/0: the guarded UPDATE matched no row", st.Repaired, st.Cleared) + } +} + +// Resolving against a cold prefix index or neighbour graph would move the +// watermark past rows that cannot resolve yet, so the pass waits for the +// post-build graph instead of writing anything. +func TestResolvedPathRouteRepair_WaitsForIndexAndGraph_289(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + id := seedObservation289(t, store, routeDirect188, mask289(mixedMask289), floodFrame289, `["c3"]`, ts190) + + ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + if err := store.repairResolvedPathRouteTypes(ctx, store.db); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("pass on a cold index returned %v, want context.DeadlineExceeded", err) + } + if st := repairState289(t, store); st.Watermark != 0 { + t.Fatalf("watermark = %d, want 0: no batch may run before the index and graph are ready", st.Watermark) + } + if rp := resolvedPathOf188(t, store, id); rp.Valid { + t.Fatalf("resolved_path = %s, want the untouched NULL", rp.String) + } + + primeIndexAndGraph188(t, store) + runRepair289(t, store) + if rp := resolvedPathOf188(t, store, id); !rp.Valid || rp.String != rp289(c3a) { + t.Fatalf("resolved_path = %v, want %s once the graph is ready", rp, rp289(c3a)) + } +} + +// The pass is scheduled as an async migration, is recorded done, and a later +// start does not run it again. +func TestResolvedPathRouteRepair_RunsOnceAsAnAsyncMigration_289(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + id := seedObservation289(t, store, routeFlood188, mask289(mixedMask289), directFrame289, `["c3"]`, ts190) + setResolvedPath289(t, store, id, rp289(c3a)) + primeIndexAndGraph188(t, store) + + store.StartResolvedPathRouteRepair(context.Background()) + store.WaitForAsyncMigrations() + + if status, err := store.AsyncMigrationStatus(resolvedPathRouteRepairMigration); err != nil || status != "done" { + t.Fatalf("migration status = %q, %v; want done", status, err) + } + if rp := resolvedPathOf188(t, store, id); rp.Valid { + t.Fatalf("resolved_path = %s, want NULL", rp.String) + } + + resetRepairWatermark289(t, store) + store.StartResolvedPathRouteRepair(context.Background()) + store.WaitForAsyncMigrations() + if st := repairState289(t, store); st.Watermark != 0 { + t.Fatalf("watermark = %d, want 0: a migration recorded done must not run again", st.Watermark) + } +} diff --git a/cmd/ingestor/resolved_path_route_repair_bench_289_test.go b/cmd/ingestor/resolved_path_route_repair_bench_289_test.go new file mode 100644 index 000000000..78b414225 --- /dev/null +++ b/cmd/ingestor/resolved_path_route_repair_bench_289_test.go @@ -0,0 +1,203 @@ +package main + +import ( + "context" + "database/sql" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// Sizing for the #289 M2 repair pass. The plan on #289 asks for the run time +// at the ~6.5M observations of the #89 sizing comment, so this builds a +// synthetic database of that shape at a fraction of the size and reports the +// per-observation cost, the writer hold percentiles and the extrapolation. +// +// Run it with: +// +// go test ./cmd/ingestor -run '^$' -bench ResolvedPathRouteRepairPass_289 -benchtime 1x -timeout 60m +// +// The row mix follows the plan's reasoning: a transmission row is shared by +// receptions of the same payload, most of which carry the route type of the +// first reception, and only a small share carry the other family. The share is +// a package var so a sizing run can try a different one. +var ( + repairBenchObservations = 1_200_000 + repairBenchObsPerTx = 3 + repairBenchCandidateEvery = 100 // 1 in N observations is a #289 candidate + repairBenchTargetRows = 6_500_000 +) + +// obsBench289 are the observers the synthetic rows are spread over: the +// observations table rejects two rows with the same transmission, observer and +// path (idx_observations_dedup), so several receptions of one transmission +// need different observers. Each gets the same obs <-> c3a edge as obs188, so +// a flood ["c3"] resolves to c3a for all of them. +var obsBench289 = []string{ + obs188, + "0b5e000000000000000000000000000000000000000000000000000000000002", + "0b5e000000000000000000000000000000000000000000000000000000000003", +} + +// seedRepairBenchDB fills store with repairBenchObservations observations over +// repairBenchObservations/repairBenchObsPerTx transmissions. Every third +// transmission carries a mixed route_mask; inside those, one observation in +// repairBenchCandidateEvery carries a frame of the other route family, which +// makes it a repair candidate. Every other observation carries a frame of the +// transmission's own family, so the SQL prefilter drops it without the pass +// reading the row. +func seedRepairBenchDB(b *testing.B, store *Store) (candidates int) { + b.Helper() + if repairBenchObsPerTx > len(obsBench289) { + b.Fatalf("repairBenchObsPerTx = %d, only %d observers are seeded", repairBenchObsPerTx, len(obsBench289)) + } + obsIdx := make([]int64, 0, repairBenchObsPerTx) + for _, id := range obsBench289[:repairBenchObsPerTx] { + if err := store.UpsertObserver(strings.ToUpper(id), "bench-289", "", nil); err != nil { + b.Fatal(err) + } + if _, err := store.db.Exec(`INSERT OR IGNORE INTO neighbor_edges (node_a, node_b, count, last_seen) + VALUES (?, ?, 5, '2026-06-01T00:00:00Z')`, id, c3a); err != nil { + b.Fatal(err) + } + var idx int64 + if err := store.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, strings.ToUpper(id)).Scan(&idx); err != nil { + b.Fatal(err) + } + obsIdx = append(obsIdx, idx) + } + txCount := repairBenchObservations / repairBenchObsPerTx + start := time.Now() + rows := 0 + // The insert runs in its own function so a failure rolls the transaction + // back before b.Fatal unwinds the goroutine: the store holds a single + // SQLite connection, and an open transaction would deadlock the cleanup. + err := func() error { + tx, err := store.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() // no-op after Commit + insTx, err := tx.Prepare(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, route_mask) + VALUES ('00', ?, '2026-06-01T00:00:00Z', ?, 5, '{}', ?)`) + if err != nil { + return err + } + defer insTx.Close() + insObs, err := tx.Prepare(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, raw_hex, resolved_path) + VALUES (?, ?, '["c3"]', ?, ?, ?)`) + if err != nil { + return err + } + defer insObs.Close() + for i := 0; i < txCount; i++ { + mask := sql.NullInt64{Int64: floodMask289, Valid: true} + mixed := i%3 == 0 + if mixed { + mask = sql.NullInt64{Int64: mixedMask289, Valid: true} + } + res, err := insTx.Exec(fmt.Sprintf("bench289-%d", i), routeFlood188, mask) + if err != nil { + return fmt.Errorf("transmission %d: %w", i, err) + } + txID, _ := res.LastInsertId() + for j := 0; j < repairBenchObsPerTx; j++ { + rows++ + frame := floodFrame289 + if mixed && rows%repairBenchCandidateEvery == 0 { + // A DIRECT reception the pre-#308 backfill anchored on the + // observer: the pass must clear the stored relay. + frame = directFrame289 + candidates++ + } + if _, err := insObs.Exec(txID, obsIdx[j], ts190+int64(rows), frame, rp289(c3a)); err != nil { + return fmt.Errorf("observation %d: %w", rows, err) + } + } + } + return tx.Commit() + }() + if err != nil { + b.Fatal(err) + } + b.Logf("seeded %d transmissions / %d observations (%d candidates) in %s", + txCount, rows, candidates, time.Since(start).Round(time.Second)) + return candidates +} + +// BenchmarkResolvedPathRouteRepairPass_289 runs the whole pass once over the +// synthetic database and reports the cost per observation scanned, the writer +// hold percentiles and the extrapolation to repairBenchTargetRows. +func BenchmarkResolvedPathRouteRepairPass_289(b *testing.B) { + for i := 0; i < b.N; i++ { + func() { + dbPath := filepath.Join(b.TempDir(), "ingest.db") + store := backfillFixture188(b, dbPath, true) + defer store.Close() + wantCandidates := seedRepairBenchDB(b, store) + primeIndexAndGraph188(b, store) + store.Checkpoint() + if fi, err := os.Stat(dbPath); err == nil { + b.Logf("synthetic database: %.0f MB", float64(fi.Size())/(1<<20)) + } + + oldPause := resolvedPathRouteRepairPause + resolvedPathRouteRepairPause = 0 // the pause is wall-clock, not cost + defer func() { resolvedPathRouteRepairPause = oldPause }() + before := writerStatsAgg.get(resolvedPathRouteRepairComponent).snapshot() + + b.ResetTimer() + start := time.Now() + if err := store.repairResolvedPathRouteTypes(context.Background(), store.db); err != nil { + b.Fatal(err) + } + elapsed := time.Since(start) + b.StopTimer() + + st, err := store.ResolvedPathRouteRepairState() + if err != nil || st == nil { + b.Fatalf("state: %v", err) + } + if int(st.Cleared) != wantCandidates { + b.Fatalf("cleared %d rows, want the %d seeded candidates", st.Cleared, wantCandidates) + } + w := writerStatsAgg.get(resolvedPathRouteRepairComponent).snapshot() + perRow := float64(elapsed.Nanoseconds()) / float64(st.Ceiling) + b.Logf("pass over %d observations: %s wall (%.0f ns/observation), %d candidates, %d cleared, %d batches", + st.Ceiling, elapsed.Round(time.Millisecond), perRow, st.Candidates, st.Cleared, w.Count-before.Count) + b.Logf("writer hold ms: p50 %.1f p95 %.1f p99 %.1f max %.1f over %d transactions (window %d, batch %d)", + w.HoldMsP50, w.HoldMsP95, w.HoldMsP99, w.HoldMsMax, w.Count-before.Count, + resolvedPathRouteRepairIDWindow, resolvedPathRouteRepairBatchSize) + b.Logf("extrapolated to %d observations: %s of CPU plus the %s pause per batch", + repairBenchTargetRows, + time.Duration(perRow*float64(repairBenchTargetRows)).Round(time.Second), + oldPause) + b.ReportMetric(perRow, "ns/observation") + }() + } +} + +// BenchmarkResolvedPathRouteRepairBatch_289 is the per-batch cost on the same +// synthetic database: one window read, resolved and written. +func BenchmarkResolvedPathRouteRepairBatch_289(b *testing.B) { + store := backfillFixture188(b, filepath.Join(b.TempDir(), "ingest.db"), true) + defer store.Close() + seedRepairBenchDB(b, store) + primeIndexAndGraph188(b, store) + st, err := store.resolvedPathRouteRepairStart() + if err != nil { + b.Fatal(err) + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + if st.Watermark >= st.Ceiling { + st.Watermark = 0 // wrap: the rows are already repaired, the scan cost is the same + } + if _, err := store.resolvedPathRouteRepairBatch(context.Background(), st); err != nil { + b.Fatal(err) + } + } +} diff --git a/scripts/resolved-path-route-repair-estimate.sh b/scripts/resolved-path-route-repair-estimate.sh new file mode 100755 index 000000000..b75d6a184 --- /dev/null +++ b/scripts/resolved-path-route-repair-estimate.sh @@ -0,0 +1,147 @@ +#!/usr/bin/env bash +# resolved-path-route-repair-estimate.sh — read-only sizing for the #289 M2 +# resolved_path route-type repair. +# +# Prints the candidate counts and an estimated cost for one full pass. It NEVER +# writes: the database is opened through an immutable URI, so SQLite cannot +# create a -wal/-shm file or take a write lock, and it is safe to run next to a +# live ingestor. No pubkeys, paths or frame bytes are printed — counts only. +# +# RUN IT ON A SNAPSHOT COPY. immutable=1 is what makes this read-only, and it +# also tells SQLite to ignore the -wal file: against a live database the counts +# miss every row that has not been checkpointed yet, so they are a lower bound, +# and a file changing underneath an immutable reader can make that reader +# report wrong counts or a spurious SQLITE_CORRUPT (the database itself is +# never damaged). A backup file or a stopped container's volume gives exact +# numbers. Two further approximations, on any input: +# - the candidate count is the SQL prefilter the pass uses, which is a +# superset: the pass re-checks each row in Go with the #289/#308 helper and +# writes only the rows whose route type actually changes the resolve; +# - the scan rate is the BenchmarkResolvedPathRouteRepairPass_289 figure +# below, measured warm-cache on a synthetic database. It is not a +# measurement of this host or of cold disk I/O. +# It also runs several full-table scans of observations, so on a multi-GB +# database the script's own wall clock is a minute or more. +# +# Usage: +# scripts/resolved-path-route-repair-estimate.sh /path/to/corescope.db +# +# The scan rate used for the estimate is the measured benchmark figure (see the +# PR body for the hardware); pass a different one as the second argument: +# scripts/resolved-path-route-repair-estimate.sh /path/to/db 1200000 +set -eu + +DB=${1:-} +ROWS_PER_SEC=${2:-2400000} +if [ -z "$DB" ]; then + echo "usage: $0 [observation_rows_per_second]" >&2 + exit 2 +fi +if [ ! -r "$DB" ]; then + echo "error: $DB is not readable" >&2 + exit 2 +fi +command -v sqlite3 >/dev/null 2>&1 || { echo "error: sqlite3 not found" >&2; exit 2; } + +# file: URI with immutable=1 — read-only, no locking, no journal files. +ABS=$(cd "$(dirname "$DB")" && pwd)/$(basename "$DB") +URI="file:${ABS}?immutable=1" + +q() { sqlite3 -readonly "$URI" "$1"; } + +# has_col — pre-#881 databases have no observations.raw_hex +# and pre-#89 ones no transmissions.route_mask; querying either would be an +# error rather than a count of zero. +has_col() { + sqlite3 -readonly "$URI" "SELECT COUNT(*) FROM pragma_table_info('$1') WHERE name = '$2';" +} +RAW_COL=$(has_col observations raw_hex) +MASK_COL=$(has_col transmissions route_mask) + +echo "database: $ABS" +echo "file size (bytes): $(wc -c < "$DB" | tr -d ' ')" +echo + +# Scanned extent: the pass walks observations.id from 0 to MAX(id), which is +# why NULL rows below the #188 backfill watermark are covered too. +echo "observations" +echo " rows: $(q 'SELECT COUNT(*) FROM observations;')" +echo " max id (scan end): $(q 'SELECT COALESCE(MAX(id), 0) FROM observations;')" +if [ "$RAW_COL" = "1" ]; then + echo " with a frame: $(q 'SELECT COUNT(*) FROM observations WHERE length(raw_hex) >= 2;')" +else + echo " with a frame: column absent (pre-#881 database)" +fi +echo + +# Candidates, split by bug direction. Both use the pass's own prefilter: the +# route family is bit 1 of the low nibble of the frame's header byte, i.e. of +# raw_hex's second character, and (route_type & 2) is the transmission's. +HDR_CAND=0 +HDR_NULL=0 +MASK_CAND=0 +if [ "$RAW_COL" = "1" ]; then + HDR="length(o.raw_hex) >= 2 + AND ((instr('0123456789abcdef', lower(substr(o.raw_hex, 2, 1))) - 1) & 2) != (COALESCE(t.route_type, -1) & 2) + AND COALESCE(o.path_json, '') NOT IN ('', '[]')" + HDR_CAND=$(q "SELECT COUNT(*) FROM observations o JOIN transmissions t ON t.id = o.transmission_id WHERE $HDR;") + HDR_NULL=$(q "SELECT COUNT(*) FROM observations o JOIN transmissions t ON t.id = o.transmission_id WHERE $HDR AND o.resolved_path IS NULL;") +fi +echo "candidates: frame header disagrees with t.route_type" +echo " rows: $HDR_CAND" +echo " of those NULL: $HDR_NULL (FLOOD read as DIRECT; never revisited by the #188 backfill)" +echo " of those stored: $((HDR_CAND - HDR_NULL)) (DIRECT read as FLOOD; may carry a wrong relay)" +echo + +if [ "$RAW_COL" = "1" ] && [ "$MASK_COL" = "1" ]; then + MASK_CAND=$(q "SELECT COUNT(*) FROM observations o JOIN transmissions t ON t.id = o.transmission_id + WHERE COALESCE(length(o.raw_hex), 0) < 2 + AND t.route_mask IS NOT NULL + AND (t.route_mask & 3) != 0 AND (t.route_mask & 12) != 0 + AND COALESCE(t.route_type, -1) IN (0, 1) + AND COALESCE(o.path_json, '') NOT IN ('', '[]');") + echo "candidates: no frame, route_mask holds both families, FLOOD transmission" + echo " rows: $MASK_CAND" +else + echo "candidates: no frame, route_mask holds both families, FLOOD transmission" + echo " column absent (pre-#881 or pre-#89 database): not counted" +fi +echo + +# Progress of a pass that already ran on this database. +if [ "$(q "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'resolved_path_route_repair_state';")" = "1" ]; then + echo "recorded run" + q "SELECT ' watermark/ceiling: ' || watermark || '/' || ceiling || char(10) || + ' scanned: ' || scanned || char(10) || + ' candidates: ' || candidates || char(10) || + ' repaired: ' || repaired || char(10) || + ' cleared (to NULL): ' || cleared || char(10) || + ' started at: ' || started_at || char(10) || + ' updated at: ' || updated_at + FROM resolved_path_route_repair_state WHERE id = 1;" +else + echo "recorded run: none (no #289 M2 pass has run on this database)" +fi +echo + +# Estimate. The pass walks the id range once in bounded windows; the deliberate +# pause between windows dominates the wall clock, so both parts are reported. +MAXID=$(q 'SELECT COALESCE(MAX(id), 0) FROM observations;') +CAND=$((HDR_CAND + MASK_CAND)) +echo "estimated cost of one full pass (defaults: idWindow 20000, batchSize 500, pause 250 ms)" +awk -v maxid="$MAXID" -v cand="$CAND" -v rps="$ROWS_PER_SEC" 'BEGIN { + work = (rps > 0) ? maxid / rps : 0; + windows = int((maxid + 19999) / 20000); + batches = (cand > 0) ? int((cand + 499) / 500) : 0; + if (batches < windows) batches = windows; + pause = batches * 0.25; + printf " candidate rows: %d\n", cand; + printf " batches: %d\n", batches; + printf " scan+write: %.1f s (at %d observation rows/s)\n", work, rps; + printf " deliberate pauses: %.1f s\n", pause; + printf " total wall clock: %.1f s (%.1f min)\n", work + pause, (work + pause) / 60; +}' +echo +echo "Counts are exact on a snapshot copy; on a live WAL database they omit" +echo "un-checkpointed rows and the batch count is a lower bound (see the header)." +echo "This script wrote nothing. Verify with: ls -l \"$ABS\"-wal \"$ABS\"-shm 2>/dev/null" diff --git a/scripts/test-resolved-path-route-repair-estimate.sh b/scripts/test-resolved-path-route-repair-estimate.sh new file mode 100755 index 000000000..c2b7515ba --- /dev/null +++ b/scripts/test-resolved-path-route-repair-estimate.sh @@ -0,0 +1,131 @@ +#!/usr/bin/env bash +# test-resolved-path-route-repair-estimate.sh — unit tests for the #289 M2 +# read-only measurement script. The point under test is the read-only claim: +# the coordinator runs this against a staging copy next to a live ingestor, so +# it must not create a journal, must not touch the file, and must print counts +# rather than row contents. The counted candidate split is the second point: it +# is what decides whether the repair is worth running on a given database. +set -eu + +HERE=$(cd "$(dirname "$0")" && pwd) +SCRIPT="$HERE/resolved-path-route-repair-estimate.sh" +TMP=$(mktemp -d) +trap 'rm -rf "$TMP"' EXIT +fails=0 + +fail() { echo "FAIL: $*" >&2; fails=$((fails + 1)); } +ok() { echo "ok: $*"; } + +DB="$TMP/t.db" +sqlite3 "$DB" <<'SQL' +CREATE TABLE transmissions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, raw_hex TEXT NOT NULL, + hash TEXT NOT NULL UNIQUE, first_seen TEXT NOT NULL, + route_type INTEGER, route_mask INTEGER, payload_type INTEGER, decoded_json TEXT); +CREATE TABLE observations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, transmission_id INTEGER NOT NULL, + observer_idx INTEGER NOT NULL, path_json TEXT, timestamp INTEGER, + raw_hex TEXT, resolved_path TEXT); +-- tx 1: FLOOD first reception, both families seen. tx 2: DIRECT, both seen. +-- tx 3: FLOOD, flood only. tx 4: DIRECT, direct only. +INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, route_mask, payload_type, decoded_json) VALUES + ('00', 'h1', 't', 1, 6, 5, '{}'), + ('00', 'h2', 't', 2, 6, 5, '{}'), + ('00', 'h3', 't', 1, 2, 5, '{}'), + ('00', 'h4', 't', 2, 4, 5, '{}'); +INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, raw_hex, resolved_path) VALUES + -- header candidates: DIRECT frame (16) under the FLOOD tx, stored relay. + (1, 1, '["c3"]', 1, '1601c3', '["c355"]'), + (1, 2, '["c3"]', 2, '1601c3', '["c355"]'), + -- header candidate: FLOOD frame (15) under the DIRECT tx, still NULL. + (2, 1, '["c3"]', 3, '1501c3', NULL), + -- not candidates: the frame and t.route_type agree. + (1, 3, '["c3"]', 4, '1501c3', '["c355"]'), + (2, 2, '["c3"]', 5, '1601c3', NULL), + -- not a candidate: no hops at all. + (1, 4, '[]', 6, '1601c3', NULL), + -- route_mask candidate: no frame, both families, FLOOD tx. + (1, 5, '["c3"]', 7, NULL, '["c355"]'), + -- not candidates: no frame, mixed mask but a DIRECT tx; single-family mask. + (2, 3, '["c3"]', 8, NULL, NULL), + (3, 1, '["c3"]', 9, NULL, '["c355"]'), + (4, 1, '["c3"]', 10, NULL, NULL); +SQL +# Drop the journal the seeding created, so the read-only assertion below is +# about this script and not about sqlite3's own write. +sqlite3 "$DB" 'PRAGMA journal_mode=delete;' >/dev/null +rm -f "$DB-wal" "$DB-shm" +BEFORE=$(cksum < "$DB") + +OUT=$("$SCRIPT" "$DB") + +expect() { + if printf '%s' "$OUT" | grep -qE "$1"; then ok "$2"; else fail "$2 (pattern: $1)"; fi +} +expect '^ rows: 10$' "10 observations counted" +expect '^ max id \(scan end\): 10$' "the scan end is MAX(observations.id)" +expect '^ with a frame: 6$' "6 rows carry a frame" +expect '^ rows: 3$' "3 header candidates counted (both directions)" +expect '^ of those NULL: 1 ' "1 header candidate is a NULL row" +expect '^ of those stored: 2 ' "2 header candidates carry a stored path" +expect '^ rows: 1$' "1 route_mask candidate counted" +expect 'recorded run: none' "no recorded run on a fresh database" +expect ' candidate rows: 4$' "candidate total is header + route_mask" +expect 'total wall clock' "an estimate is printed" + +# Read-only: no journal files, and the database byte-identical. +for sfx in -wal -shm -journal; do + if [ -e "$DB$sfx" ]; then fail "the script created $DB$sfx"; fi +done +ok "no journal files created" +if [ "$(cksum < "$DB")" != "$BEFORE" ]; then fail "the script modified the database"; else ok "database unchanged"; fi + +# A recorded run is reported back. +sqlite3 "$DB" "CREATE TABLE resolved_path_route_repair_state (id INTEGER PRIMARY KEY, watermark INTEGER, + ceiling INTEGER, scanned INTEGER, candidates INTEGER, repaired INTEGER, cleared INTEGER, + started_at TEXT, updated_at TEXT); +INSERT INTO resolved_path_route_repair_state VALUES (1, 10, 10, 4, 4, 1, 3, 's', 'u');" >/dev/null +sqlite3 "$DB" 'PRAGMA journal_mode=delete;' >/dev/null +rm -f "$DB-wal" "$DB-shm" +RUNOUT=$("$SCRIPT" "$DB") +if printf '%s' "$RUNOUT" | grep -qE '^ cleared \(to NULL\): 3$'; then + ok "a recorded run is reported" +else + fail "a recorded run was not reported: $RUNOUT" +fi + +# A pre-#881 database (no observations.raw_hex) must degrade to a note. +OLD="$TMP/old.db" +sqlite3 "$OLD" "CREATE TABLE transmissions (id INTEGER PRIMARY KEY, raw_hex TEXT, hash TEXT UNIQUE, + first_seen TEXT, route_type INTEGER, payload_type INTEGER); +CREATE TABLE observations (id INTEGER PRIMARY KEY, transmission_id INTEGER, observer_idx INTEGER, + path_json TEXT, timestamp INTEGER, resolved_path TEXT); +INSERT INTO transmissions VALUES (1, '00', 'o1', 't', 1, 5); +INSERT INTO observations VALUES (1, 1, 1, '[\"c3\"]', 1, NULL);" >/dev/null +sqlite3 "$OLD" 'PRAGMA journal_mode=delete;' >/dev/null +rm -f "$OLD-wal" "$OLD-shm" +if OLDOUT=$("$SCRIPT" "$OLD" 2>&1); then + if printf '%s' "$OLDOUT" | grep -q 'pre-#881 database'; then + ok "pre-#881 database reports the absent column" + else + fail "pre-#881 database did not report the absent column: $OLDOUT" + fi + if printf '%s' "$OLDOUT" | grep -qi 'error'; then fail "pre-#881 run printed an error: $OLDOUT"; else ok "pre-#881 run is error-free"; fi + if printf '%s' "$OLDOUT" | grep -qE ' candidate rows: 0$'; then + ok "pre-#881 database has no candidates" + else + fail "pre-#881 database reported candidates: $OLDOUT" + fi +else + fail "the script exited non-zero on a pre-#881 database: $OLDOUT" +fi + +# Argument handling. +if "$SCRIPT" >/dev/null 2>&1; then fail "a missing argument must fail"; else ok "missing argument rejected"; fi +if "$SCRIPT" "$TMP/does-not-exist.db" >/dev/null 2>&1; then fail "an unreadable path must fail"; else ok "unreadable path rejected"; fi + +if [ "$fails" -ne 0 ]; then + echo "$fails check(s) failed" >&2 + exit 1 +fi +echo "all checks passed" From 69e71c09e28e65490708d72c5083adfe1301b2c9 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 25 Oct 2026 11:23:43 +0100 Subject: [PATCH 2/2] reviewfix(ingestor): only repair rows the pre-#308 rule wrote (#289 M2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 2 of the review on PR #590. F1 (high): the repair now writes a row only when the stored value is exactly what the pre-#308 rule produces today (stored == had), not merely when it differs from the new answer. A family mismatch is not by itself a #289 defect: live ingest has always resolved with the frame's own route type, so most mismatched rows hold a correct ingest value, which differs from today's `had` as soon as the neighbour graph or the prefix index has drifted. Those rows were being rewritten, or cleared to NULL. The reviewer's two probe rows on the #188 fixture are now a test. F2: the rollback section said a row cleared to NULL would be re-resolved by the #188 backfill. It would not — that backfill never revisits a row below its watermark — so a clear is permanent without a snapshot. The PR body says so now. F3: the pass is behind resolvedPathRouteRepair.enabled, default false. It overwrites values in place and keeps no copy, so it must not run on the first boot of a build that happens to contain it. Documented in config.example.json and docs/user-guide/configuration.md, both of which say to snapshot first. Customizer entry is a follow-up (rule 8). F4: tests for the transport route types 0 and 3, which differ from the other family's non-transport type in bit 1 only. They kill the mutant that reads bit 0 instead. F5: the batch's readiness check is a seam, so the loop's retry of a batch that finds the snapshot cold after the gate has passed is covered. F6: the sizing script splits the candidates by header family first (the bug direction), each direction then into stored and NULL rows, with unparseable headers on their own. One query instead of two scans. F7: the benchmark mix now includes rows that gain a path, not only rows that are cleared. 1.2M observations: 511 ms, p99 write hold 13.3 ms. Relates to #289 Co-Authored-By: Claude Opus 5 --- cmd/ingestor/config.go | 6 + cmd/ingestor/main.go | 6 +- .../observation_route_type_289_test.go | 8 + cmd/ingestor/resolved_path_route_repair.go | 93 +++++++-- .../resolved_path_route_repair_289_test.go | 196 +++++++++++++++++- ...solved_path_route_repair_bench_289_test.go | 59 ++++-- config.example.json | 5 + docs/user-guide/configuration.md | 20 ++ .../resolved-path-route-repair-estimate.sh | 58 ++++-- ...est-resolved-path-route-repair-estimate.sh | 33 ++- 10 files changed, 419 insertions(+), 65 deletions(-) diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 914408151..aec656991 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -141,6 +141,12 @@ type Config struct { // observations.resolved_path rows (#188, resolved_path_backfill.go). ResolvedPathBackfill *ResolvedPathBackfillConfig `json:"resolvedPathBackfill,omitempty"` + // ResolvedPathRouteRepair opts into the one-off repair of + // observations.resolved_path rows written with the transmission's first + // route type (#289 M2, resolved_path_route_repair.go). Off by default: it + // overwrites values in place and keeps no copy of the previous ones. + ResolvedPathRouteRepair *ResolvedPathRouteRepairConfig `json:"resolvedPathRouteRepair,omitempty"` + // DecodeBackfill tunes the re-decoding of historical rows after the // effective channel/region key set grows (#380, decode_backfill.go). DecodeBackfill *DecodeBackfillConfig `json:"decodeBackfill,omitempty"` diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index a24c86281..ac1c25937 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -533,12 +533,14 @@ func main() { } // #289 M2: repair resolved_path rows the pre-#308 backfill wrote with the - // transmission's first route type, once per database. Same placement as + // transmission's first route type, once per database. Opt-in: the pass + // overwrites values in place and keeps no copy, so it waits for + // resolvedPathRouteRepair.enabled (PR #590 review F3). Same placement as // the backfill above (it needs the same primed index and graph) and the // same shutdown: cancelled here, resumed from its watermark next start. routeRepairCtx, stopResolvedPathRouteRepair := context.WithCancel(context.Background()) defer stopResolvedPathRouteRepair() - store.StartResolvedPathRouteRepair(routeRepairCtx) + store.StartResolvedPathRouteRepair(routeRepairCtx, cfg.ResolvedPathRouteRepairEnabled()) // #380: re-decode historical enc_ GRP_TXT rows and unknown-scoped // transport rows whenever the effective key set grows (config/SIGHUP, diff --git a/cmd/ingestor/observation_route_type_289_test.go b/cmd/ingestor/observation_route_type_289_test.go index 381cc63e7..eef7c6181 100644 --- a/cmd/ingestor/observation_route_type_289_test.go +++ b/cmd/ingestor/observation_route_type_289_test.go @@ -25,6 +25,14 @@ const ( directFrame289 = "1601c3aabb" mixedMask289 = int64(0b0110) // routes 1 (FLOOD) and 2 (DIRECT) floodMask289 = int64(0b0010) // route 1 only + + // The transport route types, 0 (TRANSPORT_FLOOD) and 3 + // (TRANSPORT_DIRECT). They share the low bit with the other family's + // non-transport type, so only bit 1 separates the families: a candidate + // test that looked at bit 0 would mis-classify exactly these two + // (PR #590 review F4). + transportFloodFrame289 = "1401c3aabb" + transportDirectFrame289 = "1701c3aabb" ) func mask289(m int64) sql.NullInt64 { return sql.NullInt64{Int64: m, Valid: true} } diff --git a/cmd/ingestor/resolved_path_route_repair.go b/cmd/ingestor/resolved_path_route_repair.go index cec06a470..bb7c1fd06 100644 --- a/cmd/ingestor/resolved_path_route_repair.go +++ b/cmd/ingestor/resolved_path_route_repair.go @@ -44,22 +44,33 @@ import ( // // What it writes. For each candidate the pass resolves the hops twice with the // current graph and index: once with the observation's own route type (`want`) -// and once with t.route_type (`had`, what the pre-#308 reader produced). It -// writes only when +// and once with t.route_type (`had`, what the pre-#308 reader would produce +// now). It writes only when // // 1. the two differ — so a row whose answer does not depend on the route type // is left alone even if its stored value is stale for some other reason // (that is not #289, and rewriting it would turn this pass into an // unbounded re-resolve of history); and -// 2. the stored value is not already `want` — so a row live ingest resolved -// correctly, and a row an earlier run of this pass already repaired, is -// not written again. +// 2. the stored value is exactly `had` — the signature of a row the pre-#308 +// rule wrote. A family mismatch on its own is not that signature: live +// ingest has always resolved with the frame's own route type, so most +// mismatched rows hold a correct ingest value, which differs from today's +// `had` as soon as the graph or the prefix index has drifted. Those rows +// are read and left alone, and so is a row an earlier run of this pass +// already repaired, whose stored value is `want` (PR #590 review F1). // // `want` may be empty, and then the row is set back to NULL: that is the // repair for a DIRECT reception stored with a wrong relay. Every UPDATE is // guarded on the value the batch read, so a row live ingest rewrote in the // meantime keeps ingest's value and a repeated batch is a no-op. // +// The residual the two conditions cannot cover: a row whose stored value is +// both a correct ingest value and exactly what the pre-#308 rule produces +// today. Nothing in the database distinguishes the two writers, so such a row +// is rewritten. It is also the case where the rewrite is least harmful — the +// stored value is the one the buggy rule gives — but it is not free, and it is +// why the rollback section asks for a snapshot. +// // Bounds and resume (the #89 / #380 shape): // - the pass needs the primed prefix index and a post-build neighbour graph // with at least one edge (resolvedPathBackfillReady) — resolving against a @@ -85,6 +96,26 @@ const ( resolvedPathRouteRepairMigration = "resolved_path_route_type_repair_v1" ) +// ResolvedPathRouteRepairConfig is the opt-in switch for the repair +// (PR #590 review F3). The pass overwrites observations.resolved_path in +// place and keeps no copy of the previous value, so it must not run on the +// first boot of a build that happens to contain it: an operator turns it on +// deliberately, after taking a database snapshot. Rows it cleared to NULL are +// below the #188 backfill's watermark and are never revisited, so the +// snapshot is the only way back. +// +// AGENTS.md rule 8: a customizer entry for this switch, and for the three +// bounds below, is a follow-up and not part of this PR. +type ResolvedPathRouteRepairConfig struct { + Enabled bool `json:"enabled,omitempty"` +} + +// ResolvedPathRouteRepairEnabled reports whether the one-off #289 M2 repair +// may run. Default false: an absent or empty block leaves it off. +func (c *Config) ResolvedPathRouteRepairEnabled() bool { + return c != nil && c.ResolvedPathRouteRepair != nil && c.ResolvedPathRouteRepair.Enabled +} + // Bounds. Package vars so tests can drive the multi-batch loop with a few // rows; customizer candidates later (AGENTS.md rule 8), like the #188 // backfill's batchSize/pauseMs. @@ -101,17 +132,34 @@ var ( // resolvedPathRouteRepairPreWriteHook is a test seam, nil in production: // called after a batch has resolved its rows and before it writes them. resolvedPathRouteRepairPreWriteHook func() + + // resolvedPathRouteRepairBatchReady is the batch's own readiness check, + // re-run because the loop's gate can pass and the snapshot stop being + // usable before the batch reads it. A variable so the #590 F5 test can + // make one batch see a cold snapshot without winning that race for real. + resolvedPathRouteRepairBatchReady = func(s *Store, idx prefixIndex, graph *NeighborGraph) bool { + return s.resolvedPathBackfillReady(idx, graph) + } ) // resolvedPathRouteRepairProgressEvery logs progress every N batches. const resolvedPathRouteRepairProgressEvery = 50 -// StartResolvedPathRouteRepair schedules the repair as an async migration. -// Call it after the neighbour-edges builder is running, next to the #188 -// backfill: the pass needs that builder's graph and must not delay boot. -// Cancelling ctx stops it at the next statement; it is then recorded as -// unfinished and resumes from its watermark on the next start. -func (s *Store) StartResolvedPathRouteRepair(ctx context.Context) { +// StartResolvedPathRouteRepair schedules the repair as an async migration when +// the operator has opted in (resolvedPathRouteRepair.enabled). Call it after +// the neighbour-edges builder is running, next to the #188 backfill: the pass +// needs that builder's graph and must not delay boot. Cancelling ctx stops it +// at the next statement; it is then recorded as unfinished and resumes from +// its watermark on the next start. +// +// With enabled false nothing is scheduled and nothing is recorded in +// _async_migrations, so turning the switch on later still runs the pass. +func (s *Store) StartResolvedPathRouteRepair(ctx context.Context, enabled bool) { + if !enabled { + log.Printf("[%s] disabled: set resolvedPathRouteRepair.enabled to run the one-off #289 repair (take a database snapshot first: it overwrites resolved_path in place)", + resolvedPathRouteRepairComponent) + return + } // PREFLIGHT: async=true reason="#289 M2 one-off repair: windowed scan of observations joined to transmissions with batched rewrites; must not block ingestor boot" if err := s.RunAsyncMigration(ctx, resolvedPathRouteRepairMigration, s.repairResolvedPathRouteTypes); err != nil { log.Printf("[%s] scheduling %s failed: %v", resolvedPathRouteRepairComponent, resolvedPathRouteRepairMigration, err) @@ -333,7 +381,7 @@ var resolvedPathRouteRepairCandidateSQL = fmt.Sprintf(`( func (s *Store) resolvedPathRouteRepairBatch(ctx context.Context, st *ResolvedPathRouteRepairState) (resolvedPathRouteRepairBatchResult, error) { var b resolvedPathRouteRepairBatchResult graph, idx := s.neighborGraph.load(), s.prefixIdx.load() - if !s.resolvedPathBackfillReady(idx, graph) { + if !resolvedPathRouteRepairBatchReady(s, idx, graph) { return b, errResolvedPathBackfillNotReady } windowEnd := st.Watermark + resolvedPathRouteRepairIDWindow @@ -403,11 +451,24 @@ func (s *Store) resolvedPathRouteRepairBatch(ctx context.Context, st *ResolvedPa from = strings.ToLower(r.fromPubkey) } want := marshalResolvedPath(resolvedPathRouteRepairResolve(hops, from, r.observerID, rt, graph, idx)) - // What the pre-#308 reader produced for this row with today's graph. - // When the route type does not change the answer, the stored value is - // not a #289 defect and this pass leaves it alone. + // What the pre-#308 reader would produce for this row with today's + // graph. Two conditions have to hold before the row is rewritten: + // + // - want != had: the route type actually changes the answer. A + // candidate whose answer is route-type-independent is not a #289 + // defect, and rewriting it would turn this into an unbounded + // re-resolve of history. + // - r.stored == had: the stored value is exactly what the pre-#308 + // rule produces, which is the signature of the defect. A family + // mismatch alone is not: live ingest has always used the frame's + // own route type, so most mismatched rows were written correctly + // by ingest, and when the graph or the index has drifted since, + // today's answer differs from theirs. Requiring the stored value + // to match `had` leaves those rows — and a row an earlier run + // already repaired, whose stored value is `want` — alone + // (PR #590 review F1). had := marshalResolvedPath(resolvedPathRouteRepairResolve(hops, from, r.observerID, r.txRoute, graph, idx)) - if want == had || r.stored == want { + if want == had || r.stored != had { continue } updates = append(updates, update{r.id, want, r.stored}) diff --git a/cmd/ingestor/resolved_path_route_repair_289_test.go b/cmd/ingestor/resolved_path_route_repair_289_test.go index 219c72bb0..305869c8e 100644 --- a/cmd/ingestor/resolved_path_route_repair_289_test.go +++ b/cmd/ingestor/resolved_path_route_repair_289_test.go @@ -82,8 +82,10 @@ type repairCase289 struct { id int64 } -// Both bug directions, the route_mask fallback, the rows that must not change, -// and the mismatch whose answer does not depend on the route type. +// Both bug directions (the plain and the transport route types), the +// route_mask fallback, the rows that must not change, the drifted live-ingest +// rows that must not change either, and the mismatch whose answer does not +// depend on the route type. func repairCases289() []repairCase289 { return []repairCase289{ // #289 direction 1: a DIRECT reception anchored on the observer stored @@ -99,6 +101,24 @@ func repairCases289() []repairCase289 { {name: "no frame, mixed mask, FLOOD transmission", txRoute: routeFlood188, mask: mask289(mixedMask289), rawHex: "", path: `["c3"]`, stored: rp289(c3a), want: ""}, + // The same two directions with the transport route types, which differ + // from the other family's non-transport type in bit 1 only + // (PR #590 review F4). + {name: "TRANSPORT_DIRECT frame under a FLOOD transmission", txRoute: routeFlood188, mask: mask289(mixedMask289), + rawHex: transportDirectFrame289, path: `["c3"]`, stored: rp289(c3a), want: ""}, + {name: "TRANSPORT_FLOOD frame under a DIRECT transmission", txRoute: routeDirect188, mask: mask289(mixedMask289), + rawHex: transportFloodFrame289, path: `["c3"]`, stored: "", want: rp289(c3a)}, + + // Drifted live-ingest rows (PR #590 review F1): the family mismatch is + // real, but the stored value is not what the pre-#308 rule produces + // today, so the row was not written by that rule. Live ingest has + // always used the frame's own route type, so these are correct values + // resolved against an older graph, and the pass must leave them alone. + {name: "DIRECT frame under a FLOOD transmission, drifted stored relay", txRoute: routeFlood188, + mask: mask289(mixedMask289), rawHex: directFrame289, path: `["c3"]`, stored: rp289(c3b), want: rp289(c3b)}, + {name: "FLOOD frame under a DIRECT transmission, drifted stored relay", txRoute: routeDirect188, + mask: mask289(mixedMask289), rawHex: floodFrame289, path: `["c3"]`, stored: rp289(c3b), want: rp289(c3b)}, + // Not candidates: the frame header and t.route_type agree. {name: "FLOOD frame under a FLOOD transmission", txRoute: routeFlood188, mask: mask289(floodMask289), rawHex: floodFrame289, path: `["c3"]`, stored: rp289(c3a), want: rp289(c3a)}, @@ -112,11 +132,15 @@ func repairCases289() []repairCase289 { {name: "no frame, NULL mask", txRoute: routeFlood188, mask: sql.NullInt64{}, rawHex: "", path: `["c3"]`, stored: rp289(c3a), want: rp289(c3a)}, - // A candidate whose answer does not depend on the route type: ["c355"] - // names c3a uniquely either way. Its stored value is stale for some - // other reason, which is not #289, so the pass leaves it alone. + // Candidates whose answer does not depend on the route type: ["c355"] + // names c3a uniquely either way. One has a stored value that is stale + // for some other reason, which is not #289; the other already holds + // the answer. Neither is written — the second one would be a write of + // the value the row already has. {name: "DIRECT frame under a FLOOD transmission, unique 2-byte hop", txRoute: routeFlood188, mask: mask289(mixedMask289), rawHex: directFrame289, path: `["c355"]`, stored: rp289(c3b), want: rp289(c3b)}, + {name: "DIRECT frame under a FLOOD transmission, unique 2-byte hop, already correct", txRoute: routeFlood188, + mask: mask289(mixedMask289), rawHex: directFrame289, path: `["c355"]`, stored: rp289(c3a), want: rp289(c3a)}, } } @@ -175,11 +199,11 @@ func TestResolvedPathRouteRepair_RepairsBothDirections_289(t *testing.T) { if st.Watermark < st.Ceiling { t.Fatalf("watermark %d did not reach the ceiling %d", st.Watermark, st.Ceiling) } - if st.Repaired != 1 || st.Cleared != 2 { - t.Fatalf("state = %d repaired, %d cleared; want 1 repaired (the NULL flood row) and 2 cleared (the two wrong relays)", st.Repaired, st.Cleared) + if st.Repaired != 2 || st.Cleared != 3 { + t.Fatalf("state = %d repaired, %d cleared; want 2 repaired (the two NULL flood rows) and 3 cleared (the three wrong relays)", st.Repaired, st.Cleared) } - if st.Candidates != 4 { - t.Fatalf("candidates = %d, want 4 (three defects plus the route-type-independent row)", st.Candidates) + if st.Candidates != 9 { + t.Fatalf("candidates = %d, want 9 (five defects, two drifted rows and two route-type-independent rows)", st.Candidates) } } @@ -408,7 +432,7 @@ func TestResolvedPathRouteRepair_RunsOnceAsAnAsyncMigration_289(t *testing.T) { setResolvedPath289(t, store, id, rp289(c3a)) primeIndexAndGraph188(t, store) - store.StartResolvedPathRouteRepair(context.Background()) + store.StartResolvedPathRouteRepair(context.Background(), true) store.WaitForAsyncMigrations() if status, err := store.AsyncMigrationStatus(resolvedPathRouteRepairMigration); err != nil || status != "done" { @@ -419,9 +443,159 @@ func TestResolvedPathRouteRepair_RunsOnceAsAnAsyncMigration_289(t *testing.T) { } resetRepairWatermark289(t, store) - store.StartResolvedPathRouteRepair(context.Background()) + store.StartResolvedPathRouteRepair(context.Background(), true) store.WaitForAsyncMigrations() if st := repairState289(t, store); st.Watermark != 0 { t.Fatalf("watermark = %d, want 0: a migration recorded done must not run again", st.Watermark) } } + +// PR #590 review F1. A family mismatch is not by itself a #289 defect: live +// ingest has always resolved with the frame's own route type, so most +// mismatched rows were written correctly by ingest. When the neighbour graph +// or the prefix index has drifted since, today's answer differs from the +// stored one, and rewriting the row would destroy a correct value. The pass +// therefore writes only a row whose stored value is exactly what the pre-#308 +// rule produces today. +// +// Both rows below are the reviewer's repro on the #188 fixture: live ingest +// resolved them to c3b when c3b was the only "c3" node, and the edge that now +// picks c3a was added later. +func TestResolvedPathRouteRepair_LeavesDriftedLiveIngestRowsUntouched_289(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + // A DIRECT reception under a FLOOD transmission. Today DIRECT resolves to + // nothing and the pre-#308 FLOOD rule to c3a, so an unguarded pass clears + // the row. + drifted1 := seedObservation289(t, store, routeFlood188, mask289(mixedMask289), directFrame289, `["c3"]`, ts190) + setResolvedPath289(t, store, drifted1, rp289(c3b)) + // A FLOOD reception under a DIRECT transmission. Today the flood rule + // resolves to c3a, so an unguarded pass overwrites the stored c3b. + drifted2 := seedObservation289(t, store, routeDirect188, mask289(mixedMask289), floodFrame289, `["c3"]`, ts190+1) + setResolvedPath289(t, store, drifted2, rp289(c3b)) + primeIndexAndGraph188(t, store) + + runRepair289(t, store) + + for _, c := range []struct { + name string + id int64 + }{ + {"a DIRECT reception whose stored relay is not what the pre-#308 rule gives", drifted1}, + {"a FLOOD reception whose stored relay is not what the pre-#308 rule gives", drifted2}, + } { + rp := resolvedPathOf188(t, store, c.id) + if !rp.Valid || rp.String != rp289(c3b) { + t.Errorf("%s: resolved_path = %v, want the untouched %s", c.name, rp, rp289(c3b)) + } + } + st := repairState289(t, store) + if st.Repaired != 0 || st.Cleared != 0 { + t.Fatalf("state = %d repaired, %d cleared; want 0/0: neither row carries the value the pre-#308 rule produces", st.Repaired, st.Cleared) + } + if st.Candidates != 2 { + t.Fatalf("candidates = %d, want 2: both rows are read and then left alone", st.Candidates) + } + if st.Watermark < st.Ceiling { + t.Fatalf("watermark %d did not reach the ceiling %d", st.Watermark, st.Ceiling) + } +} + +// PR #590 review F3. The pass overwrites data in place and keeps no copy of +// the previous value, so it must not run on the first boot of any build that +// contains it: it is opt-in, and an absent or empty block leaves it off. +func TestResolvedPathRouteRepairEnabled_OptIn_289(t *testing.T) { + var nilCfg *Config + for _, c := range []struct { + name string + cfg *Config + want bool + }{ + {"nil config", nilCfg, false}, + {"no block", &Config{}, false}, + {"empty block", &Config{ResolvedPathRouteRepair: &ResolvedPathRouteRepairConfig{}}, false}, + {"enabled: false", &Config{ResolvedPathRouteRepair: &ResolvedPathRouteRepairConfig{Enabled: false}}, false}, + {"enabled: true", &Config{ResolvedPathRouteRepair: &ResolvedPathRouteRepairConfig{Enabled: true}}, true}, + } { + if got := c.cfg.ResolvedPathRouteRepairEnabled(); got != c.want { + t.Errorf("%s: ResolvedPathRouteRepairEnabled() = %v, want %v", c.name, got, c.want) + } + } +} + +// The switch reaches the scheduler: with the repair off, nothing is recorded +// in _async_migrations, no state table is written and the defect row keeps the +// value the pre-#308 backfill left. Turning it on repairs the same row. +func TestResolvedPathRouteRepair_NotScheduledWhenDisabled_289(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + id := seedObservation289(t, store, routeFlood188, mask289(mixedMask289), directFrame289, `["c3"]`, ts190) + setResolvedPath289(t, store, id, rp289(c3a)) + primeIndexAndGraph188(t, store) + + store.StartResolvedPathRouteRepair(context.Background(), false) + store.WaitForAsyncMigrations() + + if status, err := store.AsyncMigrationStatus(resolvedPathRouteRepairMigration); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("migration status = %q, err = %v; want no recorded run", status, err) + } + st, err := store.ResolvedPathRouteRepairState() + if err != nil { + t.Fatal(err) + } + if st != nil { + t.Fatalf("state row = %+v, want none: a disabled repair must not write", st) + } + if rp := resolvedPathOf188(t, store, id); !rp.Valid || rp.String != rp289(c3a) { + t.Fatalf("resolved_path = %v, want the untouched %s", rp, rp289(c3a)) + } + + store.StartResolvedPathRouteRepair(context.Background(), true) + store.WaitForAsyncMigrations() + if status, err := store.AsyncMigrationStatus(resolvedPathRouteRepairMigration); err != nil || status != "done" { + t.Fatalf("migration status = %q, %v; want done once the operator opts in", status, err) + } + if rp := resolvedPathOf188(t, store, id); rp.Valid { + t.Fatalf("resolved_path = %s, want NULL once the repair is enabled", rp.String) + } +} + +// PR #590 review F5. The loop's gate can pass and the snapshot stop being +// usable before the batch reads it (a graph rebuild publishes an empty graph +// between the two). That batch writes nothing and reports +// errResolvedPathBackfillNotReady; the loop must retry the same watermark +// rather than fail the migration. The batch's readiness check is a seam so the +// race does not have to be won for real. +func TestResolvedPathRouteRepair_RetriesABatchThatSeesAColdSnapshot_289(t *testing.T) { + store := backfillFixture188(t, filepath.Join(t.TempDir(), "ingest.db"), true) + defer store.Close() + id := seedObservation289(t, store, routeDirect188, mask289(mixedMask289), floodFrame289, `["c3"]`, ts190) + primeIndexAndGraph188(t, store) + + checks := 0 + old := resolvedPathRouteRepairBatchReady + resolvedPathRouteRepairBatchReady = func(s *Store, idx prefixIndex, graph *NeighborGraph) bool { + checks++ + if checks == 1 { + return false // the snapshot went cold after the loop's gate passed + } + return old(s, idx, graph) + } + t.Cleanup(func() { resolvedPathRouteRepairBatchReady = old }) + + runRepair289(t, store) + + if checks < 2 { + t.Fatalf("the batch ran %d time(s), want a retry after the not-ready batch", checks) + } + if rp := resolvedPathOf188(t, store, id); !rp.Valid || rp.String != rp289(c3a) { + t.Fatalf("resolved_path = %v, want %s from the retried batch", rp, rp289(c3a)) + } + st := repairState289(t, store) + if st.Watermark < st.Ceiling { + t.Fatalf("watermark %d did not reach the ceiling %d", st.Watermark, st.Ceiling) + } + if st.Repaired != 1 { + t.Fatalf("state = %d repaired, want 1 from the retried batch", st.Repaired) + } +} diff --git a/cmd/ingestor/resolved_path_route_repair_bench_289_test.go b/cmd/ingestor/resolved_path_route_repair_bench_289_test.go index 78b414225..a88b9bb99 100644 --- a/cmd/ingestor/resolved_path_route_repair_bench_289_test.go +++ b/cmd/ingestor/resolved_path_route_repair_bench_289_test.go @@ -49,7 +49,14 @@ var obsBench289 = []string{ // makes it a repair candidate. Every other observation carries a frame of the // transmission's own family, so the SQL prefilter drops it without the pass // reading the row. -func seedRepairBenchDB(b *testing.B, store *Store) (candidates int) { +// +// The mixed transmissions alternate between the two route families, so both +// repair directions are measured (PR #590 review F7): a candidate under a +// FLOOD transmission has a stored relay the pass clears, and one under a +// DIRECT transmission is a NULL row that gains a path. A cleared row writes +// a NULL and a repaired row writes a 66-byte JSON array, so the two are not +// the same amount of work inside the write transaction. +func seedRepairBenchDB(b *testing.B, store *Store) (cleared, repaired int) { b.Helper() if repairBenchObsPerTx > len(obsBench289) { b.Fatalf("repairBenchObsPerTx = %d, only %d observers are seeded", repairBenchObsPerTx, len(obsBench289)) @@ -96,24 +103,40 @@ func seedRepairBenchDB(b *testing.B, store *Store) (candidates int) { for i := 0; i < txCount; i++ { mask := sql.NullInt64{Int64: floodMask289, Valid: true} mixed := i%3 == 0 + // Half the mixed transmissions have a DIRECT first reception, so + // their candidates are the direction that gains a path. + direct := mixed && (i/3)%2 == 1 + txRoute, ownFrame, ownStored := routeFlood188, floodFrame289, rp289(c3a) if mixed { mask = sql.NullInt64{Int64: mixedMask289, Valid: true} } - res, err := insTx.Exec(fmt.Sprintf("bench289-%d", i), routeFlood188, mask) + if direct { + // A DIRECT transmission: its own receptions resolve to + // nothing, as a DIRECT ["c3"] has no observer anchor. + txRoute, ownFrame, ownStored = routeDirect188, directFrame289, "" + } + res, err := insTx.Exec(fmt.Sprintf("bench289-%d", i), txRoute, mask) if err != nil { return fmt.Errorf("transmission %d: %w", i, err) } txID, _ := res.LastInsertId() for j := 0; j < repairBenchObsPerTx; j++ { rows++ - frame := floodFrame289 + frame, stored := ownFrame, ownStored if mixed && rows%repairBenchCandidateEvery == 0 { - // A DIRECT reception the pre-#308 backfill anchored on the - // observer: the pass must clear the stored relay. - frame = directFrame289 - candidates++ + if direct { + // A FLOOD reception the pre-#308 backfill read as + // DIRECT: the row stayed NULL and gains c3a. + frame, stored = floodFrame289, "" + repaired++ + } else { + // A DIRECT reception the pre-#308 backfill anchored on + // the observer: the pass clears the stored relay. + frame, stored = directFrame289, rp289(c3a) + cleared++ + } } - if _, err := insObs.Exec(txID, obsIdx[j], ts190+int64(rows), frame, rp289(c3a)); err != nil { + if _, err := insObs.Exec(txID, obsIdx[j], ts190+int64(rows), frame, nilIfEmpty(stored)); err != nil { return fmt.Errorf("observation %d: %w", rows, err) } } @@ -123,9 +146,9 @@ func seedRepairBenchDB(b *testing.B, store *Store) (candidates int) { if err != nil { b.Fatal(err) } - b.Logf("seeded %d transmissions / %d observations (%d candidates) in %s", - txCount, rows, candidates, time.Since(start).Round(time.Second)) - return candidates + b.Logf("seeded %d transmissions / %d observations (%d candidates: %d to clear, %d to repair) in %s", + txCount, rows, cleared+repaired, cleared, repaired, time.Since(start).Round(time.Second)) + return cleared, repaired } // BenchmarkResolvedPathRouteRepairPass_289 runs the whole pass once over the @@ -137,7 +160,7 @@ func BenchmarkResolvedPathRouteRepairPass_289(b *testing.B) { dbPath := filepath.Join(b.TempDir(), "ingest.db") store := backfillFixture188(b, dbPath, true) defer store.Close() - wantCandidates := seedRepairBenchDB(b, store) + wantCleared, wantRepaired := seedRepairBenchDB(b, store) primeIndexAndGraph188(b, store) store.Checkpoint() if fi, err := os.Stat(dbPath); err == nil { @@ -161,13 +184,17 @@ func BenchmarkResolvedPathRouteRepairPass_289(b *testing.B) { if err != nil || st == nil { b.Fatalf("state: %v", err) } - if int(st.Cleared) != wantCandidates { - b.Fatalf("cleared %d rows, want the %d seeded candidates", st.Cleared, wantCandidates) + if wantCleared == 0 || wantRepaired == 0 { + b.Fatalf("the mix must measure both directions: %d to clear, %d to repair", wantCleared, wantRepaired) + } + if int(st.Cleared) != wantCleared || int(st.Repaired) != wantRepaired { + b.Fatalf("%d cleared / %d repaired, want the seeded %d / %d", + st.Cleared, st.Repaired, wantCleared, wantRepaired) } w := writerStatsAgg.get(resolvedPathRouteRepairComponent).snapshot() perRow := float64(elapsed.Nanoseconds()) / float64(st.Ceiling) - b.Logf("pass over %d observations: %s wall (%.0f ns/observation), %d candidates, %d cleared, %d batches", - st.Ceiling, elapsed.Round(time.Millisecond), perRow, st.Candidates, st.Cleared, w.Count-before.Count) + b.Logf("pass over %d observations: %s wall (%.0f ns/observation), %d candidates, %d cleared, %d repaired, %d batches", + st.Ceiling, elapsed.Round(time.Millisecond), perRow, st.Candidates, st.Cleared, st.Repaired, w.Count-before.Count) b.Logf("writer hold ms: p50 %.1f p95 %.1f p99 %.1f max %.1f over %d transactions (window %d, batch %d)", w.HoldMsP50, w.HoldMsP95, w.HoldMsP99, w.HoldMsMax, w.Count-before.Count, resolvedPathRouteRepairIDWindow, resolvedPathRouteRepairBatchSize) diff --git a/config.example.json b/config.example.json index a2f186782..eac3379b5 100644 --- a/config.example.json +++ b/config.example.json @@ -38,6 +38,11 @@ "pauseMs": 250, "_comment": "Optional (ingestor, #188). Once per ingestor start, re-resolves observations stored with resolved_path = NULL, from a persisted watermark up to the newest observation at the start of the pass. It waits until a neighbour-edge build has caught up with the stored observations and found at least one edge, then works in batches: each batch reads batchSize rows, resolves them outside any lock, and writes the results and the new watermark in one short write transaction, then sleeps pauseMs. Rows that still do not resolve stay NULL and are not retried. disabled: true skips the pass. batchSize: rows per batch, default 500. pauseMs: milliseconds between batches, default 250, which caps the pass at about 2,000 rows/s. For both numbers, 0 or omitted means the default, so the pause cannot be turned off; the smallest pause is 1. The server sees backfilled values only after its next restart." }, + "resolvedPathRouteRepair": { + "enabled": false, + "_comment": "Opt-in, OFF by default (ingestor, #289 M2). One-off repair of observations.resolved_path rows the pre-#308 batch readers wrote with the transmission's FIRST route type instead of each reception's own: a DIRECT reception under a FLOOD transmission was anchored on the observer and stored a relay the observer never heard the packet from, and a FLOOD reception under a DIRECT transmission got no anchor and stayed NULL. The pass walks observations.id from 0 to the newest row at its start, in bounded windows, resolving outside any lock and writing each window with the new watermark in one short transaction; it runs once per database (recorded in _async_migrations) and resumes from its watermark after a crash or restart. It rewrites a row only when the route type changes the answer AND the stored value is exactly what the pre-#308 rule produces, so rows live ingest wrote are left alone. TAKE A DATABASE SNAPSHOT FIRST: the rewrite is in place, the previous value is kept nowhere, and a row cleared back to NULL sits below the #188 backfill's watermark, which never revisits it — so a clear is permanent without that snapshot. Measure before enabling: scripts/resolved-path-route-repair-estimate.sh prints the candidate counts and the estimated cost of one pass, read-only, on a snapshot copy. Progress and the repaired/cleared counts are in the resolved_path_route_repair_state table. The server sees the rewritten values only after its next restart.", + "_comment_rollback": "Setting this back to false stops any further repair but does not undo one that ran. A recorded migration is never re-run; re-arming after a rollback to a pre-#308 ingestor means clearing the _async_migrations row by hand." + }, "decodeBackfill": { "disabled": false, "batchSize": 300, diff --git a/docs/user-guide/configuration.md b/docs/user-guide/configuration.md index 62413da5f..c02a48cfb 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -161,6 +161,26 @@ Once per ingestor start, observations stored with `resolved_path = NULL` are res | `resolvedPathBackfill.batchSize` | `500` | Rows per batch; `0` means the default | | `resolvedPathBackfill.pauseMs` | `250` | Milliseconds between batches; `0` means the default, so the pause cannot be turned off (minimum `1`) | +### Resolved-path route-type repair (ingestor, opt-in) + +A one-off repair of `observations.resolved_path` rows that the batch readers wrote before [#308](https://github.com/Kpa-clawbot/CoreScope/issues/308) using the **transmission's first** route type instead of each reception's own. DIRECT and FLOOD receptions of the same payload share one `transmissions` row, so a reception carrying the other route family was resolved with the wrong rule: a DIRECT reception under a FLOOD transmission was anchored on the observer and stored a relay the observer never heard the packet from, and a FLOOD reception under a DIRECT transmission got no anchor and stayed `NULL`. + +| Field | Default | Description | +|-------|---------|-------------| +| `resolvedPathRouteRepair.enabled` | `false` | Run the one-off repair. Off unless set: the pass rewrites values in place. | + +> **Take a database snapshot before enabling this.** The rewrite is in place and +> the previous value is kept nowhere. A row the pass clears back to `NULL` sits +> below the resolved-path backfill's watermark, which never revisits a row it has +> passed, so **a clear is permanent** without that snapshot. Setting `enabled` +> back to `false` stops any further repair but undoes nothing. + +**Measure first.** `scripts/resolved-path-route-repair-estimate.sh /path/to/snapshot.db` prints the candidate counts, split by bug direction, and the estimated cost of one pass. It is read-only (the database is opened through an `immutable=1` URI) but should still be pointed at a snapshot copy rather than a live file. + +The pass walks `observations.id` from 0 to the newest row at its start — which is how it reaches `NULL` rows below the backfill's watermark — in bounded windows, resolving outside any lock and committing each window's rewrites together with the new watermark. It runs once per database, recorded in `_async_migrations`, and resumes from its watermark after a crash or a restart. A row is rewritten only when the route type changes the resolved answer **and** the stored value is exactly what the pre-#308 rule produces, so values written by live ingest (which has always used the frame's own route type) are read and left alone. Progress and the repaired/cleared counts are in the `resolved_path_route_repair_state` table. The server sees the rewritten values after its next restart. + +A recorded migration is never re-run. An ingestor rolled back to a pre-#308 build writes new wrong rows that this pass will not revisit; re-arming means clearing its `_async_migrations` row by hand. + ## Channel decryption | Field | Description | diff --git a/scripts/resolved-path-route-repair-estimate.sh b/scripts/resolved-path-route-repair-estimate.sh index b75d6a184..ac6817d01 100755 --- a/scripts/resolved-path-route-repair-estimate.sh +++ b/scripts/resolved-path-route-repair-estimate.sh @@ -16,7 +16,9 @@ # numbers. Two further approximations, on any input: # - the candidate count is the SQL prefilter the pass uses, which is a # superset: the pass re-checks each row in Go with the #289/#308 helper and -# writes only the rows whose route type actually changes the resolve; +# writes only a row whose route type actually changes the resolve AND whose +# stored value is exactly what the pre-#308 rule produces, so the repaired +# count is normally well below the candidate count printed here; # - the scan rate is the BenchmarkResolvedPathRouteRepairPass_289 figure # below, measured warm-cache on a synthetic database. It is not a # measurement of this host or of cold disk I/O. @@ -74,23 +76,55 @@ else fi echo -# Candidates, split by bug direction. Both use the pass's own prefilter: the -# route family is bit 1 of the low nibble of the frame's header byte, i.e. of -# raw_hex's second character, and (route_type & 2) is the transmission's. +# Candidates, split by bug direction: which route family the frame's own header +# carries. That is bit 1 of the low nibble of the header byte, i.e. of +# raw_hex's second character, and (route_type & 2) is the transmission's — the +# pass's own prefilter. Bit 1 alone separates the families, because the +# transport types (0 and 3) share bit 0 with the other family's non-transport +# type. +# +# Each direction is then split by whether the row carries a stored path, which +# is a different question from the direction and is reported inside it: a +# DIRECT-frame row can be NULL (the resolve found nothing) and a FLOOD-frame +# row can carry a stored value (PR #590 review F6). A stored value is not by +# itself a defect either: live ingest has always used the frame's own route +# type, so a mismatched row may well hold a correct ingest value, and the pass +# rewrites a row only when its stored value is exactly what the pre-#308 rule +# produces (PR #590 review F1). One query, so this is a single scan. HDR_CAND=0 -HDR_NULL=0 +HDR_DIR=0 +HDR_DIR_NULL=0 +HDR_FLOOD=0 +HDR_FLOOD_NULL=0 +HDR_BAD=0 MASK_CAND=0 if [ "$RAW_COL" = "1" ]; then - HDR="length(o.raw_hex) >= 2 - AND ((instr('0123456789abcdef', lower(substr(o.raw_hex, 2, 1))) - 1) & 2) != (COALESCE(t.route_type, -1) & 2) - AND COALESCE(o.path_json, '') NOT IN ('', '[]')" - HDR_CAND=$(q "SELECT COUNT(*) FROM observations o JOIN transmissions t ON t.id = o.transmission_id WHERE $HDR;") - HDR_NULL=$(q "SELECT COUNT(*) FROM observations o JOIN transmissions t ON t.id = o.transmission_id WHERE $HDR AND o.resolved_path IS NULL;") + SPLIT=$(q "SELECT COUNT(*), + COALESCE(SUM(CASE WHEN hex_ok > 0 AND hdr_fam != 0 THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN hex_ok > 0 AND hdr_fam != 0 AND rp IS NULL THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN hex_ok > 0 AND hdr_fam = 0 THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN hex_ok > 0 AND hdr_fam = 0 AND rp IS NULL THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN hex_ok = 0 THEN 1 ELSE 0 END), 0) + FROM ( + SELECT instr('0123456789abcdef', lower(substr(o.raw_hex, 2, 1))) AS hex_ok, + ((instr('0123456789abcdef', lower(substr(o.raw_hex, 2, 1))) - 1) & 2) AS hdr_fam, + o.resolved_path AS rp + FROM observations o JOIN transmissions t ON t.id = o.transmission_id + WHERE length(o.raw_hex) >= 2 + AND ((instr('0123456789abcdef', lower(substr(o.raw_hex, 2, 1))) - 1) & 2) != (COALESCE(t.route_type, -1) & 2) + AND COALESCE(o.path_json, '') NOT IN ('', '[]') + );") + IFS='|' read -r HDR_CAND HDR_DIR HDR_DIR_NULL HDR_FLOOD HDR_FLOOD_NULL HDR_BAD <