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/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 af2a86d4f..ac1c25937 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -532,6 +532,16 @@ 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. 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, cfg.ResolvedPathRouteRepairEnabled()) + // #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/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 new file mode 100644 index 000000000..bb7c1fd06 --- /dev/null +++ b/cmd/ingestor/resolved_path_route_repair.go @@ -0,0 +1,520 @@ +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 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 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 +// 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" +) + +// 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. +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() + + // 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 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) + } +} + +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 !resolvedPathRouteRepairBatchReady(s, 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 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 != had { + 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..305869c8e --- /dev/null +++ b/cmd/ingestor/resolved_path_route_repair_289_test.go @@ -0,0 +1,601 @@ +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 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 + // 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: ""}, + + // 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)}, + {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)}, + + // 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)}, + } +} + +// 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 != 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 != 9 { + t.Fatalf("candidates = %d, want 9 (five defects, two drifted rows and two route-type-independent rows)", 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(), true) + 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(), 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 new file mode 100644 index 000000000..a88b9bb99 --- /dev/null +++ b/cmd/ingestor/resolved_path_route_repair_bench_289_test.go @@ -0,0 +1,230 @@ +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. +// +// 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)) + } + 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 + // 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} + } + 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, stored := ownFrame, ownStored + if mixed && rows%repairBenchCandidateEvery == 0 { + 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, nilIfEmpty(stored)); 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: %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 +// 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() + wantCleared, wantRepaired := 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 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 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) + 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/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 new file mode 100755 index 000000000..ac6817d01 --- /dev/null +++ b/scripts/resolved-path-route-repair-estimate.sh @@ -0,0 +1,181 @@ +#!/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 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. +# 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: 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_DIR=0 +HDR_DIR_NULL=0 +HDR_FLOOD=0 +HDR_FLOOD_NULL=0 +HDR_BAD=0 +MASK_CAND=0 +if [ "$RAW_COL" = "1" ]; then + 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 < 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..fb0a7ad80 --- /dev/null +++ b/scripts/test-resolved-path-route-repair-estimate.sh @@ -0,0 +1,148 @@ +#!/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), + -- The split is by header family, not by NULL/stored (PR #590 review F6): + -- a DIRECT-frame candidate that is NULL, and a FLOOD-frame candidate that + -- carries a stored path. + (1, 6, '["c3"]', 11, '1601c3', NULL), + (2, 4, '["c3"]', 12, '1501c3', '["c355"]'), + -- The transport route types, 3 (17) and 0 (14): they differ from the other + -- family's non-transport type in bit 1 only (PR #590 review F4). + (1, 7, '["c3"]', 13, '1701c3', '["c355"]'), + (2, 5, '["c3"]', 14, '1401c3', NULL), + -- An unparseable header: reported on its own, because the pass re-checks + -- it in Go rather than trusting the prefilter's family bit. + (1, 8, '["c3"]', 15, 'zz01c3', 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: 15$' "15 observations counted" +expect '^ max id \(scan end\): 15$' "the scan end is MAX(observations.id)" +expect '^ with a frame: 11$' "11 rows carry a frame" +expect '^ rows: 8$' "8 header candidates counted (both directions)" +# The split is by header family first. Each bucket mixes NULL and stored rows, +# so a NULL/non-NULL split could not produce these four numbers. +expect '^ DIRECT frame under a FLOOD transmission' "the first direction is labelled by header family" +expect '^ rows: 4 \(stored 3, NULL 1\)$' "4 DIRECT-frame candidates: 3 stored, 1 NULL" +expect '^ FLOOD frame under a DIRECT transmission' "the second direction is labelled by header family" +expect '^ rows: 3 \(stored 1, NULL 2\)$' "3 FLOOD-frame candidates: 1 stored, 2 NULL" +expect '^ unparseable header: 1 ' "the unparseable header is reported on its own" +expect '^ rows: 1$' "1 route_mask candidate counted" +expect 'recorded run: none' "no recorded run on a fresh database" +expect ' candidate rows: 9$' "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"