persist: stream HNSW node blobs in bounded windows; compact/persist progress logs

internal/index/hnsw_compact.go

@@ -1,5 +1,12 @@
 package index
 
+import "log"
+
+// compactProgressEvery paces the in-compact progress logs; the whole pass
+// runs under the write lock, so these lines are the only liveness signal.
+// A var (not const) so tests can lower it below the fixture size.
+var compactProgressEvery = 10_000_000
+
 // Rebuild constructs a fresh HNSW with the same parameters as h and inserts
 // every valid (vec != nil) node from h via AddPassage. Unlike Compact, which
 // merely removes zombies and rewires existing edges, Rebuild reconstructs
@@ -69,6 +76,9 @@ func (h *HNSW) Compact() (removed int) {
 		newCodes = make([][]uint16, 0, len(h.codes))
 	}
 	for i := range h.nodes {
+		if i > 0 && i%compactProgressEvery == 0 {
+			log.Printf("hnsw compact: scanning %d/%d nodes", i, len(h.nodes))
+		}
 		if len(h.nodes[i].vec) == 0 {
 			remap[i] = -1
 			continue
@@ -88,6 +98,9 @@ func (h *HNSW) Compact() (removed int) {
 	// 2. Remap neighbor lists. Iterating in increasing order so writes only
 	//    touch slots we've already read from the source array.
 	for i := range newNodes {
+		if i > 0 && i%compactProgressEvery == 0 {
+			log.Printf("hnsw compact: rewiring neighbors %d/%d nodes", i, len(newNodes))
+		}
 		for lvl := range newNodes[i].neighbors {
 			old := newNodes[i].neighbors[lvl]
 			out := make([]int, 0, len(old))

internal/index/hnsw_compact_test.go

@@ -1,9 +1,13 @@
 package index
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"log"
 	"math/rand"
+	"os"
+	"strings"
 	"testing"
 )
 
@@ -118,3 +122,32 @@ func TestHNSWZombieCompaction(t *testing.T) {
 		t.Errorf("rebuild did not restore recall: clean=%.3f rebuilt=%.3f", clean, rebuiltRecall)
 	}
 }
+
+// TestHNSWCompactProgressLogs pins the write-lock liveness lines.
+func TestHNSWCompactProgressLogs(t *testing.T) {
+	h := buildTestHNSW(250, 16, 3, 5)
+	for i := 0; i < 40; i++ {
+		h.nodes[i*5].vec = nil
+	}
+
+	savedEvery := compactProgressEvery
+	compactProgressEvery = 100
+	var buf bytes.Buffer
+	log.SetOutput(&buf)
+	defer func() {
+		compactProgressEvery = savedEvery
+		log.SetOutput(os.Stderr)
+	}()
+
+	removed := h.Compact()
+	if removed != 40 {
+		t.Errorf("removed: want 40, got %d", removed)
+	}
+	out := buf.String()
+	if !strings.Contains(out, "hnsw compact: scanning") {
+		t.Errorf("missing scanning progress line in:\n%s", out)
+	}
+	if !strings.Contains(out, "hnsw compact: rewiring neighbors") {
+		t.Errorf("missing rewiring progress line in:\n%s", out)
+	}
+}

internal/index/hnsw_persist.go

@@ -41,12 +41,21 @@ import (
 	"fmt"
 	"log"
 	"math"
+	"time"
 
 	"github.com/pilot-protocol/cosift/internal/store"
 )
 
 const hnswMetaMagic = "HSW1"
 
+// persistWindowBytes bounds encoded blobs held in memory at once: a full
+// persist that materializes every blob first costs ~vec-bytes of extra heap
+// (~240 GB at 80M nodes) and OOMs before writing anything. Var for tests.
+var persistWindowBytes = 1 << 30
+
+// persistFlushed is a test hook observing each flushed window (nil in prod).
+var persistFlushed func(nodes, bytes int)
+
 // Persist serializes every node + meta into the PebbleStore. Safe to call
 // during ongoing search (acquires RLock); does NOT acquire the write lock,
 // so concurrent Add() during Persist will partially leak into the saved
@@ -55,7 +64,7 @@ func (h *HNSW) Persist(ctx context.Context, ps *store.PebbleStore) error {
 	return h.PersistFrom(ctx, ps, 0)
 }
 
-// PersistFrom writes meta + nodes[fromIdx:] in a single Pebble batch. The
+// PersistFrom writes nodes[fromIdx:] in bounded windows, then meta. The
 // crawl-time checkpoint goroutine uses this with fromIdx = last-persisted
 // count, so each checkpoint touches only the newly-added nodes. Meta is
 // always re-written so a reader can size the slice correctly.
@@ -81,15 +90,48 @@ func (h *HNSW) PersistFrom(ctx context.Context, ps *store.PebbleStore, fromIdx i
 	// New order: meta ALWAYS lags or equals nodes-on-disk. Worst case after
 	// partial write: meta says N nodes, disk has N+M; the M extras are
 	// orphan but harmless (LoadHNSW caps at meta.nodeCount).
-	entries := make([]store.VectorNodeEntry, 0, len(h.nodes)-fromIdx)
+	total := len(h.nodes) - fromIdx
+	start := time.Now()
+	window := make([]store.VectorNodeEntry, 0, 4096)
+	windowBytes, written, flushes := 0, 0, 0
+	var bytesWritten int64
+	flush := func() error {
+		if len(window) == 0 {
+			return nil
+		}
+		if err := ps.PutVectorNodesBatch(ctx, window); err != nil {
+			return fmt.Errorf("put vector nodes batch: %w", err)
+		}
+		written += len(window)
+		bytesWritten += int64(windowBytes)
+		flushes++
+		if persistFlushed != nil {
+			persistFlushed(len(window), windowBytes)
+		}
+		if flushes > 1 || written < total {
+			elapsed := max(time.Since(start).Seconds(), 0.001)
+			rate := float64(written) / elapsed
+			eta := time.Duration(float64(total-written) / rate * float64(time.Second)).Round(time.Second)
+			log.Printf("hnsw persist: %d/%d nodes (%.1f GiB, %.0f nodes/s, eta %s)",
+				written, total, float64(bytesWritten)/(1<<30), rate, eta)
+		}
+		clear(window)
+		window = window[:0]
+		windowBytes = 0
+		return nil
+	}
 	for i := fromIdx; i < len(h.nodes); i++ {
-		entries = append(entries, store.VectorNodeEntry{
-			ID:   uint64(i),
-			Blob: encodeHNSWNode(&h.nodes[i]),
-		})
+		blob := encodeHNSWNode(&h.nodes[i])
+		window = append(window, store.VectorNodeEntry{ID: uint64(i), Blob: blob})
+		windowBytes += len(blob) + 16
+		if windowBytes >= persistWindowBytes {
+			if err := flush(); err != nil {
+				return err
+			}
+		}
 	}
-	if err := ps.PutVectorNodesBatch(ctx, entries); err != nil {
-		return fmt.Errorf("put vector nodes batch: %w", err)
+	if err := flush(); err != nil {
+		return err
 	}
 	meta := encodeHNSWMeta(h.dim, h.maxLevel, h.entryPoint, len(h.nodes))
 	if err := ps.PutVectorMeta(ctx, meta); err != nil {