diff --git a/cmd/cosift/runtime_metrics.go b/cmd/cosift/runtime_metrics.go new file mode 100644 index 0000000..28511ce --- /dev/null +++ b/cmd/cosift/runtime_metrics.go @@ -0,0 +1,56 @@ +package main + +import ( + "math" + "runtime/metrics" +) + +type runtimeStats struct { + HeapObjects, HeapLive, HeapGoal, MemLimit, MemTotal, GCCycles, Goroutines int64 +} + +// readRuntimeStats samples runtime/metrics (no STW). MemLimit is -1 when no +// GOMEMLIMIT is set. +func readRuntimeStats() runtimeStats { + samples := []metrics.Sample{ + {Name: "/memory/classes/heap/objects:bytes"}, + {Name: "/gc/heap/live:bytes"}, + {Name: "/gc/heap/goal:bytes"}, + {Name: "/gc/memory/limit:bytes"}, + {Name: "/memory/classes/total:bytes"}, + {Name: "/gc/cycles/total:gc-cycles"}, + {Name: "/sched/goroutines:goroutines"}, + } + metrics.Read(samples) + u := func(i int) int64 { + if samples[i].Value.Kind() != metrics.KindUint64 { + return 0 + } + return int64(samples[i].Value.Uint64()) + } + rs := runtimeStats{ + HeapObjects: u(0), + HeapLive: u(1), + HeapGoal: u(2), + MemLimit: u(3), + MemTotal: u(4), + GCCycles: u(5), + Goroutines: u(6), + } + if rs.MemLimit <= 0 || rs.MemLimit == math.MaxInt64 { + rs.MemLimit = -1 + } + return rs +} + +func (rs runtimeStats) statsMap() map[string]any { + return map[string]any{ + "heap_objects_bytes": rs.HeapObjects, + "heap_live_bytes": rs.HeapLive, + "heap_goal_bytes": rs.HeapGoal, + "mem_limit_bytes": rs.MemLimit, + "mem_total_bytes": rs.MemTotal, + "gc_cycles": rs.GCCycles, + "goroutines": rs.Goroutines, + } +} diff --git a/cmd/cosift/runtime_metrics_test.go b/cmd/cosift/runtime_metrics_test.go new file mode 100644 index 0000000..05da79f --- /dev/null +++ b/cmd/cosift/runtime_metrics_test.go @@ -0,0 +1,69 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "runtime" + "strconv" + "strings" + "testing" +) + +func TestStatsAndMetricsExposeRuntimeMemory(t *testing.T) { + f := populatedPebbleStore(t) + srv := f.makeServer(nil) + runtime.GC() + + body, err := srv.buildStatsBody(context.Background()) + if err != nil { + t.Fatalf("buildStatsBody: %v", err) + } + var out struct { + Runtime map[string]float64 `json:"runtime"` + } + if err := json.Unmarshal(body, &out); err != nil { + t.Fatalf("unmarshal: %v", err) + } + rt := out.Runtime + for _, k := range []string{"heap_objects_bytes", "heap_live_bytes", "heap_goal_bytes", "mem_limit_bytes", "mem_total_bytes", "gc_cycles", "goroutines"} { + if _, ok := rt[k]; !ok { + t.Errorf("/stats runtime missing %q", k) + } + } + if !(rt["heap_goal_bytes"] >= rt["heap_live_bytes"] && rt["heap_live_bytes"] > 0) { + t.Errorf("want heap_goal >= heap_live > 0, got goal=%v live=%v", rt["heap_goal_bytes"], rt["heap_live_bytes"]) + } + if rt["mem_total_bytes"] < rt["heap_objects_bytes"] || rt["goroutines"] < 1 || rt["gc_cycles"] < 1 { + t.Errorf("implausible runtime stats: %v", rt) + } + if rt["mem_limit_bytes"] != -1 && rt["mem_limit_bytes"] <= 0 { + t.Errorf("mem_limit_bytes: got %v", rt["mem_limit_bytes"]) + } + + rec := httptest.NewRecorder() + srv.handleMetrics(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("/metrics: code %d", rec.Code) + } + vals := map[string]float64{} + for _, line := range strings.Split(rec.Body.String(), "\n") { + if strings.HasPrefix(line, "cosift_go_") { + name, v, _ := strings.Cut(line, " ") + n, err := strconv.ParseFloat(v, 64) + if err != nil { + t.Errorf("parse %q: %v", line, err) + } + vals[name] = n + } + } + for _, k := range []string{"cosift_go_heap_objects_bytes", "cosift_go_heap_live_bytes", "cosift_go_heap_goal_bytes", "cosift_go_mem_limit_bytes", "cosift_go_mem_total_bytes", "cosift_go_gc_cycles_total", "cosift_go_goroutines"} { + if _, ok := vals[k]; !ok { + t.Errorf("/metrics missing %q", k) + } + } + if !(vals["cosift_go_heap_goal_bytes"] >= vals["cosift_go_heap_live_bytes"] && vals["cosift_go_heap_live_bytes"] > 0) { + t.Errorf("want heap_goal >= heap_live > 0, got %v", vals) + } +} diff --git a/cmd/cosift/serve_admin.go b/cmd/cosift/serve_admin.go index b4f570e..65bd379 100644 --- a/cmd/cosift/serve_admin.go +++ b/cmd/cosift/serve_admin.go @@ -91,9 +91,7 @@ func (s *pebbleHTTP) handleCheckpoint(w http.ResponseWriter, r *http.Request) { writeProblem(w, http.StatusUnauthorized, "missing or invalid admin token") return } - if rc := http.NewResponseController(w); rc != nil { - _ = rc.SetWriteDeadline(time.Time{}) - } + liftWriteDeadline(w) base := os.Getenv("COSIFT_CHECKPOINT_DIR") if base == "" { base = "/tmp" @@ -254,9 +252,7 @@ func (s *pebbleHTTP) handleEvalQuick(w http.ResponseWriter, r *http.Request) { // this long-running admin endpoint via ResponseController. Pairs with // bounded-parallel dispatch below so the 10-query batch finishes in // ~one chat-LLM round-trip instead of ten. - if rc := http.NewResponseController(w); rc != nil { - _ = rc.SetWriteDeadline(time.Time{}) - } + liftWriteDeadline(w) type queryResult struct { Query string `json:"query"` Verdict string `json:"verdict"` // answered | no_info | empty | error @@ -504,9 +500,7 @@ func (s *pebbleHTTP) handleHNSWCompact(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusAccepted, map[string]any{"status": "started", "watch": "/stats hnsw_compact"}) return } - if rc := http.NewResponseController(w); rc != nil { - _ = rc.SetWriteDeadline(time.Time{}) - } + liftWriteDeadline(w) <-done resp, code := j.resultJSON() writeJSON(w, code, resp) diff --git a/cmd/cosift/serve_answer.go b/cmd/cosift/serve_answer.go index eb76373..77ebedb 100644 --- a/cmd/cosift/serve_answer.go +++ b/cmd/cosift/serve_answer.go @@ -678,6 +678,8 @@ func (r *recordingWriter) Flush() { } } +func (r *recordingWriter) Unwrap() http.ResponseWriter { return r.ResponseWriter } + // wantsSSE reports whether the request opts into Server-Sent Events, // either via ?stream=true or an Accept: text/event-stream header. Both // /answer, /query, and /research check the same envelope. @@ -724,9 +726,7 @@ func newAnswerSSE(w http.ResponseWriter, start time.Time) *answerSSE { w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("X-Accel-Buffering", "no") - if rc := http.NewResponseController(w); rc != nil { - _ = rc.SetWriteDeadline(time.Time{}) - } + liftWriteDeadline(w) w.WriteHeader(http.StatusOK) return &answerSSE{w: w, flusher: flusher, start: start, last: start} } diff --git a/cmd/cosift/serve_crawl.go b/cmd/cosift/serve_crawl.go index a8efc13..ce06638 100644 --- a/cmd/cosift/serve_crawl.go +++ b/cmd/cosift/serve_crawl.go @@ -98,9 +98,7 @@ func (s *pebbleHTTP) handleFrontierClear(w http.ResponseWriter, r *http.Request) writeProblem(w, http.StatusUnauthorized, "missing or invalid peer token") return } - if rc := http.NewResponseController(w); rc != nil { - _ = rc.SetWriteDeadline(time.Time{}) - } + liftWriteDeadline(w) if err := s.store.ClearFrontier(r.Context()); err != nil { writeProblem(w, http.StatusInternalServerError, err.Error()) return diff --git a/cmd/cosift/serve_setup.go b/cmd/cosift/serve_setup.go index ebba9a8..9dbb4ee 100644 --- a/cmd/cosift/serve_setup.go +++ b/cmd/cosift/serve_setup.go @@ -1566,6 +1566,17 @@ func (s *statusCapturingWriter) Flush() { } } +func (s *statusCapturingWriter) Unwrap() http.ResponseWriter { return s.ResponseWriter } + +var liftWriteDeadlineWarn sync.Once + +// liftWriteDeadline clears the server WriteTimeout for a long-running handler. +func liftWriteDeadline(w http.ResponseWriter) { + if err := http.NewResponseController(w).SetWriteDeadline(time.Time{}); err != nil { + liftWriteDeadlineWarn.Do(func() { log.Printf("warning: cannot lift write deadline: %v", err) }) + } +} + //go:embed assets/landing.html var landingHTML []byte diff --git a/cmd/cosift/serve_stats.go b/cmd/cosift/serve_stats.go index 6737900..e59096b 100644 --- a/cmd/cosift/serve_stats.go +++ b/cmd/cosift/serve_stats.go @@ -474,6 +474,7 @@ func (s *pebbleHTTP) buildStatsBody(ctx context.Context) ([]byte, error) { } } out["dense_resolution_drops"] = s.denseResolutionDrops.Load() + out["runtime"] = readRuntimeStats().statsMap() return json.Marshal(out) } @@ -609,6 +610,28 @@ func (s *pebbleHTTP) handleMetrics(w http.ResponseWriter, r *http.Request) { fmt.Fprintf(w, "# TYPE cosift_hnsw_compact_running gauge\n") fmt.Fprintf(w, "cosift_hnsw_compact_running %d\n", running) } + rs := readRuntimeStats() + fmt.Fprintf(w, "# HELP cosift_go_heap_objects_bytes Bytes occupied by heap objects (live + not yet collected).\n") + fmt.Fprintf(w, "# TYPE cosift_go_heap_objects_bytes gauge\n") + fmt.Fprintf(w, "cosift_go_heap_objects_bytes %d\n", rs.HeapObjects) + fmt.Fprintf(w, "# HELP cosift_go_heap_live_bytes Heap bytes marked live by the last GC.\n") + fmt.Fprintf(w, "# TYPE cosift_go_heap_live_bytes gauge\n") + fmt.Fprintf(w, "cosift_go_heap_live_bytes %d\n", rs.HeapLive) + fmt.Fprintf(w, "# HELP cosift_go_heap_goal_bytes Heap size the next GC cycle targets.\n") + fmt.Fprintf(w, "# TYPE cosift_go_heap_goal_bytes gauge\n") + fmt.Fprintf(w, "cosift_go_heap_goal_bytes %d\n", rs.HeapGoal) + fmt.Fprintf(w, "# HELP cosift_go_mem_limit_bytes GOMEMLIMIT soft memory limit, -1 when unset.\n") + fmt.Fprintf(w, "# TYPE cosift_go_mem_limit_bytes gauge\n") + fmt.Fprintf(w, "cosift_go_mem_limit_bytes %d\n", rs.MemLimit) + fmt.Fprintf(w, "# HELP cosift_go_mem_total_bytes Total memory mapped by the Go runtime.\n") + fmt.Fprintf(w, "# TYPE cosift_go_mem_total_bytes gauge\n") + fmt.Fprintf(w, "cosift_go_mem_total_bytes %d\n", rs.MemTotal) + fmt.Fprintf(w, "# HELP cosift_go_gc_cycles_total Completed GC cycles since process start.\n") + fmt.Fprintf(w, "# TYPE cosift_go_gc_cycles_total counter\n") + fmt.Fprintf(w, "cosift_go_gc_cycles_total %d\n", rs.GCCycles) + fmt.Fprintf(w, "# HELP cosift_go_goroutines Live goroutines.\n") + fmt.Fprintf(w, "# TYPE cosift_go_goroutines gauge\n") + fmt.Fprintf(w, "cosift_go_goroutines %d\n", rs.Goroutines) // PromQL // rate(cosift_request_duration_seconds_sum) / rate(cosift_requests_total) // gives mean latency in any window. Labels = path; misrouted calls (404) diff --git a/cmd/cosift/write_deadline_test.go b/cmd/cosift/write_deadline_test.go new file mode 100644 index 0000000..d68e77c --- /dev/null +++ b/cmd/cosift/write_deadline_test.go @@ -0,0 +1,62 @@ +package main + +import ( + "errors" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestLiftWriteDeadlineThroughMiddleware(t *testing.T) { + srv := &pebbleHTTP{rl: newRateLimiterFromEnv()} + h := srv.count(srv.rateLimit(func(w http.ResponseWriter, r *http.Request) { + liftWriteDeadline(w) + time.Sleep(500 * time.Millisecond) + _, _ = w.Write([]byte("slow-ok")) + })) + ts := httptest.NewUnstartedServer(h) + ts.Config.WriteTimeout = 200 * time.Millisecond + ts.Start() + defer ts.Close() + + resp, err := http.Get(ts.URL + "/slow") + if err != nil { + t.Fatalf("GET: %v", err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + if string(body) != "slow-ok" { + t.Fatalf("body: got %q want %q", body, "slow-ok") + } +} + +type noUnwrapWriter struct{ http.ResponseWriter } + +func TestWrappedWritersSupportSetWriteDeadline(t *testing.T) { + var scErr, rwErr, bareErr error + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + scErr = http.NewResponseController(&statusCapturingWriter{ResponseWriter: w}).SetWriteDeadline(time.Time{}) + rwErr = http.NewResponseController(&recordingWriter{ResponseWriter: w}).SetWriteDeadline(time.Time{}) + bareErr = http.NewResponseController(&noUnwrapWriter{w}).SetWriteDeadline(time.Time{}) + })) + defer ts.Close() + resp, err := http.Get(ts.URL) + if err != nil { + t.Fatalf("GET: %v", err) + } + resp.Body.Close() + if scErr != nil { + t.Errorf("statusCapturingWriter: %v", scErr) + } + if rwErr != nil { + t.Errorf("recordingWriter: %v", rwErr) + } + if !errors.Is(bareErr, http.ErrNotSupported) { + t.Errorf("wrapper without Unwrap: got %v want ErrNotSupported", bareErr) + } +} diff --git a/docs/ENV.md b/docs/ENV.md index da8f5f9..10b8915 100644 --- a/docs/ENV.md +++ b/docs/ENV.md @@ -226,6 +226,10 @@ not re-enable them without the confidentiality decision in that section. | `COSIFT_PEBBLE_MEMTABLE_MB` | int (MB) | `32` | Pebble memtable size in MB; must be `> 0`. | `internal/store/pebble.go:154` | | `COSIFT_PEBBLE_MEMTABLES` | int | `2` | Memtable count; `MemTableStopWritesThreshold = value + 2`. Must be `> 0`. | `internal/store/pebble.go:155` | | `COSIFT_PEBBLE_COMPACTIONS` | int | `1` | Pebble `MaxConcurrentCompactions`. `1` (Pebble's default) serializes every background compaction on one slot — the cause of the 128K-SSTable pile and the 12 MB/s full persist observed on the production box; `4` is a sane value on a 64-core host. Does not parallelise the manual range compaction that clears a node slot (one contiguous range = one job). Must be `> 0`. | `internal/store/pebble.go` (`openPebble`) | +| `COSIFT_PEBBLE_TARGET_FILE_MB` | int (MB) | unset → Pebble default (`2`, doubling per level) | Sets `Levels[i].TargetFileSize` for L0–L6: L0 = value, doubling each level. Larger files mean fewer SSTables per level and fewer, bigger compactions. Only applied when set. | `internal/store/pebble.go` (`applyLevelOptsFromEnv`) | +| `COSIFT_PEBBLE_LBASE_MB` | int (MB) | unset → Pebble default (`64`) | Pebble `LBaseMaxBytes`: size of the first non-L0 level; each deeper level is 10× the previous. Only applied when set. | `internal/store/pebble.go` (`applyLevelOptsFromEnv`) | +| `COSIFT_PEBBLE_L0_COMPACTION_FILES` | int | unset → Pebble default (`500`) | Pebble `L0CompactionFileThreshold`: L0 file count that triggers an L0→Lbase compaction. Only applied when set. | `internal/store/pebble.go` (`applyLevelOptsFromEnv`) | +| `COSIFT_PEBBLE_L0_STOP_WRITES` | int | unset → Pebble default (`12`) | Pebble `L0StopWritesThreshold`: L0 sublevel count at which writes stall until compaction catches up. Must be ≥ Pebble's `L0CompactionThreshold` (`4`). Only applied when set. | `internal/store/pebble.go` (`applyLevelOptsFromEnv`) | | `COSIFT_PEBBLE_SYNC` | bool (`"false"` disables) | unset → `Sync` (fsync each commit) | Set to `"false"` to use `NoSync` writes (skips per-commit fsync — faster crawls, drops durability vs OS crash; WAL still written so process-crash durability holds). | `internal/store/pebble.go:176` | --- diff --git a/internal/index/hnsw_compact.go b/internal/index/hnsw_compact.go index d49d0e1..521f20c 100644 --- a/internal/index/hnsw_compact.go +++ b/internal/index/hnsw_compact.go @@ -70,7 +70,7 @@ func (h *HNSW) Rebuild() *HNSW { func (h *HNSW) Compact() (removed int) { h.persistMu.Lock() defer h.persistMu.Unlock() - return h.compactLocked() + return h.compactLocked(nil) } // DirtyCount reports how many nodes await an incremental persist. @@ -82,7 +82,7 @@ func (h *HNSW) DirtyCount() int { // CompactProgress is the observable state of a CompactPersist run. type CompactProgress struct { - Phase string // compact | persist | cleanup | done | error + Phase string // compact | compact:url-index | compact:entry-point | persist | cleanup | done | error NodesBefore, NodesAfter, Removed int Written, Total int // persist progress } @@ -116,7 +116,9 @@ func (h *HNSW) CompactPersist(ctx context.Context, ps *store.PebbleStore, forceP res := CompactResult{NodesBefore: h.Len()} report(CompactProgress{Phase: "compact", NodesBefore: res.NodesBefore}) t0 := time.Now() - res.Removed = h.compactLocked() + res.Removed = h.compactLocked(func(phase string) { + report(CompactProgress{Phase: phase, NodesBefore: res.NodesBefore}) + }) res.CompactDur = time.Since(t0) res.NodesAfter = h.Len() base := CompactProgress{NodesBefore: res.NodesBefore, NodesAfter: res.NodesAfter, Removed: res.Removed} @@ -174,9 +176,12 @@ func (h *HNSW) CompactPersist(ctx context.Context, ps *store.PebbleStore, forceP return res, nil } -func (h *HNSW) compactLocked() (removed int) { +func (h *HNSW) compactLocked(phase func(string)) (removed int) { h.mu.Lock() defer h.mu.Unlock() + if phase == nil { + phase = func(string) {} + } if len(h.nodes) == 0 { return 0 @@ -235,6 +240,9 @@ func (h *HNSW) compactLocked() (removed int) { // 3. Rebuild the URL index and counters; every id changed, so the dirty // set is meaningless until the caller's full persist rewrites the graph. + phase("compact:url-index") + t0 := time.Now() + log.Printf("hnsw compact: rebuilding url index for %d nodes", len(newNodes)) h.nodes = newNodes h.codes = newCodes h.byURL = make(map[string][]int32, len(h.byURL)) @@ -244,20 +252,20 @@ func (h *HNSW) compactLocked() (removed int) { h.valid = len(h.nodes) h.dirty = make(map[int32]struct{}) h.renumbered = true + log.Printf("hnsw compact: url index rebuilt in %s", time.Since(t0).Round(time.Millisecond)) // 4. Pick new entry point as the highest-level surviving node. - if len(h.nodes) == 0 { - h.entryPoint = -1 - h.maxLevel = 0 - return removed - } - h.entryPoint = 0 - h.maxLevel = h.nodes[0].level - for i := 1; i < len(h.nodes); i++ { - if h.nodes[i].level > h.maxLevel { + phase("compact:entry-point") + t0 = time.Now() + log.Printf("hnsw compact: scanning %d nodes for entry point", len(h.nodes)) + h.entryPoint = -1 + h.maxLevel = 0 + for i := range h.nodes { + if i == 0 || h.nodes[i].level > h.maxLevel { h.entryPoint = i h.maxLevel = h.nodes[i].level } } + log.Printf("hnsw compact: entry point %d (level %d) in %s", h.entryPoint, h.maxLevel, time.Since(t0).Round(time.Millisecond)) return removed } diff --git a/internal/index/hnsw_compact_test.go b/internal/index/hnsw_compact_test.go index c68f185..83fad5c 100644 --- a/internal/index/hnsw_compact_test.go +++ b/internal/index/hnsw_compact_test.go @@ -150,4 +150,35 @@ func TestHNSWCompactProgressLogs(t *testing.T) { if !strings.Contains(out, "hnsw compact: rewiring neighbors") { t.Errorf("missing rewiring progress line in:\n%s", out) } + for _, want := range []string{"rebuilding url index", "url index rebuilt in", "for entry point", "hnsw compact: entry point"} { + if !strings.Contains(out, want) { + t.Errorf("missing %q progress line in:\n%s", want, out) + } + } +} + +func TestHNSWCompactPersistReportsSubPhases(t *testing.T) { + ps := openTestStore(t) + ctx := context.Background() + h := buildTestHNSW(120, 8, 3, 5) + if err := h.Persist(ctx, ps); err != nil { + t.Fatal(err) + } + h.MarkURLPassagesInvalid("https://x/3") + var phases []string + res, err := h.CompactPersist(ctx, ps, false, nil, func(p CompactProgress) { + if n := len(phases); n == 0 || phases[n-1] != p.Phase { + phases = append(phases, p.Phase) + } + }) + if err != nil || res.Removed != 1 { + t.Fatalf("compact: %+v %v", res, err) + } + want := []string{"compact", "compact:url-index", "compact:entry-point", "persist", "cleanup", "done"} + if fmt.Sprint(phases) != fmt.Sprint(want) { + t.Fatalf("phases: got %v want %v", phases, want) + } + if h.entryPoint < 0 || h.entryPoint >= h.Len() { + t.Fatalf("entry point %d out of range", h.entryPoint) + } } diff --git a/internal/store/pebble.go b/internal/store/pebble.go index 18ab6e8..d9bdc68 100644 --- a/internal/store/pebble.go +++ b/internal/store/pebble.go @@ -28,6 +28,7 @@ import ( "encoding/gob" "errors" "fmt" + "log" "math" "net/url" "os" @@ -131,6 +132,10 @@ const ( // COSIFT_PEBBLE_CACHE_MB — block cache size in MB (default 128) // COSIFT_PEBBLE_MEMTABLE_MB — single memtable size in MB (default 32) // COSIFT_PEBBLE_MEMTABLES — max memtables in memory (default 2) +// COSIFT_PEBBLE_TARGET_FILE_MB — L0 TargetFileSize in MB, doubling per level (default: Pebble's 2) +// COSIFT_PEBBLE_LBASE_MB — LBaseMaxBytes in MB (default: Pebble's 64) +// COSIFT_PEBBLE_L0_COMPACTION_FILES — L0CompactionFileThreshold (default: Pebble's 500) +// COSIFT_PEBBLE_L0_STOP_WRITES — L0StopWritesThreshold (default: Pebble's 12) // // Total Pebble memory ceiling ≈ cache + memtables × memtable_size, so the // defaults pin Pebble at roughly 128 + 2×32 = 192 MB. Real working set @@ -164,6 +169,7 @@ func openPebble(path string, readOnly bool) (*PebbleStore, error) { MaxConcurrentCompactions: func() int { return compactions }, ReadOnly: readOnly, } + applyLevelOptsFromEnv(opts) db, err := pebble.Open(path, opts) if err != nil { return nil, fmt.Errorf("pebble.Open(%s): %w", path, err) @@ -248,6 +254,35 @@ func (p *PebbleStore) Checkpoint(destDir string) error { return p.db.Checkpoint(destDir) } +// applyLevelOptsFromEnv sets LSM shape options only when the matching env +// var is set, so unset leaves Pebble's defaults untouched. +func applyLevelOptsFromEnv(opts *pebble.Options) { + targetMB := envInt("COSIFT_PEBBLE_TARGET_FILE_MB", 0) + lbaseMB := envInt("COSIFT_PEBBLE_LBASE_MB", 0) + l0Files := envInt("COSIFT_PEBBLE_L0_COMPACTION_FILES", 0) + l0Stop := envInt("COSIFT_PEBBLE_L0_STOP_WRITES", 0) + if targetMB == 0 && lbaseMB == 0 && l0Files == 0 && l0Stop == 0 { + return + } + if targetMB > 0 { + opts.Levels = make([]pebble.LevelOptions, 7) + for i := range opts.Levels { + opts.Levels[i].TargetFileSize = int64(targetMB) << 20 << i + } + } + if lbaseMB > 0 { + opts.LBaseMaxBytes = int64(lbaseMB) << 20 + } + if l0Files > 0 { + opts.L0CompactionFileThreshold = l0Files + } + if l0Stop > 0 { + opts.L0StopWritesThreshold = l0Stop + } + log.Printf("pebble: level opts target_file_mb=%d lbase_mb=%d l0_compaction_files=%d l0_stop_writes=%d (0 = pebble default)", + targetMB, lbaseMB, l0Files, l0Stop) +} + // envInt reads an env var as int with a default. Empty / malformed → default. func envInt(name string, defaultV int) int { v := os.Getenv(name) diff --git a/internal/store/pebble_level_opts_test.go b/internal/store/pebble_level_opts_test.go new file mode 100644 index 0000000..943aca6 --- /dev/null +++ b/internal/store/pebble_level_opts_test.go @@ -0,0 +1,83 @@ +package store + +import ( + "testing" + + "github.com/cockroachdb/pebble" +) + +func TestLevelOptsUnsetMatchPebbleDefaults(t *testing.T) { + for _, k := range []string{"COSIFT_PEBBLE_TARGET_FILE_MB", "COSIFT_PEBBLE_LBASE_MB", "COSIFT_PEBBLE_L0_COMPACTION_FILES", "COSIFT_PEBBLE_L0_STOP_WRITES"} { + t.Setenv(k, "") + } + got := &pebble.Options{} + applyLevelOptsFromEnv(got) + if got.Levels != nil || got.LBaseMaxBytes != 0 || got.L0CompactionFileThreshold != 0 || got.L0StopWritesThreshold != 0 { + t.Fatalf("unset env must not touch options: %+v", got) + } + got.EnsureDefaults() + want := (&pebble.Options{}).EnsureDefaults() + for i := 0; i < 7; i++ { + if g, w := got.Level(i).TargetFileSize, want.Level(i).TargetFileSize; g != w { + t.Errorf("L%d TargetFileSize: got %d want %d", i, g, w) + } + } + if got.LBaseMaxBytes != want.LBaseMaxBytes || got.L0CompactionFileThreshold != want.L0CompactionFileThreshold || got.L0StopWritesThreshold != want.L0StopWritesThreshold { + t.Errorf("defaults drifted: got lbase=%d l0files=%d l0stop=%d want %d %d %d", + got.LBaseMaxBytes, got.L0CompactionFileThreshold, got.L0StopWritesThreshold, + want.LBaseMaxBytes, want.L0CompactionFileThreshold, want.L0StopWritesThreshold) + } +} + +func TestLevelOptsFromEnv(t *testing.T) { + t.Setenv("COSIFT_PEBBLE_TARGET_FILE_MB", "16") + t.Setenv("COSIFT_PEBBLE_LBASE_MB", "512") + t.Setenv("COSIFT_PEBBLE_L0_COMPACTION_FILES", "50") + t.Setenv("COSIFT_PEBBLE_L0_STOP_WRITES", "24") + opts := &pebble.Options{} + applyLevelOptsFromEnv(opts) + opts.EnsureDefaults() + for i := 0; i < 7; i++ { + if g, w := opts.Level(i).TargetFileSize, int64(16)<<20< "$bin/curl" <<'STUB' +#!/usr/bin/env bash +printf '{"path":"%s"}' "$SNAPTEST_CKPT" +STUB +cat > "$bin/gcloud" <<'STUB' +#!/usr/bin/env bash +echo "$*" >> "$SNAPTEST_GCLOUD_LOG" +STUB +cat > "$bin/pigz" <<'STUB' +#!/usr/bin/env bash +sleep "${SNAPTEST_PIGZ_DELAY:-0}" +exec gzip -c +STUB +chmod +x "$bin"/* +if ! command -v python3 >/dev/null 2>&1; then + cat > "$bin/python3" <<'STUB' +#!/usr/bin/env bash +sed -n 's/.*"path": *"\([^"]*\)".*/\1/p' +STUB + chmod +x "$bin/python3" +fi +export PATH="$bin:$PATH" + +fail() { printf "FAIL: %b\n" "$*" >&2; exit 1; } + +make_fixture() { + local ckpt="$work/$1" + mkdir -p "$ckpt" + head -c 4194304 /dev/zero > "$ckpt/000001.sst" + echo '{"cluster":{"peer_auth_token":"t"}}' > "$work/cosift.json" + echo "$ckpt" +} + +run_snapshot() { + local rc=0 + COSIFT_ADMIN_TOKEN=t COSIFT_GCS_BUCKET=gs://test COSIFT_KEEP=14 \ + bash "$snapshot" > "$work/out.log" 2>&1 || rc=$? + return $rc +} + +# tar rc 1: pigz stalls so tar blocks mid-file; the file is appended to meanwhile. +ckpt="$(make_fixture ckpt-changed)" +export SNAPTEST_CKPT="$ckpt" SNAPTEST_GCLOUD_LOG="$work/gcloud.log" SNAPTEST_PIGZ_DELAY=1 +: > "$SNAPTEST_GCLOUD_LOG" +export COSIFT_CONFIG="$work/cosift.json" +( sleep 0.3; echo changed >> "$ckpt/000001.sst" ) & +rc=0; run_snapshot || rc=$? +wait +(( rc == 0 )) || fail "changed-file run exited $rc (want 0):\n$(cat "$work/out.log")" +grep -q 'tar reported changed files (rc 1)' "$work/out.log" || fail "rc 1 was not observed; the mutation did not race tar:\n$(cat "$work/out.log")" +grep -q '^storage cp ' "$SNAPTEST_GCLOUD_LOG" || fail "upload did not run after rc 1" +[[ ! -d "$ckpt" ]] || fail "checkpoint dir not cleaned up" +echo "ok: tar rc 1 tolerated" + +# tar rc 2: config file missing -> "Cannot stat" -> abort before upload. +ckpt="$(make_fixture ckpt-fatal)" +export SNAPTEST_CKPT="$ckpt" SNAPTEST_PIGZ_DELAY=0 +: > "$SNAPTEST_GCLOUD_LOG" +export COSIFT_CONFIG="$work/missing.json" +rc=0; run_snapshot || rc=$? +(( rc == 2 )) || fail "missing-config run exited $rc (want 2):\n$(cat "$work/out.log")" +grep -q 'snapshot: tar failed (rc 2)' "$work/out.log" || fail "rc 2 message missing:\n$(cat "$work/out.log")" +[[ ! -s "$SNAPTEST_GCLOUD_LOG" ]] || fail "upload ran despite tar rc 2" +echo "ok: tar rc 2 aborts" diff --git a/scripts/snapshot.sh b/scripts/snapshot.sh index e9bafc5..20d9f26 100644 --- a/scripts/snapshot.sh +++ b/scripts/snapshot.sh @@ -53,9 +53,19 @@ log "checkpoint dir: $ckpt" archive="$tmp/cosift-snapshot.tar.gz" log "tarring $ckpt + $CONFIG → $archive" -tar -C "$(dirname "$ckpt")" -I "pigz -p ${COSIFT_PIGZ_THREADS:-8}" -cf "$archive" \ +# tar rc 1 (file changed as we read it) is harmless: checkpoint SSTs are hard-linked and immutable +set +e +tar --warning=no-file-changed -C "$(dirname "$ckpt")" -I "pigz -p ${COSIFT_PIGZ_THREADS:-8}" -cf "$archive" \ "$(basename "$ckpt")" \ -C "$(dirname "$CONFIG")" "$(basename "$CONFIG")" +tar_rc=$? +set -e +if (( tar_rc > 1 )); then + echo "snapshot: tar failed (rc $tar_rc)" >&2 + exit "$tar_rc" +elif (( tar_rc == 1 )); then + log "tar reported changed files (rc 1); continuing" +fi size=$(stat -c%s "$archive") log "$((size / 1024 / 1024)) MiB"