bb852faa27
Implements the top findings from the codebase review, verified with tests and live CLI/MCP checks. **Gossip integrity** - Push validation: 4 MiB body cap, 1000-row batch cap; rows claiming the local node id (vector-poisoning), empty node ids, and negative HCLs rejected (internal/watch/gossip.go, internal/db/gossip.go) - Reconcile-on-pull: Run returns the pulled count, syncGossip rebuilds derived state when > 0 — entry-count comparison could never fire, so synced observations never materialized into searchable entries **Data-layer safety** - HLC resumed from MAX(hcl) at Open (hlc.SeekTo): a restart with a regressed wall clock cannot reissue values the (node_id, hcl) locator and pull cursors depend on - Writer serialization: _txlock=immediate DSN + SetMaxOpenConns(1) + per-KnoxDB mutex around RecordObservation's check-then-insert dedup (closes duplicate-row race) **Watch daemon** - Ticker guard flags now atomic.Bool (was a cross-goroutine data race) - Trailing-edge per-path debounce (timer-based, pruned on fire/delete) - Recursive watches (startup tree walk + watcher.Add on dir Create); Rename re-ingests, Remove cancels pending ingests **MCP + CLI** - Strict arg validation, no silent clamping: thread_id 0 errors instead of renaming thread #1; empty knox_thread_link {} errors instead of false success; thread existence checked before writes; golden-thread tool nil-safe - --page 0 errors instead of panicking; query/recent pagination actually pages (page x limit) **Tests** (new internal/hlc and internal/db packages): SeekTo monotonicity, concurrent dedup race, push validation, reopen HCL monotonicity, batch caps, self-spoof rejection, idempotency on observation counts. Verified: go build, go vet, full suite with -race, live MCP stdio transcripts against a scratch DB. Reviewed-on: #3 Co-authored-by: David Gwilliam <dhgwilliam@gmail.com> Co-committed-by: David Gwilliam <dhgwilliam@gmail.com>
157 lines
4.2 KiB
Go
157 lines
4.2 KiB
Go
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)
|
|
}
|
|
}
|