ops fixes for v0.2.5: Pebble level sizing knobs, lift the write deadline through the middleware, tolerate tar rc 1 in snapshot.sh, compact phase liveness, Go runtime memory metrics

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,
+	}
+}

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)
+	}
+}

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)

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}
 }

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

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
 

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)

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)
+	}
+}

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` |
 
 ---