diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 5f60ba85..0279e18d 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -254,6 +254,24 @@ type RetentionConfig struct { // region-discovery stream would otherwise grow without limit. Consumed by // Task 6's pruner. ClientRegionsDays int `json:"clientRegionsDays"` + // HopIndexDays turns on the observation_hops inverted index (the + // accelerator behind /api/nodes/{pk}/reach) and bounds it to this many + // days. 0 or unset = off: no table, no triggers, and a previously installed + // index is removed at startup. The index costs roughly 200 bytes per + // observation (about 30% of the observations table, measured on a real + // 10M-observation database; real paths average 10 to 12 hops) plus a + // trigger on every observation insert. Reach windows longer than this fall + // back to the substring scan. + HopIndexDays int `json:"hopIndexDays"` +} + +// HopIndexDaysOrZero returns retention.hopIndexDays, or 0 (off) when unset. +// LoadConfig rejects negative values, so none reach here from a config file. +func (c *Config) HopIndexDaysOrZero() int { + if c.Retention != nil && c.Retention.HopIndexDays > 0 { + return c.Retention.HopIndexDays + } + return 0 } // PacketDaysOrZero returns the configured retention.packetDays or 0 @@ -539,6 +557,11 @@ func LoadConfig(path string) (*Config, error) { if err := json.Unmarshal(data, &cfg); err != nil { return nil, fmt.Errorf("parsing config %s: %w", path, err) } + // observerDays reads -1 as "keep forever"; here a negative value would + // silently remove the index, so refuse it instead. + if cfg.Retention != nil && cfg.Retention.HopIndexDays < 0 { + return nil, fmt.Errorf("config %s: retention.hopIndexDays is %d; it must be 0 (off) or a number of days", path, cfg.Retention.HopIndexDays) + } } // Env var overrides diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index b616cd19..f69c815f 100644 --- a/cmd/ingestor/db.go +++ b/cmd/ingestor/db.go @@ -87,6 +87,9 @@ type Store struct { stmtBumpTxLastSeen *sql.Stmt stmtInsertObservation *sql.Stmt stmtInsertAdvertEvidence *sql.Stmt + hopBackfillCancel context.CancelFunc // stops the observation_hops backfill on Close + hopIndexDays int // retention.hopIndexDays; 0 = no hop index (see ConfigureObservationHops) + hopMaintMu sync.Mutex // serializes the hop index backfill and trim stmtUpsertNode *sql.Stmt stmtIncrementAdvertCount *sql.Stmt stmtUpsertObserver *sql.Stmt @@ -233,6 +236,9 @@ func OpenStoreWithInterval(dbPath string, sampleIntervalSec int) (*Store, error) var evidenceStatus string _ = db.QueryRow(`SELECT status FROM _async_migrations WHERE name='advert_route_evidence_v1'`).Scan(&evidenceStatus) s.advertEvidenceComplete.Store(evidenceStatus == "done") + // The observation_hops backfill is deliberately NOT scheduled here; see + // ScheduleObservationHopsBackfill (called from main once ingest is live). + return s, nil } @@ -1422,6 +1428,11 @@ func observerMetaColumns(meta *ObserverMeta) (model, firmware, clientVersion, ra // Close checkpoints the WAL and closes the database. func (s *Store) Close() error { + // Long resumable backfills must not hold shutdown hostage; they pick up + // from their stored progress on the next boot. + if s.hopBackfillCancel != nil { + s.hopBackfillCancel() + } s.backfillWg.Wait() s.Checkpoint() return s.db.Close() diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 90edcd9d..cd5a5a3a 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -282,6 +282,14 @@ func main() { log.Printf("Running — %d MQTT source(s) connected", connectedCount) } + // Opt-in reach hop index: installs it, or removes it when off (the default). + // Here, with the MQTT subscription up and the ingest gate still closed: a + // removal holds the single writer for about a minute on a large database, + // and the buffer keeps that minute's packets instead of losing them. + if err := store.ConfigureObservationHops(cfg.HopIndexDaysOrZero()); err != nil { + log.Fatalf("db: hop index: %v", err) + } + // Node retention: move stale nodes to inactive_nodes on startup nodeDays := cfg.NodeDaysOrDefault() store.MoveStaleNodes(nodeDays) @@ -315,6 +323,13 @@ func main() { } } + // Keep observation_hops bounded to its window (no-op when the index is off). + if n, err := store.TrimObservationHops(); err != nil { + log.Printf("[hop-index] trim error: %v", err) + } else if n > 0 { + log.Printf("[hop-index] startup trimmed %d hop rows", n) + } + // Client-RX coverage retention: bound the opt-in coverage tables (#1727). // Independent of the feature flag, so data persists are reaped even after // the feature is turned off. 0 = disabled. @@ -384,6 +399,9 @@ func main() { keyBackfillCtx, stopKeyBackfill := context.WithCancel(context.Background()) defer stopKeyBackfill() keyBackfillDone := keyBackfill.Start(keyBackfillCtx) + // After the gate: ingest is live, the backfill yields to it between chunks. + // No-op when the index is off or already covers its window. + store.ScheduleObservationHopsBackfill() if d := ingestBuffer.Dropped(); d > 0 { log.Printf("[ingest-buffer] write path ready; draining backlog (dropped %d during startup — consider raising ingestBufferSize)", d) } else { @@ -423,6 +441,19 @@ func main() { } }() + // Daily trim of the reach hop index, independent of packet retention. + if cfg.HopIndexDaysOrZero() > 0 { + go func() { + for range time.Tick(24 * time.Hour) { + if n, err := store.TrimObservationHops(); err != nil { + log.Printf("[hop-index] trim error: %v", err) + } else if n > 0 { + log.Printf("[hop-index] trimmed %d hop rows", n) + } + } + }() + } + // Daily ticker for transmission retention (#1283). var packetRetentionTicker *time.Ticker if packetDays > 0 { diff --git a/cmd/ingestor/observation_hops_backfill.go b/cmd/ingestor/observation_hops_backfill.go new file mode 100644 index 00000000..acdf1c1e --- /dev/null +++ b/cmd/ingestor/observation_hops_backfill.go @@ -0,0 +1,277 @@ +package main + +import ( + "context" + "database/sql" + "fmt" + "log" + "time" + + "github.com/meshcore-analyzer/dbschema" +) + +// Chunk size and pacing for the observation_hops backfill. vars so tests can +// run it without sleeping. +var ( + // hopBackfillChunkSeconds is the timestamp span indexed per writer + // transaction. 15 minutes is ~7k observations (~80k hop rows) on a busy + // mesh: on a small EBS-backed host an hour-sized chunk held the writer for + // 2-7s, which stalls live MQTT ingest behind it (writerMu, see WriterTx). + hopBackfillChunkSeconds int64 = 900 + // hopBackfillPause is the idle gap between chunks. The backfill reads the + // whole observations table once; on a box whose page cache is far smaller + // than the DB that is real I/O contention, so it yields between chunks. + hopBackfillPause = 250 * time.Millisecond + // hopBackfillAfterChunk, when set, runs after each committed backfill + // chunk. Tests only. + hopBackfillAfterChunk func() + // hopMaintAfterRead, when set, runs after the backfill or the trim has read + // observations to pick its next chunk and before the transaction that moves + // the coverage mark. Tests only: a row inserted here is what those + // transactions must still account for. + hopMaintAfterRead func() + // hopNow is the clock the index window is measured from; tests pin it. + hopNow = time.Now +) + +// ConfigureObservationHops applies retention.hopIndexDays. Called once from +// main's startup block, after the MQTT subscription is up and before the ingest +// gate opens: removing the index held the writer for 45 to 71 s on a copy of a +// 10.5M-observation database, and run there that time is buffered ingest, not +// lost packets. It precedes the startup trim and the backfill. +// +// days == 0 (the default; LoadConfig rejects negative values) removes the index +// if a previous run installed it, so an upgrade that never sets the key changes +// nothing, and switching it off leaves no table and no triggers behind. +// days > 0 installs the index (or refreshes its triggers); +// ScheduleObservationHopsBackfill then fills whatever part of the window is not +// yet covered. +func (s *Store) ConfigureObservationHops(days int) error { + s.hopIndexDays = days + if days <= 0 { + start := time.Now() + removed, err := dbschema.DropObservationHopsSchema(s.db) + if err != nil { + return err + } + if removed { + log.Printf("[hop-index] off (retention.hopIndexDays unset or 0): removed observation_hops and its triggers in %s", time.Since(start).Round(time.Millisecond)) + } + return nil + } + if err := dbschema.EnsureObservationHopsSchema(s.db, log.Printf); err != nil { + return err + } + since, err := s.hopCoverage() + if err != nil { + return err + } + log.Printf("[hop-index] on: window %d days, covered since %s", days, time.Unix(since, 0).UTC().Format(time.RFC3339)) + return nil +} + +func (s *Store) hopCoverage() (int64, error) { + var since int64 + err := s.db.QueryRow(`SELECT since FROM ` + dbschema.ObservationHopsCoverageTable + ` WHERE id = 1`).Scan(&since) + if err != nil { + return 0, fmt.Errorf("read hop coverage: %w", err) + } + return since, nil +} + +func (s *Store) hopFloor() int64 { + return hopNow().Unix() - int64(s.hopIndexDays)*86400 +} + +// ScheduleObservationHopsBackfill starts the backfill in the background when the +// index is on and its coverage does not yet reach back hopIndexDays: on first +// install, after an interrupted run, and after the window was raised. It must +// be called AFTER the ingest buffer is opened: main gates ingest on +// WaitForAsyncMigrations(), and a backfill that long (15 minutes on a 13GB +// database) scheduled earlier keeps live ingest dark the whole time. Run after +// the gate, it shares the writer with ingest one short chunk at a time. Close() +// cancels it; it resumes from the stored coverage mark on the next boot. +func (s *Store) ScheduleObservationHopsBackfill() { + if s.hopIndexDays <= 0 { + return + } + since, err := s.hopCoverage() + if err != nil { + log.Printf("[hop-index] backfill not scheduled: %v", err) + return + } + if since <= s.hopFloor() { + return // the whole window is covered; the daily trim keeps it that way + } + ctx, cancel := context.WithCancel(context.Background()) + s.hopBackfillCancel = cancel + s.backfillWg.Add(1) + go func() { + defer s.backfillWg.Done() + log.Printf("[hop-index] backfilling observation_hops from %s back to %d days (newest first)", + time.Unix(since, 0).UTC().Format(time.RFC3339), s.hopIndexDays) + if err := s.backfillObservationHops(ctx); err != nil { + log.Printf("[hop-index] backfill stopped, resumes on next start: %v", err) + } + }() +} + +// backfillObservationHops indexes existing observations into observation_hops, +// newest first, lowering observation_hops_coverage.since as each chunk commits +// so readers can use the index for any window the backfill has already covered +// while it is still running. +// +// Idempotent and resumable: it continues from the stored coverage mark, and the +// inserts are OR IGNORE (the insert trigger may already have indexed a row that +// arrived late with an old timestamp). +// +// Each transaction indexes every second it moves the mark across, not only the +// chunk around the observation it found: the read that picks the chunk runs +// outside the transaction, and a row with a timestamp in that range can arrive +// in between, too old for the trigger (the mark has not moved yet). Covering the +// whole range makes the mark's claim true whatever arrived meanwhile. The same +// rule applies to the last step, which claims down to the floor. +// +// It holds hopMaintMu for its whole run, so a trim cannot interleave with it. +// Without that, a trim that raised the mark and deleted rows between two chunks +// would be undone by the next chunk lowering the mark again, and readers would +// trust a range whose rows the trim had deleted. +func (s *Store) backfillObservationHops(ctx context.Context) error { + s.hopMaintMu.Lock() + defer s.hopMaintMu.Unlock() + since, err := s.hopCoverage() + if err != nil { + return err + } + start := time.Now() + var rows int64 + // Only the rolling window is indexed. since never drops below the floor, so + // the coverage mark honestly says "indexed from here". + floor := s.hopFloor() + for since > floor { + if err := ctx.Err(); err != nil { + return err + } + // Jump straight to the newest observation below the coverage mark + // instead of stepping a fixed span: a bad-clock observer can leave rows + // at timestamp 0 or in 1970, and walking decades of empty hourly + // chunks would take days. timestamp is indexed, so this is a seek. + var next sql.NullInt64 + if err := s.db.QueryRow(`SELECT MAX(timestamp) FROM observations WHERE timestamp < ?`, since).Scan(&next); err != nil { + return fmt.Errorf("next observation below %d: %w", since, err) + } + // Cover [lower, since): the chunk below the newest observation found, or + // exactly down to the floor when nothing is left inside the window. The + // INSERT runs over that whole range, so the mark never claims a second + // the transaction did not index (see the function comment). + lower := floor + if next.Valid && next.Int64 >= floor { + lower = next.Int64 - hopBackfillChunkSeconds + 1 + if lower < floor { + lower = floor + } + } + if hopMaintAfterRead != nil { + hopMaintAfterRead() + } + var n int64 + err := s.WriterTx("observation_hops_backfill", func(tx *sql.Tx) error { + var err error + if n, err = dbschema.IndexObservationHops(tx, lower, since); err != nil { + return err + } + _, err = tx.Exec(`UPDATE `+dbschema.ObservationHopsCoverageTable+` SET since = ? WHERE id = 1`, lower) + return err + }) + if err != nil { + return fmt.Errorf("backfill chunk [%d,%d): %w", lower, since, err) + } + rows += n + since = lower + if hopBackfillAfterChunk != nil { + hopBackfillAfterChunk() + } + if since <= floor { + break + } + // Yield between chunks, not after the last one. + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(hopBackfillPause): + } + } + log.Printf("[hop-index] backfill complete: %d rows in %s", rows, time.Since(start).Round(time.Second)) + return nil +} + +// TrimObservationHops slides the index window forward: it deletes the hop rows +// of the observations below now-hopIndexDays, one chunk per writer +// transaction, raising the coverage mark to the chunk's upper bound in the same +// transaction. Each chunk commits or rolls back whole, so a trim that stops +// midway leaves the mark exactly where its deletions end and the next trim +// continues from there. It waits for a running backfill (hopMaintMu). Returns +// hop rows deleted. +// +// Every transaction deletes from the current mark up to the new one, not only +// from the oldest observation it found: that read runs outside the transaction, +// and a row with a timestamp between the mark and that observation can arrive +// in between, indexed by the trigger because it is above the mark. Deleting the +// whole range the mark moves across removes it too, instead of leaving its hop +// rows under the raised mark where nothing would ever delete them. +func (s *Store) TrimObservationHops() (int64, error) { + if s.hopIndexDays <= 0 { + return 0, nil + } + s.hopMaintMu.Lock() + defer s.hopMaintMu.Unlock() + floor := s.hopFloor() + since, err := s.hopCoverage() + if err != nil { + return 0, err + } + // The backfill has not reached the floor yet (it lowers the mark down to + // it on its next run); there is nothing below the window to remove. + if since >= floor { + return 0, nil + } + // Size the chunks from the oldest observation at or above the mark, not + // from a possibly tiny mark: stepping from timestamp 0 in 15-minute chunks + // would take hundreds of thousands of empty transactions. + var minTS sql.NullInt64 + if err := s.db.QueryRow(`SELECT MIN(timestamp) FROM observations WHERE timestamp >= ?`, since).Scan(&minTS); err != nil { + return 0, fmt.Errorf("min observation timestamp: %w", err) + } + if hopMaintAfterRead != nil { + hopMaintAfterRead() + } + // Nothing indexed below the window: one transaction moves the mark to the + // floor (and still deletes across the range, in case a row arrived). + a := floor + if minTS.Valid && minTS.Int64 < floor { + a = minTS.Int64 + } + var total int64 + mark := since + for mark < floor { + b := a + hopBackfillChunkSeconds + if b > floor { + b = floor + } + var n int64 + err := s.WriterTx("observation_hops_trim", func(tx *sql.Tx) error { + if _, err := tx.Exec(`UPDATE `+dbschema.ObservationHopsCoverageTable+` SET since = ? WHERE id = 1`, b); err != nil { + return err + } + var err error + n, err = dbschema.UnindexObservationHops(tx, mark, b) + return err + }) + if err != nil { + return total, fmt.Errorf("trim chunk [%d,%d): %w", mark, b, err) + } + total += n + mark, a = b, b + } + return total, nil +} diff --git a/cmd/ingestor/observation_hops_backfill_test.go b/cmd/ingestor/observation_hops_backfill_test.go new file mode 100644 index 00000000..60828805 --- /dev/null +++ b/cmd/ingestor/observation_hops_backfill_test.go @@ -0,0 +1,656 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "os" + "sort" + "strings" + "testing" + "time" + + "github.com/meshcore-analyzer/dbschema" +) + +const day = int64(86400) + +func fastHopBackfill(t *testing.T, chunk int64) { + t.Helper() + c, p := hopBackfillChunkSeconds, hopBackfillPause + t.Cleanup(func() { hopBackfillChunkSeconds, hopBackfillPause = c, p }) + hopBackfillChunkSeconds, hopBackfillPause = chunk, 0 +} + +func hopCoverage(s *Store) (since int64) { + s.db.QueryRow(`SELECT since FROM observation_hops_coverage`).Scan(&since) + return +} + +func hopRows(s *Store) (n int) { + s.db.QueryRow(`SELECT COUNT(*) FROM observation_hops`).Scan(&n) + return +} + +// hopSchemaObjects lists the index's tables and triggers present in the DB. +func hopSchemaObjects(t *testing.T, s *Store) []string { + t.Helper() + rows, err := s.db.Query(`SELECT type || ' ' || name FROM sqlite_master + WHERE name LIKE 'observation_hops%' OR name LIKE 'trg_observation_hops%' ORDER BY name`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []string + for rows.Next() { + var n string + rows.Scan(&n) + out = append(out, n) + } + return out +} + +// checkHopIndex is the oracle for the index invariant: observation_hops holds +// exactly one row per (upper-cased text hop, observation) for every observation +// at or above the coverage mark, and nothing else. The expected set is built in +// Go from path_json, independently of the production SQL. +func checkHopIndex(t *testing.T, s *Store) { + t.Helper() + since := hopCoverage(s) + want := map[string]bool{} + rows, err := s.db.Query(`SELECT id, timestamp, path_json FROM observations WHERE path_json IS NOT NULL`) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var id, ts int64 + var pj string + rows.Scan(&id, &ts, &pj) + if ts < since { + continue + } + var elems []interface{} + if json.Unmarshal([]byte(pj), &elems) != nil { + continue + } + for _, e := range elems { + if h, ok := e.(string); ok { + want[fmt.Sprintf("%s|%d|%d", strings.ToUpper(h), ts, id)] = true + } + } + } + rows.Close() + got := map[string]bool{} + rows, err = s.db.Query(`SELECT hop, timestamp, observation_id FROM observation_hops`) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var h string + var ts, id int64 + rows.Scan(&h, &ts, &id) + got[fmt.Sprintf("%s|%d|%d", h, ts, id)] = true + } + rows.Close() + var missing, extra []string + for k := range want { + if !got[k] { + missing = append(missing, k) + } + } + for k := range got { + if !want[k] { + extra = append(extra, k) + } + } + sort.Strings(missing) + sort.Strings(extra) + if len(missing)+len(extra) > 0 { + t.Fatalf("hop index out of step with observations at/after since=%d:\n missing %v\n extra %v", since, missing, extra) + } +} + +func assertNear(t *testing.T, what string, got, want int64) { + t.Helper() + if got < want-5 || got > want+5 { + t.Fatalf("%s = %d, want ~%d", what, got, want) + } +} + +type ob = struct { + path string + ts int64 +} + +func insertObs(t *testing.T, s *Store, obs ...ob) { + t.Helper() + for i, o := range obs { + if _, err := s.db.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (1, ?, ?, ?)`, i, o.path, o.ts); err != nil { + t.Fatal(err) + } + } +} + +// legacyStore returns a store with the index on (days) whose observations +// predate it: no hop rows, and a coverage mark claiming only the present. +func legacyStore(t *testing.T, days int, obs ...ob) *Store { + t.Helper() + s := newTestStore(t) + if _, err := s.db.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen) VALUES (1,'00','h1','2026-01-01T00:00:00Z')`); err != nil { + t.Fatal(err) + } + insertObs(t, s, obs...) + if err := s.ConfigureObservationHops(days); err != nil { + t.Fatal(err) + } + return s +} + +// restart re-applies the setting and runs the boot-time backfill to completion, +// as main does on the next ingestor start. +func restart(t *testing.T, s *Store, days int) { + t.Helper() + if err := s.ConfigureObservationHops(days); err != nil { + t.Fatal(err) + } + if _, err := s.TrimObservationHops(); err != nil { + t.Fatal(err) + } + s.ScheduleObservationHopsBackfill() + s.WaitForAsyncMigrations() +} + +// The default install and an upgrade that never sets the key: no table, no +// triggers, no backfill, and ingest still works. +func TestHopIndex_OffByDefault_InstallsNothing(t *testing.T) { + s := newTestStore(t) + if got := hopSchemaObjects(t, s); len(got) != 0 { + t.Fatalf("OpenStore installed hop index objects: %v", got) + } + cfg := &Config{} + restart(t, s, cfg.HopIndexDaysOrZero()) + if got := hopSchemaObjects(t, s); len(got) != 0 { + t.Fatalf("hopIndexDays unset installed hop index objects: %v", got) + } + if _, err := s.db.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen) VALUES (1,'00','h1','2026-01-01T00:00:00Z')`); err != nil { + t.Fatal(err) + } + insertObs(t, s, ob{`["AA"]`, time.Now().Unix()}) + if n, err := s.TrimObservationHops(); n != 0 || err != nil { + t.Fatalf("trim with index off: %d, %v", n, err) + } +} + +func TestHopIndexDaysOrZero(t *testing.T) { + for _, c := range []struct { + r *RetentionConfig + want int + }{{nil, 0}, {&RetentionConfig{}, 0}, {&RetentionConfig{HopIndexDays: 14}, 14}} { + if got := (&Config{Retention: c.r}).HopIndexDaysOrZero(); got != c.want { + t.Errorf("HopIndexDaysOrZero(%+v) = %d, want %d", c.r, got, c.want) + } + } +} + +// Review #2214 round 2: observerDays reads -1 as "keep forever". The same value +// here would remove the index without a word, so LoadConfig refuses it; 0 and +// positive values load as before. +func TestLoadConfig_RejectsNegativeHopIndexDays(t *testing.T) { + path := t.TempDir() + "/config.json" + for _, c := range []struct { + json string + wantErr bool + }{ + {`{"retention":{"hopIndexDays":-1}}`, true}, + {`{"retention":{"hopIndexDays":-14}}`, true}, + {`{"retention":{"hopIndexDays":0}}`, false}, + {`{"retention":{"hopIndexDays":14}}`, false}, + {`{"retention":{}}`, false}, + } { + if err := os.WriteFile(path, []byte(c.json), 0o644); err != nil { + t.Fatal(err) + } + _, err := LoadConfig(path) + switch { + case c.wantErr && (err == nil || !strings.Contains(err.Error(), "retention.hopIndexDays")): + t.Errorf("%s: LoadConfig error = %v, want one naming retention.hopIndexDays", c.json, err) + case !c.wantErr && err != nil: + t.Errorf("%s: LoadConfig error = %v, want nil", c.json, err) + } + } +} + +// Switching the index off removes the table and triggers, and ingest, path +// repair and deletes keep working without them. Switching it back on rebuilds +// the window. +func TestHopIndex_OffRemovesEverything_OnRebuilds(t *testing.T) { + fastHopBackfill(t, 3600) + now := time.Now().Unix() + s := legacyStore(t, 14, ob{`["AA","BB"]`, now - 2*day}) + restart(t, s, 14) + checkHopIndex(t, s) + if hopRows(s) != 2 { + t.Fatalf("on: %d hop rows, want 2", hopRows(s)) + } + restart(t, s, 0) + if got := hopSchemaObjects(t, s); len(got) != 0 { + t.Fatalf("hopIndexDays 0 left objects behind: %v", got) + } + insertObs(t, s, ob{`["CC"]`, now - day}) + if _, err := s.db.Exec(`UPDATE observations SET path_json = '["DD"]' WHERE path_json = '["CC"]'`); err != nil { + t.Fatalf("path update with index off: %v", err) + } + if _, err := s.db.Exec(`DELETE FROM observations WHERE path_json = '["DD"]'`); err != nil { + t.Fatalf("delete with index off: %v", err) + } + insertObs(t, s, ob{`["EE"]`, now - 3*day}) + restart(t, s, 14) + checkHopIndex(t, s) + assertNear(t, "coverage after re-enable", hopCoverage(s), now-14*day) + if hopRows(s) != 3 { // AA, BB, EE + t.Fatalf("re-enabled: %d hop rows, want 3", hopRows(s)) + } +} + +func TestBackfillObservationHops_IndexesWindowAndLowersCoverageToFloor(t *testing.T) { + fastHopBackfill(t, 100) + now := time.Now().Unix() + s := legacyStore(t, 14, + ob{`["AA","BB"]`, now - 1000}, + ob{`["BB"]`, now - 800}, + ob{`not json`, now - 600}, + ob{`["CC","AA"]`, now - 400}, + ob{`["OLD"]`, now - 17*day}, // outside the window: must NOT be indexed + ) + if err := s.backfillObservationHops(context.Background()); err != nil { + t.Fatal(err) + } + if n := hopRows(s); n != 5 { // AA,BB | BB | - | CC,AA + t.Fatalf("hop rows = %d, want 5", n) + } + assertNear(t, "coverage", hopCoverage(s), now-14*day) + checkHopIndex(t, s) + // Idempotent / resumable. + if err := s.backfillObservationHops(context.Background()); err != nil { + t.Fatal(err) + } + if n := hopRows(s); n != 5 { + t.Fatalf("hop rows after rerun = %d, want 5", n) + } +} + +// Review #2214: raising hopIndexDays on an install whose backfill already +// finished must extend coverage on the next start, not leave it at the old +// window. +func TestHopIndex_RaisingWindowBackfillsOnNextStart(t *testing.T) { + fastHopBackfill(t, 3600) + now := time.Now().Unix() + s := legacyStore(t, 3, + ob{`["AA"]`, now - day}, + ob{`["BB"]`, now - 5*day}, + ob{`["CC"]`, now - 9*day}, + ob{`["DD"]`, now - 12*day}, // beyond even the raised window + ) + restart(t, s, 3) + assertNear(t, "coverage at 3 days", hopCoverage(s), now-3*day) + checkHopIndex(t, s) + if hopRows(s) != 1 { + t.Fatalf("3-day window: %d hop rows, want 1", hopRows(s)) + } + restart(t, s, 10) + assertNear(t, "coverage after raising to 10 days", hopCoverage(s), now-10*day) + checkHopIndex(t, s) + if hopRows(s) != 3 { + t.Fatalf("10-day window: %d hop rows, want 3", hopRows(s)) + } +} + +func TestHopIndex_LoweringWindowTrimsOnNextStart(t *testing.T) { + fastHopBackfill(t, 3600) + now := time.Now().Unix() + s := legacyStore(t, 10, ob{`["AA"]`, now - day}, ob{`["BB","CC"]`, now - 5*day}, ob{`["DD"]`, now - 9*day}) + restart(t, s, 10) + if hopRows(s) != 4 { + t.Fatalf("10-day window: %d hop rows, want 4", hopRows(s)) + } + restart(t, s, 3) + assertNear(t, "coverage after lowering to 3 days", hopCoverage(s), now-3*day) + checkHopIndex(t, s) + if hopRows(s) != 1 { + t.Fatalf("3-day window: %d hop rows, want 1", hopRows(s)) + } +} + +func TestBackfillObservationHops_StopsOnCancelKeepingProgress(t *testing.T) { + fastHopBackfill(t, 100) + now := time.Now().Unix() + s := legacyStore(t, 14, ob{`["AA"]`, now - 1000}) + before := hopCoverage(s) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := s.backfillObservationHops(ctx); err == nil { + t.Fatal("cancelled backfill must return an error") + } + if got := hopCoverage(s); got != before { + t.Fatalf("cancelled-before-start run moved coverage %d -> %d", before, got) + } + // Resumes on the next start. + restart(t, s, 14) + checkHopIndex(t, s) + assertNear(t, "coverage after resume", hopCoverage(s), now-14*day) +} + +// A bad-clock observer leaves rows near timestamp 0. The backfill must jump the +// gap, not walk empty chunks (which hung the whole test suite). +func TestBackfillObservationHops_SkipsHugeEmptyGaps(t *testing.T) { + fastHopBackfill(t, 3600) + now := time.Now().Unix() + s := legacyStore(t, 14, ob{`["AA"]`, 1}, ob{`["BB"]`, now - 3*day}) + done := make(chan error, 1) + go func() { done <- s.backfillObservationHops(context.Background()) }() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(10 * time.Second): + t.Fatal("backfill did not finish: walking empty chunks instead of skipping gaps") + } + if n := hopRows(s); n != 1 { + t.Fatalf("hop rows = %d, want 1 (only the in-window row)", n) + } + checkHopIndex(t, s) +} + +func TestTrimObservationHops_SlidesWindowAndDeletesByKey(t *testing.T) { + fastHopBackfill(t, 3600) + now := time.Now().Unix() + s := legacyStore(t, 14, + ob{`["AA","BB"]`, now - 12*day}, // stays + ob{`["CC"]`, now - 15*day}, // falls out + ob{`["DD","EE","FF"]`, now - 16*day}, + ) + // An older, deeper window was indexed (as with a larger setting before). + s.db.Exec(`UPDATE observation_hops_coverage SET since = ?`, now-19*day) + if _, err := dbschema.IndexObservationHops(s.db, 0, now); err != nil { + t.Fatal(err) + } + if n := hopRows(s); n != 6 { + t.Fatalf("setup: %d hop rows, want 6", n) + } + n, err := s.TrimObservationHops() + if err != nil { + t.Fatal(err) + } + if n != 4 || hopRows(s) != 2 { + t.Fatalf("trim deleted %d, left %d; want 4 deleted, 2 left", n, hopRows(s)) + } + assertNear(t, "coverage", hopCoverage(s), now-14*day) + checkHopIndex(t, s) + if n, _ := s.TrimObservationHops(); n != 0 { + t.Fatalf("second trim deleted %d, want 0 (nothing newly out of window)", n) + } +} + +// Stored hops are upper-case whichever path wrote them (insert trigger, update +// trigger, backfill), so a lookup matches what the case-insensitive LIKE scan +// matches. +func TestHopIndex_StoresHopsUpperCase(t *testing.T) { + fastHopBackfill(t, 3600) + now := time.Now().Unix() + s := legacyStore(t, 14, ob{`["ab12","Cd"]`, now - 2*day}) // indexed by the backfill + restart(t, s, 14) + insertObs(t, s, ob{`["ef"]`, now - day}) // indexed by the insert trigger + if _, err := s.db.Exec(`UPDATE observations SET path_json = '["gh","ab12"]' WHERE path_json = '["ef"]'`); err != nil { + t.Fatal(err) + } + checkHopIndex(t, s) + var lower int + s.db.QueryRow(`SELECT COUNT(*) FROM observation_hops WHERE hop != UPPER(hop)`).Scan(&lower) + if lower != 0 { + t.Fatalf("%d hop rows are not upper-case", lower) + } + // Deletes find the upper-cased rows. + s.db.Exec(`DELETE FROM observations`) + if n := hopRows(s); n != 0 { + t.Fatalf("delete left %d hop rows", n) + } +} + +// Review #2214: a trim must never interleave with a running backfill. Here the +// window shrinks while the backfill is past the new floor, which makes the trim +// want to delete rows the backfill has just indexed; if it ran between chunks, +// the next chunk would lower the mark again over the deleted rows. +func TestTrimWaitsForRunningBackfill(t *testing.T) { + fastHopBackfill(t, 3600) + now := time.Now().Unix() + var obs []ob + for d := int64(1); d <= 9; d++ { + obs = append(obs, ob{fmt.Sprintf(`["H%d"]`, d), now - d*day}) + } + s := legacyStore(t, 10, obs...) + trimDone := make(chan error, 1) + fired := false + t.Cleanup(func() { hopBackfillAfterChunk = nil }) + hopBackfillAfterChunk = func() { + if fired || hopCoverage(s) > now-5*day { + return + } + fired = true + s.hopIndexDays = 3 + go func() { _, err := s.TrimObservationHops(); trimDone <- err }() + select { + case <-trimDone: + t.Error("trim ran while the backfill was between chunks") + trimDone <- nil + case <-time.After(300 * time.Millisecond): + } + } + if err := s.backfillObservationHops(context.Background()); err != nil { + t.Fatal(err) + } + if !fired { + t.Fatal("test hook never fired") + } + select { + case err := <-trimDone: + if err != nil { + t.Fatal(err) + } + case <-time.After(10 * time.Second): + t.Fatal("trim never finished after the backfill") + } + checkHopIndex(t, s) + assertNear(t, "coverage", hopCoverage(s), now-3*day) +} + +// Scheduling the backfill inside OpenStore keeps live ingest dark for its whole +// duration: main gates ingest on WaitForAsyncMigrations(). It must be started +// explicitly, after the gate. +func TestOpenStoreDoesNotScheduleHopBackfill(t *testing.T) { + fastHopBackfill(t, 900) + now := time.Now().Unix() + s := legacyStore(t, 14, ob{`["AA"]`, now - day}) + s.WaitForAsyncMigrations() + if hopRows(s) != 0 { + t.Fatal("hop rows indexed before ScheduleObservationHopsBackfill; it would block the ingest gate") + } + s.ScheduleObservationHopsBackfill() + s.WaitForAsyncMigrations() + if hopRows(s) != 1 { + t.Fatalf("explicit schedule: %d hop rows, want 1", hopRows(s)) + } + checkHopIndex(t, s) +} + +// Both shipped defaults leave the index off: no config file (zero-config +// startup) and config.example.json, which docker/entrypoint.sh copies to +// config.json on a fresh install. On the store side anything at or below 0 is +// off; a negative value never gets this far from a config file +// (TestLoadConfig_RejectsNegativeHopIndexDays). +func TestHopIndex_ShippedDefaultsAreOff(t *testing.T) { + for _, path := range []string{"../../config.example.json", t.TempDir() + "/missing.json"} { + cfg, err := LoadConfig(path) + if err != nil { + t.Fatal(err) + } + if d := cfg.HopIndexDaysOrZero(); d != 0 { + t.Errorf("%s: hopIndexDays = %d, want 0 (opt-in)", path, d) + } + } + s := newTestStore(t) + restart(t, s, -1) + if got := hopSchemaObjects(t, s); len(got) != 0 { + t.Fatalf("hopIndexDays -1 installed %v", got) + } +} + +// A trim that fails midway (crash, disk full) must leave the mark exactly where +// its committed deletions end: readers never trust a range whose rows are gone. +func TestTrimStoppedMidwayKeepsMarkHonest(t *testing.T) { + fastHopBackfill(t, 3600) + now := time.Now().Unix() + s := legacyStore(t, 3, + ob{`["AA"]`, now - day}, + ob{`["BB"]`, now - 6*day}, + ob{`["ZZ"]`, now - 5*day}, // its chunk fails + ob{`["CC"]`, now - 4*day}, + ) + restart(t, s, 10) + if hopRows(s) != 4 { + t.Fatalf("setup: %d hop rows, want 4", hopRows(s)) + } + if _, err := s.db.Exec(`CREATE TRIGGER test_trim_fails BEFORE DELETE ON observation_hops + WHEN OLD.hop = 'ZZ' BEGIN SELECT RAISE(ABORT, 'injected'); END`); err != nil { + t.Fatal(err) + } + s.hopIndexDays = 3 + if _, err := s.TrimObservationHops(); err == nil { + t.Fatal("trim succeeded despite the injected failure") + } + checkHopIndex(t, s) + if since := hopCoverage(s); since > now-5*day { + t.Fatalf("mark %d moved past the failed chunk at %d", since, now-5*day) + } + if hopRows(s) != 3 { // BB's chunk committed; ZZ, CC, AA remain + t.Fatalf("%d hop rows after failed trim, want 3", hopRows(s)) + } + s.db.Exec(`DROP TRIGGER test_trim_fails`) + if _, err := s.TrimObservationHops(); err != nil { + t.Fatal(err) + } + checkHopIndex(t, s) + assertNear(t, "coverage after retry", hopCoverage(s), now-3*day) +} + +// Review #2214 round 2: the read that picks a chunk runs outside the chunk's +// transaction. A row whose timestamp falls inside the range that transaction is +// about to claim can arrive in between; the trigger skips it (the mark has not +// moved yet), so the transaction itself must index it. Both the chunk around an +// observation and the final step down to the floor get such a row. +func TestBackfill_RowArrivingBeforeTheMarkMovesIsIndexed(t *testing.T) { + fastHopBackfill(t, 3600) + fixed := time.Now().Truncate(time.Second) + t.Cleanup(func() { hopNow = time.Now }) + hopNow = func() time.Time { return fixed } + now := fixed.Unix() + floor := now - 14*day + s := legacyStore(t, 14, ob{`["AA"]`, now - day}) + t.Cleanup(func() { hopMaintAfterRead = nil }) + calls := 0 + hopMaintAfterRead = func() { + calls++ + switch calls { + case 1: // the chunk below AA is about to be claimed + insertObs(t, s, ob{`["LATE1"]`, now - day - 600}) + case 2: // nothing is left; the claim goes down to the floor + insertObs(t, s, ob{`["LATE2"]`, floor + 100}) + } + } + if err := s.backfillObservationHops(context.Background()); err != nil { + t.Fatal(err) + } + if calls != 2 { + t.Fatalf("hook fired %d times, want 2 (one chunk, then the step to the floor)", calls) + } + if got := hopCoverage(s); got != floor { + t.Fatalf("coverage = %d, want the floor %d", got, floor) + } + checkHopIndex(t, s) // both late rows are above the mark, so both must be indexed + if hopRows(s) != 3 { + t.Fatalf("%d hop rows, want 3 (AA, LATE1, LATE2)", hopRows(s)) + } +} + +// Review #2214 round 2: the trim reads the oldest observation outside its first +// transaction. A row with a timestamp between the mark and that observation can +// arrive in between; the trigger indexes it (it is above the mark), and a trim +// that deleted only from the observation it had found would leave those hop +// rows under the raised mark for good. Each transaction deletes from the current +// mark instead. +func TestTrim_RowArrivingBeforeTheMarkMovesIsRemoved(t *testing.T) { + fastHopBackfill(t, 3600) + now := time.Now().Unix() + // BB sits a non-whole number of chunks below the new floor, so the last + // chunk has to be clamped to the floor or the mark overshoots it. + s := legacyStore(t, 10, ob{`["AA"]`, now - day}, ob{`["BB"]`, now - 6*day - 1234}) + restart(t, s, 10) + if hopRows(s) != 2 { + t.Fatalf("setup: %d hop rows, want 2", hopRows(s)) + } + s.hopIndexDays = 3 // the window shrinks: BB falls out + t.Cleanup(func() { hopMaintAfterRead = nil }) + fired := false + hopMaintAfterRead = func() { + if fired { + return + } + fired = true + insertObs(t, s, ob{`["LATE"]`, now - 8*day}) // between the mark (10 d) and BB (6 d) + if n := hopRows(s); n != 3 { + t.Fatalf("the trigger did not index the late row: %d hop rows", n) + } + } + n, err := s.TrimObservationHops() + if err != nil { + t.Fatal(err) + } + if !fired { + t.Fatal("hook never fired") + } + if n != 2 { + t.Fatalf("trim deleted %d hop rows, want 2 (BB and LATE)", n) + } + assertNear(t, "coverage", hopCoverage(s), now-3*day) + checkHopIndex(t, s) // LATE is below the mark now: no hop row may remain for it + if hopRows(s) != 1 { + t.Fatalf("%d hop rows, want 1 (AA)", hopRows(s)) + } +} + +// The backfill stops exactly at the floor: a row at the floor second is indexed +// (and so covered), a row one second older is not, and the mark lands on the +// floor itself, not a chunk below it. +func TestBackfillObservationHops_StopsExactlyAtFloor(t *testing.T) { + fastHopBackfill(t, 3600) + fixed := time.Now().Truncate(time.Second) + t.Cleanup(func() { hopNow = time.Now }) + hopNow = func() time.Time { return fixed } + floor := fixed.Unix() - 14*day + s := legacyStore(t, 14, + ob{`["NEW"]`, fixed.Unix() - 10}, + ob{`["IN"]`, floor + 5000}, // more than one chunk above the floor + ob{`["EDGE"]`, floor}, + ob{`["OUT"]`, floor - 1}, + ) + if err := s.backfillObservationHops(context.Background()); err != nil { + t.Fatal(err) + } + if got := hopCoverage(s); got != floor { + t.Fatalf("coverage = %d, want exactly the floor %d", got, floor) + } + checkHopIndex(t, s) + if hopRows(s) != 3 { + t.Fatalf("%d hop rows, want 3 (NEW, IN, EDGE)", hopRows(s)) + } +} diff --git a/cmd/ingestor/startup_order_test.go b/cmd/ingestor/startup_order_test.go new file mode 100644 index 00000000..98e00e35 --- /dev/null +++ b/cmd/ingestor/startup_order_test.go @@ -0,0 +1,70 @@ +package main + +import ( + "go/ast" + "go/parser" + "go/token" + "testing" +) + +// The hop index is configured in main's startup block: after the ingest buffer +// and the MQTT connections are up (ingestBuffer.Start, client.Connect; the +// subscription follows the connect) and before the ingest gate opens +// (ingestBuffer.Ready). Removing the index holds the single writer for about a +// minute on a large database; run before the subscription, that minute's +// packets are lost instead of buffered (#2214 review, round 2). +// +// main has no seam for a behaviour test, so this pins the call order in its +// source. It is a guard, not the proof: the behaviour was checked against a +// broker with the built binaries. +func TestMainConfiguresHopIndexBetweenSubscribeAndGate(t *testing.T) { + fset := token.NewFileSet() + f, err := parser.ParseFile(fset, "main.go", nil, 0) + if err != nil { + t.Fatal(err) + } + var mainFn *ast.FuncDecl + for _, d := range f.Decls { + if fn, ok := d.(*ast.FuncDecl); ok && fn.Recv == nil && fn.Name.Name == "main" { + mainFn = fn + } + } + if mainFn == nil { + t.Fatal("func main not found in main.go") + } + order := []string{"ingestBuffer.Start", "client.Connect", "store.ConfigureObservationHops", "ingestBuffer.Ready"} + pos := map[string]token.Pos{} + ast.Inspect(mainFn.Body, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + sel, ok := call.Fun.(*ast.SelectorExpr) + if !ok { + return true + } + recv, ok := sel.X.(*ast.Ident) + if !ok { + return true + } + key := recv.Name + "." + sel.Sel.Name + for _, want := range order { + if key == want { + if _, dup := pos[key]; dup { + t.Fatalf("%s is called more than once in main; this test assumes one call site", key) + } + pos[key] = call.Pos() + } + } + return true + }) + for i, key := range order { + if pos[key] == token.NoPos { + t.Fatalf("%s not found in main", key) + } + if i > 0 && pos[key] <= pos[order[i-1]] { + t.Fatalf("%s (line %d) must come after %s (line %d) in main", + key, fset.Position(pos[key]).Line, order[i-1], fset.Position(pos[order[i-1]]).Line) + } + } +} diff --git a/cmd/server/node_reach.go b/cmd/server/node_reach.go index 3ec2567f..0d4ab925 100644 --- a/cmd/server/node_reach.go +++ b/cmd/server/node_reach.go @@ -15,6 +15,7 @@ import ( "time" "github.com/gorilla/mux" + "github.com/meshcore-analyzer/dbschema" "golang.org/x/sync/singleflight" ) @@ -769,6 +770,20 @@ func (s *Server) scanReachRows(ctx context.Context, tokens map[string]bool, sinc if len(tokens) == 0 { return nil, nil // defensive: an empty LIKE chain would render `AND ()` (SQL error) } + // Prefer the inverted hop index when the ingestor's backfill has covered + // this window: a token seek instead of a substring scan of every + // observation in the window (which, cold, outlives the HTTP WriteTimeout). + if s.reachHopIndexCovers(ctx, sinceEpoch) { + rows, err := s.scanReachRowsIndexed(ctx, tokens, sinceEpoch) + if err == nil { + return rows, nil + } + // The coverage check and the scan are separate queries, and the ingestor + // drops the index when hopIndexDays is set to 0: a removal between the + // two fails the scan with "no such table". Answer from the substring scan + // instead; an error there still reaches the caller as a 500. + log.Printf("[reach] hop index scan failed, using the path_json scan: %v", err) + } args := []interface{}{sinceEpoch} // Sort tokens so the generated SQL text is byte-stable across requests // with the same token set — preserves the driver's prepared-statement @@ -782,7 +797,14 @@ func (s *Server) scanReachRows(ctx context.Context, tokens map[string]bool, sinc args = append(args, "%\""+tok+"\"%") } args = append(args, reachScanRowLimit) - rows, err := s.db.conn.QueryContext(ctx, reachScanSQL(len(toks)), args...) + return s.readReachRows(ctx, reachScanSQL(len(toks)), args) +} + +// readReachRows runs a reach scan query (column order fixed by the callers) and +// folds the result into pathRows. Shared by the LIKE scan and the indexed scan +// so both apply identical row handling and truncation accounting. +func (s *Server) readReachRows(ctx context.Context, q string, args []interface{}) ([]pathRow, error) { + rows, err := s.db.conn.QueryContext(ctx, q, args...) if err != nil { log.Printf("[reach] scan query failed: %v", err) return nil, err @@ -844,3 +866,43 @@ func (s *Server) reachAcquireBuildSlot() (func(), bool) { return nil, false } } + +// reachHopIndexCovers reports whether observation_hops can answer a window that +// starts at sinceEpoch. The table is an optional accelerator maintained by the +// ingestor when retention.hopIndexDays > 0 (see +// internal/dbschema/observation_hops.go): a missing table or coverage row +// (index off, older DB, ingestor not yet started) just means "no". +func (s *Server) reachHopIndexCovers(ctx context.Context, sinceEpoch int64) bool { + var covered int64 + err := s.db.conn.QueryRowContext(ctx, + `SELECT since FROM `+dbschema.ObservationHopsCoverageTable+` WHERE id = 1`).Scan(&covered) + return err == nil && covered <= sinceEpoch +} + +// scanReachRowsIndexed is scanReachRows answered from observation_hops. It +// returns the same rows as the LIKE scan: hop tokens are matched as whole path +// elements (the LIKE pattern "%\"TOK\"%" is exactly that), both sides are +// upper-cased with SQLite's ASCII UPPER (LIKE ignores ASCII case; the index +// stores UPPER(hop)), and IN (subquery) keeps an observation that matches +// several tokens to one row. +func (s *Server) scanReachRowsIndexed(ctx context.Context, tokens map[string]bool, sinceEpoch int64) ([]pathRow, error) { + toks := make([]string, 0, len(tokens)) + for tok := range tokens { + toks = append(toks, tok) + } + sort.Strings(toks) + args := make([]interface{}, 0, len(toks)+2) + marks := make([]string, len(toks)) + for i, tok := range toks { + marks[i] = "UPPER(?)" + args = append(args, tok) + } + args = append(args, sinceEpoch, reachScanRowLimit) + q := `SELECT LOWER(COALESCE(obs.id,'')), LOWER(COALESCE(t.from_pubkey,'')), COALESCE(t.payload_type,0), o.path_json, o.snr + FROM observations o + JOIN transmissions t ON t.id = o.transmission_id + LEFT JOIN observers obs ON obs.rowid = o.observer_idx + WHERE o.id IN (SELECT observation_id FROM observation_hops WHERE hop IN (` + strings.Join(marks, ",") + `) AND timestamp >= ?) + LIMIT ?` + return s.readReachRows(ctx, q, args) +} diff --git a/cmd/server/node_reach_hops_bench_test.go b/cmd/server/node_reach_hops_bench_test.go new file mode 100644 index 00000000..d3580ff1 --- /dev/null +++ b/cmd/server/node_reach_hops_bench_test.go @@ -0,0 +1,99 @@ +package main + +import ( + "context" + "database/sql" + "fmt" + "math/rand" + "path/filepath" + "testing" + + "github.com/meshcore-analyzer/dbschema" +) + +// BenchmarkScanReachRows compares the path_json LIKE scan with the +// observation_hops index on a synthetic DB: 11 hops per path on average (the +// real mean is 9.6 to 12.2), a 7-day window inside 14 days of data, one hot +// relay token in about 3% of paths and one rare token in about 0.01%. +// +// go test ./cmd/server -run xxx -bench ScanReachRows -benchtime 5x +// +// The cold-cache, EBS-bound case that motivated the index is not reproducible +// in a unit benchmark; see the PR description for measurements on real data. +func BenchmarkScanReachRows(b *testing.B) { + const ( + nObs = 600_000 + window = 7 * 86400 + span = 14 * 86400 + now = int64(1_800_000_000) + ) + conn, err := sql.Open("sqlite3", "file:"+filepath.Join(b.TempDir(), "r.db")+"?_journal_mode=WAL") + if err != nil { + b.Fatal(err) + } + defer conn.Close() + conn.SetMaxOpenConns(1) + for _, s := range []string{ + `CREATE TABLE transmissions (id INTEGER PRIMARY KEY, from_pubkey TEXT, payload_type INTEGER)`, + `CREATE TABLE observers (id TEXT, inactive INTEGER)`, + `INSERT INTO observers (id) VALUES ('OBS1')`, + `CREATE TABLE observations (id INTEGER PRIMARY KEY, transmission_id INTEGER, observer_idx INTEGER, snr REAL, path_json TEXT, timestamp INTEGER)`, + `CREATE INDEX idx_observations_timestamp ON observations(timestamp)`, + } { + if _, err := conn.Exec(s); err != nil { + b.Fatal(err) + } + } + rng := rand.New(rand.NewSource(1)) + tx, _ := conn.Begin() + for i := 1; i <= nObs; i++ { + tx.Exec(`INSERT INTO transmissions (id, from_pubkey, payload_type) VALUES (?, 'FF00', 4)`, i) + p := "[" + for h, n := 0, 1+rng.Intn(21); h < n; h++ { + if h > 0 { + p += "," + } + tok := fmt.Sprintf("%02X%02X", rng.Intn(256), rng.Intn(256)) + switch r := rng.Intn(10000); { + case r < 300: + tok = "HOT1" + case r < 301: + tok = "RARE" + } + p += `"` + tok + `"` + } + p += "]" + tx.Exec(`INSERT INTO observations (id, transmission_id, observer_idx, snr, path_json, timestamp) VALUES (?,?,1,-5,?,?)`, + i, i, p, now-int64(rng.Intn(span))) + } + tx.Commit() + if err := dbschema.EnsureObservationHopsSchema(conn, nil); err != nil { + b.Fatal(err) + } + // Index everything and claim full coverage, as a finished backfill does. + conn.Exec(`INSERT OR IGNORE INTO observation_hops SELECT DISTINCT j.value, o.timestamp, o.id FROM observations o, json_each(o.path_json) j`) + srv := &Server{db: &DB{conn: conn}} + since := now - window + for _, tc := range []struct{ name, tok string }{{"hot_token", "HOT1"}, {"rare_token", "RARE"}} { + for _, indexed := range []bool{false, true} { + mode := "like" + if indexed { + mode = "indexed" + conn.Exec(`UPDATE observation_hops_coverage SET since = 0`) + } else { + conn.Exec(`UPDATE observation_hops_coverage SET since = ?`, now+1) // not covered: LIKE path + } + b.Run(tc.name+"/"+mode, func(b *testing.B) { + for i := 0; i < b.N; i++ { + rows, err := srv.scanReachRows(context.Background(), map[string]bool{tc.tok: true}, since) + if err != nil { + b.Fatal(err) + } + if len(rows) == 0 { + b.Fatal("no rows") + } + } + }) + } + } +} diff --git a/cmd/server/node_reach_hops_test.go b/cmd/server/node_reach_hops_test.go new file mode 100644 index 00000000..dff86a2e --- /dev/null +++ b/cmd/server/node_reach_hops_test.go @@ -0,0 +1,158 @@ +package main + +import ( + "context" + "fmt" + "math" + "sort" + "testing" + + "github.com/meshcore-analyzer/dbschema" +) + +// canon makes pathRow sets comparable regardless of row order. +func canon(rows []pathRow) []string { + out := make([]string, 0, len(rows)) + for _, r := range rows { + out = append(out, fmt.Sprintf("%s|%s|%d|%v|%v|%.1f", r.observerPK, r.fromPubkey, r.payloadType, r.path, r.snrValid, r.snr)) + } + sort.Strings(out) + return out +} + +func newHopIndexedTestDB(t *testing.T) *DB { + t.Helper() + db := newReachScanTestDB(t) + db.conn.SetMaxOpenConns(1) + // Extra rows: mixed hop widths, a path hitting two tokens, one row outside + // the window, and one lower-case hop (LIKE ignores ASCII case, so the index + // must too). Inserted BEFORE the schema so the backfill path is what + // indexes them, as on an upgraded DB. + for _, s := range []string{ + `INSERT INTO transmissions (id, from_pubkey, payload_type) VALUES (4,'',5),(5,'',5),(6,'',5),(7,'',5)`, + `INSERT INTO observations (id, transmission_id, observer_idx, snr, path_json, timestamp) VALUES + (4,4,1,1.0,'["01FA","01FAB2"]',2000), + (5,5,1,2.0,'["01FAB2","DD"]',2000), + (6,6,1,3.0,'["01FA"]',10), + (7,7,1,4.0,'["01fab2"]',2000)`, + } { + if _, err := db.conn.Exec(s); err != nil { + t.Fatal(err) + } + } + return db +} + +// The indexed scan must return exactly what the LIKE scan returns, for upper- +// and lower-case tokens alike. +func TestScanReachRows_IndexedMatchesLike(t *testing.T) { + for _, toks := range []map[string]bool{ + {"01FA": true, "01FAB2": true}, + {"01fa": true, "01fab2": true}, + } { + t.Run(fmt.Sprint(toks), func(t *testing.T) { indexedMatchesLike(t, toks) }) + } +} + +func indexedMatchesLike(t *testing.T, toks map[string]bool) { + db := newHopIndexedTestDB(t) + defer db.conn.Close() + srv := &Server{db: db} + + if srv.reachHopIndexCovers(context.Background(), 500) { + t.Fatal("no hop schema (index off) must not count as coverage") + } + like, err := srv.scanReachRows(context.Background(), toks, 500) // no hop schema → LIKE + if err != nil { + t.Fatal(err) + } + + if err := dbschema.EnsureObservationHopsSchema(db.conn, nil); err != nil { + t.Fatal(err) + } + // The backfill's chunk function over all rows, then claim full coverage. + if _, err := dbschema.IndexObservationHops(db.conn, 0, math.MaxInt64); err != nil { + t.Fatal(err) + } + if srv.reachHopIndexCovers(context.Background(), 500) { + // seeded coverage is "now": window starting at 500 is NOT covered yet. + t.Fatal("coverage seeded at now must not cover an old window before the backfill") + } + db.conn.Exec(`UPDATE observation_hops_coverage SET since = 0`) + if !srv.reachHopIndexCovers(context.Background(), 500) { + t.Fatal("coverage since=0 must cover the window") + } + + idx, err := srv.scanReachRows(context.Background(), toks, 500) + if err != nil { + t.Fatal(err) + } + a, b := canon(like), canon(idx) + if len(a) != 5 { // obs 1,2,4,5,7 — not 3 (no token) and not 6 (before window) + t.Fatalf("LIKE scan returned %d rows, want 5: %v", len(a), a) + } + if fmt.Sprint(a) != fmt.Sprint(b) { + t.Fatalf("indexed scan differs from LIKE scan:\nlike %v\nindexed %v", a, b) + } +} + +// Review #2214 round 2: the coverage check and the indexed scan are separate +// queries, and the ingestor removes the index (hopIndexDays set to 0) in one +// transaction between them for a request that is in flight. That request must +// still get its answer, from the substring scan; and a failure in both paths +// still surfaces as an error (#1631), not as an empty reach. +func TestScanReachRows_IndexRemovedMidRequestFallsBackToLike(t *testing.T) { + db := newHopIndexedTestDB(t) + defer db.conn.Close() + srv := &Server{db: db} + toks := map[string]bool{"01FA": true, "01FAB2": true} + like, err := srv.scanReachRows(context.Background(), toks, 500) + if err != nil { + t.Fatal(err) + } + if err := dbschema.EnsureObservationHopsSchema(db.conn, nil); err != nil { + t.Fatal(err) + } + db.conn.Exec(`UPDATE observation_hops_coverage SET since = 0`) + // The state the in-flight request sees: the coverage row its check read, + // and no table left to scan. + if _, err := db.conn.Exec(`DROP TABLE observation_hops`); err != nil { + t.Fatal(err) + } + if !srv.reachHopIndexCovers(context.Background(), 500) { + t.Fatal("setup: the coverage check must still pass") + } + rows, err := srv.scanReachRows(context.Background(), toks, 500) + if err != nil { + t.Fatalf("scan with the hop table gone: %v, want the LIKE scan's answer", err) + } + if a, b := canon(like), canon(rows); fmt.Sprint(a) != fmt.Sprint(b) { + t.Fatalf("fallback differs from the LIKE scan:\nlike %v\nfallback %v", a, b) + } + if _, err := db.conn.Exec(`DROP TABLE observations`); err != nil { + t.Fatal(err) + } + if _, err := srv.scanReachRows(context.Background(), toks, 500); err == nil { + t.Fatal("both scans failed, but scanReachRows returned no error") + } +} + +// A window that starts before the coverage mark must use the LIKE scan, so a +// half-finished backfill can never silently drop old observations. +func TestScanReachRows_PartialCoverageFallsBackToLike(t *testing.T) { + db := newHopIndexedTestDB(t) + defer db.conn.Close() + srv := &Server{db: db} + if err := dbschema.EnsureObservationHopsSchema(db.conn, nil); err != nil { + t.Fatal(err) + } + // Index is empty (backfill not run) and since = now: a correct reader must + // ignore it and still find rows. + rows, err := srv.scanReachRows(context.Background(), map[string]bool{"01FA": true}, 0) + if err != nil { + t.Fatal(err) + } + if len(rows) != 4 { // obs 1,2,4,6 contain exactly "01FA" + t.Fatalf("fallback scan returned %d rows, want 4", len(rows)) + } +} diff --git a/config.example.json b/config.example.json index 70049f56..847d4111 100644 --- a/config.example.json +++ b/config.example.json @@ -16,7 +16,8 @@ "clientRxObsDays": 30, "clientRfDays": 30, "clientRegionsDays": 30, - "_comment": "nodeDays: nodes not seen in N days moved to inactive_nodes (default 7). observerDays: observers not sending data in N days are removed (-1 = keep forever, default 14). observerPurgeDays: observers already marked inactive AND not seen in N days are hard-deleted, but only when no observation, observer_metrics or dropped_packets row still references them (0 = disabled, the default). Set it above both observerDays and packetDays — below those the reference guards keep every row anyway. packetDays: transmissions older than N days are deleted (0 = disabled). clientRxDays: mobile client-RX coverage rows (client_receptions/client_observers) older than N days are deleted (0 = disabled) — bounds the opt-in coverage tables (#1727). clientRxObsDays: diagnostic client_rx_observations rows (by rx_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRxObservations table. clientRfDays: RF environment sample rows (client_rf_samples, by sampled_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRfSamples table; unset means unbounded growth at roughly 240 rows/hour per driver. clientRegionsDays: region-discovery answer rows (node_declared_regions, by observed_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRegions table. NOTE (#1283): all retention fields are consumed by the INGESTOR process. The server is read-only and never prunes." + "hopIndexDays": 0, + "_comment": "nodeDays: nodes not seen in N days moved to inactive_nodes (default 7). observerDays: observers not sending data in N days are removed (-1 = keep forever, default 14). observerPurgeDays: observers already marked inactive AND not seen in N days are hard-deleted, but only when no observation, observer_metrics or dropped_packets row still references them (0 = disabled, the default). Set it above both observerDays and packetDays — below those the reference guards keep every row anyway. packetDays: transmissions older than N days are deleted (0 = disabled). clientRxDays: mobile client-RX coverage rows (client_receptions/client_observers) older than N days are deleted (0 = disabled) — bounds the opt-in coverage tables (#1727). clientRxObsDays: diagnostic client_rx_observations rows (by rx_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRxObservations table. clientRfDays: RF environment sample rows (client_rf_samples, by sampled_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRfSamples table; unset means unbounded growth at roughly 240 rows/hour per driver. clientRegionsDays: region-discovery answer rows (node_declared_regions, by observed_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRegions table. hopIndexDays: opt-in index that speeds up the node reach panel (/api/nodes/{pk}/reach) on large databases; 0 = off (the default): no table, no triggers. Negative values are rejected at startup (unlike observerDays, -1 does not mean keep forever). N > 0 indexes the last N days (14 covers the UI's 7-day default with margin). Costs roughly 200 bytes per observation (about 30% of the observations table) and a trigger on every observation insert; reach windows longer than N days use the slower path_json scan. Changes apply at the next ingestor start: raising N backfills the older days in the background, lowering it trims, and 0 removes the table and triggers (about a minute on a large database; the ingestor is already subscribed and buffers packets meanwhile, up to ingestBufferSize). Before rolling back to a version without this setting, set it to 0 and start the ingestor once; an older ingestor keeps the triggers but never trims, so the table would grow without limit. NOTE (#1283): all retention fields are consumed by the INGESTOR process. The server is read-only and never prunes." }, "db": { "vacuumOnStartup": false, diff --git a/internal/dbschema/observation_hops.go b/internal/dbschema/observation_hops.go new file mode 100644 index 00000000..c912b0e4 --- /dev/null +++ b/internal/dbschema/observation_hops.go @@ -0,0 +1,221 @@ +package dbschema + +import ( + "database/sql" + "fmt" +) + +// observation_hops is an inverted index over observations.path_json: one row +// per (hop token, observation). It exists so "which observations relayed +// through hop X in the last N days" is an index range seek instead of a +// substring scan of path_json over every observation in the window. +// +// The reach endpoint was the motivating reader: on a multi-GB DB the LIKE scan +// reads gigabytes of observations from a cold page cache and outlives the HTTP +// WriteTimeout, so the request 500s and nothing is ever cached. +// +// Layout: +// +// observation_hops(hop, timestamp, observation_id) PRIMARY KEY, WITHOUT ROWID +// +// The PK leads with hop so a token lookup is a seek, then timestamp so the +// window bound is a range on the same b-tree, then observation_id to make the +// row unique. WITHOUT ROWID keeps it a single b-tree (no separate rowid index). +// +// Maintenance is by trigger, not by writer code, so every path that touches +// observations.path_json keeps the index in step and the prune/purge paths need +// no changes: the delete trigger removes exactly the rows the insert trigger +// created, by primary key (hop, timestamp, observation_id), never a scan. +// +// Opt-in: none of this is part of Apply. The ingestor installs it only when +// retention.hopIndexDays > 0 (EnsureObservationHopsSchema) and removes it when +// the setting is 0 or unset (DropObservationHopsSchema), so an install that +// never sets the key has no table, no triggers and no per-insert cost. +// +// Invariant: observation_hops holds rows for exactly the observations with +// timestamp >= observation_hops_coverage.since, and none below it. Real paths +// average 10 to 12 hops, so the index holds that many rows (roughly 200 bytes) +// per observation and cannot be kept for the full packet retention on a large +// instance; the ingestor therefore bounds it to a window (see +// cmd/ingestor/observation_hops_backfill.go) and the triggers ignore anything +// below the mark. +// +// Case: hops are stored upper-cased (SQLite UPPER, ASCII only) and readers +// upper-case their tokens the same way, so a lookup matches exactly what the +// case-insensitive LIKE "%\"TOK\"%" scan matches. The decoder already writes +// upper-case hex; this makes the two paths equal by construction, not by habit. +// +// Coverage: the index is only trustworthy for timestamps at or above +// observation_hops_coverage.since. A fresh install of the triggers sets since to +// "now" (only new inserts are indexed); a chunked backfill then walks backwards +// and lowers since as each chunk commits. Readers MUST fall back to the +// substring scan for windows that start before since. +// +// The server never asserts this schema (see AssertReady): it is an optional +// accelerator, and a server whose DB has no index simply uses the slow path. + +// ObservationHopsCoverageTable is read by cmd/server to decide whether the +// index may answer a query window. +const ObservationHopsCoverageTable = "observation_hops_coverage" + +// observationHopsTriggers lists the triggers EnsureObservationHopsSchema +// creates and DropObservationHopsSchema removes. +var observationHopsTriggers = []string{"trg_observation_hops_ai", "trg_observation_hops_ad", "trg_observation_hops_au"} + +// EnsureObservationHopsSchema installs the index tables and (re)creates its +// triggers. The triggers are dropped and created on every call, so a changed +// definition reaches an existing install; the tables and the coverage mark are +// kept. Called by the ingestor when retention.hopIndexDays > 0, and by tests +// that build minimal fixture DBs. +func EnsureObservationHopsSchema(rw *sql.DB, logf Logger) error { + if logf == nil { + logf = func(string, ...interface{}) {} + } + tx, err := rw.Begin() + if err != nil { + return err + } + defer tx.Rollback() + + // PREFLIGHT: async=false reason="creates empty tables and triggers only — no row scan or rewrite of observations; the backfill is a separate chunked, resumable pass run after the ingest gate (ScheduleObservationHopsBackfill)" + stmts := []string{ + `CREATE TABLE IF NOT EXISTS observation_hops ( + hop TEXT NOT NULL, + timestamp INTEGER NOT NULL, + observation_id INTEGER NOT NULL, + PRIMARY KEY (hop, timestamp, observation_id) + ) WITHOUT ROWID`, + `CREATE TABLE IF NOT EXISTS ` + ObservationHopsCoverageTable + ` ( + id INTEGER PRIMARY KEY CHECK (id = 1), + since INTEGER NOT NULL + )`, + } + for _, name := range observationHopsTriggers { + stmts = append(stmts, `DROP TRIGGER IF EXISTS `+name) + } + stmts = append(stmts, + // json_valid guards every trigger: json_each raises on malformed JSON, + // and a trigger error would fail the observation INSERT itself. + `CREATE TRIGGER trg_observation_hops_ai + AFTER INSERT ON observations + WHEN NEW.path_json IS NOT NULL AND json_valid(NEW.path_json) + AND NEW.timestamp >= (SELECT since FROM observation_hops_coverage WHERE id = 1) + BEGIN + INSERT OR IGNORE INTO observation_hops (hop, timestamp, observation_id) + SELECT DISTINCT UPPER(value), NEW.timestamp, NEW.id FROM json_each(NEW.path_json) WHERE type = 'text'; + END`, + `CREATE TRIGGER trg_observation_hops_ad + AFTER DELETE ON observations + WHEN OLD.path_json IS NOT NULL AND json_valid(OLD.path_json) + AND OLD.timestamp >= (SELECT since FROM observation_hops_coverage WHERE id = 1) + BEGIN + DELETE FROM observation_hops + WHERE observation_id = OLD.id AND timestamp = OLD.timestamp + AND hop IN (SELECT UPPER(value) FROM json_each(OLD.path_json) WHERE type = 'text'); + END`, + // The path-repair pass in the ingestor rewrites path_json in place. + `CREATE TRIGGER trg_observation_hops_au + AFTER UPDATE OF path_json ON observations + WHEN IFNULL(OLD.path_json, '') != IFNULL(NEW.path_json, '') + BEGIN + DELETE FROM observation_hops + WHERE observation_id = OLD.id AND timestamp = OLD.timestamp + AND json_valid(OLD.path_json) + AND OLD.timestamp >= (SELECT since FROM observation_hops_coverage WHERE id = 1) + AND hop IN (SELECT UPPER(value) FROM json_each(OLD.path_json) WHERE type = 'text'); + INSERT OR IGNORE INTO observation_hops (hop, timestamp, observation_id) + SELECT DISTINCT UPPER(value), NEW.timestamp, NEW.id FROM json_each(NEW.path_json) + WHERE json_valid(NEW.path_json) AND type = 'text' + AND NEW.timestamp >= (SELECT since FROM observation_hops_coverage WHERE id = 1); + END`, + ) + for _, s := range stmts { + if _, err := tx.Exec(s); err != nil { + return fmt.Errorf("observation_hops schema: %w", err) + } + } + // Seed coverage only if absent. From this instant the triggers index every + // new row, so "now" is the correct starting claim; the backfill lowers it. + res, err := tx.Exec(`INSERT OR IGNORE INTO ` + ObservationHopsCoverageTable + + ` (id, since) VALUES (1, CAST(strftime('%s','now') AS INTEGER))`) + if err != nil { + return fmt.Errorf("seed observation_hops coverage: %w", err) + } + seeded, _ := res.RowsAffected() + if err := tx.Commit(); err != nil { + return err + } + if seeded > 0 { + logf("[dbschema] observation_hops installed; backfill pending") + } + return nil +} + +// DropObservationHopsSchema removes the index: its triggers first (so no insert +// touches a half-dropped table), then both tables. Reports whether anything was +// there. Called by the ingestor when retention.hopIndexDays is 0 or unset, so +// switching the setting off leaves nothing behind; it is also the documented +// step before rolling back to a version that does not know this setting. +func DropObservationHopsSchema(rw *sql.DB) (bool, error) { + var n int + if err := rw.QueryRow(`SELECT COUNT(*) FROM sqlite_master + WHERE name IN ('observation_hops', '` + ObservationHopsCoverageTable + `', 'trg_observation_hops_ai', 'trg_observation_hops_ad', 'trg_observation_hops_au')`).Scan(&n); err != nil { + return false, err + } + if n == 0 { + return false, nil + } + tx, err := rw.Begin() + if err != nil { + return false, err + } + defer tx.Rollback() + stmts := []string{} + for _, name := range observationHopsTriggers { + stmts = append(stmts, `DROP TRIGGER IF EXISTS `+name) + } + stmts = append(stmts, `DROP TABLE IF EXISTS observation_hops`, `DROP TABLE IF EXISTS `+ObservationHopsCoverageTable) + for _, s := range stmts { + if _, err := tx.Exec(s); err != nil { + return false, fmt.Errorf("drop observation_hops: %w", err) + } + } + return true, tx.Commit() +} + +// Execer is satisfied by *sql.DB and *sql.Tx. +type Execer interface { + Exec(query string, args ...interface{}) (sql.Result, error) +} + +// IndexObservationHops adds the hop rows of the observations with +// lo <= timestamp < hi, the same rows the insert trigger writes. Used by the +// ingestor's backfill, one chunk per writer transaction. Returns rows added. +func IndexObservationHops(ex Execer, lo, hi int64) (int64, error) { + res, err := ex.Exec(` + INSERT OR IGNORE INTO observation_hops (hop, timestamp, observation_id) + SELECT DISTINCT UPPER(j.value), o.timestamp, o.id + FROM observations o, json_each(o.path_json) j + WHERE o.timestamp >= ? AND o.timestamp < ? + AND json_valid(o.path_json) AND j.type = 'text'`, lo, hi) + if err != nil { + return 0, err + } + return res.RowsAffected() +} + +// UnindexObservationHops deletes the hop rows of the observations with +// lo <= timestamp < hi, by primary key. Used by the ingestor's trim. Returns +// rows deleted. +func UnindexObservationHops(ex Execer, lo, hi int64) (int64, error) { + res, err := ex.Exec(` + DELETE FROM observation_hops WHERE (hop, timestamp, observation_id) IN ( + SELECT UPPER(j.value), o.timestamp, o.id + FROM observations o, json_each(o.path_json) j + WHERE o.timestamp >= ? AND o.timestamp < ? + AND json_valid(o.path_json) AND j.type = 'text')`, lo, hi) + if err != nil { + return 0, err + } + return res.RowsAffected() +} diff --git a/internal/dbschema/observation_hops_bench_test.go b/internal/dbschema/observation_hops_bench_test.go new file mode 100644 index 00000000..7855b2dd --- /dev/null +++ b/internal/dbschema/observation_hops_bench_test.go @@ -0,0 +1,53 @@ +package dbschema + +import ( + "database/sql" + "path/filepath" + "testing" +) + +// BenchmarkObservationInsert measures what the observation_hops triggers add +// to the ingest write path. Each iteration is one observation in its own +// transaction on a file DB opened with WriterDSN (WAL, synchronous=FULL), the +// way the ingestor commits, with a realistic path of 11 hops. +// +// go test ./internal/dbschema -run xxx -bench ObservationInsert -benchtime 20000x +func BenchmarkObservationInsert(b *testing.B) { + path := `["0B","1C","D2","25","E4","3D","9B","CF","A1","77","5E"]` + for _, withIndex := range []bool{false, true} { + name := "without_index" + if withIndex { + name = "with_index" + } + b.Run(name, func(b *testing.B) { + db, err := sql.Open("sqlite3", WriterDSN(filepath.Join(b.TempDir(), "w.db"))) + if err != nil { + b.Fatal(err) + } + defer db.Close() + db.SetMaxOpenConns(1) + if _, err := db.Exec(`CREATE TABLE observations (id INTEGER PRIMARY KEY AUTOINCREMENT, path_json TEXT, timestamp INTEGER NOT NULL)`); err != nil { + b.Fatal(err) + } + if withIndex { + if err := EnsureObservationHopsSchema(db, nil); err != nil { + b.Fatal(err) + } + db.Exec(`UPDATE ` + ObservationHopsCoverageTable + ` SET since = 0`) + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + tx, err := db.Begin() + if err != nil { + b.Fatal(err) + } + if _, err := tx.Exec(`INSERT INTO observations (path_json, timestamp) VALUES (?, ?)`, path, 1_800_000_000+i); err != nil { + b.Fatal(err) + } + if err := tx.Commit(); err != nil { + b.Fatal(err) + } + } + }) + } +} diff --git a/internal/dbschema/observation_hops_test.go b/internal/dbschema/observation_hops_test.go new file mode 100644 index 00000000..eac38486 --- /dev/null +++ b/internal/dbschema/observation_hops_test.go @@ -0,0 +1,261 @@ +package dbschema + +import ( + "database/sql" + "reflect" + "sort" + "testing" + + _ "github.com/mattn/go-sqlite3" +) + +func hopTestDB(t *testing.T) *sql.DB { + t.Helper() + db, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatal(err) + } + db.SetMaxOpenConns(1) // :memory: is per-connection + if _, err := db.Exec(`CREATE TABLE observations (id INTEGER PRIMARY KEY AUTOINCREMENT, path_json TEXT, timestamp INTEGER NOT NULL)`); err != nil { + t.Fatal(err) + } + if err := EnsureObservationHopsSchema(db, nil); err != nil { + t.Fatal(err) + } + // Fresh installs seed coverage at "now"; most tests use small timestamps. + db.Exec(`UPDATE ` + ObservationHopsCoverageTable + ` SET since = 0`) + return db +} + +func hopsFor(t *testing.T, db *sql.DB, id int) []string { + t.Helper() + rows, err := db.Query(`SELECT hop FROM observation_hops WHERE observation_id = ? ORDER BY hop`, id) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []string + for rows.Next() { + var h string + rows.Scan(&h) + out = append(out, h) + } + return out +} + +func TestObservationHops_TriggersTrackInsertUpdateDelete(t *testing.T) { + db := hopTestDB(t) + defer db.Close() + db.Exec(`INSERT INTO observations (path_json, timestamp) VALUES ('["AA","01FA","AA"]', 100)`) // id 1, dup hop + if got, want := hopsFor(t, db, 1), []string{"01FA", "AA"}; !reflect.DeepEqual(got, want) { + t.Fatalf("after insert: %v want %v", got, want) + } + db.Exec(`UPDATE observations SET path_json = '["BB"]' WHERE id = 1`) + if got, want := hopsFor(t, db, 1), []string{"BB"}; !reflect.DeepEqual(got, want) { + t.Fatalf("after update: %v want %v", got, want) + } + db.Exec(`DELETE FROM observations WHERE id = 1`) + var n int + db.QueryRow(`SELECT COUNT(*) FROM observation_hops`).Scan(&n) + if n != 0 { + t.Fatalf("delete left %d hop rows", n) + } +} + +// A trigger error would fail the observation INSERT itself, taking ingest down. +func TestObservationHops_BadPathJSONNeverBlocksInsert(t *testing.T) { + db := hopTestDB(t) + defer db.Close() + for _, p := range []interface{}{nil, "", "[]", "not json", `{"a":1}`, `[1,2]`} { + if _, err := db.Exec(`INSERT INTO observations (path_json, timestamp) VALUES (?, 100)`, p); err != nil { + t.Fatalf("insert path_json=%v failed: %v", p, err) + } + } + if _, err := db.Exec(`UPDATE observations SET path_json = 'garbage'`); err != nil { + t.Fatalf("update failed: %v", err) + } + if _, err := db.Exec(`DELETE FROM observations`); err != nil { + t.Fatalf("delete failed: %v", err) + } +} + +func TestObservationHops_IdempotentAndCoverageSeededOnce(t *testing.T) { + db := hopTestDB(t) + defer db.Close() + db.Exec(`UPDATE ` + ObservationHopsCoverageTable + ` SET since = 5`) + if err := EnsureObservationHopsSchema(db, nil); err != nil { + t.Fatal(err) + } + var since int64 + db.QueryRow(`SELECT since FROM ` + ObservationHopsCoverageTable).Scan(&since) + if since != 5 { + t.Fatalf("re-running Apply reset coverage to %d", since) + } +} + +// The reach query is a seek on (hop, timestamp): pin that it never degrades to +// a scan of the whole table. +func TestObservationHops_QueryUsesPrimaryKey(t *testing.T) { + db := hopTestDB(t) + defer db.Close() + // Same shape as cmd/server scanReachRowsIndexed: tokens upper-cased in SQL. + rows, err := db.Query(`EXPLAIN QUERY PLAN SELECT observation_id FROM observation_hops WHERE hop IN (UPPER(?),UPPER(?)) AND timestamp >= ?`, "aa", "BB", 10) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var plan []string + for rows.Next() { + var a, b, c int + var d string + rows.Scan(&a, &b, &c, &d) + plan = append(plan, d) + } + sort.Strings(plan) + if len(plan) == 0 || !contains(plan[0], "PRIMARY KEY") || contains(plan[0], "SCAN") { + t.Fatalf("expected PK SEARCH, got %v", plan) + } +} + +// Rows below the coverage mark are deliberately not indexed or maintained, so +// the index stays bounded to its window. +func TestObservationHops_IgnoresRowsBelowCoverageMark(t *testing.T) { + db := hopTestDB(t) + defer db.Close() + db.Exec(`UPDATE ` + ObservationHopsCoverageTable + ` SET since = 1000`) + db.Exec(`INSERT INTO observations (path_json, timestamp) VALUES ('["AA"]', 500)`) // id 1, below + db.Exec(`INSERT INTO observations (path_json, timestamp) VALUES ('["BB"]', 1500)`) // id 2, in window + if got := hopsFor(t, db, 1); len(got) != 0 { + t.Fatalf("below-mark row was indexed: %v", got) + } + if got, want := hopsFor(t, db, 2), []string{"BB"}; !reflect.DeepEqual(got, want) { + t.Fatalf("in-window row: %v want %v", got, want) + } + db.Exec(`UPDATE observations SET path_json = '["CC"]' WHERE id IN (1,2)`) + if got := hopsFor(t, db, 1); len(got) != 0 { + t.Fatalf("update indexed a below-mark row: %v", got) + } + if got, want := hopsFor(t, db, 2), []string{"CC"}; !reflect.DeepEqual(got, want) { + t.Fatalf("update in-window: %v want %v", got, want) + } + db.Exec(`DELETE FROM observations`) + var n int + db.QueryRow(`SELECT COUNT(*) FROM observation_hops`).Scan(&n) + if n != 0 { + t.Fatalf("delete left %d rows", n) + } +} + +func hopObjects(t *testing.T, db *sql.DB) []string { + t.Helper() + rows, err := db.Query(`SELECT name FROM sqlite_master + WHERE name LIKE 'observation_hops%' OR name LIKE 'trg_observation_hops%' ORDER BY name`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []string + for rows.Next() { + var n string + rows.Scan(&n) + out = append(out, n) + } + return out +} + +// The index is opt-in: the shared schema (Apply, used by the ingestor and the +// migrate tool) must not install any of it. +func TestApplyDoesNotInstallObservationHops(t *testing.T) { + db := minimalDB(t) + defer db.Close() + if err := Apply(db, nil); err != nil { + t.Fatal(err) + } + if got := hopObjects(t, db); len(got) != 0 { + t.Fatalf("Apply installed %v", got) + } +} + +func TestDropObservationHopsSchemaRemovesEverything(t *testing.T) { + db := hopTestDB(t) + defer db.Close() + if got := hopObjects(t, db); len(got) != 5 { + t.Fatalf("setup: %v, want 2 tables + 3 triggers", got) + } + removed, err := DropObservationHopsSchema(db) + if err != nil || !removed { + t.Fatalf("drop: removed=%v err=%v", removed, err) + } + if got := hopObjects(t, db); len(got) != 0 { + t.Fatalf("left behind: %v", got) + } + if _, err := db.Exec(`INSERT INTO observations (path_json, timestamp) VALUES ('["AA"]', 5)`); err != nil { + t.Fatalf("insert after drop: %v", err) + } + removed, err = DropObservationHopsSchema(db) + if err != nil || removed { + t.Fatalf("second drop: removed=%v err=%v, want false, nil", removed, err) + } +} + +// A trigger definition changed in a later version must reach an install that +// already has the old one; the tables and coverage mark are kept. +func TestEnsureObservationHopsSchemaRefreshesTriggers(t *testing.T) { + db := hopTestDB(t) + defer db.Close() + db.Exec(`UPDATE ` + ObservationHopsCoverageTable + ` SET since = 7`) + db.Exec(`DROP TRIGGER trg_observation_hops_ai`) + if _, err := db.Exec(`CREATE TRIGGER trg_observation_hops_ai AFTER INSERT ON observations + BEGIN INSERT OR IGNORE INTO observation_hops SELECT value, NEW.timestamp, NEW.id FROM json_each(NEW.path_json); END`); err != nil { + t.Fatal(err) + } + if err := EnsureObservationHopsSchema(db, nil); err != nil { + t.Fatal(err) + } + db.Exec(`INSERT INTO observations (path_json, timestamp) VALUES ('["ab"]', 10)`) + if got, want := hopsFor(t, db, 1), []string{"AB"}; !reflect.DeepEqual(got, want) { + t.Fatalf("hops %v, want %v (old trigger still installed?)", got, want) + } + var since int64 + db.QueryRow(`SELECT since FROM ` + ObservationHopsCoverageTable).Scan(&since) + if since != 7 { + t.Fatalf("coverage mark reset to %d, want 7 kept", since) + } +} + +// The backfill and trim chunk functions add and remove exactly the rows the +// triggers would, upper-cased, and only inside [lo, hi). +func TestIndexAndUnindexObservationHops(t *testing.T) { + db := hopTestDB(t) + defer db.Close() + removed, _ := DropObservationHopsSchema(db) // insert with no triggers, as before an upgrade + if !removed { + t.Fatal("setup: drop") + } + for _, r := range []struct { + p string + ts int + }{{`["aa","BB","aa"]`, 10}, {`["CC",5,["DD"]]`, 20}, {`bad`, 20}, {`["EE"]`, 30}} { + db.Exec(`INSERT INTO observations (path_json, timestamp) VALUES (?, ?)`, r.p, r.ts) + } + if err := EnsureObservationHopsSchema(db, nil); err != nil { + t.Fatal(err) + } + n, err := IndexObservationHops(db, 10, 30) + if err != nil || n != 3 { + t.Fatalf("index [10,30): %d rows, err %v; want 3 (AA, BB, CC)", n, err) + } + if got, want := hopsFor(t, db, 1), []string{"AA", "BB"}; !reflect.DeepEqual(got, want) { + t.Fatalf("obs 1 hops %v, want %v", got, want) + } + if got := hopsFor(t, db, 4); len(got) != 0 { + t.Fatalf("obs 4 at hi bound indexed: %v", got) + } + n, err = UnindexObservationHops(db, 10, 20) + if err != nil || n != 2 { + t.Fatalf("unindex [10,20): %d rows, err %v; want 2", n, err) + } + if got, want := hopsFor(t, db, 2), []string{"CC"}; !reflect.DeepEqual(got, want) { + t.Fatalf("obs 2 hops after unindex %v, want %v", got, want) + } +}