fix: harden gossip, HLC restarts, watcher races, MCP args, pagination

- gossip: validate push batches (4 MiB / 1000-row caps); reject rows claiming the local node id (vector-poisoning), empty ids, negative HCLs - gossip: reconcile derived state after pulls (pulls only append to the observation log, so entry-count comparison could never trigger it) - hlc: seek clock from persisted MAX(hcl) at Open so a restart with a regressed wall clock cannot reissue values (locator/cursor safety) - db: serialize writers via BEGIN IMMEDIATE DSN, single conn per pool, and a per-KnoxDB mutex around RecordObservation's dedup - watch: atomic ticker guards (was a cross-goroutine data race), trailing-edge per-path debounce, recursive directory watches, rename re-ingest, remove cancels pending ingests - mcp: strict argument validation (no silent clamping), thread existence checks before writes, nil-safe golden-thread tool - cli: --page 0 no longer panics; query/recent pagination actually pages - tests: hlc SeekTo monotonicity, concurrent dedup race, push validation, batch caps, idempotency on observation counts
2026-09-17 01:52:06 -07:00
parent 876d2aa45f
commit 6845975b7b
10 changed files with 733 additions and 116 deletions
@@ -22,6 +22,12 @@ type KnoxDB struct {
 	db     *sql.DB
 	nodeID string
 	clock  *hlc.Clock
+
+	// writeMu serializes the check-then-insert dedup in RecordObservation within
+	// this process. Cross-process serialization comes from _txlock=immediate (the
+	// write lock is taken at BEGIN, before the dedup read) plus a single
+	// connection per pool.
+	writeMu sync.Mutex
 }

 type Observation struct {
@@ -93,10 +99,13 @@ func Open(path string) (*KnoxDB, error) {
 		return nil, fmt.Errorf("create db dir: %w", err)
 	}

-	db, err := sql.Open("sqlite", path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)")
+	db, err := sql.Open("sqlite", path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_txlock=immediate")
 	if err != nil {
 		return nil, fmt.Errorf("open db: %w", err)
 	}
+	// One connection per process: WAL has a single writer; serializing on one
+	// connection avoids pool contention surfacing as busy_timeout errors.
+	db.SetMaxOpenConns(1)

 	if _, err := db.Exec(Schema); err != nil {
 		return nil, fmt.Errorf("init schema: %w", err)
@@ -135,6 +144,13 @@ func Open(path string) (*KnoxDB, error) {
 	if _, err := db.Exec("UPDATE observations SET hcl=id WHERE hcl IS NULL"); err != nil {
 		return nil, fmt.Errorf("backfill hcl: %w", err)
 	}
+	// Resume this node's HLC from its persisted max: a restart with a regressed
+	// wall clock must not reissue already-persisted values (see hlc.SeekTo).
+	var maxHCL int64
+	if err := db.QueryRow(`SELECT COALESCE(MAX(hcl), 0) FROM observations WHERE node_id=?`, nodeID).Scan(&maxHCL); err != nil {
+		return nil, fmt.Errorf("seed hlc: %w", err)
+	}
+	kdb.clock.SeekTo(maxHCL)

 	// Locator uniqueness: (node_id, hcl) is the merge key for gossip; a given
 	// node's HCL is strictly monotonic so this never throws a false conflict.
@@ -204,6 +220,9 @@ func (k *KnoxDB) ObservationEntryEstimate() int {
 // Idempotent: if the latest observation for this fingerprint has an identical
 // content signature, nothing is recorded — re-ingesting unchanged content is a no-op.
 func (k *KnoxDB) RecordObservation(o ObservationRecord) (obsID int64, isNew bool, err error) {
+	k.writeMu.Lock()
+	defer k.writeMu.Unlock()
+
 	tx, err := k.db.Begin()
 	if err != nil {
 		return 0, false, fmt.Errorf("begin tx: %w", err)
@@ -0,0 +1,156 @@
+package db
+
+import (
+	"path/filepath"
+	"sync"
+	"testing"
+)
+
+func tmpDB(t *testing.T) *KnoxDB {
+	t.Helper()
+	kdb, err := Open(filepath.Join(t.TempDir(), "index.db"))
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	t.Cleanup(func() { kdb.Close() })
+	return kdb
+}
+
+func record(t *testing.T, k *KnoxDB, fp, title string) {
+	t.Helper()
+	_, _, err := k.RecordObservation(ObservationRecord{
+		Fingerprint:     fp,
+		SourceID:        "test",
+		SourcePath:      fp,
+		Project:         "itest",
+		ContentType:     "test",
+		Title:           title,
+		Summary:         "summary",
+		CreatedAt:       "2026-08-29T00:00:00Z",
+		LineEnd:         0,
+		Confidence:      0.9,
+		IngesterVersion: "test/v1",
+	})
+	if err != nil {
+		t.Errorf("record %s: %v", fp, err)
+	}
+}
+
+func obsCount(t *testing.T, k *KnoxDB) int {
+	t.Helper()
+	stats, err := k.Stats()
+	if err != nil {
+		t.Fatalf("stats: %v", err)
+	}
+	n, _ := stats["total_observations"].(int)
+	return n
+}
+
+// TestRecordObservationConcurrentDedup: N goroutines ingesting identical
+// content must produce exactly one observation row. This exercises the
+// check-then-insert dedup under the writeMu + BEGIN IMMEDIATE serialization.
+func TestRecordObservationConcurrentDedup(t *testing.T) {
+	k := tmpDB(t)
+	var wg sync.WaitGroup
+	for i := 0; i < 20; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			record(t, k, "fp:concurrent", "same title")
+		}()
+	}
+	wg.Wait()
+	if n := obsCount(t, k); n != 1 {
+		t.Fatalf("expected exactly 1 observation after concurrent identical ingests, got %d", n)
+	}
+}
+
+// TestRecordObservationDistinctFingerprints: different content must never be
+// deduped away (the constraint is per-fingerprint signature, not global).
+func TestRecordObservationDistinctFingerprints(t *testing.T) {
+	k := tmpDB(t)
+	record(t, k, "fp:a", "alpha")
+	record(t, k, "fp:b", "beta")
+	record(t, k, "fp:a", "alpha changed")
+	if n := obsCount(t, k); n != 3 {
+		t.Fatalf("expected 3 observations, got %d", n)
+	}
+}
+
+// TestPushObservationsValidation: forged/malformed rows are rejected without
+// error — self node_id (poisoning vector), empty node_id, negative HCL.
+func TestPushObservationsValidation(t *testing.T) {
+	k := tmpDB(t)
+	foreign := GossipObservation{
+		NodeID: "0123456789abcdef0123456789abcdef", HCL: 42,
+		Fingerprint: "fp:foreign", SourceID: "test", Title: "t", Summary: "s",
+		CollectedAt: "2026-08-29T00:00:00Z",
+	}
+	rows := []GossipObservation{
+		foreign,
+		{NodeID: k.NodeID(), HCL: 100, Fingerprint: "fp:self"}, // spoof poisoning attempt
+		{NodeID: "", HCL: 1, Fingerprint: "fp:empty"},
+		{NodeID: "other", HCL: -5, Fingerprint: "fp:neg"},
+	}
+	n, err := k.PushObservations(rows)
+	if err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	if n != 1 {
+		t.Fatalf("expected exactly the one valid row inserted, got %d", n)
+	}
+	if got := obsCount(t, k); got != 1 {
+		t.Fatalf("expected 1 observation in the log, got %d", got)
+	}
+}
+
+// TestOpenReopenHCLMonotonicAcrossRestart: reopening a DB must resume the HLC
+// from its persisted max (clock seeding), keep the node identity, and order the
+// new observation above every previous one.
+func TestOpenReopenHCLMonotonicAcrossRestart(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "index.db")
+
+	k1, err := Open(path)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	record(t, k1, "fp:r1", "one")
+	record(t, k1, "fp:r2", "two")
+	record(t, k1, "fp:r3", "three")
+	nodeID1 := k1.NodeID()
+
+	maxBefore := int64(0)
+	rows, err := k1.ObservationsAfter(nodeID1, 0, 100)
+	if err != nil {
+		t.Fatalf("obs after: %v", err)
+	}
+	for _, r := range rows {
+		if r.HCL > maxBefore {
+			maxBefore = r.HCL
+		}
+	}
+	if err := k1.Close(); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+
+	k2, err := Open(path)
+	if err != nil {
+		t.Fatalf("reopen: %v", err)
+	}
+	defer k2.Close()
+	if k2.NodeID() != nodeID1 {
+		t.Errorf("node id changed across reopen: %q -> %q", nodeID1, k2.NodeID())
+	}
+	record(t, k2, "fp:r4", "four")
+
+	after, err := k2.ObservationsAfter(nodeID1, maxBefore, 100)
+	if err != nil {
+		t.Fatalf("obs after (reopened): %v", err)
+	}
+	if len(after) != 1 {
+		t.Fatalf("expected exactly the new observation above the pre-restart max, got %d rows", len(after))
+	}
+	if after[0].Fingerprint != "fp:r4" {
+		t.Errorf("unexpected row above max: %s", after[0].Fingerprint)
+	}
+}
@@ -3,6 +3,7 @@ package db
 import (
 	"database/sql"
 	"fmt"
+	"log"
 )

 // GossipObservation is the serializable wire form of an observation exchanged
@@ -40,6 +41,19 @@ func (k *KnoxDB) PushObservations(rows []GossipObservation) (int, error) {

 	inserted := 0
 	for _, o := range rows {
+		// Reject malformed or forged rows. Nodes only ever push their own
+		// observations, so a row claiming this node's id cannot be legitimate:
+		// accepting it would let a peer poison our knowledge vector (a forged
+		// max-HCL makes peers believe they have our whole history and stop
+		// pulling). Empty node ids and negative HCLs are likewise never produced
+		// by a real node.
+		if o.NodeID == "" || o.HCL < 0 {
+			continue
+		}
+		if o.NodeID == k.nodeID {
+			log.Printf("[gossip] dropped pushed row claiming local node_id (spoof?)")
+			continue
+		}
 		res, err := tx.Exec(
 			`INSERT OR IGNORE INTO observations
 			 (fingerprint, source_id, source_path, project, content_type, title, summary,