diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 9b0c08581..28354afad 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -203,6 +203,23 @@ func (c *Config) IncrementalVacuumPages() int { return 1024 } +// AnalysisLimit returns the per-index row cap for the planner stats refresh +// (#2058). A negative setting disables the refresh; zero means unset, matching +// IncrementalVacuumPages above. +// +// 10000 rather than the 400 SQLite's documentation offers: measured on the +// 9.4 GB staging database, 400 and 1000 leave the channel-query plan exactly as +// it was, 10000 produces the same plan as an unbounded ANALYZE, and it costs 2.0s +// against that ANALYZE's 242.9s, both timed warm. The full ladder, and the cold +// figure that matters at startup, are in Store.RefreshPlannerStats and +// Store.EnsurePlannerStats. +func (c *Config) AnalysisLimit() int { + if c.DB != nil && c.DB.AnalysisLimit != 0 { + return c.DB.AnalysisLimit + } + return 10000 +} + // ShouldValidateSignatures returns true (default) unless explicitly disabled. func (c *Config) ShouldValidateSignatures() bool { if c.ValidateSignatures != nil { diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index 042af9153..714b8115d 100644 --- a/cmd/ingestor/db.go +++ b/cmd/ingestor/db.go @@ -1707,6 +1707,126 @@ func (s *Store) RunIncrementalVacuum(pages int) { } } +// RefreshPlannerStats rebuilds the query planner's cardinality statistics +// (#2058). +// +// Without a sqlite_stat1 table the planner works from built-in guesses, and on +// the channel queries it guesses wrong: it drives from the plain +// idx_transmissions_payload_type rather than idx_tx_channel_hash, the partial +// index (WHERE payload_type = 5) this schema already carries for that exact +// filter. +// +// Measured on the 9.4 GB staging database (1,250,489 transmissions, 14,169,329 +// observations), region-filtered GetChannels, counting page-cache misses +// because wall time there is dominated by the OS page cache (56.7s cold, 0.80s +// warm, for the same query and plan): +// +// analysis_limit ANALYZE driving index page misses +// none (no stats) - idx_transmissions_payload_type 143,442 +// 400 171ms idx_transmissions_payload_type 143,449 +// 1000 171ms idx_transmissions_payload_type 143,450 +// 10000 2.0s* idx_tx_channel_hash 107,429 +// 0 (unbounded) 242.9s idx_tx_channel_hash 107,429 +// +// (*) Every ANALYZE duration in that table was timed on a warm page cache, one +// after another. The same statement at limit 10000 took 3m43.9s cold, on a +// freshly started container. See EnsurePlannerStats. +// +// So 10000 buys the whole plan change, and the four-minute unbounded ANALYZE +// buys nothing beyond it. 400, the value SQLite's documentation offers for the +// bounded form, changes nothing at all on this data: it samples too few rows to +// separate the 126,336-row partial index from the 920,700-row plain one. +// +// ANALYZE, not PRAGMA optimize. Measured on the same database: optimize is a +// no-op here, because it only analyzes tables that the calling connection has +// itself queried during the session, and a maintenance call has queried none. +// PRAGMA optimize(0x03) returned no statements and sqlite_stat1 was not +// created. +// +// Note that analysis_limit=0 means *no* limit to SQLite, not "use a default". +// Config.AnalysisLimit maps an unset config to 10000 for that reason, and a +// negative value here disables the refresh. +// +// Returns whether the statistics were refreshed. This function owns its +// logging; callers need add nothing. +func (s *Store) RefreshPlannerStats(analysisLimit int) bool { + if analysisLimit < 0 { + return false + } + first := !s.hasPlannerStats() + start := time.Now() + // Tagged for /api/perf writer-lock visibility (#1340). + if _, err := s.instrumentedExec("analyze", fmt.Sprintf("PRAGMA analysis_limit=%d", analysisLimit)); err != nil { + log.Printf("[analyze] could not set analysis_limit: %v", err) + return false + } + if _, err := s.instrumentedExec("analyze", "ANALYZE"); err != nil { + log.Printf("[analyze] ANALYZE failed: %v", err) + return false + } + elapsed := time.Since(start).Round(time.Millisecond) + if first { + log.Printf("[analyze] planner statistics built in %v (analysis_limit=%d, first run against this database)", elapsed, analysisLimit) + } else { + log.Printf("[analyze] planner statistics refreshed in %v (analysis_limit=%d)", elapsed, analysisLimit) + } + return true +} + +// EnsurePlannerStats builds planner statistics when the database has none, and +// reports whether it did (#2058). +// +// This exists to close the window the startup stagger opens. The refresh ticker +// waits 2 minutes before its first run, and a query arriving in that window +// against a database with no statistics at all gets the plan measured in +// RefreshPlannerStats above: 143,442 pages read, which timed at 56.7s cold on +// the 9.4 GB staging file. +// +// It fires once per database, not once per restart. sqlite_stat1 is an ordinary +// table, so once written it stays in the file and a fresh read-only connection +// reads it back (verified across connection close, and through a mode=ro +// handle). Every later start therefore costs one query against sqlite_master and +// leaves the work to the ticker. +// +// The trade is a one-time ANALYZE early in startup, and it is not cheap on a +// cold page cache. Observed on staging at 9.4 GB: 3m43.9s, against the 2.0s the +// same statement takes warm. For those 3m44s it holds the store's single write +// connection (SetMaxOpenConns(1) in OpenStore), so ingest stalls and buffers: the +// observations table took zero rows for four minutes and then 1027 in the minute +// the ANALYZE finished, against about 130 a minute either side, with nothing +// dropped. Hence the warning below, so an operator watching a first deploy can +// tell this apart from a hang. +// +// That cost belongs to the first ANALYZE, not to running it here. The refresh +// ticker would pay exactly the same 3m44s two minutes later; this only moves it +// earlier, where it overlaps the startup burst the ingest buffer is already +// sized for. +// +// A negative analysisLimit disables it without a word: the warning is only +// true when an ANALYZE follows it. +func (s *Store) EnsurePlannerStats(analysisLimit int) bool { + if analysisLimit < 0 || s.hasPlannerStats() { + return false + } + log.Printf("[analyze] this database has no planner statistics; building them now. " + + "ANALYZE holds the single write connection until it finishes (3m43.9s measured on 9.4 GB, cold), " + + "so ingest will buffer and catch up. Once per database, not once per restart.") + return s.RefreshPlannerStats(analysisLimit) +} + +// hasPlannerStats reports whether ANALYZE has ever run against this database. +// A query error reads as "no stats", which is the safe direction for both +// callers: it costs the log line a wrong word, and costs EnsurePlannerStats one +// ANALYZE that was not needed, rather than skipping one that was. +func (s *Store) hasPlannerStats() bool { + var n int + if err := s.db.QueryRow( + `SELECT count(*) FROM sqlite_master WHERE type='table' AND name='sqlite_stat1'`).Scan(&n); err != nil { + return false + } + return n > 0 +} + // Checkpoint runs a WAL checkpoint (TRUNCATE mode). // Returns the number of WAL frames checkpointed (0 if WAL was already empty). // TRUNCATE resets the WAL file to zero bytes when all frames are checkpointed; diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 3ab0626b1..d0f382a5d 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -327,9 +327,16 @@ func main() { // DB) is absorbed by IngestBuffer instead of delaying start-up. Cancelled // on shutdown so store.Close() does not wait for it; it resumes on the // next start. + // + // It shares its start with the one-off planner-statistics build (#2058, + // issue #100): both hold the single writer for one long statement, so + // they run one after the other, statistics first, each behind a drain of + // the ingest buffer, never back to back. Why that order, and where the + // other startup tickers fall: planner_stats_startup.go. + analysisLimit := cfg.AnalysisLimit() routeMaskCtx, stopRouteMaskBackfill := context.WithCancel(context.Background()) defer stopRouteMaskBackfill() - store.StartRouteMaskBackfill(routeMaskCtx) + startupWritersDone := store.startPlannerStatsThenRouteMaskBackfill(routeMaskCtx, analysisLimit, ingestBuffer.Pending) // Daily ticker for node retention retentionTicker := time.NewTicker(1 * time.Hour) @@ -412,6 +419,25 @@ func main() { }() log.Printf("[db] WAL checkpoint scheduled every 1h") + // Daily planner statistics refresh (#2058). + // + // The one-off build for a database that has never been analyzed runs in + // the startup sequence above (Store.EnsurePlannerStats carries what it + // costs). The routine refresh is staggered 2 minutes past the point that + // sequence hands over, so it never lands on the startup ANALYZE or the + // backfill's index build, and so it also sees the rows the startup burst + // added. On a first start it re-runs the build two minutes later, warm. + // + // Bounded by analysis_limit either way, so neither grows with the file the way + // an unbounded ANALYZE does: 2.0s against 242.9s on a 9.4 GB database, both + // timed warm. Cold, on a first start, it is 3m43.9s. + stopPlannerStatsRefresh, plannerStatsEnabled := store.schedulePlannerStatsRefresh(analysisLimit, startupWritersDone, 2*time.Minute, 24*time.Hour) + if plannerStatsEnabled { + log.Printf("[analyze] planner statistics refresh scheduled every 24h (analysis_limit=%d)", analysisLimit) + } else { + log.Printf("[analyze] planner statistics build and refresh disabled (db.analysisLimit=%d)", analysisLimit) + } + // Daily neighbor_edges retention (#1287 — moved from cmd/server). { nDays := cfg.NeighborEdgesDaysOrDefault() @@ -509,6 +535,7 @@ func main() { statsTicker.Stop() pruneQueueTicker.Stop() walCheckpointTicker.Stop() + stopPlannerStatsRefresh() stopWatchdog() store.LogStats() // final stats on shutdown for _, c := range clients { diff --git a/cmd/ingestor/planner_stats_startup.go b/cmd/ingestor/planner_stats_startup.go new file mode 100644 index 000000000..d8d6c8903 --- /dev/null +++ b/cmd/ingestor/planner_stats_startup.go @@ -0,0 +1,124 @@ +package main + +import ( + "context" + "log" + "sync" + "time" +) + +// Startup ordering of the one-off single-writer holds (issue #100, the fork +// adaptation of `Kpa-clawbot/CoreScope#2072`/`#2074`). +// +// Two startup steps hold the store's single write connection (writerMu, +// SetMaxOpenConns(1)) for one long statement each: +// +// - EnsurePlannerStats: ANALYZE on a database that has never been analyzed. +// Once per database; 3m43.9s cold on upstream's 9.4 GB staging file. +// - the #89 route_mask backfill: CREATE INDEX of its pending index (~20 s on +// a staging-sized DB), then short batches with a yield between them. +// +// While either runs, IngestBuffer's consumer is blocked and packets queue. +// Started side by side, the two holds would run back to back (writerMu is +// FIFO once contended) and the ingest pause would be their sum; started the +// moment the buffer opens, either would also sit on top of the backlog the +// buffer collected during the startup migrations. So they run in sequence on +// one goroutine, each behind a drain of the buffer: +// +// drain → EnsurePlannerStats → drain → StartRouteMaskBackfill +// +// ANALYZE goes first because it is what the read path is waiting for (the +// channel queries run on the wrong index until sqlite_stat1 exists), while the +// backfill is background repair whose batch phase alone runs for minutes on a +// staging-sized table. On a restart EnsurePlannerStats is one sqlite_master +// query, so the backfill starts after nothing more than the first drain. +// +// The routine refresh (schedulePlannerStatsRefresh) starts its stagger only +// after this sequence hands over, so it cannot land on the startup ANALYZE. +// The other startup tickers (WAL checkpoint +30s, observer retention +90s, +// multibyte persist +2m, neighbor prune +4m) are left where they were: each is +// a short write that queues behind whichever hold is running and then yields +// writerMu back, so they add seconds, not a second long hold. + +// plannerStatsDrainTimeout bounds each drain wait: if ingest outpaces the +// writer the buffer may never read empty, and the next step must still run. +// Vars so tests can shrink them. +var ( + plannerStatsDrainTimeout = 2 * time.Minute + plannerStatsDrainPoll = 250 * time.Millisecond +) + +// startPlannerStatsThenRouteMaskBackfill runs the sequence above on its own +// goroutine and returns a channel closed once the backfill has been scheduled +// (or skipped because ctx was cancelled). pending reports the ingest backlog; +// main passes IngestBuffer.Pending. +func (s *Store) startPlannerStatsThenRouteMaskBackfill(ctx context.Context, analysisLimit int, pending func() int) <-chan struct{} { + done := make(chan struct{}) + go func() { + defer close(done) + waitForIngestDrain(ctx, pending, "planner statistics") + if ctx.Err() != nil { + return + } + s.EnsurePlannerStats(analysisLimit) + waitForIngestDrain(ctx, pending, "route_mask backfill") + if ctx.Err() != nil { + return + } + s.StartRouteMaskBackfill(ctx) + }() + return done +} + +// waitForIngestDrain blocks until pending() reads zero, ctx is cancelled, or +// plannerStatsDrainTimeout passes. +func waitForIngestDrain(ctx context.Context, pending func() int, next string) { + deadline := time.Now().Add(plannerStatsDrainTimeout) + for pending() > 0 { + if time.Now().After(deadline) { + log.Printf("[startup] ingest buffer still has %d pending after %v; starting %s anyway", pending(), plannerStatsDrainTimeout, next) + return + } + select { + case <-ctx.Done(): + return + case <-time.After(plannerStatsDrainPoll): + } + } +} + +// schedulePlannerStatsRefresh starts the routine refresh (#2058): once after +// the stagger, then every interval. The stagger starts when after closes (the +// startup sequence above has handed over). A negative analysisLimit disables +// it and reports enabled=false. stop ends the ticker. +func (s *Store) schedulePlannerStatsRefresh(analysisLimit int, after <-chan struct{}, stagger, interval time.Duration) (stop func(), enabled bool) { + if analysisLimit < 0 { + return func() {}, false + } + quit := make(chan struct{}) + go func() { + select { + case <-after: + case <-quit: + return + } + select { + case <-time.After(stagger): + case <-quit: + return + } + s.RefreshPlannerStats(analysisLimit) + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + s.RefreshPlannerStats(analysisLimit) + case <-quit: + return + } + } + }() + var once sync.Once + return func() { once.Do(func() { close(quit) }) }, true +} diff --git a/cmd/ingestor/planner_stats_startup_test.go b/cmd/ingestor/planner_stats_startup_test.go new file mode 100644 index 000000000..6c38d3902 --- /dev/null +++ b/cmd/ingestor/planner_stats_startup_test.go @@ -0,0 +1,443 @@ +package main + +import ( + "bytes" + "context" + "database/sql" + "fmt" + "log" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/meshcore-analyzer/dbschema" +) + +// Fork adaptation of `Kpa-clawbot/CoreScope#2072`/`#2074` (issue #100): the +// one-off ANALYZE and the #89 route_mask backfill's pending-index build are +// both long holds of the single writer. startPlannerStatsThenRouteMaskBackfill +// runs them one after the other with an ingest-buffer drain in front of each, +// so the ingest pause is never the sum of the two (or of either and the startup +// backlog). + +// syncLogBuffer is a log sink that is safe to read while goroutines write. +type syncLogBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncLogBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncLogBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +func capturePlannerLog(t *testing.T) *syncLogBuffer { + t.Helper() + buf := &syncLogBuffer{} + orig := log.Writer() + log.SetOutput(buf) + t.Cleanup(func() { log.SetOutput(orig) }) + return buf +} + +func shrinkPlannerDrainWait(t *testing.T, timeout time.Duration) { + t.Helper() + oldTimeout, oldPoll := plannerStatsDrainTimeout, plannerStatsDrainPoll + plannerStatsDrainTimeout, plannerStatsDrainPoll = timeout, time.Millisecond + t.Cleanup(func() { plannerStatsDrainTimeout, plannerStatsDrainPoll = oldTimeout, oldPoll }) +} + +// plannerStartupStore returns a store holding legacy rows (route_mask NULL, +// so the backfill has real work) and no sqlite_stat1. +func plannerStartupStore(t *testing.T) *Store { + t.Helper() + routeMaskShrinkBackfill(t, 2) + s := routeMaskStore(t, filepath.Join(t.TempDir(), "startup.db")) + for i := 0; i < 6; i++ { + routeMaskInsert(t, s, routeMaskObs{firstIngestedFloodRaw, "obs-a", fmt.Sprintf("2026-09-24T10:%02d:00Z", i)}) + routeMaskInsertRaw(t, s, fmt.Sprintf("legacy-%d", i), 1) + } + routeMaskMakeLegacy(t, s) + if hasStat1(t, s) { + t.Fatal("setup: the store already carries sqlite_stat1") + } + return s +} + +func plannerIngestPacket(i int) *PacketData { + return &PacketData{ + RawHex: "1102a1b2", + Timestamp: time.Now().UTC().Format(time.RFC3339), + ObserverID: "obs-a", + Hash: fmt.Sprintf("live-%06d", i), + RouteType: 1, + PayloadType: 5, + ChannelHash: fmt.Sprintf("ch-%d", i%3), + } +} + +func waitClosed(t *testing.T, ch <-chan struct{}, what string) { + t.Helper() + select { + case <-ch: + case <-time.After(30 * time.Second): + t.Fatalf("%s did not finish within 30s (deadlock on the single writer?)", what) + } +} + +func waitAsyncMigrations(t *testing.T, s *Store) { + t.Helper() + done := make(chan struct{}) + go func() { s.WaitForAsyncMigrations(); close(done) }() + waitClosed(t, done, "the route_mask backfill") +} + +func countTx(t *testing.T, s *Store, where string) int { + t.Helper() + var n int + if err := s.db.QueryRow(`SELECT count(*) FROM transmissions WHERE ` + where).Scan(&n); err != nil { + t.Fatal(err) + } + return n +} + +// Both one-off startup writers run against live ingest through the real +// IngestBuffer: no deadlock, statistics built, backfill complete, every +// buffered packet written, and ingest still writing afterwards. +func TestPlannerStatsAndRouteMaskBackfillAtStartup_Issue100(t *testing.T) { + shrinkPlannerDrainWait(t, 10*time.Second) + s := plannerStartupStore(t) + defer s.Close() + if countTx(t, s, "route_mask IS NULL") == 0 { + t.Fatal("setup: no legacy rows for the backfill") + } + + buf := NewIngestBuffer(1000) + buf.Start() + defer buf.Stop() + var written, failed atomic.Int64 + submit := func(i int) { + pd := plannerIngestPacket(i) + buf.Submit(func() { + if _, err := s.InsertTransmission(pd); err != nil { + failed.Add(1) + return + } + written.Add(1) + }) + } + // A startup backlog, as the subscription would have buffered it. + for i := 0; i < 50; i++ { + submit(i) + } + buf.Ready() + + stopFeed := make(chan struct{}) + feedDone := make(chan struct{}) + var fed atomic.Int64 + fed.Store(50) + go func() { + defer close(feedDone) + // Bounded: live traffic is ~130-600 packets a minute; this is far + // denser, but finite, so the buffer is never the thing under test. + for i := 50; i < 350; i++ { + select { + case <-stopFeed: + return + case <-time.After(2 * time.Millisecond): + submit(i) + fed.Add(1) + } + } + }() + + scheduled := s.startPlannerStatsThenRouteMaskBackfill(context.Background(), 10000, buf.Pending) + waitClosed(t, scheduled, "EnsurePlannerStats + scheduling the backfill") + waitAsyncMigrations(t, s) + close(stopFeed) + <-feedDone + + if !hasStat1(t, s) { + t.Error("sqlite_stat1 absent: EnsurePlannerStats did not run at startup") + } + if st, err := s.AsyncMigrationStatus(dbschema.RouteMaskBackfillMigration); err != nil || st != "done" { + t.Errorf("route_mask backfill status = %q, %v; want done", st, err) + } + if n := countTx(t, s, "route_mask IS NULL"); n != 0 { + t.Errorf("%d rows still without route_mask after the backfill", n) + } + + // Every packet fed so far must land, and a packet submitted after both + // startup writers must land too. + deadline := time.Now().Add(10 * time.Second) + for written.Load()+failed.Load() < fed.Load() && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if failed.Load() > 0 || buf.Dropped() > 0 { + t.Fatalf("ingest lost packets: failed=%d dropped=%d", failed.Load(), buf.Dropped()) + } + if got, want := written.Load(), fed.Load(); got != want { + t.Fatalf("ingest wrote %d of %d packets", got, want) + } + after := plannerIngestPacket(999999) + ok := make(chan struct{}) + buf.Submit(func() { + if _, err := s.InsertTransmission(after); err == nil { + close(ok) + } + }) + waitClosed(t, ok, "an insert after both startup writers") + if countTx(t, s, "hash = 'live-999999'") != 1 { + t.Error("the post-startup packet is not in transmissions") + } +} + +// The order this fork chose: drain, ANALYZE, drain, backfill. Swapping the +// two writers, or dropping either drain, fails here. +func TestPlannerStatsBeforeRouteMaskBackfillWithDrains_Issue100(t *testing.T) { + shrinkPlannerDrainWait(t, 10*time.Second) + s := plannerStartupStore(t) + defer s.Close() + logs := capturePlannerLog(t) + + // Stand-in for IngestBuffer.Pending: reports a backlog for a few polls, + // then empty, and logs each time it reports empty so the order is visible. + var polls atomic.Int64 + pending := func() int { + n := polls.Add(1) + if n%4 != 0 { + return 3 + } + log.Printf("[test] buffer drained (poll %d)", n) + return 0 + } + + waitClosed(t, s.startPlannerStatsThenRouteMaskBackfill(context.Background(), 10000, pending), "startup sequence") + waitAsyncMigrations(t, s) + + out := logs.String() + marks := []string{ + "[test] buffer drained (poll 4)", + "no planner statistics", + "planner statistics built in", + "[test] buffer drained (poll 8)", + fmt.Sprintf("%q starting", dbschema.RouteMaskBackfillMigration), + "[route-mask] pending index ready", + } + last := -1 + for _, m := range marks { + i := strings.Index(out, m) + if i == -1 { + t.Fatalf("log line %q missing; got:\n%s", m, out) + } + if i < last { + t.Fatalf("log line %q is out of order; want %q in that order; got:\n%s", m, marks, out) + } + last = i + } +} + +// A restart: sqlite_stat1 is already there, so no ANALYZE at startup. The +// sentinel row would be rewritten by any ANALYZE, so it is what proves it. +func TestStartupSkipsAnalyzeWhenStatsExist_Issue100(t *testing.T) { + shrinkPlannerDrainWait(t, 10*time.Second) + s := plannerStartupStore(t) + defer s.Close() + s.RefreshPlannerStats(10000) + if _, err := s.db.Exec(`UPDATE sqlite_stat1 SET stat = 'sentinel' WHERE tbl = 'transmissions'`); err != nil { + t.Fatal(err) + } + logs := capturePlannerLog(t) + + waitClosed(t, s.startPlannerStatsThenRouteMaskBackfill(context.Background(), 10000, func() int { return 0 }), "startup sequence") + waitAsyncMigrations(t, s) + + var sentinels int + if err := s.db.QueryRow(`SELECT count(*) FROM sqlite_stat1 WHERE stat = 'sentinel'`).Scan(&sentinels); err != nil { + t.Fatal(err) + } + if sentinels == 0 { + t.Error("the startup ran ANALYZE on a database that already had planner statistics") + } + if out := logs.String(); strings.Contains(out, "no planner statistics") || strings.Contains(out, "planner statistics built") { + t.Errorf("startup logged an ANALYZE on a database with statistics:\n%s", out) + } + if st, _ := s.AsyncMigrationStatus(dbschema.RouteMaskBackfillMigration); st != "done" { + t.Errorf("route_mask backfill status = %q, want done (it must still run on a restart)", st) + } +} + +// analysis_limit < 0 turns off both the startup build and the routine +// refresh, and nothing else: the backfill still runs. +func TestNegativeAnalysisLimitDisablesBothStartupAndRefresh_Issue100(t *testing.T) { + shrinkPlannerDrainWait(t, 10*time.Second) + s := plannerStartupStore(t) + defer s.Close() + logs := capturePlannerLog(t) + + scheduled := s.startPlannerStatsThenRouteMaskBackfill(context.Background(), -1, func() int { return 0 }) + stop, enabled := s.schedulePlannerStatsRefresh(-1, scheduled, time.Millisecond, time.Millisecond) + defer stop() + if enabled { + t.Error("schedulePlannerStatsRefresh reported the refresh enabled at analysis_limit=-1") + } + waitClosed(t, scheduled, "startup sequence") + waitAsyncMigrations(t, s) + time.Sleep(50 * time.Millisecond) // many refresh intervals, had it been scheduled + + if hasStat1(t, s) { + t.Error("sqlite_stat1 built although analysis_limit < 0") + } + if out := logs.String(); strings.Contains(out, "no planner statistics") { + t.Errorf("warned about an ANALYZE that is disabled:\n%s", out) + } + if st, _ := s.AsyncMigrationStatus(dbschema.RouteMaskBackfillMigration); st != "done" { + t.Errorf("route_mask backfill status = %q, want done: disabling ANALYZE must not disable the backfill", st) + } +} + +// The routine refresh starts only once the startup sequence has handed over, +// then runs on its interval. +func TestPlannerStatsRefreshWaitsForStartupThenTicks_Issue100(t *testing.T) { + s := newTestStore(t) + defer s.Close() + logs := capturePlannerLog(t) + + after := make(chan struct{}) + stop, enabled := s.schedulePlannerStatsRefresh(10000, after, time.Millisecond, 5*time.Millisecond) + defer stop() + if !enabled { + t.Fatal("refresh not scheduled at a positive limit") + } + time.Sleep(30 * time.Millisecond) + if hasStat1(t, s) { + t.Fatal("the routine refresh ran before the startup sequence handed over") + } + close(after) + deadline := time.Now().Add(5 * time.Second) + for strings.Count(logs.String(), "[analyze] planner statistics") < 2 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if n := strings.Count(logs.String(), "[analyze] planner statistics"); n < 2 { + t.Fatalf("want the staggered refresh and at least one tick, got %d refresh lines:\n%s", n, logs.String()) + } +} + +// A drain that never happens (ingest faster than the writer) must not hold +// the backfill back forever. +func TestStartupDrainWaitIsBounded_Issue100(t *testing.T) { + shrinkPlannerDrainWait(t, 20*time.Millisecond) + s := plannerStartupStore(t) + defer s.Close() + + waitClosed(t, s.startPlannerStatsThenRouteMaskBackfill(context.Background(), 10000, func() int { return 1 }), "startup sequence with a buffer that never drains") + waitAsyncMigrations(t, s) + if st, _ := s.AsyncMigrationStatus(dbschema.RouteMaskBackfillMigration); st != "done" { + t.Errorf("route_mask backfill status = %q, want done", st) + } +} + +// Shutdown during the startup ANALYZE: the backfill is not started on a +// cancelled context (it resumes on the next start, as before). The stand-in +// for IngestBuffer.Pending cancels once the statistics exist, i.e. between +// the two writers, which is the window the ctx check after the second drain +// covers. +func TestStartupSequenceStopsOnCancel_Issue100(t *testing.T) { + shrinkPlannerDrainWait(t, time.Hour) + s := plannerStartupStore(t) + defer s.Close() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + pending := func() int { + var n int + s.db.QueryRow(`SELECT count(*) FROM sqlite_master WHERE name = 'sqlite_stat1'`).Scan(&n) + if n == 0 { + return 0 // first drain: nothing buffered, go on to ANALYZE + } + cancel() // shutdown arrives while the ANALYZE's backlog drains + return 1 + } + waitClosed(t, s.startPlannerStatsThenRouteMaskBackfill(ctx, 10000, pending), "startup sequence after cancel") + waitAsyncMigrations(t, s) + if !hasStat1(t, s) { + t.Fatal("setup: the ANALYZE did not run before the cancel") + } + if st, err := s.AsyncMigrationStatus(dbschema.RouteMaskBackfillMigration); err != sql.ErrNoRows { + t.Errorf("the backfill was scheduled after shutdown was requested (status %q, err %v)", st, err) + } +} + +// The routine refresh must rewrite statistics that already exist; that is its +// whole job after the first start. ANALYZE does. PRAGMA optimize, which +// upstream's first draft used, would leave them alone here: it only +// re-analyzes a table whose row count has moved a lot. (Upstream's +// TestRefreshPlannerStatsWritesStatistics_Issue2058 guards the same mistake +// on SQLite 3.45.1, where optimize wrote nothing on a fresh database; the +// SQLite 3.46.0 bundled with modernc.org/sqlite here does analyze a +// never-analyzed table from optimize, so that test alone no longer catches it.) +func TestRefreshPlannerStatsRewritesExistingStats_Issue100(t *testing.T) { + s := newTestStore(t) + defer s.Close() + for i := 0; i < 20; i++ { + s.InsertTransmission(plannerIngestPacket(i)) + } + if !s.RefreshPlannerStats(10000) { + t.Fatal("first refresh failed") + } + if _, err := s.db.Exec(`UPDATE sqlite_stat1 SET stat = 'sentinel' WHERE tbl = 'transmissions'`); err != nil { + t.Fatal(err) + } + if !s.RefreshPlannerStats(10000) { + t.Fatal("second refresh failed") + } + var sentinels int + if err := s.db.QueryRow(`SELECT count(*) FROM sqlite_stat1 WHERE stat = 'sentinel'`).Scan(&sentinels); err != nil { + t.Fatal(err) + } + if sentinels != 0 { + t.Errorf("%d stale sqlite_stat1 rows survived a refresh; the refresh does not re-analyze", sentinels) + } +} + +// hasPlannerStats gates a write, so its error direction is pinned: a failing +// query reads as "no statistics" (one unneeded ANALYZE), never as "has +// statistics" (a database left without them). +func TestHasPlannerStatsErrorReadsAsAbsent_Issue100(t *testing.T) { + s := newTestStore(t) + s.RefreshPlannerStats(10000) + if !s.hasPlannerStats() { + t.Fatal("hasPlannerStats false right after a refresh") + } + s.WaitForAsyncMigrations() + if err := s.db.Close(); err != nil { + t.Fatal(err) + } + if s.hasPlannerStats() { + t.Error("hasPlannerStats read a failed query as \"statistics present\"") + } +} + +// Upstream's EnsurePlannerStats logged its "building them now, ingest will +// pause" warning even at a negative limit, where nothing is built. +func TestEnsurePlannerStatsDisabledIsSilent_Issue100(t *testing.T) { + s := newTestStore(t) + defer s.Close() + logs := capturePlannerLog(t) + if s.EnsurePlannerStats(-1) { + t.Error("EnsurePlannerStats(-1) reported a build") + } + if out := logs.String(); strings.Contains(out, "no planner statistics") { + t.Errorf("warned about an ANALYZE that is disabled:\n%s", out) + } +} diff --git a/cmd/ingestor/refresh_planner_stats_test.go b/cmd/ingestor/refresh_planner_stats_test.go new file mode 100644 index 000000000..6df97fa2c --- /dev/null +++ b/cmd/ingestor/refresh_planner_stats_test.go @@ -0,0 +1,267 @@ +package main + +import ( + "bytes" + "log" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/meshcore-analyzer/dbconfig" +) + +// #2058: the planner has no cardinality statistics because ANALYZE has never +// run, so it picks a plain index over the partial index built for the query. +// These pin the three things that make the refresh work at all: the statement is +// one that actually writes statistics, the pragma reaches the connection, and a +// negative limit leaves the database untouched. + +func hasStat1(t *testing.T, s *Store) bool { + t.Helper() + var n int + if err := s.db.QueryRow( + `SELECT count(*) FROM sqlite_master WHERE type='table' AND name='sqlite_stat1'`).Scan(&n); err != nil { + t.Fatalf("query sqlite_master: %v", err) + } + return n > 0 +} + +// This is the guard against going back to PRAGMA optimize, which is what the +// first draft of this change used. Measured against the 9.4 GB staging database: +// optimize analyzes only tables the calling connection has itself queried during +// the session, so from a maintenance call it writes nothing and sqlite_stat1 +// never appears. A fresh store here has queried nothing either, so this test +// fails on that mistake instead of passing on a no-op. +func TestRefreshPlannerStatsWritesStatistics_Issue2058(t *testing.T) { + s := newTestStore(t) + defer s.Close() + + if hasStat1(t, s) { + t.Fatal("a fresh store already carries sqlite_stat1, so this test cannot tell whether the refresh did anything") + } + + if !s.RefreshPlannerStats(10000) { + t.Fatal("RefreshPlannerStats reported no refresh") + } + + if !hasStat1(t, s) { + t.Error("sqlite_stat1 was not created, so the planner still has no statistics") + } +} + +func TestRefreshPlannerStatsAppliesTheLimit_Issue2058(t *testing.T) { + s := newTestStore(t) + defer s.Close() + + s.RefreshPlannerStats(250) + + // analysis_limit is per connection. The store runs SetMaxOpenConns(1) + // (in OpenStore), which is the only reason setting it through Exec is sound + // here: on a multi-connection pool the pragma could land on a connection + // the ANALYZE never uses, and the limit would silently not apply. + var limit int + if err := s.db.QueryRow("PRAGMA analysis_limit").Scan(&limit); err != nil { + t.Fatalf("read back analysis_limit: %v", err) + } + if limit != 250 { + t.Errorf("analysis_limit did not reach the connection: want 250, got %d", limit) + } +} + +func TestRefreshPlannerStatsNegativeLimitIsANoop_Issue2058(t *testing.T) { + s := newTestStore(t) + defer s.Close() + + if s.RefreshPlannerStats(-1) { + t.Error("a negative limit must not report a refresh") + } + if hasStat1(t, s) { + t.Error("a negative limit still built sqlite_stat1; the refresh is not actually disabled") + } +} + +func TestRefreshPlannerStatsIsRepeatable_Issue2058(t *testing.T) { + s := newTestStore(t) + defer s.Close() + + // The ticker calls this every 24h for the life of the process. A second + // call must not error or undo the first, which is the part a single-call + // test would not notice. + s.RefreshPlannerStats(10000) + if !s.RefreshPlannerStats(10000) { + t.Fatal("the second refresh reported failure") + } + if !hasStat1(t, s) { + t.Error("sqlite_stat1 disappeared across two refreshes") + } +} + +func TestAnalysisLimitConfigDefault_Issue2058(t *testing.T) { + cases := []struct { + name string + cfg *Config + want int + }{ + // Zero means "no limit" to SQLite, so an unset config must not be + // passed through as 0: that would turn a 2 second refresh into the + // 242.9s unbounded ANALYZE measured on the staging database. + {"no db section", &Config{}, 10000}, + {"db section, limit unset", &Config{DB: &dbconfig.DBConfig{}}, 10000}, + {"explicit limit", &Config{DB: &dbconfig.DBConfig{AnalysisLimit: 1000}}, 1000}, + {"disabled", &Config{DB: &dbconfig.DBConfig{AnalysisLimit: -1}}, -1}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := tc.cfg.AnalysisLimit(); got != tc.want { + t.Errorf("AnalysisLimit() = %d, want %d", got, tc.want) + } + }) + } +} + +// The default is not a free choice: 400 and 1000 were measured to leave the +// channel-query plan unchanged on the staging database, so a well-meant edit +// back to SQLite's documented 400 would quietly return this to a no-op. +func TestAnalysisLimitDefaultIsHighEnoughToMatter_Issue2058(t *testing.T) { + const measuredIneffective = 1000 + if got := (&Config{}).AnalysisLimit(); got <= measuredIneffective { + t.Errorf("default analysis_limit is %d; %d and below were measured to leave the plan unchanged on a 9.4 GB database", + got, measuredIneffective) + } +} + +func TestAnalysisLimitSurvivesTheConfigFile_Issue2058(t *testing.T) { + // The knob is only useful if it survives the config file. The field lives in + // internal/dbconfig, so a wrong json tag there would leave the accessor + // returning the default however the operator set it. + path := filepath.Join(t.TempDir(), "config.json") + if err := os.WriteFile(path, []byte(`{"db":{"analysisLimit":123}}`), 0o600); err != nil { + t.Fatal(err) + } + cfg, err := LoadConfig(path) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if got := cfg.AnalysisLimit(); got != 123 { + t.Errorf("analysisLimit did not survive the config file: got %d, want 123", got) + } +} + +// EnsurePlannerStats closes the 2 minute window the startup stagger opens, and +// must do so exactly once per database rather than once per restart. + +func TestEnsurePlannerStatsBuildsWhenAbsent_Issue2058(t *testing.T) { + s := newTestStore(t) + defer s.Close() + + if !s.EnsurePlannerStats(10000) { + t.Fatal("EnsurePlannerStats did not build statistics on a database that has none") + } + if !hasStat1(t, s) { + t.Error("sqlite_stat1 absent after EnsurePlannerStats") + } +} + +func TestEnsurePlannerStatsSkipsWhenPresent_Issue2058(t *testing.T) { + s := newTestStore(t) + defer s.Close() + + s.RefreshPlannerStats(10000) + // The point of the skip: on every restart after the first this must cost one + // sqlite_master query, not an ANALYZE. If it ever returns true here it is + // running the 2s refresh on every boot. + if s.EnsurePlannerStats(10000) { + t.Error("EnsurePlannerStats rebuilt statistics that were already there") + } +} + +func TestEnsurePlannerStatsRespectsDisabled_Issue2058(t *testing.T) { + s := newTestStore(t) + defer s.Close() + + if s.EnsurePlannerStats(-1) { + t.Error("a negative limit must not build statistics") + } + if hasStat1(t, s) { + t.Error("sqlite_stat1 built despite the refresh being disabled") + } +} + +// The whole design rests on this: statistics live in the file, so the boot-time +// build is a one-off per database. If a future change ever wrote them somewhere +// per-process, EnsurePlannerStats would silently run a full ANALYZE on every +// restart and this test is what says so. +func TestPlannerStatsSurviveReopen_Issue2058(t *testing.T) { + dir := t.TempDir() + dbPath := dir + "/reopen.db" + + first, err := OpenStore(dbPath) + if err != nil { + t.Fatal(err) + } + first.WaitForAsyncMigrations() + if !first.EnsurePlannerStats(10000) { + t.Fatal("first store did not build statistics") + } + first.Close() + + second, err := OpenStore(dbPath) + if err != nil { + t.Fatal(err) + } + defer second.Close() + second.WaitForAsyncMigrations() + + if !hasStat1(t, second) { + t.Fatal("statistics did not survive reopening the database") + } + if second.EnsurePlannerStats(10000) { + t.Error("the reopened store rebuilt statistics, so a restart would pay for an ANALYZE it does not need") + } +} + +// An operator watching a first deploy sees ingest stop for minutes. The warning +// is what tells them it is an ANALYZE and not a hang, so it is worth pinning: +// measured on staging, the build held the write connection 3m43.9s and the +// observations table took zero rows for four minutes. +func TestEnsurePlannerStatsWarnsBeforeBuilding_Issue2058(t *testing.T) { + s := newTestStore(t) + defer s.Close() + + var buf bytes.Buffer + orig := log.Writer() + log.SetOutput(&buf) + defer log.SetOutput(orig) + + s.EnsurePlannerStats(10000) + + out := buf.String() + for _, want := range []string{"no planner statistics", "write connection", "Once per database"} { + if !strings.Contains(out, want) { + t.Errorf("the first-build warning does not mention %q; an operator seeing ingest stall gets no explanation.\ngot: %s", want, out) + } + } + if i, j := strings.Index(out, "no planner statistics"), strings.Index(out, "statistics built in"); i == -1 || j == -1 || i > j { + t.Errorf("the warning must come before the completion line, so it is visible while the write path is held; got: %s", out) + } +} + +// The counterpart: a restart must not log the warning, or every boot looks like +// it is about to stall. +func TestEnsurePlannerStatsIsQuietWhenPresent_Issue2058(t *testing.T) { + s := newTestStore(t) + defer s.Close() + s.RefreshPlannerStats(10000) + + var buf bytes.Buffer + orig := log.Writer() + log.SetOutput(&buf) + defer log.SetOutput(orig) + + s.EnsurePlannerStats(10000) + + if out := buf.String(); strings.Contains(out, "no planner statistics") { + t.Errorf("warned about building on a database that already has statistics: %s", out) + } +} diff --git a/cmd/ingestor/route_mask_backfill_test.go b/cmd/ingestor/route_mask_backfill_test.go index 7b27eddb8..b5156346a 100644 --- a/cmd/ingestor/route_mask_backfill_test.go +++ b/cmd/ingestor/route_mask_backfill_test.go @@ -306,7 +306,10 @@ func TestMain_StartsRouteMaskBackfillAfterBufferReady(t *testing.T) { s := string(src) subscribe := strings.Index(s, "c.Subscribe(") ready := strings.Index(s, "ingestBuffer.Ready()") - start := strings.Index(s, "store.StartRouteMaskBackfill(") + // Since issue #100 main starts the backfill through the startup sequence + // that runs the one-off ANALYZE first (planner_stats_startup.go; the + // sequence itself is pinned in planner_stats_startup_test.go). + start := strings.Index(s, "store.startPlannerStatsThenRouteMaskBackfill(") shutdown := strings.Index(s, `log.Println("Shutting down...")`) stop := strings.LastIndex(s, "stopRouteMaskBackfill()") disconnect := strings.LastIndex(s, "c.Disconnect(5000)") @@ -314,7 +317,7 @@ func TestMain_StartsRouteMaskBackfillAfterBufferReady(t *testing.T) { t.Fatalf("markers not found: subscribe=%d ready=%d start=%d shutdown=%d stop=%d disconnect=%d", subscribe, ready, start, shutdown, stop, disconnect) } if !(subscribe < ready && ready < start) { - t.Errorf("StartRouteMaskBackfill must come after MQTT subscribe and ingestBuffer.Ready()") + t.Errorf("the route_mask backfill (via startPlannerStatsThenRouteMaskBackfill) must start after MQTT subscribe and ingestBuffer.Ready()") } if !(shutdown < stop && stop < disconnect) { t.Errorf("shutdown must cancel the route_mask backfill before disconnecting MQTT clients") diff --git a/cmd/server/planner_stats_readonly_test.go b/cmd/server/planner_stats_readonly_test.go new file mode 100644 index 000000000..fb9bc070b --- /dev/null +++ b/cmd/server/planner_stats_readonly_test.go @@ -0,0 +1,165 @@ +package main + +import ( + "database/sql" + "fmt" + "path/filepath" + "strings" + "testing" +) + +// Issue #100 (port of `Kpa-clawbot/CoreScope#2072`/`#2074`): the ingestor +// writes sqlite_stat1; the server only reads it, through its mode=ro handle. +// These pin that the statistics reach that handle and change the plan of the +// region-filtered channel query, and that the handle cannot write them itself. + +// plannerStatsFixture builds a v3-shaped database with the ingestor's two +// competing indexes and a staging-like payload mix: ~10% GRP_TXT +// (payload_type 5), the rest spread over the other types, ~8 observations per +// transmission. Built through a separate read-write connection, the way the +// ingestor owns the file. +func plannerStatsFixture(t *testing.T) (path string, rw *sql.DB) { + t.Helper() + path = filepath.Join(t.TempDir(), "stats.db") + rw, err := sql.Open("sqlite", path+"?_journal_mode=WAL") + if err != nil { + t.Fatal(err) + } + rw.SetMaxOpenConns(1) + t.Cleanup(func() { rw.Close() }) + for _, q := range []string{ + `CREATE TABLE transmissions (id INTEGER PRIMARY KEY, raw_hex TEXT, hash TEXT UNIQUE, first_seen TEXT, + route_type INTEGER, payload_type INTEGER, payload_version INTEGER, decoded_json TEXT, channel_hash TEXT)`, + `CREATE TABLE observers (rowid INTEGER PRIMARY KEY, id TEXT UNIQUE, name TEXT, iata TEXT)`, + `CREATE TABLE observations (id INTEGER PRIMARY KEY, transmission_id INTEGER, observer_idx INTEGER, + direction TEXT, snr REAL, rssi REAL, score INTEGER, path_json TEXT, timestamp INTEGER)`, + `CREATE INDEX idx_transmissions_payload_type ON transmissions(payload_type)`, + `CREATE INDEX idx_tx_channel_hash ON transmissions(channel_hash) WHERE payload_type = 5`, + `CREATE INDEX idx_observations_transmission_id ON observations(transmission_id)`, + `BEGIN`, + } { + if _, err := rw.Exec(q); err != nil { + t.Fatalf("%s: %v", q, err) + } + } + for i := 1; i <= 20; i++ { + iata := "AAA" + if i%2 == 0 { + iata = "BBB" + } + rw.Exec(`INSERT INTO observers (rowid, id, name, iata) VALUES (?, ?, ?, ?)`, i, fmt.Sprintf("obs%02d", i), fmt.Sprintf("Obs %d", i), iata) + } + types := []int{4, 4, 4, 2, 1, 1, 0, 3, 9} // plus 5 on every 10th row + for i := 1; i <= 4000; i++ { + pt, ch := types[i%len(types)], interface{}(nil) + if i%10 == 0 { + pt, ch = 5, fmt.Sprintf("#ch%d", i%40) + } + if _, err := rw.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json, channel_hash) + VALUES (?, '00', ?, ?, 1, ?, '{"text":"a: b"}', ?)`, i, fmt.Sprintf("h%05d", i), fmt.Sprintf("2026-09-01T00:%02d:00Z", i%60), pt, ch); err != nil { + t.Fatal(err) + } + for o := 0; o < 8; o++ { + rw.Exec(`INSERT INTO observations (transmission_id, observer_idx, timestamp) VALUES (?, ?, ?)`, i, 1+(i+o)%20, 1788000000+i) + } + } + if _, err := rw.Exec(`COMMIT`); err != nil { + t.Fatal(err) + } + return path, rw +} + +// The same shape as the v3 region branch of DB.GetChannels. +const plannerStatsChannelsSQL = `SELECT t.channel_hash, COUNT(*) AS msg_count, MAX(t.first_seen) AS last_activity + FROM transmissions t + JOIN observations o ON o.transmission_id = t.id + LEFT JOIN observers obs ON obs.rowid = o.observer_idx + WHERE t.payload_type = 5 + AND t.channel_hash IS NOT NULL + AND t.channel_hash NOT LIKE 'enc_%' + AND obs.rowid IS NOT NULL AND UPPER(TRIM(obs.iata)) IN (?) + GROUP BY t.channel_hash + ORDER BY last_activity DESC` + +func channelsPlan(t *testing.T, conn *sql.DB) string { + t.Helper() + rows, err := conn.Query(`EXPLAIN QUERY PLAN `+plannerStatsChannelsSQL, "AAA") + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var plan []string + for rows.Next() { + var id, parent, notused int + var detail string + if err := rows.Scan(&id, &parent, ¬used, &detail); err != nil { + t.Fatal(err) + } + plan = append(plan, detail) + } + return strings.Join(plan, " | ") +} + +func TestServerReadsPlannerStatsThroughReadOnlyHandle_Issue100(t *testing.T) { + path, rw := plannerStatsFixture(t) + + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if !db.isV3() { + t.Fatal("fixture not detected as v3; the plan below would be for the wrong branch") + } + + before := channelsPlan(t, db.conn) + if !strings.Contains(before, "idx_transmissions_payload_type") { + t.Fatalf("without statistics the plan should drive from idx_transmissions_payload_type (the #2058 plan); got %s", before) + } + + // The server cannot build the statistics itself: its handle is mode=ro. + if _, err := db.conn.Exec(`ANALYZE`); err == nil { + t.Fatal("ANALYZE succeeded on the server's handle; it is not read-only") + } + + // The ingestor's side: bounded ANALYZE on its own read-write connection. + if _, err := rw.Exec(`PRAGMA analysis_limit=10000`); err != nil { + t.Fatal(err) + } + if _, err := rw.Exec(`ANALYZE`); err != nil { + t.Fatal(err) + } + + // The already-open read-only handle picks the statistics up without a + // restart: ANALYZE changes the schema cookie, so the next statement + // reloads sqlite_stat1. + var rowsStat int + if err := db.conn.QueryRow(`SELECT count(*) FROM sqlite_stat1`).Scan(&rowsStat); err != nil || rowsStat == 0 { + t.Fatalf("server handle sees %d sqlite_stat1 rows (err %v)", rowsStat, err) + } + after := channelsPlan(t, db.conn) + t.Logf("plan without stats: %s", before) + t.Logf("plan with stats: %s", after) + if !strings.Contains(after, "idx_tx_channel_hash") { + t.Fatalf("with statistics the plan should drive from the partial idx_tx_channel_hash; got %s", after) + } + + // And a freshly opened read-only handle (a server restart) reads the same. + db2, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + defer db2.Close() + if p := channelsPlan(t, db2.conn); !strings.Contains(p, "idx_tx_channel_hash") { + t.Fatalf("a reopened read-only handle lost the statistics; plan %s", p) + } + + // The real query still answers through the read-only handle. + chans, err := db2.GetChannels("AAA") + if err != nil { + t.Fatal(err) + } + if len(chans) == 0 { + t.Fatal("GetChannels returned nothing for region AAA") + } +} diff --git a/config.example.json b/config.example.json index b0282f606..24f4089a7 100644 --- a/config.example.json +++ b/config.example.json @@ -17,11 +17,12 @@ "db": { "vacuumOnStartup": false, "incrementalVacuumPages": 1024, + "analysisLimit": 10000, "load": { "chunkSize": 10000, "_comment": "chunkSize: rows fetched per chunk by PacketStore.LoadChunked during startup (#1009). Default 10000. Lower values surface the early-HTTP-readiness signal sooner (the listener binds after the first chunk) at the cost of more SQL round-trips. Higher values reduce per-chunk overhead but delay first-chunk readiness. The X-CoreScope-Load-Status response header reports loading|ready and progress= until the load completes." }, - "_comment": "vacuumOnStartup: run one-time full VACUUM to enable incremental auto-vacuum on existing DBs. Executed by the INGESTOR at startup, BEFORE the MQTT subscriber starts (#1283), so there is no contention with concurrent writes. Blocks ingestor startup for minutes on large DBs; requires 2x DB file size in free disk space. incrementalVacuumPages: free pages returned to OS after each retention reaper cycle (default 1024). See #919. load.chunkSize: see nested _comment (#1009).", + "_comment": "vacuumOnStartup: run one-time full VACUUM to enable incremental auto-vacuum on existing DBs. Executed by the INGESTOR at startup, BEFORE the MQTT subscriber starts (#1283), so there is no contention with concurrent writes. Blocks ingestor startup for minutes on large DBs; requires 2x DB file size in free disk space. incrementalVacuumPages: free pages returned to OS after each retention reaper cycle (default 1024). See #919. analysisLimit: index rows a bounded ANALYZE visits per index when the INGESTOR refreshes the SQLite planner statistics (sqlite_stat1): once at startup on a database that has none, then daily (default 10000; 0 in the file means the default, not unbounded; negative disables both). The first build holds the write lock once per database, so ingest buffers for its duration (see docs/user-guide/database.md). load.chunkSize: see nested _comment (#1009).", "_comment_slowWriterMs": "#1340 — SQLite writer-lock log threshold (default 500). Any wrapped writer call (tagged neighbor_builder, mqtt_handler, prune_packets, prune_observers, prune_metrics, vacuum) whose hold_ms exceeds this emits a single [db-slow-writer] log line. Configured per-process via the CORESCOPE_DB_SLOW_WRITER_MS environment variable on the INGESTOR (e.g. CORESCOPE_DB_SLOW_WRITER_MS=200 for tighter alerting). Per-component wait_ms / hold_ms / contention_total histograms are surfaced via /api/perf/write-sources under .writer_perf regardless of this threshold." }, "listLimits": { diff --git a/docs/user-guide/configuration.md b/docs/user-guide/configuration.md index 89b61eaed..cbe232047 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -110,6 +110,7 @@ How long (in hours) before a node is marked degraded or silent: |-------|---------|-------------| | `db.vacuumOnStartup` | `false` | Run a one-time full `VACUUM` on startup to enable incremental auto-vacuum (blocks for minutes on large DBs) | | `db.incrementalVacuumPages` | `1024` | Free pages returned to the OS after each retention reaper cycle | +| `db.analysisLimit` | `10000` | Index rows per index for the ingestor's bounded `ANALYZE` (planner statistics). Negative disables it. See [Database](database.md#planner-statistics-analyze) | See [Database](database.md) for details on SQLite auto-vacuum, WAL, and manual maintenance. See [#919](https://github.com/Kpa-clawbot/CoreScope/issues/919) for background. diff --git a/docs/user-guide/database.md b/docs/user-guide/database.md index feaf6c1e2..a664b74ba 100644 --- a/docs/user-guide/database.md +++ b/docs/user-guide/database.md @@ -55,6 +55,33 @@ and holds an exclusive lock. |-------|---------|-------------| | `db.vacuumOnStartup` | `false` | One-time full VACUUM to enable incremental auto-vacuum | | `db.incrementalVacuumPages` | `1024` | Pages returned to OS per reaper cycle | +| `db.analysisLimit` | `10000` | Index rows per index for the planner-statistics `ANALYZE` (negative disables it) | + +## Planner statistics (ANALYZE) + +The ingestor keeps SQLite's query-planner statistics (`sqlite_stat1`) up to +date with a bounded `ANALYZE` (`PRAGMA analysis_limit`, default 10000 rows per +index). Without them SQLite picks the plain `payload_type` index for the +channel queries instead of the partial `idx_tx_channel_hash`. + +- **First start on a database without statistics:** the ingestor builds them + right after startup. The `ANALYZE` holds the single write connection until + it finishes, so ingest pauses and the ingest buffer catches up afterwards. + The log says so before it starts (`[analyze] this database has no planner + statistics; building them now …`). This happens once per database: the + statistics are stored in the file. +- **Every start after that:** one `sqlite_master` lookup, then a refresh + 2 minutes after startup and every 24 h. +- **Order at startup:** the statistics build runs before the `route_mask` + backfill, and each waits for the ingest buffer to drain first, so the two + write-lock holds never run back to back. +- **Disable:** set `db.analysisLimit` to a negative value. +- **Remove the statistics:** `DROP TABLE sqlite_stat1;` against the database, + with the ingestor stopped (the server's connection is read-only). An older + image simply ignores the table. With `analysisLimit` still enabled the next + ingestor start rebuilds it. + +The server only reads the statistics; it never runs `ANALYZE` itself. ## Manual VACUUM diff --git a/internal/dbconfig/dbconfig.go b/internal/dbconfig/dbconfig.go index 7526fa676..21ac79416 100644 --- a/internal/dbconfig/dbconfig.go +++ b/internal/dbconfig/dbconfig.go @@ -6,6 +6,7 @@ package dbconfig type DBConfig struct { VacuumOnStartup bool `json:"vacuumOnStartup"` // one-time full VACUUM on startup if auto_vacuum is not INCREMENTAL IncrementalVacuumPages int `json:"incrementalVacuumPages"` // pages returned to OS per reaper cycle (default 1024) + AnalysisLimit int `json:"analysisLimit"` // index rows ANALYZE visits per index (default 10000); negative disables the planner stats refresh (#2058) // Load controls chunked startup loading (#1009). Load *LoadConfig `json:"load,omitempty"`