diff --git a/cmd/ingestor/async_migration.go b/cmd/ingestor/async_migration.go index 32133db4e..2ab75a314 100644 --- a/cmd/ingestor/async_migration.go +++ b/cmd/ingestor/async_migration.go @@ -32,6 +32,7 @@ package main import ( "context" "database/sql" + "errors" "fmt" "log" ) @@ -109,6 +110,12 @@ func (s *Store) RunAsyncMigration(ctx context.Context, name string, fn func(cont WHERE name = ?`, runErr.Error(), name); err != nil { log.Printf("[async-migration] failed to record failure for %q: %v", name, err) } + if errors.Is(runErr, context.Canceled) { + // A stop at shutdown: recorded as unfinished like a failure so + // the next start re-runs it, but not a failure. + log.Printf("[async-migration] %q cancelled (will resume): %v", name, runErr) + return + } log.Printf("[async-migration] %q FAILED: %v", name, runErr) return } diff --git a/cmd/ingestor/hash_migrate.go b/cmd/ingestor/hash_migrate.go new file mode 100644 index 000000000..61422dcd9 --- /dev/null +++ b/cmd/ingestor/hash_migrate.go @@ -0,0 +1,358 @@ +package main + +import ( + "context" + "database/sql" + "fmt" + "log" + "time" +) + +// Content-hash migration (#215). +// +// transmissions.hash is the content hash of the raw packet, and the formula +// has changed over the project's life (#786, #787). Rows written before a +// change carry a stale hash. They are rehashed here, in the process that owns +// the DB, once: this runs as the async migration below and is recorded in +// _async_migrations, so it does not repeat at the next start. It used to run in +// cmd/server on the server's mode=ro handle, where every UPDATE and DELETE +// failed, was logged as a collision, and was retried at every start. +// +// New rows are written with the current formula (InsertTransmission hashes with +// ComputeContentHash), so only rows older than the formula can be stale. +// +// Rehashing can make two rows collide: the same packet stored under two stale +// hashes, or a stale row and a row already written with the current hash. +// transmissions.hash is UNIQUE, so the rows are merged instead: +// +// - the row with the LOWEST ID survives. cmd/server's in-memory migration +// applies the same rule, so a server that has not restarted yet and one +// that has agree on which row stays; +// - the duplicate's observations move to it. An observation the survivor +// already has (same observer and path, which idx_observations_dedup rejects) +// is dropped instead, and the survivor's copy wins, as it does when ingest +// meets a repeat reception. The earlier copy could win only by copying every +// observation column, the optional ones (resolved_path, raw_hex) included, +// over the survivor's row, and by the same in the server's in-memory +// observation; what is lost is the dropped copy's SNR, RSSI and time (same +// 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), 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. +// +// If the formula changes again, bump the migration name: a recorded migration +// is never re-run. +const contentHashMigration = "content_hash_formula_v1" + +// StartContentHashMigration schedules the migration. Like the route_mask +// backfill it is started by main once the ingest buffer is draining, not by +// OpenStore: it scans the whole of transmissions and writes in batches, which +// the buffer absorbs, and OpenStore callers (tests, tools) that seed their own +// rows must not have them rehashed under them. Cancelled on shutdown; a +// migration that did not finish resumes at the next start. +func (s *Store) StartContentHashMigration(ctx context.Context) { + // PREFLIGHT: async=true reason="full-table scan of transmissions.raw_hex (hash recompute) with batched writes; must not block ingestor boot" + if err := s.RunAsyncMigration(ctx, contentHashMigration, s.migrateContentHashes); err != nil { + log.Printf("[hash-migrate] scheduling %s failed: %v", contentHashMigration, err) + } +} + +// Package vars, not consts, so tests can drive the multi-batch loop with a few +// rows. +var ( + contentHashMigrationBatchSize = 2000 + contentHashMigrationYield = 20 * time.Millisecond + + // contentHashMigrationHook is a test seam, nil in production. It is called + // after a batch was scanned and before it is rewritten ("scanned"), and + // after the rewrite ("batch"); n is the number of rows scanned so far. + contentHashMigrationHook func(stage string, n int) +) + +type staleContentHash struct { + id int64 + oldHash string + newHash string +} + +// migrateContentHashes is the content_hash_formula_v1 async migration body. +// Each batch is read, then rewritten in one transaction under writerMu like +// every other writer, so live inserts queue behind a batch rather than the +// whole migration (the same shape as backfillTxRouteMask). +func (s *Store) migrateContentHashes(ctx context.Context, d *sql.DB) error { + start := time.Now() + ex := hashMigrationTables{ + pingTriggers: tableExists(d, "ping_triggers"), + routeMaskChanges: tableExists(d, "route_mask_changes"), + } + fill, err := fillableColumns(d) + if err != nil { + return fmt.Errorf("columns: %w", err) + } + ex.fillCols = fill + var scanned, rehashed, merged int64 + var lastID int64 + for { + // A cancel (shutdown) is an error, not a normal end: the run must be + // recorded as unfinished so the next start resumes it. + if err := ctx.Err(); err != nil { + return err + } + stale, next, n, err := scanStaleContentHashes(ctx, d, lastID, contentHashMigrationBatchSize) + if err != nil { + return fmt.Errorf("scan: %w", err) + } + if n == 0 { + break + } + scanned += int64(n) + lastID = next + if h := contentHashMigrationHook; h != nil { + h("scanned", int(scanned)) + } + if len(stale) > 0 { + r, m, err := s.rewriteContentHashes(ctx, d, stale, ex) + if err != nil { + return fmt.Errorf("rewrite: %w", err) + } + rehashed += r + merged += m + } + if h := contentHashMigrationHook; h != nil { + h("batch", int(scanned)) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(contentHashMigrationYield): + } + } + if rehashed > 0 || merged > 0 { + log.Printf("[hash-migrate] rehashed %d of %d transmissions to the current formula, merged %d duplicates, in %s", + rehashed, scanned, merged, time.Since(start).Round(time.Millisecond)) + } + return nil +} + +// tableExists reports whether a table is present. Both optional tables are +// created by dbschema.Apply before this runs; the check keeps the migration +// from failing on a DB that predates them. +func tableExists(d *sql.DB, name string) bool { + var n int + return d.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?`, name).Scan(&n) == nil && n > 0 +} + +type hashMigrationTables struct { + pingTriggers bool + routeMaskChanges bool + // fillCols are the columns of fillColumns the table has: a merge fills them + // from the duplicate when the survivor's value is NULL. + fillCols []string +} + +// fillColumns are the nullable columns of transmissions a merge fills with +// COALESCE: the ones that describe the packet and are NULL only because the +// row that was ingested first did not have the value (an older schema, a copy +// that was not transport-scoped). It is an allow-list on purpose: a nullable +// column added later may use NULL for "pending" (a backfill that has not run +// yet), and filling it from a duplicate would pass that off as a value. A new +// column that should be filled is added here, with a test. +var fillColumns = []string{ + "route_type", "payload_type", "payload_version", "decoded_json", + "from_pubkey", "channel_hash", "scope_name", +} + +// fillableColumns returns the columns of fillColumns that transmissions has, in +// the order of fillColumns: an older schema lacks some of them. +func fillableColumns(d *sql.DB) ([]string, error) { + rows, err := d.Query(`SELECT name FROM pragma_table_info('transmissions')`) + if err != nil { + return nil, err + } + defer rows.Close() + have := make(map[string]bool) + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + return nil, err + } + have[name] = true + } + if err := rows.Err(); err != nil { + return nil, err + } + var cols []string + for _, c := range fillColumns { + if have[c] { + cols = append(cols, c) + } + } + return cols, nil +} + +// scanStaleContentHashes reads the next batch of rows after afterID and returns +// those whose hash differs from the current formula, the last id scanned, and +// how many rows it scanned (0 at the end of the table). +func scanStaleContentHashes(ctx context.Context, d *sql.DB, afterID int64, limit int) ([]staleContentHash, int64, int, error) { + rows, err := d.QueryContext(ctx, `SELECT id, raw_hex, hash FROM transmissions WHERE id > ? ORDER BY id LIMIT ?`, afterID, limit) + if err != nil { + return nil, afterID, 0, err + } + defer rows.Close() + var stale []staleContentHash + last := afterID + n := 0 + for rows.Next() { + var id int64 + var raw, hash string + if err := rows.Scan(&id, &raw, &hash); err != nil { + return nil, afterID, 0, err + } + last = id + n++ + if raw == "" { + continue + } + if want := ComputeContentHash(raw); want != hash { + stale = append(stale, staleContentHash{id: id, oldHash: hash, newHash: want}) + } + } + return stale, last, n, rows.Err() +} + +// rewriteContentHashes applies one batch in a single transaction. +func (s *Store) rewriteContentHashes(ctx context.Context, d *sql.DB, stale []staleContentHash, ex hashMigrationTables) (rehashed, merged int64, err error) { + waitStart := time.Now() + writerMu.Lock() + wait := time.Since(waitStart) + holdStart := time.Now() + defer func() { + hold := time.Since(holdStart) + writerMu.Unlock() + recordWriterTiming("content_hash_migration", wait, hold, "rewriteContentHashes") + }() + + tx, err := d.BeginTx(ctx, nil) + if err != nil { + return 0, 0, err + } + defer tx.Rollback() // no-op after Commit + + for _, u := range stale { + // The scan ran before writerMu was taken: retention may have removed + // the row since. + var current string + if err := tx.QueryRowContext(ctx, `SELECT hash FROM transmissions WHERE id = ?`, u.id).Scan(¤t); err == sql.ErrNoRows { + continue + } else if err != nil { + return 0, 0, err + } + if current != u.oldHash { + continue + } + var holder int64 + finalID := u.id + err := tx.QueryRowContext(ctx, `SELECT id FROM transmissions WHERE hash = ?`, u.newHash).Scan(&holder) + switch { + case err == sql.ErrNoRows: + // No collision: the hash is simply rewritten. + case err != nil: + return 0, 0, err + default: + winner, loser := u.id, holder + if holder < u.id { + winner, loser = holder, u.id + } + if err := mergeTransmissions(ctx, tx, winner, loser, ex); err != nil { + return 0, 0, fmt.Errorf("merge tx %d into %d: %w", loser, winner, err) + } + merged++ + finalID = winner + } + if finalID == u.id { + // The survivor already holds the new hash when it is the holder; + // otherwise the duplicate holding it is gone and the hash is free. + if _, err := tx.ExecContext(ctx, `UPDATE transmissions SET hash = ? WHERE id = ?`, u.newHash, u.id); err != nil { + return 0, 0, err + } + } + if ex.pingTriggers { + // ping_triggers keeps a copy of the hash. + if _, err := tx.ExecContext(ctx, `UPDATE ping_triggers SET hash = ? WHERE tx_id = ?`, u.newHash, finalID); err != nil { + return 0, 0, err + } + } + rehashed++ + } + if err := tx.Commit(); err != nil { + return 0, 0, err + } + return rehashed, merged, nil +} + +// mergeTransmissions folds loser into winner and deletes loser. See the file +// comment for the rules. The winner's hash is not touched here: when the loser +// holds the new hash, deleting it frees the hash for the caller's UPDATE. +func mergeTransmissions(ctx context.Context, tx *sql.Tx, winner, loser int64, ex hashMigrationTables) error { + // OR IGNORE: an observation the survivor already has violates + // idx_observations_dedup; it stays behind and is deleted just below. + if _, err := tx.ExecContext(ctx, `UPDATE OR IGNORE observations SET transmission_id = ? WHERE transmission_id = ?`, winner, loser); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `DELETE FROM observations WHERE transmission_id = ?`, loser); err != nil { + return err + } + + var oldMask sql.NullInt64 + if err := tx.QueryRowContext(ctx, `SELECT route_mask FROM transmissions WHERE id = ?`, winner).Scan(&oldMask); err != nil { + return err + } + 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" + 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) + args = append(args, loser) + } + args = append(args, winner) + if _, err := tx.ExecContext(ctx, "UPDATE transmissions SET "+set+" WHERE id = ?", args...); err != nil { + return err + } + + if ex.pingTriggers { + // A ping is one row per transmission: keep the survivor's if it has one, + // else it takes over the duplicate's. + if _, err := tx.ExecContext(ctx, `DELETE FROM ping_triggers WHERE tx_id = ? AND EXISTS (SELECT 1 FROM ping_triggers WHERE tx_id = ?)`, loser, winner); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `UPDATE ping_triggers SET tx_id = ? WHERE tx_id = ?`, winner, loser); err != nil { + return err + } + } + if ex.routeMaskChanges { + 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) { + // Same record InsertTransmission leaves when an observation adds a + // route bit: servers that hold the survivor pick up the full mask. + 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 { + return err + } + } + } + _, err := tx.ExecContext(ctx, `DELETE FROM transmissions WHERE id = ?`, loser) + return err +} diff --git a/cmd/ingestor/hash_migrate_215_test.go b/cmd/ingestor/hash_migrate_215_test.go new file mode 100644 index 000000000..293c63d92 --- /dev/null +++ b/cmd/ingestor/hash_migrate_215_test.go @@ -0,0 +1,601 @@ +package main + +import ( + "context" + "database/sql" + "fmt" + "log" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "testing" + "time" +) + +// Issue #215: the content-hash migration (rehash with the current formula, +// merge rows that collide, delete the duplicate) is a DB write and belongs to +// the ingestor. The server used to run it on its read-only handle, where every +// statement failed and was retried at every start. + +func hm215Raw(key int) string { + return fmt.Sprintf("0A00D69FD7A5A7475DB07337749AE61FA53A4788E9%02X", key) +} + +// hm215Prepare opens the DB at path, drops the migration's record, lets fn seed +// it, and reopens it WITHOUT starting the migration. The record is dropped +// because it is marked done on the first open: the seeded DB then looks like one +// written by an ingestor from before the migration existed. +func hm215Prepare(t *testing.T, path string, fn func(db *sql.DB)) *Store { + t.Helper() + s, err := OpenStore(path) + if err != nil { + t.Fatal(err) + } + s.WaitForAsyncMigrations() + if _, err := s.db.Exec(`DELETE FROM _async_migrations WHERE name LIKE '%content_hash%'`); err != nil { + t.Fatal(err) + } + fn(s.db) + if err := s.Close(); err != nil { + t.Fatal(err) + } + s, err = OpenStore(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { s.Close() }) + return s +} + +// hm215Reopen is hm215Prepare followed by running the migration to the end. +func hm215Reopen(t *testing.T, path string, fn func(db *sql.DB)) *Store { + t.Helper() + s := hm215Prepare(t, path, fn) + s.StartContentHashMigration(context.Background()) + s.WaitForAsyncMigrations() + return s +} + +func hm215Exec(t *testing.T, db *sql.DB, q string, args ...interface{}) { + t.Helper() + if _, err := db.Exec(q, args...); err != nil { + t.Fatalf("%v\n%s", err, q) + } +} + +func hm215Tx(t *testing.T, db *sql.DB, id int, raw, hash string, lastSeen, mask int) { + t.Helper() + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json, last_seen, route_mask) + VALUES (?, ?, ?, ?, 1, 4, '{}', ?, ?)`, id, raw, hash, fmt.Sprintf("2026-01-01T00:00:%02dZ", id%60), lastSeen, mask) +} + +func hm215Obs(t *testing.T, db *sql.DB, txID, observer int, path string) { + t.Helper() + hm215Exec(t, db, `INSERT INTO observations (transmission_id, observer_idx, direction, snr, rssi, score, path_json, timestamp) + VALUES (?, ?, 'RX', -5.0, -90.0, 5, ?, 1767225600)`, txID, observer, path) +} + +func hm215Count(t *testing.T, db *sql.DB, q string, args ...interface{}) int { + t.Helper() + var n int + if err := db.QueryRow(q, args...).Scan(&n); err != nil { + t.Fatalf("%v\n%s", err, q) + } + return n +} + +// hm215Seed writes the rows every test below migrates: +// +// A: ids 11,12,13 share raw A. Rows overlap in observations (12 repeats an +// observation of 11, 13 one of 12), which the observations dedup index +// would reject when re-parented. +// B: id 20 is stale, id 25 already carries the current hash of the same raw +// packet. The lowest id (20) must survive although 25 holds the hash. +// C: id 30 is stale and has no duplicate. +// D: id 40 is current and must not be touched. +func hm215Seed(t *testing.T, db *sql.DB) { + t.Helper() + for i := 1; i <= 3; i++ { + 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')`, fmt.Sprintf("o%d", i), fmt.Sprintf("O%d", i)) + } + rowid := func(id string) int { + return hm215Count(t, db, `SELECT rowid FROM observers WHERE id = ?`, id) + } + o1, o2, o3 := rowid("o1"), rowid("o2"), rowid("o3") + + hm215Tx(t, db, 11, hm215Raw(1), "stale-a-1", 100, 1) + hm215Tx(t, db, 12, hm215Raw(1), "stale-a-2", 300, 2) + hm215Tx(t, db, 13, hm215Raw(1), "stale-a-3", 200, 4) + hm215Obs(t, db, 11, o1, `["aa"]`) + hm215Obs(t, db, 11, o1, `["bb"]`) + hm215Obs(t, db, 12, o1, `["aa"]`) // same as 11's: rejected by the dedup index once re-parented + hm215Obs(t, db, 12, o2, `["cc"]`) + hm215Obs(t, db, 13, o2, `["cc"]`) // same as 12's + hm215Obs(t, db, 13, o2, `["dd"]`) + hm215Exec(t, db, `INSERT INTO ping_triggers (tx_id, hash, channel_hash, sender, first_seen) VALUES (12, 'stale-a-2', 'c', 's', '2026-01-01T00:00:12Z')`) + hm215Exec(t, db, `INSERT INTO ping_triggers (tx_id, hash, channel_hash, sender, first_seen) VALUES (13, 'stale-a-3', 'c', 's', '2026-01-01T00:00:13Z')`) + hm215Exec(t, db, `INSERT INTO route_mask_changes (transmission_id, route_mask, created_at) VALUES (13, 4, 1)`) + + hm215Tx(t, db, 20, hm215Raw(2), "stale-b-20", 50, 1) + hm215Tx(t, db, 25, hm215Raw(2), ComputeContentHash(hm215Raw(2)), 60, 2) + hm215Obs(t, db, 20, o1, `["ee"]`) + hm215Obs(t, db, 25, o3, `["ff"]`) + hm215Obs(t, db, 25, o1, `["ee"]`) // same as 20's + + hm215Tx(t, db, 30, hm215Raw(3), "stale-c-30", 10, 1) + hm215Obs(t, db, 30, o1, `["01"]`) + + hm215Tx(t, db, 40, hm215Raw(4), ComputeContentHash(hm215Raw(4)), 10, 1) + hm215Obs(t, db, 40, o2, `["02"]`) +} + +func hm215Open(t *testing.T) (*Store, string) { + t.Helper() + path := filepath.Join(t.TempDir(), "hm215.db") + return hm215Reopen(t, path, func(db *sql.DB) { hm215Seed(t, db) }), path +} + +func TestContentHashMigration_ConvergesInTheIngestor_215(t *testing.T) { + for _, batch := range []int{1, 2, 3, 1000} { + t.Run(fmt.Sprintf("batch%d", batch), func(t *testing.T) { + oldBatch, oldYield := contentHashMigrationBatchSize, contentHashMigrationYield + contentHashMigrationBatchSize, contentHashMigrationYield = batch, 0 + t.Cleanup(func() { contentHashMigrationBatchSize, contentHashMigrationYield = oldBatch, oldYield }) + + s, _ := hm215Open(t) + db := s.db + + // One row per content hash, none of them stale. + rows, err := db.Query(`SELECT id, raw_hex, hash FROM transmissions ORDER BY id`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var ids []int + seen := map[string]int{} + for rows.Next() { + var id int + var raw, hash string + if err := rows.Scan(&id, &raw, &hash); err != nil { + t.Fatal(err) + } + if want := ComputeContentHash(raw); hash != want { + t.Errorf("tx %d hash = %s, want %s", id, hash, want) + } + seen[hash]++ + ids = append(ids, id) + } + rows.Close() + // A survives as its lowest id, B as 20 (not 25, the current holder). + if got, want := fmt.Sprint(ids), "[11 20 30 40]"; got != want { + t.Fatalf("surviving transmissions = %s, want %s", got, want) + } + for h, n := range seen { + if n != 1 { + t.Errorf("hash %s is held by %d rows", h, n) + } + } + + // Every observation survives once: A has 4 distinct of 6, B 2 of 3. + for id, want := range map[int]int{11: 4, 20: 2, 30: 1, 40: 1} { + if got := hm215Count(t, db, `SELECT COUNT(*) FROM observations WHERE transmission_id = ?`, id); got != want { + t.Errorf("tx %d holds %d observations, want %d", id, got, want) + } + } + if got := hm215Count(t, db, `SELECT COUNT(*) FROM observations WHERE transmission_id NOT IN (SELECT id FROM transmissions)`); got != 0 { + t.Errorf("%d observations point at a transmission that no longer exists", got) + } + if got := hm215Count(t, db, `SELECT COUNT(*) FROM pragma_foreign_key_check`); got != 0 { + t.Errorf("foreign_key_check reports %d violations", got) + } + + // The survivor carries what the duplicates knew: the latest + // last_seen and every route type seen. first_seen stays its own. + if got := hm215Count(t, db, `SELECT last_seen FROM transmissions WHERE id = 11`); got != 300 { + t.Errorf("survivor last_seen = %d, want the latest of the duplicates (300)", got) + } + if got := hm215Count(t, db, `SELECT route_mask FROM transmissions WHERE id = 11`); got != 7 { + t.Errorf("survivor route_mask = %d, want the union 7", got) + } + if got := hm215Count(t, db, `SELECT route_mask FROM transmissions WHERE id = 20`); got != 3 { + t.Errorf("tx 20 route_mask = %d, want the union with the deleted row 25 (3)", got) + } + var firstSeen string + if err := db.QueryRow(`SELECT first_seen FROM transmissions WHERE id = 11`).Scan(&firstSeen); err != nil || firstSeen != "2026-01-01T00:00:11Z" { + t.Errorf("survivor first_seen = %q (%v), want its own", firstSeen, err) + } + + // Rows hung off the transmission follow the survivor. + if got := hm215Count(t, db, `SELECT COUNT(*) FROM ping_triggers`); got != 1 { + t.Errorf("ping_triggers holds %d rows, want one for the merged ping", got) + } + var pingTx int + var pingHash string + if err := db.QueryRow(`SELECT tx_id, hash FROM ping_triggers`).Scan(&pingTx, &pingHash); err != nil { + t.Fatal(err) + } + if pingTx != 11 || pingHash != ComputeContentHash(hm215Raw(1)) { + t.Errorf("ping trigger = tx %d hash %s, want tx 11 with the new hash", pingTx, pingHash) + } + if got := hm215Count(t, db, `SELECT COUNT(*) FROM route_mask_changes WHERE transmission_id NOT IN (SELECT id FROM transmissions)`); got != 0 { + t.Errorf("%d route_mask_changes rows name a deleted transmission", got) + } + // A running server that holds the survivor learns the grown mask. + if got := hm215Count(t, db, `SELECT COALESCE(MAX(route_mask), 0) FROM route_mask_changes WHERE transmission_id = 11`); got != 7 { + t.Errorf("route_mask_changes announces mask %d for the survivor, want 7", got) + } + }) + } +} + +// The migration runs once. The next start must not scan the table again, and +// a database that has converged is left byte for byte as it is. +func TestContentHashMigration_RunsOnce_215(t *testing.T) { + s, path := hm215Open(t) + var status string + if err := s.db.QueryRow(`SELECT status FROM _async_migrations WHERE name LIKE '%content_hash%'`).Scan(&status); err != nil { + t.Fatalf("migration is not recorded: %v", err) + } + if status != "done" { + t.Fatalf("migration status = %q, want done", status) + } + before := hm215Count(t, s.db, `SELECT COUNT(*) FROM transmissions`) + if err := s.Close(); err != nil { + t.Fatal(err) + } + + // A stale row appearing after convergence is not touched on the next + // start: the ingestor writes the current formula, so none can appear, and + // the migration does not pay a table scan on every boot to look. + s2 := hm215Reopen(t, path, func(db *sql.DB) { + hm215Exec(t, db, `INSERT INTO _async_migrations (name, status) VALUES ('content_hash_formula_v1', 'done') ON CONFLICT(name) DO UPDATE SET status='done'`) + hm215Tx(t, db, 90, hm215Raw(9), "stale-after", 1, 1) + }) + _ = s2 + if hm215Count(t, s2.db, `SELECT COUNT(*) FROM transmissions WHERE hash = 'stale-after'`) != 1 { + t.Error("a completed migration ran again at the next start") + } + if got := hm215Count(t, s2.db, `SELECT COUNT(*) FROM transmissions`); got != before+1 { + t.Errorf("transmissions = %d, want %d", got, before+1) + } +} + +// New rows are written with the current formula, so the migration is only +// needed for rows older than the formula (#786/#787), never for new ones. +func TestInsertWritesTheCurrentContentHash_215(t *testing.T) { + s := newTestStore(t) + raw := hm215Raw(7) + pd := &PacketData{RawHex: raw, Hash: ComputeContentHash(raw), Timestamp: time.Now().UTC().Format(time.RFC3339), PayloadType: 4, RouteType: 1, DecodedJSON: "{}", PathJSON: "[]"} + if _, err := s.InsertTransmission(pd); err != nil { + t.Fatal(err) + } + var hash string + if err := s.db.QueryRow(`SELECT hash FROM transmissions WHERE raw_hex = ?`, raw).Scan(&hash); err != nil { + t.Fatal(err) + } + if want := ComputeContentHash(raw); hash != want || strings.HasPrefix(hash, "stale") { + t.Fatalf("inserted hash = %s, want %s", hash, want) + } +} + +// hm215Dump is the migrated state in a comparable form: the rows and their +// foreign rows, without the autoincrement ids and clock values that differ +// between two runs of the same data. +func hm215Dump(t *testing.T, db *sql.DB) string { + t.Helper() + var b strings.Builder + dump := func(title, q string) { + b.WriteString("== " + title + "\n") + rows, err := db.Query(q) + if err != nil { + t.Fatalf("%v\n%s", err, q) + } + defer rows.Close() + cols, _ := rows.Columns() + vals := make([]interface{}, len(cols)) + ptrs := make([]interface{}, len(cols)) + for i := range vals { + ptrs[i] = &vals[i] + } + for rows.Next() { + if err := rows.Scan(ptrs...); err != nil { + t.Fatal(err) + } + fmt.Fprintln(&b, vals...) + } + } + dump("transmissions", `SELECT id, raw_hex, hash, first_seen, last_seen, COALESCE(route_mask, -1), COALESCE(scope_name, ''), COALESCE(channel_hash, ''), COALESCE(from_pubkey, '') FROM transmissions ORDER BY id`) + dump("observations", `SELECT transmission_id, observer_idx, COALESCE(path_json, ''), timestamp, COALESCE(snr, 0) FROM observations ORDER BY transmission_id, observer_idx, COALESCE(path_json, ''), timestamp`) + dump("ping_triggers", `SELECT tx_id, hash FROM ping_triggers ORDER BY tx_id`) + dump("route_mask_changes", `SELECT transmission_id, route_mask FROM route_mask_changes ORDER BY transmission_id, route_mask`) + return b.String() +} + +func hm215Status(t *testing.T, s *Store) string { + t.Helper() + var status string + if err := s.db.QueryRow(`SELECT status FROM _async_migrations WHERE name = ?`, contentHashMigration).Scan(&status); err != nil { + t.Fatalf("migration is not recorded: %v", err) + } + return status +} + +// A run that is stopped half way (a deploy restart) must not be recorded as +// done, and the restarted run must end in exactly the state an uninterrupted run +// ends in. The migration deletes rows, so a stop that was recorded as done would +// leave the rest unconverted for good. +func TestContentHashMigration_InterruptedRunResumesToTheSameState_215(t *testing.T) { + oldBatch, oldYield := contentHashMigrationBatchSize, contentHashMigrationYield + contentHashMigrationBatchSize, contentHashMigrationYield = 1, 0 + t.Cleanup(func() { contentHashMigrationBatchSize, contentHashMigrationYield = oldBatch, oldYield }) + + ref, _ := hm215Open(t) + want := hm215Dump(t, ref.db) + + path := filepath.Join(t.TempDir(), "interrupted.db") + s := hm215Prepare(t, path, func(db *sql.DB) { hm215Seed(t, db) }) + seeded := hm215Count(t, s.db, `SELECT COUNT(*) FROM transmissions WHERE hash LIKE 'stale-%'`) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + // Stop after the fourth row was scanned: rows 11, 12, 13 and 20 are done, + // 30 is still stale. + contentHashMigrationHook = func(stage string, n int) { + if stage == "batch" && n == 4 { + cancel() + } + } + t.Cleanup(func() { contentHashMigrationHook = nil }) + s.StartContentHashMigration(ctx) + s.WaitForAsyncMigrations() + contentHashMigrationHook = nil + + if got := hm215Status(t, s); got == "done" { + t.Fatal("a run that was cancelled half way is recorded as done: the rest is never converted") + } + left := hm215Count(t, s.db, `SELECT COUNT(*) FROM transmissions WHERE hash LIKE 'stale-%'`) + if left == 0 || left >= seeded { + t.Fatalf("setup: %d of %d stale rows left, the cancel did not land half way", left, seeded) + } + if got := hm215Count(t, s.db, `SELECT COUNT(*) FROM pragma_foreign_key_check`); got != 0 { + t.Errorf("foreign_key_check reports %d violations at the cancel point", got) + } + if got := hm215Count(t, s.db, `SELECT COUNT(*) FROM observations WHERE transmission_id NOT IN (SELECT id FROM transmissions)`); got != 0 { + t.Errorf("%d orphan observations at the cancel point", got) + } + + s.StartContentHashMigration(context.Background()) + s.WaitForAsyncMigrations() + if got := hm215Status(t, s); got != "done" { + t.Fatalf("restarted run status = %q, want done", got) + } + if got := hm215Count(t, s.db, `SELECT COUNT(*) FROM transmissions WHERE hash LIKE 'stale-%'`); got != 0 { + t.Errorf("%d stale rows left after the restart", got) + } + if got := hm215Dump(t, s.db); got != want { + t.Errorf("the restarted run ends in a different state than an uninterrupted one:\n got:\n%s\n want:\n%s", got, want) + } +} + +// A row the scan saw can be gone (retention) or changed by the time its batch +// runs, because the scan is outside writerMu. Neither may fail the migration. +func TestContentHashMigration_RowsChangedSinceTheScanAreSkipped_215(t *testing.T) { + oldBatch, oldYield := contentHashMigrationBatchSize, contentHashMigrationYield + contentHashMigrationBatchSize, contentHashMigrationYield = 1000, 0 + t.Cleanup(func() { contentHashMigrationBatchSize, contentHashMigrationYield = oldBatch, oldYield }) + + path := filepath.Join(t.TempDir(), "race.db") + s := hm215Prepare(t, path, func(db *sql.DB) { hm215Seed(t, db) }) + fired := false + contentHashMigrationHook = func(stage string, n int) { + if stage != "scanned" || fired { + return + } + fired = true + // Retention removes row 30 and something rewrites the hash of row 11 + // after the scan and before the batch. + hm215Exec(t, s.db, `DELETE FROM observations WHERE transmission_id = 30`) + hm215Exec(t, s.db, `DELETE FROM transmissions WHERE id = 30`) + hm215Exec(t, s.db, `UPDATE transmissions SET hash = 'changed-11' WHERE id = 11`) + } + t.Cleanup(func() { contentHashMigrationHook = nil }) + s.StartContentHashMigration(context.Background()) + s.WaitForAsyncMigrations() + contentHashMigrationHook = nil + + if got := hm215Status(t, s); got != "done" { + t.Fatalf("status = %q: a row that changed since the scan failed the migration", got) + } + if got := hm215Count(t, s.db, `SELECT COUNT(*) FROM transmissions WHERE id = 30`); got != 0 { + t.Errorf("deleted row 30 is back") + } + var hash string + if err := s.db.QueryRow(`SELECT hash FROM transmissions WHERE id = 11`).Scan(&hash); err != nil || hash != "changed-11" { + t.Errorf("row 11 hash = %q (%v): a row whose hash changed since the scan must be left for the next run, not rewritten from stale data", hash, err) + } + // The next run picks the skipped row up. + if err := s.migrateContentHashes(context.Background(), s.db); err != nil { + t.Fatal(err) + } + if got := hm215Count(t, s.db, `SELECT COUNT(*) FROM transmissions WHERE hash NOT LIKE '________________'`); got != 0 { + t.Errorf("%d rows still carry a non-current hash after the next run", got) + } +} + +// A stop at shutdown is not a failure: it resumes at the next start, and the log +// must not read like one during a deploy restart. A real failure still does. +func TestAsyncMigration_CancelIsLoggedAsResumable_215(t *testing.T) { + var buf strings.Builder + prev := log.Writer() + log.SetOutput(&buf) + t.Cleanup(func() { log.SetOutput(prev) }) + + s := newTestStore(t) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := s.RunAsyncMigration(ctx, "cancel_probe_v1", func(ctx context.Context, d *sql.DB) error { return ctx.Err() }); err != nil { + t.Fatal(err) + } + if err := s.RunAsyncMigration(context.Background(), "failure_probe_v1", func(ctx context.Context, d *sql.DB) error { return fmt.Errorf("boom") }); err != nil { + t.Fatal(err) + } + s.WaitForAsyncMigrations() + out := buf.String() + if !strings.Contains(out, `"cancel_probe_v1" cancelled (will resume)`) { + t.Errorf("a cancelled migration is not logged as resumable:\n%s", out) + } + if strings.Contains(out, `"cancel_probe_v1" FAILED`) { + t.Errorf("a cancelled migration is logged as FAILED:\n%s", out) + } + if !strings.Contains(out, `"failure_probe_v1" FAILED: boom`) { + t.Errorf("a real failure is no longer logged as FAILED:\n%s", out) + } + if got, _ := s.AsyncMigrationStatus("cancel_probe_v1"); got == "done" { + t.Error("a cancelled migration is recorded as done") + } +} + +// Rows that merge keep what the duplicate knew: the earliest first_seen and +// every column the survivor has no value for. The survivor's own non-null values +// stay. A dropped observation (same observer and path) keeps the survivor's +// copy, as ingest does for a repeat reception. +func TestContentHashMigration_MergeKeepsEarliestFirstSeenAndFillsNulls_215(t *testing.T) { + s := hm215Reopen(t, filepath.Join(t.TempDir(), "rpi3.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'`) + ins := func(id int, raw, hash, first string, scope, channel, from interface{}) { + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json, last_seen, route_mask, scope_name, channel_hash, from_pubkey) + VALUES (?, ?, ?, ?, 1, 4, '{}', 1, 1, ?, ?, ?)`, id, raw, hash, first, scope, channel, from) + } + // Pair 1: the survivor (lowest id, later first_seen) has nothing set. + ins(50, hm215Raw(5), "stale-r-50", "2026-01-02T00:00:00Z", nil, nil, nil) + ins(51, hm215Raw(5), "stale-r-51", "2026-01-01T00:00:00Z", "#x", "ch", "pk") + // Pair 2: the survivor has its own values; the duplicate's differ. + ins(60, hm215Raw(6), "stale-r-60", "2026-01-01T00:00:00Z", "#keep", "keepch", "keeppk") + ins(61, hm215Raw(6), "stale-r-61", "2026-01-02T00:00:00Z", "#other", "otherch", "otherpk") + // The same observation in both rows of pair 1: the survivor's copy + // (SNR -10, later) stays although the duplicate's (SNR 5) is earlier. + hm215Exec(t, db, `INSERT INTO observations (transmission_id, observer_idx, direction, snr, rssi, path_json, timestamp) VALUES (50, ?, 'RX', -10.0, -100, '["aa"]', 1767312000)`, o1) + hm215Exec(t, db, `INSERT INTO observations (transmission_id, observer_idx, direction, snr, rssi, path_json, timestamp) VALUES (51, ?, 'RX', 5.0, -60, '["aa"]', 1767225600)`, o1) + }) + type row struct{ first, scope, channel, from string } + get := func(id int) row { + var r row + if err := s.db.QueryRow(`SELECT first_seen, COALESCE(scope_name, ''), COALESCE(channel_hash, ''), COALESCE(from_pubkey, '') FROM transmissions WHERE id = ?`, id).Scan(&r.first, &r.scope, &r.channel, &r.from); err != nil { + t.Fatalf("tx %d: %v", id, err) + } + return r + } + if got, want := get(50), (row{"2026-01-01T00:00:00Z", "#x", "ch", "pk"}); got != want { + t.Errorf("survivor 50 = %+v, want the earliest first_seen and the duplicate's values for what it lacked: %+v", got, want) + } + if got, want := get(60), (row{"2026-01-01T00:00:00Z", "#keep", "keepch", "keeppk"}); got != want { + t.Errorf("survivor 60 = %+v, want its own non-null values kept: %+v", got, want) + } + for _, id := range []int{51, 61} { + if hm215Count(t, s.db, `SELECT COUNT(*) FROM transmissions WHERE id = ?`, id) != 0 { + t.Errorf("duplicate %d was not deleted", id) + } + } + var snr float64 + if err := s.db.QueryRow(`SELECT snr FROM observations WHERE transmission_id = 50`).Scan(&snr); err != nil || snr != -10.0 { + t.Errorf("the surviving observation has SNR %v (%v), want the survivor's own -10 (documented rule)", snr, err) + } +} + +// Perf check, not a pass/fail test: with CORESCOPE_PERF_215= it times +// migrateContentHashes over 2000 stale rows in 1000 colliding pairs (the shape +// of the server's TestPerf_HashMigrateMerge_202) and over 5000 stale rows that +// do not collide (TestPerf_HashMigrateRehash_215), printing one RESULT line each. +func TestPerf_ContentHashMigration_215(t *testing.T) { + runs, _ := strconv.Atoi(os.Getenv("CORESCOPE_PERF_215")) + if runs <= 0 { + t.Skip("set CORESCOPE_PERF_215= to measure") + } + measure := func(name string, rows int, paired bool) { + var times []time.Duration + for r := 0; r < runs; r++ { + s := newTestStore(t) + db := s.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'`) + tx, err := db.Begin() + if err != nil { + t.Fatal(err) + } + for id := 1; id <= rows; id++ { + key := id + if paired { + key = (id + 1) / 2 + } + if _, err := tx.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json, last_seen, route_mask) + VALUES (?, ?, ?, '2026-01-01T00:00:00Z', 1, 4, '{}', ?, 1)`, id, fmt.Sprintf("0A00D69FD7A5A7475DB07337749AE61FA53A4788%04X", key), fmt.Sprintf("old-%d", id), id); err != nil { + t.Fatal(err) + } + for k := 0; k < 2; k++ { + if _, err := tx.Exec(`INSERT INTO observations (transmission_id, observer_idx, direction, path_json, timestamp) VALUES (?, ?, 'RX', ?, 1767225600)`, + id, o1, fmt.Sprintf(`["%02X","%02X"]`, id%256, k)); err != nil { + t.Fatal(err) + } + } + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + oldBatch, oldYield := contentHashMigrationBatchSize, contentHashMigrationYield + contentHashMigrationBatchSize, contentHashMigrationYield = 500, 0 + t0 := time.Now() + if err := s.migrateContentHashes(context.Background(), db); err != nil { + t.Fatal(err) + } + times = append(times, time.Since(t0)) + contentHashMigrationBatchSize, contentHashMigrationYield = oldBatch, oldYield + if got := hm215Count(t, db, `SELECT COUNT(*) FROM transmissions`); paired && got != rows/2 || !paired && got != rows { + t.Fatalf("%s: %d transmissions after the migration", name, got) + } + } + sort.Slice(times, func(i, j int) bool { return times[i] < times[j] }) + fmt.Printf("RESULT %s median_ns=%d\n", name, times[len(times)/2].Nanoseconds()) + } + measure("ingestor-hash-migrate-merge-2000", 2000, true) + measure("ingestor-hash-migrate-rehash-5000", 5000, false) +} + +// OpenStore does not start the migration: callers that seed their own rows +// (tests, tools) would have them rehashed under them. main starts it, once the +// ingest buffer is draining, and cancels it on shutdown, like the route_mask +// backfill. +func TestContentHashMigration_StartsAfterBufferReady_215(t *testing.T) { + path := filepath.Join(t.TempDir(), "start.db") + s, err := OpenStore(path) + if err != nil { + t.Fatal(err) + } + defer s.Close() + s.WaitForAsyncMigrations() + if hm215Count(t, s.db, `SELECT COUNT(*) FROM _async_migrations WHERE name LIKE '%content_hash%'`) != 0 { + t.Error("OpenStore scheduled the content-hash migration") + } + + src, err := os.ReadFile("main.go") + if err != nil { + t.Fatal(err) + } + m := string(src) + ready := strings.Index(m, "ingestBuffer.Ready()") + start := strings.Index(m, "store.StartContentHashMigration(") + shutdown := strings.Index(m, `log.Println("Shutting down...")`) + stop := strings.LastIndex(m, "stopContentHashMigration()") + disconnect := strings.LastIndex(m, "c.Disconnect(5000)") + if ready < 0 || start < 0 || shutdown < 0 || stop < 0 || disconnect < 0 { + t.Fatalf("markers not found: ready=%d start=%d shutdown=%d stop=%d disconnect=%d", ready, start, shutdown, stop, disconnect) + } + if !(ready < start) { + t.Errorf("StartContentHashMigration must come after ingestBuffer.Ready()") + } + if !(shutdown < stop && stop < disconnect) { + t.Errorf("shutdown must cancel the content-hash migration before disconnecting MQTT clients") + } +} diff --git a/cmd/ingestor/hash_migrate_round3_215_test.go b/cmd/ingestor/hash_migrate_round3_215_test.go new file mode 100644 index 000000000..95d0a0f06 --- /dev/null +++ b/cmd/ingestor/hash_migrate_round3_215_test.go @@ -0,0 +1,93 @@ +package main + +import ( + "database/sql" + "path/filepath" + "reflect" + "sort" + "testing" +) + +// The columns a merge fills with COALESCE are an explicit allow-list (#215 +// review, NEW-4). A nullable column added to transmissions later, whose NULL +// may mean "pending" and not "unknown", must not be filled from a duplicate +// unless someone adds it here on purpose. +func TestFillableColumns_IsAnExplicitAllowList_215(t *testing.T) { + want := []string{"channel_hash", "decoded_json", "from_pubkey", "payload_type", "payload_version", "route_type", "scope_name"} + s := hm215Prepare(t, filepath.Join(t.TempDir(), "fill.db"), func(db *sql.DB) {}) + got, err := fillableColumns(s.db) + if err != nil { + t.Fatal(err) + } + sort.Strings(got) + if !reflect.DeepEqual(got, want) { + t.Fatalf("fillableColumns on today's schema = %v, want exactly %v", got, want) + } + + // A nullable column nobody listed is not filled. + hm215Exec(t, s.db, `ALTER TABLE transmissions ADD COLUMN zz_pending_marker TEXT`) + got, err = fillableColumns(s.db) + if err != nil { + t.Fatal(err) + } + sort.Strings(got) + if !reflect.DeepEqual(got, want) { + t.Errorf("fillableColumns after adding a nullable column = %v, want it unchanged: %v", got, want) + } + + // A listed column the table does not have (an older schema) is left out, not + // an error. + hm215Exec(t, s.db, `ALTER TABLE transmissions DROP COLUMN zz_pending_marker`) + hm215Exec(t, s.db, `DROP INDEX IF EXISTS idx_tx_scope_name`) + hm215Exec(t, s.db, `ALTER TABLE transmissions DROP COLUMN scope_name`) + got, err = fillableColumns(s.db) + if err != nil { + t.Fatal(err) + } + sort.Strings(got) + if want2 := []string{"channel_hash", "decoded_json", "from_pubkey", "payload_type", "payload_version", "route_type"}; !reflect.DeepEqual(got, want2) { + t.Errorf("fillableColumns without scope_name = %v, want %v", got, want2) + } +} + +// NULL (not transport-scoped) and "" (transport-scoped, region unknown) are +// different values of scope_name (scopeNameForDB), and the merge keeps them +// apart: only NULL is filled from the duplicate (#215 review, NEW-4). The +// server's in-memory merge follows the same rule +// (TestHashMigrate_ScopeFillMatchesTheIngestorCoalesce_215). +func TestContentHashMigration_ScopeFillOnlyFillsNull_215(t *testing.T) { + s := hm215Reopen(t, filepath.Join(t.TempDir(), "scope.db"), func(db *sql.DB) { + ins := func(id int, raw, hash string, scope interface{}) { + hm215Exec(t, db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json, last_seen, route_mask, scope_name) + VALUES (?, ?, ?, '2026-01-01T00:00:00Z', 1, 4, '{}', 1, 1, ?)`, id, raw, hash, scope) + } + ins(70, hm215Raw(70), "stale-s-70", "") // transport-scoped, region unknown + ins(71, hm215Raw(70), "stale-s-71", "#x") + ins(72, hm215Raw(72), "stale-s-72", nil) // not transport-scoped + ins(73, hm215Raw(72), "stale-s-73", "") + ins(74, hm215Raw(74), "stale-s-74", nil) + ins(75, hm215Raw(74), "stale-s-75", "#y") + ins(76, hm215Raw(76), "stale-s-76", nil) + ins(77, hm215Raw(76), "stale-s-77", nil) + }) + get := func(id int) (v sql.NullString) { + if err := s.db.QueryRow(`SELECT scope_name FROM transmissions WHERE id = ?`, id).Scan(&v); err != nil { + t.Fatalf("tx %d: %v", id, err) + } + return v + } + for _, c := range []struct { + id int + want sql.NullString + why string + }{ + {70, sql.NullString{String: "", Valid: true}, `"" is a value: the duplicate's "#x" must not replace it`}, + {72, sql.NullString{String: "", Valid: true}, `NULL is filled with the duplicate's "", which stays "" and not NULL`}, + {74, sql.NullString{String: "#y", Valid: true}, "NULL is filled with the duplicate's value"}, + {76, sql.NullString{}, "NULL and NULL stay NULL"}, + } { + if got := get(c.id); got != c.want { + t.Errorf("survivor %d scope_name = %+v, want %+v: %s", c.id, got, c.want, c.why) + } + } +} diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index fbf585b43..4de249f02 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -284,6 +284,13 @@ func main() { defer stopRouteMaskBackfill() store.StartRouteMaskBackfill(routeMaskCtx) + // #215: rehash rows older than the current content-hash formula and merge + // the duplicates that creates, once (it is recorded as done). Same + // placement and shutdown as the route_mask backfill above. + contentHashCtx, stopContentHashMigration := context.WithCancel(context.Background()) + defer stopContentHashMigration() + store.StartContentHashMigration(contentHashCtx) + // Daily ticker for node retention retentionTicker := time.NewTicker(1 * time.Hour) go func() { @@ -461,6 +468,7 @@ func main() { log.Println("Shutting down...") stopRouteMaskBackfill() + stopContentHashMigration() retentionTicker.Stop() metricsRetentionTicker.Stop() if packetRetentionTicker != nil { diff --git a/cmd/server/hash_migrate.go b/cmd/server/hash_migrate.go index f51cbcb41..0066189f7 100644 --- a/cmd/server/hash_migrate.go +++ b/cmd/server/hash_migrate.go @@ -2,13 +2,97 @@ package main import ( "log" + "slices" + "strings" + "sync/atomic" "time" ) +// Content-hash migration, server half (#215). +// +// The DB half (rehash, merge the rows that collide, delete the duplicate) is +// done once by the ingestor (cmd/ingestor/hash_migrate.go): cmd/server is the +// read path and opens SQLite with mode=ro, so it must not write. Everything +// here happens in memory and is safe to repeat: a store loaded from a DB the +// ingestor has already converged finds nothing to do. +// +// What it still does matters for a server that loaded rows before the ingestor +// got to them. Packets whose hash differs from the current formula are rehashed +// in memory. When two of them collide, or one collides with a transmission that +// already has the hash, they are merged the way the ingestor merges the DB +// rows, so the two agree on which transmission is the survivor: the lowest ID. +// The duplicate's observations move to it (an observation the survivor already +// has, same observer and path, is dropped, as the DB's dedup index drops it), +// and the duplicate leaves every index. +// +// A merge keeps what the duplicate knew, as the ingestor does: the earliest +// first_seen (the survivor stays where it is in s.packets and the byPayloadType +// and byNode lists it is in, which are then put back in order), and the scope and +// route type the survivor has no value for. decoded_json and payload_type are not +// filled in memory: the content hash includes the payload type and the payload, +// so the duplicates agree on them, and filling would change the survivor's +// charge and its index membership. Observations that collide on observer and +// path keep the survivor's copy (see the ingestor); the earliest first_seen is +// what carries the earliest time. +// +// Scope fill. The ingestor fills a NULL scope_name and keeps a value, "" (a +// transport-scoped packet whose region is unknown) included. In memory NULL +// and "" are both "" (nullStrVal), and StoreTx has no spare byte for a flag +// (it stays 320 bytes), so a survivor whose ScopeName is "" is filled from the +// duplicate. The two agree for a NULL survivor, the usual one: a row written +// before scope_name existed, merged with the same packet heard since. They +// differ for a survivor with "" and a duplicate with a region, which needs the +// same packet heard twice, once with a region key configured that matched its +// transport code and once without (the keys were changed in between). There +// the DB keeps "" and memory shows the region until the server reloads, which +// the deploy plan requires after the migration anyway. Filling is the more +// informative of the two and is never wrong about a region that was matched. +// +// Known gap. A reception stored on a row that the in-memory merge has already +// removed is not picked up until the server reloads: IngestNewObservations +// finds no transmission for it, and the poller's cursor moves on. It happens in +// the window between this pass and the ingestor's batch for that row, so the +// server must be restarted once after the ingestor's migration is done (see +// the PR's deploy plan). A bounded loser-to-survivor id alias would close it, +// but it has to live until the ingestor finishes, which the read-only server +// cannot see, so it is left out. +// +// Relay indexes are the one thing not carried over. A merged observation's +// resolved relays are not re-derived for the survivor (that needs resolved_path +// from the DB, and the survivor is not re-indexed in memory); they come back at +// the next load, when the DB holds the merged row. Until then they are +// attributed to nothing, where before the merge they were attributed to a +// ghost transmission that counted the same packet twice. + +// hashRekeySweeps counts the passes over all of nodeHashes or byNode that a +// batch needed. Test observability only: with the resolved-pubkey index on, the +// keys of a transmission are found through the transmission and none is needed. +var hashRekeySweeps atomic.Int64 + +// lockHold accumulates how long the migration held the store's write lock, one +// sample per batch (NEW-1 of the #215 review: a batch that merges costs time +// that grows with the store, and staging should be able to read it off the log). +type lockHold struct { + max, total time.Duration + batches int +} + +func (h *lockHold) record(d time.Duration) { + h.max = max(h.max, d) + h.total += d + h.batches++ +} + +func millis(d time.Duration) float64 { return float64(d) / float64(time.Millisecond) } + +type hashUpdate struct { + tx *StoreTx + oldHash string + newHash string +} + // migrateContentHashesAsync recomputes content hashes in batches after the -// server is already serving HTTP. Packets whose hash changes are updated in -// both the DB and the in-memory byHash index. The migration is idempotent: -// once all hashes match the current formula it completes instantly. +// server is already serving HTTP. It only reads and rewrites in-memory state. func migrateContentHashesAsync(store *PacketStore, batchSize int, yieldDuration time.Duration) { defer func() { if r := recover(); r != nil { @@ -17,142 +101,475 @@ func migrateContentHashesAsync(store *PacketStore, batchSize int, yieldDuration store.hashMigrationComplete.Store(true) }() - // Snapshot the packet slice length under lock (packets only grow). + // Walk a snapshot, not the live slice: a merge removes elements from + // s.packets and eviction trims its head, either of which would shift an + // offset into it. Transmissions that are gone by the time their batch is + // applied are skipped. store.mu.RLock() - total := len(store.packets) + snapshot := slices.Clone(store.packets) store.mu.RUnlock() - // Keep evidence only for hashes that actually collide during this one - // migration run. The legacy migration retains duplicate in-memory rows, so - // a later batch must also update ghosts created by an earlier collision. - // This registry is bounded by collision members and is discarded when the - // migration returns. - type collisionEvidence struct { - mask uint8 - members map[*StoreTx]struct{} + if batchSize < 1 { + batchSize = 1 } - collisionEvidenceByHash := make(map[string]*collisionEvidence) + rehashed, merged := 0, 0 + var hold lockHold + indexesReady := false + for offset := 0; offset < len(snapshot); offset += batchSize { + end := min(offset+batchSize, len(snapshot)) - migrated := 0 - for offset := 0; offset < total; offset += batchSize { - end := offset + batchSize - if end > total { - end = total + store.mu.RLock() + updates := staleContentHashes(store, snapshot[offset:end]) + store.mu.RUnlock() + if len(updates) == 0 { + continue } - // Collect stale hashes in this batch under RLock. - type hashUpdate struct { - tx *StoreTx - oldHash string - newHash string + // A merge edits the subpath and path-hop indexes: let their + // background build finish first, or it could index a duplicate that + // was just removed. + if !indexesReady { + store.WaitIndexesReady(30 * time.Second) + indexesReady = true } - var updates []hashUpdate + store.mu.Lock() + held := time.Now() + r, m := store.applyContentHashUpdates(updates) + hold.record(time.Since(held)) + store.mu.Unlock() + rehashed += r + merged += m - store.mu.RLock() - for _, tx := range store.packets[offset:end] { - if tx.RawHex == "" { - continue - } - newHash := ComputeContentHash(tx.RawHex) - if newHash != tx.Hash { - updates = append(updates, hashUpdate{tx: tx, oldHash: tx.Hash, newHash: newHash}) - } - } - store.mu.RUnlock() + // Yield to let HTTP handlers run. + time.Sleep(yieldDuration) + } - if len(updates) == 0 { + if rehashed > 0 { + log.Printf("[hash-migrate] Rehashed %d transmissions in memory to the current formula, merged %d duplicates; max write-lock hold %.1f ms over %d batches (total %.1f ms)", + rehashed, merged, millis(hold.max), hold.batches, millis(hold.total)) + } +} + +// staleContentHashes returns the transmissions of batch whose hash differs from +// the current formula. Must hold s.mu (read). +func staleContentHashes(s *PacketStore, batch []*StoreTx) []hashUpdate { + var updates []hashUpdate + for _, tx := range batch { + if tx.RawHex == "" || s.byTxID[tx.ID] != tx { continue } - // A UNIQUE collision merges DB observations into one survivor, while - // this legacy migration intentionally keeps its existing in-memory - // cardinality/index behaviour. Track the survivor IDs so the observed - // path-width evidence can nevertheless be made consistent across every - // same-content in-memory row after the existing update loop. - collisionSurvivors := make(map[string][]int) - - // Write batch to DB in a single transaction. - dbTx, err := store.db.conn.Begin() - if err != nil { - log.Printf("[hash-migrate] begin tx: %v", err) - continue + if newHash := ComputeContentHash(tx.RawHex); newHash != tx.Hash { + updates = append(updates, hashUpdate{tx: tx, oldHash: tx.Hash, newHash: newHash}) } - stmt, err := dbTx.Prepare("UPDATE transmissions SET hash = ? WHERE id = ?") - if err != nil { - log.Printf("[hash-migrate] prepare: %v", err) - dbTx.Rollback() + } + return updates +} + +// applyContentHashUpdates rehashes the transmissions of updates and merges the +// ones that collide. Returns how many it rehashed and how many duplicates it +// merged away. Must hold s.mu for writing. +func (s *PacketStore) applyContentHashUpdates(updates []hashUpdate) (rehashed, merged int) { + // The batch was chosen under a read lock: drop what changed since + // (eviction, a concurrent rehash). + live := updates[:0] + for _, u := range updates { + if s.byTxID[u.tx.ID] != u.tx || u.tx.Hash != u.oldHash { continue } + live = append(live, u) + } + if len(live) == 0 { + return 0, 0 + } + keys := &nodeKeyFinder{s: s} + s.rekeyNodeHashes(live, keys) - for _, u := range updates { - if _, err := stmt.Exec(u.newHash, u.tx.ID); err != nil { - // UNIQUE constraint = two old hashes map to the same new hash (duplicate). - // Merge observations to the surviving tx, delete the duplicate. - log.Printf("[hash-migrate] tx %d collides — merging duplicate", u.tx.ID) - var survID int - if err2 := dbTx.QueryRow("SELECT id FROM transmissions WHERE hash = ?", u.newHash).Scan(&survID); err2 == nil { - dbTx.Exec("UPDATE observations SET transmission_id = ? WHERE transmission_id = ?", survID, u.tx.ID) - dbTx.Exec("DELETE FROM transmissions WHERE id = ?", u.tx.ID) - collisionSurvivors[u.newHash] = append(collisionSurvivors[u.newHash], survID) - } - } + m := newHashMerge() + m.keys = keys + for _, u := range live { + tx := u.tx + if s.byHash[u.oldHash] == tx { + delete(s.byHash, u.oldHash) } - stmt.Close() + tx.Hash = u.newHash + // The hash length is part of the charge (estimateStoreTxBytes). + s.trackedBytes += rechargeTx(tx) - if err := dbTx.Commit(); err != nil { - log.Printf("[hash-migrate] commit: %v", err) + holder := s.byHash[u.newHash] + if holder == nil || holder == tx { + s.byHash[u.newHash] = tx continue } + // A collision. The lowest ID survives, as in the ingestor's merge of + // the DB rows, so later polls by transmission_id find the same row. + winner, loser := tx, holder + if holder.ID < tx.ID { + winner, loser = holder, tx + } + s.byHash[u.newHash] = winner + s.moveObservations(winner, loser, m) + m.add(winner, loser) + } + merged = s.finishHashMerge(m) + return len(live), merged +} - // Update in-memory index under write lock. - store.mu.Lock() - // A merged duplicate stays in memory with its own observations (see - // collisionSurvivors above): the DB rows were re-parented, the - // in-memory ones are not, so trackedBytes keeps charging each - // observation to the one tx that holds it and eviction credits it - // once (#202). Copying them to the survivor here would credit them - // twice, as each of the two txs is evicted. A fix that removes the - // duplicate must move them (see - // TestHashMigrationMerge_EvictionCreditsEveryObservationOnce_202). - for _, u := range updates { - delete(store.byHash, u.oldHash) - u.tx.Hash = u.newHash - store.byHash[u.newHash] = u.tx - } - for newHash, survivorIDs := range collisionSurvivors { - evidence := collisionEvidenceByHash[newHash] - if evidence == nil { - evidence = &collisionEvidence{members: make(map[*StoreTx]struct{})} - collisionEvidenceByHash[newHash] = evidence +// nodeKeyFinder finds the pubkeys a transmission is indexed under in byNode and +// nodeHashes without walking those indexes. A transmission is indexed under its +// decoded pubkeys, the relays the path_json fallback resolved (recorded in +// fallbackByNode) and the relays its observations' resolved_path names; the +// last are recorded only as hashes, in the resolved-pubkey index, so they are +// mapped back to pubkeys through the pubkeys nodeHashes holds. That map costs +// one pass over the keys of nodeHashes (the nodes, not their transmissions), +// built on first use and only when a transmission of the batch has resolved +// relays. +type nodeKeyFinder struct { + s *PacketStore + byHash map[uint64][]string +} + +// usable reports whether the keys can be found this way. With the +// resolved-pubkey index off nothing records a transmission's resolved relays. +func (f *nodeKeyFinder) usable() bool { return f.s.useResolvedPathIndex } + +// keys returns the pubkeys tx is indexed under. Must hold s.mu. +func (f *nodeKeyFinder) keys(tx *StoreTx) []string { + var out []string + if decoded := tx.ParsedDecoded(); decoded != nil { + for _, field := range [...]string{"pubKey", "destPubKey", "srcPubKey"} { + if v, ok := decoded[field].(string); ok && v != "" { + out = append(out, v) } - addMember := func(tx *StoreTx) { - if tx == nil { - return - } - evidence.mask |= tx.pathHashSizeMask - evidence.members[tx] = struct{}{} + } + } + out = append(out, f.s.fallbackByNode[tx]...) + if rev := f.s.resolvedPubkeyReverse[tx.ID]; len(rev) > 0 { + if f.byHash == nil { + f.byHash = make(map[uint64][]string, len(f.s.nodeHashes)) + for pk := range f.s.nodeHashes { + h := resolvedPubkeyHash(pk) + f.byHash[h] = append(f.byHash[h], pk) } - for _, u := range updates { - if u.newHash == newHash { - addMember(u.tx) - } + } + for _, h := range rev { + out = append(out, f.byHash[h]...) + } + } + return out +} + +// rekeyNodeHashes renames the hash keys of s.nodeHashes. nodeHashes[pubkey] is +// the set of hashes of the transmissions byNode[pubkey] holds; a transmission +// that changes hash must change its key in every set it is in, or a later +// observation of it is indexed a second time and eviction, which removes the +// key by the transmission's current hash, leaves the old one behind. +// +// The keys of each transmission are found through the transmission +// (nodeKeyFinder), so a batch costs what the batch holds. Only with the +// resolved-pubkey index off, where a transmission's resolved relays are +// recorded nowhere, does it fall back to one pass over all of nodeHashes. +func (s *PacketStore) rekeyNodeHashes(updates []hashUpdate, keys *nodeKeyFinder) { + if !keys.usable() { + renames := make(map[string]string, len(updates)) + for _, u := range updates { + renames[u.oldHash] = u.newHash + } + s.rekeyNodeHashesSweep(renames) + return + } + for _, u := range updates { + for _, pk := range keys.keys(u.tx) { + if hashes := s.nodeHashes[pk]; hashes[u.oldHash] { + delete(hashes, u.oldHash) + hashes[u.newHash] = true } - for _, survivorID := range survivorIDs { - addMember(store.byTxID[survivorID]) + } + } +} + +// rekeyNodeHashesSweep is the fallback: one pass over all of nodeHashes. +func (s *PacketStore) rekeyNodeHashesSweep(renames map[string]string) { + hashRekeySweeps.Add(1) + var moved []string + for _, hashes := range s.nodeHashes { + moved = moved[:0] + for h := range hashes { + if _, ok := renames[h]; ok { + moved = append(moved, h) } - addMember(store.byHash[newHash]) - for member := range evidence.members { - member.pathHashSizeMask |= evidence.mask + } + for _, h := range moved { + delete(hashes, h) + hashes[renames[h]] = true + } + } +} + +// moveObservations moves loser's observations to winner, dropping the ones +// winner already has (same observer and path). The observations keep their +// place in byObsID and byObserver and their charge; a dropped one is credited +// here. loser keeps its observations list until finishHashMerge has removed it +// from the path-hop index, which reads it; the index entries of loser itself +// are removed there. +func (s *PacketStore) moveObservations(winner, loser *StoreTx, m *hashMerge) { + if winner.obsKeys == nil { + winner.obsKeys = make(map[string]bool) + winner.observerSet = make(map[string]bool) + } + for _, o := range loser.Observations { + dk := o.ObserverID + "|" + o.PathJSON + if winner.obsKeys[dk] { + delete(s.byObsID, o.ID) + s.totalObs-- + s.trackedBytes -= estimateStoreObsBytes(o) + m.dropped[o.ID] = struct{}{} + if o.ObserverID != "" { + m.droppedObservers[o.ObserverID] = struct{}{} } + continue } - store.mu.Unlock() + o.TransmissionID = winner.ID + winner.Observations = append(winner.Observations, o) + winner.obsKeys[dk] = true + if o.ObserverID != "" && !winner.observerSet[o.ObserverID] { + winner.observerSet[o.ObserverID] = true + winner.UniqueObserverCount++ + } + winner.ObservationCount++ + if o.Timestamp > winner.LatestSeen { + winner.LatestSeen = o.Timestamp + } + } + if loser.LatestSeen > winner.LatestSeen { + winner.LatestSeen = loser.LatestSeen + } + if loser.FirstSeen != "" && loser.FirstSeen < winner.FirstSeen { + winner.FirstSeen = loser.FirstSeen + m.firstSeenMoved[winner] = struct{}{} + } + // The ingestor's COALESCE fills a NULL scope_name and keeps any value, "" + // (transport-scoped, region unknown) included. StoreTx.ScopeName cannot tell + // the two apart (nullStrVal reads NULL as ""), and StoreTx has no padding left + // for a flag, so this fills on "": see "Scope fill" + // in the file header. + if winner.ScopeName == "" { + winner.ScopeName = loser.ScopeName + } + if winner.RouteType == nil { + winner.RouteType = loser.RouteType + } + winner.pathHashSizeMask |= loser.pathHashSizeMask + if loser.routeMaskKnown { + winner.routeMask |= loser.routeMask + winner.routeMaskKnown = true + } +} - migrated += len(updates) +// hashMerge collects what one batch of collisions removed and touched. +type hashMerge struct { + losers []*StoreTx + loserSet map[*StoreTx]bool + winners map[*StoreTx]string // survivor → its best path before the first merge + dropped map[int]struct{} // observation ids dropped as duplicates + droppedObservers map[string]struct{} + firstSeenMoved map[*StoreTx]struct{} // survivors that took an earlier first_seen + keys *nodeKeyFinder +} - // Yield to let HTTP handlers run. - time.Sleep(yieldDuration) +func newHashMerge() *hashMerge { + return &hashMerge{ + loserSet: make(map[*StoreTx]bool), + winners: make(map[*StoreTx]string), + dropped: make(map[int]struct{}), + droppedObservers: make(map[string]struct{}), + firstSeenMoved: make(map[*StoreTx]struct{}), + } +} + +// add records a merge. winner's best path is saved before moveObservations' +// caller lets it change; add must run after moveObservations, which does not +// touch PathJSON, so the saved path is still the pre-merge one. +func (m *hashMerge) add(winner, loser *StoreTx) { + if _, ok := m.winners[winner]; !ok { + m.winners[winner] = winner.PathJSON + } + m.losers = append(m.losers, loser) + m.loserSet[loser] = true +} + +// finishHashMerge removes every merged-away transmission from every index and +// from trackedBytes, then refreshes the survivors (best observation, charge, +// path-derived indexes). Returns the number of duplicates removed. +func (s *PacketStore) finishHashMerge(m *hashMerge) int { + if len(m.losers) == 0 { + return 0 + } + affectedPayloadTypes := make(map[int]struct{}) + // The pubkeys the duplicates are indexed under, read before their records + // (fallback relays, resolved-pubkey index) are removed below. + var nodePks map[string]struct{} + if m.keys != nil && m.keys.usable() { + nodePks = make(map[string]struct{}) + for _, l := range m.losers { + for _, pk := range m.keys.keys(l) { + nodePks[pk] = struct{}{} + } + } + } + for _, l := range m.losers { + delete(s.byTxID, l.ID) + // Observations moved or were credited above: this is the tx alone. + s.trackedBytes -= s.txChargedBytes(l) + s.untrackAdvertPubkey(l) + s.removeFromResolvedPubkeyIndex(l.ID) + if removeTxFromSubpathIndexFull(s.spIndex, s.spTxIndex, l) { + s.spTotalPaths-- + } + delete(s.pathHopResolved, l) + delete(s.fallbackByNode, l) + if l.PayloadType != nil { + affectedPayloadTypes[*l.PayloadType] = struct{}{} + } + } + // The resolved relay keys of a duplicate in byPathHop are found through the + // hops of its observed paths, so it must still hold its observations here: + // they were moved, not detached. A duplicate that was itself a survivor + // earlier in the batch holds the ones it took over too. + evictFromPathHopIndex(s.byPathHop, m.loserSet) + for _, l := range m.losers { + l.Observations = nil + l.ObservationCount = 0 + } + s.removeTxsFromByNode(m.loserSet, nodePks) + for pt := range affectedPayloadTypes { + kept := slices.DeleteFunc(s.byPayloadType[pt], func(t *StoreTx) bool { return m.loserSet[t] }) + if len(kept) == 0 { + delete(s.byPayloadType, pt) + } else { + s.byPayloadType[pt] = kept + } + } + for obsID := range m.droppedObservers { + kept := slices.DeleteFunc(s.byObserver[obsID], func(o *StoreObs) bool { + _, dropped := m.dropped[o.ID] + return dropped + }) + if len(kept) == 0 { + delete(s.byObserver, obsID) + } else { + s.byObserver[obsID] = kept + } + } + s.compactDistIndex(m.loserSet) + s.packets = slices.DeleteFunc(s.packets, func(t *StoreTx) bool { return m.loserSet[t] }) + s.reorderAfterFirstSeenMoved(m) + + // A survivor now holds observations it did not have: its best one may be a + // different path, which moves it in the path-derived indexes, and its + // charge follows. A survivor that was itself merged into another one later + // in the batch is gone. + var changed []*StoreTx + for w, oldPath := range m.winners { + if m.loserSet[w] { + continue + } + pickBestObservation(w) + s.trackedBytes += rechargeTx(w) + if w.PathJSON != oldPath { + s.reindexTxPath(w, oldPath) + changed = append(changed, w) + } + } + if len(changed) > 0 { + s.updateDistanceIndexForTxs(changed) + } + + s.invalidateRelayStatsCache() + s.invalidateCachesFor(cacheInvalidation{eviction: true}) + s.hashSizeInfoMu.Lock() + s.hashSizeInfoCache = nil + s.hashSizeInfoMu.Unlock() + return len(m.losers) +} + +// removeTxsFromByNode removes the given transmissions from the byNode lists +// and rebuilds nodeHashes for each list it changed, from the transmissions the +// list still holds: a duplicate and its survivor share one hash after the +// merge, so the key cannot be deleted by hash without also dropping the +// survivor's, and a key the duplicate alone held must go. pks are the pubkeys +// the transmissions are indexed under (nodeKeyFinder); nil means unknown, and +// every list is looked at. +func (s *PacketStore) removeTxsFromByNode(remove map[*StoreTx]bool, pks map[string]struct{}) { + one := func(pk string) { + list := s.byNode[pk] + hit := false + for _, t := range list { + if remove[t] { + hit = true + break + } + } + if !hit { + return + } + kept := slices.DeleteFunc(list, func(t *StoreTx) bool { return remove[t] }) + if len(kept) == 0 { + delete(s.byNode, pk) + delete(s.nodeHashes, pk) + return + } + s.byNode[pk] = kept + hashes := make(map[string]bool, len(kept)) + for _, t := range kept { + hashes[t.Hash] = true + } + s.nodeHashes[pk] = hashes + } + if pks != nil { + for pk := range pks { + one(pk) + } + return + } + hashRekeySweeps.Add(1) + for pk := range s.byNode { + one(pk) } +} - if migrated > 0 { - log.Printf("[hash-migrate] Migrated %d content hashes to new formula", migrated) +// reorderAfterFirstSeenMoved puts the lists that are ordered by first_seen back +// in order after a survivor took an earlier one: s.packets (eviction cuts from +// its head), the byPayloadType lists and the byNode lists the survivor is in. +// Stable, so equal times keep their load order. Only runs when it happened. +func (s *PacketStore) reorderAfterFirstSeenMoved(m *hashMerge) { + if len(m.firstSeenMoved) == 0 { + return + } + byFirstSeen := func(a, b *StoreTx) int { return strings.Compare(a.FirstSeen, b.FirstSeen) } + slices.SortStableFunc(s.packets, byFirstSeen) + types := make(map[int]struct{}) + pks := make(map[string]struct{}) + for w := range m.firstSeenMoved { + if w.PayloadType != nil { + types[*w.PayloadType] = struct{}{} + } + if m.keys != nil && m.keys.usable() { + for _, pk := range m.keys.keys(w) { + pks[pk] = struct{}{} + } + } + } + for pt := range types { + slices.SortStableFunc(s.byPayloadType[pt], byFirstSeen) + } + if m.keys != nil && m.keys.usable() { + for pk := range pks { + slices.SortStableFunc(s.byNode[pk], byFirstSeen) + } + return + } + hashRekeySweeps.Add(1) + for _, list := range s.byNode { + slices.SortStableFunc(list, byFirstSeen) } } diff --git a/cmd/server/hash_migrate_readonly_215_test.go b/cmd/server/hash_migrate_readonly_215_test.go new file mode 100644 index 000000000..dfd4ccfff --- /dev/null +++ b/cmd/server/hash_migrate_readonly_215_test.go @@ -0,0 +1,774 @@ +package main + +import ( + "bytes" + "context" + "database/sql" + "database/sql/driver" + "fmt" + "log" + "os" + "path/filepath" + "reflect" + "sort" + "strings" + "sync" + "testing" + "time" +) + +// Issue #215: the content-hash migration used to UPDATE/DELETE transmissions +// and observations on the server's mode=ro handle. Every write failed, was +// logged as a collision, and left in-memory duplicates ("ghosts") sharing one +// hash. The DB half of the migration now lives in the ingestor; the server +// only rehashes and merges in memory. These tests pin that. + +const hm215Observers = 4 + +// Relay paths: hop i of an observation is the first byte of resolved pubkey i +// (acct113PK(k) starts with byte k), as the store assumes when it sweeps +// byPathHop by hop prefix. + +// hm215Raw is a distinct, decodable raw packet per key. ComputeContentHash +// differs for every key, and never equals the "stale-*" hashes below. +func hm215Raw(key int) string { + return fmt.Sprintf("0A00D69FD7A5A7475DB07337749AE61FA53A4788E9%02X", key) +} + +func hm215Time(offset time.Duration) string { + return time.Now().UTC().Add(offset).Format(time.RFC3339) +} + +type hm215DB struct { + conn *sql.DB + path string + obs int +} + +func hm215Create(t testing.TB) *hm215DB { + t.Helper() + path := filepath.Join(t.TempDir(), "hm215.db") + acct113CreateDBOpts(t, path, 0, 0, 0) // schema + observers obs0..obs3, no rows + conn, err := sql.Open("sqlite", path+"?_journal_mode=WAL") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + return &hm215DB{conn: conn, path: path} +} + +func (d *hm215DB) tx(t testing.TB, id int, raw, hash, firstSeen string, payloadType int, decoded string) { + t.Helper() + if _, err := d.conn.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json) + VALUES (?, ?, ?, ?, 1, ?, 1, ?)`, id, raw, hash, firstSeen, payloadType, decoded); err != nil { + t.Fatal(err) + } +} + +func (d *hm215DB) observation(t testing.TB, txID, observer int, path, resolved, ts string) { + t.Helper() + d.obs++ + var rp interface{} + if resolved != "" { + rp = resolved + } + if _, err := d.conn.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, snr, rssi, score, path_json, timestamp, resolved_path) + VALUES (?, ?, ?, ?, 'RX', -5.0, -90.0, 5, ?, ?, ?)`, + d.obs, txID, fmt.Sprintf("obs%d", observer%hm215Observers), fmt.Sprintf("Observer%d", observer%hm215Observers), path, ts, rp); err != nil { + t.Fatal(err) + } +} + +func hm215Advert(pk string) string { + return fmt.Sprintf(`{"type":"ADVERT","pubKey":%q,"name":"n"}`, pk) +} + +// hm215Ballast is a recent, correctly hashed transmission that stays live. It +// shares relay keys with the stale rows so a shared byNode list is exercised. +func (d *hm215DB) ballast(t testing.TB) { + raw := hm215Raw(250) + ts := hm215Time(-time.Hour) + d.tx(t, 1000, raw, ComputeContentHash(raw), ts, 4, hm215Advert(acct113PK(240))) + // PK(18) is also the relay of the second row of every stale group, which + // the survivor (the first row) does not have: removing that row leaves the + // ballast alone in byNode[PK(18)], and its nodeHashes entry must then hold + // the ballast's hash only. + d.observation(t, 1000, 0, `["05","11","12"]`, fmt.Sprintf(`[%q,%q,%q]`, acct113PK(5), acct113PK(17), acct113PK(18)), ts) + d.observation(t, 1000, 1, `["aa"]`, "", ts) // fallback relay +} + +// hm215Stale adds the stale rows: groups of `dups` duplicate rows that share a +// raw packet (so one content hash), and `singles` rows that migrate without a +// collision. Group g runs its first_seen order with the ids for odd g and +// against them for even g, so both "the first row processed survives" and +// "a later row takes over" happen. Observation (obs0, ["aa"]) is identical +// across the rows of a group: the DB would reject it as a duplicate, so the +// merge must keep it once. +func (d *hm215DB) stale(t testing.TB, groups, dups, singles int) { + old := -72 * time.Hour + for g := 1; g <= groups; g++ { + raw := hm215Raw(g) + pk := acct113PK(100 + g) + for k := 1; k <= dups; k++ { + id := g*10 + k + order := k + if g%2 == 0 { + order = dups - k + 1 + } + ts := hm215Time(old + time.Duration(order)*time.Second) + d.tx(t, id, raw, fmt.Sprintf("stale-%d-%d", g, k), ts, 4, hm215Advert(pk)) + d.observation(t, id, 0, `["aa"]`, "", ts) // same in every row: fallback relay + // Observer 0 again, with a path of its own: not a duplicate of the + // one above or of the other rows' (the dedup key is observer AND + // path), so the merge must keep every one of them. + d.observation(t, id, 0, fmt.Sprintf(`["%02x"]`, 32+k), "", ts) + path, resolved := fmt.Sprintf(`["05","%02x"]`, 16+k), fmt.Sprintf(`[%q,%q]`, acct113PK(5), acct113PK(16+k)) + if k == dups { + // The last row has the longest path: once its observations + // move to the survivor, the survivor's best path changes. + path = fmt.Sprintf(`["05","%02x","11"]`, 16+k) + resolved = fmt.Sprintf(`[%q,%q,%q]`, acct113PK(5), acct113PK(16+k), acct113PK(17)) + } + d.observation(t, id, k, path, resolved, ts) + } + } + for s := 1; s <= singles; s++ { + id := 500 + s + ts := hm215Time(old + time.Duration(s)*time.Second) + d.tx(t, id, hm215Raw(100+s), fmt.Sprintf("stale-single-%d", s), ts, 4, hm215Advert(acct113PK(200+s))) + d.observation(t, id, s, fmt.Sprintf(`["05","%02x"]`, 30+s), fmt.Sprintf(`[%q,%q]`, acct113PK(5), acct113PK(30+s)), ts) + } +} + +func hm215Open(t testing.TB, d *hm215DB) *PacketStore { return hm215OpenIdx(t, d, true) } + +// hm215OpenIdx opens the store with the resolved-pubkey index on or off (the +// feature flag; off is the conservative path). +func hm215OpenIdx(t testing.TB, d *hm215DB, resolvedIndex bool) *PacketStore { + t.Helper() + db, err := OpenDB(d.path) + if err != nil { + t.Fatal(err) + } + cfg := &PacketStoreConfig{} + store := NewPacketStore(db, cfg) + store.useResolvedPathIndex = resolvedIndex + leak202SeedNodes(store, true) // hop "aa" resolves to a unique node + if err := store.Load(); err != nil { + t.Fatal(err) + } + if !store.WaitIndexesReady(30 * time.Second) { + t.Fatal("background index builds did not finish") + } + t.Cleanup(func() { db.conn.Close() }) + return store +} + +// hm215Snapshot is the part of the store a leftover would show up in. +// spTotalPaths is left out on purpose: eviction never decrements it (master +// behaviour, unrelated to this change), so it cannot return to a baseline. +type hm215Snapshot struct { + Packets int + TrackedBytes int64 + ByNode map[string][]int + NodeHashes map[string][]string + ByPayloadType map[int]int + ByObserver map[string]int + ByObsID int + ByTxID int + ByHash int + PathHopRecs int + FallbackRecs int + ResolvedIdx int + TotalObs int + AdvertKeys map[string]int + ByPathHop map[string]int + SpIndex map[string]int + SpTxIndex map[string]int +} + +func hm215Snap(s *PacketStore) hm215Snapshot { + s.mu.RLock() + defer s.mu.RUnlock() + sn := hm215Snapshot{ + Packets: len(s.packets), TrackedBytes: s.trackedBytes, + ByNode: map[string][]int{}, NodeHashes: map[string][]string{}, + ByPayloadType: map[int]int{}, ByObserver: map[string]int{}, + ByObsID: len(s.byObsID), ByTxID: len(s.byTxID), ByHash: len(s.byHash), + PathHopRecs: len(s.pathHopResolved), FallbackRecs: len(s.fallbackByNode), + ResolvedIdx: len(s.resolvedPubkeyIndex), TotalObs: s.totalObs, + AdvertKeys: map[string]int{}, + ByPathHop: map[string]int{}, SpIndex: map[string]int{}, SpTxIndex: map[string]int{}, + } + for pk, list := range s.byNode { + for _, tx := range list { + sn.ByNode[pk] = append(sn.ByNode[pk], tx.ID) + } + sort.Ints(sn.ByNode[pk]) + } + for pk, hashes := range s.nodeHashes { + for h := range hashes { + sn.NodeHashes[pk] = append(sn.NodeHashes[pk], h) + } + sort.Strings(sn.NodeHashes[pk]) + } + for pt, list := range s.byPayloadType { + sn.ByPayloadType[pt] = len(list) + } + for o, list := range s.byObserver { + sn.ByObserver[o] = len(list) + } + for pk, n := range s.advertPubkeys { + sn.AdvertKeys[pk] = n + } + for k, list := range s.byPathHop { + sn.ByPathHop[k] = len(list) + } + for k, n := range s.spIndex { + sn.SpIndex[k] = n + } + for k, list := range s.spTxIndex { + sn.SpTxIndex[k] = len(list) + } + return sn +} + +func hm215Migrate(s *PacketStore, batch int) string { + var buf bytes.Buffer + prev := log.Writer() + log.SetOutput(&buf) + defer log.SetOutput(prev) + migrateContentHashesAsync(s, batch, 0) + return buf.String() +} + +// stmtLog records every statement a connection prepares or every transaction it +// begins. database/sql falls back to Prepare/Begin for a driver connection that +// implements nothing else, so wrapping driver.Conn sees all of them. +type stmtLog struct { + mu sync.Mutex + stmts []string +} + +func (l *stmtLog) add(q string) { + l.mu.Lock() + l.stmts = append(l.stmts, q) + l.mu.Unlock() +} + +func (l *stmtLog) all() []string { + l.mu.Lock() + defer l.mu.Unlock() + return append([]string(nil), l.stmts...) +} + +type recordingConn struct { + driver.Conn + log *stmtLog +} + +func (c *recordingConn) Prepare(q string) (driver.Stmt, error) { + c.log.add(q) + return c.Conn.Prepare(q) +} + +func (c *recordingConn) Begin() (driver.Tx, error) { //nolint:staticcheck // the fallback database/sql uses + c.log.add("BEGIN") + return c.Conn.Begin() //nolint:staticcheck +} + +type recordingConnector struct { + drv driver.Driver + dsn string + log *stmtLog +} + +func (c *recordingConnector) Connect(context.Context) (driver.Conn, error) { + conn, err := c.drv.Open(c.dsn) + if err != nil { + return nil, err + } + return &recordingConn{Conn: conn, log: c.log}, nil +} + +func (c *recordingConnector) Driver() driver.Driver { return c.drv } + +// hm215OpenRecording opens the store the way the server does (mode=ro, same +// DSN as OpenDB) on a connection that records every statement. +func hm215OpenRecording(t testing.TB, d *hm215DB) (*PacketStore, *stmtLog) { + t.Helper() + dsn := fmt.Sprintf("file:%s?mode=ro&_journal_mode=WAL&_busy_timeout=5000", d.path) + probe, err := sql.Open("sqlite", dsn) + if err != nil { + t.Fatal(err) + } + drv := probe.Driver() + probe.Close() + rec := &stmtLog{} + conn := sql.OpenDB(&recordingConnector{drv: drv, dsn: dsn, log: rec}) + conn.SetMaxOpenConns(4) + t.Cleanup(func() { conn.Close() }) + db := &DB{conn: conn, path: d.path, schemaHealerStop: make(chan struct{})} + db.detectSchema() + store := NewPacketStore(db, &PacketStoreConfig{}) + leak202SeedNodes(store, true) + if err := store.Load(); err != nil { + t.Fatal(err) + } + if !store.WaitIndexesReady(30 * time.Second) { + t.Fatal("background index builds did not finish") + } + return store, rec +} + +// A probe on a connection that records statements, not only a log line: the +// migration may run reads, but it must not prepare a single write statement or +// begin a transaction, whether or not the read-only handle would reject it. +func TestHashMigrate_IssuesNoWriteStatements_215(t *testing.T) { + d := hm215Create(t) + d.ballast(t) + d.stale(t, 2, 3, 2) + s, rec := hm215OpenRecording(t, d) + loaded := len(rec.all()) + if loaded == 0 { + t.Fatal("setup: the probe recorded no statements from the load, so it would see nothing of the migration either") + } + hm215Migrate(s, 2) + if len(s.byHash) != 5 { + t.Fatalf("setup: the migration did not run: %d transmissions by hash, want 5", len(s.byHash)) + } + for _, q := range rec.all() { + head := strings.ToUpper(strings.TrimSpace(q)) + if strings.HasPrefix(head, "SELECT") || strings.HasPrefix(head, "PRAGMA") || strings.HasPrefix(head, "WITH") { + continue + } + t.Errorf("the server prepared a non-read statement: %q", q) + } +} + +// The migration must not try to write: no statement may fail on the mode=ro +// handle, so nothing may be logged as a collision or an error, and the DB +// stays exactly as the ingestor left it. +func TestHashMigrate_NeverWritesOnTheReadOnlyHandle_215(t *testing.T) { + d := hm215Create(t) + d.ballast(t) + d.stale(t, 2, 3, 2) + s := hm215Open(t, d) + + out := hm215Migrate(s, 2) + for _, bad := range []string{"collides", "begin tx", "prepare", "commit", "readonly", "read-only"} { + if strings.Contains(out, bad) { + t.Errorf("the migration logged %q, which means it tried to write on the read-only handle:\n%s", bad, out) + } + } + var rows int + if err := d.conn.QueryRow(`SELECT COUNT(*) FROM transmissions WHERE hash LIKE 'stale-%'`).Scan(&rows); err != nil { + t.Fatal(err) + } + if want := 2*3 + 2; rows != want { + t.Fatalf("%d stale rows left in the DB, want all %d untouched (the ingestor owns that rewrite)", rows, want) + } +} + +// After the migration there is exactly one transmission per content hash, it +// is the lowest id (the ingestor keeps the same one), it holds every +// observation once, and every index agrees. +func TestHashMigrate_MergesDuplicatesInMemory_215(t *testing.T) { + for _, batch := range []int{1, 2, 100} { + t.Run(fmt.Sprintf("batch%d", batch), func(t *testing.T) { + d := hm215Create(t) + d.ballast(t) + d.stale(t, 2, 3, 2) + s := hm215Open(t, d) + hm215Migrate(s, batch) + + s.mu.RLock() + defer s.mu.RUnlock() + // 1 ballast + 2 merged groups + 2 singles. + if len(s.packets) != 5 || len(s.byTxID) != 5 || len(s.byHash) != 5 { + t.Fatalf("packets/byTxID/byHash = %d/%d/%d, want 5 each", len(s.packets), len(s.byTxID), len(s.byHash)) + } + if got := len(s.byPayloadType[4]); got != 5 { + t.Errorf("byPayloadType[ADVERT] holds %d transmissions, want 5 (ballast, 2 merged groups, 2 singles)", got) + } + seen := map[string]int{} + for _, tx := range s.packets { + seen[tx.Hash]++ + if s.byTxID[tx.ID] != tx || s.byHash[tx.Hash] != tx { + t.Errorf("tx %d is not what byTxID/byHash hold for it", tx.ID) + } + if want := ComputeContentHash(tx.RawHex); tx.Hash != want { + t.Errorf("tx %d hash = %s, want %s", tx.ID, tx.Hash, want) + } + } + for h, n := range seen { + if n != 1 { + t.Errorf("hash %s is held by %d transmissions in s.packets", h, n) + } + } + for g := 1; g <= 2; g++ { + tx := s.byHash[ComputeContentHash(hm215Raw(g))] + if tx == nil || tx.ID != g*10+1 { + t.Fatalf("group %d survivor = %v, want the lowest id %d", g, tx, g*10+1) + } + // (obs0,["aa"]) is shared by the 3 rows; each row has one + // observation of its own observer and one of observer 0 with a + // path of its own. + if want := 1 + 3 + 3; len(tx.Observations) != want || tx.ObservationCount != want { + t.Errorf("group %d survivor holds %d observations (count %d), want %d", g, len(tx.Observations), tx.ObservationCount, want) + } + keys := map[string]bool{} + for _, o := range tx.Observations { + if o.TransmissionID != tx.ID { + t.Errorf("observation %d still names tx %d, want %d", o.ID, o.TransmissionID, tx.ID) + } + k := o.ObserverID + "|" + o.PathJSON + if keys[k] { + t.Errorf("observation %s is held twice", k) + } + keys[k] = true + if s.byObsID[o.ID] != o { + t.Errorf("byObsID[%d] is not the observation of tx %d", o.ID, tx.ID) + } + } + if int(tx.UniqueObserverCount) != len(tx.observerSet) || len(tx.observerSet) != 4 { + t.Errorf("group %d survivor has %d unique observers (%d in the set), want 4", g, tx.UniqueObserverCount, len(tx.observerSet)) + } + for k := 1; k <= 3; k++ { + if loser := s.byTxID[g*10+k]; k != 1 && loser != nil { + t.Errorf("duplicate tx %d is still in byTxID", g*10+k) + } + } + pk := acct113PK(100 + g) + if list := s.byNode[pk]; len(list) != 1 || list[0] != tx { + t.Errorf("byNode[%s...] = %d entries, want exactly the survivor", pk[:4], len(list)) + } + if got := s.nodeHashes[pk]; len(got) != 1 || !got[tx.Hash] { + t.Errorf("nodeHashes[%s...] = %v, want exactly the new hash", pk[:4], got) + } + if n := s.advertPubkeys[pk]; n != 1 { + t.Errorf("advertPubkeys[%s...] = %d, want 1 (one advert transmission)", pk[:4], n) + } + } + for pk, hashes := range s.nodeHashes { + for h := range hashes { + if strings.HasPrefix(h, "stale-") { + t.Errorf("nodeHashes[%s...] still holds the old hash %s", pk[:4], h) + } + } + } + // The subpath total follows the merge: it counts transmissions with a + // path of 2+ hops, and a merged-away duplicate no longer counts. + multiHop := 0 + for _, tx := range s.packets { + if len(txGetParsedPath(tx)) >= 2 { + multiHop++ + } + } + if s.spTotalPaths != multiHop { + t.Errorf("spTotalPaths = %d, want %d (transmissions with 2+ hop paths)", s.spTotalPaths, multiHop) + } + // Every observation of the DB survives, except the 4 identical + // (obs0, ["aa"]) ones that merge into 1 per group. + if want := 2 /*ballast*/ + 2*(1+3+3) + 2; s.totalObs != want { + t.Errorf("totalObs = %d, want %d", s.totalObs, want) + } + if n := len(s.byObsID); n != s.totalObs { + t.Errorf("byObsID holds %d observations, totalObs = %d", n, s.totalObs) + } + }) + } + t.Run("QueryPackets", func(t *testing.T) { + d := hm215Create(t) + d.ballast(t) + d.stale(t, 2, 3, 2) + s := hm215Open(t, d) + if before := s.QueryPackets(PacketQuery{Limit: 100}).Total; before != 1+6+2 { + t.Fatalf("setup: %d transmissions before the migration, want 9", before) + } + hm215Migrate(s, 3) + if got := s.QueryPackets(PacketQuery{Limit: 100}).Total; got != 5 { + t.Fatalf("QueryPackets total = %d after the merge, want one per hash (5)", got) + } + if got := s.QueryPackets(PacketQuery{Limit: 100, Hash: ComputeContentHash(hm215Raw(1))}).Total; got != 1 { + t.Fatalf("QueryPackets by hash = %d, want 1", got) + } + }) +} + +// trackedBytes, every per-transmission charge and every index must account for +// exactly what the store holds after the migration. +func TestHashMigrate_AccountingIsExactAfterTheMerge_215(t *testing.T) { + d := hm215Create(t) + d.ballast(t) + d.stale(t, 3, 3, 3) + s := hm215Open(t, d) + acct113Check(t, s, "before the migration") // stale hashes are a different length: charges are stale only after a rehash + hm215Migrate(s, 2) + acct113Check(t, s, "after the migration") +} + +// After the migration and the eviction of every migrated row the store is +// what it would be had those rows never existed. Compared against a second +// store that never held them, so a double credit cannot hide behind the clamp +// at zero, and a leftover (a node key, a relay record, an advert refcount) +// shows up as a difference. +func TestHashMigrate_EvictionReturnsToBaseline_215(t *testing.T) { + for _, resolvedIndex := range []bool{true, false} { + for _, batch := range []int{1, 3, 1000} { + t.Run(fmt.Sprintf("resolvedIndex=%v/batch%d", resolvedIndex, batch), func(t *testing.T) { + base := hm215Create(t) + base.ballast(t) + baseline := hm215OpenIdx(t, base, resolvedIndex) + want := hm215Snap(baseline) + + d := hm215Create(t) + d.ballast(t) + d.stale(t, 3, 3, 3) + s := hm215OpenIdx(t, d, resolvedIndex) + hm215Migrate(s, batch) + s.mu.Lock() + s.retentionHours = 24 // the stale rows are 3 days old; the ballast is an hour old + s.mu.Unlock() + if n := s.RunEviction(); n != 3+3 { + t.Fatalf("evicted %d transmissions, want the 3 merged groups and 3 singles (6)", n) + } + acct113Check(t, s, "after evicting the migrated rows") + got := hm215Snap(s) + if !reflect.DeepEqual(got, want) { + t.Errorf("store after migrate+evict differs from a store that never had the rows:\n got %+v\n want %+v", got, want) + } + s.mu.RLock() + defer s.mu.RUnlock() + if got.TrackedBytes != want.TrackedBytes { + t.Errorf("trackedBytes = %d, want exactly the baseline %d (drift %d)", got.TrackedBytes, want.TrackedBytes, got.TrackedBytes-want.TrackedBytes) + } + if len(s.fallbackByNode) != len(baseline.fallbackByNode) { + t.Errorf("fallbackByNode holds %d records, baseline %d", len(s.fallbackByNode), len(baseline.fallbackByNode)) + } + }) + } + } +} + +// Renaming the nodeHashes keys of a batch must cost what the batch holds, not +// what the whole index holds: with the resolved-pubkey index on, the keys of a +// transmission are found through the transmission (its decoded pubkeys, its +// fallback relays, its resolved relays), and no pass over all of nodeHashes +// runs under the store's write lock (#215 review, N2). +func TestHashMigrate_RenameDoesNotWalkTheWholeIndex_215(t *testing.T) { + d := hm215Create(t) + d.ballast(t) + d.stale(t, 3, 3, 3) + s := hm215Open(t, d) + before := hashRekeySweeps.Load() + hm215Migrate(s, 2) + if n := hashRekeySweeps.Load() - before; n != 0 { + t.Errorf("the rename walked the whole of nodeHashes %d times, want 0 with the resolved-pubkey index on", n) + } + // With the index off the relays of a transmission are not recorded + // anywhere, so the one pass per batch stays as the fallback. + d2 := hm215Create(t) + d2.ballast(t) + d2.stale(t, 1, 3, 0) + s2 := hm215OpenIdx(t, d2, false) + before = hashRekeySweeps.Load() + hm215Migrate(s2, 100) + if n := hashRekeySweeps.Load() - before; n == 0 { + t.Error("with the resolved-pubkey index off no pass ran: the relay keys would be left under the old hash") + } +} + +// A merge keeps what the duplicate knew: the earliest first_seen (the survivor +// stays where it is in s.packets, so the slice has to be put back in order) and +// the columns the survivor has no value for. The survivor's own values stay. +func TestHashMigrate_MergeKeepsEarliestFirstSeenAndFillsNulls_215(t *testing.T) { + d := hm215Create(t) + if _, err := d.conn.Exec(`ALTER TABLE transmissions ADD COLUMN scope_name TEXT`); err != nil { + t.Fatal(err) + } + d.ballast(t) + raw := hm215Raw(7) + t0 := time.Now().UTC().Add(-72 * time.Hour) + at := func(s int) string { return t0.Add(time.Duration(s) * time.Second).Format(time.RFC3339) } + // 50 is the survivor (lowest id) with the LATER first_seen and no scope; + // 51 is the duplicate with the earlier one and a scope. A bystander (52) + // sits between the two in time. + d.tx(t, 50, raw, "stale-r-50", at(30), 4, hm215Advert(acct113PK(210))) + d.tx(t, 51, raw, "stale-r-51", at(10), 4, hm215Advert(acct113PK(210))) + d.tx(t, 52, hm215Raw(8), "stale-r-52", at(20), 4, hm215Advert(acct113PK(211))) + if _, err := d.conn.Exec(`UPDATE transmissions SET scope_name = '#x' WHERE id = 51`); err != nil { + t.Fatal(err) + } + // Another pair where the survivor has its own scope. + raw2 := hm215Raw(9) + d.tx(t, 60, raw2, "stale-r-60", at(40), 4, hm215Advert(acct113PK(212))) + d.tx(t, 61, raw2, "stale-r-61", at(41), 4, hm215Advert(acct113PK(212))) + if _, err := d.conn.Exec(`UPDATE transmissions SET scope_name = CASE id WHEN 60 THEN '#keep' ELSE '#other' END WHERE id IN (60, 61)`); err != nil { + t.Fatal(err) + } + s := hm215Open(t, d) + hm215Migrate(s, 2) + + s.mu.RLock() + w := s.byTxID[50] + if w == nil || s.byTxID[51] != nil { + s.mu.RUnlock() + t.Fatal("tx 50 must survive and 51 be merged into it") + } + if w.FirstSeen != at(10) { + t.Errorf("survivor first_seen = %s, want the duplicate's earlier %s", w.FirstSeen, at(10)) + } + if w.ScopeName != "#x" { + t.Errorf("survivor scope = %q, want %q filled from the duplicate", w.ScopeName, "#x") + } + if s.byTxID[60].ScopeName != "#keep" { + t.Errorf("tx 60 scope = %q, want its own %q kept", s.byTxID[60].ScopeName, "#keep") + } + for i := 1; i < len(s.packets); i++ { + if s.packets[i-1].FirstSeen > s.packets[i].FirstSeen { + t.Errorf("s.packets is not sorted by first_seen at %d: %s > %s (eviction cuts from the head)", i, s.packets[i-1].FirstSeen, s.packets[i].FirstSeen) + break + } + } + s.mu.RUnlock() + + // Eviction cuts from the head by first_seen, so it reaches all of them. + s.mu.Lock() + s.retentionHours = 24 + s.mu.Unlock() + if n := s.RunEviction(); n != 3 { + t.Errorf("retention evicted %d transmissions, want the survivors of both pairs and the bystander (3)", n) + } + acct113Check(t, s, "after retention") +} + +// The batch was chosen under a read lock: what was evicted or rehashed since is +// dropped when it is applied under the write lock, not resurrected. +func TestHashMigrate_BatchRechecksUnderTheWriteLock_215(t *testing.T) { + d := hm215Create(t) + d.ballast(t) + d.stale(t, 2, 3, 2) + s := hm215Open(t, d) + s.mu.RLock() + updates := staleContentHashes(s, s.packets) + s.mu.RUnlock() + if len(updates) == 0 { + t.Fatal("setup: nothing stale") + } + // Eviction removes every stale row (they are 3 days old). + s.mu.Lock() + s.retentionHours = 24 + s.mu.Unlock() + if n := s.RunEviction(); n != 2*3+2 { + t.Fatalf("setup: evicted %d, want all %d stale rows", n, 2*3+2) + } + s.mu.Lock() + r, m := s.applyContentHashUpdates(updates) + s.mu.Unlock() + if r != 0 || m != 0 { + t.Errorf("applied %d rehashes and %d merges for transmissions that are gone", r, m) + } + s.mu.RLock() + defer s.mu.RUnlock() + if len(s.byHash) != 1 || len(s.byTxID) != 1 || len(s.packets) != 1 { + t.Errorf("byHash/byTxID/packets = %d/%d/%d after applying a stale batch, want only the ballast", len(s.byHash), len(s.byTxID), len(s.packets)) + } +} + +// The in-memory pass walks a snapshot of s.packets taken when it starts, so +// main must start it after the whole startup load (every chunk, the background +// fill included): in production it started right after the first chunk and saw +// almost none of the store (#215 review, N3). +func TestMain_StartsHashMigrationAfterStartupLoad_215(t *testing.T) { + src, err := os.ReadFile("main.go") + if err != nil { + t.Fatal(err) + } + m := string(src) + done := strings.Index(m, "<-store.StartupLoadDone()") + start := strings.Index(m, "migrateContentHashesAsync(store") + if done < 0 || start < 0 { + t.Fatalf("markers not found: StartupLoadDone wait=%d, migrateContentHashesAsync=%d", done, start) + } + if !(done < start) { + t.Error("migrateContentHashesAsync must start after <-store.StartupLoadDone()") + } + if strings.Contains(m[:done], "go migrateContentHashesAsync(") { + t.Error("the migration is also started before the startup load has finished") + } +} + +// A second run, and a store that was already migrated, change nothing. +func TestHashMigrate_IsIdempotent_215(t *testing.T) { + d := hm215Create(t) + d.ballast(t) + d.stale(t, 2, 3, 2) + s := hm215Open(t, d) + hm215Migrate(s, 4) + first := hm215Snap(s) + out := hm215Migrate(s, 4) + if strings.Contains(out, "Migrated") { + t.Errorf("a second run migrated again: %s", out) + } + if second := hm215Snap(s); !reflect.DeepEqual(first, second) { + t.Errorf("a second run changed the store:\n first %+v\n second %+v", first, second) + } + if !s.hashMigrationComplete.Load() { + t.Error("hashMigrationComplete not set") + } +} + +// A late observation of a migrated transmission must find the node index +// already holding it: a stale nodeHashes key would index it a second time. +func TestHashMigrate_RekeysNodeHashesSoLateObservationsDoNotDuplicate_215(t *testing.T) { + d := hm215Create(t) + d.ballast(t) + d.stale(t, 0, 0, 1) + s := hm215Open(t, d) + hm215Migrate(s, 10) + + s.mu.Lock() + defer s.mu.Unlock() + tx := s.byTxID[501] + pk := acct113PK(201) + if tx == nil || len(s.byNode[pk]) != 1 { + t.Fatalf("setup: byNode[%s...] = %d entries", pk[:4], len(s.byNode[pk])) + } + s.addToByNode(tx, pk) + s.addToByNode(tx, acct113PK(5)) // a relay the observation was resolved through + if n := len(s.byNode[pk]); n != 1 { + t.Errorf("a second index call added the migrated tx again: byNode has %d entries", n) + } + count := 0 + for _, t2 := range s.byNode[acct113PK(5)] { + if t2 == tx { + count++ + } + } + if count != 1 { + t.Errorf("the migrated tx is in byNode[relay] %d times, want 1", count) + } +} + +// perFallbackRelayBytes must stay a real allowance. The charge is derived, not +// measured on its own, so pin it from both sides in the default test run +// (the heap comparison is env-gated, #209 review): at least what a fallback +// relay structurally holds, at most what a resolved relay holds, of which the +// fallback relay is a strict subset. +func TestPerFallbackRelayBytes_IsPinned_215(t *testing.T) { + // A byNode slot holds a pointer (8). The nodeHashes entry holds a key + // string header (16) and a bool. The record in fallbackByNode holds the + // pubkey string header (16). + const floor = 8 + 16 + 16 + 1 + if got := fallbackRelayBytes(1) - fallbackRelayBytes(0); got < perFallbackRecordBytes+floor { + t.Errorf("one fallback relay is charged %d bytes, want at least %d", got, perFallbackRecordBytes+floor) + } + if got := fallbackRelayBytes(2) - fallbackRelayBytes(1); got != perFallbackRelayBytes || got < floor { + t.Errorf("a further fallback relay is charged %d bytes, want perFallbackRelayBytes=%d and at least %d", got, perFallbackRelayBytes, floor) + } + if perFallbackRelayBytes > perResolvedRelayBytes || perFallbackRecordBytes > perResolvedRecordBytes { + t.Errorf("a fallback relay (%d+%d) is charged more than a resolved relay (%d+%d), which holds a superset", + perFallbackRelayBytes, perFallbackRecordBytes, perResolvedRelayBytes, perResolvedRecordBytes) + } +} diff --git a/cmd/server/hash_migrate_round3_215_test.go b/cmd/server/hash_migrate_round3_215_test.go new file mode 100644 index 000000000..c2abb0b5c --- /dev/null +++ b/cmd/server/hash_migrate_round3_215_test.go @@ -0,0 +1,243 @@ +package main + +import ( + "fmt" + "reflect" + "regexp" + "strconv" + "testing" + "time" +) + +// Round 3 of #215 (review of round 2: NEW-1, NEW-3, NEW-4). + +func hm215TextMsg(dest, src string) string { + return fmt.Sprintf(`{"type":"TXT_MSG","destPubKey":%q,"srcPubKey":%q}`, dest, src) +} + +// hm215AssertOrdered fails when s.packets, a byPayloadType list or a byNode list +// is not in first_seen order (eviction cuts from the head of each). +func hm215AssertOrdered(t *testing.T, s *PacketStore) { + t.Helper() + check := func(name string, list []*StoreTx) { + for i := 1; i < len(list); i++ { + if list[i-1].FirstSeen > list[i].FirstSeen { + t.Errorf("%s is not in first_seen order at %d: tx %d (%s) before tx %d (%s)", + name, i, list[i-1].ID, list[i-1].FirstSeen, list[i].ID, list[i].FirstSeen) + return + } + } + } + s.mu.RLock() + defer s.mu.RUnlock() + check("s.packets", s.packets) + for pt, list := range s.byPayloadType { + check(fmt.Sprintf("byPayloadType[%d]", pt), list) + } + for pk, list := range s.byNode { + check(fmt.Sprintf("byNode[%s...]", pk[:4]), list) + } +} + +func hm215IDs(list []*StoreTx) []int { + ids := make([]int, len(list)) + for i, tx := range list { + ids[i] = tx.ID + } + return ids +} + +// A transmission is also indexed under the destPubKey and srcPubKey of its +// decoded JSON (indexByNode), not only pubKey: the rename and the removal of a +// merged-away duplicate must find those keys too. Every other fixture here uses +// ADVERTs, which carry pubKey only (#215 review, NEW-3, mutant M2). +func TestHashMigrate_RekeysDestAndSrcPubkeys_215(t *testing.T) { + const textMsg = 2 + for _, resolvedIndex := range []bool{true, false} { + for _, batch := range []int{1, 2, 100} { + t.Run(fmt.Sprintf("resolvedIndex=%v/batch%d", resolvedIndex, batch), func(t *testing.T) { + base := hm215Create(t) + base.ballast(t) + want := hm215Snap(hm215OpenIdx(t, base, resolvedIndex)) + + d := hm215Create(t) + d.ballast(t) + old := hm215Time(-72 * time.Hour) + older := hm215Time(-72*time.Hour + time.Second) + // A colliding pair and a single, none of them an ADVERT. + d.tx(t, 80, hm215Raw(80), "stale-t-80", old, textMsg, hm215TextMsg(acct113PK(150), acct113PK(151))) + d.tx(t, 81, hm215Raw(80), "stale-t-81", older, textMsg, hm215TextMsg(acct113PK(150), acct113PK(151))) + d.tx(t, 82, hm215Raw(82), "stale-t-82", old, textMsg, hm215TextMsg(acct113PK(152), acct113PK(153))) + s := hm215OpenIdx(t, d, resolvedIndex) + hm215Migrate(s, batch) + + s.mu.RLock() + pairHash, singleHash := ComputeContentHash(hm215Raw(80)), ComputeContentHash(hm215Raw(82)) + for _, pk := range []string{acct113PK(150), acct113PK(151)} { + if got := s.nodeHashes[pk]; len(got) != 1 || !got[pairHash] { + t.Errorf("nodeHashes[%s...] = %v, want exactly the new hash of the merged pair", pk[:4], got) + } + if list := s.byNode[pk]; len(list) != 1 || list[0].ID != 80 { + t.Errorf("byNode[%s...] = %v, want only the survivor 80", pk[:4], hm215IDs(list)) + } + } + for _, pk := range []string{acct113PK(152), acct113PK(153)} { + if got := s.nodeHashes[pk]; len(got) != 1 || !got[singleHash] { + t.Errorf("nodeHashes[%s...] = %v, want exactly the new hash of the single", pk[:4], got) + } + } + s.mu.RUnlock() + acct113Check(t, s, "after the migration") + + s.mu.Lock() + s.retentionHours = 24 + s.mu.Unlock() + if n := s.RunEviction(); n != 2 { + t.Fatalf("evicted %d transmissions, want the survivor of the pair and the single (2)", n) + } + acct113Check(t, s, "after evicting the migrated rows") + if got := hm215Snap(s); !reflect.DeepEqual(got, want) { + t.Errorf("store after migrate+evict differs from one that never had the rows:\n got %+v\n want %+v", got, want) + } + }) + } + } +} + +// A survivor that takes an earlier first_seen from its duplicate moves in every +// list that is ordered by first_seen: s.packets, byPayloadType and the byNode +// lists it is in. The order is asserted, not just the membership (#215 review, +// NEW-3, mutants M6 and M7). +func TestHashMigrate_FirstSeenMoveKeepsEveryListOrdered_215(t *testing.T) { + for _, resolvedIndex := range []bool{true, false} { + t.Run(fmt.Sprintf("resolvedIndex=%v", resolvedIndex), func(t *testing.T) { + d := hm215Create(t) + d.ballast(t) + t0 := time.Now().UTC().Add(-72 * time.Hour) + at := func(s int) string { return t0.Add(time.Duration(s) * time.Second).Format(time.RFC3339) } + pk := acct113PK(210) + // 50 survives with the LATER first_seen; its duplicate 51 is the earliest + // of all; 52 is a different packet of the same node, between the two. + d.tx(t, 50, hm215Raw(7), "stale-o-50", at(30), 4, hm215Advert(pk)) + d.tx(t, 51, hm215Raw(7), "stale-o-51", at(10), 4, hm215Advert(pk)) + d.tx(t, 52, hm215Raw(8), "stale-o-52", at(20), 4, hm215Advert(pk)) + s := hm215OpenIdx(t, d, resolvedIndex) + + s.mu.RLock() + if got, want := hm215IDs(s.byNode[pk]), []int{51, 52, 50}; !reflect.DeepEqual(got, want) { + s.mu.RUnlock() + t.Fatalf("setup: byNode[%s...] = %v before the migration, want load order %v", pk[:4], got, want) + } + s.mu.RUnlock() + hm215Migrate(s, 100) + + s.mu.RLock() + ballast := 1000 + if got, want := hm215IDs(s.packets), []int{50, 52, ballast}; !reflect.DeepEqual(got, want) { + t.Errorf("s.packets = %v, want %v", got, want) + } + if got, want := hm215IDs(s.byPayloadType[4]), []int{50, 52, ballast}; !reflect.DeepEqual(got, want) { + t.Errorf("byPayloadType[ADVERT] = %v, want %v", got, want) + } + if got, want := hm215IDs(s.byNode[pk]), []int{50, 52}; !reflect.DeepEqual(got, want) { + t.Errorf("byNode[%s...] = %v, want %v", pk[:4], got, want) + } + s.mu.RUnlock() + hm215AssertOrdered(t, s) + }) + } +} + +// A merge fills a survivor's empty scope from the duplicate. The ingestor's +// COALESCE fills a NULL scope_name only and keeps "" (transport-scoped, region +// unknown); StoreTx.ScopeName cannot tell the two apart and has no spare byte +// for a flag (it stays 320 bytes), so in memory "" is filled. The result is the +// same for a NULL survivor, the usual one, and differs for a "" survivor with a +// duplicate that matched a region: the DB keeps "" and memory shows the region +// until the reload the deploy plan requires (#215 review, NEW-4). This pins both +// halves of that documented difference, so changing either is a decision. +func TestHashMigrate_ScopeFillAgreesWithTheIngestorForNull_215(t *testing.T) { + d := hm215Create(t) + if _, err := d.conn.Exec(`ALTER TABLE transmissions ADD COLUMN scope_name TEXT`); err != nil { + t.Fatal(err) + } + d.ballast(t) + t0 := time.Now().UTC().Add(-72 * time.Hour) + at := func(s int) string { return t0.Add(time.Duration(s) * time.Second).Format(time.RFC3339) } + pair := func(survivor, dup, key int, sScope, dScope interface{}) { + raw := hm215Raw(key) + d.tx(t, survivor, raw, fmt.Sprintf("stale-c-%d", survivor), at(survivor), 4, hm215Advert(acct113PK(key))) + d.tx(t, dup, raw, fmt.Sprintf("stale-c-%d", dup), at(dup), 4, hm215Advert(acct113PK(key))) + if _, err := d.conn.Exec(`UPDATE transmissions SET scope_name = ? WHERE id = ?`, sScope, survivor); err != nil { + t.Fatal(err) + } + if _, err := d.conn.Exec(`UPDATE transmissions SET scope_name = ? WHERE id = ?`, dScope, dup); err != nil { + t.Fatal(err) + } + } + pair(70, 71, 70, nil, "#x") // NULL survivor: the ingestor fills it too + pair(72, 73, 72, nil, "") // NULL survivor, "" duplicate: "" in both + pair(74, 75, 74, "#k", "#z") + pair(76, 77, 76, nil, nil) + pair(78, 79, 78, "", "#w") // the documented difference: the DB keeps "" + s := hm215Open(t, d) + hm215Migrate(s, 100) + + s.mu.RLock() + defer s.mu.RUnlock() + for _, c := range []struct { + id int + want string + why string + }{ + {70, "#x", "NULL is filled with the duplicate's value, as the ingestor does"}, + {72, "", `NULL filled with ""`}, + {74, "#k", "its own value stays"}, + {76, "", "NULL and NULL"}, + {78, "#w", `a "" survivor is filled in memory (the DB keeps ""), see the comment above`}, + } { + tx := s.byTxID[c.id] + if tx == nil { + t.Fatalf("survivor %d is gone", c.id) + } + if tx.ScopeName != c.want { + t.Errorf("survivor %d scope = %q, want %q: %s", c.id, tx.ScopeName, c.want, c.why) + } + } +} + +// NEW-1 (measurement only): the migration reports the longest time one batch +// held the write lock, so a staging run can read it off the log. +func TestHashMigrate_LogsMaxWriteLockHold_215(t *testing.T) { + re := regexp.MustCompile(`\[hash-migrate\] .*max write-lock hold ([0-9]+\.[0-9]) ms over ([0-9]+) batches`) + for _, c := range []struct{ batch, want int }{{1, 8}, {100, 1}} { + d := hm215Create(t) + d.ballast(t) + d.stale(t, 2, 3, 2) // 8 stale rows + s := hm215Open(t, d) + out := hm215Migrate(s, c.batch) + m := re.FindStringSubmatch(out) + if m == nil { + t.Fatalf("batch %d: no max write-lock hold in the log:\n%s", c.batch, out) + } + if n, _ := strconv.Atoi(m[2]); n != c.want { + t.Errorf("batch %d: %d batches in the log, want %d (one per batch that had something to apply)", c.batch, n, c.want) + } + } + // A store with nothing stale applies nothing and logs nothing. + d := hm215Create(t) + d.ballast(t) + if out := hm215Migrate(hm215Open(t, d), 10); re.MatchString(out) { + t.Errorf("nothing was stale, but the log says: %s", out) + } +} + +func TestLockHold_TracksTheMaximumAndTheCount_215(t *testing.T) { + var h lockHold + for _, d := range []time.Duration{3 * time.Millisecond, 9 * time.Millisecond, 5 * time.Millisecond} { + h.record(d) + } + if h.max != 9*time.Millisecond || h.batches != 3 || h.total != 17*time.Millisecond { + t.Errorf("lockHold = %+v, want max 9ms, total 17ms over 3 batches", h) + } +} diff --git a/cmd/server/hash_migrate_test.go b/cmd/server/hash_migrate_test.go index 4868d3de5..3ed643c5a 100644 --- a/cmd/server/hash_migrate_test.go +++ b/cmd/server/hash_migrate_test.go @@ -41,13 +41,15 @@ func TestMigrateContentHashesAsync(t *testing.T) { t.Error("new hash should be in index") } + // The server never writes (#215): the DB keeps the stale hash until the + // ingestor's migration rewrites it. var dbHash string err = db.conn.QueryRow("SELECT hash FROM transmissions WHERE raw_hex = ?", rawHex).Scan(&dbHash) if err != nil { t.Fatal(err) } - if dbHash != correctHash { - t.Errorf("DB hash = %s, want %s", dbHash, correctHash) + if dbHash != wrongHash { + t.Errorf("DB hash = %s, want the stale %s: the server must not write", dbHash, wrongHash) } } @@ -78,7 +80,7 @@ func TestMigrateContentHashesAsync_NoOp(t *testing.T) { } } -func TestMigrateContentHashesAsync_CollisionUnionsObservedPathHashSizesWithoutChangingLegacyIndexes(t *testing.T) { +func TestMigrateContentHashesAsync_CollisionUnionsObservedPathHashSizesAndMerges(t *testing.T) { db := setupTestDBv2(t) store := NewPacketStore(db, nil) @@ -124,32 +126,31 @@ func TestMigrateContentHashesAsync_CollisionUnionsObservedPathHashSizesWithoutCh authoritative.observedPathHashSizes(), wantSizes) } - // Characterize, do not silently fix, the pre-existing duplicate-migration - // index bug. A separate change must remove the deleted row from every - // PacketStore index and re-parent its observations atomically. - if len(store.packets) != 2 || len(store.byPayloadType[PayloadGRP_TXT]) != 2 { - t.Fatalf("legacy in-memory cardinality changed: packets/payload = %d/%d, want 2/2", + // The duplicate is merged in memory (#215): one transmission remains, the + // lowest id, holding both observations. + if len(store.packets) != 1 || len(store.byPayloadType[PayloadGRP_TXT]) != 1 { + t.Fatalf("in-memory cardinality: packets/payload = %d/%d, want 1/1", len(store.packets), len(store.byPayloadType[PayloadGRP_TXT])) } - if got := store.QueryPackets(PacketQuery{Limit: 10}).Total; got != 2 { - t.Fatalf("legacy QueryPackets cardinality changed: got %d, want 2", got) + if got := store.QueryPackets(PacketQuery{Limit: 10}).Total; got != 1 { + t.Fatalf("QueryPackets cardinality: got %d, want 1", got) } - if store.byTxID[2] == nil || store.byTxID[2].Observations[0].TransmissionID != 2 { - t.Fatal("legacy duplicate/observation ownership changed unexpectedly") + if store.byTxID[2] != nil || store.byTxID[1] == nil || len(store.byTxID[1].Observations) != 2 { + t.Fatal("the lowest id must survive with both observations") + } + for _, o := range store.byTxID[1].Observations { + if o.TransmissionID != 1 { + t.Fatalf("observation %d still names tx %d", o.ID, o.TransmissionID) + } } - var txCount, canonicalObservationCount int + // ...and the DB is left to the ingestor: the server wrote nothing. + var txCount int if err := db.conn.QueryRow(`SELECT COUNT(*) FROM transmissions WHERE raw_hex = ?`, rawHex).Scan(&txCount); err != nil { t.Fatal(err) } - if txCount != 1 { - t.Fatalf("DB transmission rows = %d, want 1 after duplicate merge", txCount) - } - if err := db.conn.QueryRow(`SELECT COUNT(*) FROM observations WHERE transmission_id = 1`).Scan(&canonicalObservationCount); err != nil { - t.Fatal(err) - } - if canonicalObservationCount != 2 { - t.Fatalf("DB canonical observations = %d, want 2 after duplicate merge", canonicalObservationCount) + if txCount != 2 { + t.Fatalf("DB transmission rows = %d, want the 2 the ingestor has not merged yet", txCount) } } @@ -194,8 +195,11 @@ func TestMigrateContentHashesAsync_CollisionIncludesAlreadyCurrentSurvivorEviden if got := store.byHash[correctHash].observedPathHashSizes(); !reflect.DeepEqual(got, want) { t.Fatalf("authoritative observed path hash sizes = %v, want %v", got, want) } - if len(store.packets) != 2 || store.QueryPackets(PacketQuery{Limit: 10}).Total != 2 { - t.Fatal("legacy in-memory duplicate cardinality changed unexpectedly") + if len(store.packets) != 1 || store.QueryPackets(PacketQuery{Limit: 10}).Total != 1 { + t.Fatal("the duplicate was not merged into the already-current survivor") + } + if store.byHash[correctHash].ID != 1 { + t.Fatalf("survivor = tx %d, want the already-current tx 1 (lowest id)", store.byHash[correctHash].ID) } } @@ -245,34 +249,24 @@ func TestMigrateContentHashesAsync_CollisionCarriesEvidenceAcrossBatches(t *test t.Fatalf("authoritative observed path hash sizes = %v, want %v", got, want) } - // Keep characterizing the separate legacy index bug rather than hiding it - // inside this evidence-only feature change. - if len(store.packets) != 3 || len(store.byPayloadType[PayloadGRP_TXT]) != 3 { - t.Fatalf("legacy in-memory cardinality changed: packets/payload = %d/%d, want 3/3", + // Three rows, one per batch, all merge into the lowest id, in memory. + if len(store.packets) != 1 || len(store.byPayloadType[PayloadGRP_TXT]) != 1 { + t.Fatalf("in-memory cardinality: packets/payload = %d/%d, want 1/1", len(store.packets), len(store.byPayloadType[PayloadGRP_TXT])) } - if got := store.QueryPackets(PacketQuery{Limit: 10}).Total; got != 3 { - t.Fatalf("legacy QueryPackets cardinality changed: got %d, want 3", got) + if got := store.QueryPackets(PacketQuery{Limit: 10}).Total; got != 1 { + t.Fatalf("QueryPackets cardinality: got %d, want 1", got) } - for _, duplicateID := range []int{2, 3} { - duplicate := store.byTxID[duplicateID] - if duplicate == nil || len(duplicate.Observations) != 1 || - duplicate.Observations[0].TransmissionID != duplicateID { - t.Fatalf("legacy duplicate %d observation ownership changed unexpectedly", duplicateID) - } + if survivor := store.byTxID[1]; survivor == nil || len(survivor.Observations) != 3 { + t.Fatal("the lowest id must survive with all three observations") } - var txCount, canonicalObservationCount int + // The DB is the ingestor's: all three rows are still there. + var txCount int if err := db.conn.QueryRow(`SELECT COUNT(*) FROM transmissions WHERE raw_hex = ?`, rawHex).Scan(&txCount); err != nil { t.Fatal(err) } - if txCount != 1 { - t.Fatalf("DB transmission rows = %d, want 1 after duplicate merges", txCount) - } - if err := db.conn.QueryRow(`SELECT COUNT(*) FROM observations WHERE transmission_id = 1`).Scan(&canonicalObservationCount); err != nil { - t.Fatal(err) - } - if canonicalObservationCount != 3 { - t.Fatalf("DB canonical observations = %d, want 3 after duplicate merges", canonicalObservationCount) + if txCount != 3 { + t.Fatalf("DB transmission rows = %d, want the 3 the ingestor has not merged yet", txCount) } } diff --git a/cmd/server/main.go b/cmd/server/main.go index 0c9cc2694..14221cdc2 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -633,8 +633,15 @@ func main() { // process. The server reads the results via the periodic // recompNeighborGraph / fetchResolvedPathForObs paths. - // Migrate old content hashes in background (one-time, idempotent). - go migrateContentHashesAsync(store, 5000, 100*time.Millisecond) + // Rehash content hashes in memory in the background (idempotent). The DB + // rewrite is the ingestor's (#215); the server never writes. + // It walks a snapshot of s.packets taken when it starts, so it waits for the + // whole startup load: LoadChunked and the background fill, not just the + // first chunk the HTTP listener binds after (#215). + go func() { + <-store.StartupLoadDone() + migrateContentHashesAsync(store, 5000, 100*time.Millisecond) + }() if err := httpServer.ListenAndServe(); err != http.ErrServerClosed { log.Fatalf("[server] %v", err) diff --git a/cmd/server/readonly_invariant_test.go b/cmd/server/readonly_invariant_test.go index cc2fdde70..398cc3b35 100644 --- a/cmd/server/readonly_invariant_test.go +++ b/cmd/server/readonly_invariant_test.go @@ -203,3 +203,90 @@ func nodeTableWritePattern(verb, trailer string) *regexp.Regexp { } return regexp.MustCompile(expr) } + +// txTableWritePattern matches DML against the ingestor-owned packet tables +// (transmissions, observations) and the tables hung off them, in the shapes +// nodeTableWritePattern covers. +func txTableWritePattern(verb string) *regexp.Regexp { + const table = "[\"`\\[]?(transmissions|observations|ping_triggers|route_mask_changes)[\"`\\]]?" + return regexp.MustCompile(`(?i)` + verb + `\s+` + table + `\b`) +} + +// TestServerHasNoPacketTableWrites enforces #215: the content-hash migration +// used to UPDATE transmissions/observations and DELETE transmissions from the +// server's mode=ro handle. Every statement failed, was logged as a collision, +// and was retried at every start. cmd/server/ may not rewrite the packet +// tables; that is the ingestor's job (cmd/ingestor/hash_migrate.go). +// +// UPDATE, DELETE and REPLACE are forbidden in every server source file. The +// hash migration files (hash_migrate*.go) may not issue any statement at all: +// no INSERT either, and no transaction or Exec on the connection. +// +// Known gap, outside this change: handlePostPacket (routes.go) INSERTs into +// transmissions/observations on the same read-only handle. It is not part of +// the hash migration and is left alone here; INSERT is therefore only +// checked in the migration files. +func TestServerHasNoPacketTableWrites(t *testing.T) { + entries, err := os.ReadDir(".") + if err != nil { + t.Fatal(err) + } + everywhere := []*regexp.Regexp{ + txTableWritePattern(`UPDATE(\s+OR\s+\w+)?`), + txTableWritePattern(`DELETE\s+FROM`), + txTableWritePattern(`REPLACE\s+INTO`), + } + migrationOnly := []*regexp.Regexp{ + txTableWritePattern(`INSERT\s+(OR\s+\w+\s+)?INTO`), + regexp.MustCompile(`\.conn\.(Begin|BeginTx|Exec|ExecContext|Prepare|PrepareContext)\s*\(`), + } + var violations []string + for _, e := range entries { + name := e.Name() + if e.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") { + continue + } + b, err := os.ReadFile(filepath.Join(".", name)) + if err != nil { + t.Fatal(err) + } + patterns := everywhere + if strings.HasPrefix(name, "hash_migrate") { + patterns = append(append([]*regexp.Regexp{}, everywhere...), migrationOnly...) + } + for _, p := range patterns { + if loc := p.FindIndex(b); loc != nil { + line := 1 + strings.Count(string(b[:loc[0]]), "\n") + violations = append(violations, fmt.Sprintf("%s:%d: %s", name, line, p.String())) + } + } + } + if len(violations) > 0 { + t.Errorf("cmd/server/ writes the packet tables (#215):\n %s", strings.Join(violations, "\n ")) + } +} + +// TestServerHasNoPacketTableWritesIsSensitive keeps the guard honest: it must +// match the statements the old migration issued. +func TestServerHasNoPacketTableWritesIsSensitive(t *testing.T) { + for _, stmt := range []string{ + `UPDATE transmissions SET hash = ? WHERE id = ?`, + `UPDATE observations SET transmission_id = ? WHERE transmission_id = ?`, + `DELETE FROM transmissions WHERE id = ?`, + `update "observations" set x = 1`, + `UPDATE OR IGNORE observations SET transmission_id = 1`, + } { + matched := false + for _, p := range []*regexp.Regexp{ + txTableWritePattern(`UPDATE(\s+OR\s+\w+)?`), + txTableWritePattern(`DELETE\s+FROM`), + } { + if p.MatchString(stmt) { + matched = true + } + } + if !matched { + t.Errorf("the guard does not match %q", stmt) + } + } +} diff --git a/cmd/server/readonly_sql_literal_test.go b/cmd/server/readonly_sql_literal_test.go index 8a63b1e98..91681efcd 100644 --- a/cmd/server/readonly_sql_literal_test.go +++ b/cmd/server/readonly_sql_literal_test.go @@ -25,11 +25,11 @@ var writeSQLPattern = regexp.MustCompile(`(?is)\b(INSERT\s+(OR\s+\w+\s+)?INTO|RE // shared one. // - backup.go: VACUUM INTO writes a snapshot file, not the database. // - openapi.go: prose mentioning VACUUM INTO. -// - hash_migrate.go, routes.go: pre-existing writes to the shared -// database, tracked separately. +// - routes.go: pre-existing writes to the shared database (the packet +// POST handler), tracked separately. hash_migrate.go had 3 until the +// content-hash migration moved to the ingestor (#215). var knownServerWriteSQL = map[string]int{ "backup.go": 1, - "hash_migrate.go": 3, "openapi.go": 1, "ping_score_history.go": 15, "routes.go": 3, diff --git a/cmd/server/store.go b/cmd/server/store.go index a2ba63611..fe4915847 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -3701,31 +3701,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] pathHopMutated := false for txID, tx := range updatedTxs { if tx.PathJSON != oldPaths[txID] { - // Path changed — remove old subpaths, add new ones. - oldHops := parsePathJSON(oldPaths[txID]) - if len(oldHops) >= 2 { - // Temporarily set parsedPath to old hops for removal. - saved, savedFlag := tx.parsedPath, tx.pathParsed - tx.parsedPath, tx.pathParsed = oldHops, true - if removeTxFromSubpathIndexFull(s.spIndex, s.spTxIndex, tx) { - s.spTotalPaths-- - } - tx.parsedPath, tx.pathParsed = saved, savedFlag - } - // Remove old path-hop index entries using old hops. - // Resolved pubkey entries are managed via resolvedPubkeyIndex, not byPathHop. - if len(oldHops) > 0 { - saved, savedFlag := tx.parsedPath, tx.pathParsed - tx.parsedPath, tx.pathParsed = oldHops, true - removeTxFromPathHopIndex(s.byPathHop, tx) - tx.parsedPath, tx.pathParsed = saved, savedFlag - } - // pickBestObservation already set pathParsed=false so - // addTxToSubpathIndex will re-parse the new path. - if addTxToSubpathIndexFull(s.spIndex, s.spTxIndex, tx) { - s.spTotalPaths++ - } - addTxToPathHopIndex(s.byPathHop, tx) + s.reindexTxPath(tx, oldPaths[txID]) // #1164: coalesce — one invalidate after the loop, not per-tx. pathHopMutated = true } @@ -3769,6 +3745,38 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] return broadcastMaps } +// reindexTxPath moves tx in the subpath and path-hop indexes after its best +// path changed from oldPath to tx.PathJSON (pickBestObservation, which also +// cleared tx.pathParsed). The caller invalidates the relay-stats cache once for +// a batch. Must hold s.mu for writing. +func (s *PacketStore) reindexTxPath(tx *StoreTx, oldPath string) { + // Path changed — remove old subpaths, add new ones. + oldHops := parsePathJSON(oldPath) + if len(oldHops) >= 2 { + // Temporarily set parsedPath to old hops for removal. + saved, savedFlag := tx.parsedPath, tx.pathParsed + tx.parsedPath, tx.pathParsed = oldHops, true + if removeTxFromSubpathIndexFull(s.spIndex, s.spTxIndex, tx) { + s.spTotalPaths-- + } + tx.parsedPath, tx.pathParsed = saved, savedFlag + } + // Remove old path-hop index entries using old hops. + // Resolved pubkey entries are managed via resolvedPubkeyIndex, not byPathHop. + if len(oldHops) > 0 { + saved, savedFlag := tx.parsedPath, tx.pathParsed + tx.parsedPath, tx.pathParsed = oldHops, true + removeTxFromPathHopIndex(s.byPathHop, tx) + tx.parsedPath, tx.pathParsed = saved, savedFlag + } + // pickBestObservation already set pathParsed=false so + // addTxToSubpathIndex will re-parse the new path. + if addTxToSubpathIndexFull(s.spIndex, s.spTxIndex, tx) { + s.spTotalPaths++ + } + addTxToPathHopIndex(s.byPathHop, tx) +} + // MaxTransmissionID returns the highest transmission ID in the store. func (s *PacketStore) MaxTransmissionID() int { s.mu.RLock() diff --git a/cmd/server/tracked_bytes_leaks_202_test.go b/cmd/server/tracked_bytes_leaks_202_test.go index 498f06726..53e353a06 100644 --- a/cmd/server/tracked_bytes_leaks_202_test.go +++ b/cmd/server/tracked_bytes_leaks_202_test.go @@ -53,7 +53,9 @@ func leak202TxCharge(t *testing.T, s *PacketStore, id int) int64 { } // A hash migration that merges two duplicate transmissions must credit each -// observation exactly once when the store later evicts them. +// observation exactly once when the store later evicts them. The merge moves +// the duplicate's observations to the survivor (#215), so the survivor holds +// all five and the one transmission is evicted. func TestHashMigrationMerge_EvictionCreditsEveryObservationOnce_202(t *testing.T) { db := setupTestDBv2(t) store := NewPacketStore(db, nil) @@ -92,22 +94,19 @@ func TestHashMigrationMerge_EvictionCreditsEveryObservationOnce_202(t *testing.T if !store.WaitIndexesReady(30 * time.Second) { t.Fatal("background index builds did not finish") } - ballast := leak202TxCharge(t, store, 900) - migrateContentHashesAsync(store, 1, 0) if !store.hashMigrationComplete.Load() { t.Fatal("hash migration did not complete") } - var dbTx int - if err := db.conn.QueryRow(`SELECT COUNT(*) FROM transmissions WHERE raw_hex = ?`, rawHex).Scan(&dbTx); err != nil { - t.Fatal(err) - } - if dbTx != 1 { - t.Fatalf("setup: %d duplicate rows left in the DB, want the merge to leave 1", dbTx) + if len(store.packets) != 2 { + t.Fatalf("setup: %d transmissions after the merge, want the survivor and the ballast", len(store.packets)) } + // Measured after the migration: the ballast's stored hash is not the + // current formula either, so it is rehashed too, and its charge follows the + // hash length (#215). + ballast := leak202TxCharge(t, store, 900) - // Whatever the merge did in memory, trackedBytes must still equal what the - // store holds, before anything is evicted. + // trackedBytes must equal what the store holds, before anything is evicted. if got, want := store.trackedBytes, leak202LiveCharge(store); got != want { t.Fatalf("after the merge trackedBytes = %d, live charges sum to %d (drift %d)", got, want, got-want) } diff --git a/cmd/server/tracked_bytes_perf_202_test.go b/cmd/server/tracked_bytes_perf_202_test.go index 96c7ea8a8..a33d7a95e 100644 --- a/cmd/server/tracked_bytes_perf_202_test.go +++ b/cmd/server/tracked_bytes_perf_202_test.go @@ -89,6 +89,7 @@ func TestPerf_FallbackChargeVsHeap_202(t *testing.T) { func TestPerf_HashMigrateMerge_202(t *testing.T) { runs := perf202Runs(t) var migrate []time.Duration + packets := 0 for i := 0; i < runs; i++ { db := setupTestDBv2(t) tx, err := db.conn.Begin() @@ -120,10 +121,59 @@ func TestPerf_HashMigrateMerge_202(t *testing.T) { t0 := time.Now() migrateContentHashesAsync(store, 500, 0) migrate = append(migrate, time.Since(t0)) - if len(store.packets) != 2000 { - t.Fatalf("%d packets after the migration, want 2000", len(store.packets)) + // Master leaves the 2000 ghosts in memory; since #215 the duplicates + // are merged (1000 remain). Printed, not asserted, so the same file + // compiles and runs against both. + packets = len(store.packets) + db.conn.Close() + } + fmt.Printf("RESULT hash-migrate-merge-2000 median_ns=%d packets_after=%d\n", perf202Median(migrate).Nanoseconds(), packets) +} + +// Hash migration of CORESCOPE_PERF_215_ROWS (default 5000) transmissions that all carry a stale hash and do not +// collide, each indexed under three node keys (#215). It measures the rename of +// the nodeHashes keys that the migration now does per batch. Compiles against +// master too (where the migration also writes the DB, on this in-memory +// handle). +func TestPerf_HashMigrateRehash_215(t *testing.T) { + runs := perf202Runs(t) + rows, _ := strconv.Atoi(os.Getenv("CORESCOPE_PERF_215_ROWS")) + if rows <= 0 { + rows = 5000 + } + var migrate []time.Duration + for i := 0; i < runs; i++ { + db := setupTestDBv2(t) + tx, err := db.conn.Begin() + if err != nil { + t.Fatal(err) + } + for id := 1; id <= rows; id++ { + raw := fmt.Sprintf("0A00D69FD7A5A7475DB07337749AE61FA53A478%05X", id) + decoded := fmt.Sprintf(`{"type":"ADVERT","pubKey":"%064x","destPubKey":"%064x","srcPubKey":"%064x"}`, id%800, (id+1)%800, (id+2)%800) + if _, err := tx.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES (?, ?, ?, '2026-01-01T00:00:00Z', 1, 4, ?)`, id, raw, fmt.Sprintf("old-%d", id), decoded); err != nil { + t.Fatal(err) + } + if _, err := tx.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, path_json, timestamp) + VALUES (?, ?, 'observer', 'Observer', '["AA","BB"]', ?)`, id, id, id); err != nil { + t.Fatal(err) + } + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + store := NewPacketStore(db, nil) + if err := store.Load(); err != nil { + t.Fatal(err) + } + t0 := time.Now() + migrateContentHashesAsync(store, 500, 0) + migrate = append(migrate, time.Since(t0)) + if len(store.packets) != rows { + t.Fatalf("%d packets after the migration, want %d", len(store.packets), rows) } db.conn.Close() } - fmt.Printf("RESULT hash-migrate-merge-2000 median_ns=%d\n", perf202Median(migrate).Nanoseconds()) + fmt.Printf("RESULT hash-migrate-rehash-%d median_ns=%d\n", rows, perf202Median(migrate).Nanoseconds()) }