brain/index: refuse --rebuild while a live brain holds the db

bin/brain/index.go

@@ -33,10 +33,21 @@ func main() {
 type indexFlags struct {
 	db, factsJSON, withChats, since                                              string
 	corpus                                                                       []string
-	rebuild, noDefaults, withMail, withFacts, dryRun, skipIndexes, jsonOut, skip bool
+	rebuild, noDefaults, withMail, withFacts, dryRun, skipIndexes, jsonOut, skip, force bool
 	limit, workers, batch, progress                                              int
 }
 
+// gitRepoRoot resolves the actual repository checkout (independent of
+// KB_ROOT, which points the corpus/db at arbitrary roots for tests and
+// throwaway builds).
+func gitRepoRoot() string {
+	out, err := exec.Command("git", "rev-parse", "--show-toplevel").Output()
+	if err != nil {
+		return ""
+	}
+	return strings.TrimSpace(string(out))
+}
+
 func run(args []string) int {
 	// historic wrapper prepended --with-mail; keep default off unless passed
 	v := indexFlags{}
@@ -57,6 +68,7 @@ func run(args []string) int {
 	p.Int(&v.batch, "", "batch", "leafs per transaction (default 64)")
 	p.Int(&v.progress, "", "progress", "progress/ETA line every N seconds")
 	p.Bool(&v.skip, "", "skip", "skip leafs already in the db (resume)")
+	p.Bool(&v.force, "", "force", "rebuild even if the db is open by a live process")
 	p.Bool(&v.jsonOut, "", "json", "JSON stats")
 	// --corpus may repeat: parse manually from args leftovers after flaggy
 	if err := cliparse.Parse(p, filterCorpusArgs(args, &v.corpus)); err != nil {
@@ -72,6 +84,10 @@ func run(args []string) int {
 	if dbpath == "" {
 		dbpath = filepath.Join(root, "var", "kb.lbug")
 	}
+	port := os.Getenv("KB_PORT")
+	if port == "" {
+		port = "8630"
+	}
 
 	var leafs []brain.CorpusLeaf
 	var err error
@@ -143,6 +159,30 @@ func run(args []string) int {
 	}
 
 	if !v.skip {
+		// A live holder of this file (e.g. brain-serve in a compose container
+		// bind-mounting the same var/) would keep serving the removed inode:
+		// reads go stale and the fresh db lands with the writer's uid, locking
+		// out the service. Refuse unless --force. Two probes: same-namespace
+		// fd holders via /proc, and (for the default repo db) any brain API
+		// answering on 127.0.0.1:$KB_PORT — container fds are invisible to
+		// host /proc when the service runs as another uid.
+		var reasons []string
+		if holders, err := brain.LiveHolders(dbpath); err != nil {
+			fmt.Fprintf(os.Stderr, "brain/index: live-holder check: %v\n", err)
+			return 1
+		} else if len(holders) > 0 {
+			reasons = append(reasons, fmt.Sprintf("%s is open by %d process(es):\n  %s", dbpath, len(holders), strings.Join(holders, "\n  ")))
+		}
+		if gitRoot := gitRepoRoot(); gitRoot != "" &&
+			dbpath == filepath.Join(gitRoot, "var", "kb.lbug") &&
+			os.Getenv("KB_INDEX_ALLOW_LIVE") != "1" &&
+			brain.BrainAPIAlive("127.0.0.1:" + port) {
+			reasons = append(reasons, fmt.Sprintf("a brain API is answering on 127.0.0.1:%s (compose brain bind-mounts this db)", port))
+		}
+		if len(reasons) > 0 && !v.force {
+			fmt.Fprintf(os.Stderr, "brain/index: refuse --rebuild; stop/restart the brain first (or pass --force):\n%s\n", strings.Join(reasons, "\n"))
+			return 2
+		}
 		_ = os.Remove(dbpath)
 		_ = os.Remove(dbpath + ".wal")
 	}

compose.yaml

@@ -81,7 +81,10 @@ services:
       context: .
       dockerfile: Dockerfile
       target: api
-    environment: *env
+    environment:
+      <<: *env
+      # sanctioned swap+restart flow: the live-holder guard must not block it
+      KB_INDEX_ALLOW_LIVE: "1"
     volumes:
       - ./var:/data/var
       - ./var/lbdb-extension:/data/.lbdb/extension:ro

docs/brain-rebuild.md

@@ -33,6 +33,14 @@ already-written corpus and goes straight to index build.
 >     KB_BUFFER_POOL=10737418240 bin-build/brain-index --skip \
 >         --with-mail --with-facts --workers 12 --batch 256 --progress 5
 
+> Warning: `--rebuild` refuses to run while a brain holds the db open (fd
+> holders via /proc, or an API answering on `127.0.0.1:$KB_PORT` when
+> rebuilding the repo-default `var/kb.lbug`) — deleting the file under a live
+> serve leaves it serving the removed inode and can lock the service out of
+> the fresh db. Stop/restart the brain first, pass `--force` to override, or
+> set `KB_INDEX_ALLOW_LIVE=1` in environments where the swap+restart flow is
+> intended (the compose `index` service already sets this).
+
 Observed rates: embedding is fast (~16k/s, 256-dim); the db **write** phase is
 the bottleneck (~100/s, ~40 min for 242k leafs).
 

internal/brain/livecheck.go

@@ -0,0 +1,120 @@
+// Live-process detection for the db file. Cgo-free so it unit-tests without
+// the Ladybug library.
+package brain
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"syscall"
+	"time"
+)
+
+// LiveHolders returns one "pid N (cmdline)" string per process currently
+// holding path open. Detection is by device+inode of each /proc/<pid>/fd
+// target, so holders are found across bind mounts and container PID
+// namespaces: a compose brain service holding /data/var/kb.lbug is flagged
+// when a host-side run checks the same file at ~/2dph/var/kb.lbug. Holders of
+// a deleted inode whose path (in their own namespace) equals path are also
+// reported — that state means a server is up but serving stale data.
+//
+// Best effort: unreadable /proc entries are skipped, and on non-Linux systems
+// the result is always empty.
+func LiveHolders(path string) ([]string, error) {
+	if runtime.GOOS != "linux" {
+		return nil, nil
+	}
+	abs, err := filepath.Abs(path)
+	if err != nil {
+		return nil, err
+	}
+	var wantDev, wantIno uint64
+	if fi, err := os.Stat(abs); err == nil {
+		if st, ok := fi.Sys().(*syscall.Stat_t); ok {
+			wantDev, wantIno = st.Dev, st.Ino
+		}
+	} else if !os.IsNotExist(err) {
+		return nil, err
+	}
+
+	entries, err := os.ReadDir("/proc")
+	if err != nil {
+		return nil, err
+	}
+	var out []string
+	for _, e := range entries {
+		pid := e.Name()
+		if !isNumeric(pid) {
+			continue
+		}
+		cmd := procCmdline(pid)
+		fds, err := os.ReadDir("/proc/" + pid + "/fd")
+		if err != nil {
+			continue // kernel threads, other users, races
+		}
+		held := false
+		for _, fd := range fds {
+			link, err := os.Readlink("/proc/" + pid + "/fd/" + fd.Name())
+			if err != nil {
+				continue
+			}
+			clean := strings.TrimSuffix(link, " (deleted)")
+			if clean == abs && strings.HasSuffix(link, " (deleted)") {
+				held = true // stale deleted-inode holder at our path
+				break
+			}
+			if wantIno == 0 {
+				continue
+			}
+			if fi, err := os.Stat("/proc/" + pid + "/fd/" + fd.Name()); err == nil {
+				if st, ok := fi.Sys().(*syscall.Stat_t); ok && st.Dev == wantDev && st.Ino == wantIno {
+					held = true
+					break
+				}
+			}
+		}
+		if held {
+			out = append(out, fmt.Sprintf("pid %s (%s)", pid, cmd))
+		}
+	}
+	return out, nil
+}
+
+// BrainAPIAlive reports whether a brain HTTP API answers /stats on addr
+// (e.g. "127.0.0.1:8630"). The compose brain runs in its own container as
+// another uid, so its fds are invisible to LiveHolders from the host; a
+// healthy /stats response is the reliable cross-namespace signal.
+func BrainAPIAlive(addr string) bool {
+	client := http.Client{Timeout: 2 * time.Second}
+	resp, err := client.Get("http://" + addr + "/stats")
+	if err != nil {
+		return false
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return false
+	}
+	b, err := io.ReadAll(io.LimitReader(resp.Body, 4096))
+	return err == nil && strings.Contains(string(b), `"total"`)
+}
+
+func procCmdline(pid string) string {
+	b, err := os.ReadFile("/proc/" + pid + "/cmdline")
+	if err != nil || len(b) == 0 {
+		return "?"
+	}
+	return strings.TrimSpace(strings.ReplaceAll(string(b), "\x00", " "))
+}
+
+func isNumeric(s string) bool {
+	for i := 0; i < len(s); i++ {
+		if s[i] < '0' || s[i] > '9' {
+			return false
+		}
+	}
+	return s != ""
+}

internal/brain/livecheck_test.go

@@ -0,0 +1,98 @@
+package brain
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+)
+
+// findSelf reports whether LiveHolders flags this test process holding path.
+func findSelf(t *testing.T, path string) []string {
+	t.Helper()
+	out, err := LiveHolders(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	self := "pid " + strconv.Itoa(os.Getpid()) + " ("
+	var mine []string
+	for _, h := range out {
+		if strings.HasPrefix(h, self) {
+			mine = append(mine, h)
+		}
+	}
+	return mine
+}
+
+func TestLiveHoldersDetectsOpenFile(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "kb.lbug")
+	f, err := os.Create(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer f.Close()
+
+	mine := findSelf(t, path)
+	if len(mine) != 1 {
+		t.Fatalf("want exactly 1 self holder, got %v", mine)
+	}
+
+	// Closed fd must clear the detection.
+	if err := f.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if got := findSelf(t, path); len(got) != 0 {
+		t.Fatalf("holder still flagged after close: %v", got)
+	}
+}
+
+func TestLiveHoldersDetectsDeletedInodeHolder(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "kb.lbug")
+	f, err := os.Create(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer f.Close()
+	if err := os.Remove(path); err != nil {
+		t.Fatal(err)
+	}
+
+	if mine := findSelf(t, path); len(mine) != 1 {
+		t.Fatalf("deleted-inode holder not flagged, got %v", mine)
+	}
+}
+
+func TestLiveHoldersMissingFileClean(t *testing.T) {
+	out, err := LiveHolders(filepath.Join(t.TempDir(), "nope.lbug"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(out) != 0 {
+		t.Fatalf("want no holders for missing file, got %v", out)
+	}
+}
+
+func TestBrainAPIAlive(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/stats" {
+			http.NotFound(w, r)
+			return
+		}
+		fmt.Fprint(w, `{"by_root":{"facts":3},"db":"/data/var/kb.lbug","total":3}`)
+	}))
+	defer srv.Close()
+
+	addr := strings.TrimPrefix(srv.URL, "http://")
+	if !BrainAPIAlive(addr) {
+		t.Fatal("want alive for serving /stats with total")
+	}
+	if BrainAPIAlive("127.0.0.1:1") { // nothing listens on port 1
+		t.Fatal("want dead when nothing listens")
+	}
+}