diff --git a/CHANGELOG.md b/CHANGELOG.md index f905bfa..f7b6363 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,18 @@ All notable changes to SuperBased Observer are documented here. ## [Unreleased] +### Changed + +- **perf(cost): push historical noise filtering into SQLite** — cost + summaries now exclude synthetic-model and all-zero token rows in the source + queries before Go allocates raw rows or runs per-turn rollups. The Go-side + noise guard remains as a defensive backstop. +- **perf(db): gate low-value `observer db vacuum` rebuilds** — the CLI and + dashboard vacuum path now checks reclaimable free pages first, skips the + full-file rebuild below `max(64 MiB, 2% of DB size)`, bounds the rebuild + with a 10-minute context timeout, and exposes `--force` for explicit + operator compaction. + ## [1.32.0] — 2026-08-26 ### Fixed @@ -2982,6 +2994,7 @@ All notable changes to SuperBased Observer are documented here. - **Contact email domain migrated `marmut.app` → `superbased.app`** across documentation, package metadata, and the website source. + ## [1.8.4] — 2026-06-12 ### Changed diff --git a/README.md b/README.md index 8701aae..df5e937 100644 --- a/README.md +++ b/README.md @@ -1056,7 +1056,7 @@ lint-gated policy editor, budget guardrails, evidence downloads) — see the | `observer contract [--json]` | The published stability contract: every MCP tool with its tier (stable / conditional / experimental) plus each adapter's public capability row. `--json` emits the `contract_version`-stamped artifact an integrator pins. Prose: [`docs/mcp-contract.md`](docs/mcp-contract.md). | | `observer adapters [--json]` | The full adapter capability matrix (proxy / surface / hook / MCP / native / token / handoff / attach / resume), generated from the capability registry. | | `observer prune` | Run retention now. | -| `observer db stats\|vacuum\|backup` | Storage manager: per-table size breakdown (index + FTS5 shadow bytes folded in), reclaim free pages with bytes-freed report, online snapshot via `VACUUM INTO` (safe while the daemon runs; refuses overwrite). | +| `observer db stats\|vacuum\|backup` | Storage manager: per-table size breakdown (index + FTS5 shadow bytes folded in), reclaim free pages with bytes-freed report, online snapshot via `VACUUM INTO` (safe while the daemon runs; refuses overwrite). `db vacuum` skips the full rebuild when reclaimable space is below `max(64 MiB, 2% of DB size)`; pass `--force` to compact anyway. | | `observer db import [--dry-run]` | Merge another `observer.db` (a stranded install from another OS / home dir) into this node's. Idempotent single-transaction merge; `--dry-run` rolls the same transaction back for exact counts. Migrates the source first — point it at a copy. | | `observer metrics [--port N]` | Prometheus `/metrics` endpoint. | | `observer export {json\|csv\|xlsx}` | Dump tables for external analysis. | diff --git a/cmd/observer/dbcmd.go b/cmd/observer/dbcmd.go index 7a2522d..06a01e2 100644 --- a/cmd/observer/dbcmd.go +++ b/cmd/observer/dbcmd.go @@ -70,7 +70,10 @@ func newDBStatsCmd() *cobra.Command { } func newDBVacuumCmd() *cobra.Command { - var configPath string + var ( + configPath string + force bool + ) cmd := &cobra.Command{ Use: "vacuum", Short: "Rebuild the database file to reclaim free pages (needs the write lock; pick a quiet moment)", @@ -80,19 +83,40 @@ func newDBVacuumCmd() *cobra.Command { return err } defer cleanup() - fmt.Fprintln(cmd.OutOrStdout(), "vacuum: rebuilding — this can take a while on large databases…") - freed, err := db.Vacuum(cmd.Context(), database) + opts := db.VacuumOptions{ + MinReclaimableBytes: defaultVacuumMinReclaimableBytes, + MinReclaimableRatio: defaultVacuumMinReclaimableRatio, + Timeout: defaultVacuumTimeout, + } + if force { + opts.MinReclaimableBytes = 0 + opts.MinReclaimableRatio = 0 + } + fmt.Fprintln(cmd.OutOrStdout(), "vacuum: checking reclaimable free pages…") + res, err := db.VacuumWithOptions(cmd.Context(), database, opts) if err != nil { return err } - fmt.Fprintf(cmd.OutOrStdout(), "vacuum complete: %s reclaimed\n", fmtBytesIEC(freed)) + if !res.Ran { + fmt.Fprintf(cmd.OutOrStdout(), "vacuum skipped: %s reclaimable is below the %s threshold for this %s database (use --force to run anyway)\n", + fmtBytesIEC(res.ReclaimableBytes), fmtBytesIEC(res.RequiredBytes), fmtBytesIEC(res.TotalBytes)) + return nil + } + fmt.Fprintf(cmd.OutOrStdout(), "vacuum complete: %s reclaimed\n", fmtBytesIEC(res.FreedBytes)) return nil }, } cmd.Flags().StringVar(&configPath, "config", "", "Path to config.toml") + cmd.Flags().BoolVar(&force, "force", false, "Run VACUUM even when reclaimable free pages are below the default threshold") return cmd } +const ( + defaultVacuumMinReclaimableBytes int64 = 64 << 20 + defaultVacuumMinReclaimableRatio = 0.02 + defaultVacuumTimeout = 10 * time.Minute +) + func newDBBackupCmd() *cobra.Command { var ( configPath string diff --git a/internal/db/storage.go b/internal/db/storage.go index 84f73b0..3c1f5b9 100644 --- a/internal/db/storage.go +++ b/internal/db/storage.go @@ -8,6 +8,7 @@ import ( "path/filepath" "sort" "strings" + "time" ) // Storage manager primitives (usability arc P6.8). This file is the @@ -161,22 +162,90 @@ func resolveOwner(name string, owners map[string]string, ftsTables []string) str return name } +// VacuumOptions controls the guarded VACUUM path. +type VacuumOptions struct { + // MinReclaimableBytes skips the rebuild when the freelist is smaller + // than this threshold. Zero always runs. + MinReclaimableBytes int64 + // MinReclaimableRatio skips the rebuild when reclaimable bytes are less + // than this fraction of the current DB size. Zero disables the ratio gate. + MinReclaimableRatio float64 + // Timeout bounds the rebuild. Zero leaves the caller's context as-is. + Timeout time.Duration +} + +// VacuumResult reports whether VACUUM ran and how many bytes it reclaimed. +type VacuumResult struct { + Ran bool + FreedBytes int64 + ReclaimableBytes int64 + TotalBytes int64 + RequiredBytes int64 +} + // Vacuum rebuilds the database file, returning the freed bytes // (before − after, from page accounting). VACUUM needs the write lock // and temporarily doubles disk usage; run it at a quiet moment. func Vacuum(ctx context.Context, database *sql.DB) (freedBytes int64, err error) { - before, err := fileBytes(ctx, database) + res, err := VacuumWithOptions(ctx, database, VacuumOptions{}) if err != nil { return 0, err } + return res.FreedBytes, nil +} + +// VacuumWithOptions is Vacuum plus an optional reclaimable-byte gate and +// timeout. It lets automatic/UI-triggered maintenance avoid a full-file +// rebuild when SQLite has little or nothing to reclaim, while preserving the +// old unconditional behavior when options are zero. +func VacuumWithOptions(ctx context.Context, database *sql.DB, opts VacuumOptions) (VacuumResult, error) { + if opts.Timeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, opts.Timeout) + defer cancel() + } + reclaimable, err := reclaimableBytes(ctx, database) + if err != nil { + return VacuumResult{}, err + } + total, err := fileBytes(ctx, database) + if err != nil { + return VacuumResult{}, err + } + required := vacuumRequiredBytes(total, opts) + if required > 0 && reclaimable < required { + return VacuumResult{ + Ran: false, + ReclaimableBytes: reclaimable, + TotalBytes: total, + RequiredBytes: required, + }, nil + } if _, err := database.ExecContext(ctx, "VACUUM"); err != nil { - return 0, fmt.Errorf("db.Vacuum: %w", err) + return VacuumResult{}, fmt.Errorf("db.Vacuum: %w", err) } after, err := fileBytes(ctx, database) if err != nil { - return 0, err + return VacuumResult{}, err + } + return VacuumResult{ + Ran: true, + FreedBytes: total - after, + ReclaimableBytes: reclaimable, + TotalBytes: total, + RequiredBytes: required, + }, nil +} + +func vacuumRequiredBytes(total int64, opts VacuumOptions) int64 { + required := opts.MinReclaimableBytes + if opts.MinReclaimableRatio > 0 && total > 0 { + ratioRequired := int64(float64(total) * opts.MinReclaimableRatio) + if ratioRequired > required { + required = ratioRequired + } } - return before - after, nil + return required } // BackupInto writes a consistent snapshot of the live database to @@ -209,3 +278,14 @@ func fileBytes(ctx context.Context, database *sql.DB) (int64, error) { } return pageSize * pageCount, nil } + +func reclaimableBytes(ctx context.Context, database *sql.DB) (int64, error) { + var pageSize, freelist int64 + if err := database.QueryRowContext(ctx, "PRAGMA page_size").Scan(&pageSize); err != nil { + return 0, fmt.Errorf("db: page_size: %w", err) + } + if err := database.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&freelist); err != nil { + return 0, fmt.Errorf("db: freelist_count: %w", err) + } + return pageSize * freelist, nil +} diff --git a/internal/db/storage_test.go b/internal/db/storage_test.go index 2700b05..ff65ac4 100644 --- a/internal/db/storage_test.go +++ b/internal/db/storage_test.go @@ -137,3 +137,42 @@ func TestVacuumAndBackupInto(t *testing.T) { t.Errorf("freed = %d, want >= -%d (one page of VACUUM rounding)", freed, sqlitePageSize) } } + +func TestVacuumWithOptionsSkipsBelowReclaimableThreshold(t *testing.T) { + ctx := context.Background() + database, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "s.db")}) + if err != nil { + t.Fatal(err) + } + defer database.Close() + + res, err := VacuumWithOptions(ctx, database, VacuumOptions{MinReclaimableBytes: 1 << 30}) + if err != nil { + t.Fatalf("VacuumWithOptions: %v", err) + } + if res.Ran { + t.Fatal("VacuumWithOptions ran despite reclaimable bytes being below threshold") + } + if res.ReclaimableBytes >= 1<<30 { + t.Fatalf("test database unexpectedly has %d reclaimable bytes", res.ReclaimableBytes) + } + if res.FreedBytes != 0 { + t.Errorf("skipped vacuum freed bytes = %d, want 0", res.FreedBytes) + } +} + +func TestVacuumRequiredBytesUsesLargerOfAbsoluteAndRatioThresholds(t *testing.T) { + opts := VacuumOptions{ + MinReclaimableBytes: 64 << 20, + MinReclaimableRatio: 0.02, + } + if got, want := vacuumRequiredBytes(1<<30, opts), int64(64<<20); got != want { + t.Errorf("small DB threshold = %d, want %d", got, want) + } + if got, want := vacuumRequiredBytes(40<<30, opts), int64(858993459); got != want { + t.Errorf("40 GiB DB threshold = %d, want %d", got, want) + } + if got := vacuumRequiredBytes(40<<30, VacuumOptions{}); got != 0 { + t.Errorf("force/zero gate threshold = %d, want 0", got) + } +} diff --git a/internal/intelligence/cost/summary.go b/internal/intelligence/cost/summary.go index 1268a22..80882ad 100644 --- a/internal/intelligence/cost/summary.go +++ b/internal/intelligence/cost/summary.go @@ -729,6 +729,7 @@ func (e *Engine) loadProxyRows(ctx context.Context, db *sql.DB, opts Options, si where = append(where, "at.timestamp >= ?") args = append(args, since.UTC().Format(time.RFC3339Nano)) } + where = append(where, proxyNoiseWhere("at")) if !opts.Until.IsZero() { where = append(where, "at.timestamp < ?") args = append(args, opts.Until.UTC().Format(time.RFC3339Nano)) @@ -815,6 +816,7 @@ func (e *Engine) loadJSONLRows(ctx context.Context, db *sql.DB, opts Options, si where = append(where, "tu.timestamp >= ?") args = append(args, since.UTC().Format(time.RFC3339Nano)) } + where = append(where, jsonlNoiseWhere("tu")) if !opts.Until.IsZero() { where = append(where, "tu.timestamp < ?") args = append(args, opts.Until.UTC().Format(time.RFC3339Nano)) @@ -900,6 +902,7 @@ func (e *Engine) loadSummaryCallRows(ctx context.Context, db *sql.DB, opts Optio where = append(where, "timestamp >= ?") args = append(args, since.UTC().Format(time.RFC3339Nano)) } + where = append(where, summaryCallNoiseWhere()) if !opts.Until.IsZero() { where = append(where, "timestamp < ?") args = append(args, opts.Until.UTC().Format(time.RFC3339Nano)) @@ -953,6 +956,41 @@ func (e *Engine) loadSummaryCallRows(ctx context.Context, db *sql.DB, opts Optio return out, nil } +func proxyNoiseWhere(alias string) string { + return fmt.Sprintf(`COALESCE(%[1]s.model, '') != '' AND ( + COALESCE(%[1]s.input_tokens, 0) != 0 OR + COALESCE(%[1]s.output_tokens, 0) != 0 OR + COALESCE(%[1]s.cache_read_tokens, 0) != 0 OR + COALESCE(%[1]s.cache_creation_tokens, 0) != 0 OR + COALESCE(%[1]s.cache_creation_1h_tokens, 0) != 0 OR + COALESCE(%[1]s.web_search_requests, 0) != 0 OR + COALESCE(%[1]s.cost_usd, 0) != 0 + )`, alias) +} + +func jsonlNoiseWhere(alias string) string { + return fmt.Sprintf(`COALESCE(%[1]s.model, '') != '' AND ( + COALESCE(%[1]s.input_tokens, 0) != 0 OR + COALESCE(%[1]s.output_tokens, 0) != 0 OR + COALESCE(%[1]s.cache_read_tokens, 0) != 0 OR + COALESCE(%[1]s.cache_creation_tokens, 0) != 0 OR + COALESCE(%[1]s.cache_creation_1h_tokens, 0) != 0 OR + COALESCE(%[1]s.reasoning_tokens, 0) != 0 OR + COALESCE(%[1]s.web_search_requests, 0) != 0 OR + COALESCE(%[1]s.estimated_cost_usd, 0) != 0 + )`, alias) +} + +func summaryCallNoiseWhere() string { + return `COALESCE(model, '') != '' AND ( + COALESCE(input_tokens, 0) != 0 OR + COALESCE(output_tokens, 0) != 0 OR + COALESCE(cache_read_tokens, 0) != 0 OR + COALESCE(cache_creation_tokens, 0) != 0 OR + COALESCE(cost_usd, 0) != 0 + )` +} + // bucket holds the in-progress aggregation for one group key. type bucket struct { tokens TokenBundle