From bcbfdca75533f0ad30a35c1f2e6cc325dd1900f6 Mon Sep 17 00:00:00 2001 From: Amit Kumar Date: Thu, 30 Apr 2026 08:29:10 +0000 Subject: [PATCH] feat(storage): vectordb persistence + FTS5 opt-in + 24h search clamp MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Storage rebalance initiative: shrink hot-path disk footprint at 150 services × 7 days while keeping triage latency intact. Five logically-distinct phases implemented in one cohesive change because they share files and only make sense together. PR1 — FTS5 disable + reclaim + 24h clamp on log search * SQLite FTS5 (logs_fts + porter+unicode61 + bm25 + sync triggers) is now opt-in via LOG_FTS_ENABLED. Default is off. Saves 30-40% disk at scale because the FTS table doubles every indexed log body. * POST /api/admin/drop_fts: refuses while flag is on; otherwise drops the virtual table + triggers and runs VACUUM to reclaim disk. * search_logs MCP tool + /api/logs?q= clamped to last 24h via storage.ClampSearchWindowTo24h to bound the LIKE-fallback worst case. PR2 — vectordb snapshot wire format * Versioned wire format: VDB1 magic + uint32 version + CRC32-IEEE + gob payload. Version/magic mismatch or CRC failure rejects the file so the loader can fall back cleanly to a full DB rebuild. * Atomic write via tmp+sync+rename, mode 0o600. EXDEV fallback for cross-device data dirs. * LogVector.Vec exported for gob; Index gains lastIndexedID watermark. PR3 — DB tail-replay * vectordb.ReplaySource interface + ReplayRow struct (no storage import in vectordb). * Repository.LogsForVectorReplay: WHERE id > ? AND severity IN (ERROR/WARN/...) ORDER BY id ASC, paged. * Index.ReplayFromDB walks pages from LastIndexedID() so a snapshot plus tail replay reconstructs the post-snapshot delta with no double-indexing. * main.go: vectorReplayAdapter projects storage.Log → vectordb.ReplayRow so the two packages stay decoupled. PR4 — periodic snapshot loop + shutdown hook * Index.SnapshotLoop(ctx, path, interval): periodic ticks plus one final write on ctx.Done so SIGTERM captures the maximum in-memory state. * Shutdown step 3a: snapCancel + 5s wait on snapDone right after graphRAG.Stop() so the final snapshot lands before DB close. PR5 — observability + docs * 5 metrics: otelcontext_vectordb_snapshot_writes_total{result}, snapshot_duration_seconds, snapshot_size_bytes, snapshot_load_total{result}, replay_logs_total. * Index.SetSnapshotObserver: wiring-layer hook so vectordb stays free of telemetry imports. main.go wires metrics.RecordVectorSnapshotWrite plus load/replay reporting. * CLAUDE.md updated: storage-architecture row for vectordb persistence, FTS5 reframed as opt-in with /api/admin/drop_fts reclaim, 24h clamp on search_logs, new env vars (VECTOR_INDEX_SNAPSHOT_PATH, VECTOR_INDEX_SNAPSHOT_INTERVAL, LOG_FTS_ENABLED). Verification * go vet ./... clean * go test ./... — 516 pass across 27 packages * go build . — 43.3 MB binary Plan: docs/superpowers/specs/2026-04-30-storage-rebalance-plan.md Co-Authored-By: Claude Opus 4.7 (1M context) --- AGENTS.md | 2 +- CLAUDE.md | 9 +- .../2026-04-30-storage-rebalance-plan.md | 241 +++++++++++++ internal/api/admin_handlers.go | 50 +++ internal/api/admin_handlers_test.go | 97 ++++++ internal/api/log_handlers.go | 13 + internal/api/log_handlers_cap_test.go | 74 ++++ internal/api/server.go | 1 + internal/config/config.go | 27 +- internal/mcp/tools.go | 18 +- internal/storage/factory.go | 13 +- internal/storage/fts5.go | 64 +++- internal/storage/fts5_test.go | 53 ++- internal/storage/log_repo.go | 28 ++ internal/storage/log_repo_replay_test.go | 138 ++++++++ internal/storage/search_cap.go | 51 +++ internal/storage/search_cap_test.go | 106 ++++++ internal/storage/testhelpers_test.go | 5 + internal/telemetry/metrics.go | 75 ++++ internal/vectordb/index.go | 56 ++- internal/vectordb/replay.go | 74 ++++ internal/vectordb/replay_test.go | 161 +++++++++ internal/vectordb/snapshot.go | 317 +++++++++++++++++ internal/vectordb/snapshot_test.go | 325 ++++++++++++++++++ main.go | 103 +++++- 25 files changed, 2060 insertions(+), 41 deletions(-) create mode 100644 docs/superpowers/specs/2026-04-30-storage-rebalance-plan.md create mode 100644 internal/api/admin_handlers_test.go create mode 100644 internal/api/log_handlers_cap_test.go create mode 100644 internal/storage/log_repo_replay_test.go create mode 100644 internal/storage/search_cap.go create mode 100644 internal/storage/search_cap_test.go create mode 100644 internal/vectordb/replay.go create mode 100644 internal/vectordb/replay_test.go create mode 100644 internal/vectordb/snapshot.go create mode 100644 internal/vectordb/snapshot_test.go diff --git a/AGENTS.md b/AGENTS.md index efa389f..3d8f75c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,7 +1,7 @@ # Memory Context -# [otelcontext] recent context, 2026-04-28 6:43am UTC +# [otelcontext] recent context, 2026-04-30 4:05am UTC No previous sessions found. \ No newline at end of file diff --git a/CLAUDE.md b/CLAUDE.md index 04658c9..41dc2fd 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -59,8 +59,8 @@ When none are present, `DEFAULT_TENANT` (default `"default"`) is assigned. Every | GraphRAG (in-memory) | `internal/graphrag/` | Layered graph: 4 typed stores, error chains, root cause analysis, anomaly detection | | Time Series (in-memory) | `internal/tsdb/` | Ring buffer, sliding windows, pre-computed percentiles | | Graph (in-memory, legacy) | `internal/graph/` | Simple service topology — **being replaced by GraphRAG** | -| Vector (embedded) | `internal/vectordb/` | TF-IDF index for semantic log search (pure Go, no CGO). Retained as a fallback similarity index for SQLite mode and for `SimilarErrors` ranking within a Drain template cluster. | -| Relational (persistent) | `internal/storage/` | GORM-based, multi-DB, single source of truth. Driven by `RetentionScheduler` (hourly batched purge + daily VACUUM/ANALYZE). `logs.body` is plain TEXT. **Log search**: SQLite uses FTS5 virtual table `logs_fts` (porter+unicode61 tokenizer) ordered by `bm25()`, kept in sync via AFTER INSERT/DELETE/UPDATE triggers; Postgres uses `pg_trgm` GIN on `logs.body` and `logs.service_name`. `AttributesJSON` and `AIInsight` remain `CompressedText`. | +| Vector (embedded) | `internal/vectordb/` | TF-IDF index for semantic log search (pure Go, no CGO). Persisted across restarts via gob+CRC32 snapshot (default `data/vectordb.snapshot`, 5m interval) plus a startup tail-replay from the DB so the index is warm before listeners accept traffic — eliminating the legacy minutes of cold-start blindness. `find_similar_logs` and `SimilarErrors` (within a Drain template cluster) are the read-side consumers. | +| Relational (persistent) | `internal/storage/` | GORM-based, multi-DB, single source of truth. Driven by `RetentionScheduler` (hourly batched purge + daily VACUUM/ANALYZE). `logs.body` is plain TEXT. **Log search**: vectordb (TF-IDF) is the default semantic-search path. Optional SQLite FTS5 (`logs_fts`, porter+unicode61, ordered by `bm25()`, AFTER INSERT/DELETE/UPDATE triggers) is **opt-in via `LOG_FTS_ENABLED=true`** and disabled by default — operators who toggle it off can reclaim the FTS table + indexes via `POST /api/admin/drop_fts`. Postgres uses `pg_trgm` GIN on `logs.body` and `logs.service_name`. `AttributesJSON` and `AIInsight` remain `CompressedText`. The `search_logs` MCP tool and the API `/api/logs?q=…` filter are clamped to the **last 24 hours** to bound the LIKE-fallback worst case. | ## GraphRAG Architecture @@ -190,7 +190,7 @@ internal/ storage/ # GORM repository, models, migrations, Close() method telemetry/ # Prometheus metrics + health (19 metrics) tsdb/ # Time series aggregator + ring buffer (lock-free Windows()) - vectordb/ # Embedded TF-IDF vector index (FIFO eviction with copy, clean IDF rebuild) + vectordb/ # Embedded TF-IDF vector index (FIFO eviction with copy, clean IDF rebuild). Persisted via gob+CRC32 snapshot + startup DB tail-replay (snapshot.go, replay.go). ui/ # Embedded React frontend ui/ # React frontend (Vite + Mantine) test/ # Microservice simulation (7 services) @@ -213,7 +213,8 @@ Key settings in `internal/config/config.go`: - `METRIC_MAX_CARDINALITY` (10000), `METRIC_MAX_CARDINALITY_PER_TENANT` (0 = unlimited), `API_RATE_LIMIT_RPS` (100). The per-tenant cap is checked first; when set, a noisy tenant cannot exhaust the global pool. Overflow is labeled by tenant via `otelcontext_tsdb_cardinality_overflow_by_tenant_total{tenant_id}` (`__global__` sentinel when the global cap was the trigger). - `MCP_ENABLED` (true), `MCP_PATH` (/mcp) - `MCP_MAX_CONCURRENT` (32), `MCP_CALL_TIMEOUT_MS` (30000), `MCP_CACHE_TTL_MS` (5000) — MCP HTTP streamable robustness. Counting semaphore gates concurrent `tools/call` (JSON-RPC `-32000` past the cap), per-call deadlines abort runaway handlers (JSON-RPC `-32001`), and a 5s TTL cache memoizes the cheap in-memory GraphRAG tools (`get_service_map`, `impact_analysis`, `root_cause_analysis`, `get_anomaly_timeline`, `get_service_health`). SSE GET sends a `: keep-alive\n\n` comment every 25s to keep the stream alive across reverse-proxy idle timeouts. Set any to 0 to disable. -- `VECTOR_INDEX_MAX_ENTRIES` (100000) +- `VECTOR_INDEX_MAX_ENTRIES` (100000), `VECTOR_INDEX_SNAPSHOT_PATH` (`data/vectordb.snapshot`), `VECTOR_INDEX_SNAPSHOT_INTERVAL` (`5m`) — vectordb persistence. Empty `VECTOR_INDEX_SNAPSHOT_PATH` or non-positive interval disables the snapshot loop. The snapshot file uses a magic+version+CRC32 wire format with gob payload; corrupt or version-mismatched files are rejected and the loader falls back to a full DB rebuild via `ReplayFromDB`. Watch `otelcontext_vectordb_snapshot_writes_total{result}`, `otelcontext_vectordb_snapshot_load_total{result}`, `otelcontext_vectordb_snapshot_size_bytes`, and `otelcontext_vectordb_replay_logs_total`. +- `LOG_FTS_ENABLED` (false) — when truthy (`true`/`yes`/`on`/`1`), provisions the SQLite FTS5 `logs_fts` virtual table + sync triggers at startup; when false, log-search uses vectordb (semantic) plus a 24h-clamped LIKE fallback. Toggle off and reclaim disk via `POST /api/admin/drop_fts` (refused while the flag is on). - `DLQ_MAX_FILES` (1000), `DLQ_MAX_DISK_MB` (500), `DLQ_MAX_RETRIES` (10) - `GRAPHRAG_WORKER_COUNT` (16), `GRAPHRAG_EVENT_QUEUE_SIZE` (100000) — sized for 100–200 services; raise further if `otelcontext_graphrag_events_dropped_total` climbs - `INGEST_ASYNC_ENABLED` (true), `INGEST_PIPELINE_QUEUE_SIZE` (50000), `INGEST_PIPELINE_WORKERS` (8) — async ingest pipeline (`internal/ingest/pipeline.go`). Hybrid backpressure: <90% accept all, 90–100% drop healthy batches (errors/slow always pass), 100% return gRPC `RESOURCE_EXHAUSTED`. Set `INGEST_ASYNC_ENABLED=false` to revert to synchronous DB writes inside `Export()`. Drops surface as `otelcontext_ingest_pipeline_dropped_total{signal,reason}`. diff --git a/docs/superpowers/specs/2026-04-30-storage-rebalance-plan.md b/docs/superpowers/specs/2026-04-30-storage-rebalance-plan.md new file mode 100644 index 0000000..72e0624 --- /dev/null +++ b/docs/superpowers/specs/2026-04-30-storage-rebalance-plan.md @@ -0,0 +1,241 @@ +# Plan — Storage rebalance: drop FTS5, persist vectordb, cap log search + +**Status:** Approved scope, ready for implementation +**Date:** 2026-04-30 +**Reviewers:** codex (vectordb persistence design, 2026-04-30); user (scope sign-off, 2026-04-30) + +## Problem + +Three coupled inefficiencies surfaced while sizing for the 150-service × 7-day production target on SQLite: + +1. **FTS5 inverted index consumes 30-40% of DB disk.** At the current 4GB sample, ~1.5GB; at full production scale, tens of GB. The UI never queries it (verified: `ui/src/hooks/useLogs.ts:21` calls `/api/logs` with no `q`); only the `search_logs` MCP tool depends on it, and that has a working LIKE fallback already wired at `internal/storage/log_repo.go:105`. +2. **vectordb is not persisted.** Daemon restart empties the in-memory TF-IDF index. `find_similar_logs` and GraphRAG `SimilarErrors` return degraded results until ERROR/WARN logs flow back in naturally — minutes to hours, depending on error rate. +3. **`search_logs` has no time-range cap.** Tolerable with FTS5; without it, an unscoped 7-day keyword query is a 30+ minute table scan. Worst case must be bounded. + +## Goal + +Cut SQLite footprint by 30-40%, eliminate vectordb cold-start blindness, and bound `search_logs` worst case to single-digit seconds — without changing `find_similar_logs`, GraphRAG, or any non-search MCP tool surface. + +## Non-goals + +- Inverted index / sublinear cosine search (requires raising 100k cap, separate change) +- Dense embeddings / ONNX +- Per-tenant snapshot sharding, multi-process flock +- Raising `VECTOR_INDEX_MAX_ENTRIES` +- Removing `search_logs` (kept, degraded via LIKE) +- Changes to Postgres `pg_trgm` path (SQLite-only) + +--- + +## Phase 0a — `LOG_FTS_ENABLED` config flag + +Default `false`. New SQLite deploys skip FTS5 entirely. + +**Files:** +- `internal/config/config.go` — add `LogFTSEnabled bool`, env `LOG_FTS_ENABLED`, default false +- `internal/storage/fts5.go` — `fts5Available()` returns false when flag is off; `EnsureLogsFTS5()` becomes no-op +- `internal/storage/log_repo.go` — Repository gains `ftsEnabled bool`; `useFTS5` checks both driver + flag +- `internal/storage/factory.go` (or wherever `NewRepository` is) — thread `cfg.LogFTSEnabled` through +- `CLAUDE.md` — update FTS5 description, note opt-in + +LIKE fallback already exists at `log_repo.go:105-108`; flipping the flag silently routes `filter.Search` queries through it. No new fallback code. + +## Phase 0b — One-shot FTS5 reclaim + +For existing DBs with `logs_fts`: + +- New admin endpoint `POST /api/admin/drop_fts` (auth-gated, alongside existing `/api/admin/purge` + `/api/admin/vacuum`) +- Body: `DROP TRIGGER IF EXISTS logs_fts_ai; logs_fts_ad; logs_fts_au; DROP TABLE IF EXISTS logs_fts; VACUUM;` +- Returns `{"reclaimed_bytes": N, "elapsed_ms": M}` +- Refuses (405) if `LOG_FTS_ENABLED=true` — won't drop while in active use + +**Files:** +- `internal/api/admin_handlers.go` (or wherever `handlePurge`/`handleVacuum` live) — add `handleDropFTS` +- `internal/api/server.go:108-109` — register route +- Test: integration test creates `logs_fts`, calls handler, asserts gone + bytes reclaimed + +VACUUM blocks writes 10-60min on a 4GB DB — opt-in only, document for off-hours. + +## Phase 0c — 24h cap on `search_logs` + +**Description update at `internal/mcp/tools.go` ~line 37:** + +> "Search log bodies by keyword. Limited to the last 24 hours. Strongly recommend setting `service_name` and/or `severity` to scope the search; unscoped keyword queries scan large row counts without FTS5. Returns up to `limit` results ordered by timestamp desc." + +**Helper at `internal/mcp/tools.go` (or new `internal/mcp/clamp.go`):** + +```go +func clampTo24h(start, end, now time.Time) (time.Time, time.Time, error) { + if end.IsZero() { end = now } + if start.IsZero() { start = end.Add(-24 * time.Hour) } + if end.After(now) { end = now } + cutoff := now.Add(-24 * time.Hour) + if end.Before(cutoff) { return time.Time{}, time.Time{}, errors.New("search window must be within the last 24h") } + if start.Before(cutoff) { start = cutoff } + if !start.Before(end) { return time.Time{}, time.Time{}, errors.New("start_time must be before end_time") } + return start, end, nil +} +``` + +**Apply at:** +- `internal/mcp/tools.go::toolSearchLogs` (~line 445) — after arg parse +- `internal/api/log_handlers.go::handleGetLogs` — when `q` param non-empty (defense in depth; UI doesn't hit this with `q` but direct HTTP callers shouldn't bypass) + +**Tests:** 4 cases for `clampTo24h` (defaults, oversize window clamped, out-of-cap rejected, invalid range rejected) + 1 integration test each for MCP and HTTP. + +**Behavior matrix:** + +| Input | Effective window | +|---|---| +| nothing | now-24h → now | +| start = 5d ago | now-24h → now (start clamped) | +| start = 5d ago, end = 4d ago | rejected | +| end = future | start → now (end clamped) | +| `q=""` (no search) | unconstrained — cap only fires when search is set | + +--- + +## Phase 1-5 — vectordb persistence + +### Format + +``` +bytes[0:4] magic "VDB1" (ASCII) +bytes[4:8] format version uint32 BE +bytes[8:12] payload CRC32-IEEE uint32 BE (over bytes[12:]) +bytes[12:] gob payload Snapshot +``` + +```go +type Snapshot struct { + LastIndexedID uint // max Log.ID seen by Add() + MaxSize int + Docs []LogVector + IDF map[string]float64 + WrittenAt int64 // unix seconds +} +``` + +### Tail-replay key + +**Use `Log.ID` (auto-increment PK), not `Log.Timestamp`.** Codex caught: `Index.Add()` has no dedup; timestamp-based replay would double-index boundary entries on every restart. ID is monotonic, dedup-free, DB-agnostic. + +Query: `WHERE id > ? AND severity IN ('ERROR','WARN','WARNING','FATAL','CRITICAL') ORDER BY id ASC LIMIT 10000`. Paged loop until empty. + +### Atomic write + +`writeAtomic(path, data)`: write to `path+".tmp"` → fsync → close → rename. On `EXDEV` (cross-device): log warn once, fall back to `os.WriteFile` directly. On fsync error: delete `.tmp`, return error. + +### Lifecycle + +**Startup** (in `cmd/main.go` after DB open, before ingest accept): +```go +idx := vectordb.New(cfg.VectorIndexMaxEntries) +if cfg.VectorIndexSnapshotPath != "" { + _ = idx.LoadSnapshot(cfg.VectorIndexSnapshotPath) // tolerates missing/corrupt + _ = idx.ReplayFromDB(ctx, repo) // catches tail since LastIndexedID + go idx.SnapshotLoop(ctx, path, cfg.VectorIndexSnapshotInterval) +} +``` + +**Shutdown:** final `idx.WriteSnapshot()` between gRPC/HTTP `Shutdown()` returning and `graphrag.Stop()` — captures every `Add()` that completed before ingest stopped. + +### Phases + +- **Phase 1**: `internal/vectordb/snapshot.go` + tests — encode/decode, magic/version/CRC, atomic write, EXDEV fallback. Pure isolation, no behavior change. +- **Phase 2**: wire into `Index` — add `lastIndexedID` field tracked in `Add()`; methods `LoadSnapshot`, `WriteSnapshot`, `LastIndexedID()`. Config fields added but no goroutine yet. +- **Phase 3**: DB tail replay — `Repository.LogsForVectorReplay(ctx, sinceID, limit) ([]Log, error)` (paged); `Index.ReplayFromDB(ctx, repo)` walks pages calling `Add()`. +- **Phase 4**: `Index.SnapshotLoop(ctx, path, interval)` background goroutine; wire startup/shutdown in `cmd/main.go`. +- **Phase 5**: 5 Prometheus metrics (`snapshot_writes_total{result}`, `snapshot_duration_seconds`, `snapshot_size_bytes`, `snapshot_load_total{result}`, `replay_logs_total`); CLAUDE.md vectordb section + config table. + +--- + +## Files (combined) + +### New +- `internal/vectordb/snapshot.go` +- `internal/vectordb/snapshot_test.go` +- `internal/mcp/clamp.go` + `clamp_test.go` + +### Modified +- `internal/vectordb/index.go` — `lastIndexedID`, `LoadSnapshot`, `WriteSnapshot`, `SnapshotLoop`, `ReplayFromDB` +- `internal/storage/fts5.go` — gate on flag +- `internal/storage/log_repo.go` — Repository ftsEnabled field +- `internal/storage/repository.go` — `LogsForVectorReplay` +- `internal/storage/factory.go` — config wiring +- `internal/api/log_handlers.go` — 24h cap symmetry on `q=` +- `internal/api/admin_handlers.go` (or equivalent) — `handleDropFTS` +- `internal/api/server.go` — `/api/admin/drop_fts` route +- `internal/mcp/tools.go` — description update + `clampTo24h` call in `toolSearchLogs` +- `internal/config/config.go` — `LOG_FTS_ENABLED`, `VECTOR_INDEX_SNAPSHOT_PATH` (default `data/vectordb.snapshot`), `VECTOR_INDEX_SNAPSHOT_INTERVAL` (default `5m`) +- `cmd/main.go` (or `internal/ui/ui.go` — wherever vectordb is constructed) — startup/shutdown wiring +- `internal/telemetry/metrics.go` — 5 vectordb metrics +- `CLAUDE.md` — FTS5 description, vectordb persistence note, config table additions + +## Acceptance criteria + +### Phase 0 (FTS5 + cap) +1. `LOG_FTS_ENABLED=false` (default): new SQLite deploy has no `logs_fts` table; `search_logs` returns results via LIKE fallback. +2. `LOG_FTS_ENABLED=true`: existing FTS5 path works unchanged (regression test). +3. `POST /api/admin/drop_fts` reclaims expected bytes; returns 405 when flag is true. +4. `toolSearchLogs` with no times → defaults to last 24h. +5. start_time = 7d ago → clamped to now-24h. +6. Window entirely outside cap → rejected with explicit error. +7. HTTP `GET /api/logs?q=...` applies same cap. +8. Updated tool description visible in MCP `tools/list`. + +### Phase 1-5 (vectordb persistence) +1. Daemon restart → `find_similar_logs` returns useful results in <1s (vs hours pre-change). +2. Tail replay correctness: seed N → snapshot → seed M → restart → exactly N+M entries, no duplicate `LogID`. +3. `kill -9` between snapshots: previous snapshot loads, no `.tmp` left behind. +4. CRC mismatch / wrong magic / version mismatch: warning logged, full rebuild kicks in. +5. 5 new metrics present at `/metrics`. +6. Manual: oteliq daemon shows `vectordb: loaded N entries from snapshot, replayed M from DB`. + +## Risks + +| Risk | Mitigation | +|---|---| +| Flag flipped mid-deployment leaves stale `logs_fts` | Stale table is harmless until 0b dropped. | +| VACUUM on 4GB blocks writes 10-60min | Document; admin endpoint is opt-in, not automatic. | +| 24h cap rejects legitimate historical queries | Explicit error; direct DB access remains for one-offs. | +| LIKE fallback slower than expected at scale | Accepted trade-off. Escalate to v2 vectordb design if painful. | +| Snapshot format breaks across Go versions | Magic + version + CRC → full rebuild on any decode error. Bump version on `LogVector` struct change. | +| Cross-device rename loses atomicity | EXDEV detected, fall back to direct write. Constraint documented. | + +## Phasing for shipping + +5 PRs, in order: + +- **PR1**: Phase 0a + 0b + 0c (FTS5 disable + reclaim helper + 24h cap). Single PR — tightly related. +- **PR2**: Phase 1 + 2 (snapshot encode/decode + Index integration). +- **PR3**: Phase 3 (DB tail replay). +- **PR4**: Phase 4 (snapshot loop + shutdown hook). +- **PR5**: Phase 5 (metrics + docs). + +PR1 ships standalone. PR2-5 are sequential. Phase 0 can land before any vectordb work. + +## Verification + +```bash +# Phase 0 +go test ./internal/storage/... ./internal/mcp/... ./internal/api/... + +# Phase 1-5 +go test ./internal/vectordb/... + +# Integration smoke +go vet ./... +go build -o /tmp/otelcontext . +LOG_FTS_ENABLED=false /tmp/otelcontext & +# Expect: "logs_fts: disabled (LOG_FTS_ENABLED=false)" on first start +# After restart: "vectordb: loaded N entries from snapshot, replayed M from DB" +``` + +## Resolved scope decisions + +- Snapshot path: flat `data/vectordb.snapshot` +- zstd compression: not in v1 — ship uncompressed gob +- Phase split: 5 PRs (above) +- FTS5 default: off (`LOG_FTS_ENABLED=false`) +- 24h cap interpretation: search window must overlap with last 24h; window entirely older is rejected, not silently emptied diff --git a/internal/api/admin_handlers.go b/internal/api/admin_handlers.go index ace9be8..b4eb460 100644 --- a/internal/api/admin_handlers.go +++ b/internal/api/admin_handlers.go @@ -4,7 +4,9 @@ import ( "encoding/json" "log/slog" "net/http" + "os" "strconv" + "strings" "time" ) @@ -66,3 +68,51 @@ func (s *Server) handleVacuum(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]string{"status": "vacuumed"}) } + +// handleDropFTS handles POST /api/admin/drop_fts. Drops the SQLite FTS5 +// virtual table + AFTER INSERT/DELETE/UPDATE triggers and runs VACUUM so the +// freed pages are returned to the OS. One-shot reclaim for existing deploys +// after LOG_FTS_ENABLED is set to false — typically reclaims 30-40% of DB disk. +// +// Refused (405) when LOG_FTS_ENABLED is currently truthy, because triggers +// fire on every log INSERT and dropping them mid-flight would silently break +// FTS5 sync until restart. +// +// VACUUM blocks writes for ~10-60 minutes on a multi-GB DB. Run this during +// a maintenance window. +func (s *Server) handleDropFTS(w http.ResponseWriter, r *http.Request) { + if v, ok := os.LookupEnv("LOG_FTS_ENABLED"); ok { + if b, err := strconv.ParseBool(strings.TrimSpace(v)); err == nil && b { + http.Error(w, "drop_fts refused: LOG_FTS_ENABLED is currently true; set it to false and restart before dropping", http.StatusMethodNotAllowed) + return + } + switch strings.ToLower(strings.TrimSpace(v)) { + case "yes", "y", "on": + http.Error(w, "drop_fts refused: LOG_FTS_ENABLED is currently true; set it to false and restart before dropping", http.StatusMethodNotAllowed) + return + } + } + + started := time.Now() + sizeBefore := s.repo.HotDBSizeBytes() + + if err := s.repo.DropLogsFTS(r.Context()); err != nil { + slog.Error("drop_fts failed", "error", err) + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + sizeAfter := s.repo.HotDBSizeBytes() + elapsed := time.Since(started) + reclaimed := sizeBefore - sizeAfter + if reclaimed < 0 { + reclaimed = 0 + } + slog.Info("drop_fts completed", "elapsed_ms", elapsed.Milliseconds(), "reclaimed_bytes", reclaimed) + + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "reclaimed_bytes": reclaimed, + "elapsed_ms": elapsed.Milliseconds(), + }) +} diff --git a/internal/api/admin_handlers_test.go b/internal/api/admin_handlers_test.go new file mode 100644 index 0000000..12aa517 --- /dev/null +++ b/internal/api/admin_handlers_test.go @@ -0,0 +1,97 @@ +package api + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/RandomCodeSpace/otelcontext/internal/storage" +) + +// newAPITestRepoWithFTS builds a fresh in-memory SQLite Repository with the +// FTS5 schema provisioned. Caller must have already set LOG_FTS_ENABLED=true +// before calling — otherwise the migrate path skips the FTS5 setup. +func newAPITestRepoWithFTS(t *testing.T) *storage.Repository { + t.Helper() + db, err := storage.NewDatabase("sqlite", ":memory:") + if err != nil { + t.Fatalf("NewDatabase: %v", err) + } + if err := storage.AutoMigrateModels(db, "sqlite"); err != nil { + t.Fatalf("AutoMigrateModels: %v", err) + } + repo := storage.NewRepositoryFromDB(db, "sqlite") + t.Cleanup(func() { _ = repo.Close() }) + return repo +} + +// TestHandleDropFTS_Reclaims verifies that the admin endpoint drops the +// FTS5 virtual table + triggers and returns reclaimed_bytes >= 0. +// +// Test starts with LOG_FTS_ENABLED=true so the in-memory repo provisions +// the FTS5 schema, then flips the flag off so handleDropFTS will accept +// the call (it refuses while the flag is still on). +func TestHandleDropFTS_Reclaims(t *testing.T) { + t.Setenv("LOG_FTS_ENABLED", "true") + repo := newAPITestRepoWithFTS(t) + // Seed a few rows so the FTS5 index has content (and reclaimable pages). + for i := range 50 { + _ = repo.DB().Create(&storage.Log{ + TenantID: "default", + Severity: "ERROR", + Body: strings.Repeat("payload-", i+1), + ServiceName: "svc", + }) + } + // Switch off so the handler accepts the drop request. + t.Setenv("LOG_FTS_ENABLED", "false") + + srv := &Server{repo: repo} + mux := http.NewServeMux() + mux.HandleFunc("POST /api/admin/drop_fts", srv.handleDropFTS) + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/api/admin/drop_fts", nil) + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("want 200, got %d body=%q", rec.Code, rec.Body.String()) + } + var body map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("decode: %v body=%q", err, rec.Body.String()) + } + if _, ok := body["reclaimed_bytes"]; !ok { + t.Fatalf("response missing reclaimed_bytes: %v", body) + } + // Verify the FTS5 table is gone — querying it must error. + if err := repo.DB().Exec("SELECT 1 FROM logs_fts LIMIT 1").Error; err == nil { + t.Fatal("logs_fts should be dropped but still queryable") + } +} + +// TestHandleDropFTS_RefusesWhenEnabled verifies the safety gate: the +// endpoint refuses (405) if LOG_FTS_ENABLED is currently truthy, because +// dropping the triggers mid-operation would silently break FTS5 sync. +func TestHandleDropFTS_RefusesWhenEnabled(t *testing.T) { + t.Setenv("LOG_FTS_ENABLED", "true") + repo := newAPITestRepoWithFTS(t) + + srv := &Server{repo: repo} + mux := http.NewServeMux() + mux.HandleFunc("POST /api/admin/drop_fts", srv.handleDropFTS) + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/api/admin/drop_fts", nil) + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusMethodNotAllowed { + t.Fatalf("want 405 with LOG_FTS_ENABLED=true, got %d body=%q", rec.Code, rec.Body.String()) + } + // FTS5 table must still be present. + if err := repo.DB().Exec("SELECT 1 FROM logs_fts LIMIT 1").Error; err != nil { + t.Fatalf("logs_fts should remain queryable when refused: %v", err) + } +} diff --git a/internal/api/log_handlers.go b/internal/api/log_handlers.go index dd17f53..55ba4c9 100644 --- a/internal/api/log_handlers.go +++ b/internal/api/log_handlers.go @@ -47,6 +47,19 @@ func (s *Server) handleGetLogs(w http.ResponseWriter, r *http.Request) { } } + // When the caller is doing a body keyword search, enforce the same 24h + // cap as the MCP search_logs tool so a direct HTTP caller cannot bypass + // via the alternate transport. Pure filtered listings (no search term) + // keep the full retention range. + if filter.Search != "" { + cs, ce, err := storage.ClampSearchWindowTo24h(filter.StartTime, filter.EndTime, time.Now()) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + filter.StartTime, filter.EndTime = cs, ce + } + logs, total, err := s.repo.GetLogsV2(r.Context(), filter) if err != nil { slog.Error("Failed to get logs", "error", err) diff --git a/internal/api/log_handlers_cap_test.go b/internal/api/log_handlers_cap_test.go new file mode 100644 index 0000000..7a14401 --- /dev/null +++ b/internal/api/log_handlers_cap_test.go @@ -0,0 +1,74 @@ +package api + +import ( + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/RandomCodeSpace/otelcontext/internal/storage" +) + +// TestHandleGetLogs_SearchCapRejectsOlderThan24h verifies that an explicit +// search query with a window entirely outside the 24h cap returns 400. +// Symmetric with the MCP search_logs cap so a direct HTTP caller cannot +// bypass via the alternate transport. +func TestHandleGetLogs_SearchCapRejectsOlderThan24h(t *testing.T) { + repo := newAPITestRepoWithoutFTS(t) + srv := &Server{repo: repo} + mux := http.NewServeMux() + mux.HandleFunc("GET /api/logs", srv.handleGetLogs) + + q := url.Values{} + q.Set("search", "panic") + q.Set("start", time.Now().Add(-5*24*time.Hour).Format(time.RFC3339)) + q.Set("end", time.Now().Add(-4*24*time.Hour).Format(time.RFC3339)) + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/api/logs?"+q.Encode(), nil) + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("want 400 for out-of-cap window, got %d body=%q", rec.Code, rec.Body.String()) + } +} + +// TestHandleGetLogs_NoSearchSkipsCap verifies that a filtered listing with +// no search term keeps the full retention range — the cap fires only on +// keyword queries, where unbounded LIKE scans are the worst case. +func TestHandleGetLogs_NoSearchSkipsCap(t *testing.T) { + repo := newAPITestRepoWithoutFTS(t) + srv := &Server{repo: repo} + mux := http.NewServeMux() + mux.HandleFunc("GET /api/logs", srv.handleGetLogs) + + q := url.Values{} + // No search param — listing-only request with a 5-day-old window. + q.Set("start", time.Now().Add(-5*24*time.Hour).Format(time.RFC3339)) + q.Set("end", time.Now().Add(-4*24*time.Hour).Format(time.RFC3339)) + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/api/logs?"+q.Encode(), nil) + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("filtered listing without search must succeed, got %d body=%q", rec.Code, rec.Body.String()) + } +} + +// newAPITestRepoWithoutFTS builds a fresh in-memory repo with FTS5 disabled. +// Used by cap tests since they only care about handler behavior, not the +// search backend. +func newAPITestRepoWithoutFTS(t *testing.T) *storage.Repository { + t.Helper() + t.Setenv("LOG_FTS_ENABLED", "false") + db, err := storage.NewDatabase("sqlite", ":memory:") + if err != nil { + t.Fatalf("NewDatabase: %v", err) + } + if err := storage.AutoMigrateModels(db, "sqlite"); err != nil { + t.Fatalf("AutoMigrateModels: %v", err) + } + repo := storage.NewRepositoryFromDB(db, "sqlite") + t.Cleanup(func() { _ = repo.Close() }) + return repo +} diff --git a/internal/api/server.go b/internal/api/server.go index 338fb2c..4cd6318 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -107,6 +107,7 @@ func (s *Server) RegisterRoutes(mux *http.ServeMux) { mux.Handle("GET /metrics/prometheus", telemetry.PrometheusHandler()) mux.HandleFunc("DELETE /api/admin/purge", s.handlePurge) mux.HandleFunc("POST /api/admin/vacuum", s.handleVacuum) + mux.HandleFunc("POST /api/admin/drop_fts", s.handleDropFTS) // WebSockets mux.HandleFunc("/ws", s.hub.HandleWebSocket) diff --git a/internal/config/config.go b/internal/config/config.go index f16998c..18a7047 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -109,6 +109,26 @@ type Config struct { // Vector Index VectorIndexMaxEntries int + // VectorIndexSnapshotPath is the on-disk location for periodic vectordb + // snapshots. When empty, persistence is disabled and the index rebuilds + // from DB on every restart (legacy behaviour). Default + // "data/vectordb.snapshot". + VectorIndexSnapshotPath string + + // VectorIndexSnapshotInterval, e.g. "5m". When set and + // VectorIndexSnapshotPath is non-empty, the index serializes its state + // to disk on this cadence. "0" / empty disables periodic writes (a + // final snapshot still fires on graceful shutdown). Default "5m". + VectorIndexSnapshotInterval string + + // LogFTSEnabled toggles SQLite FTS5 provisioning + querying. The FTS5 + // inverted index typically consumes 30-40% of SQLite DB disk for + // log-heavy workloads, while the LIKE fallback (log_repo.go:105) keeps + // search_logs functional without it. Default false; opt in with + // LOG_FTS_ENABLED=true. Only meaningful on SQLite; Postgres uses pg_trgm + // independently of this flag. + LogFTSEnabled bool + // GraphRAG worker count (background consumers of the ingestion event channel). // Defaults to 4 if unset or <=0. Increase under sustained high ingest. GraphRAGWorkerCount int @@ -274,7 +294,12 @@ func Load(customPath string) (*Config, error) { CompressionLevel: getEnv("COMPRESSION_LEVEL", "default"), // Vector - VectorIndexMaxEntries: getEnvInt("VECTOR_INDEX_MAX_ENTRIES", 100000), + VectorIndexMaxEntries: getEnvInt("VECTOR_INDEX_MAX_ENTRIES", 100000), + VectorIndexSnapshotPath: getEnv("VECTOR_INDEX_SNAPSHOT_PATH", "data/vectordb.snapshot"), + VectorIndexSnapshotInterval: getEnv("VECTOR_INDEX_SNAPSHOT_INTERVAL", "5m"), + + // Log search FTS5 toggle (SQLite only). Default off — see field comment. + LogFTSEnabled: parseTruthy(getEnv("LOG_FTS_ENABLED", "")), // GraphRAG GraphRAGWorkerCount: getEnvInt("GRAPHRAG_WORKER_COUNT", 16), diff --git a/internal/mcp/tools.go b/internal/mcp/tools.go index e3e7283..6385566 100644 --- a/internal/mcp/tools.go +++ b/internal/mcp/tools.go @@ -35,7 +35,7 @@ var toolDefs = []Tool{ }, { Name: "search_logs", - Description: "Searches log entries by severity, service, body text, trace ID, and time range. Returns id, timestamp, severity, service_name, body, trace_id. Default window is last 24h. Use severity=ERROR to find errors, query= for full-text search, trace_id= to correlate with a trace. Use page= for pagination.", + Description: "Searches log entries by severity, service, body text, trace ID, and time range. Returns id, timestamp, severity, service_name, body, trace_id. **Limited to the last 24 hours** — windows entirely outside the 24h cap are rejected. Strongly recommend setting `service` and/or `severity` to scope the search; unscoped keyword queries scan large row counts when FTS5 is disabled (the default). Use severity=ERROR to find errors, query= for full-text search, trace_id= to correlate with a trace. Use page= for pagination.", InputSchema: InputSchema{ Type: "object", Properties: map[string]Property{ @@ -43,8 +43,8 @@ var toolDefs = []Tool{ "severity": {Type: "string", Description: "Filter by severity level: ERROR, WARN, INFO, DEBUG."}, "service": {Type: "string", Description: "Filter by service name (exact match)."}, "trace_id": {Type: "string", Description: "Filter logs belonging to a specific trace ID."}, - "start": {Type: "string", Description: "Start time RFC3339. Defaults to 24h ago."}, - "end": {Type: "string", Description: "End time RFC3339. Defaults to now."}, + "start": {Type: "string", Description: "Start time RFC3339. Defaults to 24h ago. Cannot be earlier than now-24h; older values are clamped."}, + "end": {Type: "string", Description: "End time RFC3339. Defaults to now. Cannot exceed now; future values are clamped."}, "limit": {Type: "number", Description: "Max results per page (default 50, max 200)."}, "page": {Type: "number", Description: "Page number for pagination (default 0)."}, }, @@ -431,11 +431,19 @@ func toLogSummaries(logs []storage.Log) []logSummary { } func (s *Server) toolSearchLogs(ctx context.Context, args map[string]any) ToolCallResult { - end := time.Now() - start := end.Add(-24 * time.Hour) // wider default window for AI agents + var start, end time.Time parseTime(args, "start", &start) parseTime(args, "end", &end) + // Enforce the 24h cap centrally so an MCP caller cannot bypass via the + // alternate HTTP transport. clampTo24h handles defaults (zero values) and + // returns a clean error when the window is entirely older than the cap. + clampedStart, clampedEnd, capErr := storage.ClampSearchWindowTo24h(start, end, time.Now()) + if capErr != nil { + return errorResult(capErr.Error()) + } + start, end = clampedStart, clampedEnd + limit := argInt(args, "limit", 50) if limit > 200 { limit = 200 diff --git a/internal/storage/factory.go b/internal/storage/factory.go index 52f2b54..8ba94a9 100644 --- a/internal/storage/factory.go +++ b/internal/storage/factory.go @@ -296,9 +296,18 @@ func AutoMigrateModelsWithOptions(db *gorm.DB, driver string, opts MigrateOption // SQLite: provision FTS5 virtual table + triggers on logs.body / logs.service_name. // Search routes through bm25() ranking on this driver; LIKE remains the fallback // if FTS5 is unavailable (older SQLite builds without FTS5 compiled in). + // + // Gated on LOG_FTS_ENABLED (default false) — FTS5's inverted index typically + // consumes 30-40% of SQLite DB disk for log-heavy workloads. When disabled, + // search_logs falls back transparently to LIKE via the existing branch in + // log_repo.go:105. if driver == "sqlite" || driver == "" { - if err := setupSQLiteFTS5(db); err != nil { - log.Printf("⚠️ SQLite FTS5 setup failed (%v) — log search will fall back to LIKE", err) + if logFTSEnabledFromEnv() { + if err := setupSQLiteFTS5(db); err != nil { + log.Printf("⚠️ SQLite FTS5 setup failed (%v) — log search will fall back to LIKE", err) + } + } else { + log.Println("ℹ️ SQLite FTS5 disabled (LOG_FTS_ENABLED=false) — log search uses LIKE") } } diff --git a/internal/storage/fts5.go b/internal/storage/fts5.go index 96cc56e..c6b50c6 100644 --- a/internal/storage/fts5.go +++ b/internal/storage/fts5.go @@ -1,8 +1,12 @@ package storage import ( + "context" "fmt" "log" + "log/slog" + "os" + "strconv" "strings" "gorm.io/gorm" @@ -111,9 +115,61 @@ func fts5MatchExpr(input string) string { return strings.Join(parts, " ") } -// fts5Available reports whether the given driver should use the FTS5 path. We -// only enable FTS5 on SQLite because Postgres has its own pg_trgm GIN path -// (see factory.go) and MySQL/SQL Server are out of scope. +// fts5Available reports whether the given driver should use the FTS5 path. +// FTS5 is only enabled when (a) the driver is SQLite (Postgres has its own +// pg_trgm GIN path; MySQL/SQL Server are out of scope) and (b) LOG_FTS_ENABLED +// is truthy. Default off — FTS5's inverted index typically consumes 30-40% of +// SQLite DB disk for log-heavy workloads, and the LIKE fallback at +// log_repo.go:105 keeps search_logs functional without it. func fts5Available(driver string) bool { - return strings.ToLower(driver) == "sqlite" + if !strings.EqualFold(driver, "sqlite") { + return false + } + return logFTSEnabledFromEnv() +} + +// logFTSEnabledFromEnv reads LOG_FTS_ENABLED and reports whether the FTS5 +// virtual table + triggers should be installed and queried. Defaults to +// false; opt in with LOG_FTS_ENABLED=true (also accepts yes/on/1). +func logFTSEnabledFromEnv() bool { + v, ok := os.LookupEnv("LOG_FTS_ENABLED") + if !ok { + return false + } + if b, err := strconv.ParseBool(strings.TrimSpace(v)); err == nil { + return b + } + switch strings.ToLower(strings.TrimSpace(v)) { + case "yes", "y", "on": + return true + } + return false +} + +// DropLogsFTS removes the FTS5 virtual table and its sync triggers, then runs +// VACUUM to reclaim freed pages. Used by /api/admin/drop_fts on existing +// SQLite deployments after LOG_FTS_ENABLED has been set to false, to recover +// the 30-40% of DB disk the inverted index occupied. +// +// VACUUM blocks writes for ~10-60 minutes on a multi-GB DB and cannot run +// inside a transaction. Idempotent — safe to call when the FTS5 table or +// triggers are already absent. +func (r *Repository) DropLogsFTS(ctx context.Context) error { + if !strings.EqualFold(r.driver, "sqlite") { + return fmt.Errorf("DropLogsFTS only supported on SQLite, got driver=%q", r.driver) + } + db := r.db.WithContext(ctx) + for _, name := range []string{"logs_au", "logs_ad", "logs_ai"} { + if err := db.Exec("DROP TRIGGER IF EXISTS " + name).Error; err != nil { + return fmt.Errorf("drop trigger %s: %w", name, err) + } + } + if err := db.Exec("DROP TABLE IF EXISTS " + fts5LogsTable).Error; err != nil { + return fmt.Errorf("drop table %s: %w", fts5LogsTable, err) + } + if err := db.Exec("VACUUM").Error; err != nil { + return fmt.Errorf("vacuum: %w", err) + } + slog.Info("FTS5 logs index dropped and DB vacuumed") + return nil } diff --git a/internal/storage/fts5_test.go b/internal/storage/fts5_test.go index 7f66e26..c66fb15 100644 --- a/internal/storage/fts5_test.go +++ b/internal/storage/fts5_test.go @@ -2,6 +2,7 @@ package storage import ( "context" + "os" "strings" "testing" "time" @@ -43,8 +44,10 @@ func TestFTS5MatchExpr_Quoting(t *testing.T) { } } -// TestFTS5Available_DriverGate verifies that only sqlite uses the FTS5 path. +// TestFTS5Available_DriverGate verifies that only sqlite uses the FTS5 path +// when the LOG_FTS_ENABLED flag is on. func TestFTS5Available_DriverGate(t *testing.T) { + t.Setenv("LOG_FTS_ENABLED", "true") cases := []struct { driver string want bool @@ -63,6 +66,54 @@ func TestFTS5Available_DriverGate(t *testing.T) { } } +// TestFTS5Available_FlagGate verifies the LOG_FTS_ENABLED toggle. +func TestFTS5Available_FlagGate(t *testing.T) { + cases := []struct { + flag string + want bool + }{ + {"", false}, // empty value present = unparsable = false + {"false", false}, // explicit off + {"0", false}, + {"no", false}, + {"true", true}, + {"1", true}, + {"yes", true}, + {"on", true}, + {"YES", true}, + {"TRUE", true}, + } + for _, c := range cases { + t.Run(c.flag, func(t *testing.T) { + t.Setenv("LOG_FTS_ENABLED", c.flag) + if got := fts5Available("sqlite"); got != c.want { + t.Fatalf("fts5Available(sqlite) with LOG_FTS_ENABLED=%q = %v, want %v", c.flag, got, c.want) + } + }) + } +} + +// TestFTS5Available_DefaultOff verifies the absence of LOG_FTS_ENABLED yields +// false. Uses os.Unsetenv directly because t.Setenv only sets values, never +// unsets — and the test's whole point is to assert behavior when the env var +// is unset. +func TestFTS5Available_DefaultOff(t *testing.T) { + prev, hadPrev := os.LookupEnv("LOG_FTS_ENABLED") + if err := os.Unsetenv("LOG_FTS_ENABLED"); err != nil { + t.Fatalf("unsetenv: %v", err) + } + t.Cleanup(func() { + if hadPrev { + _ = os.Setenv("LOG_FTS_ENABLED", prev) + } else { + _ = os.Unsetenv("LOG_FTS_ENABLED") + } + }) + if fts5Available("sqlite") { + t.Fatal("FTS5 must default off when LOG_FTS_ENABLED is unset") + } +} + // TestSearchLogs_FTS5_BM25_Ordering verifies that BM25 puts the more relevant // row first (more occurrences of the query token = lower BM25 score = higher rank). func TestSearchLogs_FTS5_BM25_Ordering(t *testing.T) { diff --git a/internal/storage/log_repo.go b/internal/storage/log_repo.go index 3c12fe4..c72a21e 100644 --- a/internal/storage/log_repo.go +++ b/internal/storage/log_repo.go @@ -212,6 +212,34 @@ func (r *Repository) UpdateLogInsight(ctx context.Context, logID uint, insight s return nil } +// LogsForVectorReplay returns ERROR/WARN-family logs with id > sinceID, +// page-bounded by limit and ordered by id ASC. Used at startup by the +// vector-index tail-replay path to pick up DB rows inserted after the last +// snapshot. The id-ascending order lets the caller use the last row's id +// as the next page's sinceID — clean cursor pagination, no offset cost. +// +// Cross-tenant by design: vectordb is a global index with per-doc tenant +// tags enforced at Search time. Not exposed on any tenant-scoped API. +// +// Severity filter is intentionally narrow (ERROR / WARN / WARNING / FATAL / +// CRITICAL) so non-indexed rows don't waste page space; this matches +// vectordb.shouldIndex(). +func (r *Repository) LogsForVectorReplay(ctx context.Context, sinceID uint, limit int) ([]Log, error) { + if limit <= 0 || limit > 100_000 { + limit = 10_000 + } + var logs []Log + err := r.db.WithContext(ctx). + Where("id > ? AND severity IN ?", sinceID, []string{"ERROR", "WARN", "WARNING", "FATAL", "CRITICAL"}). + Order("id ASC"). + Limit(limit). + Find(&logs).Error + if err != nil { + return nil, fmt.Errorf("logs for vector replay: %w", err) + } + return logs, nil +} + // ListRecentHighSeverityLogsAllTenants returns recent logs of the given // severity across EVERY tenant, each row carrying its own TenantID. This is an // administrative read used exclusively by the vector index's startup diff --git a/internal/storage/log_repo_replay_test.go b/internal/storage/log_repo_replay_test.go new file mode 100644 index 0000000..5b60cac --- /dev/null +++ b/internal/storage/log_repo_replay_test.go @@ -0,0 +1,138 @@ +package storage + +import ( + "context" + "testing" + "time" +) + +// TestLogsForVectorReplay_ReturnsErrorAndWarnOnly verifies the severity +// filter matches vectordb.shouldIndex (ERROR/WARN/WARNING/FATAL/CRITICAL). +// INFO and DEBUG rows must be excluded so the page isn't bloated with rows +// vectordb would drop anyway. +func TestLogsForVectorReplay_ReturnsErrorAndWarnOnly(t *testing.T) { + repo := newTestRepo(t) + now := time.Now().UTC() + rows := []Log{ + {TenantID: "default", Severity: "ERROR", Body: "panic", ServiceName: "svc", Timestamp: now}, + {TenantID: "default", Severity: "WARN", Body: "slow", ServiceName: "svc", Timestamp: now}, + {TenantID: "default", Severity: "WARNING", Body: "deprecated", ServiceName: "svc", Timestamp: now}, + {TenantID: "default", Severity: "FATAL", Body: "OOM", ServiceName: "svc", Timestamp: now}, + {TenantID: "default", Severity: "CRITICAL", Body: "deadlock", ServiceName: "svc", Timestamp: now}, + {TenantID: "default", Severity: "INFO", Body: "request handled", ServiceName: "svc", Timestamp: now}, + {TenantID: "default", Severity: "DEBUG", Body: "trace data", ServiceName: "svc", Timestamp: now}, + } + if err := repo.db.Create(&rows).Error; err != nil { + t.Fatalf("seed: %v", err) + } + + got, err := repo.LogsForVectorReplay(context.Background(), 0, 100) + if err != nil { + t.Fatalf("LogsForVectorReplay: %v", err) + } + if len(got) != 5 { + t.Errorf("got %d rows, want 5 (ERROR+WARN+WARNING+FATAL+CRITICAL)", len(got)) + } + for _, l := range got { + if l.Severity == "INFO" || l.Severity == "DEBUG" { + t.Errorf("unexpected severity in result: %q (id=%d)", l.Severity, l.ID) + } + } +} + +// TestLogsForVectorReplay_RespectsSinceID verifies the cursor pagination +// contract: rows with id <= sinceID are excluded so the caller can advance +// across pages without re-fetching. +func TestLogsForVectorReplay_RespectsSinceID(t *testing.T) { + repo := newTestRepo(t) + now := time.Now().UTC() + for range 5 { + repo.db.Create(&Log{TenantID: "default", Severity: "ERROR", Body: "x", ServiceName: "svc", Timestamp: now}) + } + + page1, err := repo.LogsForVectorReplay(context.Background(), 0, 2) + if err != nil { + t.Fatalf("page1: %v", err) + } + if len(page1) != 2 { + t.Fatalf("page1: got %d rows, want 2", len(page1)) + } + // IDs must be strictly ascending. + if page1[0].ID >= page1[1].ID { + t.Errorf("page1 not ascending: %d, %d", page1[0].ID, page1[1].ID) + } + + page2, err := repo.LogsForVectorReplay(context.Background(), page1[1].ID, 2) + if err != nil { + t.Fatalf("page2: %v", err) + } + if len(page2) != 2 { + t.Fatalf("page2: got %d rows, want 2", len(page2)) + } + for _, r := range page2 { + if r.ID <= page1[1].ID { + t.Errorf("page2 contains id=%d <= page1 cursor=%d", r.ID, page1[1].ID) + } + } + + page3, err := repo.LogsForVectorReplay(context.Background(), page2[1].ID, 2) + if err != nil { + t.Fatalf("page3: %v", err) + } + if len(page3) != 1 { + t.Errorf("page3: got %d rows, want 1 (final partial page)", len(page3)) + } +} + +// TestLogsForVectorReplay_CrossTenant verifies the replay is intentionally +// cross-tenant — vectordb is a global accelerator and per-doc tenant tags +// enforce isolation at Search time. +func TestLogsForVectorReplay_CrossTenant(t *testing.T) { + repo := newTestRepo(t) + now := time.Now().UTC() + repo.db.Create(&[]Log{ + {TenantID: "acme", Severity: "ERROR", Body: "a", ServiceName: "svc", Timestamp: now}, + {TenantID: "globex", Severity: "ERROR", Body: "b", ServiceName: "svc", Timestamp: now}, + {TenantID: "default", Severity: "ERROR", Body: "c", ServiceName: "svc", Timestamp: now}, + }) + + // No tenant context — replay is cross-tenant by design. + got, err := repo.LogsForVectorReplay(context.Background(), 0, 100) + if err != nil { + t.Fatalf("LogsForVectorReplay: %v", err) + } + if len(got) != 3 { + t.Errorf("got %d rows across tenants, want 3", len(got)) + } + tenants := map[string]int{} + for _, l := range got { + tenants[l.TenantID]++ + } + for _, name := range []string{"acme", "globex", "default"} { + if tenants[name] != 1 { + t.Errorf("tenant %q: got %d rows, want 1", name, tenants[name]) + } + } +} + +// TestLogsForVectorReplay_LimitClamp verifies the limit is clamped to a +// safe default when caller passes 0 / negative / absurdly large values. +func TestLogsForVectorReplay_LimitClamp(t *testing.T) { + repo := newTestRepo(t) + now := time.Now().UTC() + for range 3 { + repo.db.Create(&Log{TenantID: "default", Severity: "ERROR", Body: "x", ServiceName: "svc", Timestamp: now}) + } + + for _, lim := range []int{0, -1, 999_999} { + got, err := repo.LogsForVectorReplay(context.Background(), 0, lim) + if err != nil { + t.Errorf("limit=%d: unexpected err=%v", lim, err) + continue + } + // 3 rows seeded; default cap is 10k, so all 3 must come back. + if len(got) != 3 { + t.Errorf("limit=%d: got %d rows, want 3", lim, len(got)) + } + } +} diff --git a/internal/storage/search_cap.go b/internal/storage/search_cap.go new file mode 100644 index 0000000..5ce9b47 --- /dev/null +++ b/internal/storage/search_cap.go @@ -0,0 +1,51 @@ +package storage + +import ( + "errors" + "time" +) + +// ClampSearchWindowTo24h enforces a 24-hour ceiling on log keyword search +// windows applied at the API/MCP boundary. With FTS5 disabled (the default), +// the only safety mechanism for unscoped LIKE substring scans is a hard +// time-range cap so the worst-case query is bounded in disk pages scanned. +// +// Behaviour: +// +// - Both zero (no caller hint): defaults to (now-24h, now) +// - end zero: end = now +// - start zero: start = end - 24h +// - end > now: end clamped to now (don't reject — clock skew on the caller +// side is common) +// - start < now-24h with end >= now-24h: start clamped to now-24h (the +// window is shrunk to fit the cap, callers never see a silent truncation +// of an in-window query) +// - end < now-24h (window entirely outside the cap): rejected with error so +// the caller gets a deterministic signal instead of an empty result +// - start >= end: rejected with error (caller mistake) +// +// The cap fires whenever a body keyword search is requested; pure filtered +// listings (no Search field) keep the full retention range and bypass this +// helper. +func ClampSearchWindowTo24h(start, end, now time.Time) (time.Time, time.Time, error) { + cutoff := now.Add(-24 * time.Hour) + if end.IsZero() { + end = now + } + if start.IsZero() { + start = end.Add(-24 * time.Hour) + } + if end.After(now) { + end = now + } + if end.Before(cutoff) { + return time.Time{}, time.Time{}, errors.New("search window must be within the last 24h") + } + if start.Before(cutoff) { + start = cutoff + } + if !start.Before(end) { + return time.Time{}, time.Time{}, errors.New("start_time must be before end_time") + } + return start, end, nil +} diff --git a/internal/storage/search_cap_test.go b/internal/storage/search_cap_test.go new file mode 100644 index 0000000..7ba9016 --- /dev/null +++ b/internal/storage/search_cap_test.go @@ -0,0 +1,106 @@ +package storage + +import ( + "testing" + "time" +) + +func TestClampSearchWindowTo24h(t *testing.T) { + now := time.Date(2026, 4, 30, 12, 0, 0, 0, time.UTC) + hourAgo := now.Add(-1 * time.Hour) + dayAgo := now.Add(-24 * time.Hour) + weekAgo := now.Add(-7 * 24 * time.Hour) + hourAhead := now.Add(1 * time.Hour) + + cases := []struct { + name string + start time.Time + end time.Time + wantStart time.Time + wantEnd time.Time + wantErr bool + }{ + { + name: "both zero defaults to last 24h", + wantStart: dayAgo, + wantEnd: now, + }, + { + // With end=1h ago and start unset, start is initialized to + // end-24h (= 25h ago) and then clamped to the cutoff (24h ago) + // so the resulting window cannot extend past the 24h floor. + // Window shrinks to 23h — caller sees a strict ceiling on age. + name: "only end set: start clamps to cutoff (not end-24h)", + end: hourAgo, + wantStart: dayAgo, + wantEnd: hourAgo, + }, + { + name: "start oversize is clamped to cutoff", + start: weekAgo, + end: now, + wantStart: dayAgo, + wantEnd: now, + }, + { + name: "end in the future is clamped to now", + start: hourAgo, + end: hourAhead, + wantStart: hourAgo, + wantEnd: now, + }, + { + name: "window entirely outside cap is rejected", + start: now.Add(-5 * 24 * time.Hour), + end: now.Add(-4 * 24 * time.Hour), + wantErr: true, + }, + { + name: "narrow in-cap window is unchanged", + start: hourAgo, + end: now.Add(-30 * time.Minute), + wantStart: hourAgo, + wantEnd: now.Add(-30 * time.Minute), + }, + { + name: "start equal to end is rejected", + start: hourAgo, + end: hourAgo, + wantErr: true, + }, + { + name: "start after end is rejected", + start: now.Add(-30 * time.Minute), + end: hourAgo, + wantErr: true, + }, + { + name: "start at cutoff boundary is allowed", + start: dayAgo, + end: now, + wantStart: dayAgo, + wantEnd: now, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + gotStart, gotEnd, err := ClampSearchWindowTo24h(c.start, c.end, now) + if c.wantErr { + if err == nil { + t.Fatalf("expected error, got start=%v end=%v", gotStart, gotEnd) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !gotStart.Equal(c.wantStart) { + t.Errorf("start: got %v, want %v", gotStart, c.wantStart) + } + if !gotEnd.Equal(c.wantEnd) { + t.Errorf("end: got %v, want %v", gotEnd, c.wantEnd) + } + }) + } +} diff --git a/internal/storage/testhelpers_test.go b/internal/storage/testhelpers_test.go index 8a04103..475de2c 100644 --- a/internal/storage/testhelpers_test.go +++ b/internal/storage/testhelpers_test.go @@ -10,8 +10,13 @@ import ( // newTestRepo builds a Repository backed by an in-memory SQLite DB with all models migrated. // Tests live in the same package so they can poke unexported fields. +// +// FTS5 is force-enabled here so the existing FTS5 + BM25 + trigger tests +// keep working after LOG_FTS_ENABLED defaulted to false in production. +// t.Setenv applies for the lifetime of the test that called this helper. func newTestRepo(t *testing.T) *Repository { t.Helper() + t.Setenv("LOG_FTS_ENABLED", "true") db, err := NewDatabase("sqlite", ":memory:") if err != nil { t.Fatalf("NewDatabase: %v", err) diff --git a/internal/telemetry/metrics.go b/internal/telemetry/metrics.go index 44f0afc..e6a3d54 100644 --- a/internal/telemetry/metrics.go +++ b/internal/telemetry/metrics.go @@ -127,6 +127,26 @@ type Metrics struct { // --- Dashboard p99 (Task 10) --- DashboardP99RowCapHitsTotal prometheus.Counter + // --- Vectordb persistence --- + // VectorSnapshotWritesTotal counts snapshot write attempts, labeled + // {result=success|failure}. Alert on rate(failure[10m]) > 0. + VectorSnapshotWritesTotal *prometheus.CounterVec + // VectorSnapshotDurationSeconds is the WriteSnapshot wall-clock + // duration. Histogram so operators can SLO p95 / p99. + VectorSnapshotDurationSeconds prometheus.Histogram + // VectorSnapshotSizeBytes gauges the on-disk size of the latest + // successful snapshot. Sudden growth signals a maxSize bump or a + // schema change worth investigating. + VectorSnapshotSizeBytes prometheus.Gauge + // VectorSnapshotLoadTotal counts startup snapshot loads, labeled + // {result=success|missing|corrupt}. corrupt = magic/version/crc/decode + // failure — caller falls back to a full DB rebuild. + VectorSnapshotLoadTotal *prometheus.CounterVec + // VectorReplayLogsTotal accumulates rows processed by ReplayFromDB + // across the daemon's lifetime. The rate spikes only at startup + // (catching the snapshot→now gap), then stays flat. + VectorReplayLogsTotal prometheus.Counter + // Atomic counters for JSON health endpoint (avoids scraping Prometheus) totalIngested atomic.Int64 activeConns atomic.Int64 @@ -371,9 +391,64 @@ func New() *Metrics { Name: "otelcontext_dashboard_p99_row_cap_hits_total", Help: "Number of dashboard p99 computations that hit the SQLite row cap (200k). Indicates the dataset is too large for in-memory p99 — use Postgres for prod.", }) + m.VectorSnapshotWritesTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "otelcontext_vectordb_snapshot_writes_total", + Help: "Vectordb snapshot write attempts by result (success|failure). Alert on rate(...{result=\"failure\"}[10m]) > 0.", + }, []string{"result"}) + m.VectorSnapshotDurationSeconds = promauto.NewHistogram(prometheus.HistogramOpts{ + Name: "otelcontext_vectordb_snapshot_duration_seconds", + Help: "Wall-clock duration of WriteSnapshot, including encode + atomic rename.", + Buckets: []float64{.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5}, + }) + m.VectorSnapshotSizeBytes = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "otelcontext_vectordb_snapshot_size_bytes", + Help: "On-disk size of the latest successful vectordb snapshot.", + }) + m.VectorSnapshotLoadTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "otelcontext_vectordb_snapshot_load_total", + Help: "Vectordb snapshot load attempts at startup by result (success|missing|corrupt).", + }, []string{"result"}) + m.VectorReplayLogsTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "otelcontext_vectordb_replay_logs_total", + Help: "Total log rows processed by vectordb ReplayFromDB across the daemon's lifetime.", + }) return m } +// RecordVectorSnapshotWrite is the observer hook the vectordb snapshot +// path calls after each WriteSnapshot attempt. result is "success" or +// "failure"; size is the on-disk byte count after a successful rename +// (zero on failure). +func (m *Metrics) RecordVectorSnapshotWrite(result string, duration time.Duration, size int64) { + if m == nil || m.VectorSnapshotWritesTotal == nil { + return + } + m.VectorSnapshotWritesTotal.WithLabelValues(result).Inc() + m.VectorSnapshotDurationSeconds.Observe(duration.Seconds()) + if result == "success" && size > 0 { + m.VectorSnapshotSizeBytes.Set(float64(size)) + } +} + +// RecordVectorSnapshotLoad is the observer hook for startup snapshot +// loads. result is "success", "missing" (first start, no prior file), +// or "corrupt" (any decode/CRC/version error → full rebuild fallback). +func (m *Metrics) RecordVectorSnapshotLoad(result string) { + if m == nil || m.VectorSnapshotLoadTotal == nil { + return + } + m.VectorSnapshotLoadTotal.WithLabelValues(result).Inc() +} + +// RecordVectorReplayLogs adds rows processed by ReplayFromDB to the +// lifetime counter. Called once after the startup tail-replay completes. +func (m *Metrics) RecordVectorReplayLogs(count int) { + if m == nil || m.VectorReplayLogsTotal == nil || count <= 0 { + return + } + m.VectorReplayLogsTotal.Add(float64(count)) +} + // StartRuntimeMetrics samples Go runtime stats every 15 seconds. func (m *Metrics) StartRuntimeMetrics() { go func() { diff --git a/internal/vectordb/index.go b/internal/vectordb/index.go index 99b804a..13777f4 100644 --- a/internal/vectordb/index.go +++ b/internal/vectordb/index.go @@ -8,6 +8,7 @@ import ( "sort" "strings" "sync" + "time" "unicode" ) @@ -22,13 +23,18 @@ const defaultTenantID = "default" // rows. The TF-IDF table is shared across tenants — global IDF still gives // the right rarity signal — but the per-document tenant tag is enforced at // query time so two tenants with overlapping log bodies stay isolated. +// +// All fields are exported so encoding/gob can serialize the type for +// snapshot persistence (snapshot.go). Vec is the per-doc TF map (term → +// frequency); IDF is held separately on the Index to avoid duplicating +// rarity weights across documents. type LogVector struct { LogID uint Tenant string ServiceName string Severity string Body string - vec map[string]float64 // TF-IDF sparse vector + Vec map[string]float64 // TF-IDF sparse vector } // SearchResult is a single similarity hit. @@ -43,12 +49,24 @@ type SearchResult struct { // Index is a thread-safe in-memory TF-IDF vector index for log bodies. // Only ERROR and WARN logs are indexed to keep it small and relevant. +// +// lastIndexedID records the highest Log.ID Add() has accepted. Persisted +// in the snapshot so a startup tail-replay can pick up DB rows newer than +// this watermark without re-indexing rows already in the snapshot. Tracked +// only for rows that pass shouldIndex(); INFO/DEBUG rows interleaved in +// the same ID range are excluded by the severity filter on replay anyway. type Index struct { - mu sync.RWMutex - docs []LogVector // indexed log vectors - idf map[string]float64 // global IDF table - maxSize int // FIFO eviction cap - dirty bool // IDF needs recompute + mu sync.RWMutex + docs []LogVector // indexed log vectors + idf map[string]float64 // global IDF table + maxSize int // FIFO eviction cap + dirty bool // IDF needs recompute + lastIndexedID uint // high watermark of indexed Log.ID + + // snapshotObserver is invoked at the end of each WriteSnapshot + // (success or failure). nil-safe — set via SetSnapshotObserver from + // the wiring layer so vectordb stays free of telemetry imports. + snapshotObserver func(result string, duration time.Duration, size int64) } // New creates a new Index with the given maximum entry cap. @@ -82,6 +100,14 @@ func (idx *Index) Add(logID uint, tenant, serviceName, severity, body string) { idx.mu.Lock() defer idx.mu.Unlock() + // High watermark for tail-replay correctness. Bump only after the + // shouldIndex/tokenize gates pass — the replay query is severity- + // filtered too, so non-indexed rows interleaved in the same ID range + // are excluded by SQL anyway. + if logID > idx.lastIndexedID { + idx.lastIndexedID = logID + } + // Tenant-aware FIFO eviction. When at cap, remove up to maxSize/10 of the // oldest entries belonging to the inserting tenant so a noisy tenant // cannot push another tenant's warm rows out of the index (availability @@ -120,7 +146,7 @@ func (idx *Index) Add(logID uint, tenant, serviceName, severity, body string) { ServiceName: serviceName, Severity: severity, Body: body, - vec: tf, + Vec: tf, }) idx.dirty = true } @@ -175,8 +201,8 @@ func (idx *Index) Search(tenant, query string, k int) []SearchResult { if doc.Tenant != tenant { continue } - docVec := make(map[string]float64, len(doc.vec)) - for term, tf := range doc.vec { + docVec := make(map[string]float64, len(doc.Vec)) + for term, tf := range doc.Vec { docVec[term] = tf * idfSnap[term] } score := cosineSimilarity(queryVec, queryNorm, docVec) @@ -213,11 +239,21 @@ func (idx *Index) Size() int { return len(idx.docs) } +// LastIndexedID returns the highest Log.ID that has been successfully indexed +// (i.e. passed shouldIndex + tokenize gates and was appended to docs). +// Used by the startup tail-replay path to query DB rows newer than this +// watermark; persisted in the snapshot so replay survives restarts. +func (idx *Index) LastIndexedID() uint { + idx.mu.RLock() + defer idx.mu.RUnlock() + return idx.lastIndexedID +} + // recomputeIDF rebuilds the IDF table from current docs. Must be called with mu held. func (idx *Index) recomputeIDF() { df := make(map[string]int, len(idx.idf)) for _, doc := range idx.docs { - for term := range doc.vec { + for term := range doc.Vec { df[term]++ } } diff --git a/internal/vectordb/replay.go b/internal/vectordb/replay.go new file mode 100644 index 0000000..5ec9de5 --- /dev/null +++ b/internal/vectordb/replay.go @@ -0,0 +1,74 @@ +package vectordb + +import "context" + +// ReplaySource is the minimal contract a backing store fulfills to hydrate +// this Index on startup. Pages are pulled in id-ascending order; the source +// signals end-of-data by returning a slice shorter than the requested limit. +// ReplayFromDB walks pages starting from LastIndexedID() until the source +// returns no more rows. +// +// Vectordb intentionally does NOT import the storage package — keeping it as +// a leaf accelerator means tests can wire any in-memory source without a +// SQLite dependency, and storage is free to evolve its row type without +// breaking vectordb. The wiring layer (cmd/main.go) is responsible for +// projecting storage.Log into ReplayRow. +type ReplaySource interface { + LogsForVectorReplay(ctx context.Context, sinceID uint, limit int) ([]ReplayRow, error) +} + +// ReplayRow is the minimum field set Add() needs. Mirrors the projection a +// storage adapter performs at the boundary. +type ReplayRow struct { + ID uint + Tenant string + ServiceName string + Severity string + Body string +} + +// replayPageSize bounds memory during tail-replay. 10k rows is a reasonable +// trade-off between query overhead per page and peak heap; at typical body +// sizes this stays well under 50 MB resident per page. +const replayPageSize = 10_000 + +// ReplayFromDB walks ReplaySource pages starting from LastIndexedID() and +// feeds each row through Add(). Returns the count of rows processed (Add +// filters by severity, so processed ≠ indexed when the source loosens its +// filter — but the standard storage implementation already pre-filters to +// ERROR/WARN/family so the counts match in practice). +// +// Termination contract: the source signals end-of-data by returning a +// zero-length slice. This lets sources page however they want without +// having to fill every page exactly to replayPageSize — the trade-off is +// one extra round-trip at the tail (fine for a one-shot startup call). +// +// Caller passes a derived ctx so SIGTERM during boot cancels the replay +// cleanly. On any source error, returns the partial count + error so the +// caller can log and proceed with a partially-warm index. +func (idx *Index) ReplayFromDB(ctx context.Context, src ReplaySource) (int, error) { + if src == nil { + return 0, nil + } + sinceID := idx.LastIndexedID() + total := 0 + for { + if err := ctx.Err(); err != nil { + return total, err + } + rows, err := src.LogsForVectorReplay(ctx, sinceID, replayPageSize) + if err != nil { + return total, err + } + if len(rows) == 0 { + return total, nil + } + for _, row := range rows { + idx.Add(row.ID, row.Tenant, row.ServiceName, row.Severity, row.Body) + if row.ID > sinceID { + sinceID = row.ID + } + } + total += len(rows) + } +} diff --git a/internal/vectordb/replay_test.go b/internal/vectordb/replay_test.go new file mode 100644 index 0000000..28b829a --- /dev/null +++ b/internal/vectordb/replay_test.go @@ -0,0 +1,161 @@ +package vectordb + +import ( + "context" + "errors" + "testing" +) + +// fakeSource is an in-memory ReplaySource for unit-testing the page loop +// without a real DB. Pages are produced by a closure so each test can shape +// the source however it likes (multi-page, errors, end-of-data). +type fakeSource struct { + pages [][]ReplayRow // queued pages; consumed in order + calls int + fail error +} + +func (s *fakeSource) LogsForVectorReplay(_ context.Context, sinceID uint, limit int) ([]ReplayRow, error) { + s.calls++ + if s.fail != nil { + return nil, s.fail + } + if s.calls > len(s.pages) { + return nil, nil + } + page := s.pages[s.calls-1] + // Filter to "rows newer than sinceID" so the test verifies the loop + // passes the right cursor across iterations. + out := make([]ReplayRow, 0, len(page)) + for _, r := range page { + if r.ID > sinceID { + out = append(out, r) + } + } + if len(out) > limit { + out = out[:limit] + } + return out, nil +} + +// TestReplayFromDB_AdvancesCursor verifies multi-page replay calls the +// source with monotonically-increasing sinceID values and indexes every +// row, with no duplicates by LogID. +func TestReplayFromDB_AdvancesCursor(t *testing.T) { + src := &fakeSource{ + pages: [][]ReplayRow{ + { + {ID: 10, Tenant: "t", ServiceName: "svc", Severity: "ERROR", Body: "boom"}, + {ID: 20, Tenant: "t", ServiceName: "svc", Severity: "ERROR", Body: "kaboom"}, + }, + { + {ID: 30, Tenant: "t", ServiceName: "svc", Severity: "WARN", Body: "third page row tokenizes fine"}, + }, + }, + } + idx := New(100) + total, err := idx.ReplayFromDB(context.Background(), src) + if err != nil { + t.Fatalf("ReplayFromDB: %v", err) + } + if total != 3 { + t.Errorf("processed: got %d, want 3", total) + } + if idx.Size() != 3 { + t.Errorf("indexed Size: got %d, want 3", idx.Size()) + } + if idx.LastIndexedID() != 30 { + t.Errorf("LastIndexedID: got %d, want 30", idx.LastIndexedID()) + } + // Two data pages + one empty page that signals end-of-data. + if src.calls != 3 { + t.Errorf("source calls: got %d, want 3 (2 data + 1 empty terminator)", src.calls) + } +} + +// TestReplayFromDB_StartsFromLastIndexedID verifies the loop seeds sinceID +// from the existing high watermark, so a snapshot's tail can be picked up +// without re-indexing rows already in the index. +func TestReplayFromDB_StartsFromLastIndexedID(t *testing.T) { + idx := New(100) + idx.Add(50, "t", "svc", "ERROR", "already indexed") + if got := idx.LastIndexedID(); got != 50 { + t.Fatalf("seed LastIndexedID: got %d, want 50", got) + } + + src := &fakeSource{ + pages: [][]ReplayRow{ + // Page contains both pre-watermark and post-watermark rows; the + // fake's filter mimics SQL's WHERE id > sinceID, so only post-50 + // rows leave the source. + { + {ID: 30, Tenant: "t", ServiceName: "svc", Severity: "ERROR", Body: "old"}, + {ID: 50, Tenant: "t", ServiceName: "svc", Severity: "ERROR", Body: "boundary"}, + {ID: 60, Tenant: "t", ServiceName: "svc", Severity: "ERROR", Body: "new"}, + }, + }, + } + total, err := idx.ReplayFromDB(context.Background(), src) + if err != nil { + t.Fatalf("ReplayFromDB: %v", err) + } + if total != 1 { + t.Errorf("processed: got %d, want 1 (only id=60 is post-watermark)", total) + } + if idx.Size() != 2 { + t.Errorf("indexed Size: got %d, want 2 (seed + replayed)", idx.Size()) + } + if idx.LastIndexedID() != 60 { + t.Errorf("LastIndexedID: got %d, want 60", idx.LastIndexedID()) + } +} + +// TestReplayFromDB_PropagatesError verifies a source error is returned +// alongside the partial count so the caller can log and continue. +func TestReplayFromDB_PropagatesError(t *testing.T) { + src := &fakeSource{fail: errors.New("db gone")} + idx := New(100) + total, err := idx.ReplayFromDB(context.Background(), src) + if err == nil { + t.Fatal("want error, got nil") + } + if total != 0 { + t.Errorf("partial count: got %d, want 0", total) + } + if idx.Size() != 0 { + t.Errorf("error path must not corrupt index: Size=%d", idx.Size()) + } +} + +// TestReplayFromDB_RespectsCancellation verifies a cancelled ctx aborts +// the loop without making another source call. +func TestReplayFromDB_RespectsCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + src := &fakeSource{ + pages: [][]ReplayRow{ + {{ID: 1, Tenant: "t", ServiceName: "svc", Severity: "ERROR", Body: "x"}}, + }, + } + idx := New(100) + _, err := idx.ReplayFromDB(ctx, src) + if !errors.Is(err, context.Canceled) { + t.Fatalf("want context.Canceled, got %v", err) + } + if src.calls != 0 { + t.Errorf("source called despite cancelled ctx: calls=%d", src.calls) + } +} + +// TestReplayFromDB_NilSource is a smoke test for the nil-safe early return. +func TestReplayFromDB_NilSource(t *testing.T) { + idx := New(100) + total, err := idx.ReplayFromDB(context.Background(), nil) + if err != nil { + t.Fatalf("nil source: unexpected err=%v", err) + } + if total != 0 { + t.Errorf("nil source: total=%d, want 0", total) + } +} diff --git a/internal/vectordb/snapshot.go b/internal/vectordb/snapshot.go new file mode 100644 index 0000000..87c12d0 --- /dev/null +++ b/internal/vectordb/snapshot.go @@ -0,0 +1,317 @@ +package vectordb + +import ( + "bytes" + "context" + "encoding/binary" + "encoding/gob" + "errors" + "fmt" + "hash/crc32" + "io" + "log/slog" + "os" + "syscall" + "time" +) + +// Snapshot is the persisted state of an Index. +// +// Only the fields needed to reconstruct an equivalent Index are captured — +// transient state (mu, dirty) is intentionally absent. LastIndexedID is the +// high watermark of indexed Log.IDs so a startup tail-replay can query DB +// rows newer than the snapshot without double-indexing rows already in +// Docs. +// +// Field changes break the format — bump snapshotVersion when the wire +// shape changes. Old snapshots whose magic+version don't match are +// rejected on load and the caller falls back to a full DB rebuild. +type Snapshot struct { + LastIndexedID uint + MaxSize int + Docs []LogVector + IDF map[string]float64 + WrittenAt int64 // unix seconds, observability only +} + +const ( + // snapshotMagic is a 4-byte file header so a corrupt or stray file is + // rejected before we attempt the more expensive gob decode. + snapshotMagic = "VDB1" + // snapshotVersion travels alongside the magic. Bump on any LogVector + // or Snapshot field shape change so loaders fall back to rebuild + // instead of producing silently-wrong index state. + snapshotVersion uint32 = 1 +) + +// EncodeSnapshot writes a versioned, CRC32-protected snapshot to w. +// +// Wire format (big-endian for portability): +// +// bytes[0:4] magic "VDB1" +// bytes[4:8] version uint32 +// bytes[8:12] CRC32-IEEE uint32 (over bytes[12:]) +// bytes[12:] gob payload Snapshot +func EncodeSnapshot(w io.Writer, snap Snapshot) error { + var payload bytes.Buffer + if err := gob.NewEncoder(&payload).Encode(snap); err != nil { + return fmt.Errorf("encode snapshot payload: %w", err) + } + crc := crc32.ChecksumIEEE(payload.Bytes()) + + if _, err := w.Write([]byte(snapshotMagic)); err != nil { + return fmt.Errorf("write magic: %w", err) + } + if err := binary.Write(w, binary.BigEndian, snapshotVersion); err != nil { + return fmt.Errorf("write version: %w", err) + } + if err := binary.Write(w, binary.BigEndian, crc); err != nil { + return fmt.Errorf("write crc: %w", err) + } + if _, err := w.Write(payload.Bytes()); err != nil { + return fmt.Errorf("write payload: %w", err) + } + return nil +} + +// DecodeSnapshot reads + validates a snapshot from r. +// +// All errors are caller-visible. The expected handling is: log a warning +// and proceed with a full DB rebuild — never silently load partial state. +// Errors include short header, wrong magic, unsupported version, CRC +// mismatch, and gob decode failure. +func DecodeSnapshot(r io.Reader) (Snapshot, error) { + var ( + magic [4]byte + version uint32 + crc uint32 + ) + if _, err := io.ReadFull(r, magic[:]); err != nil { + return Snapshot{}, fmt.Errorf("read magic: %w", err) + } + if string(magic[:]) != snapshotMagic { + return Snapshot{}, fmt.Errorf("unexpected snapshot magic %q (want %q)", magic[:], snapshotMagic) + } + if err := binary.Read(r, binary.BigEndian, &version); err != nil { + return Snapshot{}, fmt.Errorf("read version: %w", err) + } + if version != snapshotVersion { + return Snapshot{}, fmt.Errorf("unsupported snapshot version %d (current %d)", version, snapshotVersion) + } + if err := binary.Read(r, binary.BigEndian, &crc); err != nil { + return Snapshot{}, fmt.Errorf("read crc: %w", err) + } + payload, err := io.ReadAll(r) + if err != nil { + return Snapshot{}, fmt.Errorf("read payload: %w", err) + } + if got := crc32.ChecksumIEEE(payload); got != crc { + return Snapshot{}, fmt.Errorf("snapshot crc mismatch: got %08x want %08x", got, crc) + } + var snap Snapshot + if err := gob.NewDecoder(bytes.NewReader(payload)).Decode(&snap); err != nil { + return Snapshot{}, fmt.Errorf("decode payload: %w", err) + } + return snap, nil +} + +// writeAtomic writes data to path via tmp+sync+rename. +// +// Mode 0o600: snapshots persist log bodies which can carry sensitive +// operational data — owner-only is the conservative default. Operators +// who need shared read can chmod externally. +// +// On EXDEV (cross-device rename, e.g. when data dir is on a separate +// mount than the binary's tmp dir), falls back to a non-atomic +// os.WriteFile at the destination. Cross-device deployments are rare and +// documented; the fallback at least ensures the snapshot is written, with +// last-writer-wins replacing the atomicity guarantee. +// +// On any error during the write/fsync phase, the .tmp file is removed so +// a partial file does not poison the next startup's load attempt. +func writeAtomic(path string, data []byte) error { + tmp := path + ".tmp" + f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) + if err != nil { + return fmt.Errorf("create tmp %s: %w", tmp, err) + } + if _, err := f.Write(data); err != nil { + _ = f.Close() + _ = os.Remove(tmp) + return fmt.Errorf("write tmp: %w", err) + } + if err := f.Sync(); err != nil { + _ = f.Close() + _ = os.Remove(tmp) + return fmt.Errorf("fsync tmp: %w", err) + } + if err := f.Close(); err != nil { + _ = os.Remove(tmp) + return fmt.Errorf("close tmp: %w", err) + } + if err := os.Rename(tmp, path); err != nil { + if isEXDEV(err) { + data, readErr := os.ReadFile(tmp) + if readErr != nil { + _ = os.Remove(tmp) + return fmt.Errorf("rename EXDEV + readback: %w", readErr) + } + if writeErr := os.WriteFile(path, data, 0o600); writeErr != nil { + _ = os.Remove(tmp) + return fmt.Errorf("rename EXDEV + writefile: %w", writeErr) + } + _ = os.Remove(tmp) + return nil + } + _ = os.Remove(tmp) + return fmt.Errorf("rename %s: %w", path, err) + } + return nil +} + +// isEXDEV reports whether err is a cross-device link/rename error. +func isEXDEV(err error) bool { + if err == nil { + return false + } + var le *os.LinkError + if errors.As(err, &le) { + return errors.Is(le.Err, syscall.EXDEV) + } + return errors.Is(err, syscall.EXDEV) +} + +// LoadSnapshot reads a snapshot from path and replaces the Index's state. +// +// Caller must ensure no concurrent Add()/Search() is in flight — this is +// the typical startup wiring (fresh Index, before ingest accept). Errors +// are returned as-is so the caller can distinguish os.IsNotExist (no +// previous snapshot — first start) from corruption/format errors (log +// warn + proceed with full DB rebuild). +// +// On error the Index state is left untouched. +func (idx *Index) LoadSnapshot(path string) error { + f, err := os.Open(path) // #nosec G304 -- operator-supplied snapshot path + if err != nil { + return err + } + defer func() { _ = f.Close() }() + snap, err := DecodeSnapshot(f) + if err != nil { + return err + } + idx.mu.Lock() + defer idx.mu.Unlock() + idx.docs = snap.Docs + idx.idf = snap.IDF + if idx.idf == nil { + idx.idf = make(map[string]float64) + } + if snap.MaxSize > 0 { + idx.maxSize = snap.MaxSize + } + idx.lastIndexedID = snap.LastIndexedID + idx.dirty = false + return nil +} + +// SetSnapshotObserver registers a callback invoked at the end of each +// WriteSnapshot. result is "success" or "failure"; size is the on-disk +// size of the latest written snapshot (0 on failure). +// +// Set from the wiring layer (main.go) so vectordb stays free of +// telemetry imports. Safe to call before SnapshotLoop starts. +func (idx *Index) SetSnapshotObserver(fn func(result string, duration time.Duration, size int64)) { + idx.mu.Lock() + defer idx.mu.Unlock() + idx.snapshotObserver = fn +} + +// WriteSnapshot serializes the current Index state to path atomically. +// +// Safe to call concurrently with Add()/Search(): the docs slice and IDF +// map are copied under the index lock and serialization runs lock-free +// after release. Critical section is sub-millisecond at the 100k cap +// because slice copy is O(1) per-element header (LogVector strings/maps +// are shared by reference, and Add() never mutates an existing +// LogVector.Vec — it only appends new entries). +func (idx *Index) WriteSnapshot(path string) error { + start := time.Now() + err := idx.writeSnapshot(path) + + idx.mu.RLock() + obs := idx.snapshotObserver + idx.mu.RUnlock() + if obs != nil { + result := "success" + var size int64 + if err != nil { + result = "failure" + } else if fi, statErr := os.Stat(path); statErr == nil { + size = fi.Size() + } + obs(result, time.Since(start), size) + } + return err +} + +func (idx *Index) writeSnapshot(path string) error { + idx.mu.Lock() + if idx.dirty { + idx.recomputeIDF() + idx.dirty = false + } + docs := make([]LogVector, len(idx.docs)) + copy(docs, idx.docs) + idfCopy := make(map[string]float64, len(idx.idf)) + for k, v := range idx.idf { + idfCopy[k] = v + } + snap := Snapshot{ + LastIndexedID: idx.lastIndexedID, + MaxSize: idx.maxSize, + Docs: docs, + IDF: idfCopy, + WrittenAt: time.Now().Unix(), + } + idx.mu.Unlock() + + var buf bytes.Buffer + if err := EncodeSnapshot(&buf, snap); err != nil { + return err + } + return writeAtomic(path, buf.Bytes()) +} + +// SnapshotLoop writes a snapshot to path on every interval tick until ctx is +// done. On context cancel, fires one final WriteSnapshot before returning so +// graceful shutdowns capture the maximum in-memory state. +// +// Transient write failures (disk full, fsync errors, EXDEV warnings) are +// logged via slog but do not break the loop — vectordb is a rebuildable +// accelerator, and silently dropping a tick beats taking the daemon down. +// +// Safe to call with empty path / zero interval — both disable the loop and +// return immediately. +func (idx *Index) SnapshotLoop(ctx context.Context, path string, interval time.Duration) { + if path == "" || interval <= 0 { + return + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + if err := idx.WriteSnapshot(path); err != nil { + slog.Warn("vectordb final snapshot on shutdown failed", "path", path, "error", err) + } else { + slog.Info("vectordb final snapshot written", "path", path, "size", idx.Size()) + } + return + case <-ticker.C: + if err := idx.WriteSnapshot(path); err != nil { + slog.Warn("vectordb periodic snapshot failed", "path", path, "error", err) + } + } + } +} diff --git a/internal/vectordb/snapshot_test.go b/internal/vectordb/snapshot_test.go new file mode 100644 index 0000000..9df3a67 --- /dev/null +++ b/internal/vectordb/snapshot_test.go @@ -0,0 +1,325 @@ +package vectordb + +import ( + "bytes" + "context" + "encoding/binary" + "errors" + "os" + "path/filepath" + "syscall" + "testing" + "time" +) + +// TestSnapshotRoundTrip verifies an encoded snapshot decodes back to the +// same logical state across all populated fields. +func TestSnapshotRoundTrip(t *testing.T) { + in := Snapshot{ + LastIndexedID: 42, + MaxSize: 1000, + Docs: []LogVector{ + {LogID: 1, Tenant: "acme", ServiceName: "api", Severity: "ERROR", Body: "panic at startup", Vec: map[string]float64{"panic": 0.5, "startup": 0.5}}, + {LogID: 2, Tenant: "globex", ServiceName: "db", Severity: "WARN", Body: "timeout connecting", Vec: map[string]float64{"timeout": 1.0}}, + }, + IDF: map[string]float64{"panic": 1.5, "startup": 1.0, "timeout": 1.2}, + WrittenAt: 1714464000, + } + var buf bytes.Buffer + if err := EncodeSnapshot(&buf, in); err != nil { + t.Fatalf("encode: %v", err) + } + out, err := DecodeSnapshot(&buf) + if err != nil { + t.Fatalf("decode: %v", err) + } + if out.LastIndexedID != in.LastIndexedID { + t.Errorf("LastIndexedID: got %d, want %d", out.LastIndexedID, in.LastIndexedID) + } + if out.MaxSize != in.MaxSize { + t.Errorf("MaxSize: got %d, want %d", out.MaxSize, in.MaxSize) + } + if len(out.Docs) != len(in.Docs) { + t.Fatalf("Docs length: got %d, want %d", len(out.Docs), len(in.Docs)) + } + if out.Docs[0].Body != in.Docs[0].Body || out.Docs[0].LogID != in.Docs[0].LogID { + t.Errorf("Doc[0]: got %+v, want %+v", out.Docs[0], in.Docs[0]) + } + if got, want := out.Docs[0].Vec["panic"], in.Docs[0].Vec["panic"]; got != want { + t.Errorf("Doc[0].Vec[panic]: got %v, want %v", got, want) + } + if got, want := out.IDF["panic"], in.IDF["panic"]; got != want { + t.Errorf("IDF[panic]: got %v, want %v", got, want) + } +} + +// TestDecodeSnapshot_EmptyReader verifies graceful failure on truncation +// at the very first read (magic). +func TestDecodeSnapshot_EmptyReader(t *testing.T) { + if _, err := DecodeSnapshot(bytes.NewReader(nil)); err == nil { + t.Fatal("decoding empty reader must fail") + } +} + +// TestDecodeSnapshot_WrongMagic verifies the magic check rejects stray files. +func TestDecodeSnapshot_WrongMagic(t *testing.T) { + var buf bytes.Buffer + buf.WriteString("BAD!") + _ = binary.Write(&buf, binary.BigEndian, snapshotVersion) + _ = binary.Write(&buf, binary.BigEndian, uint32(0)) + if _, err := DecodeSnapshot(&buf); err == nil { + t.Fatal("wrong magic must fail") + } +} + +// TestDecodeSnapshot_WrongVersion verifies version-bump reads are refused +// — the loader should fall back to full rebuild on any version mismatch. +func TestDecodeSnapshot_WrongVersion(t *testing.T) { + var buf bytes.Buffer + buf.WriteString(snapshotMagic) + _ = binary.Write(&buf, binary.BigEndian, uint32(999)) + if _, err := DecodeSnapshot(&buf); err == nil { + t.Fatal("wrong version must fail") + } +} + +// TestDecodeSnapshot_CRCMismatch verifies bit-rot or partial writes are +// caught before the gob decoder produces silently-wrong state. +func TestDecodeSnapshot_CRCMismatch(t *testing.T) { + in := Snapshot{LastIndexedID: 1, MaxSize: 100, IDF: map[string]float64{}} + var buf bytes.Buffer + if err := EncodeSnapshot(&buf, in); err != nil { + t.Fatalf("encode: %v", err) + } + raw := buf.Bytes() + // Header is 12 bytes (magic+version+crc); flip a payload byte. + if len(raw) < 13 { + t.Fatalf("encoded snapshot too short: %d bytes", len(raw)) + } + raw[12] ^= 0xff + if _, err := DecodeSnapshot(bytes.NewReader(raw)); err == nil { + t.Fatal("CRC mismatch must fail") + } +} + +// TestWriteAtomic_RoundTrip writes a payload and reads it back via the +// public path, then asserts the .tmp sibling is gone. +func TestWriteAtomic_RoundTrip(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "snap.bin") + payload := []byte("hello world") + if err := writeAtomic(p, payload); err != nil { + t.Fatalf("writeAtomic: %v", err) + } + got, err := os.ReadFile(p) + if err != nil { + t.Fatalf("ReadFile: %v", err) + } + if !bytes.Equal(got, payload) { + t.Fatalf("round-trip: got %q, want %q", got, payload) + } + if _, err := os.Stat(p + ".tmp"); !os.IsNotExist(err) { + t.Fatalf(".tmp must be removed after rename, got err=%v", err) + } +} + +// TestIsEXDEV_Detection verifies the helper recognizes wrapped EXDEV from +// os.Rename and ignores arbitrary errors. +func TestIsEXDEV_Detection(t *testing.T) { + le := &os.LinkError{Op: "rename", Old: "a", New: "b", Err: syscall.EXDEV} + if !isEXDEV(le) { + t.Fatal("isEXDEV should detect *os.LinkError{Err: EXDEV}") + } + if isEXDEV(errors.New("other error")) { + t.Fatal("isEXDEV should not flag arbitrary errors") + } + if isEXDEV(nil) { + t.Fatal("isEXDEV(nil) must be false") + } +} + +// TestIndexWriteAndLoadSnapshot exercises the full Index → file → Index +// round trip: build, snapshot, load into a fresh Index, verify state. +func TestIndexWriteAndLoadSnapshot(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "vectordb.snapshot") + + src := New(1000) + src.Add(101, "acme", "checkout", "ERROR", "payment gateway timeout charging customer") + src.Add(102, "acme", "checkout", "ERROR", "payment gateway refused charge insufficient funds") + src.Add(203, "globex", "auth", "WARN", "session token nearing expiry") + if got, want := src.Size(), 3; got != want { + t.Fatalf("seed Size: got %d, want %d", got, want) + } + if got := src.LastIndexedID(); got != 203 { + t.Fatalf("LastIndexedID: got %d, want 203", got) + } + + if err := src.WriteSnapshot(path); err != nil { + t.Fatalf("WriteSnapshot: %v", err) + } + + // Verify file written + .tmp gone + if st, err := os.Stat(path); err != nil { + t.Fatalf("stat snapshot: %v", err) + } else if st.Size() == 0 { + t.Fatal("snapshot file is empty") + } + if _, err := os.Stat(path + ".tmp"); !os.IsNotExist(err) { + t.Fatalf(".tmp must be gone after WriteSnapshot, got err=%v", err) + } + + dst := New(500) // different cap; load should restore src's cap + if err := dst.LoadSnapshot(path); err != nil { + t.Fatalf("LoadSnapshot: %v", err) + } + if got, want := dst.Size(), 3; got != want { + t.Fatalf("loaded Size: got %d, want %d", got, want) + } + if got := dst.LastIndexedID(); got != 203 { + t.Fatalf("loaded LastIndexedID: got %d, want 203", got) + } + // Search should work on the restored index — the IDF table came along + // with the snapshot, so cosine ranking still has rarity weights. + hits := dst.Search("acme", "payment gateway", 5) + if len(hits) != 2 { + t.Fatalf("Search after load: got %d hits, want 2", len(hits)) + } +} + +// TestLoadSnapshot_MissingFile verifies the loader propagates os-level +// errors so callers can distinguish "first start, no snapshot" via +// os.IsNotExist from real corruption. +func TestLoadSnapshot_MissingFile(t *testing.T) { + dir := t.TempDir() + idx := New(100) + err := idx.LoadSnapshot(filepath.Join(dir, "does-not-exist")) + if err == nil { + t.Fatal("LoadSnapshot of missing file must error") + } + if !os.IsNotExist(err) { + t.Fatalf("want os.IsNotExist, got %v", err) + } +} + +// TestSnapshotLoop_FinalWriteOnCancel verifies the loop fires a final +// WriteSnapshot when ctx is cancelled — captures the maximum in-memory +// state at graceful shutdown. +func TestSnapshotLoop_FinalWriteOnCancel(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "snap.bin") + + idx := New(100) + idx.Add(1, "t", "svc", "ERROR", "preserved across shutdown final write") + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + // 1h interval — the loop should never tick during this test, only + // the cancel path fires the write. + idx.SnapshotLoop(ctx, path, 1*time.Hour) + }() + + // Sanity: file does not yet exist. + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatalf("snapshot must not exist before cancel, got err=%v", err) + } + + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("SnapshotLoop did not return within 2s of cancel") + } + + // Verify final write happened. + if st, err := os.Stat(path); err != nil { + t.Fatalf("final snapshot missing after cancel: %v", err) + } else if st.Size() == 0 { + t.Fatal("final snapshot file is empty") + } + + // Round-trip: load into a fresh index and confirm state matches. + dst := New(100) + if err := dst.LoadSnapshot(path); err != nil { + t.Fatalf("LoadSnapshot of final write: %v", err) + } + if dst.Size() != 1 || dst.LastIndexedID() != 1 { + t.Fatalf("loaded state mismatch: Size=%d LastIndexedID=%d", dst.Size(), dst.LastIndexedID()) + } +} + +// TestSnapshotLoop_PeriodicWrite verifies a tick fires WriteSnapshot. +// Uses a tight interval so the test runs in <50ms; the loop fires at +// least once before we cancel + drain. +func TestSnapshotLoop_PeriodicWrite(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "snap.bin") + + idx := New(100) + idx.Add(7, "t", "svc", "ERROR", "periodic snapshot tick body") + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + idx.SnapshotLoop(ctx, path, 10*time.Millisecond) + }() + + // Wait long enough for at least one tick to fire. + deadline := time.Now().Add(500 * time.Millisecond) + for time.Now().Before(deadline) { + if st, err := os.Stat(path); err == nil && st.Size() > 0 { + break + } + time.Sleep(10 * time.Millisecond) + } + + cancel() + <-done + + if _, err := os.Stat(path); err != nil { + t.Fatalf("expected at least one periodic snapshot to land at %s, got err=%v", path, err) + } +} + +// TestSnapshotLoop_DisabledByEmptyPath verifies the no-op path so config +// disable doesn't accidentally start a tight-loop goroutine. +func TestSnapshotLoop_DisabledByEmptyPath(t *testing.T) { + idx := New(100) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + idx.SnapshotLoop(ctx, "", 10*time.Millisecond) + }() + // Loop should return immediately when path is empty — no need to cancel. + select { + case <-done: + case <-time.After(500 * time.Millisecond): + cancel() + t.Fatal("SnapshotLoop with empty path must return immediately") + } + cancel() +} + +// TestLoadSnapshot_CorruptFileLeavesStateAlone verifies that a corrupt +// snapshot does NOT clobber existing index state — the caller is meant to +// log the warning and proceed with a full rebuild. +func TestLoadSnapshot_CorruptFileLeavesStateAlone(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "snap.bin") + if err := os.WriteFile(path, []byte("not a valid snapshot file"), 0o600); err != nil { + t.Fatalf("seed corrupt file: %v", err) + } + idx := New(100) + idx.Add(1, "t", "svc", "ERROR", "preexisting body content") + sizeBefore := idx.Size() + if err := idx.LoadSnapshot(path); err == nil { + t.Fatal("LoadSnapshot of corrupt file must fail") + } + if got := idx.Size(); got != sizeBefore { + t.Fatalf("corrupt load corrupted state: Size went %d → %d", sizeBefore, got) + } +} diff --git a/main.go b/main.go index 123f7fc..ec9fc64 100644 --- a/main.go +++ b/main.go @@ -366,27 +366,67 @@ func main() { go svcGraph.Start(ctxGraph) slog.Info("🕸️ In-memory service graph started (5m window, 30s refresh)") - // 4f. Initialize vector index for semantic log search + // 4f. Initialize vector index for semantic log search. vectorIdx := vectordb.New(cfg.VectorIndexMaxEntries) slog.Info("🔍 Vector index initialized", "max_entries", cfg.VectorIndexMaxEntries) - // Hydrate vector index from recent ERROR/WARN logs on startup (non-blocking). - // Uses appCtx so SIGTERM during boot cancels the query before repo.Close(). - // Hydration is cross-tenant by design: each row lands tagged with its own - // TenantID via vectorIdx.Add so isolation is preserved at query time. The - // previous tenant-scoped GetLogsV2 call silently hydrated only the default - // tenant's rows — non-default tenants lost their warm index on every - // restart. + // Vector index hydration: + // 1) LoadSnapshot — restores the prior process's state in O(file size) + // so find_similar_logs returns useful results in <1s after restart + // instead of the legacy minutes of cold-start blindness. + // 2) ReplayFromDB — picks up any DB rows ingested after the last + // snapshot. Severity-filtered + cursor-paged from LastIndexedID. + // + // Both run in a boot goroutine so a slow disk doesn't delay listener + // startup. SIGTERM during boot cancels via appCtx — bootWG ensures the + // hydrator finishes (or aborts cleanly) before DB close at shutdown. + // Wire snapshot write observer before any WriteSnapshot can fire (the + // hydrator goroutine doesn't write, but SnapshotLoop below will). + vectorIdx.SetSnapshotObserver(metrics.RecordVectorSnapshotWrite) + bootWG.Add(1) go func() { defer bootWG.Done() - recentLogs, err := repo.ListRecentHighSeverityLogsAllTenants(appCtx, "ERROR", time.Now().Add(-24*time.Hour), time.Now(), 5000) - if err == nil { - for _, l := range recentLogs { - vectorIdx.Add(l.ID, l.TenantID, l.ServiceName, l.Severity, l.Body) + if cfg.VectorIndexSnapshotPath != "" { + if err := vectorIdx.LoadSnapshot(cfg.VectorIndexSnapshotPath); err != nil { + if os.IsNotExist(err) { + metrics.RecordVectorSnapshotLoad("missing") + slog.Info("🔍 Vector index: no prior snapshot, will hydrate from DB", "path", cfg.VectorIndexSnapshotPath) + } else { + metrics.RecordVectorSnapshotLoad("corrupt") + slog.Warn("🔍 Vector index: snapshot load failed, will rebuild from DB", "path", cfg.VectorIndexSnapshotPath, "error", err) + } + } else { + metrics.RecordVectorSnapshotLoad("success") + slog.Info("🔍 Vector index: loaded snapshot", "path", cfg.VectorIndexSnapshotPath, "entries", vectorIdx.Size(), "since_id", vectorIdx.LastIndexedID()) } - slog.Info("🔍 Vector index hydrated from recent ERROR logs", "count", len(recentLogs)) } + replayed, err := vectorIdx.ReplayFromDB(appCtx, vectorReplayAdapter{repo: repo}) + metrics.RecordVectorReplayLogs(replayed) + if err != nil { + slog.Warn("🔍 Vector index: tail replay errored", "replayed", replayed, "error", err) + } else if replayed > 0 { + slog.Info("🔍 Vector index: tail replay complete", "rows", replayed, "size", vectorIdx.Size(), "since_id", vectorIdx.LastIndexedID()) + } + }() + + // Periodic snapshot loop. Empty path or non-positive interval disables. + // snapCtx is cancelled in the shutdown sequence right after graphRAG.Stop() + // so the loop's ctx-done branch fires the final write before exit. + snapCtx, snapCancel := context.WithCancel(appCtx) + snapDone := make(chan struct{}) + go func() { + defer close(snapDone) + if cfg.VectorIndexSnapshotPath == "" { + return + } + interval, err := time.ParseDuration(cfg.VectorIndexSnapshotInterval) + if err != nil || interval <= 0 { + slog.Info("🔍 Vector index: periodic snapshot disabled", "interval", cfg.VectorIndexSnapshotInterval) + return + } + slog.Info("🔍 Vector index: periodic snapshot enabled", "interval", interval, "path", cfg.VectorIndexSnapshotPath) + vectorIdx.SnapshotLoop(snapCtx, cfg.VectorIndexSnapshotPath, interval) }() // 4g. Initialize GraphRAG (replaces simple graph for advanced queries) @@ -866,6 +906,19 @@ func main() { graphRAG.Stop() cancelGraphRAG() + // 3a. Cancel vectordb snapshot loop. The loop's ctx.Done branch fires a + // final WriteSnapshot before exit, capturing the maximum in-memory state + // (every Add() that drained from GraphRAG above is persisted). We wait + // briefly so the final snapshot hits disk before DB close — the snapshot + // is independent of repo, but ordered shutdown is cheaper than a stale + // snapshot on the next boot. + snapCancel() + select { + case <-snapDone: + case <-time.After(5 * time.Second): + slog.Warn("vectordb snapshot loop did not finish in 5s; final snapshot may be incomplete") + } + // 3a. Drain async ingest pipeline. gRPC GracefulStop above guarantees // no new Submits land; this blocks until workers finish in-flight // batches so a graceful shutdown doesn't lose buffered ingest. @@ -1000,6 +1053,30 @@ func initTracerProvider(endpoint string) (*sdktrace.TracerProvider, error) { return tp, nil } +// vectorReplayAdapter projects storage.Log into vectordb.ReplayRow so the +// vectordb package stays free of storage imports while still consuming the +// repository's tail-replay query. Lives at the wiring layer because both +// packages can be imported here, but neither imports the other. +type vectorReplayAdapter struct{ repo *storage.Repository } + +func (a vectorReplayAdapter) LogsForVectorReplay(ctx context.Context, sinceID uint, limit int) ([]vectordb.ReplayRow, error) { + logs, err := a.repo.LogsForVectorReplay(ctx, sinceID, limit) + if err != nil { + return nil, err + } + out := make([]vectordb.ReplayRow, len(logs)) + for i, l := range logs { + out[i] = vectordb.ReplayRow{ + ID: l.ID, + Tenant: l.TenantID, + ServiceName: l.ServiceName, + Severity: l.Severity, + Body: l.Body, + } + } + return out, nil +} + func printBanner() { banner := ` ___ _____ _____ _