diff --git a/bench/load/first_index_measurement_test.go b/bench/load/first_index_measurement_test.go new file mode 100644 index 000000000..086c2f87a --- /dev/null +++ b/bench/load/first_index_measurement_test.go @@ -0,0 +1,173 @@ +package main + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "sync" + "testing" + "time" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + + "github.com/zzet/gortex/internal/config" + "github.com/zzet/gortex/internal/graph/store_sqlite" + "github.com/zzet/gortex/internal/indexer" + "github.com/zzet/gortex/internal/parser" + "github.com/zzet/gortex/internal/parser/languages" +) + +// TestMeasureStandaloneFirstIndex is an opt-in, single-build harness. Run the +// identical harness on the control and candidate with an owned clean corpus; +// compare reports separately. It never grants a ratio from a single run. +func TestMeasureStandaloneFirstIndex(t *testing.T) { + corpus := os.Getenv("GORTEX_FIRST_INDEX_CORPUS") + if corpus == "" { + t.Skip("set GORTEX_FIRST_INDEX_CORPUS and GORTEX_FIRST_INDEX_REPORT for an owned-fixture measurement") + } + reportPath := os.Getenv("GORTEX_FIRST_INDEX_REPORT") + if reportPath == "" { + t.Fatal("GORTEX_FIRST_INDEX_REPORT is required") + } + corpus, err := filepath.Abs(corpus) + if err != nil { + t.Fatal(err) + } + git := func(args ...string) string { + t.Helper() + result, err := exec.Command("git", append([]string{"-C", corpus}, args...)...).CombinedOutput() + if err != nil { + t.Fatalf("fixture git %v: %v: %s", args, err, result) + } + return string(result) + } + if status := git("status", "--porcelain", "--untracked-files=all"); status != "" { + t.Fatalf("fixture must be clean: %s", status) + } + sourceSHA := strings.TrimSpace(git("rev-parse", "HEAD")) + manifest := sha256.Sum256([]byte(git("ls-files", "-s", "-z"))) + output, err := os.OpenFile(reportPath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) + if err != nil { + t.Fatal(err) + } + defer output.Close() + store, err := store_sqlite.Open(filepath.Join(t.TempDir(), "fresh.sqlite")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := store.Close(); err != nil { + t.Error(err) + } + }) + sampler := newProcessSampler() + type sample struct { + At time.Time `json:"at"` + Metrics map[string]Metric `json:"metrics"` + } + var mu sync.Mutex + var samples []sample + var before, after map[string]Metric + var startAt, endAt time.Time + started, finished, joined := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var startOnce, endOnce sync.Once + core, logs := observer.New(zap.InfoLevel) + logger := zap.New(core, zap.Hooks(func(entry zapcore.Entry) error { + switch entry.Message { + case "indexer: drain start (shadow → disk)": + startOnce.Do(func() { + mu.Lock() + startAt = entry.Time + before = sampler.sample(os.Getpid()) + samples = append(samples, sample{time.Now(), before}) + mu.Unlock() + close(started) + }) + case "indexer: FlushBulk complete": + endOnce.Do(func() { + mu.Lock() + endAt = entry.Time + after = sampler.sample(os.Getpid()) + samples = append(samples, sample{time.Now(), after}) + mu.Unlock() + close(finished) + }) + } + return nil + })) + go func() { + defer close(joined) + select { + case <-started: + case <-finished: + return + } + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for { + select { + case <-finished: + return + case <-ticker.C: + mu.Lock() + if endAt.IsZero() { + samples = append(samples, sample{time.Now(), sampler.sample(os.Getpid())}) + } + mu.Unlock() + } + } + }() + registry := parser.NewRegistry() + languages.RegisterAll(registry) + idx := indexer.New(store, registry, config.Default().Index, logger) + t.Cleanup(idx.Close) + result, indexErr := idx.IndexCtx(t.Context(), corpus) + endOnce.Do(func() { close(finished) }) + <-joined + complete := indexErr == nil && !startAt.IsZero() && !endAt.IsZero() + deltas := make(map[string]Metric) + for _, name := range []string{"diskio_written_bytes", "logical_written_bytes", "syscall_written_bytes"} { + if complete { + deltas[name] = counterDelta(before[name], after[name]) + } else { + deltas[name] = unavailable("bytes", "complete drain endpoints unavailable") + } + } + entries := logs.FilterMessage("indexer: FlushBulk complete").All() + var drainFields map[string]any + if len(entries) == 1 { + drainFields = entries[0].ContextMap() + } + report := map[string]any{ + "go_version": runtime.Version(), "goos": runtime.GOOS, "goarch": runtime.GOARCH, "gomaxprocs": runtime.GOMAXPROCS(0), + "complete": complete, "corpus_sha": sourceSHA, "tracked_manifest_sha256": hex.EncodeToString(manifest[:]), + "result": result, "drain_start": startAt, "drain_end": endAt, "drain_fields": drainFields, + "process_write_deltas": deltas, "process_samples": samples, + "qualification": "process-wide counters over drain log boundaries; concurrent process activity can contribute; macOS logical writes exclude external-volume ledger; diskio is the physical meter; unavailable observations remain null; five-second samples plus endpoints; compare identical frozen corpus/harness/host/storage separately; no acceptance inferred", + } + if complete { + report["drain_and_seal_seconds"] = endAt.Sub(startAt).Seconds() + } + if indexErr != nil { + report["error"] = indexErr.Error() + } + if err := json.NewEncoder(output).Encode(report); err != nil { + t.Fatal(err) + } + if status := git("status", "--porcelain", "--untracked-files=all"); status != "" { + t.Fatalf("fixture changed: %s", status) + } + if strings.TrimSpace(git("rev-parse", "HEAD")) != sourceSHA || sha256.Sum256([]byte(git("ls-files", "-s", "-z"))) != manifest { + t.Fatal("fixture identity changed") + } + if !complete { + t.Fatalf("first index did not complete one measured drain: %v", indexErr) + } +} diff --git a/cmd/gortex/daemon_controller.go b/cmd/gortex/daemon_controller.go index df8ec5892..14d526e61 100644 --- a/cmd/gortex/daemon_controller.go +++ b/cmd/gortex/daemon_controller.go @@ -1782,6 +1782,14 @@ func storageStatusFor(g graph.Store) *daemon.StorageStatus { CloseCheckpointEstimateMS: estimate.Milliseconds(), WALReclaim: walReclaimStatus(reporter.WALReclaimStats()), } + if reporter, ok := g.(interface{ DenseIndexesMissing() []string }); ok { + out.DenseIndexesMissing = reporter.DenseIndexesMissing() + } + if reporter, ok := g.(interface { + LastFirstIndexDrains() map[string]graph.FirstIndexDrain + }); ok { + out.LastFirstIndexDrain = reporter.LastFirstIndexDrains() + } if policy, ok := g.(interface{ CloseCheckpointPolicy() string }); ok { out.CloseCheckpointPolicy = policy.CloseCheckpointPolicy() } diff --git a/cmd/gortex/daemon_state.go b/cmd/gortex/daemon_state.go index 2d7d81663..785de26c7 100644 --- a/cmd/gortex/daemon_state.go +++ b/cmd/gortex/daemon_state.go @@ -745,7 +745,7 @@ func warmupDaemonStateContext(ctx context.Context, state *daemonState, logger *z coordinatedBulkActive := coordinatedBulk != nil && coordinatedBulk.BeginCoordinatedBulkLoad() defer func() { if coordinatedBulkActive { - if err := coordinatedBulk.EndCoordinatedBulkLoad(); err != nil { + if err := graph.EndOrAbortCoordinatedBulkLoad(coordinatedBulk); err != nil { logger.Error("daemon: coordinated cold bulk-load cleanup failed", zap.Error(err)) } } @@ -844,9 +844,14 @@ func warmupDaemonStateContext(ctx context.Context, state *daemonState, logger *z // final seal includes them in bulk index creation, checkpointing, and // planner statistics instead of reopening ordinary write transactions // immediately after a large cold load. + // The finalize is the window's last call: the deferred end above is + // skipped once it has run, and nothing retries a failed seal. So a + // failed end is followed by an abort, which closes the window and + // leaves the indexes it could not rebuild to the store's repair, rather + // than keeping the warmup's relaxed-durability writer until shutdown. var finalizeBulk func() error if coordinatedBulkActive { - finalizeBulk = coordinatedBulk.EndCoordinatedBulkLoad + finalizeBulk = func() error { return graph.EndOrAbortCoordinatedBulkLoad(coordinatedBulk) } } drain := state.multiIndexer.RunDeferredGoModAll if batchCtx.Err() != nil { diff --git a/cmd/gortex/daemon_warmup_bulk_test.go b/cmd/gortex/daemon_warmup_bulk_test.go index 393bf6000..61b3980ab 100644 --- a/cmd/gortex/daemon_warmup_bulk_test.go +++ b/cmd/gortex/daemon_warmup_bulk_test.go @@ -1,10 +1,45 @@ package main import ( + "context" + "encoding/json" "errors" + "io" + "log" + "os" + "path/filepath" + "strings" "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" + + "github.com/zzet/gortex/internal/config" + "github.com/zzet/gortex/internal/graph" + "github.com/zzet/gortex/internal/graph/store_sqlite" + "github.com/zzet/gortex/internal/indexer" + "github.com/zzet/gortex/internal/parser" + "github.com/zzet/gortex/internal/parser/languages" ) +// The existing missing-index log is emitted after the atomic record is installed +// and before repair is scheduled. Observe status at that causal boundary instead +// of racing a successful background repair after warmup returns. +type warmupSealStatusProbe struct { + output io.Writer + missing func() +} + +func (p warmupSealStatusProbe) Write(data []byte) (int, error) { + if strings.Contains(string(data), "store_sqlite: dense indexes missing after bulk window close") { + p.missing() + } + return p.output.Write(data) +} + func TestDrainGoModAndFinalizeWarmupBulkOrdersDrainBeforeFinalize(t *testing.T) { var order []string wantErr := errors.New("finalize failed") @@ -38,3 +73,92 @@ func TestDrainGoModAndFinalizeWarmupBulkWithoutBulkFinalizer(t *testing.T) { t.Fatalf("finalize elapsed = %s, want 0", finalizeElapsed) } } + +// TestWarmupFailedFinalSealGivesUpTheCoordinatedWindow: a cold warmup whose +// final seal fails logs the failure and closes the coordinated window, rather +// than keeping its relaxed-durability writer until the daemon shuts down. The +// writer is back at its own synchronous level, the planner probe no longer +// sees a live window, and the store's repair rebuilds the index the seal +// could not. +func TestWarmupFailedFinalSealGivesUpTheCoordinatedWindow(t *testing.T) { + t.Setenv("GORTEX_PARSER_ISOLATION", "0") + root := t.TempDir() + t.Setenv("HOME", filepath.Join(root, "home")) + t.Setenv("XDG_CONFIG_HOME", filepath.Join(root, "config")) + t.Setenv("XDG_DATA_HOME", filepath.Join(root, "data")) + t.Setenv("XDG_CACHE_HOME", filepath.Join(root, "cache")) + repo := filepath.Join(root, "repo") + require.NoError(t, os.Mkdir(repo, 0o700)) + require.NoError(t, os.WriteFile(filepath.Join(repo, "main.go"), []byte("package main\nfunc Main() {}\n"), 0o600)) + store, err := store_sqlite.Open(filepath.Join(root, "store.sqlite")) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, store.Close()) }) + ctx := context.Background() + writerSync, err := store.WriterPragmaForTest(ctx, "synchronous") + require.NoError(t, err) + cm, err := config.NewConfigManager(filepath.Join(root, "config.yaml")) + require.NoError(t, err) + require.NoError(t, cm.Global().AddRepo(config.RepoEntry{Path: repo})) + reg := parser.NewRegistry() + reg.Register(languages.NewGoExtractor()) + logs, observed := observer.New(zap.InfoLevel) + logger := zap.New(logs) + idx := indexer.New(store, reg, config.Default().Index, logger) + mi := indexer.NewMultiIndexer(store, reg, idx.Search(), cm, logger) + t.Cleanup(func() { require.NoError(t, mi.Close(context.Background())) }) + state := &daemonState{graph: store, indexer: idx, multiIndexer: mi, configManager: cm} + injected := errors.New("injected final seal failure") + restore := store.SetBulkSealHookForTest(func(index string) error { + if index == "nodes_by_kind" { + return injected + } + return nil + }) + + missingStatus := make(chan []string, 1) + priorOutput := log.Writer() + log.SetOutput(warmupSealStatusProbe{output: priorOutput, missing: func() { + select { + case missingStatus <- storageStatusFor(store).DenseIndexesMissing: + default: + } + }}) + defer log.SetOutput(priorOutput) + _, _ = warmupDaemonState(state, logger, func() {}) + select { + case missing := <-missingStatus: + assert.Contains(t, missing, "nodes_by_kind", "status reads the existing missing-index record before repair is scheduled") + default: + t.Error("the failed seal never published its missing-index boundary") + } + restore() + require.NotEmpty(t, observed.FilterMessage("daemon: coordinated cold bulk-load finalize failed").All(), + "the fixture must fail the coordinated window's final seal") + gotSync, err := store.WriterPragmaForTest(ctx, "synchronous") + require.NoError(t, err) + assert.Equal(t, writerSync, gotSync, "writer durability must be restored") + health, err := store.PlannerStatsHealth(ctx) + require.NoError(t, err) + assert.NotEqual(t, "bulk_window_active", health.Reason, "the failed window must not stay open") + deadline := time.Now().Add(10 * time.Second) + for health.Reason == "dense_indexes_missing" && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + health, err = store.PlannerStatsHealth(ctx) + require.NoError(t, err) + } + assert.NotEqual(t, "dense_indexes_missing", health.Reason, "the repair must rebuild the index the seal left out") +} + +func TestStorageStatusReportsFirstIndexDrainWithoutInventingEmptyResult(t *testing.T) { + store, err := store_sqlite.Open(filepath.Join(t.TempDir(), "status.sqlite")) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, store.Close()) }) + empty, err := json.Marshal(storageStatusFor(store)) + require.NoError(t, err) + assert.NotContains(t, string(empty), "last_first_index_drain") + store.RecordFirstIndexDrain("repo", graph.FirstIndexDrain{Window: "coordinated", Nodes: 7, Edges: 9}) + response := storageStatusFor(store) + require.NotNil(t, response) + assert.Equal(t, 7, response.LastFirstIndexDrain["repo"].Nodes) + assert.Equal(t, "coordinated", response.LastFirstIndexDrain["repo"].Window) +} diff --git a/docs/mcp.md b/docs/mcp.md index df13918d7..61cc4028f 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -526,7 +526,7 @@ Gortex captures every large tool response into a bounded per-session ring; these |------|-------------| | `analyze` | Unified graph analysis dispatcher. `kind` ∈ `dead_code`, `hotspots`, `cycles`, `would_create_cycle`, `connectivity_health`, `todos`, `blame`, `coverage`, `coverage_gaps`, `coverage_summary`, `stale_code`, `stale_flags`, `ownership`, `releases`, `cgo_users`, `wasm_users`, `orphan_tables`, `unreferenced_tables`, `channel_ops`, `goroutine_spawns`, `field_writers`, `race_writes`, `unclosed_channels`, `unsafe_patterns`, `health_score`, `impact`, `annotation_users`, `config_readers`, `env_var_users`, `sql_call_sites`, `fixes_history`, `edge_audit`, `domain`, `named`, `tests_as_edges`, `clusters`, `event_emitters`, `pubsub`, `string_emitters`, `error_surface`, `log_events`, `sql_rebuild`, `external_calls`, `routes`, `models`, `components`, `k8s_resources`, `images`, `kustomize`, `cross_repo`, `dbt_models`, `synthesizers`, `resolution_outcomes`. `clusters` takes an `algorithm` arg (`leiden` / `louvain` / `spectral`). `impact` takes an optional `target` (`{symbol}` or `{file}`): with one it ranks that target's blast radius — the target row first, then its transitive dependents with their `depth` — and reports the closure width plus whether it is exact; without one it ranks the whole repo. An unresolvable target is a structured error, never a silent fall-back to the repo-wide ranking. `synthesizers` rolls up every framework-dispatch-synthesized edge by the pass that produced it; `resolution_outcomes` classifies unresolved call/reference edges by why the resolver gave up (`ambiguous_multi_match` / `candidate_out_of_scope` / `cross_language_only` / `stub_only` / `no_definition`) | | `find_clones` | Near-duplicate function/method clusters from the MinHash + LSH `similar_to` layer; `dead_only: true` finds dead duplicates of live code | -| `index_health` | Health score, parse failures, stale files, language coverage, tracked-repo path liveness (`tracked_repo_paths_ok` + `missing_repo_paths` — a repo whose directory was deleted still holds its registration and silently drops out of workspace-wide answers), per-(repo, provider) semantic-enrichment lifecycle (`semantic_enrichment`: running / completed / partial / abandoned / failed with edge counts, plus a `semantic_enrichment_ok` rollup) — a green file count with a `partial` enrichment state means LSP-tier edges are incomplete. `path_liveness` asks the same question one level down, per file: it stats the paths the graph itself claims and reports how many indexed files no longer exist on disk (`orphan_files` / `orphan_rate` / `orphans_by_repo`, sampled with `truncated: true` past 20k files). `stale_files` only covers files the daemon still tracks, so a deletion it never witnessed shows up here and nowhere else; a non-zero `orphan_files` caps `health_score`. On a SQLite-backed daemon a `planner_stats` object reports the query planner's own view of the store: `nodes` / `edges` carry `believed` (the leading token of the index's `sqlite_stat1` row) against `actual_from_counters` (the `repo_index_state` counter sum across ALL view generations, flagged by `counters_known`; no counter row describes the external-call bucket or nodes carrying no repo prefix, and a leftover empty-prefix row on a store that later became multi-repo is summed on top of the per-repo rows, so the pair is an approximation of index cardinality and a difference between the two is not a measured discrepancy), `receivers` carries a `bounded: true` LIMIT-capped probe of the Go receiver index plus `complete` (false means the probe stopped at its cap, so `actual` is a lower bound and must not be compared with `believed`) and is omitted entirely when the receiver index is not in the schema — a bulk load has dropped it, or this store never had it, and a `believed: 0` there would read as the poisoned near-zero row rather than as an absent index. `stale` says whether the graph has outgrown what the planner believes — which can invert join order on receiver/edge queries — while `reason` names every verdict, staleness or not: `bulk_window_active` is why a cold load in progress reports zeros. `last_refresh_at` / `last_refresh_reason` appear once this process has rebuilt them. The field is read-only: statistics refresh at the next repository index, commit/HEAD move, whole-graph resolve or generation build, never from reading this payload — and that refresh is cooperative, so a busy writer, a spent per-pass budget or a per-index timeout defers the rest of the work to the following boundary (which resumes where the last one stopped) rather than stalling the pipeline that noticed | +| `index_health` | Health score, parse failures, stale files, language coverage, tracked-repo path liveness (`tracked_repo_paths_ok` + `missing_repo_paths` — a repo whose directory was deleted still holds its registration and silently drops out of workspace-wide answers), per-(repo, provider) semantic-enrichment lifecycle (`semantic_enrichment`: running / completed / partial / abandoned / failed with edge counts, plus a `semantic_enrichment_ok` rollup) — a green file count with a `partial` enrichment state means LSP-tier edges are incomplete. `path_liveness` asks the same question one level down, per file: it stats the paths the graph itself claims and reports how many indexed files no longer exist on disk (`orphan_files` / `orphan_rate` / `orphans_by_repo`, sampled with `truncated: true` past 20k files). `stale_files` only covers files the daemon still tracks, so a deletion it never witnessed shows up here and nowhere else; a non-zero `orphan_files` caps `health_score`. On a SQLite-backed daemon a `planner_stats` object reports the query planner's own view of the store: `nodes` / `edges` carry `believed` (the leading token of the index's `sqlite_stat1` row) against `actual_from_counters` (the `repo_index_state` counter sum across ALL view generations, flagged by `counters_known`; no counter row describes the external-call bucket or nodes carrying no repo prefix, and a leftover empty-prefix row on a store that later became multi-repo is summed on top of the per-repo rows, so the pair is an approximation of index cardinality and a difference between the two is not a measured discrepancy), `receivers` carries a `bounded: true` LIMIT-capped probe of the Go receiver index plus `complete` (false means the probe stopped at its cap, so `actual` is a lower bound and must not be compared with `believed`) and is omitted entirely when the receiver index is not in the schema — a bulk load has dropped it, or this store never had it, and a `believed: 0` there would read as the poisoned near-zero row rather than as an absent index. `stale` says whether the graph has outgrown what the planner believes — which can invert join order on receiver/edge queries — while `reason` names every verdict, staleness or not: `bulk_window_active` is why a cold load in progress reports zeros, and `dense_indexes_missing` means a cold load closed without rebuilding some of its dense indexes (typically because the volume filled up during the final index rebuild): `dense_indexes_missing` in the same object names them, lookups that use them scan until they are back, and the store rebuilds them in the background once its volume has free space at least the size of the store (a daemon restart also rebuilds them). `last_refresh_at` / `last_refresh_reason` appear once this process has rebuilt them. The field is read-only: statistics refresh at the next repository index, commit/HEAD move, whole-graph resolve or generation build, never from reading this payload — and that refresh is cooperative, so a busy writer, a spent per-pass budget or a per-index timeout defers the rest of the work to the following boundary (which resumes where the last one stopped) rather than stalling the pipeline that noticed | | `get_symbol_history` | Symbols modified this session with counts; flags churning (3+ edits) | The `analyze` dispatcher also accepts a set of **facade-aliased kinds** that route to the captured legacy handler instead of the dispatcher switch: `processes` → `get_processes`, `communities` → `get_communities`, `contracts` → `contracts`, `architecture` → `get_architecture`, `clones` → `find_clones`, `health` → `audit_health`, `inspections` → `run_inspections`, `recent_changes` → `get_recent_changes`, and the other entries of the facade analyze migration table (see `mcp-facade-v1.md`). These aliases are **surface-independent**: they work for client (facade-v1), explicitly legacy-preset, and session-less HTTP callers alike, with no `tools_search` promotion — the HTTP dashboard endpoints depend on this under the `core`/`defer` default. @@ -774,3 +774,15 @@ Read-only, URI-addressable, no args. Clients that speak resources can `resources | `pre_commit` | Review uncommitted changes — shows changed symbols, blast radius, risk level, affected tests | | `orientation` | Orient in an unfamiliar codebase — graph stats, communities, execution flows, key symbols | | `safe_to_change` | Analyze whether it's safe to change specific symbols — blast radius, edit plan, affected tests | + +`index_health` includes `last_first_index_drain` after a full shadow index has +persisted, keyed by repository prefix. Daemon status exposes the same snapshot +under `storage`, together with `dense_indexes_missing` when a failed seal owes +index repair. A drain records its actual window kind, node/edge counts, drain +and seal-call durations, completion time and any seal error. A coordinated +window seals at its outer boundary, so its per-repository seal call can be a +no-op. `wal_peak_scope=shared_store_drain_chunk_samples` identifies physical WAL +observations at the start, bounded graph/FTS chunks and seal boundaries. This +is a sampled peak on the shared store; concurrent repository work can contribute +and a peak between samples can be missed. Missing observations +are omitted, and these process-local snapshots reset when the store reopens. diff --git a/internal/daemon/proto.go b/internal/daemon/proto.go index 1e02423bc..143c8e491 100644 --- a/internal/daemon/proto.go +++ b/internal/daemon/proto.go @@ -5,6 +5,7 @@ package daemon import ( "encoding/json" "fmt" + "github.com/zzet/gortex/internal/graph" "io" "time" ) @@ -525,8 +526,10 @@ type PublicationPhaseOffset struct { // StorageStatus is the store's write-ahead-log state for daemon status. type StorageStatus struct { - DBBytes int64 `json:"db_bytes"` - WALBytes int64 `json:"wal_bytes"` + DenseIndexesMissing []string `json:"dense_indexes_missing,omitempty"` + LastFirstIndexDrain map[string]graph.FirstIndexDrain `json:"last_first_index_drain,omitempty"` + DBBytes int64 `json:"db_bytes"` + WALBytes int64 `json:"wal_bytes"` // WALPendingFrames is the backlog not yet copied into the database // (mxFrame - nBackfill), -1 when the wal-index could not be read. WALPendingFrames int64 `json:"wal_pending_frames"` diff --git a/internal/graph/first_index_drain.go b/internal/graph/first_index_drain.go new file mode 100644 index 000000000..7f1fc1e24 --- /dev/null +++ b/internal/graph/first_index_drain.go @@ -0,0 +1,21 @@ +package graph + +import "time" + +// FirstIndexDrain is the last full shadow persisted for one repository during +// this store's lifetime. It is an observation, not a readiness or durability +// verdict. A coordinated window defers its final seal to the outer owner. +type FirstIndexDrain struct { + Window string `json:"window"` + Nodes int `json:"nodes"` + Edges int `json:"edges"` + DrainMS float64 `json:"drain_ms"` + SealMS float64 `json:"seal_ms"` + CompletedAt time.Time `json:"completed_at"` + Error string `json:"error,omitempty"` + WALPeakBytes int64 `json:"wal_peak_bytes,omitempty"` + WALPeakObserved bool `json:"wal_peak_observed"` + // WAL peak is sampled at bounded drain/FTS chunks and seal boundaries. + // The WAL is shared: overlapping work may contribute to these observations. + WALPeakScope string `json:"wal_peak_scope"` +} diff --git a/internal/graph/planner_stats.go b/internal/graph/planner_stats.go index 636c45674..d5dd0f6a5 100644 --- a/internal/graph/planner_stats.go +++ b/internal/graph/planner_stats.go @@ -55,6 +55,11 @@ type PlannerStatsFreshness struct { Refreshed bool Reason string + // DenseIndexesMissing names secondary indexes the backend knows are absent + // and owes a rebuild for (an interrupted bulk load), reported with its own + // Reason instead of a verdict. Empty in the ordinary case. + DenseIndexesMissing []string + LastRefreshAt time.Time LastRefreshReason string diff --git a/internal/graph/store.go b/internal/graph/store.go index 5f4ffacc9..ddbdfb58b 100644 --- a/internal/graph/store.go +++ b/internal/graph/store.go @@ -2,6 +2,7 @@ package graph import ( "context" + "errors" "iter" "sync" ) @@ -525,9 +526,27 @@ type BulkLoader interface { // multi-repository cold load. Implementations engage only on an empty store; // nested per-repository BulkLoader calls must leave the outer window open so // secondary indexes are rebuilt once after every repository drains. +// +// A failed EndCoordinatedBulkLoad may leave the window open so its owner can +// retry it. AbortCoordinatedBulkLoad is how an owner that will not retry gives +// it up: the window closes unfinished, ordinary durability comes back, and +// whatever the end could not rebuild is left to the backend to repair. type CoordinatedBulkLoader interface { BeginCoordinatedBulkLoad() bool EndCoordinatedBulkLoad() error + AbortCoordinatedBulkLoad() error +} + +// EndOrAbortCoordinatedBulkLoad ends l's coordinated window and, when the end +// fails, aborts it. It is for an owner's last call on the window: an end that +// fails with the window still open would otherwise leave it open until the +// store closes, with every later write at the window's relaxed durability. +func EndOrAbortCoordinatedBulkLoad(l CoordinatedBulkLoader) error { + err := l.EndCoordinatedBulkLoad() + if err == nil { + return nil + } + return errors.Join(err, l.AbortCoordinatedBulkLoad()) } // SymbolHit is a single full-text-search result: the matched node ID diff --git a/internal/graph/store_sqlite/add_batch_set.go b/internal/graph/store_sqlite/add_batch_set.go index c9dfed5d3..aa7299787 100644 --- a/internal/graph/store_sqlite/add_batch_set.go +++ b/internal/graph/store_sqlite/add_batch_set.go @@ -551,6 +551,26 @@ func insertEdgeChunksTxLimitedContext(ctx context.Context, tx *sql.Tx, viewGen i return rowsInserted, statements, insertedKeys, nil } +// SetBatchWriteHookForTest installs a hook every AddBatch-family write on this +// store calls inside its transaction, under the write gate and before any row +// is written, with the batch's node and edge counts. A returned error fails the +// batch exactly as a failed insert would (the transaction rolls back and the +// error reaches the caller the way that caller reports storage errors); a panic +// unwinds the batch the same way. The hook runs under the write gate, so it +// must not call the store's write methods. restore puts the previous hook back. +func (s *Store) SetBatchWriteHookForTest(hook func(nodeRows, edgeRows int) error) (restore func()) { + previous := s.batchWriteHook.Swap(&hook) + return func() { s.batchWriteHook.Store(previous) } +} + +func (s *Store) runBatchWriteHook(nodeRows, edgeRows int) error { + hook := s.batchWriteHook.Load() + if hook == nil || *hook == nil { + return nil + } + return (*hook)(nodeRows, edgeRows) +} + func (s *Store) addBatchSetOriented(nodes []*graph.Node, edges []*graph.Edge) (sqliteAddBatchStats, error) { return s.addBatchSetOrientedContext(context.Background(), nodes, edges) } @@ -708,6 +728,9 @@ func (s *Store) addBatchSetOrientedContext(ctx context.Context, nodes []*graph.N } defer owner.finish(&err) tx := owner.tx + if err := s.runBatchWriteHook(len(nodes), len(edges)); err != nil { + return stats, err + } var ( receiptDelta *sqliteMutationReceiptAccumulator diff --git a/internal/graph/store_sqlite/bulk_begin_observation_test.go b/internal/graph/store_sqlite/bulk_begin_observation_test.go new file mode 100644 index 000000000..39a0cd577 --- /dev/null +++ b/internal/graph/store_sqlite/bulk_begin_observation_test.go @@ -0,0 +1,94 @@ +package store_sqlite + +import ( + "bytes" + "log" + "strings" + "sync" + "testing" + + "github.com/zzet/gortex/internal/graph" +) + +type bulkAdmissionLogBuffer struct { + mu sync.Mutex + bytes.Buffer +} + +func (b *bulkAdmissionLogBuffer) Write(data []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.Buffer.Write(data) +} +func (b *bulkAdmissionLogBuffer) reset() { b.mu.Lock(); defer b.mu.Unlock(); b.Reset() } +func (b *bulkAdmissionLogBuffer) text() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.String() +} + +func TestBulkBeginReportsEveryAdmissionOutcome(t *testing.T) { + s, _ := openTempStore(t) + var output bulkAdmissionLogBuffer + prior := log.Writer() + log.SetOutput(&output) + t.Cleanup(func() { log.SetOutput(prior) }) + expect := func(window, outcome, reason string, begin func()) { + t.Helper() + output.reset() + begin() + lines := strings.Split(strings.TrimSpace(output.text()), "\n") + var admissions []string + for _, line := range lines { + if strings.Contains(line, "bulk begin ") { + admissions = append(admissions, line) + } + } + want := "window=" + window + " outcome=" + outcome + " reason=" + reason + if len(admissions) != 1 || !strings.Contains(admissions[0], want) { + t.Fatalf("admission logs=%q, want exactly one %q", admissions, want) + } + } + expect("standalone", "engaged", "cold_store", s.BeginBulkLoad) + expect("standalone", "nested_noop", "active_window", s.BeginBulkLoad) + expect("coordinated", "nested_noop", "active_window", func() { s.BeginCoordinatedBulkLoad() }) + if err := s.FlushBulk(); err != nil { + t.Fatal(err) + } + s.AddBatch([]*graph.Node{{ID: "repo/a.go::A", Name: "A", Kind: graph.KindFunction}}, nil) + expect("standalone", "refused_not_cold", "nodes", s.BeginBulkLoad) + expect("coordinated", "refused_not_cold", "nodes", func() { s.BeginCoordinatedBulkLoad() }) + memory, err := Open(":memory:") + if err != nil { + t.Fatal(err) + } + defer memory.Close() + expect("standalone", "memory_store", "memory_store", memory.BeginBulkLoad) + expect("coordinated", "memory_store", "memory_store", func() { memory.BeginCoordinatedBulkLoad() }) + for _, predicate := range []string{"edges", "file_mtimes", "repo_index_state"} { + warm, _ := openTempStore(t) + switch predicate { + case "edges": + nodes, edges := bulkFixture(2, 1) + warm.AddBatch(nodes, edges) + if _, err := warm.writerDB.Exec("DELETE FROM nodes"); err != nil { + t.Fatal(err) + } + case "file_mtimes": + if err := warm.SetFileMtime("repo", "a.go", 1); err != nil { + t.Fatal(err) + } + case "repo_index_state": + if err := warm.SetRepoIndexState(graph.RepoIndexState{RepoPrefix: "repo"}); err != nil { + t.Fatal(err) + } + } + expect("coordinated", "refused_not_cold", predicate, func() { warm.BeginCoordinatedBulkLoad() }) + } + generation, _ := openTempStore(t) + if active, err := generation.BeginGenerationBulkLoad(42); err != nil || !active { + t.Fatalf("generation active=%v err=%v", active, err) + } + expect("standalone", "refused_generation_window", "generation_window", generation.BeginBulkLoad) + expect("coordinated", "refused_generation_window", "generation_window", func() { generation.BeginCoordinatedBulkLoad() }) +} diff --git a/internal/graph/store_sqlite/bulk_load.go b/internal/graph/store_sqlite/bulk_load.go index 4d69e3b1b..3ffbbd75f 100644 --- a/internal/graph/store_sqlite/bulk_load.go +++ b/internal/graph/store_sqlite/bulk_load.go @@ -383,10 +383,8 @@ func (s *Store) activeWriteConnLocked(ctx context.Context) (*sql.Conn, func(), e func (s *Store) BeginBulkLoad() { s.writeMu.Lock() defer s.writeMu.Unlock() - if s.coordinatedBulkLoad { - return - } - s.beginBulkLoadLocked() + outcome, reason := s.beginBulkLoadLocked() + log.Printf("store_sqlite: bulk begin window=standalone outcome=%s reason=%s", outcome, reason) } // BeginCoordinatedBulkLoad opens one outer cold-load window around a set of @@ -398,11 +396,9 @@ func (s *Store) BeginBulkLoad() { func (s *Store) BeginCoordinatedBulkLoad() bool { s.writeMu.Lock() defer s.writeMu.Unlock() - if s.coordinatedBulkLoad || s.bulkConn != nil { - return false - } - s.beginBulkLoadLocked() - if s.bulkConn == nil { + outcome, reason := s.beginBulkLoadLocked() + log.Printf("store_sqlite: bulk begin window=coordinated outcome=%s reason=%s", outcome, reason) + if outcome != "engaged" { return false } s.coordinatedBulkLoad = true @@ -410,14 +406,17 @@ func (s *Store) BeginCoordinatedBulkLoad() bool { return true } -func (s *Store) beginBulkLoadLocked() { - - // Re-entrancy / non-disk guard: a second BeginBulkLoad without an - // intervening FlushBulk, or an in-memory store, stays a no-op. - if s.bulkConn != nil || isMemoryPath(s.dbPath) { - return +func (s *Store) beginBulkLoadLocked() (outcome, reason string) { + if s.generationBulkLoad != 0 { + return "refused_generation_window", "generation_window" } - + if s.coordinatedBulkLoad || s.bulkConn != nil { + return "nested_noop", "active_window" + } + if isMemoryPath(s.dbPath) { + return "memory_store", "memory_store" + } + outcome, reason = "refused_setup", "writer_connection" ctx := context.Background() conn, err := s.writerDB.Conn(ctx) if err != nil { @@ -427,10 +426,11 @@ func (s *Store) beginBulkLoadLocked() { // Nodes alone are not a cold-store proof. A degraded warm store can have // zero nodes while retaining millions of edges, and sidecar receipts prove // an incremental lifecycle even when both graph tables happen to be empty. - if !coldGraphStoreEmpty(ctx, conn) { + if predicate := coldGraphStoreRefusal(ctx, conn); predicate != "" { _ = conn.Close() - return + return "refused_not_cold", predicate } + reason = "pragma_setup" // Capture prior pragma values so FlushBulk (and every early-return / // error path) can restore them. If they can't be read, don't engage — @@ -500,6 +500,7 @@ func (s *Store) beginBulkLoadLocked() { // The bulk path changes durability and secondary-index maintenance outside // the ordinary row mutation protocol. Active receipts therefore fail closed. s.markMutationReceiptsIncompleteLocked() + return "engaged", "cold_store" } // ErrGenerationBulkLoadPopulated refuses a generation-scoped bulk window over @@ -924,13 +925,62 @@ func (s *Store) scheduleWALDrainAboveLine(result walCheckpointResult, boundary s s.scheduleWALDrain(boundary) } +// bulkWindowExit is what a bulk-window exit still owes once the write gate is +// released. The locked half of FlushBulk / EndCoordinatedBulkLoad releases the +// gate with a defer, so a panic anywhere under it (a seal DDL, an FTS merge, a +// test hook) unwinds with the gate free and every later writer can still get +// in; the owed work below is then skipped along with the rest of the caller. +type bulkWindowExit struct { + // durabilityCheckpoint: a standalone window closed, so the frames it + // committed at synchronous=OFF are synced by one TRUNCATE on a NORMAL + // connection (checkpointBulkWAL). + durabilityCheckpoint bool + // residue is what a coordinated window's final PASSIVE left in the WAL; + // residueBoundary names that boundary for the residue gate, and is empty + // when no such PASSIVE ran. + residue walCheckpointResult + residueBoundary string + // repairDenseIndexes: the window closed with its seal still owed, so it + // recorded missing dense indexes and the maintenance lane owes a repair. + repairDenseIndexes bool +} + +// finishBulkWindowExit runs what exit owes after writeMu is released, in the +// order the exits have always run it. The durability checkpoint must run on a +// NORMAL connection, and it must not wait for a second writer slot while the +// bulk connection is still pinned, so neither step may move under the gate. +func (s *Store) finishBulkWindowExit(exit bulkWindowExit) error { + if exit.residueBoundary != "" { + s.scheduleWALDrainAboveLine(exit.residue, exit.residueBoundary) + } + var err error + if exit.durabilityCheckpoint { + err = s.checkpointBulkWAL() + } + if exit.repairDenseIndexes { + s.scheduleDenseIndexRepair() + } + return err +} + // FlushBulk exits the bulk-load fast path: it rebuilds every dropped index, // restores synchronous + cache_size + wal_autocheckpoint, releases the pinned writer and write gate, // then performs one bounded TRUNCATE checkpoint. The ordering matters: the // durability checkpoint must run on a NORMAL connection, and it must not wait // for a second writer slot while the bulk connection is still pinned. func (s *Store) FlushBulk() error { + exit, err := s.flushBulkLocked() + return errors.Join(err, s.finishBulkWindowExit(exit)) +} + +// flushBulkLocked is FlushBulk's write-gate half. A seal that panics leaves the +// pinned writer and its deferred indexes in place, so the next FlushBulk +// retries the same window. A seal that returns an error closes the window +// anyway: the writer is restored and released, and the indexes it could not +// rebuild stay recorded as missing for the maintenance lane's repair. +func (s *Store) flushBulkLocked() (bulkWindowExit, error) { s.writeMu.Lock() + defer s.writeMu.Unlock() if s.coordinatedBulkLoad { // The outer coordinated window owns index sealing and its final // durability checkpoint. WAL pressure inside that window is already @@ -938,8 +988,7 @@ func (s *Store) FlushBulk() error { // repository boundary can coincide with an unrelated read snapshot and // spend the full passive-checkpoint timeout without advancing the WAL; // repeating that timeout once per repository only stalls the cold path. - s.writeMu.Unlock() - return nil + return bulkWindowExit{}, nil } if s.generationBulkLoad != 0 { // A generation-scoped window belongs to whoever opened it, exactly as @@ -959,31 +1008,47 @@ func (s *Store) FlushBulk() error { // index (bulkIndexesDeferred is false for it) and its WAL is already // bounded inside the payload by noteBulkRowsLocked's row intervals, so // leaving it open costs the pass nothing it would otherwise have paid. - s.writeMu.Unlock() - return nil + return bulkWindowExit{}, nil } hadBulk := s.bulkConn != nil sealErr := s.sealBulkIndexesLocked("flush") + repair := s.bulkIndexesDeferred closeErr := s.closeBulkConnectionLocked() s.jsonbIngestBuffers.release() - s.writeMu.Unlock() - if !hadBulk { - return errors.Join(sealErr, closeErr) - } - return errors.Join(sealErr, closeErr, s.checkpointBulkWAL()) + return bulkWindowExit{durabilityCheckpoint: hadBulk, repairDenseIndexes: repair}, errors.Join(sealErr, closeErr) } // EndCoordinatedBulkLoad closes an outer multi-repository cold-load window. // It is idempotent and safe to defer: durability/cache restoration and index // rebuild happen even when one repository failed or panicked. func (s *Store) EndCoordinatedBulkLoad() error { + exit, err := s.endCoordinatedBulkLoadLocked() + return errors.Join(err, s.finishBulkWindowExit(exit)) +} + +// endCoordinatedBulkLoadLocked is EndCoordinatedBulkLoad's write-gate half. +func (s *Store) endCoordinatedBulkLoadLocked() (bulkWindowExit, error) { s.writeMu.Lock() + defer s.writeMu.Unlock() if !s.coordinatedBulkLoad { - s.writeMu.Unlock() - return nil + return bulkWindowExit{}, nil } s.coordinatedBulkLoad = false s.syncBulkWindowLocked() + // The flag is cleared above, before the FTS merges and the seal, so a panic + // in either would otherwise leave the pinned writer live behind a window + // nobody owns: the owner's next End would find no coordinated load and + // return nil, and the store would keep synchronous=OFF until Close. Put the + // window back the way the seal-error branch below keeps it, retryable by + // the same owner. Registered after the early return so it never claims a + // standalone window, and inert once closeBulkConnectionLocked has released + // the connection or the seal-error branch has restored the flag itself. + defer func() { + if s.bulkConn != nil && !s.coordinatedBulkLoad { + s.coordinatedBulkLoad = true + s.syncBulkWindowLocked() + } + }() if s.deferredFTSOptimize { // A full FTS5 optimize is unbounded and previously sat directly on the // cold-start critical path. One bounded merge keeps segment growth in @@ -1008,8 +1073,7 @@ func (s *Store) EndCoordinatedBulkLoad() error { // sealBulkIndexesLocked intentionally requires the pinned bulk writer. s.coordinatedBulkLoad = true s.syncBulkWindowLocked() - s.writeMu.Unlock() - return sealErr + return bulkWindowExit{}, sealErr } // Give the accumulated WAL one longer, still-bounded PASSIVE drain before // planner statistics traverse the large graph indexes. This runs on the @@ -1044,7 +1108,7 @@ func (s *Store) EndCoordinatedBulkLoad() error { // waiting for readers. Committed WAL frames are durable either way; later // passive checkpoints plus SQLite's next WAL reset/journal_size_limit reclaim // the file after the reader leaves. - var residue walCheckpointResult + var exit bulkWindowExit if hadBulk { ctx, cancel := context.WithTimeout(context.Background(), bulkFinalCheckpointTimeout) started := time.Now() @@ -1066,26 +1130,40 @@ func (s *Store) EndCoordinatedBulkLoad() error { // readers allowed. Hand that number to the residue gate rather than // treating "we tried once" as a drained log — the follow-up TRUNCATE // runs on the lane, after this window's readers are gone. - residue = result + exit.residue = result } + exit.residueBoundary = "cold_load_finalize" s.jsonbIngestBuffers.release() - s.writeMu.Unlock() - s.scheduleWALDrainAboveLine(residue, "cold_load_finalize") - return errors.Join(sealErr, statsErr, closeErr) + return exit, errors.Join(sealErr, statsErr, closeErr) } // AbortCoordinatedBulkLoad gives up retryable finalization and restores the -// writer connection unconditionally. It is for terminal owners only: dense -// indexes whose rebuild failed may remain absent until normal schema repair or -// restart, but the store no longer holds synchronous=OFF, disabled automatic -// checkpoints, or a pinned writer indefinitely. +// writer connection unconditionally, unless a generation window holds it. It +// is for terminal owners only: dense indexes whose rebuild failed stay absent, +// recorded as missing, until the maintenance lane's repair or a restart +// rebuilds them, but the store no longer holds synchronous=OFF, disabled +// automatic checkpoints, or a pinned writer indefinitely. func (s *Store) AbortCoordinatedBulkLoad() error { + exit, err := s.abortCoordinatedBulkLoadLocked() + return errors.Join(err, s.finishBulkWindowExit(exit)) +} + +func (s *Store) abortCoordinatedBulkLoadLocked() (bulkWindowExit, error) { s.writeMu.Lock() defer s.writeMu.Unlock() + if s.generationBulkLoad != 0 { + // Never a coordinated window (the two refuse each other), and not this + // caller's to close. An owner that aborts after an end which failed + // once the window was already closed can find the writer pinned by a + // payload generation opened since; closing that connection would leave + // its checkpoint lease bound and its EndGenerationBulkLoad inert. + return bulkWindowExit{}, nil + } s.coordinatedBulkLoad = false + repair := s.bulkIndexesDeferred err := s.closeBulkConnectionLocked() s.jsonbIngestBuffers.release() - return err + return bulkWindowExit{repairDenseIndexes: repair}, err } // noteBulkRowsLocked advances independent index-seal and WAL-checkpoint budgets @@ -1206,7 +1284,10 @@ func (s *Store) sealBulkIndexesLocked(reason string) error { _ = s.checkpointBulkWALPassiveLocked("index_seal_before") for _, idx := range bulkDroppableIndexes { indexStarted := time.Now() - _, err := conn.ExecContext(ctx, idx.ddl) + err := s.runBulkSealHook(idx.name) + if err == nil { + _, err = conn.ExecContext(ctx, idx.ddl) + } s.emitBulkFinalizeEvent(bulkFinalizeEvent{Stage: "index", Name: idx.name, Elapsed: time.Since(indexStarted), Err: err}) if err != nil { sealErr = errors.Join(sealErr, fmt.Errorf("store_sqlite: rebuild index %s: %w", idx.name, err)) @@ -1229,9 +1310,29 @@ func (s *Store) sealBulkIndexesLocked(reason string) error { return sealErr } s.bulkIndexesDeferred = false + // Every dense index is back, including any an earlier window left missing. + s.setDenseIndexesMissingLocked(nil) return nil } +// SetBulkSealHookForTest installs a hook this store's dense-index seal calls +// before each index's DDL, with the index name. A returned error stands in for +// that DDL failing (the DDL is skipped and the seal stays retryable, as on a +// real failure); a panic unwinds the seal. The hook runs under the write gate +// and must not call back into the store. restore puts the previous hook back. +func (s *Store) SetBulkSealHookForTest(hook func(index string) error) (restore func()) { + previous := s.bulkSealHook.Swap(&hook) + return func() { s.bulkSealHook.Store(previous) } +} + +func (s *Store) runBulkSealHook(index string) error { + hook := s.bulkSealHook.Load() + if hook == nil || *hook == nil { + return nil + } + return (*hook)(index) +} + // closeBulkConnectionLocked restores connection-local pragmas and releases the // pinned writer. Dense-index rebuilding deliberately lives in // sealBulkIndexesLocked so an early seal cannot end the outer window. @@ -1250,6 +1351,14 @@ func (s *Store) closeBulkConnectionLocked() error { _, syncErr := conn.ExecContext(ctx, fmt.Sprintf("PRAGMA synchronous = %d", s.bulkPrevSync)) _, cacheErr := conn.ExecContext(ctx, fmt.Sprintf("PRAGMA cache_size = %d", s.bulkPrevCacheSize)) _, autoCheckpointErr := conn.ExecContext(ctx, fmt.Sprintf("PRAGMA wal_autocheckpoint = %d", s.bulkPrevAutoCheckpoint)) + if s.bulkIndexesDeferred { + // The window is closing with its seal still owed: a failed seal at + // FlushBulk or Close, or an abort that never sealed. The indexes it + // dropped stay absent, so say which ones before the connection that + // can still read the schema goes. Asked after the PRAGMA restores so + // it can never delay durability coming back. + s.noteDenseIndexesMissingLocked(ctx, conn) + } closeErr := conn.Close() s.bulkIndexesDeferred = false s.bulkDeferredNodeRows = 0 @@ -1268,6 +1377,77 @@ func (s *Store) closeBulkConnectionLocked() error { ) } +// noteDenseIndexesMissingLocked records which bulkDroppableIndexes are absent +// from the schema as a cold window closes unsealed. One schema read on the +// pinned connection gives the exact set even after a partial seal; when that +// read fails every droppable index is listed instead, which only costs the +// repair a few CREATE INDEX IF NOT EXISTS that find the index already there. +// Only a cold window defers indexes (a generation window opens with +// bulkIndexesDeferred false), so this never runs for one. +func (s *Store) noteDenseIndexesMissingLocked(ctx context.Context, conn *sql.Conn) { + present, err := schemaIndexNames(ctx, conn) + var missing []bulkDroppableIndex + for _, idx := range bulkDroppableIndexes { + if err != nil || !present[idx.name] { + missing = append(missing, idx) + } + } + s.setDenseIndexesMissingLocked(missing) + if len(missing) > 0 { + log.Printf("store_sqlite: dense indexes missing after bulk window close indexes=%s", strings.Join(s.denseIndexesMissingNames(), ",")) + } +} + +// setDenseIndexesMissingLocked replaces the missing-index record. The caller +// holds writeMu; the slice is never modified after it is published. +func (s *Store) setDenseIndexesMissingLocked(missing []bulkDroppableIndex) { + if len(missing) == 0 { + s.denseIndexesMissing.Store(nil) + return + } + s.denseIndexesMissing.Store(&missing) +} + +// denseIndexesMissingRecord returns a copy of the missing-index record, or nil. +func (s *storeCore) denseIndexesMissingRecord() []bulkDroppableIndex { + missing := s.denseIndexesMissing.Load() + if missing == nil { + return nil + } + return append([]bulkDroppableIndex(nil), (*missing)...) +} + +// denseIndexesMissingNames returns the names in the missing-index record, or +// nil. +func (s *storeCore) denseIndexesMissingNames() []string { + missing := s.denseIndexesMissing.Load() + if missing == nil { + return nil + } + names := make([]string, 0, len(*missing)) + for _, idx := range *missing { + names = append(names, idx.name) + } + return names +} + +func schemaIndexNames(ctx context.Context, conn *sql.Conn) (map[string]bool, error) { + rows, err := conn.QueryContext(ctx, `SELECT name FROM sqlite_schema WHERE type = 'index'`) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + present := make(map[string]bool) + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + return nil, err + } + present[name] = true + } + return present, rows.Err() +} + func wrapBulkRestoreError(pragma string, err error) error { if err == nil { return nil @@ -1381,14 +1561,48 @@ func shouldBackoffBulkRowCheckpoint(err error) bool { // rows has been indexed before, whichever view wrote them, so it is not the // cold store this fast path is for. func coldGraphStoreEmpty(ctx context.Context, conn *sql.Conn) bool { - var empty int + return coldGraphStoreRefusal(ctx, conn) == "" +} + +// One bounded probe retains the cold-store proof and identifies its first +// failed predicate without counting or hydrating any graph rows. +func coldGraphStoreRefusal(ctx context.Context, conn *sql.Conn) string { + var reason string err := conn.QueryRowContext(ctx, ` -SELECT NOT EXISTS(SELECT 1 FROM nodes WHERE view_gen = ?) - AND NOT EXISTS(SELECT 1 FROM edges WHERE view_gen = ?) - AND NOT EXISTS(SELECT 1 FROM file_mtimes) - AND NOT EXISTS(SELECT 1 FROM repo_index_state)`, - baseViewGeneration, baseViewGeneration).Scan(&empty) - return err == nil && empty == 1 +SELECT CASE + WHEN EXISTS(SELECT 1 FROM nodes WHERE view_gen = ?) THEN 'nodes' + WHEN EXISTS(SELECT 1 FROM edges WHERE view_gen = ?) THEN 'edges' + WHEN EXISTS(SELECT 1 FROM file_mtimes) THEN 'file_mtimes' + WHEN EXISTS(SELECT 1 FROM repo_index_state) THEN 'repo_index_state' + ELSE '' END`, baseViewGeneration, baseViewGeneration).Scan(&reason) + if err != nil { + return "cold_probe_error" + } + return reason +} + +// WriterPragmaForTest reads an integer PRAGMA off the store's writer +// connection: the pinned bulk connection while a window holds one, the single +// writer-pool connection otherwise. PRAGMAs such as synchronous are +// connection-local, so a read-pool connection cannot show what the writer runs +// with. It takes the write gate under ctx, so a deadline also proves the gate +// is free. name must be a bare PRAGMA name. +func (s *Store) WriterPragmaForTest(ctx context.Context, name string) (int64, error) { + for _, r := range name { + if (r < 'a' || r > 'z') && r != '_' { + return 0, fmt.Errorf("store_sqlite: not a pragma name: %q", name) + } + } + if err := s.writeMu.LockContext(ctx); err != nil { + return 0, err + } + defer s.writeMu.Unlock() + conn, release, err := s.activeWriteConnLocked(ctx) + if err != nil { + return 0, err + } + defer release() + return pragmaInt(ctx, conn, name) } // pragmaInt reads a single-integer PRAGMA (synchronous, cache_size, wal_autocheckpoint) off the diff --git a/internal/graph/store_sqlite/bulk_load_generation_test.go b/internal/graph/store_sqlite/bulk_load_generation_test.go index 67ae10223..0f3597b08 100644 --- a/internal/graph/store_sqlite/bulk_load_generation_test.go +++ b/internal/graph/store_sqlite/bulk_load_generation_test.go @@ -210,6 +210,35 @@ func TestGenerationBulkLoadRefusesWhatItCannotProveEmpty(t *testing.T) { }) } +// TestAbortCoordinatedBulkLoadLeavesAGenerationWindowToItsOwner: an owner that +// aborts after its coordinated end failed may find the writer pinned by a +// payload generation opened since. The abort must leave that window, and the +// checkpoint lease only its own end releases, to the generation's owner. +func TestAbortCoordinatedBulkLoadLeavesAGenerationWindowToItsOwner(t *testing.T) { + store, _ := openTempStore(t) + engaged, err := store.BeginGenerationBulkLoad(3) + if err != nil || !engaged { + t.Fatalf("BeginGenerationBulkLoad = (%v, %v), want engaged", engaged, err) + } + if err := store.AbortCoordinatedBulkLoad(); err != nil { + t.Fatalf("AbortCoordinatedBulkLoad: %v", err) + } + if id, open := store.InGenerationBulkLoad(); id != 3 || !open { + t.Fatalf("generation window after abort = (%d, %v), want (3, true)", id, open) + } + if err := store.EndGenerationBulkLoad(); err != nil { + t.Fatalf("EndGenerationBulkLoad: %v", err) + } + // The owner's end released the lease: the next generation can open. + engaged, err = store.BeginGenerationBulkLoad(4) + if err != nil || !engaged { + t.Fatalf("next BeginGenerationBulkLoad = (%v, %v), want engaged", engaged, err) + } + if err := store.EndGenerationBulkLoad(); err != nil { + t.Fatalf("EndGenerationBulkLoad: %v", err) + } +} + func TestGenerationBulkLoadLeavesBaseReadersAndIndexesAlone(t *testing.T) { store, _ := openTempStore(t) baseNodes, baseEdges := bulkFixture(256, 512) diff --git a/internal/graph/store_sqlite/bulk_load_test.go b/internal/graph/store_sqlite/bulk_load_test.go index 528c7b659..dd846d7c1 100644 --- a/internal/graph/store_sqlite/bulk_load_test.go +++ b/internal/graph/store_sqlite/bulk_load_test.go @@ -835,6 +835,83 @@ func TestUncoordinatedBulkSealFailureRetriesAtFlushBoundary(t *testing.T) { integrityOK(t, s.db) } +// TestFlushBulkSealPanicDoesNotDeadlock: a seal that panics with a non-error +// value must unwind with the write gate released and the window still +// retryable, so the owner's next exit call seals, restores and closes it. +func TestFlushBulkSealPanicDoesNotDeadlock(t *testing.T) { + for _, tc := range []struct { + name string + begin func(*Store) bool + end func(*Store) error + }{ + {"standalone", func(s *Store) bool { s.BeginBulkLoad(); return s.bulkConn != nil }, (*Store).FlushBulk}, + {"coordinated", (*Store).BeginCoordinatedBulkLoad, (*Store).EndCoordinatedBulkLoad}, + } { + t.Run(tc.name, func(t *testing.T) { + s, _ := openTempStore(t) + syncBefore := pragmaIntDB(t, s.writerDB, "synchronous") + if !tc.begin(s) { + t.Fatal("bulk fast path did not engage") + } + s.AddNode(&graph.Node{ID: "repo/a.go::A", Kind: graph.KindFunction, Name: "A"}) + calls := 0 + defer s.SetBulkSealHookForTest(func(string) error { + calls++ + if calls == 1 { + panic("injected seal panic") + } + return nil + })() + + recovered := func() (r any) { + defer func() { r = recover() }() + _ = tc.end(s) + return nil + }() + if recovered != "injected seal panic" { + t.Fatalf("recovered %v, want the injected seal panic", recovered) + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := s.writeMu.LockContext(ctx); err != nil { + // Hand the leaked hold back so cleanup's Close cannot hang too. + s.writeMu.Unlock() + t.Fatalf("write gate still held after the seal panic: %v", err) + } + retryable := s.bulkConn != nil && s.bulkIndexesDeferred && s.bulkWindowOpen.Load() && + s.coordinatedBulkLoad == (tc.name == "coordinated") + s.writeMu.Unlock() + if !retryable { + t.Fatal("seal panic lost the window's retryable state") + } + + if err := tc.end(s); err != nil { + t.Fatalf("exit retry after the seal panic: %v", err) + } + if calls <= 1 { + t.Fatalf("seal hook calls = %d, want the retry to seal again", calls) + } + if s.bulkConn != nil || s.bulkIndexesDeferred || s.coordinatedBulkLoad || s.bulkWindowOpen.Load() { + t.Fatalf("retry left bulk state: conn=%v deferred=%v coordinated=%v open=%v", + s.bulkConn != nil, s.bulkIndexesDeferred, s.coordinatedBulkLoad, s.bulkWindowOpen.Load()) + } + if got := pragmaIntDB(t, s.writerDB, "synchronous"); got != syncBefore { + t.Fatalf("writer synchronous = %d, want restored %d", got, syncBefore) + } + present := indexNames(t, s.db) + for _, idx := range bulkDroppableIndexes { + if !present[idx.name] { + t.Fatalf("index %s missing after the retried seal", idx.name) + } + } + if got := s.NodeCount(); got != 1 { + t.Fatalf("node count = %d, want 1", got) + } + integrityOK(t, s.db) + }) + } +} + func TestCoordinatedBulkLoadRoutesSingleRowSidecarsOnPinnedConnection(t *testing.T) { s, _ := openTempStore(t) s.db.SetMaxOpenConns(1) diff --git a/internal/graph/store_sqlite/dense_index_repair.go b/internal/graph/store_sqlite/dense_index_repair.go new file mode 100644 index 000000000..a8e5a0764 --- /dev/null +++ b/internal/graph/store_sqlite/dense_index_repair.go @@ -0,0 +1,230 @@ +package store_sqlite + +import ( + "context" + "fmt" + "log" + "math" + "path/filepath" + "strings" + "time" + + "github.com/zzet/gortex/internal/platform" +) + +// Dense-index repair: the maintenance-lane job that rebuilds the dense indexes +// a cold bulk window closed without (denseIndexesMissing). The usual cause is +// the volume filling up during the seal, so the job is built around not +// making that worse: +// +// - it never runs inline. The exit that recorded the missing indexes posts +// one request after it has released the write gate, and the lane worker +// runs it as a quiescing whole-database job, like VACUUM. +// - it does not try while the disk is still full. An attempt is skipped +// while the free space on the store's volume is below the store's used +// bytes (with a floor of denseIndexRepairMinFreeBytes). The dense indexes +// are a fraction of the used pages, and a CREATE INDEX needs about twice +// the index size at its peak (the sort plus the WAL frames), so a volume +// with the store's own size free can always take the rebuild while a full +// one is never asked to. +// - it backs off. A skipped or failed attempt re-posts itself after +// denseIndexRepairBaseDelay, doubling up to denseIndexRepairMaxDelay, +// so a volume that stays full costs one statfs per backoff step and one +// log line, not a retry loop. +// - it holds the write gate one index at a time. The DDL is exactly what +// createGraphCoreIndexes runs (bulkDroppableIndexes is the single source +// for both, copied into the record by the window that closed without it, +// so the lane never reads that list), but each CREATE INDEX takes and +// releases writeMu on its own, so a writer waits behind at most one index +// build rather than the set. +// +// It is deliberately not part of the lane's settle point (waitMaintenanceIdle): +// a long rebuild must not hold AwaitMaintenanceIdle callers, which sit on +// publication paths. A completed repair asks the lane for a statistics pass, +// because the rebuilt indexes have no sqlite_stat1 rows yet. + +// Vars rather than constants only so the in-package cases can shorten them. +var ( + denseIndexRepairBaseDelay = 30 * time.Second + denseIndexRepairMaxDelay = 15 * time.Minute + denseIndexRepairMinFreeBytes = int64(64 << 20) +) + +// scheduleDenseIndexRepair posts one immediate repair request, resetting any +// backoff an earlier record had built up. The caller has just recorded the +// missing set and must not hold writeMu; like scheduleWALDrain it takes only +// maintenanceSched and never blocks. +func (s *Store) scheduleDenseIndexRepair() { + if s.coreless() { + return + } + s.maintenanceSched.Lock() + defer s.maintenanceSched.Unlock() + if s.maintenanceClosed || s.maintenanceSignal == nil { + return + } + s.stopDenseIndexRepairTimerLocked() + s.denseRepairDelay = 0 + if s.denseRepairOwed { + return + } + s.denseRepairOwed = true + select { + case s.maintenanceSignal <- struct{}{}: + default: + } +} + +// takeDenseIndexRepair claims an owed repair for the lane worker. +func (s *Store) takeDenseIndexRepair() bool { + s.maintenanceSched.Lock() + defer s.maintenanceSched.Unlock() + if !s.denseRepairOwed || s.maintenanceClosed { + return false + } + s.denseRepairOwed = false + return true +} + +// retryDenseIndexRepairLater re-posts the repair once the current backoff +// elapses and doubles the backoff for the attempt after it. It returns the +// delay it armed, or 0 when the lane is closed. +func (s *Store) retryDenseIndexRepairLater() time.Duration { + s.maintenanceSched.Lock() + defer s.maintenanceSched.Unlock() + if s.maintenanceClosed || s.maintenanceSignal == nil { + return 0 + } + delay := s.denseRepairDelay + if delay <= 0 { + delay = denseIndexRepairBaseDelay + } + s.denseRepairDelay = min(2*delay, denseIndexRepairMaxDelay) + s.stopDenseIndexRepairTimerLocked() + signal := s.maintenanceSignal + s.denseRepairTimer = time.AfterFunc(delay, func() { + s.maintenanceSched.Lock() + defer s.maintenanceSched.Unlock() + if s.maintenanceClosed || s.maintenanceSignal != signal || s.denseRepairOwed { + return + } + s.denseRepairOwed = true + select { + case signal <- struct{}{}: + default: + } + }) + return delay +} + +// stopDenseIndexRepairTimerLocked cancels a pending retry. The caller holds +// maintenanceSched. +func (s *Store) stopDenseIndexRepairTimerLocked() { + if s.denseRepairTimer != nil { + s.denseRepairTimer.Stop() + s.denseRepairTimer = nil + } +} + +// runDenseIndexRepair is one attempt, run by the lane worker. Nothing it hits +// is an error to anyone: the record stays, the attempt is logged, and the +// backoff decides when the next one comes. +func (s *Store) runDenseIndexRepair(ctx context.Context) { + missing := s.denseIndexesMissingNames() + if len(missing) == 0 { + // A later seal already rebuilt them. + s.maintenanceSched.Lock() + s.stopDenseIndexRepairTimerLocked() + s.denseRepairDelay = 0 + s.maintenanceSched.Unlock() + return + } + avail, need, ok := s.denseIndexRepairSpace() + if !ok { + s.denseIndexRepairSkips.Add(1) + delay := s.retryDenseIndexRepairLater() + log.Printf("store_sqlite: dense index repair skipped: low free space avail_bytes=%d need_bytes=%d indexes=%s retry_in=%s", avail, need, strings.Join(missing, ","), delay) + return + } + started := time.Now() + err := s.runMaintenance(ctx, maintenanceDenseIndexRepair, true, func(ctx context.Context) error { + s.denseIndexRepairAttempts.Add(1) + return s.repairDenseIndexes(ctx) + }) + if err != nil { + if ctx.Err() != nil { + return + } + delay := s.retryDenseIndexRepairLater() + log.Printf("store_sqlite: dense index repair deferred indexes=%s retry_in=%s error=%q", strings.Join(s.denseIndexesMissingNames(), ","), delay, err) + return + } + s.maintenanceSched.Lock() + s.stopDenseIndexRepairTimerLocked() + s.denseRepairDelay = 0 + s.maintenanceSched.Unlock() + s.denseIndexRepairs.Add(1) + log.Printf("store_sqlite: dense index repair complete indexes=%s elapsed=%s", strings.Join(missing, ","), time.Since(started).Round(time.Millisecond)) + // The rebuilt indexes carry no statistics yet; the lane's next pass + // notices that and analyzes them. + s.schedulePublishMaintenance() +} + +// denseIndexRepairSpace reports the free space on the store's volume, the +// amount an attempt needs, and whether the attempt may run. An unreadable +// free-space figure is a skip: the gate exists for the full volume, which is +// exactly where a probe may fail too. So is an unreadable store size +// (CompactStats reads 0, 0 when its PRAGMAs fail), which would otherwise +// shrink the need to the floor whatever the store's real size. +func (s *Store) denseIndexRepairSpace() (avail, need int64, ok bool) { + free, total := s.CompactStats() + if total <= 0 { + return -1, -1, false + } + need = max(total-free, denseIndexRepairMinFreeBytes) + probe := platform.DiskAvailBytes + if override := s.denseRepairDiskAvail.Load(); override != nil && *override != nil { + probe = *override + } + bytes, err := probe(filepath.Dir(s.dbPath)) + if err != nil { + return -1, need, false + } + avail = int64(min(bytes, uint64(math.MaxInt64))) + return avail, need, avail >= need +} + +// repairDenseIndexes rebuilds every index in the missing record, one write-gate +// hold per index. It refuses, like VACUUM, while a bulk window pins the writer: +// that window drops and seals the same indexes itself. +func (s *Store) repairDenseIndexes(ctx context.Context) error { + for _, idx := range s.denseIndexesMissingRecord() { + if err := s.repairDenseIndex(ctx, idx); err != nil { + return err + } + } + return nil +} + +func (s *Store) repairDenseIndex(ctx context.Context, idx bulkDroppableIndex) error { + if err := s.writeMu.LockContext(ctx); err != nil { + return fmt.Errorf("%w: %s: write gate: %w", ErrMaintenanceBusy, maintenanceDenseIndexRepair, err) + } + defer s.writeMu.Unlock() + if s.bulkConn != nil { + return fmt.Errorf("%w: %s: bulk writer pinned", ErrMaintenanceBusy, maintenanceDenseIndexRepair) + } + if _, err := s.writerDB.ExecContext(ctx, idx.ddl); err != nil { + return fmt.Errorf("store_sqlite: repair index %s: %w", idx.name, err) + } + // Under the hold that built it, so whichever window recorded the name, the + // index it names is present now. + var remaining []bulkDroppableIndex + for _, missing := range s.denseIndexesMissingRecord() { + if missing.name != idx.name { + remaining = append(remaining, missing) + } + } + s.setDenseIndexesMissingLocked(remaining) + return nil +} diff --git a/internal/graph/store_sqlite/dense_index_repair_test.go b/internal/graph/store_sqlite/dense_index_repair_test.go new file mode 100644 index 000000000..b65bccec6 --- /dev/null +++ b/internal/graph/store_sqlite/dense_index_repair_test.go @@ -0,0 +1,306 @@ +package store_sqlite + +import ( + "context" + "database/sql" + "errors" + "path/filepath" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/zzet/gortex/internal/graph" +) + +// shortenDenseIndexRepairBackoff makes a skipped repair retry within +// milliseconds for the rest of the test. +func shortenDenseIndexRepairBackoff(t *testing.T) { + t.Helper() + base, maxDelay := denseIndexRepairBaseDelay, denseIndexRepairMaxDelay + denseIndexRepairBaseDelay, denseIndexRepairMaxDelay = 5*time.Millisecond, 20*time.Millisecond + t.Cleanup(func() { denseIndexRepairBaseDelay, denseIndexRepairMaxDelay = base, maxDelay }) +} + +// TestFailedSealReportsDenseIndexesMissing: a cold window that closes without +// rebuilding its dense indexes must still restore durability and close, must +// say which indexes it left absent instead of letting the planner probe read +// their absence as a clean store, must not retry the rebuild while the volume +// is full, and must rebuild them in one maintenance attempt once it is not, +// followed by the statistics pass the rebuilt indexes need. +func TestFailedSealReportsDenseIndexesMissing(t *testing.T) { + shortenDenseIndexRepairBackoff(t) + allDroppable := make([]string, 0, len(bulkDroppableIndexes)) + for _, idx := range bulkDroppableIndexes { + allDroppable = append(allDroppable, idx.name) + } + for _, tc := range []struct { + name string + // fail closes the window the case's way and returns the indexes it + // expects to be left missing, in declaration order. + fail func(t *testing.T, s *Store) []string + }{ + {"flush seal error", func(t *testing.T, s *Store) []string { + s.BeginBulkLoad() + if s.bulkConn == nil { + t.Fatal("bulk fast path did not engage") + } + s.AddNode(&graph.Node{ID: "repo/a.go::A", Kind: graph.KindFunction, Name: "A"}) + injected := errors.New("injected seal failure") + defer s.SetBulkSealHookForTest(func(index string) error { + if index == "nodes_by_kind" || index == "edges_by_kind" { + return injected + } + return nil + })() + err := s.FlushBulk() + if !errors.Is(err, injected) { + t.Fatalf("FlushBulk error = %v, want the injected seal failure", err) + } + for _, name := range []string{"nodes_by_kind", "edges_by_kind"} { + if !strings.Contains(err.Error(), name) { + t.Fatalf("FlushBulk error %q does not name %s", err, name) + } + } + return []string{"nodes_by_kind", "edges_by_kind"} + }}, + {"abort before seal", func(t *testing.T, s *Store) []string { + if !s.BeginCoordinatedBulkLoad() { + t.Fatal("coordinated fast path did not engage") + } + s.AddNode(&graph.Node{ID: "repo/a.go::A", Kind: graph.KindFunction, Name: "A"}) + if err := s.AbortCoordinatedBulkLoad(); err != nil { + t.Fatalf("AbortCoordinatedBulkLoad: %v", err) + } + return allDroppable + }}, + } { + t.Run(tc.name, func(t *testing.T) { + s, _ := openTempStore(t) + var full atomic.Bool + full.Store(true) + diskAvail := func(string) (uint64, error) { + if full.Load() { + return 0, nil + } + return 1 << 40, nil + } + s.denseRepairDiskAvail.Store(&diskAvail) + syncBefore := pragmaIntDB(t, s.writerDB, "synchronous") + want := tc.fail(t, s) + + if s.bulkConn != nil || s.coordinatedBulkLoad || s.bulkWindowOpen.Load() { + t.Fatal("window left open after closing unsealed") + } + if got := pragmaIntDB(t, s.writerDB, "synchronous"); got != syncBefore { + t.Fatalf("writer synchronous = %d, want restored %d", got, syncBefore) + } + health, err := s.PlannerStatsHealth(context.Background()) + if err != nil { + t.Fatalf("PlannerStatsHealth: %v", err) + } + if health.Reason != plannerStatsDenseIndexesMissingReason || health.Stale { + t.Fatalf("health reason=%q stale=%v, want %q and not stale", health.Reason, health.Stale, plannerStatsDenseIndexesMissingReason) + } + if !reflect.DeepEqual(health.DenseIndexesMissing, want) { + t.Fatalf("DenseIndexesMissing = %v, want %v", health.DenseIndexesMissing, want) + } + present := indexNames(t, s.db) + for _, name := range allDroppable { + missing := false + for _, w := range want { + missing = missing || w == name + } + if present[name] == missing { + t.Fatalf("index %s present=%v, want present=%v", name, present[name], !missing) + } + } + + // While the volume reads as full the lane keeps skipping on its + // backoff and never reaches a CREATE INDEX. + waitUntil(t, "a repair attempt skipped for free space", func() bool { return s.denseIndexRepairSkips.Load() >= 2 }) + if got := s.denseIndexRepairAttempts.Load(); got != 0 { + t.Fatalf("repair attempts while full = %d, want 0", got) + } + if health, err := s.PlannerStatsHealth(context.Background()); err != nil || health.Reason != plannerStatsDenseIndexesMissingReason { + t.Fatalf("health while full = %q, %v; want %q", health.Reason, err, plannerStatsDenseIndexesMissingReason) + } + + // Once space is back, the next retry rebuilds every listed index. + full.Store(false) + waitUntil(t, "the dense index repair", func() bool { return s.denseIndexRepairs.Load() == 1 }) + if got := s.denseIndexRepairAttempts.Load(); got != 1 { + t.Fatalf("repair attempts = %d, want one", got) + } + health, err = s.PlannerStatsHealth(context.Background()) + if err != nil { + t.Fatalf("PlannerStatsHealth after repair: %v", err) + } + if health.Reason == plannerStatsDenseIndexesMissingReason || len(health.DenseIndexesMissing) != 0 { + t.Fatalf("after repair reason=%q missing=%v, want both cleared", health.Reason, health.DenseIndexesMissing) + } + present = indexNames(t, s.db) + for _, name := range allDroppable { + if !present[name] { + t.Fatalf("index %s missing after the repair", name) + } + } + // A rebuilt index has no sqlite_stat1 row, and once the repair is + // done nothing else asks for a statistics pass, so the repair must + // ask for one itself. nodes_by_kind holds the case's node. + waitUntil(t, "statistics for the rebuilt nodes_by_kind", func() bool { + return s.plannerStatsIndexesWithStats(context.Background())["nodes_by_kind"] + }) + integrityOK(t, s.db) + }) + } +} + +// TestDenseIndexRepairBuildsFromTheRecordedDDL: the repair runs the DDL the +// closing window recorded with each missing name, never a lookup in +// bulkDroppableIndexes made when the lane gets to it. In-package cases replace +// that list to force a seal failure and put it back while the lane may still be +// on its way to the repair; a lookup there reads the list unsynchronised with +// that restore, and finds no DDL for a name the record still holds. +func TestDenseIndexRepairBuildsFromTheRecordedDDL(t *testing.T) { + shortenDenseIndexRepairBackoff(t) + s, _ := openTempStore(t) + var full atomic.Bool + full.Store(true) + diskAvail := func(string) (uint64, error) { + if full.Load() { + return 0, nil + } + return 1 << 40, nil + } + s.denseRepairDiskAvail.Store(&diskAvail) + + originalIndexes := bulkDroppableIndexes + bulkDroppableIndexes = append(append([]bulkDroppableIndex(nil), originalIndexes...), bulkDroppableIndex{ + name: "forced_bad_index", + ddl: `CREATE INDEX forced_bad_index ON definitely_missing_table(id)`, + }) + s.BeginBulkLoad() + if s.bulkConn == nil { + bulkDroppableIndexes = originalIndexes + t.Fatal("bulk fast path did not engage") + } + s.AddNode(&graph.Node{ID: "repo/a.go::A", Kind: graph.KindFunction, Name: "A"}) + err := s.FlushBulk() + bulkDroppableIndexes = originalIndexes + if err == nil || !strings.Contains(err.Error(), "forced_bad_index") { + t.Fatalf("FlushBulk error = %v, want the forced index's DDL failure", err) + } + + // The list no longer names the index; the record does, with the DDL that + // cannot succeed. Every attempt must run it and fail, not report a repair. + full.Store(false) + waitUntil(t, "a finished repair attempt", func() bool { + return s.denseIndexRepairs.Load() > 0 || s.denseIndexRepairAttempts.Load() >= 2 + }) + if got := s.denseIndexRepairs.Load(); got != 0 { + t.Fatalf("repairs = %d for an index whose DDL cannot run, want 0", got) + } + if got := s.denseIndexesMissingNames(); !reflect.DeepEqual(got, []string{"forced_bad_index"}) { + t.Fatalf("missing record = %v, want [forced_bad_index]", got) + } +} + +// TestDenseIndexRepairSkipsWithoutTheStoreSize: an attempt is sized by the +// store's own used bytes, and CompactStats reads 0, 0 when it cannot read +// them. That must skip the attempt, as an unreadable free-space figure does, +// not shrink the need to the floor and let a large store rebuild on a volume +// that is nearly full. +func TestDenseIndexRepairSkipsWithoutTheStoreSize(t *testing.T) { + db, err := sql.Open("sqlite", ":memory:") + if err != nil { + t.Fatal(err) + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + s := &Store{storeCore: &storeCore{db: db, dbPath: filepath.Join(t.TempDir(), "store.sqlite")}} + diskAvail := func(string) (uint64, error) { return 1 << 40, nil } + s.denseRepairDiskAvail.Store(&diskAvail) + if _, total := s.CompactStats(); total != 0 { + t.Fatalf("CompactStats total = %d on a closed pool, want 0", total) + } + if avail, need, ok := s.denseIndexRepairSpace(); ok { + t.Fatalf("attempt allowed with the store size unreadable: avail=%d need=%d", avail, need) + } +} + +// TestDenseIndexRepairWaitsForAPinnedWriter: a repair that comes due while a +// window pins the writer defers, as VACUUM does, instead of building under +// that window, and builds on a later retry once the window has closed. A +// generation window is the one that can open on the populated store a failed +// seal leaves behind. +func TestDenseIndexRepairWaitsForAPinnedWriter(t *testing.T) { + shortenDenseIndexRepairBackoff(t) + s, _ := openTempStore(t) + var full atomic.Bool + full.Store(true) + diskAvail := func(string) (uint64, error) { + if full.Load() { + return 0, nil + } + return 1 << 40, nil + } + s.denseRepairDiskAvail.Store(&diskAvail) + s.BeginBulkLoad() + if s.bulkConn == nil { + t.Fatal("bulk fast path did not engage") + } + s.AddNode(&graph.Node{ID: "repo/a.go::A", Kind: graph.KindFunction, Name: "A"}) + injected := errors.New("injected seal failure") + restore := s.SetBulkSealHookForTest(func(index string) error { + if index == "nodes_by_kind" { + return injected + } + return nil + }) + err := s.FlushBulk() + restore() + if !errors.Is(err, injected) { + t.Fatalf("FlushBulk error = %v, want the injected seal failure", err) + } + engaged, err := s.BeginGenerationBulkLoad(3) + if err != nil || !engaged { + t.Fatalf("BeginGenerationBulkLoad = (%v, %v), want engaged", engaged, err) + } + + full.Store(false) + waitUntil(t, "two repair attempts under the pinned writer", func() bool { return s.denseIndexRepairAttempts.Load() >= 2 }) + if got := s.denseIndexRepairs.Load(); got != 0 { + t.Fatalf("repairs under a pinned writer = %d, want 0", got) + } + if got := s.denseIndexesMissingNames(); !reflect.DeepEqual(got, []string{"nodes_by_kind"}) { + t.Fatalf("missing record under a pinned writer = %v, want [nodes_by_kind]", got) + } + if indexNames(t, s.db)["nodes_by_kind"] { + t.Fatal("nodes_by_kind was built under the pinned writer") + } + + if err := s.EndGenerationBulkLoad(); err != nil { + t.Fatalf("EndGenerationBulkLoad: %v", err) + } + waitUntil(t, "the repair after the window closed", func() bool { return s.denseIndexRepairs.Load() == 1 }) + if !indexNames(t, s.db)["nodes_by_kind"] { + t.Fatal("nodes_by_kind missing after the repair") + } + if got := s.denseIndexesMissingNames(); len(got) != 0 { + t.Fatalf("missing record after the repair = %v, want empty", got) + } +} + +func waitUntil(t *testing.T, what string, done func() bool) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for !done() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(time.Millisecond) + } +} diff --git a/internal/graph/store_sqlite/first_index_drain.go b/internal/graph/store_sqlite/first_index_drain.go new file mode 100644 index 000000000..8bc764897 --- /dev/null +++ b/internal/graph/store_sqlite/first_index_drain.go @@ -0,0 +1,48 @@ +package store_sqlite + +import "github.com/zzet/gortex/internal/graph" + +// BulkLoadWindow snapshots the owner while the writer is between drain chunks. +func (s *Store) BulkLoadWindow() string { + s.writeMu.Lock() + defer s.writeMu.Unlock() + switch { + case s.generationBulkLoad != 0: + return "generation_noop" + case s.coordinatedBulkLoad: + return "coordinated_noop" + case s.bulkConn != nil: + return "cold" + default: + return "none" + } +} + +// RecordFirstIndexDrain retains one value per repository, shared by every +// generation handle and temporary indexer over this open store. +func (s *Store) RecordFirstIndexDrain(repo string, observation graph.FirstIndexDrain) { + s.firstIndexDrainsMu.Lock() + defer s.firstIndexDrainsMu.Unlock() + if s.firstIndexDrains == nil { + s.firstIndexDrains = make(map[string]graph.FirstIndexDrain) + } + s.firstIndexDrains[repo] = observation +} + +// LastFirstIndexDrains does no SQL or writer admission. Return a detached map +// so status and cached health payloads cannot mutate or race the live record. +func (s *Store) LastFirstIndexDrains() map[string]graph.FirstIndexDrain { + s.firstIndexDrainsMu.Lock() + defer s.firstIndexDrainsMu.Unlock() + if len(s.firstIndexDrains) == 0 { + return nil + } + out := make(map[string]graph.FirstIndexDrain, len(s.firstIndexDrains)) + for repo, observation := range s.firstIndexDrains { + out[repo] = observation + } + return out +} + +// DenseIndexesMissing reports the existing atomic repair record without SQL. +func (s *Store) DenseIndexesMissing() []string { return s.denseIndexesMissingNames() } diff --git a/internal/graph/store_sqlite/first_index_drain_test.go b/internal/graph/store_sqlite/first_index_drain_test.go new file mode 100644 index 000000000..76bfe4580 --- /dev/null +++ b/internal/graph/store_sqlite/first_index_drain_test.go @@ -0,0 +1,41 @@ +package store_sqlite + +import ( + "testing" + "time" + + "github.com/zzet/gortex/internal/graph" +) + +func TestFirstIndexDrainSnapshotAvoidsWriterAndCallerMutation(t *testing.T) { + s, _ := openTempStore(t) + s.RecordFirstIndexDrain("repo", graph.FirstIndexDrain{Window: "standalone", Nodes: 3}) + s.writeMu.Lock() + result := make(chan map[string]graph.FirstIndexDrain, 1) + go func() { result <- s.LastFirstIndexDrains() }() + var snapshot map[string]graph.FirstIndexDrain + select { + case snapshot = <-result: + s.writeMu.Unlock() + case <-time.After(5 * time.Second): + s.writeMu.Unlock() + t.Fatal("status snapshot queued behind writer") + } + delete(snapshot, "repo") + actual := s.LastFirstIndexDrains() + if actual["repo"].Nodes != 3 { + t.Fatalf("caller changed retained observation: %v", actual) + } + if err := s.RekeyRepoPrefix("repo", "renamed"); err != nil { + t.Fatal(err) + } + if got := s.LastFirstIndexDrains(); len(got) != 1 || got["renamed"].Nodes != 3 { + t.Fatalf("rekey lost or duplicated observation: %v", got) + } + if err := s.PurgeRepo("renamed"); err != nil { + t.Fatal(err) + } + if got := s.LastFirstIndexDrains(); len(got) != 0 { + t.Fatalf("purged repository retained observation: %v", got) + } +} diff --git a/internal/graph/store_sqlite/planner_stats_freshness.go b/internal/graph/store_sqlite/planner_stats_freshness.go index 86dbae87b..193aa2aef 100644 --- a/internal/graph/store_sqlite/planner_stats_freshness.go +++ b/internal/graph/store_sqlite/planner_stats_freshness.go @@ -150,6 +150,14 @@ const ( // finalize anyway. const plannerStatsBulkWindowReason = "bulk_window_active" +// plannerStatsDenseIndexesMissingReason is reported instead of a verdict while +// a cold bulk window that closed unsealed still owes dense indexes (listed in +// PlannerStatsFreshness.DenseIndexesMissing). Some of the critical indexes may +// be among them, and every rule would read an absent index as a clean store, +// so the probe names the real problem instead. No ANALYZE can fix it: the +// maintenance lane's repair rebuilds the indexes and the reason clears with it. +const plannerStatsDenseIndexesMissingReason = "dense_indexes_missing" + // Reasons a stale verdict can come back UNREFRESHED. None of them is a // failure, and none of them stamps the ledger or arms the anti-loop guard: the // verdict stands, the resume cursor keeps the indexes already finished, and the @@ -620,6 +628,13 @@ func (s *Store) plannerStatsHealth(ctx context.Context, probeReceivers bool) (gr health.Reason = plannerStatsBulkWindowReason return health, bases, nil } + // Same shape as the window above: not a judgeable state, Stale stays + // false and Known stays false, so nothing is seeded from it. + if missing := s.denseIndexesMissingNames(); len(missing) > 0 { + health.Reason = plannerStatsDenseIndexesMissingReason + health.DenseIndexesMissing = missing + return health, bases, nil + } present, err := s.plannerStatsPresentIndexes(ctx) if err != nil { diff --git a/internal/graph/store_sqlite/store.go b/internal/graph/store_sqlite/store.go index 3fec85127..a8d1ec5a9 100644 --- a/internal/graph/store_sqlite/store.go +++ b/internal/graph/store_sqlite/store.go @@ -66,6 +66,9 @@ type backgroundCheckpointCoordination struct { } type storeCore struct { + firstIndexDrainsMu sync.Mutex + firstIndexDrains map[string]graph.FirstIndexDrain + // baseFTSPages retains only exact ranked rows keyed by same-transaction // global FTS state and mutable base ownership revision. baseFTSPages symbolFTSPageCache @@ -340,6 +343,22 @@ type storeCore struct { maintenanceDrainRunning bool maintenanceDrainReason string + // Dense-index repair (dense_index_repair.go). denseRepairOwed is the + // one-slot request the lane worker consumes, posted by a bulk-window exit + // that left indexes missing or by the backoff timer; denseRepairDelay is + // the backoff the next retry waits and denseRepairTimer that pending + // retry. All three are guarded by maintenanceSched. The counters record + // attempts that reached their DDL, attempts skipped for free space, and + // completed repairs; denseRepairDiskAvail is a test override of the + // free-space probe. + denseRepairOwed bool + denseRepairDelay time.Duration + denseRepairTimer *time.Timer + denseIndexRepairAttempts atomic.Int64 + denseIndexRepairSkips atomic.Int64 + denseIndexRepairs atomic.Int64 + denseRepairDiskAvail atomic.Pointer[func(dir string) (uint64, error)] + // Lane counters. Requests counts scheduling calls, passes counts lane // passes actually started (so a burst of requests collapsing into one // pass is observable), jobs counts actions that reached their SQL, @@ -436,6 +455,11 @@ type storeCore struct { // synchronously under writeMu on cold paths; live FTS finalize emits after // releasing the gate. Cold observers must not call back into Store. bulkFinalizeObserver func(bulkFinalizeEvent) + // bulkSealHook and batchWriteHook are SetBulkSealHookForTest's and + // SetBatchWriteHookForTest's hooks. Per store rather than package-wide so + // parallel tests that each own a store do not share them. + bulkSealHook atomic.Pointer[func(index string) error] + batchWriteHook atomic.Pointer[func(nodeRows, edgeRows int) error] // Live FTS-finalize statement trace for tests; invoked only on those two // maintenance calls, never on ordinary graph/edit writes. ftsMaintenanceObserver func(string, []any) @@ -507,6 +531,16 @@ type storeCore struct { // health probe. Written only under writeMu, by syncBulkWindowLocked. bulkWindowOpen atomic.Bool + // denseIndexesMissing holds the dense indexes a cold bulk window closed + // without rebuilding (its seal failed, or it was aborted before sealing), + // in bulkDroppableIndexes order and each with the DDL that builds it; nil + // when none are owed. Written only under writeMu and replaced, never + // mutated in place, so the health probe reads it without a lock for the + // same reason it reads bulkWindowOpen, and the maintenance lane's repair + // builds from it rather than reading bulkDroppableIndexes itself. + // Not persisted: Open's createGraphCoreIndexes rebuilds them on restart. + denseIndexesMissing atomic.Pointer[[]bulkDroppableIndex] + // Prepared statements (compiled once in Open, closed in Close). stmtInsertNode *sql.Stmt stmtGetNode *sql.Stmt diff --git a/internal/graph/store_sqlite/store_compact.go b/internal/graph/store_sqlite/store_compact.go index e1a6959ab..da1ce87af 100644 --- a/internal/graph/store_sqlite/store_compact.go +++ b/internal/graph/store_sqlite/store_compact.go @@ -92,6 +92,9 @@ const ( maintenancePlannerStats maintenanceJob = "planner_stats" maintenanceVacuum maintenanceJob = "vacuum" maintenanceCheckpoint maintenanceJob = "wal_checkpoint_truncate" + // maintenanceDenseIndexRepair rebuilds the dense indexes an unsealed cold + // window left missing (dense_index_repair.go). + maintenanceDenseIndexRepair maintenanceJob = "dense_index_repair" ) // preemptible reports whether a job in flight may be asked to yield the lane. @@ -640,8 +643,16 @@ func (s *Store) runMaintenanceLane(ctx context.Context, signal <-chan struct{}, return } } + // Ahead of the pass for the same reason as the drain, and because a + // completed repair owes the pass that analyzes what it rebuilt. + if s.takeDenseIndexRepair() { + s.runDenseIndexRepair(ctx) + if ctx.Err() != nil { + return + } + } if !s.statsPassOwed() { - // The wakeup carried a drain and nothing else. + // The wakeup carried a drain or a repair and nothing else. continue } if !s.claimMaintenancePass() { @@ -731,6 +742,8 @@ func (s *Store) stopMaintenanceLane() { s.maintenanceClosed = true s.maintenanceOwed = false s.maintenanceDrainOwed = false + s.denseRepairOwed = false + s.stopDenseIndexRepairTimerLocked() cancel := s.maintenanceCancel done := s.maintenanceDone s.maintenanceSched.Unlock() diff --git a/internal/graph/store_sqlite/store_purge.go b/internal/graph/store_sqlite/store_purge.go index e3805ef9a..72ff63704 100644 --- a/internal/graph/store_sqlite/store_purge.go +++ b/internal/graph/store_sqlite/store_purge.go @@ -141,6 +141,9 @@ func (s *Store) PurgeRepo(prefix string) error { if err := tx.Commit(); err != nil { return err } + s.firstIndexDrainsMu.Lock() + delete(s.firstIndexDrains, prefix) + s.firstIndexDrainsMu.Unlock() s.finishAnalysisMutationLocked(len(ids) > 0) if changed { s.payloadInputAdminRevision.Add(1) @@ -351,6 +354,14 @@ func (s *Store) RekeyRepoPrefix(oldPrefix, newPrefix string) error { if err := tx.Commit(); err != nil { return err } + s.firstIndexDrainsMu.Lock() + if observation, ok := s.firstIndexDrains[oldPrefix]; ok { + if current, exists := s.firstIndexDrains[newPrefix]; !exists || current.CompletedAt.Before(observation.CompletedAt) { + s.firstIndexDrains[newPrefix] = observation + } + delete(s.firstIndexDrains, oldPrefix) + } + s.firstIndexDrainsMu.Unlock() if changed { s.payloadInputAdminRevision.Add(1) s.markMutationReceiptsIncompleteLocked() diff --git a/internal/indexer/drain_lane_yield.go b/internal/indexer/drain_lane_yield.go index df25c84b8..4e09a33d6 100644 --- a/internal/indexer/drain_lane_yield.go +++ b/internal/indexer/drain_lane_yield.go @@ -2,8 +2,12 @@ package indexer import ( "context" + "errors" + "fmt" "time" + "go.uber.org/zap" + "github.com/zzet/gortex/internal/graph" ) @@ -33,6 +37,12 @@ var drainYieldRows = 1024 // cooperative drain. var drainSubBatchHook func() +// shadowEvictionBranchHook, when set (tests), observes which eviction a shadow +// drain chose before its first disk write: true for the SQLite +// shadow-replacement retirement, false for the generic current-generation +// eviction. +var shadowEvictionBranchHook func(retirement bool) + type initialDrainCooperationKey struct{} // Only the initial claimed base pauses in place for announced checkout work. @@ -87,11 +97,13 @@ func drainYieldable(ctx context.Context) bool { // claimed drain uses sub-batches and stops on context cancellation. The initial // claimed drain also stands down, boundedly, for interactive demand before each // slice; ordinary and foreground drains keep their single-batch behavior. +// +// A cooperative stop returns the context's own error; anything else is the +// store refusing a write (see drainWrite). drainBatchError tells them apart. func drainAddBatch(ctx context.Context, target graph.Store, nodes []*graph.Node, edges []*graph.Edge) error { wanted := initialDrainDemand(ctx) if !drainYieldable(ctx) && wanted == nil { - target.AddBatch(nodes, edges) - return nil + return drainWrite(ctx, target, nodes, edges) } step := drainYieldRows if step <= 0 { @@ -110,13 +122,63 @@ func drainAddBatch(ctx context.Context, target graph.Store, nodes []*graph.Node, } if len(nodes) > 0 { n := min(len(nodes), step) - target.AddBatch(nodes[:n], nil) + if err := drainWrite(ctx, target, nodes[:n], nil); err != nil { + return err + } nodes = nodes[n:] continue } n := min(len(edges), step) - target.AddBatch(nil, edges[:n]) + if err := drainWrite(ctx, target, nil, edges[:n]); err != nil { + return err + } edges = edges[n:] } return nil } + +// drainCheckedWriter is the error-returning batch write a durable backend +// offers beside the void graph.Store AddBatch. +type drainCheckedWriter interface { + AddBatchCheckedContext(ctx context.Context, nodes []*graph.Node, edges []*graph.Edge) error +} + +// drainWrite writes one batch, through the checked path when target has one, +// so a storage failure (a full volume, an I/O error, a failed deferred-index +// seal) fails the drain as an error instead of panicking out of AddBatch. The +// write itself never observes ctx's cancellation: a drain stops between +// batches, never inside one. +func drainWrite(ctx context.Context, target graph.Store, nodes []*graph.Node, edges []*graph.Edge) error { + if checked, ok := target.(drainCheckedWriter); ok { + return checked.AddBatchCheckedContext(context.WithoutCancel(ctx), nodes, edges) + } + target.AddBatch(nodes, edges) + return nil +} + +// drainBatchError wraps a drainAddBatch error for the build's result: a +// cooperative stop keeps its established wording, and a refused write says the +// graph could not be persisted. +func drainBatchError(ctx context.Context, err error) error { + if ctxErr := ctx.Err(); ctxErr != nil && errors.Is(err, ctxErr) { + return fmt.Errorf("indexer: cooperative drain stopped: %w", err) + } + return fmt.Errorf("indexer: persist bulk graph batch: %w", err) +} + +// retryBulkFlushAfterPanic retries a FlushBulk that panicked, once, while that +// panic is unwinding. The retry recovers its own panic and only logs what it +// hit, so the original panic is the one that reaches the caller. Under an +// outer coordinated or generation window FlushBulk is a no-op, so this is too. +func retryBulkFlushAfterPanic(bl graph.BulkLoader, logger *zap.Logger, repo string) { + defer func() { + if r := recover(); r != nil { + logger.Error("indexer: bulk flush retry after panic panicked", + zap.String("repo", repo), zap.Any("panic", r)) + } + }() + if err := bl.FlushBulk(); err != nil { + logger.Error("indexer: bulk flush retry after panic failed", + zap.String("repo", repo), zap.Error(err)) + } +} diff --git a/internal/indexer/edit_delta_capability_test.go b/internal/indexer/edit_delta_capability_test.go index 22bff4e4e..ae8b7c639 100644 --- a/internal/indexer/edit_delta_capability_test.go +++ b/internal/indexer/edit_delta_capability_test.go @@ -34,6 +34,11 @@ var _ graph.FileIdentityBatchEvicter = (*graph.DeltaWriter)(nil) // deliberately does not implement, with the reason. The reason names what // the engine does instead. var editDeltaAbsentCapabilities = map[string]string{ + "anonymous{DBStats}": "physical durable-store file-size telemetry; full-shadow drain samples a real store, whereas DeltaWriter stages per-file changes without owning a database or WAL file", + + "anonymous{BulkLoadWindow}": "physical durable-store full-shadow window telemetry; DeltaWriter stages per-file changes and does not own a cold or generation bulk window", + "anonymous{RecordFirstIndexDrain}": "physical durable-store full-shadow completion telemetry; a per-file DeltaWriter has no full-shadow drain to record and must not forward writes to its base", + "graph.BFSAdmissionReader": "optional selected-reader traversal snapshot; the query engine retains its composed point-walk admission when a delta has no snapshot accelerator", "graph.FilteredBFSAdmissionReader": "physical indexed scoped traversal; forwarding it would discard delta ownership, so the engine keeps composed Go admission instead", "graph.KindAdjacencyReader": "optional typed adjacency accelerator; ReadKindAdjacency filters the delta's composed point reads instead of forwarding physical rows past delta ownership", @@ -215,6 +220,7 @@ var editDeltaAbsentCapabilities = map[string]string{ "indexer.repoSemanticStateReleaser": "semantic (go/types) state: the enrichment stage runs over the generation handle after the delta (runEnrichment), not over the DeltaWriter", "indexer.repoSemanticStateRetainer": "semantic (go/types) state: the enrichment stage runs over the generation handle after the delta (runEnrichment), not over the DeltaWriter", "indexer.repositoryCleanupPayload": "whole-repository or administrative operation (full index, repository reset, untrack or cleanup, warm start, vector corpus, bulk load windows, WAL maintenance); not on a file delta's per-save path", + "indexer.drainCheckedWriter": "checked write of the whole-repository shadow drain into the durable store (full index, shadow replacement); not on a file delta's per-save path, and a store without it takes the void AddBatch", "resolver.backendResolveWorkIndicator": "whole-graph pass (ResolveAll, a cold warm-up, a whole-repository derived rebuild); a delta resolves and re-derives by its receipt frontier", "resolver.batchClaimingResolver": "framework synthesizer contract, asserted on synthesizers", "resolver.claimEdgeVocabulary": "framework synthesizer contract, asserted on synthesizers", diff --git a/internal/indexer/first_index_bulk_test.go b/internal/indexer/first_index_bulk_test.go index d6e2e0ba5..30855ebfc 100644 --- a/internal/indexer/first_index_bulk_test.go +++ b/internal/indexer/first_index_bulk_test.go @@ -6,12 +6,13 @@ import ( "errors" "os" "path/filepath" - "sync" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" "github.com/zzet/gortex/internal/config" "github.com/zzet/gortex/internal/graph" @@ -19,18 +20,14 @@ import ( "github.com/zzet/gortex/internal/parser" "github.com/zzet/gortex/internal/parser/languages" "github.com/zzet/gortex/internal/progress" + "github.com/zzet/gortex/internal/search" ) // Embed the real store so the shadow and its optional persistence capabilities // stay intact. Only the drain boundaries are intercepted. type firstIndexBulkProbe struct { *store_sqlite.Store - beforeBegin func() duringDrain func() - ftsErr error - panicDrain bool - panicFlush bool - flushMu sync.Mutex begins int flushes int batches int @@ -38,41 +35,41 @@ type firstIndexBulkProbe struct { func (s *firstIndexBulkProbe) BeginBulkLoad() { s.begins++ - if s.beforeBegin != nil { - s.beforeBegin() - } s.Store.BeginBulkLoad() } -func (s *firstIndexBulkProbe) AddBatch(nodes []*graph.Node, edges []*graph.Edge) { +func (s *firstIndexBulkProbe) beforeBatch() { s.batches++ if s.duringDrain != nil { s.duringDrain() } - if s.panicDrain { - panic("injected drain panic") - } +} + +func (s *firstIndexBulkProbe) AddBatch(nodes []*graph.Node, edges []*graph.Edge) { + s.beforeBatch() s.Store.AddBatch(nodes, edges) } -func (s *firstIndexBulkProbe) BatchUpsertSymbolFTS(items []graph.SymbolFTSItem) error { - if s.ftsErr != nil { - return s.ftsErr - } - return s.Store.BatchUpsertSymbolFTS(items) +// The drain writes through the checked path when the target has one. +func (s *firstIndexBulkProbe) AddBatchCheckedContext(ctx context.Context, nodes []*graph.Node, edges []*graph.Edge) error { + s.beforeBatch() + return s.Store.AddBatchCheckedContext(ctx, nodes, edges) } func (s *firstIndexBulkProbe) FlushBulk() error { s.flushes++ - if s.panicFlush { - // Model a FlushBulk panic that leaves its writer lock held. - s.flushMu.Lock() - panic("injected flush panic") - } return s.Store.FlushBulk() } func newFirstIndexBulkFixture(t *testing.T) (*firstIndexBulkProbe, *sql.DB) { + t.Helper() + store, db := newFirstIndexStore(t) + return &firstIndexBulkProbe{Store: store}, db +} + +// newFirstIndexStore opens a fresh on-disk store for the shadow route with a +// read-only pool beside it for schema probes. +func newFirstIndexStore(t *testing.T) (*store_sqlite.Store, *sql.DB) { t.Helper() t.Setenv("GORTEX_SHADOW_MAX_FILES", "1000000") t.Setenv("GORTEX_SHADOW_MAX_BYTES", "1073741824") @@ -84,7 +81,19 @@ func newFirstIndexBulkFixture(t *testing.T) (*firstIndexBulkProbe, *sql.DB) { db, err := sql.Open("sqlite", "file:"+filepath.ToSlash(path)+"?mode=ro&_pragma=synchronous(NORMAL)") require.NoError(t, err) t.Cleanup(func() { require.NoError(t, db.Close()) }) - return &firstIndexBulkProbe{Store: store}, db + return store, db +} + +// firstIndexWriterSynchronous reads PRAGMA synchronous off the store's writer +// connection. It takes the write gate under a deadline, so it also fails, +// rather than hangs, when a failure path leaked the gate. +func firstIndexWriterSynchronous(t *testing.T, store *store_sqlite.Store) int64 { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + sync, err := store.WriterPragmaForTest(ctx, "synchronous") + require.NoError(t, err, "the store's write gate is still held") + return sync } func firstIndexFixtureRepo(t *testing.T) string { @@ -141,33 +150,194 @@ func assertFirstIndexRestored(t *testing.T, store *store_sqlite.Store, db *sql.D require.False(t, firstIndexBulkActive(t, store), "standalone window must be closed before restoration probes") assert.Equal(t, indexes, firstIndexSchemaIndexes(t, db)) // PRAGMA synchronous is connection-local, so this read-only pool cannot - // check writer restoration. TestBulkLoadDropsAndRebuildsIndexes in - // store_sqlite checks the actual writer connection instead. + // check writer restoration; firstIndexWriterSynchronous reads the writer. } -func TestFirstIndexDrainUsesBulkWindow(t *testing.T) { - probe, db := newFirstIndexBulkFixture(t) - indexes := firstIndexSchemaIndexes(t, db) - probe.beforeBegin = func() { assertFirstIndexNoReceipts(t, probe.Store, "repo") } - probe.duringDrain = func() { - assert.True(t, firstIndexBulkActive(t, probe.Store), "cold shadow drain must engage standalone bulk mode") - assertFirstIndexNoReceipts(t, probe.Store, "repo") - assert.NotEqual(t, indexes, firstIndexSchemaIndexes(t, db), "dense indexes must be deferred during drain") - } - idx := newFirstIndexBulkIndexer(t, probe, "repo") - result, err := idx.IndexCtx(context.Background(), firstIndexFixtureRepo(t)) +// sqliteFullError provokes a genuine SQLITE_FULL from the driver on a +// throwaway in-memory database capped at a few pages, so a test can hand a +// store the exact error a full volume produces without filling a disk. +func sqliteFullError(t *testing.T) error { + t.Helper() + db, err := sql.Open("sqlite", ":memory:") require.NoError(t, err) - require.Positive(t, result.NodeCount) - require.Positive(t, probe.batches, "test must observe the shadow drain") - assert.Equal(t, 1, probe.begins) - assert.Equal(t, 1, probe.flushes) - state, found, err := probe.GetRepoIndexState("repo") + defer func() { require.NoError(t, db.Close()) }() + db.SetMaxOpenConns(1) + _, err = db.Exec(`PRAGMA max_page_count = 2`) + require.NoError(t, err) + _, err = db.Exec(`CREATE TABLE filler(v BLOB)`) + require.NoError(t, err) + _, err = db.Exec(`INSERT INTO filler VALUES (randomblob(65536))`) + require.Error(t, err) + return err +} + +// A drain chunk the store refuses comes back as a typed storage error, for an +// ordinary drain and for the initial claimed one, instead of panicking out of +// the void AddBatch, and none of the refused chunk is committed. +func TestDrainAddBatchReturnsStorageFailure(t *testing.T) { + full := sqliteFullError(t) + for _, tc := range []struct { + name string + ctx context.Context + }{ + {"ordinary", context.Background()}, + {"initial claimed", withInitialDrainCooperation(context.Background(), func() bool { return false })}, + } { + t.Run(tc.name, func(t *testing.T) { + store, err := store_sqlite.Open(filepath.Join(t.TempDir(), "store.sqlite")) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, store.Close()) }) + defer store.SetBatchWriteHookForTest(func(int, int) error { return full })() + nodes := []*graph.Node{{ + ID: "app.py::fetch", Name: "fetch", Kind: graph.KindFunction, + RepoPrefix: "repo", FilePath: "app.py", Language: "python", + }} + + var drainErr error + require.NotPanics(t, func() { drainErr = drainAddBatch(tc.ctx, store, nodes, nil) }) + var storageErr *store_sqlite.StorageError + require.ErrorAs(t, drainErr, &storageErr) + assert.Contains(t, store_sqlite.SafeStorageFailureReason(drainErr), "SQLite code 13") + assert.Contains(t, drainBatchError(tc.ctx, drainErr).Error(), "indexer: persist bulk graph batch:") + assert.Zero(t, store.NodeCount(), "the refused chunk must roll back") + }) + } +} + +// firstIndexWindowWitness counts a store's batch writes, how many of them +// landed inside an engaged bulk window with the dense indexes deferred, and how +// often each dense index was sealed. fail, when set, decides whether the store +// refuses a batch. +type firstIndexWindowWitness struct { + batches, engaged, deferred int + seals map[string]int + fail func(edgeRows int) error +} + +func watchFirstIndexWindow(t *testing.T, store *store_sqlite.Store, db *sql.DB, indexes []string) *firstIndexWindowWitness { + t.Helper() + w := &firstIndexWindowWitness{seals: map[string]int{}} + t.Cleanup(store.SetBulkSealHookForTest(func(index string) error { + w.seals[index]++ + return nil + })) + t.Cleanup(store.SetBatchWriteHookForTest(func(_, edgeRows int) error { + w.batches++ + if firstIndexBulkActive(t, store) { + w.engaged++ + } + if !assert.ObjectsAreEqual(indexes, firstIndexSchemaIndexes(t, db)) { + w.deferred++ + } + if w.fail != nil { + return w.fail(edgeRows) + } + return nil + })) + // The eviction runs just before the window opens, so this is the last + // point at which a receipt could wrongly precede the drain. + shadowEvictionBranchHook = func(retirement bool) { + assert.True(t, retirement, "a SQLite target must take the shadow-replacement retirement") + assertFirstIndexNoReceipts(t, store, "repo") + } + t.Cleanup(func() { shadowEvictionBranchHook = nil }) + return w +} + +func (w *firstIndexWindowWitness) assertEngaged(t *testing.T) { + t.Helper() + require.Positive(t, w.batches, "test must observe the shadow drain") + assert.Positive(t, w.engaged, "cold shadow drain must engage standalone bulk mode") + assert.Positive(t, w.deferred, "dense indexes must be deferred during drain") + // One window per drain: a second one would seal every index again. + require.NotEmpty(t, w.seals, "the window must seal the indexes it deferred") + for index, n := range w.seals { + assert.Equal(t, 1, n, "index %s sealed %d times", index, n) + } +} + +func assertFirstIndexReceipts(t *testing.T, store *store_sqlite.Store, result *IndexResult) { + t.Helper() + state, found, err := store.GetRepoIndexState("repo") require.NoError(t, err) require.True(t, found) assert.Equal(t, result.NodeCount, state.NodeCount) assert.Equal(t, result.EdgeCount, state.EdgeCount) - assert.Contains(t, probe.LoadFileMtimes("repo"), "app.py") - assertFirstIndexRestored(t, probe.Store, db, indexes) + assert.NotEmpty(t, store.LoadFileMtimes("repo")) +} + +// A cold first index drains through a standalone bulk window on every route +// into it, and again on a retry after a failed drain left rows behind. +func TestFirstIndexDrainUsesBulkWindow(t *testing.T) { + t.Run("IndexCtx on a fresh store", func(t *testing.T) { + store, db := newFirstIndexStore(t) + indexes := firstIndexSchemaIndexes(t, db) + writerSync := firstIndexWriterSynchronous(t, store) + w := watchFirstIndexWindow(t, store, db, indexes) + + result, err := newFirstIndexBulkIndexer(t, store, "repo").IndexCtx(context.Background(), firstIndexFixtureRepo(t)) + require.NoError(t, err) + require.Positive(t, result.NodeCount) + w.assertEngaged(t) + assertFirstIndexReceipts(t, store, result) + assert.Contains(t, store.LoadFileMtimes("repo"), "app.py") + assert.Equal(t, writerSync, firstIndexWriterSynchronous(t, store)) + assertFirstIndexRestored(t, store, db, indexes) + }) + + t.Run("TrackRepoCtx on an empty multi-repo store", func(t *testing.T) { + store, db := newFirstIndexStore(t) + indexes := firstIndexSchemaIndexes(t, db) + writerSync := firstIndexWriterSynchronous(t, store) + w := watchFirstIndexWindow(t, store, db, indexes) + dir := t.TempDir() + writeFile(t, filepath.Join(dir, "main.go"), "package app\n\nfunc Fetch(url string) string { return url }\n") + mi := NewMultiIndexer(store, newTestRegistry(), search.NewNull(), newTestConfigManager(t), zap.NewNop()) + t.Cleanup(func() { require.NoError(t, mi.Close(context.Background())) }) + + result, err := mi.TrackRepoCtx(t.Context(), config.RepoEntry{Name: "repo", Path: dir}) + require.NoError(t, err) + require.Positive(t, result.NodeCount) + w.assertEngaged(t) + assertFirstIndexReceipts(t, store, result) + assert.Equal(t, writerSync, firstIndexWriterSynchronous(t, store)) + assertFirstIndexRestored(t, store, db, indexes) + }) + + t.Run("a retry after a failed drain engages the window again", func(t *testing.T) { + store, db := newFirstIndexStore(t) + indexes := firstIndexSchemaIndexes(t, db) + writerSync := firstIndexWriterSynchronous(t, store) + w := watchFirstIndexWindow(t, store, db, indexes) + injected := errors.New("injected edge batch failure") + refused := false + w.fail = func(edgeRows int) error { + if edgeRows > 0 && !refused { + refused = true + return injected + } + return nil + } + dir := firstIndexFixtureRepo(t) + + // The node batches land; the first edge batch is refused. + _, err := newFirstIndexBulkIndexer(t, store, "repo").IndexCtx(context.Background(), dir) + require.ErrorIs(t, err, injected) + require.True(t, refused, "the drain never reached an edge batch") + w.assertEngaged(t) + require.Positive(t, store.NodeCount(), "the failed drain must leave rows for the retry to evict") + assertFirstIndexNoReceipts(t, store, "repo") + assertFirstIndexRestored(t, store, db, indexes) + + // The retry evicts them, which leaves the store cold again. + w.batches, w.engaged, w.deferred, w.seals = 0, 0, 0, map[string]int{} + result, err := newFirstIndexBulkIndexer(t, store, "repo").IndexCtx(context.Background(), dir) + require.NoError(t, err) + w.assertEngaged(t) + assertFirstIndexReceipts(t, store, result) + assert.Equal(t, writerSync, firstIndexWriterSynchronous(t, store)) + assertFirstIndexRestored(t, store, db, indexes) + }) } func TestSecondRepoWarmStoreNoBulkWindow(t *testing.T) { @@ -227,81 +397,138 @@ func (firstIndexPanicReporter) Report(stage string, _, _ int) { } } +// firstIndexDrainFTSTrigger makes every symbol FTS append on store fail, the +// way a storage error inside the drain's FTS writes would. +func firstIndexDrainFTSTrigger(t *testing.T, store *store_sqlite.Store) { + t.Helper() + writer, err := sql.Open("sqlite", "file:"+filepath.ToSlash(store.Path())) + require.NoError(t, err) + _, err = writer.Exec(`CREATE TRIGGER fail_first_index_fts BEFORE INSERT ON symbol_fts_rowid +BEGIN SELECT RAISE(ABORT, 'injected drain FTS error'); END`) + require.NoError(t, err) + require.NoError(t, writer.Close()) +} + +// Every way a first index can fail, on the real store: the bulk window must +// close (no leaked write gate, writer durability back, the dense indexes back), +// no success receipt may be written, and the indexer must hand its disk target +// back. Faults inside the store come from its own test hooks, so the eviction +// the drain picks is the production one for a SQLite target. func TestFirstIndexFailureRestoresStore(t *testing.T) { - for _, mode := range []string{"drain error", "cancel before drain", "panic in body", "panic in drain"} { + for _, mode := range []string{ + "drain error", "cancel before drain", "panic in body", "panic in drain", + "storage full", "seal panic", "seal panic after a drain panic", "seal error", + } { t.Run(mode, func(t *testing.T) { - probe, db := newFirstIndexBulkFixture(t) + store, db := newFirstIndexStore(t) indexes := firstIndexSchemaIndexes(t, db) - injected := errors.New("injected drain FTS error") + writerSync := firstIndexWriterSynchronous(t, store) + var evictions []bool + shadowEvictionBranchHook = func(retirement bool) { evictions = append(evictions, retirement) } + t.Cleanup(func() { shadowEvictionBranchHook = nil }) + var fault func() error + batches, engaged := 0, false + defer store.SetBatchWriteHookForTest(func(int, int) error { + batches++ + engaged = engaged || firstIndexBulkActive(t, store) + if fault != nil { + return fault() + } + return nil + })() + seals := 0 + defer store.SetBulkSealHookForTest(func(index string) error { + seals++ + switch { + case (mode == "seal panic" || mode == "seal panic after a drain panic") && seals == 1: + panic("injected seal panic") + case mode == "seal error" && index == "nodes_by_kind": + return errors.New("injected seal failure") + } + return nil + })() ctx, cancel := context.WithCancel(context.Background()) defer cancel() switch mode { case "drain error": - probe.ftsErr = injected - case "panic in drain": - probe.panicDrain = true + firstIndexDrainFTSTrigger(t, store) + case "panic in drain", "seal panic after a drain panic": + fault = func() error { panic("injected drain panic") } + case "storage full": + full := sqliteFullError(t) + fault = func() error { return full } case "cancel before drain": ctx = progress.WithReporter(ctx, firstIndexCancelReporter{cancel: cancel}) case "panic in body": ctx = progress.WithReporter(ctx, firstIndexPanicReporter{}) } - probe.duringDrain = func() { - assert.True(t, firstIndexBulkActive(t, probe.Store), "failure fixture must exercise an engaged bulk window") + idx := newFirstIndexBulkIndexer(t, store, "repo") + if mode == "seal panic" { + idx.indexMemoryAdmission = newIndexMemoryAdmissionBudget(defaultIndexMemoryAdmissionBytes) } - idx := newFirstIndexBulkIndexer(t, probe, "repo") dir := firstIndexFixtureRepo(t) + switch mode { case "panic in body": assert.PanicsWithValue(t, "injected body panic", func() { _, _ = idx.IndexCtx(ctx, dir) }) - assert.Zero(t, probe.NodeCount(), "body panic must not drain shadow nodes") - assert.Zero(t, probe.EdgeCount(), "body panic must not drain shadow edges") + assert.Zero(t, store.NodeCount(), "body panic must not drain shadow nodes") + assert.Zero(t, store.EdgeCount(), "body panic must not drain shadow edges") case "panic in drain": require.PanicsWithValue(t, "injected drain panic", func() { _, _ = idx.IndexCtx(ctx, dir) }) + case "seal panic": + require.PanicsWithValue(t, "injected seal panic", func() { _, _ = idx.IndexCtx(ctx, dir) }) + assert.Greater(t, seals, 1, "the backstop did not retry the panicked flush") + assert.Positive(t, idx.indexMemoryAdmission.snapshot().peak, "fixture must acquire the memory lease") + assert.Zero(t, idx.indexMemoryAdmission.snapshot().used, "earlier defers must release the memory lease") + case "seal panic after a drain panic": + // The flush the drain panic's unwind runs panics too; it is + // retried the same way, and the later panic reaches the caller. + require.PanicsWithValue(t, "injected seal panic", func() { _, _ = idx.IndexCtx(ctx, dir) }) + assert.Greater(t, seals, 1, "the backstop did not retry the panicked flush") + case "cancel before drain": + _, err := idx.IndexCtx(ctx, dir) + require.ErrorIs(t, err, context.Canceled) case "drain error": _, err := idx.IndexCtx(ctx, dir) - require.ErrorIs(t, err, injected) - default: + require.ErrorContains(t, err, "injected drain FTS error") + case "storage full": _, err := idx.IndexCtx(ctx, dir) - require.ErrorIs(t, err, context.Canceled) + var storageErr *store_sqlite.StorageError + require.ErrorAs(t, err, &storageErr) + assert.Contains(t, store_sqlite.SafeStorageFailureReason(err), "SQLite code 13") + assert.Equal(t, 1, batches, "a refused write must be the drain's last") + case "seal error": + _, err := idx.IndexCtx(ctx, dir) + require.ErrorContains(t, err, "nodes_by_kind") + // Either still owed or already rebuilt by the maintenance lane, + // which runs as soon as the window closes on a volume with room. + health, healthErr := store.PlannerStatsHealth(context.Background()) + require.NoError(t, healthErr) + if health.Reason == "dense_indexes_missing" { + assert.Equal(t, []string{"nodes_by_kind"}, health.DenseIndexesMissing) + } + deadline := time.Now().Add(10 * time.Second) + for !assert.ObjectsAreEqual(indexes, firstIndexSchemaIndexes(t, db)) && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } } + if mode == "cancel before drain" || mode == "panic in body" { - assert.Zero(t, probe.begins) - assert.Zero(t, probe.batches) + assert.Empty(t, evictions, "a failure before the drain must not evict") + assert.Zero(t, batches, "a failure before the drain must not write") } else { - require.Positive(t, probe.batches) - assert.Equal(t, 1, probe.begins) + assert.Equal(t, []bool{true}, evictions, "a SQLite target must take the shadow-replacement retirement") + require.Positive(t, batches, "test must observe the shadow drain") + assert.True(t, engaged, "failure fixture must exercise an engaged bulk window") } - assertFirstIndexRestored(t, probe.Store, db, indexes) - assertFirstIndexNoReceipts(t, probe.Store, "repo") - assert.Same(t, probe, idx.graph, "panic and error paths must restore disk target") + assert.Equal(t, writerSync, firstIndexWriterSynchronous(t, store), "writer durability must be restored") + assertFirstIndexRestored(t, store, db, indexes) + assertFirstIndexNoReceipts(t, store, "repo") + assert.Same(t, store, idx.graph, "panic and error paths must restore disk target") }) } } -func TestFirstIndexFlushPanicPropagates(t *testing.T) { - // Given a FlushBulk implementation that panics while holding its writer lock. - probe, _ := newFirstIndexBulkFixture(t) - probe.panicFlush = true - idx := newFirstIndexBulkIndexer(t, probe, "repo") - idx.indexMemoryAdmission = newIndexMemoryAdmissionBudget(defaultIndexMemoryAdmissionBytes) - // The probe's lock is independent of Store; close the real bulk window at cleanup. - t.Cleanup(func() { require.NoError(t, probe.Store.FlushBulk()) }) - - // When the completed shadow drain flushes, it must not retry that locked call. - require.PanicsWithValue(t, "injected flush panic", func() { - _, _ = idx.IndexCtx(context.Background(), firstIndexFixtureRepo(t)) - }) - - // Then the panic propagates and the backstop does not re-enter FlushBulk. - // This fixture has no failure-ledger rows, so it does not cover a real - // store whose FlushBulk panics with writeMu held while the ledger is dirty. - assert.Equal(t, 1, probe.flushes) - assert.Same(t, probe, idx.graph) - assert.Positive(t, idx.indexMemoryAdmission.snapshot().peak, "fixture must acquire the memory lease") - assert.Zero(t, idx.indexMemoryAdmission.snapshot().used, "earlier defers must release the memory lease") - assertFirstIndexNoReceipts(t, probe.Store, "repo") -} - type firstIndexDrainPanicReporter struct{} func (firstIndexDrainPanicReporter) Report(stage string, _, _ int) { @@ -327,12 +554,7 @@ func TestFirstIndexFailureRestoresSQLiteReplacement(t *testing.T) { writeFile(t, baseline, "old replacement baseline") ctx := context.Background() if mode == "drain error" { - writer, err := sql.Open("sqlite", "file:"+filepath.ToSlash(probe.Path())) - require.NoError(t, err) - _, err = writer.Exec(`CREATE TRIGGER fail_first_index_fts BEFORE INSERT ON symbol_fts_rowid -BEGIN SELECT RAISE(ABORT, 'injected drain FTS error'); END`) - require.NoError(t, err) - require.NoError(t, writer.Close()) + firstIndexDrainFTSTrigger(t, probe.Store) } else { ctx = progress.WithReporter(ctx, firstIndexDrainPanicReporter{}) } @@ -356,3 +578,18 @@ BEGIN SELECT RAISE(ABORT, 'injected drain FTS error'); END`) }) } } + +func TestFirstIndexDrainReportsWindowAndSeal(t *testing.T) { + store, _ := newFirstIndexStore(t) + core, observed := observer.New(zap.InfoLevel) + idx := newFirstIndexBulkIndexer(t, store, "repo") + idx.logger = zap.New(core) + result, err := idx.IndexCtx(t.Context(), firstIndexFixtureRepo(t)) + require.NoError(t, err) + entries := observed.FilterMessage("indexer: FlushBulk complete").All() + require.Len(t, entries, 1) + fields := entries[0].ContextMap() + assert.Equal(t, "cold", fields["window"]) + assert.Contains(t, fields, "seal_elapsed") + assert.EqualValues(t, result.NodeCount, fields["nodes"]) +} diff --git a/internal/indexer/go_module_extern_test.go b/internal/indexer/go_module_extern_test.go index e8546be5f..147ee509e 100644 --- a/internal/indexer/go_module_extern_test.go +++ b/internal/indexer/go_module_extern_test.go @@ -153,8 +153,17 @@ func Caller() { library.Hello(); child.Nested() } } case "parallel": errors := make(chan error, len(repos)) + // This is a module-binding test, not an identity-admission + // saturation test. The decoy case has three roots; retain + // parallel tracking within the two native observation slots. + admitted := make(chan struct{}, 2) for _, entry := range repos { - go func() { _, e := mi.TrackRepoCtx(context.Background(), entry); errors <- e }() + go func() { + admitted <- struct{}{} + defer func() { <-admitted }() + _, e := mi.TrackRepoCtx(context.Background(), entry) + errors <- e + }() } for range repos { require.NoError(t, <-errors) diff --git a/internal/indexer/indexer.go b/internal/indexer/indexer.go index 035ab698c..d4ce2cb8d 100644 --- a/internal/indexer/indexer.go +++ b/internal/indexer/indexer.go @@ -3408,6 +3408,19 @@ func (idx *Indexer) indexCtxRaw(ctx context.Context, root string) (result *Index return } drainStart := time.Now() + var drainWALPeak int64 + var drainWALObserved bool + walReporter, hasWALReporter := diskTarget.(interface{ DBStats() (int64, int64) }) + sampleDrainWAL := func() { + if hasWALReporter { + _, walBytes := walReporter.DBStats() + if walBytes > 0 { + drainWALObserved = true + } + drainWALPeak = max(drainWALPeak, walBytes) + } + } + sampleDrainWAL() shadowNodeCount := inMemShadow.NodeCount() shadowEdgeCount := inMemShadow.EdgeCount() if !shadowEstimateReady { @@ -3438,11 +3451,16 @@ func (idx *Indexer) indexCtxRaw(ctx context.Context, root string) (result *Index restubGoModuleRepoIncoming(diskTarget, idx.RepoPrefix()) } var evictedNodes, evictedEdges int - if sqlite, ok := diskTarget.(*store_sqlite.Store); ok && sqlite.SupportsShadowReplacementRetirement() { + sqliteTarget, isSQLite := diskTarget.(*store_sqlite.Store) + retirement := isSQLite && sqliteTarget.SupportsShadowReplacementRetirement() + if shadowEvictionBranchHook != nil { + shadowEvictionBranchHook(retirement) + } + if retirement { var err error shadowReplacementStarted = true restubTrackedNpmRepoIncoming(diskTarget, idx.RepoPrefix()) - evictedNodes, evictedEdges, err = sqlite.EvictRepoForShadowReplacement(ctx, idx.RepoPrefix()) + evictedNodes, evictedEdges, err = sqliteTarget.EvictRepoForShadowReplacement(ctx, idx.RepoPrefix()) if err != nil { retErr = fmt.Errorf("indexer: retire shadow replacement rows: %w", err) return @@ -3461,17 +3479,34 @@ func (idx *Indexer) indexCtxRaw(ctx context.Context, root string) (result *Index return } bl.BeginBulkLoad() + bulkWindow := "unknown" + if reporter, ok := diskTarget.(interface{ BulkLoadWindow() string }); ok { + bulkWindow = reporter.BulkLoadWindow() + } // FlushBulk is a no-op under an outer coordinated window. Otherwise // it must also restore the standalone window if a drain write panics. bulkFlushPending := true bulkFlushInProgress := false defer func() { - // A FlushBulk panic may leave the store's writer lock held. - // Do not re-enter it; let the panic unwind the earlier defers. - if bulkFlushPending && !bulkFlushInProgress { - if err := bl.FlushBulk(); err != nil && retErr == nil { - retErr = fmt.Errorf("indexer: persist bulk graph: %w", err) - } + // A FlushBulk panicked: the explicit one below, or the one the + // next defer runs while a drain panic unwinds. The store releases + // its write gate on every exit and keeps a window whose seal did + // not finish retryable, so one more flush closes it (restoring + // durability, or recording the indexes it could not rebuild) + // instead of leaving a pinned synchronous=OFF writer behind. + if bulkFlushPending && bulkFlushInProgress { + retryBulkFlushAfterPanic(bl, idx.logger, idx.RepoPrefix()) + } + }() + defer func() { + if !bulkFlushPending || bulkFlushInProgress { + return + } + bulkFlushInProgress = true + err := bl.FlushBulk() + bulkFlushPending = false + if err != nil && retErr == nil { + retErr = fmt.Errorf("indexer: persist bulk graph: %w", err) } }() // Transfer the shadow in explicit row/byte-capped batches. The @@ -3506,6 +3541,7 @@ func (idx *Indexer) indexCtxRaw(ctx context.Context, root string) (result *Index // The builtin sentinels the drain is about to move, kept aside for // the re-assert below. See restampedBuiltins. var restampedBuiltins []*graph.Node + nodeDrainStopped := false for nodes := range inMemShadow.DrainNodeBatches(persistChunkRows, persistChunkBytes) { for _, node := range nodes { if node != nil && graph.IsBuiltinStub(node.ID) { @@ -3513,9 +3549,17 @@ func (idx *Indexer) indexCtxRaw(ctx context.Context, root string) (result *Index restampedBuiltins = append(restampedBuiltins, &copied) } } - if err := drainAddBatch(ctx, diskTarget, nodes, nil); err != nil && retErr == nil { - retErr = fmt.Errorf("indexer: cooperative drain stopped: %w", err) + if err := drainAddBatch(ctx, diskTarget, nodes, nil); err != nil { + // A refused write fails every later batch the same way, and + // a cooperative stop is final; either ends the drain, so the + // edge drain below writes nothing for this failed build. + if retErr == nil { + retErr = drainBatchError(ctx, err) + } + nodeDrainStopped = true + break } + sampleDrainWAL() if retErr != nil && drainChecksCancellation(ctx) && ctx.Err() != nil { break } @@ -3536,6 +3580,7 @@ func (idx *Indexer) indexCtxRaw(ctx context.Context, root string) (result *Index ftsReady = false return false } + sampleDrainWAL() ftsItemCount += len(ftsItems) ftsItems = make([]graph.SymbolFTSItem, 0, min(len(nodes), persistFTSChunkRows)) ftsBytes = 0 @@ -3568,6 +3613,9 @@ func (idx *Indexer) indexCtxRaw(ctx context.Context, root string) (result *Index drainPressure.afterNodeBatch(nodeRows) } for edges := range inMemShadow.DrainEdgeBatches(persistChunkRows, persistChunkBytes) { + if nodeDrainStopped { + break + } if drainChecksCancellation(ctx) && ctx.Err() != nil { if retErr == nil { retErr = fmt.Errorf("indexer: cooperative drain stopped: %w", ctx.Err()) @@ -3576,10 +3624,11 @@ func (idx *Indexer) indexCtxRaw(ctx context.Context, root string) (result *Index } if err := drainAddBatch(ctx, diskTarget, nil, edges); err != nil { if retErr == nil { - retErr = fmt.Errorf("indexer: cooperative drain stopped: %w", err) + retErr = drainBatchError(ctx, err) } break } + sampleDrainWAL() edgeRows := len(edges) drainPressure.afterEdgeBatch(edgeRows) } @@ -3605,10 +3654,17 @@ func (idx *Indexer) indexCtxRaw(ctx context.Context, root string) (result *Index // row or no composed view can equal it. The set is one row per // distinct builtin the repository references, so the copy is // bounded by the language's builtin vocabulary, not by corpus size. - if len(restampedBuiltins) > 0 { - diskTarget.AddBatch(restampedBuiltins, nil) + // + // Only after a drain that is still succeeding: a failed one publishes + // nothing and is evicted before its retry, and after a refused write + // this one would only be refused again. + if len(restampedBuiltins) > 0 && retErr == nil { + if err := drainWrite(ctx, diskTarget, restampedBuiltins, nil); err != nil { + retErr = drainBatchError(ctx, err) + } } + sampleDrainWAL() flushStart := time.Now() idx.logger.Info("indexer: FlushBulk start", zap.String("repo", idx.RepoPrefix()), @@ -3621,7 +3677,26 @@ func (idx *Indexer) indexCtxRaw(ctx context.Context, root string) (result *Index if ferr != nil && retErr == nil { retErr = fmt.Errorf("indexer: persist bulk graph: %w", ferr) } + sealElapsed := time.Since(flushStart) + sampleDrainWAL() + drainElapsed := time.Since(drainStart) + if recorder, ok := diskTarget.(interface { + RecordFirstIndexDrain(string, graph.FirstIndexDrain) + }); ok { + observation := graph.FirstIndexDrain{ + Window: bulkWindow, Nodes: shadowNodeCount, Edges: shadowEdgeCount, + DrainMS: float64(drainElapsed) / float64(time.Millisecond), + SealMS: float64(sealElapsed) / float64(time.Millisecond), CompletedAt: time.Now(), + WALPeakBytes: drainWALPeak, WALPeakObserved: drainWALObserved, WALPeakScope: "shared_store_drain_chunk_samples", + } + if retErr != nil { + observation.Error = retErr.Error() + } + recorder.RecordFirstIndexDrain(idx.RepoPrefix(), observation) + } idx.logger.Info("indexer: FlushBulk complete", + zap.String("window", bulkWindow), + zap.Duration("seal_elapsed", sealElapsed), zap.String("repo", idx.RepoPrefix()), zap.Duration("flush_elapsed", time.Since(flushStart)), zap.Duration("total_drain", time.Since(drainStart)), diff --git a/internal/indexer/multi_cold_orchestration_test.go b/internal/indexer/multi_cold_orchestration_test.go index 71f0fa479..345ce554f 100644 --- a/internal/indexer/multi_cold_orchestration_test.go +++ b/internal/indexer/multi_cold_orchestration_test.go @@ -473,6 +473,36 @@ func TestIndexMultiRepoRefFactFailureStopsGlobalConsumers(t *testing.T) { "a failed reference-fact boundary must not publish global completion") } +// TestIndexMultiRepoFailedFinalSealGivesUpTheWindow: when the coordinated +// window's final seal keeps failing, the cold pass reports the failure and then +// gives the window up instead of leaving it open behind the error. The writer +// is back at its own durability, the planner probe no longer sees a live +// window, and the store's repair rebuilds the index the seal could not. +func TestIndexMultiRepoFailedFinalSealGivesUpTheWindow(t *testing.T) { + store, db := newFirstIndexStore(t) + indexes := firstIndexSchemaIndexes(t, db) + writerSync := firstIndexWriterSynchronous(t, store) + injected := errors.New("injected final seal failure") + restore := store.SetBulkSealHookForTest(func(index string) error { + if index == "nodes_by_kind" { + return injected + } + return nil + }) + mi := newColdOrchestrationMulti(store, zap.NewNop()) + + _, err := mi.indexFixtureRepos(coldRefFactRepos(t, 2)) + restore() + require.ErrorIs(t, err, injected) + assert.Equal(t, writerSync, firstIndexWriterSynchronous(t, store), "writer durability must be restored") + assert.False(t, firstIndexBulkActive(t, store), "the failed window must not stay open") + deadline := time.Now().Add(10 * time.Second) + for !assert.ObjectsAreEqual(indexes, firstIndexSchemaIndexes(t, db)) && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + assert.Equal(t, indexes, firstIndexSchemaIndexes(t, db), "the repair must rebuild the index the seal left out") +} + // TestBeginDeferredPassesOverlapSplitsPoolFromTail pins the overlap contract: // BeginDeferredPasses drains the enrichment pool on its own goroutine while // the caller is free to run the resolve phase; contracts and the catch-up diff --git a/internal/indexer/multi_fixture_test.go b/internal/indexer/multi_fixture_test.go index 2b21c49cf..66b3c847d 100644 --- a/internal/indexer/multi_fixture_test.go +++ b/internal/indexer/multi_fixture_test.go @@ -272,7 +272,7 @@ func (mi *MultiIndexer) indexFixtureRepos(repos []config.RepoEntry) (map[string] coordinatedBulkActive := coordinatedBulk != nil && coordinatedBulk.BeginCoordinatedBulkLoad() defer func() { if coordinatedBulkActive { - if err := coordinatedBulk.EndCoordinatedBulkLoad(); err != nil { + if err := graph.EndOrAbortCoordinatedBulkLoad(coordinatedBulk); err != nil { mi.logger.Error("multi-repo bulk-load cleanup failed", zap.Error(err)) } } @@ -364,7 +364,7 @@ func (mi *MultiIndexer) indexFixtureRepos(repos []config.RepoEntry) (map[string] old.Close() } if coordinatedBulkActive { - if err := coordinatedBulk.EndCoordinatedBulkLoad(); err != nil { + if err := graph.EndOrAbortCoordinatedBulkLoad(coordinatedBulk); err != nil { return nil, fmt.Errorf("multi-repo bulk-load finalize: %w", err) } coordinatedBulkActive = false diff --git a/internal/indexer/prefix_observation_publication_test.go b/internal/indexer/prefix_observation_publication_test.go new file mode 100644 index 000000000..26df8a0ec --- /dev/null +++ b/internal/indexer/prefix_observation_publication_test.go @@ -0,0 +1,71 @@ +package indexer + +import ( + "context" + "os" + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/require" +) + +// Publication must mean the native worker has returned its allowance. A caller +// often starts another observation immediately; announcing completion before +// permit return can spuriously exhaust both slots under parallel tracking. +func TestIdentityObservationPublishesOnlyAfterPermitReturn(t *testing.T) { + root := t.TempDir() + for _, operation := range []string{"stat", "naming"} { + t.Run(operation, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var allowance prefixStatAllowance + allowance.once.Do(func() { allowance.slots = make(chan struct{}, 2) }) + nativeDone := make(chan struct{}) + // Temporarily intercept this worker's token so its release must + // wait for the test. This makes publication ordering observable + // without scheduler timing, polling or an extra production hook. + finishNative := func() { + <-allowance.slots + close(nativeDone) + } + ctx, cancel := context.WithTimeout(t.Context(), time.Minute) + defer cancel() + result := make(chan error, 1) + if operation == "stat" { + allowance.stat = func(path string) (os.FileInfo, error) { + info, err := os.Stat(path) + finishNative() + return info, err + } + go func() { result <- allowance.observe(ctx, root).err }() + } else { + go func() { + _, err := allowance.observeNaming(ctx, root, func(context.Context, string) trackNamingFacts { + finishNative() + return trackNamingFacts{canonical: root} + }) + result <- err + }() + } + <-nativeDone + synctest.Wait() + var publishedEarly bool + var err error + select { + case err = <-result: + publishedEarly = true + default: + } + // Always let the worker finish before assertions, including RED. + allowance.slots <- struct{}{} + synctest.Wait() + if !publishedEarly { + err = <-result + } + require.NoError(t, err) + require.Zero(t, len(allowance.slots)) + require.False(t, publishedEarly, "completed identity result escaped before its permit was returned") + }) + }) + } +} diff --git a/internal/indexer/prefix_ownership.go b/internal/indexer/prefix_ownership.go index a8da85371..75beaf395 100644 --- a/internal/indexer/prefix_ownership.go +++ b/internal/indexer/prefix_ownership.go @@ -60,9 +60,14 @@ func (a *prefixStatAllowance) observe(ctx context.Context, path string) prefixSt slots := a.slots done := make(chan prefixStatResult, 1) go func() { - defer func() { <-slots }() - info, err := stat(path) - done <- prefixStatResult{info: info, err: err} + result := func() prefixStatResult { + defer func() { <-slots }() + info, err := stat(path) + return prefixStatResult{info: info, err: err} + }() + // A completed result authorizes the caller's next observation. + // Return this permit first, even when the receiver runs immediately. + done <- result }() select { case <-ctx.Done(): diff --git a/internal/indexer/track_naming.go b/internal/indexer/track_naming.go index 3505de97f..72b9dfb9b 100644 --- a/internal/indexer/track_naming.go +++ b/internal/indexer/track_naming.go @@ -86,8 +86,12 @@ func (a *prefixStatAllowance) observeNaming(ctx context.Context, root string, re workerCtx, cancel := context.WithDeadline(context.Background(), deadline) defer cancel() go func() { - defer func() { <-slots }() - done <- read(workerCtx, root) + facts := func() trackNamingFacts { + defer func() { <-slots }() + return read(workerCtx, root) + }() + // Completion includes permit return, as for root Stat observations. + done <- facts }() select { case <-ctx.Done(): diff --git a/internal/mcp/contract_core_capability_table_test.go b/internal/mcp/contract_core_capability_table_test.go index cdd3d5f34..798bfd7ee 100644 --- a/internal/mcp/contract_core_capability_table_test.go +++ b/internal/mcp/contract_core_capability_table_test.go @@ -118,6 +118,9 @@ type ( capBFSContext interface { BFSContext(context.Context, []string, graph.Direction, []graph.EdgeKind, int, int) ([]graph.BFSHop, error) } + capLastFirstIndexDrains interface { + LastFirstIndexDrains() map[string]graph.FirstIndexDrain + } capRepoPrefixes interface{ RepoPrefixes() []string } ) @@ -218,6 +221,7 @@ func contractCoreCapabilityTable() []contractCoreCapRow { {"NodeLightSequencer", capOf[graph.NodeLightSequencer], capDropped, "node-only streaming; graph.NodesLightSeq falls back to the AllNodesLight forwarder, the same rows materialised"}, {"Store", capOf[graph.Store], capDropped, "the whole backend contract; a request reader is never the store"}, {"local:WALReclaimStats", capOf[walReclaimSnapshotReader], capDropped, capReasonBackend + " (physical store counters, never selected graph state)"}, + {"local:LastFirstIndexDrains", capOf[capLastFirstIndexDrains], capDropped, capReasonBackend + " (shared-store drain observations, never selected graph state)"}, {"AnalysisGenerationStore", capOf[graph.AnalysisGenerationStore], capDropped, capReasonBackend}, {"AnalysisQueryStore", capOf[graph.AnalysisQueryStore], capDropped, capReasonBackend}, {"TryAnalysisQueryStore", capOf[graph.TryAnalysisQueryStore], capDropped, capReasonBackend}, diff --git a/internal/mcp/index_health_planner_stats_test.go b/internal/mcp/index_health_planner_stats_test.go index b89263823..7ce1553bf 100644 --- a/internal/mcp/index_health_planner_stats_test.go +++ b/internal/mcp/index_health_planner_stats_test.go @@ -199,6 +199,30 @@ func TestIndexHealth_SurfacesPlannerStats(t *testing.T) { "published a receiver cardinality for an index that is not in the schema") }) + // Dense indexes an interrupted bulk load left out are named, with the + // sentence that says what they cost and how they come back. The real + // state repairs itself on the store's maintenance lane as soon as the + // volume has room, so the verdict is fixed here rather than raced. + t.Run("missing dense indexes are named", func(t *testing.T) { + _, store := newEmptySQLiteBackedServer(t) + missing := []string{"nodes_by_kind", "edges_by_kind"} + wrapped := plannerHealthOverride{Store: store, health: graph.PlannerStatsFreshness{ + Reason: "dense_indexes_missing", DenseIndexesMissing: missing, + }} + idx := indexer.New(wrapped, testRegistry(), config.Default().Index, zap.NewNop()) + missingSrv := NewServer(query.NewEngine(wrapped), wrapped, idx, nil, zap.NewNop(), nil) + + payload, err := missingSrv.buildIndexHealthPayloadCtx(context.Background()) + require.NoError(t, err) + stats, ok := payload["planner_stats"].(map[string]any) + require.True(t, ok) + assert.Equal(t, "dense_indexes_missing", stats["reason"]) + assert.Equal(t, missing, stats["dense_indexes_missing"]) + recommendation, _ := payload["recommendation"].(string) + assert.Contains(t, recommendation, "nodes_by_kind, edges_by_kind") + assert.NotContains(t, recommendation, "planner statistics are stale") + }) + // The window between a deferred cooperative pass and the boundary that // finishes it, seen from the side an agent reads. // @@ -304,6 +328,16 @@ func TestIndexHealth_SurfacesPlannerStats(t *testing.T) { }) } +// plannerHealthOverride reports a fixed planner verdict over a real store. +type plannerHealthOverride struct { + *store_sqlite.Store + health graph.PlannerStatsFreshness +} + +func (s plannerHealthOverride) PlannerStatsHealth(context.Context) (graph.PlannerStatsFreshness, error) { + return s.health, nil +} + // A backend without the capability must omit the key rather than publish a // struct of zeros, which would read as "the planner believes nothing". func TestIndexHealth_PlannerStatsAbsentOnBackendWithoutCapability(t *testing.T) { @@ -314,3 +348,26 @@ func TestIndexHealth_PlannerStatsAbsentOnBackendWithoutCapability(t *testing.T) assert.NotContains(t, payload, "planner_stats", "the in-memory graph has no planner statistics; reporting zeros would invent a defect") } + +func TestIndexHealthReportsLastFirstIndexDrain(t *testing.T) { + t.Setenv("GORTEX_STREAMING_FLUSH", "0") + t.Setenv("GORTEX_SHADOW_MAX_FILES", "1000000") + t.Setenv("GORTEX_SHADOW_MAX_BYTES", "1073741824") + srv, _ := newSQLiteBackedServer(t) + payload, err := srv.buildIndexHealthPayloadCtx(t.Context()) + require.NoError(t, err) + observations, ok := payload["last_first_index_drain"].(map[string]graph.FirstIndexDrain) + require.True(t, ok, "a completed first index must publish its actual drain observation") + require.Len(t, observations, 1) + for _, observation := range observations { + assert.Equal(t, "cold", observation.Window) + assert.Greater(t, observation.Nodes, 0) + assert.GreaterOrEqual(t, observation.DrainMS, observation.SealMS) + assert.False(t, observation.CompletedAt.IsZero()) + assert.Equal(t, "shared_store_drain_chunk_samples", observation.WALPeakScope) + } + empty, _ := newEmptySQLiteBackedServer(t) + noDrain, err := empty.buildIndexHealthPayloadCtx(t.Context()) + require.NoError(t, err) + assert.NotContains(t, noDrain, "last_first_index_drain", "absence is not a fabricated completed zero drain") +} diff --git a/internal/mcp/tools_enhancements.go b/internal/mcp/tools_enhancements.go index 65586d88f..5132d62ee 100644 --- a/internal/mcp/tools_enhancements.go +++ b/internal/mcp/tools_enhancements.go @@ -3698,6 +3698,19 @@ func (s *Server) buildIndexHealthBasePayloadCtx(ctx context.Context) (map[string plannerStats["last_refresh_at"] = planner.LastRefreshAt.UTC().Format(time.RFC3339) plannerStats["last_refresh_reason"] = planner.LastRefreshReason } + // Named, not counted: they are what an operator checks the schema for, + // and the reason above alone does not say which lookups are scanning. + if len(planner.DenseIndexesMissing) > 0 { + plannerStats["dense_indexes_missing"] = planner.DenseIndexesMissing + msg := "A bulk index load ended without rebuilding these SQLite indexes: " + strings.Join(planner.DenseIndexesMissing, ", ") + + ". Queries that use them scan instead until they are back. The store rebuilds them in the background once " + + "its volume has free space at least the size of the store, and a daemon restart rebuilds them too." + if recommendation == "" { + recommendation = msg + } else { + recommendation = msg + " " + recommendation + } + } result["planner_stats"] = plannerStats if planner.Stale { msg := "SQLite planner statistics are stale (" + planner.Reason + "): the graph has grown past what the query " + @@ -3710,6 +3723,13 @@ func (s *Server) buildIndexHealthBasePayloadCtx(ctx context.Context) (map[string } } } + if reporter, ok := s.graph.(interface { + LastFirstIndexDrains() map[string]graph.FirstIndexDrain + }); ok { + if observations := reporter.LastFirstIndexDrains(); len(observations) > 0 { + result["last_first_index_drain"] = observations + } + } if prefixAuditOK { ownership := map[string]any{ "owned_code_nodes": prefixAudit.OwnedCodeNodes,