From a558c8c6d45ed56740226b671a9f2f794e33e994 Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 6 Oct 2026 04:31:49 +0200 Subject: [PATCH 1/4] fix(ingestor): keep route_mask NULL when merging an uncomputed side (#287) A NULL transmissions.route_mask means "not computed yet", not "no routes". mergeTransmissions OR-ed COALESCE(route_mask, 0) from each side, so when only one side had a computed mask the result was non-NULL. The route_mask backfill only recomputes NULL rows, so it then skipped the merged row and the uncomputed side's route bits were lost permanently. Keep the merged route_mask NULL whenever either side is NULL, so the backfill recomputes it from the survivor's full (merged) set of observations. In the non-NULL branch both sides are known, so a plain OR is exact. The other merged columns were reviewed: first_seen/last_seen are NOT NULL (MIN/MAX), and the fill columns use COALESCE(survivor, loser), which fills an unknown survivor value from a known loser value and never fabricates a sentinel that would block a backfill. Only route_mask had the pattern. Co-Authored-By: Claude Opus 4.8 --- cmd/ingestor/hash_migrate.go | 16 ++- .../hash_migrate_route_mask_287_test.go | 109 ++++++++++++++++++ 2 files changed, 122 insertions(+), 3 deletions(-) create mode 100644 cmd/ingestor/hash_migrate_route_mask_287_test.go diff --git a/cmd/ingestor/hash_migrate.go b/cmd/ingestor/hash_migrate.go index 61422dcd9..5e224b646 100644 --- a/cmd/ingestor/hash_migrate.go +++ b/cmd/ingestor/hash_migrate.go @@ -39,7 +39,10 @@ import ( // earliest first_seen; // - the survivor takes the earliest first_seen, the later last_seen and the // union of route_mask (a grown mask is logged in route_mask_changes for -// running servers), and every nullable column it has no value for (scope_name, +// running servers), unless either side is NULL: a NULL route_mask is "not +// computed yet", so the union stays NULL and the route_mask backfill +// recomputes it from the merged observations (#287); and every nullable +// column it has no value for (scope_name, // channel_hash, from_pubkey, ...) from the duplicate; a value it has stays; // - rows hung off the duplicate follow the survivor (ping_triggers), or go // with it when the survivor already has one (route_mask_changes); @@ -314,8 +317,15 @@ func mergeTransmissions(ctx context.Context, tx *sql.Tx, winner, loser int64, ex } set := "first_seen = MIN(first_seen, (SELECT first_seen FROM transmissions WHERE id = ?)),\n" + "last_seen = MAX(last_seen, (SELECT last_seen FROM transmissions WHERE id = ?)),\n" + - "route_mask = CASE WHEN route_mask IS NULL AND (SELECT route_mask FROM transmissions WHERE id = ?) IS NULL THEN NULL\n" + - "ELSE COALESCE(route_mask, 0) | COALESCE((SELECT route_mask FROM transmissions WHERE id = ?), 0) END" + // A NULL route_mask means "not computed yet", not "no routes". If either + // side is NULL the union is unknown, so keep it NULL: a non-NULL result + // would be skipped by the route_mask backfill, losing the uncomputed + // side's bits for good (#287). The backfill recomputes the mask from the + // survivor's full set of observations, which now holds both sides'. Both + // sides are non-NULL in the ELSE branch, so a plain OR is exact; a + // COALESCE(...,0) there would reintroduce the bug. + "route_mask = CASE WHEN route_mask IS NULL OR (SELECT route_mask FROM transmissions WHERE id = ?) IS NULL THEN NULL\n" + + "ELSE route_mask | (SELECT route_mask FROM transmissions WHERE id = ?) END" args := []interface{}{loser, loser, loser, loser} for _, col := range ex.fillCols { set += fmt.Sprintf(",\n\"%s\" = COALESCE(\"%s\", (SELECT \"%s\" FROM transmissions WHERE id = ?))", col, col, col) diff --git a/cmd/ingestor/hash_migrate_route_mask_287_test.go b/cmd/ingestor/hash_migrate_route_mask_287_test.go new file mode 100644 index 000000000..c6ea2d8f0 --- /dev/null +++ b/cmd/ingestor/hash_migrate_route_mask_287_test.go @@ -0,0 +1,109 @@ +package main + +import ( + "context" + "database/sql" + "path/filepath" + "testing" +) + +// Issue #287: a NULL route_mask means "not computed yet", not "no routes". When +// a merge folds a loser into a survivor and only one side has a computed mask, +// COALESCE(route_mask, 0) injected 0 for the uncomputed side and OR-ed it in, so +// the result was non-NULL. The route_mask backfill only recomputes NULL rows, so +// it then skipped the row and the uncomputed side's route bits were lost for +// good. The fix keeps the merged mask NULL whenever either side is NULL, so the +// backfill recomputes it from the survivor's full (merged) set of observations. + +// hm287InsObs writes one observation with a raw_hex header for the backfill to +// read. Distinct observer_idx keeps the survivor's and loser's copies from +// colliding on idx_observations_dedup when the loser's is re-parented. +func hm287InsObs(t *testing.T, db *sql.DB, txID, observer int, rawHex string) { + t.Helper() + hm215Exec(t, db, `INSERT INTO observations (transmission_id, observer_idx, direction, path_json, timestamp, raw_hex) + VALUES (?, ?, 'RX', ?, 1767225600, ?)`, txID, observer, `["aa"]`, rawHex) +} + +// A survivor with a computed DIRECT mask and a loser that is still NULL but was +// heard as FLOOD: after merge + backfill the mask must be DIRECT|FLOOD, not the +// survivor's DIRECT alone. Kills the mutant that restores COALESCE(route_mask,0) +// (which leaves a non-NULL DIRECT that the backfill skips). +func TestContentHashMigration_MergeKeepsRouteMaskNullWhenLoserUnknown_287(t *testing.T) { + // A DIRECT frame header (low two bits = 2) and a FLOOD frame header + // (firstIngestedFloodRaw, low two bits = 1). + const directFrame = "12" + "00000000" + const directBit = int64(1) << 2 // route type 2 (DIRECT) + const floodBit = int64(1) << 1 // route type 1 (FLOOD) + + s := hm215Reopen(t, filepath.Join(t.TempDir(), "rm287.db"), func(db *sql.DB) { + for _, o := range []string{"o1", "o2"} { + hm215Exec(t, db, `INSERT OR IGNORE INTO observers (id, name, first_seen, last_seen) VALUES (?, ?, '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')`, o, o) + } + o1 := hm215Count(t, db, `SELECT rowid FROM observers WHERE id = 'o1'`) + o2 := hm215Count(t, db, `SELECT rowid FROM observers WHERE id = 'o2'`) + + // Survivor (lowest id): DIRECT, already computed (route_mask = directBit). + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (70, ?, 'stale-rm-70', '2026-01-01T00:00:00Z', 1, 2, 4, '{}', ?)`, hm215Raw(7), directBit) + // Loser: route_mask NULL (not computed), but heard as FLOOD. + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (71, ?, 'stale-rm-71', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', NULL)`, hm215Raw(7)) + hm287InsObs(t, db, 70, o1, directFrame) + hm287InsObs(t, db, 71, o2, firstIngestedFloodRaw) + }) + + // After the merge the mask must be NULL so the backfill owns it, and no + // route_mask_changes row may be logged for a mask that is not yet known. + var merged sql.NullInt64 + if err := s.db.QueryRow(`SELECT route_mask FROM transmissions WHERE id = 70`).Scan(&merged); err != nil { + t.Fatalf("survivor row: %v", err) + } + if merged.Valid { + t.Fatalf("merged route_mask = %d, want NULL so the backfill recomputes it", merged.Int64) + } + if n := hm215Count(t, s.db, `SELECT COUNT(*) FROM route_mask_changes WHERE transmission_id = 70`); n != 0 { + t.Fatalf("route_mask_changes logged %d rows for a not-yet-known mask, want 0", n) + } + if n := hm215Count(t, s.db, `SELECT COUNT(*) FROM transmissions WHERE id = 71`); n != 0 { + t.Fatalf("loser 71 was not deleted") + } + + // The backfill recomputes from route_type and every surviving observation + // (both sides', now re-parented to the survivor): DIRECT|FLOOD. + if err := s.backfillTxRouteMask(context.Background(), s.db); err != nil { + t.Fatalf("backfill: %v", err) + } + var final sql.NullInt64 + if err := s.db.QueryRow(`SELECT route_mask FROM transmissions WHERE id = 70`).Scan(&final); err != nil { + t.Fatalf("survivor row after backfill: %v", err) + } + if want := directBit | floodBit; !final.Valid || final.Int64 != want { + t.Fatalf("route_mask after backfill = %v, want DIRECT|FLOOD = %04b", final, want) + } +} + +// Requirement 2: the other columns a merge folds use COALESCE(survivor, loser), +// which fills an unknown (NULL) survivor value from a known loser value and +// never fabricates a sentinel. When BOTH sides are NULL the column must stay +// NULL, so a column whose NULL means "pending" keeps that meaning after a merge. +// Kills a mutant that coalesces a fill column to a literal (e.g. +// COALESCE(scope_name, '')), which would pass "pending" off as a known value. +func TestContentHashMigration_MergeKeepsFillColumnsNullWhenBothUnknown_287(t *testing.T) { + s := hm215Reopen(t, filepath.Join(t.TempDir(), "fill287.db"), func(db *sql.DB) { + hm215Exec(t, db, `INSERT OR IGNORE INTO observers (id, name, first_seen, last_seen) VALUES ('o1', 'o1', '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')`) + o1 := hm215Count(t, db, `SELECT rowid FROM observers WHERE id = 'o1'`) + // Both sides leave scope_name, channel_hash and from_pubkey NULL. + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask, scope_name, channel_hash, from_pubkey) + VALUES (80, ?, 'stale-fill-80', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', 1, NULL, NULL, NULL)`, hm215Raw(8)) + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask, scope_name, channel_hash, from_pubkey) + VALUES (81, ?, 'stale-fill-81', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', 1, NULL, NULL, NULL)`, hm215Raw(8)) + hm287InsObs(t, db, 80, o1, "11"+"00000000") + }) + var scope, channel, from sql.NullString + if err := s.db.QueryRow(`SELECT scope_name, channel_hash, from_pubkey FROM transmissions WHERE id = 80`).Scan(&scope, &channel, &from); err != nil { + t.Fatalf("survivor row: %v", err) + } + if scope.Valid || channel.Valid || from.Valid { + t.Fatalf("a both-NULL fill column was given a value: scope=%v channel=%v from=%v; want all NULL", scope, channel, from) + } +} From 635297512e9db616e6874f730aa9449af82e2951 Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 6 Oct 2026 08:23:49 +0200 Subject: [PATCH 2/4] fix(ingestor): recompute merged route_mask inline when a side is uncomputed (#287) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #295 review (CS-pve-agent3, REQUEST CHANGES). The first fix kept the merged route_mask NULL whenever either side was NULL and let the #89 backfill rebuild it. That still loses a route bit: the backfill rebuilds the mask only from the survivor's route_type and the observations that survive the merge, so a bit the computed side stored from a frame that no surviving observation can rebuild (an observation idx_observations_dedup drops during the move, or a frame a later reception overwrote) is thrown away for good — the same loss as #287, in the other direction. mergeTransmissions now recomputes the mask inline, in the merge transaction, whenever either side is NULL: it ORs both sides' stored masks (keeping every bit either side recorded) with the #89 lower bound the backfill would produce (both sides' route_type plus the surviving observation headers). The result is non-NULL, so the backfill leaves the row alone and the route_mask_changes row is logged as for any grown mask, which also tells a running server (finding 3). When both sides are already computed the merge still takes the exact OR. The per-observation header parse the backfill and the merge share is extracted into routeMaskBitFromHeader (finding 4). The file header is corrected to describe the inline recompute rather than a backfill that "holds both sides'" observations, which is not true for a dedup-dropped frame. Tests (red before the fix): the loser-NULL and the previously untested survivor-NULL directions both recompute the union inline and log the change (findings 2, 3); a dedup-dropped frame's stored bit is kept across merge and backfill (finding 1). The both-NULL fill-column guard is unchanged. Co-Authored-By: Claude Opus 4.8 --- cmd/ingestor/hash_migrate.go | 111 ++++++++--- .../hash_migrate_route_mask_287_test.go | 176 ++++++++++++++---- cmd/ingestor/route_mask_backfill.go | 15 +- 3 files changed, 235 insertions(+), 67 deletions(-) diff --git a/cmd/ingestor/hash_migrate.go b/cmd/ingestor/hash_migrate.go index 5e224b646..652af6ebe 100644 --- a/cmd/ingestor/hash_migrate.go +++ b/cmd/ingestor/hash_migrate.go @@ -6,6 +6,8 @@ import ( "fmt" "log" "time" + + "github.com/meshcore-analyzer/packetpath" ) // Content-hash migration (#215). @@ -38,12 +40,16 @@ import ( // observer, same path), while the transmission keeps the // earliest first_seen; // - the survivor takes the earliest first_seen, the later last_seen and the -// union of route_mask (a grown mask is logged in route_mask_changes for -// running servers), unless either side is NULL: a NULL route_mask is "not -// computed yet", so the union stays NULL and the route_mask backfill -// recomputes it from the merged observations (#287); and every nullable -// column it has no value for (scope_name, -// channel_hash, from_pubkey, ...) from the duplicate; a value it has stays; +// route_mask union (a grown mask is logged in route_mask_changes for running +// servers). A NULL route_mask is "not computed yet", not "no routes": when +// either side is NULL the merge cannot take a plain OR, so it recomputes the +// mask inline (#287) from both sides' stored masks (keeping a bit no +// surviving observation can rebuild) plus the #89 lower bound the backfill +// would produce (route_type and the surviving observation headers). The +// result is non-NULL, so the backfill leaves it alone and the change is +// logged like any grown mask; and every nullable column it has no value for +// (scope_name, channel_hash, from_pubkey, ...) from the duplicate; a value +// it has stays; // - rows hung off the duplicate follow the survivor (ping_triggers), or go // with it when the survivor already has one (route_mask_changes); // - the duplicate row is deleted. @@ -311,22 +317,47 @@ func mergeTransmissions(ctx context.Context, tx *sql.Tx, winner, loser int64, ex return err } - var oldMask sql.NullInt64 + var oldMask, loserMask sql.NullInt64 if err := tx.QueryRowContext(ctx, `SELECT route_mask FROM transmissions WHERE id = ?`, winner).Scan(&oldMask); err != nil { return err } + if err := tx.QueryRowContext(ctx, `SELECT route_mask FROM transmissions WHERE id = ?`, loser).Scan(&loserMask); err != nil { + return err + } + // A NULL route_mask means "not computed yet", not "no routes". + var mergedMask int64 + if oldMask.Valid && loserMask.Valid { + // Both computed: each stored mask already encodes that side's route_type + // and every observation it ever held, so their OR is the exact union. + mergedMask = oldMask.Int64 | loserMask.Int64 + } else { + // At least one side was never computed. Leaving the result non-NULL via + // COALESCE(route_mask, 0) would make the backfill skip the row and lose + // the uncomputed side's bits for good (#287). Setting it back to NULL and + // deferring to the backfill instead would lose a bit the computed side + // stored but no surviving observation can rebuild (an observation + // idx_observations_dedup dropped in the move above, or a frame a later + // reception overwrote). So recompute inline, in this transaction: keep + // every bit either side stored, then OR the #89 lower bound the backfill + // would produce (both sides' route_type plus the surviving observation + // headers). The result is non-NULL, so the backfill leaves the row alone + // and the route_mask_changes row below tells running servers. + m, err := recomputeMergedRouteMask(ctx, tx, winner, loser) + if err != nil { + return err + } + if oldMask.Valid { + m |= oldMask.Int64 + } + if loserMask.Valid { + m |= loserMask.Int64 + } + mergedMask = m + } set := "first_seen = MIN(first_seen, (SELECT first_seen FROM transmissions WHERE id = ?)),\n" + "last_seen = MAX(last_seen, (SELECT last_seen FROM transmissions WHERE id = ?)),\n" + - // A NULL route_mask means "not computed yet", not "no routes". If either - // side is NULL the union is unknown, so keep it NULL: a non-NULL result - // would be skipped by the route_mask backfill, losing the uncomputed - // side's bits for good (#287). The backfill recomputes the mask from the - // survivor's full set of observations, which now holds both sides'. Both - // sides are non-NULL in the ELSE branch, so a plain OR is exact; a - // COALESCE(...,0) there would reintroduce the bug. - "route_mask = CASE WHEN route_mask IS NULL OR (SELECT route_mask FROM transmissions WHERE id = ?) IS NULL THEN NULL\n" + - "ELSE route_mask | (SELECT route_mask FROM transmissions WHERE id = ?) END" - args := []interface{}{loser, loser, loser, loser} + "route_mask = ?" + args := []interface{}{loser, loser, mergedMask} for _, col := range ex.fillCols { set += fmt.Sprintf(",\n\"%s\" = COALESCE(\"%s\", (SELECT \"%s\" FROM transmissions WHERE id = ?))", col, col, col) args = append(args, loser) @@ -350,15 +381,13 @@ func mergeTransmissions(ctx context.Context, tx *sql.Tx, winner, loser int64, ex if _, err := tx.ExecContext(ctx, `DELETE FROM route_mask_changes WHERE transmission_id = ?`, loser); err != nil { return err } - var newMask sql.NullInt64 - if err := tx.QueryRowContext(ctx, `SELECT route_mask FROM transmissions WHERE id = ?`, winner).Scan(&newMask); err != nil { - return err - } - if newMask.Valid && (!oldMask.Valid || newMask.Int64 != oldMask.Int64) { + if !oldMask.Valid || mergedMask != oldMask.Int64 { // Same record InsertTransmission leaves when an observation adds a // route bit: servers that hold the survivor pick up the full mask. + // The merged mask is always known now, so an uncomputed survivor + // (oldMask NULL) that the merge gives a mask is announced too. if _, err := tx.ExecContext(ctx, `INSERT INTO route_mask_changes (transmission_id, route_mask, created_at) VALUES (?, ?, ?)`, - winner, newMask.Int64, time.Now().Unix()); err != nil { + winner, mergedMask, time.Now().Unix()); err != nil { return err } } @@ -366,3 +395,39 @@ func mergeTransmissions(ctx context.Context, tx *sql.Tx, winner, loser int64, ex _, err := tx.ExecContext(ctx, `DELETE FROM transmissions WHERE id = ?`, loser) return err } + +// recomputeMergedRouteMask rebuilds the #89 route-mask lower bound for a merged +// transmission whose mask cannot be a plain OR because a side was never +// computed. It mirrors backfillTxRouteMaskBatch for the one merged row: the bit +// of each side's stored route_type (the survivor keeps its own route_type or +// takes the loser's, and a route the survivor was never heard on still matters, +// so both are ORed) plus the header bit of every observation now parented on the +// winner (the move above already folded in the loser's surviving observations). +// The caller ORs in both stored masks, so a bit either side recorded from a +// frame that no surviving observation can rebuild is kept too. Unlike the +// backfill it runs once per merge, so a per-row read is fine. +func recomputeMergedRouteMask(ctx context.Context, tx *sql.Tx, winner, loser int64) (int64, error) { + var mask int64 + for _, id := range [...]int64{winner, loser} { + var routeType sql.NullInt64 + if err := tx.QueryRowContext(ctx, `SELECT route_type FROM transmissions WHERE id = ?`, id).Scan(&routeType); err != nil { + return 0, err + } + if routeType.Valid { + mask |= packetpath.RouteMaskBit(int(routeType.Int64)) + } + } + rows, err := tx.QueryContext(ctx, `SELECT substr(raw_hex, 1, 2) FROM observations WHERE transmission_id = ? AND raw_hex IS NOT NULL`, winner) + if err != nil { + return 0, err + } + defer rows.Close() + for rows.Next() { + var header sql.NullString + if err := rows.Scan(&header); err != nil { + return 0, err + } + mask |= routeMaskBitFromHeader(header.String) + } + return mask, rows.Err() +} diff --git a/cmd/ingestor/hash_migrate_route_mask_287_test.go b/cmd/ingestor/hash_migrate_route_mask_287_test.go index c6ea2d8f0..021837bbe 100644 --- a/cmd/ingestor/hash_migrate_route_mask_287_test.go +++ b/cmd/ingestor/hash_migrate_route_mask_287_test.go @@ -9,67 +9,77 @@ import ( // Issue #287: a NULL route_mask means "not computed yet", not "no routes". When // a merge folds a loser into a survivor and only one side has a computed mask, -// COALESCE(route_mask, 0) injected 0 for the uncomputed side and OR-ed it in, so -// the result was non-NULL. The route_mask backfill only recomputes NULL rows, so -// it then skipped the row and the uncomputed side's route bits were lost for -// good. The fix keeps the merged mask NULL whenever either side is NULL, so the -// backfill recomputes it from the survivor's full (merged) set of observations. +// the result must keep every bit either side stored (a bit a surviving +// observation can no longer rebuild, because idx_observations_dedup dropped its +// frame during the move, must not be lost) and OR in the bits the #89 backfill +// would recompute from the merged observations. The merge does this inline, in +// the same transaction, so the result is non-NULL: the backfill then leaves the +// row alone and the route_mask_changes row tells running servers. -// hm287InsObs writes one observation with a raw_hex header for the backfill to -// read. Distinct observer_idx keeps the survivor's and loser's copies from -// colliding on idx_observations_dedup when the loser's is re-parented. -func hm287InsObs(t *testing.T, db *sql.DB, txID, observer int, rawHex string) { +// A DIRECT frame header (low two bits = 2) and a FLOOD frame header (low two +// bits = 1). +const ( + hm287DirectFrame = "12" + "00000000" + hm287FloodFrame = "11" + "00000000" + hm287DirectBit = int64(1) << 2 // route type 2 (DIRECT) + hm287FloodBit = int64(1) << 1 // route type 1 (FLOOD) +) + +// hm287InsObs writes one observation with a raw_hex header. observer and path +// decide whether the loser's copy survives being re-parented onto the survivor: +// a distinct observer or path keeps it, a matching pair makes idx_observations_dedup +// drop it. +func hm287InsObs(t *testing.T, db *sql.DB, txID, observer int, path, rawHex string) { t.Helper() hm215Exec(t, db, `INSERT INTO observations (transmission_id, observer_idx, direction, path_json, timestamp, raw_hex) - VALUES (?, ?, 'RX', ?, 1767225600, ?)`, txID, observer, `["aa"]`, rawHex) + VALUES (?, ?, 'RX', ?, 1767225600, ?)`, txID, observer, path, rawHex) } -// A survivor with a computed DIRECT mask and a loser that is still NULL but was -// heard as FLOOD: after merge + backfill the mask must be DIRECT|FLOOD, not the -// survivor's DIRECT alone. Kills the mutant that restores COALESCE(route_mask,0) -// (which leaves a non-NULL DIRECT that the backfill skips). -func TestContentHashMigration_MergeKeepsRouteMaskNullWhenLoserUnknown_287(t *testing.T) { - // A DIRECT frame header (low two bits = 2) and a FLOOD frame header - // (firstIngestedFloodRaw, low two bits = 1). - const directFrame = "12" + "00000000" - const directBit = int64(1) << 2 // route type 2 (DIRECT) - const floodBit = int64(1) << 1 // route type 1 (FLOOD) - - s := hm215Reopen(t, filepath.Join(t.TempDir(), "rm287.db"), func(db *sql.DB) { - for _, o := range []string{"o1", "o2"} { - hm215Exec(t, db, `INSERT OR IGNORE INTO observers (id, name, first_seen, last_seen) VALUES (?, ?, '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')`, o, o) - } - o1 := hm215Count(t, db, `SELECT rowid FROM observers WHERE id = 'o1'`) - o2 := hm215Count(t, db, `SELECT rowid FROM observers WHERE id = 'o2'`) +func hm287Observers(t *testing.T, db *sql.DB) (o1, o2 int) { + t.Helper() + for _, o := range []string{"o1", "o2"} { + hm215Exec(t, db, `INSERT OR IGNORE INTO observers (id, name, first_seen, last_seen) VALUES (?, ?, '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')`, o, o) + } + return hm215Count(t, db, `SELECT rowid FROM observers WHERE id = 'o1'`), + hm215Count(t, db, `SELECT rowid FROM observers WHERE id = 'o2'`) +} - // Survivor (lowest id): DIRECT, already computed (route_mask = directBit). +// Survivor computed DIRECT + loser still NULL but heard as FLOOD, both +// observations surviving the merge: the merge must recompute the union inline +// (DIRECT|FLOOD), keep it non-NULL, and log a route_mask_changes row so a +// running server learns the grown mask. Kills the mutant that stops logging the +// merge's change row (which leaves a running server stuck on DIRECT until it +// restarts, #287 finding 3). +func TestContentHashMigration_MergeRecomputesUnionWhenLoserUnknown_287(t *testing.T) { + s := hm215Reopen(t, filepath.Join(t.TempDir(), "rm287-loser.db"), func(db *sql.DB) { + o1, o2 := hm287Observers(t, db) + // Survivor (lowest id): DIRECT, already computed. hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) - VALUES (70, ?, 'stale-rm-70', '2026-01-01T00:00:00Z', 1, 2, 4, '{}', ?)`, hm215Raw(7), directBit) + VALUES (70, ?, 'stale-rm-70', '2026-01-01T00:00:00Z', 1, 2, 4, '{}', ?)`, hm215Raw(7), hm287DirectBit) // Loser: route_mask NULL (not computed), but heard as FLOOD. hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) VALUES (71, ?, 'stale-rm-71', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', NULL)`, hm215Raw(7)) - hm287InsObs(t, db, 70, o1, directFrame) - hm287InsObs(t, db, 71, o2, firstIngestedFloodRaw) + hm287InsObs(t, db, 70, o1, `["aa"]`, hm287DirectFrame) + hm287InsObs(t, db, 71, o2, `["bb"]`, hm287FloodFrame) }) - // After the merge the mask must be NULL so the backfill owns it, and no - // route_mask_changes row may be logged for a mask that is not yet known. + want := hm287DirectBit | hm287FloodBit var merged sql.NullInt64 if err := s.db.QueryRow(`SELECT route_mask FROM transmissions WHERE id = 70`).Scan(&merged); err != nil { t.Fatalf("survivor row: %v", err) } - if merged.Valid { - t.Fatalf("merged route_mask = %d, want NULL so the backfill recomputes it", merged.Int64) + if !merged.Valid || merged.Int64 != want { + t.Fatalf("merged route_mask = %v, want DIRECT|FLOOD = %04b recomputed inline", merged, want) } - if n := hm215Count(t, s.db, `SELECT COUNT(*) FROM route_mask_changes WHERE transmission_id = 70`); n != 0 { - t.Fatalf("route_mask_changes logged %d rows for a not-yet-known mask, want 0", n) + // A running server that holds the survivor must learn the grown mask. + if got := hm215Count(t, s.db, `SELECT COALESCE(MAX(route_mask), -1) FROM route_mask_changes WHERE transmission_id = 70`); int64(got) != want { + t.Fatalf("route_mask_changes announces mask %d for the survivor, want %d", got, want) } if n := hm215Count(t, s.db, `SELECT COUNT(*) FROM transmissions WHERE id = 71`); n != 0 { t.Fatalf("loser 71 was not deleted") } - // The backfill recomputes from route_type and every surviving observation - // (both sides', now re-parented to the survivor): DIRECT|FLOOD. + // The mask is already known, so the backfill leaves it untouched. if err := s.backfillTxRouteMask(context.Background(), s.db); err != nil { t.Fatalf("backfill: %v", err) } @@ -77,7 +87,91 @@ func TestContentHashMigration_MergeKeepsRouteMaskNullWhenLoserUnknown_287(t *tes if err := s.db.QueryRow(`SELECT route_mask FROM transmissions WHERE id = 70`).Scan(&final); err != nil { t.Fatalf("survivor row after backfill: %v", err) } - if want := directBit | floodBit; !final.Valid || final.Int64 != want { + if !final.Valid || final.Int64 != want { + t.Fatalf("route_mask after backfill = %v, want DIRECT|FLOOD = %04b unchanged", final, want) + } +} + +// The reverse direction: survivor still NULL (heard FLOOD), loser already +// computed DIRECT and holding the current hash, both observations surviving. +// The merge must recompute the union inline from the survivor's own route_type +// and observations too, not just copy the loser's mask. Kills the mutant that +// recomputes only when the loser is the NULL side and otherwise defers the +// survivor-NULL case back to the backfill (#287 finding 2). +func TestContentHashMigration_MergeRecomputesUnionWhenSurvivorUnknown_287(t *testing.T) { + s := hm215Reopen(t, filepath.Join(t.TempDir(), "rm287-survivor.db"), func(db *sql.DB) { + o1, o2 := hm287Observers(t, db) + // Survivor (lowest id): route_mask NULL (not computed), heard as FLOOD. + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (72, ?, 'stale-rm-72', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', NULL)`, hm215Raw(9)) + // Loser: DIRECT, already computed, holding the current hash. + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (73, ?, ?, '2026-01-01T00:00:00Z', 1, 2, 4, '{}', ?)`, hm215Raw(9), ComputeContentHash(hm215Raw(9)), hm287DirectBit) + hm287InsObs(t, db, 72, o1, `["aa"]`, hm287FloodFrame) + hm287InsObs(t, db, 73, o2, `["bb"]`, hm287DirectFrame) + }) + + // Survivor is id 72 (lowest), holding the current hash after the merge. + want := hm287DirectBit | hm287FloodBit + var merged sql.NullInt64 + if err := s.db.QueryRow(`SELECT route_mask FROM transmissions WHERE id = 72`).Scan(&merged); err != nil { + t.Fatalf("survivor row: %v", err) + } + if !merged.Valid || merged.Int64 != want { + t.Fatalf("merged route_mask = %v, want DIRECT|FLOOD = %04b recomputed inline", merged, want) + } + if n := hm215Count(t, s.db, `SELECT COUNT(*) FROM transmissions WHERE id = 73`); n != 0 { + t.Fatalf("loser 73 was not deleted") + } +} + +// Finding 1 (blocking): a bit the computed side stored from a frame that no +// surviving observation can rebuild must not be lost. The survivor is NULL +// (heard FLOOD); the loser holds a computed DIRECT mask whose only DIRECT frame +// is an observation that idx_observations_dedup drops when it is re-parented +// (same observer and path as the survivor's FLOOD observation), and whose +// route_type is FLOOD, not DIRECT. Deferring to the backfill (what the NULL-only +// fix did) would recompute FLOOD alone and lose the stored DIRECT bit for good. +// Kills the mutant that recomputes from route_type and observations but drops +// the stored masks. +func TestContentHashMigration_MergeKeepsStoredBitsOfDedupDroppedObservation_287(t *testing.T) { + s := hm215Reopen(t, filepath.Join(t.TempDir(), "rm287-dedup.db"), func(db *sql.DB) { + o1, _ := hm287Observers(t, db) + // Survivor (lowest id): NULL, route_type FLOOD, one FLOOD observation. + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (90, ?, 'stale-rm-90', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', NULL)`, hm215Raw(6)) + // Loser: computed DIRECT mask, but route_type FLOOD, and its only DIRECT + // evidence is an observation that dedups against the survivor's. + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (91, ?, 'stale-rm-91', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', ?)`, hm215Raw(6), hm287DirectBit) + hm287InsObs(t, db, 90, o1, `["aa"]`, hm287FloodFrame) + hm287InsObs(t, db, 91, o1, `["aa"]`, hm287DirectFrame) // dropped: same observer and path + }) + + // The loser's DIRECT observation was dropped, so only the stored DIRECT bit + // carries it. The merged mask must still be DIRECT|FLOOD. + want := hm287DirectBit | hm287FloodBit + var merged sql.NullInt64 + if err := s.db.QueryRow(`SELECT route_mask FROM transmissions WHERE id = 90`).Scan(&merged); err != nil { + t.Fatalf("survivor row: %v", err) + } + if !merged.Valid || merged.Int64 != want { + t.Fatalf("merged route_mask = %v, want DIRECT|FLOOD = %04b (stored DIRECT bit kept)", merged, want) + } + // Only the survivor's FLOOD observation survives the dedup. + if n := hm215Count(t, s.db, `SELECT COUNT(*) FROM observations WHERE transmission_id = 90`); n != 1 { + t.Fatalf("survivor holds %d observations, want 1 (the loser's DIRECT one is dropped)", n) + } + // The backfill alone cannot rebuild DIRECT from the surviving observation; + // the mask must already hold it and stay unchanged. + if err := s.backfillTxRouteMask(context.Background(), s.db); err != nil { + t.Fatalf("backfill: %v", err) + } + var final sql.NullInt64 + if err := s.db.QueryRow(`SELECT route_mask FROM transmissions WHERE id = 90`).Scan(&final); err != nil { + t.Fatalf("survivor row after backfill: %v", err) + } + if !final.Valid || final.Int64 != want { t.Fatalf("route_mask after backfill = %v, want DIRECT|FLOOD = %04b", final, want) } } @@ -97,7 +191,7 @@ func TestContentHashMigration_MergeKeepsFillColumnsNullWhenBothUnknown_287(t *te VALUES (80, ?, 'stale-fill-80', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', 1, NULL, NULL, NULL)`, hm215Raw(8)) hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask, scope_name, channel_hash, from_pubkey) VALUES (81, ?, 'stale-fill-81', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', 1, NULL, NULL, NULL)`, hm215Raw(8)) - hm287InsObs(t, db, 80, o1, "11"+"00000000") + hm287InsObs(t, db, 80, o1, `["aa"]`, hm287FloodFrame) }) var scope, channel, from sql.NullString if err := s.db.QueryRow(`SELECT scope_name, channel_hash, from_pubkey FROM transmissions WHERE id = 80`).Scan(&scope, &channel, &from); err != nil { diff --git a/cmd/ingestor/route_mask_backfill.go b/cmd/ingestor/route_mask_backfill.go index 11e780c08..515622f04 100644 --- a/cmd/ingestor/route_mask_backfill.go +++ b/cmd/ingestor/route_mask_backfill.go @@ -111,6 +111,17 @@ func createRouteMaskPendingIndex(ctx context.Context, d *sql.DB) error { return nil } +// routeMaskBitFromHeader returns the #89 route-mask bit of an observation's +// raw_hex header prefix (its first two hex chars), or 0 when the prefix is +// missing or does not parse. Shared by the route_mask backfill and the +// hash-merge inline recompute (#287) so both read a frame header the same way. +func routeMaskBitFromHeader(header string) int64 { + if rt, ok := packetpath.RouteTypeFromRawHex(header); ok { + return packetpath.RouteMaskBit(rt) + } + return 0 +} + // backfillTxRouteMaskBatch fills up to limit NULL rows in one write // transaction and returns how many it updated. It holds writerMu like // InsertTransmission, so a live observation is either fully visible to the @@ -177,9 +188,7 @@ func (s *Store) backfillTxRouteMaskBatch(ctx context.Context, d *sql.DB, limit i } // Unparseable or truncated frames contribute nothing; they never stop // the backfill. - if rt, ok := packetpath.RouteTypeFromRawHex(header.String); ok { - masks[id] |= packetpath.RouteMaskBit(rt) - } + masks[id] |= routeMaskBitFromHeader(header.String) } if err := orows.Err(); err != nil { orows.Close() From 4b89c5fd2955ee63bd91aac88ef0accdb19e1940 Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 6 Oct 2026 09:08:54 +0000 Subject: [PATCH 3/4] test(ingestor): pin the #287 inline route_mask recompute sources (#287) Review round 3 (findings 1 and 3). Three guards for code shipped in round 2, red under the reviewer's mutants: - the only DIRECT evidence is a surviving observation header (kills MX2 no scan, MX3 scan the loser's rows, MX4 header bit 0); - the only DIRECT evidence is the loser's route_type (kills MX1); - both sides NULL, with and without evidence: the merge writes a known mask and logs exactly one route_mask_changes row (kills a both-NULL merge that stays NULL, and a merge that only logs for a known survivor). Co-Authored-By: Claude Opus 5.5 --- .../hash_migrate_route_mask_287_test.go | 130 ++++++++++++++++++ 1 file changed, 130 insertions(+) diff --git a/cmd/ingestor/hash_migrate_route_mask_287_test.go b/cmd/ingestor/hash_migrate_route_mask_287_test.go index 021837bbe..5818ef07b 100644 --- a/cmd/ingestor/hash_migrate_route_mask_287_test.go +++ b/cmd/ingestor/hash_migrate_route_mask_287_test.go @@ -201,3 +201,133 @@ func TestContentHashMigration_MergeKeepsFillColumnsNullWhenBothUnknown_287(t *te t.Fatalf("a both-NULL fill column was given a value: scope=%v channel=%v from=%v; want all NULL", scope, channel, from) } } + +// hm287Mask reads a transmission's route_mask. +func hm287Mask(t *testing.T, db *sql.DB, id int) sql.NullInt64 { + t.Helper() + var m sql.NullInt64 + if err := db.QueryRow(`SELECT route_mask FROM transmissions WHERE id = ?`, id).Scan(&m); err != nil { + t.Fatalf("transmission %d: %v", id, err) + } + return m +} + +// The inline recompute must read the surviving observation headers. The only +// DIRECT evidence is an observation of the not-yet-computed survivor whose +// header is DIRECT (its route_type is FLOOD, and the loser's computed mask and +// route_type are FLOOD too). The merged mask is non-NULL, so the backfill never +// gets a second chance: a merge that skips or misreads the observation scan +// leaves FLOOD for good. Kills reviewer mutants MX2 (no observation scan), MX3 +// (scan the loser's observations, empty after the move) and MX4 +// (routeMaskBitFromHeader returns 0). +func TestContentHashMigration_MergeRecomputesBitFromSurvivingObservationHeader_287(t *testing.T) { + s := hm215Reopen(t, filepath.Join(t.TempDir(), "rm287-header.db"), func(db *sql.DB) { + o1, o2 := hm287Observers(t, db) + // Survivor (lowest id): NULL, route_type FLOOD, heard once as FLOOD and + // once as DIRECT. + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (100, ?, 'stale-rm-100', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', NULL)`, hm215Raw(10)) + // Loser: computed FLOOD, route_type FLOOD. + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (101, ?, 'stale-rm-101', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', ?)`, hm215Raw(10), hm287FloodBit) + hm287InsObs(t, db, 100, o1, `["aa"]`, hm287FloodFrame) + hm287InsObs(t, db, 100, o2, `["bb"]`, hm287DirectFrame) // the only DIRECT evidence + hm287InsObs(t, db, 101, o1, `["cc"]`, hm287FloodFrame) + }) + + want := hm287DirectBit | hm287FloodBit + if merged := hm287Mask(t, s.db, 100); !merged.Valid || merged.Int64 != want { + t.Fatalf("merged route_mask = %v, want DIRECT|FLOOD = %04b (DIRECT from the surviving observation header)", merged, want) + } + if err := s.backfillTxRouteMask(context.Background(), s.db); err != nil { + t.Fatalf("backfill: %v", err) + } + if final := hm287Mask(t, s.db, 100); !final.Valid || final.Int64 != want { + t.Fatalf("route_mask after backfill = %v, want DIRECT|FLOOD = %04b", final, want) + } +} + +// The inline recompute must read the loser's route_type, not only the +// survivor's. The only DIRECT evidence is the not-yet-computed loser's +// route_type: its one observation has no stored frame (raw_hex NULL), and the +// survivor is computed FLOOD with a FLOOD observation. Master's +// COALESCE(route_mask, 0) merge ends at FLOOD here too. Kills reviewer mutant +// MX1 (read only the winner's route_type). +func TestContentHashMigration_MergeRecomputesBitFromLoserRouteType_287(t *testing.T) { + s := hm215Reopen(t, filepath.Join(t.TempDir(), "rm287-loser-rt.db"), func(db *sql.DB) { + o1, o2 := hm287Observers(t, db) + // Survivor (lowest id): computed FLOOD, route_type FLOOD. + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (110, ?, 'stale-rm-110', '2026-01-01T00:00:00Z', 1, 1, 4, '{}', ?)`, hm215Raw(11), hm287FloodBit) + // Loser: NULL, heard as DIRECT; its observation kept no raw_hex. + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (111, ?, 'stale-rm-111', '2026-01-01T00:00:00Z', 1, 2, 4, '{}', NULL)`, hm215Raw(11)) + hm287InsObs(t, db, 110, o1, `["aa"]`, hm287FloodFrame) + hm215Obs(t, db, 111, o2, `["bb"]`) + }) + + want := hm287DirectBit | hm287FloodBit + if merged := hm287Mask(t, s.db, 110); !merged.Valid || merged.Int64 != want { + t.Fatalf("merged route_mask = %v, want DIRECT|FLOOD = %04b (DIRECT from the loser's route_type)", merged, want) + } + if err := s.backfillTxRouteMask(context.Background(), s.db); err != nil { + t.Fatalf("backfill: %v", err) + } + if final := hm287Mask(t, s.db, 110); !final.Valid || final.Int64 != want { + t.Fatalf("route_mask after backfill = %v, want DIRECT|FLOOD = %04b", final, want) + } +} + +// Both sides NULL: the merge writes a known mask instead of leaving NULL, and an +// uncomputed survivor that the merge gives a mask (even 0) logs exactly one +// route_mask_changes row. With evidence the mask is the recomputed union; with +// none it is a known 0, which the backfill leaves alone (it would have written 0 +// too) and live ingest still grows. Kills the mutant that leaves a both-NULL +// merge NULL (deferring to the backfill as master did) and the mutant that logs +// a change row only when the survivor's mask was already known. +func TestContentHashMigration_MergeBothRouteMasksUnknown_287(t *testing.T) { + cases := []struct { + name string + survivorRT interface{} + loserRT interface{} + survivorFrame string // "" = no observation + loserFrame string + want int64 + }{ + {"with evidence", 1, 2, hm287FloodFrame, hm287DirectFrame, hm287DirectBit | hm287FloodBit}, + {"without evidence", nil, nil, "", "", 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + s := hm215Reopen(t, filepath.Join(t.TempDir(), "rm287-both-null.db"), func(db *sql.DB) { + o1, o2 := hm287Observers(t, db) + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (120, ?, 'stale-rm-120', '2026-01-01T00:00:00Z', 1, ?, 4, '{}', NULL)`, hm215Raw(12), tc.survivorRT) + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, last_seen, route_type, payload_type, decoded_json, route_mask) + VALUES (121, ?, 'stale-rm-121', '2026-01-01T00:00:00Z', 1, ?, 4, '{}', NULL)`, hm215Raw(12), tc.loserRT) + if tc.survivorFrame != "" { + hm287InsObs(t, db, 120, o1, `["aa"]`, tc.survivorFrame) + } + if tc.loserFrame != "" { + hm287InsObs(t, db, 121, o2, `["bb"]`, tc.loserFrame) + } + }) + + if merged := hm287Mask(t, s.db, 120); !merged.Valid || merged.Int64 != tc.want { + t.Fatalf("merged route_mask = %v, want known %04b written by the merge", merged, tc.want) + } + if n := hm215Count(t, s.db, `SELECT COUNT(*) FROM route_mask_changes WHERE transmission_id = 120`); n != 1 { + t.Fatalf("route_mask_changes rows for the survivor = %d, want 1", n) + } + if got := hm215Count(t, s.db, `SELECT route_mask FROM route_mask_changes WHERE transmission_id = 120`); int64(got) != tc.want { + t.Fatalf("route_mask_changes announces %d, want %d", got, tc.want) + } + if err := s.backfillTxRouteMask(context.Background(), s.db); err != nil { + t.Fatalf("backfill: %v", err) + } + if final := hm287Mask(t, s.db, 120); !final.Valid || final.Int64 != tc.want { + t.Fatalf("route_mask after backfill = %v, want %04b", final, tc.want) + } + }) + } +} From c0d1d51a08f8f2ea689ad0b7c81560bb31d9fc97 Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 6 Oct 2026 09:08:54 +0000 Subject: [PATCH 4/4] docs(ingestor): note the change row a NULL-mask survivor logs on merge (#287) Review round 3 (finding 4): a survivor whose route_mask was NULL always logs one route_mask_changes row, even for a known 0, so a pre-#89 DB logs one row per both-NULL merge. Comment only. Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/hash_migrate.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/cmd/ingestor/hash_migrate.go b/cmd/ingestor/hash_migrate.go index 652af6ebe..6b43bbb6b 100644 --- a/cmd/ingestor/hash_migrate.go +++ b/cmd/ingestor/hash_migrate.go @@ -47,7 +47,9 @@ import ( // surviving observation can rebuild) plus the #89 lower bound the backfill // would produce (route_type and the surviving observation headers). The // result is non-NULL, so the backfill leaves it alone and the change is -// logged like any grown mask; and every nullable column it has no value for +// logged like any grown mask, except that a survivor whose mask was NULL +// always logs one row, even for a known 0 (one per both-NULL merge on a +// pre-#89 DB); and every nullable column it has no value for // (scope_name, channel_hash, from_pubkey, ...) from the duplicate; a value // it has stays; // - rows hung off the duplicate follow the survivor (ping_triggers), or go