mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-11-11 11:57:57 +00:00
perf(reach): answer /nodes/{pk}/reach from a rolling hop index, not a path_json LIKE scan (#2214)
Refs #2146 (reach item: "an indexed hop table would remove it"). Follows #2155. ## What changes Reach finds the observations that relayed through a node with `path_json LIKE '%"XX"%'` over every observation in the window. The cost follows the window, not the node: a quiet repeater costs about as much as a busy one. When the database does not fit in the page cache this outlives the 60 s `WriteTimeout`, the request 500s, and nothing is cached. #2146 reports this endpoint at 14.9 s average and 74.7 s max on a 10.9 GB instance. Its 1.9 s figure after #2155 is a warm-cache timing; the scan reads the whole window, so cold it is far slower (below), and it holds one of the 4 read-pool connections the whole time. This PR adds an inverted index and uses it only when it is complete for the requested window: - `observation_hops(hop, timestamp, observation_id)`, `WITHOUT ROWID`. A token lookup is a seek and the window bound is a range on the same b-tree. - Triggers on `observations` maintain it (insert, delete, `path_json` update), so prune and path-repair need no changes. - `observation_hops_coverage.since` marks how far back the index is complete. If the window starts before `since`, or the table is missing (older DB, backfill unfinished), reach runs the existing `LIKE` scan unchanged. - The ingestor owns the schema, a resumable newest-first backfill, and a daily trim to `retention.hopIndexDays` (default 14). The server stays read-only and `AssertReady` does not require the table. The backfill starts after the ingest gate opens rather than inside `OpenStore`. My first version did the latter and held live ingest in the buffer for about 15 minutes on a 13.7 GB database (nothing dropped). Chunks are 15 minutes because one-hour chunks held the writer for 2 to 7 s per commit on EBS. ## Results ### Real data: vanilla vs this branch One snapshot of a real instance (11.5 GB, 10.7M observations over 14 days) served by both builds on the same machine (8 cores, 30 GB RAM, NVMe, warm cache). One cold request per cell, seconds, repeaters from busiest to quietest by 24 h relay count. | node (relays / 24 h) | 7 d vanilla | 7 d branch | 14 d vanilla | 14 d branch | |---|---|---|---|---| | 7,314 | 24.5 | 4.5 | 8.3 | 9.6 | | 7,307 | 1.7 | 2.4 | 1.3 | 3.1 | | 7,047 | 3.3 | 2.2 | 2.0 | 2.3 | | 2,151 | 20.0 | 0.44 | 29.6 | 2.0 | | 673 | 23.7 | 0.18 | 28.5 | 0.62 | | 240 | 21.8 | 0.01 | 27.8 | 0.04 | | 52 | 21.6 | 0.00 | 28.4 | 0.00 | - Nodes with few relays drop from 20 to 30 s to under a second. - The busiest relays are about the same, sometimes slower: both builds spend their time loading tens of thousands of matching rows by primary key, which the index does not speed up. - Single runs, and page-cache state moves them a lot: an earlier run, with the file freshly cached, had vanilla's quiet nodes at 4.5 to 5 s for 7 d, close to #2155's warm figure. Read orders of magnitude, not exact figures. Nothing failed on this fast machine. **Same answers.** Unit tests check indexed rows equal `LIKE` rows, fallback on partial coverage, and trigger upkeep. On real data, 21 node/window pairs (3 to 14 days) requested from both builds at the same moment were all identical once links were compared by public key. `LIKE` ignores case and the index does not; 0 of 10.1M observations in the window have a lower-case hop. Above `reachScanRowLimit` both paths truncate, and they can keep different subsets. ### Production instance t4g.large (2 vCPU, 8 GB, EBS), 13.7 GB database, about 4.5M observations per 7 days, so the cache held a fraction of it. - Before: `days=7` on a busy node returned 500 (`context canceled`), and `days=1` took 58.5 s. - With the index (38 probes of four nodes over 18 hours): `days=7` on that node returned 200 in 23 s. The busiest probe node had a median of 0.7 s and p95 18 s. The hottest (127k relay observations in 7 d) had a median of 1.3 s, p95 52 s, and one probe past 60 s at evening peak. The quietest had a median of 0.04 s. - On 2026-10-04 the host was resized to 4 vCPU and 16 GB so the cache holds the database. Same index afterwards (493 probes): hottest node median 1.4 s, p95 1.7 s, max 38 s; quietest median 0.05 s. (From 2026-10-03 the instance also ran a local change, not in this PR, that lets reach requests continue in the background, which only affects requests over 45 s.) So on a small cold host the index turned "always fails" into "usually works" but did not make the hottest relays reliably fast. That needs cheaper row loading (a compact side table measured 4 to 10x on cold reads; kept out of this PR) or background computation. ### Benchmarks (included) ``` go test ./cmd/server -run xxx -bench ScanReachRows -benchtime 5x # 600k observations, 11 hops avg, 7 d window in 14 d, warm cache hot_token/like 816 ms rare_token/like 486 ms hot_token/indexed 293 ms rare_token/indexed 1.7 ms go test ./internal/dbschema -run xxx -bench ObservationInsert -benchtime 20000x # one observation per transaction, WriterDSN (WAL, synchronous=FULL), 11 hops without_index 23 us/op with_index 127 us/op ``` The triggers add about 100 us per observation, roughly 0.1% of a core at the production instance's 8 observations per second. They cap one writer at about 8k observations per second instead of 43k. The benchmark ran in a temp directory, so it shows CPU and b-tree cost, not disk sync cost. Over 13 days on the production instance, 0 observations were dropped and newest-observation age was p99 4 s. Four of 623 samples were 20 s or more: 91 s and 20 s came within six minutes of a container restart, 31 s came 24 hours after one (when the ingestor's daily tickers fire), and 23 s had no restart nearby. Index coverage stayed between 14.0 and 15.2 days. **Disk.** On the snapshot (`dbstat`) the index is 1.98 GB for 10.1M observations in the window, about 196 bytes each, next to 6.81 GB for `observations`. It scales with hops per observation and `hopIndexDays`. Backfill of 77M rows took 15.5 min on the production instance. **Perf justification (AGENTS.md).** Reach goes from a scan over the whole window to a seek over the matching rows. Everything new is bounded: the index by `hopIndexDays` and the daily trim, the backfill by chunk size and a cancellable context, and the server adds no in-memory state. The ingest hot path pays the trigger cost above. ## Design notes and open questions - Why not the in-memory store: it defaults to 7 days (`retentionHours: 168`) and reach offers up to 30, so it could not answer the longer windows. The other option listed in #2146, driving the query from `observations.timestamp`, is what #2155 already did. - SQLite only: the triggers use `json_each`. The optional PostgreSQL backend in #2145 would need its own equivalent. - No off switch: the table and triggers are always installed and `hopIndexDays` only sets the window. If you would rather have this opt-in, or want a smaller default for small disks, I'll change it. ## Operator notes - Upgrade: the ingestor installs the table and triggers on start and backfills. Until a window is covered, reach behaves as today. - Rollback: an older binary keeps indexing new rows with no trim. To remove it with the ingestor stopped: `DROP TRIGGER trg_observation_hops_ai; DROP TRIGGER trg_observation_hops_ad; DROP TRIGGER trg_observation_hops_au; DROP TABLE observation_hops; DROP TABLE observation_hops_coverage;` - New key `retention.hopIndexDays`, documented in `config.example.json`. ## Tests `go test ./...` passes in `cmd/server`, `cmd/ingestor` and `internal/dbschema`. New: `node_reach_hops_test.go`, `observation_hops_test.go`, `observation_hops_backfill_test.go`, and the two benchmarks. ## Update after review (commit `2291bea2`) The text above is what was reviewed at `95e17976`. Where they differ, this section is current: - **Opt-in.** `retention.hopIndexDays` unset or 0 (the default, and the value in `config.example.json`) installs nothing. A value above 0 installs the table and triggers and backfills that many days. This replaces "default 14", the "No off switch" note and the Upgrade note above. - **Changing the setting** applies at the next ingestor start: raising it backfills the gap, lowering it trims, and 0 removes the triggers and both tables. On a 10.5M-observation copy the removal took 45 to 71 s, at startup before MQTT connects. - **Case.** Hops are stored and matched upper-cased, so the index ignores ASCII case the way LIKE does. This replaces the sentence on case under "Same answers". - **Rollback.** Set it to 0 and start the ingestor once, then roll back. The manual `DROP` list above still works with the ingestor stopped. - **Disk.** A full 14-day backfill on that copy built 2.17 GB for 9.68M observations, about 224 bytes each, in 13 min 37 s. - Backfill and trim are serialized, and the chunk SQL lives in `dbschema` next to the triggers. ## Update after review, round 2 (commit `95690e8c`) Where this differs from the two sections above, this one is current: - **Startup order.** The index is installed or removed in the startup block, after the MQTT subscription is up and before the ingest gate opens, so packets arriving during a removal are buffered (`ingestBufferSize`), not lost. This replaces "at startup before MQTT connects" in the first update. - **Mark moves cover their range.** Each backfill and trim transaction indexes or deletes everything between the old mark and the new one, so a row arriving between the chunk-picking read and the transaction is accounted for either way. - **Negative `hopIndexDays`** is a config error at startup; the ingestor refuses to run. This replaces "−5 is off" in the first update. - **Server.** If the index disappears between the coverage check and the scan, the request is answered from the `LIKE` scan instead of a 500. --- 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: nullrouten <201262769+nullrouten0@users.noreply.github.com> Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
13 files changed
+1925
-2
No files matched your search
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
+2
-1
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user