From 6845975b7bef74ebef5a82fa9558f072eb4cf59c Mon Sep 17 00:00:00 2001 From: David Gwilliam Date: Thu, 17 Sep 2026 01:52:06 -0700 Subject: [PATCH] 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 --- internal/cmd/mcp.go | 220 +++++++++++++++++++++++++++------- internal/cmd/status.go | 25 +++- internal/db/db.go | 21 +++- internal/db/db_test.go | 156 ++++++++++++++++++++++++ internal/db/gossip.go | 14 +++ internal/hlc/hlc.go | 17 +++ internal/hlc/hlc_test.go | 68 +++++++++++ internal/watch/gossip.go | 17 ++- internal/watch/gossip_test.go | 112 ++++++++++++++++- internal/watch/watch.go | 199 ++++++++++++++++++++---------- 10 files changed, 733 insertions(+), 116 deletions(-) create mode 100644 internal/db/db_test.go create mode 100644 internal/hlc/hlc_test.go diff --git a/internal/cmd/mcp.go b/internal/cmd/mcp.go index db652b2..c07f387 100644 --- a/internal/cmd/mcp.go +++ b/internal/cmd/mcp.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "math" "strings" "github.com/david/knox/internal/db" @@ -50,8 +51,14 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(searchTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - query, _ := req.Params.Arguments["query"].(string) - limit := clampInt(req, "limit", 10, 1, 50) + query, err := requiredStringArg(req, "query") + if err != nil { + return errorResult(err.Error()), nil + } + limit, err := optionalIntArg(req, "limit", 10, 1, 50) + if err != nil { + return errorResult(err.Error()), nil + } detail := parseDetail(getString(req, "detail", "normal")) scope := getString(req, "scope", "all") @@ -76,7 +83,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(recentTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - limit := clampInt(req, "limit", 10, 1, 50) + limit, err := optionalIntArg(req, "limit", 10, 1, 50) + if err != nil { + return errorResult(err.Error()), nil + } detail := parseDetail(getString(req, "detail", "normal")) scope := getString(req, "scope", "all") @@ -99,7 +109,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(getTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - fp := getString(req, "fingerprint", "") + fp, err := requiredStringArg(req, "fingerprint") + if err != nil { + return errorResult(err.Error()), nil + } entry, err := kdb.FindEntry(fp) if err != nil { @@ -165,7 +178,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(countTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - query, _ := req.Params.Arguments["query"].(string) + query, err := requiredStringArg(req, "query") + if err != nil { + return errorResult(err.Error()), nil + } results, err := kdb.Search(query, 500) if err != nil { return errorResult(err.Error()), nil @@ -235,7 +251,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(threadCreateTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - title := getString(req, "title", "") + title, err := requiredStringArg(req, "title") + if err != nil { + return errorResult(err.Error()), nil + } motivation := getString(req, "motivation", "") priority := getString(req, "priority", "medium") provenance := getString(req, "provenance", "{}") @@ -257,7 +276,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(threadUpdateTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - id := int64(clampInt(req, "thread_id", 0, 1, 999999)) + id, err := requiredIntArg(req, "thread_id", 1, 999999) + if err != nil { + return errorResult(err.Error()), nil + } title := getString(req, "title", "") motivation := getString(req, "motivation", "") priority := getString(req, "priority", "") @@ -281,7 +303,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(threadDraftTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - limit := clampInt(req, "limit", 20, 1, 200) + limit, err := optionalIntArg(req, "limit", 20, 1, 200) + if err != nil { + return errorResult(err.Error()), nil + } threads, err := kdb.ListThreads("") if err != nil { return errorResult(err.Error()), nil @@ -329,9 +354,18 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(threadLinkTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - threadID := int64(clampInt(req, "thread_id", 0, 1, 999999)) - obsID := int64(clampInt(req, "observation_id", 0, 1, 999999)) + threadID, err := requiredIntArg(req, "thread_id", 1, 999999) + if err != nil { + return errorResult(err.Error()), nil + } + obsID, err := requiredIntArg(req, "observation_id", 1, 999999) + if err != nil { + return errorResult(err.Error()), nil + } relevance := getString(req, "relevance", "") + if err := threadExists(kdb, threadID); err != nil { + return errorResult(err.Error()), nil + } if err := kdb.LinkObservationToThread(threadID, obsID, relevance); err != nil { return errorResult(err.Error()), nil } @@ -345,7 +379,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(threadSearchTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - query := getString(req, "query", "") + query, err := requiredStringArg(req, "query") + if err != nil { + return errorResult(err.Error()), nil + } threads, err := kdb.SearchThreadsByMotivation(query) if err != nil { return errorResult(err.Error()), nil @@ -372,9 +409,18 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(threadLinkEntryTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - threadID := int64(clampInt(req, "thread_id", 0, 1, 999999)) - fp := getString(req, "fingerprint", "") + threadID, err := requiredIntArg(req, "thread_id", 1, 999999) + if err != nil { + return errorResult(err.Error()), nil + } + fp, err := requiredStringArg(req, "fingerprint") + if err != nil { + return errorResult(err.Error()), nil + } relation := getString(req, "relation", "produced") + if err := threadExists(kdb, threadID); err != nil { + return errorResult(err.Error()), nil + } if err := kdb.LinkEntryToThread(threadID, fp, relation); err != nil { return errorResult(err.Error()), nil } @@ -390,9 +436,21 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(threadRelateTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - parentID := int64(clampInt(req, "parent_id", 0, 1, 999999)) - childID := int64(clampInt(req, "child_id", 0, 1, 999999)) + parentID, err := requiredIntArg(req, "parent_id", 1, 999999) + if err != nil { + return errorResult(err.Error()), nil + } + childID, err := requiredIntArg(req, "child_id", 1, 999999) + if err != nil { + return errorResult(err.Error()), nil + } relation := getString(req, "relation", "spawned") + if err := threadExists(kdb, parentID); err != nil { + return errorResult(err.Error()), nil + } + if err := threadExists(kdb, childID); err != nil { + return errorResult(err.Error()), nil + } if err := kdb.LinkEntryToThread(childID, db.ThreadFP(parentID), relation); err != nil { return errorResult(err.Error()), nil } @@ -406,29 +464,67 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(goldenTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - threadID := int64(clampInt(req, "thread_id", 0, 0, 999999)) - - if threadID == 0 { - // Query or clear - current, _ := kdb.GoldenThreadID() + // Absent thread_id: query the current golden thread. + if _, ok := req.Params.Arguments["thread_id"]; !ok { + current, err := kdb.GoldenThreadID() + if err != nil { + return errorResult(err.Error()), nil + } if current == 0 { return mcp.NewToolResultText("No golden thread set."), nil } - // If thread_id was explicitly 0 and a golden exists, clear it - if _, ok := req.Params.Arguments["thread_id"]; ok { - kdb.SetGoldenThread(0) - t, _ := kdb.GetThread(current) - return mcp.NewToolResultText(fmt.Sprintf("Golden thread cleared (was #%d: %s)", current, t.Title)), nil + t, err := kdb.GetThread(current) + if err != nil { + return errorResult(err.Error()), nil + } + if t == nil { + return errorResult(fmt.Sprintf("Golden thread #%d no longer exists", current)), nil } - t, _ := kdb.GetThread(current) return mcp.NewToolResultText(fmt.Sprintf("Golden thread: #%d %s [%s]\n %s", t.ID, t.Title, t.Status, t.Motivation)), nil } + threadID, err := requiredIntArg(req, "thread_id", 0, 999999) + if err != nil { + return errorResult(err.Error()), nil + } + if threadID == 0 { + // Explicit 0 clears the golden thread. + current, err := kdb.GoldenThreadID() + if err != nil { + return errorResult(err.Error()), nil + } + if current == 0 { + return mcp.NewToolResultText("No golden thread set."), nil + } + if err := kdb.SetGoldenThread(0); err != nil { + return errorResult(err.Error()), nil + } + t, err := kdb.GetThread(current) + if err != nil { + return errorResult(err.Error()), nil + } + name := "?" + if t != nil { + name = t.Title + } + return mcp.NewToolResultText(fmt.Sprintf("Golden thread cleared (was #%d: %s)", current, name)), nil + } + + if err := threadExists(kdb, threadID); err != nil { + return errorResult(err.Error()), nil + } if err := kdb.SetGoldenThread(threadID); err != nil { return errorResult(err.Error()), nil } - t, _ := kdb.GetThread(threadID) - return mcp.NewToolResultText(fmt.Sprintf("Golden thread set to #%d: %s", threadID, t.Title)), nil + t, err := kdb.GetThread(threadID) + if err != nil { + return errorResult(err.Error()), nil + } + name := "?" + if t != nil { + name = t.Title + } + return mcp.NewToolResultText(fmt.Sprintf("Golden thread set to #%d: %s", threadID, name)), nil }) // ─── knox_topics ─────────────────────────────────────────── @@ -438,7 +534,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer { ) s.AddTool(topicsTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { - limit := clampInt(req, "limit", 10, 1, 30) + limit, err := optionalIntArg(req, "limit", 10, 1, 30) + if err != nil { + return errorResult(err.Error()), nil + } entries, _ := kdb.RecentEntries(2000) tfidf := index.BuildTFIDF(entries) clusters := tfidf.Cluster(2, limit) @@ -618,18 +717,59 @@ func getString(req mcp.CallToolRequest, key, def string) string { return def } -func clampInt(req mcp.CallToolRequest, key string, def, min, max int) int { - if v, ok := req.Params.Arguments[key].(float64); ok { - n := int(v) - if n < min { - return min - } - if n > max { - return max - } - return n +// requiredStringArg returns a non-empty string argument or a descriptive error. +// Empty/missing/wrong-type values are rejected rather than silently defaulted: +// LLM clients routinely omit or zero value params, and a silent default writes +// to the wrong thread or reports false success. +func requiredStringArg(req mcp.CallToolRequest, key string) (string, error) { + v, _ := req.Params.Arguments[key].(string) + v = strings.TrimSpace(v) + if v == "" { + return "", fmt.Errorf("%s is required and must be a non-empty string", key) } - return def + return v, nil +} + +// requiredIntArg returns an integer argument validated against [min, max]. +func requiredIntArg(req mcp.CallToolRequest, key string, min, max int64) (int64, error) { + v, ok := req.Params.Arguments[key].(float64) + if !ok || v != math.Trunc(v) { + return 0, fmt.Errorf("%s is required and must be an integer", key) + } + n := int64(v) + if n < min || n > max { + return 0, fmt.Errorf("%s must be in [%d..%d], got %d", key, min, max, n) + } + return n, nil +} + +// optionalIntArg validates a present numeric argument, defaulting when absent. +func optionalIntArg(req mcp.CallToolRequest, key string, def, min, max int) (int, error) { + v, ok := req.Params.Arguments[key].(float64) + if !ok { + return def, nil + } + if v != math.Trunc(v) { + return 0, fmt.Errorf("%s must be an integer", key) + } + n := int(v) + if n < min || n > max { + return 0, fmt.Errorf("%s must be in [%d..%d], got %d", key, min, max, n) + } + return n, nil +} + +// threadExists is a guard for write tools: link/relate/set-golden must not +// silently accept ids with no matching thread. +func threadExists(kdb *db.KnoxDB, id int64) error { + t, err := kdb.GetThread(id) + if err != nil { + return err + } + if t == nil { + return fmt.Errorf("thread #%d not found", id) + } + return nil } func shortFP(fp string) string { diff --git a/internal/cmd/status.go b/internal/cmd/status.go index 2c0cbd6..c8c4130 100644 --- a/internal/cmd/status.go +++ b/internal/cmd/status.go @@ -13,7 +13,7 @@ import ( func paginate(entries []db.Entry, page, limit int) []db.Entry { start := (page - 1) * limit - if start >= len(entries) { + if start < 0 || start >= len(entries) { return nil } end := start + limit @@ -74,16 +74,22 @@ func NewQueryCmd(kdb *db.KnoxDB) *cobra.Command { if query == "" { return fmt.Errorf("search query required (positional arg or stdin pipe)") } - results, err := kdb.Search(query, limit) + if page < 1 || limit < 1 { + return fmt.Errorf("page and limit must be >= 1 (got page=%d limit=%d)", page, limit) + } + // Fetch enough rows to cover the requested page: Search caps at its + // limit argument, so slicing a limit-sized result set could never + // reach page 2+. + all, err := kdb.Search(query, page*limit) if err != nil { return err } - results = paginate(results, page, limit) + results := paginate(all, page, limit) if len(results) == 0 { fmt.Println("No results found.") return nil } - fmt.Printf("Found %d results for %q (page %d, %d per page):\n\n", len(results), query, page, limit) + fmt.Printf("Found %d results for %q — showing %d (page %d, %d per page):\n\n", len(all), query, len(results), page, limit) for _, r := range results { fmt.Printf(" %-8s %-30s [%s] %s\n", shortFP(r.Fingerprint), truncateStr(r.Title, 30), r.Project, r.SourceID) if r.Summary != "" { @@ -105,13 +111,20 @@ func NewRecentCmd(kdb *db.KnoxDB) *cobra.Command { Use: "recent", Short: "Show recent knowledge entries", RunE: func(c *cobra.Command, args []string) error { - all, err := kdb.RecentEntries(limit * 10) + if page < 1 || limit < 1 { + return fmt.Errorf("page and limit must be >= 1 (got page=%d limit=%d)", page, limit) + } + all, err := kdb.RecentEntries(page * limit) if err != nil { return err } entries := paginate(all, page, limit) if len(entries) == 0 { - fmt.Println("No entries yet. Run `knox watch` to start ingesting.") + if page > 1 { + fmt.Printf("No more entries on page %d (page %d of %d).\n", page, page, (len(all)+limit-1)/limit) + } else { + fmt.Println("No entries yet. Run `knox watch` to start ingesting.") + } return nil } fmt.Printf("Recent %d entries (page %d, %d per page):\n\n", len(entries), page, limit) diff --git a/internal/db/db.go b/internal/db/db.go index 12bb25e..4b8d453 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -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) diff --git a/internal/db/db_test.go b/internal/db/db_test.go new file mode 100644 index 0000000..f8d326f --- /dev/null +++ b/internal/db/db_test.go @@ -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) + } +} diff --git a/internal/db/gossip.go b/internal/db/gossip.go index 613f03f..4a3ffcd 100644 --- a/internal/db/gossip.go +++ b/internal/db/gossip.go @@ -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, diff --git a/internal/hlc/hlc.go b/internal/hlc/hlc.go index 2827e16..cc0370c 100644 --- a/internal/hlc/hlc.go +++ b/internal/hlc/hlc.go @@ -27,6 +27,23 @@ type Clock struct { func New() *Clock { return &Clock{} } +// SeekTo adopts the given packed HLC value when it is ahead of the clock's current +// position, so the next Now is still strictly increasing. Used to resume a node's +// clock from its persisted MAX(hcl) at startup — without it, a restart with a +// regressed wall clock would reissue already-used values and break the +// monotonicity the (node_id, hcl) locator uniqueness and gossip cursors rely on. +func (c *Clock) SeekTo(v int64) { + c.mu.Lock() + defer c.mu.Unlock() + + wall := v >> wallShift + seq := v & seqMask + if wall > c.wallMS || (wall == c.wallMS && seq > c.seq) { + c.wallMS = wall + c.seq = seq + } +} + // Now returns the next monotonic HLC value and the wall-clock time embedded in // it. The returned time is the HLC's wall component — never ahead of the local // clock beyond the current call and never rewinding across calls. diff --git a/internal/hlc/hlc_test.go b/internal/hlc/hlc_test.go new file mode 100644 index 0000000..9a530d0 --- /dev/null +++ b/internal/hlc/hlc_test.go @@ -0,0 +1,68 @@ +package hlc + +import "testing" + +// TestSeekToFutureValueKeepsMonotonic: after resuming from a persisted value +// ahead of the wall clock (clock regression), every subsequent Now must still +// be strictly greater than the resumed position. +func TestSeekToFutureValueKeepsMonotonic(t *testing.T) { + c := New() + resumed := int64(1) << 62 // packed value far ahead of any real wall clock + c.SeekTo(resumed) + + prev, _ := c.Now() + if prev <= resumed { + t.Fatalf("first Now after SeekTo = %d, want > resumed %d", prev, resumed) + } + for i := 0; i < 100; i++ { + next, _ := c.Now() + if next <= prev { + t.Fatalf("HLC regressed: %d then %d", prev, next) + } + prev = next + } +} + +// TestSeekToLowerValueIgnored: resuming from a value behind the current clock +// (or a fresh 0-padded DB) must not rewind it. +func TestSeekToLowerValueIgnored(t *testing.T) { + c := New() + first, _ := c.Now() + c.SeekTo(0) + second, _ := c.Now() + if second <= first { + t.Fatalf("SeekTo(0) rewound the clock: %d then %d", first, second) + } + + // Seek to exactly the last emitted value: the next value must exceed it. + c.SeekTo(second) + third, _ := c.Now() + if third <= second { + t.Fatalf("SeekTo(last) did not preserve monotonicity: %d then %d", second, third) + } +} + +// TestSeekToAcrossRestart mirrors Open's reopen path: a fresh clock resumed +// from the persisted max keeps issuing strictly increasing values. +func TestSeekToAcrossRestart(t *testing.T) { + c1 := New() + var last int64 + for i := 0; i < 50; i++ { + last, _ = c1.Now() + } + + c2 := New() // fresh process clock + c2.SeekTo(last) + + prev, _ := c2.Now() + if prev <= last { + t.Fatalf("reopened clock reissued a value: %d <= %d", prev, last) + } + for i := 0; i < 50; i++ { + next, _ := c2.Now() + if next <= prev { + t.Fatalf("reopened clock regressed: %d then %d", prev, next) + } + prev = next + } +} diff --git a/internal/watch/gossip.go b/internal/watch/gossip.go index 313b845..7190348 100644 --- a/internal/watch/gossip.go +++ b/internal/watch/gossip.go @@ -98,10 +98,16 @@ func nextCursor(rows []db.GossipObservation) int64 { return rows[len(rows)-1].HCL } +// maxBatchRows bounds the number of observations a peer may push in one POST. +// Pull already pages at 500 rows, so any larger batch is at best redundant and +// at worst a flood; capping keeps memory and insert work bounded. +const maxBatchRows = 1000 + func (n *Node) handleBatch(w http.ResponseWriter, r *http.Request) { + r.Body = http.MaxBytesReader(w, r.Body, 4<<20) // 4 MiB body, err := io.ReadAll(r.Body) if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) + http.Error(w, "batch too large or unreadable", http.StatusBadRequest) return } var rows []db.GossipObservation @@ -109,6 +115,10 @@ func (n *Node) handleBatch(w http.ResponseWriter, r *http.Request) { http.Error(w, "bad json: "+err.Error(), http.StatusBadRequest) return } + if len(rows) > maxBatchRows { + http.Error(w, fmt.Sprintf("batch too large: %d rows (max %d)", len(rows), maxBatchRows), http.StatusBadRequest) + return + } inserted, err := n.Kdb.PushObservations(rows) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) @@ -273,7 +283,7 @@ func (c *Client) Diff() (*DiffSummary, error) { // node pulls/pushes directly with every other node it learns about. // // m, when non-nil, receives gossip event counters (nil for one-shot CLI runs). -func Run(kdb *db.KnoxDB, m *metrics.Metrics, static []string) { +func Run(kdb *db.KnoxDB, m *metrics.Metrics, static []string) int { myID := kdb.NodeID() // Seed the work queue with static config plus persisted discoveries. @@ -284,6 +294,7 @@ func Run(kdb *db.KnoxDB, m *metrics.Metrics, static []string) { seen := make(map[string]bool) // addr → handled (also suppresses self) queue := 0 + pulledTotal := 0 for queue < len(work) { addr := strings.TrimSpace(work[queue]) queue++ @@ -338,6 +349,7 @@ func Run(kdb *db.KnoxDB, m *metrics.Metrics, static []string) { pulled += n } } + pulledTotal += pulled if m != nil { m.IncrementPull(pulled) } @@ -356,6 +368,7 @@ func Run(kdb *db.KnoxDB, m *metrics.Metrics, static []string) { log.Printf("[gossip] synced with %s (%s): in sync (pushed %d)", p.NodeID, addr, pushed) } } + return pulledTotal } func kdbVectorGet(kdb *db.KnoxDB, nodeID string) int64 { diff --git a/internal/watch/gossip_test.go b/internal/watch/gossip_test.go index 5d3ba34..9a5649c 100644 --- a/internal/watch/gossip_test.go +++ b/internal/watch/gossip_test.go @@ -1,8 +1,12 @@ package watch import ( + "bytes" + "encoding/json" + "net/http" "net/http/httptest" "path/filepath" + "strings" "testing" "github.com/david/knox/internal/db" @@ -228,16 +232,116 @@ func TestGossipIdempotent(t *testing.T) { defer sb.Close() Run(b, nil, []string{sa.URL}) - before, err := b.EntryCount() + entryBefore, err := b.EntryCount() if err != nil { t.Fatal(err) } + stats, err := b.Stats() + if err != nil { + t.Fatal(err) + } + obsBefore, _ := stats["total_observations"].(int) + Run(b, nil, []string{sa.URL}) - after, err := b.EntryCount() + + entryAfter, err := b.EntryCount() if err != nil { t.Fatal(err) } - if before != after { - t.Errorf("second sweep changed entry count: %d -> %d", before, after) + if entryBefore != entryAfter { + t.Errorf("second sweep changed entry count: %d -> %d", entryBefore, entryAfter) + } + stats, err = b.Stats() + if err != nil { + t.Fatal(err) + } + obsAfter, _ := stats["total_observations"].(int) + if obsBefore != obsAfter { + t.Errorf("second sweep duplicated observations: %d -> %d", obsBefore, obsAfter) + } +} + +// TestHandleBatchRejectsOversizedBatch: more than maxBatchRows in one POST +// must be refused up front, before any insert work. +func TestHandleBatchRejectsOversizedBatch(t *testing.T) { + b := tmpKnoxDB(t) + node := &Node{Kdb: b, Name: "B"} + sv := httptest.NewServer(node.Handler()) + defer sv.Close() + + rows := make([]db.GossipObservation, maxBatchRows+1) + for i := range rows { + rows[i] = db.GossipObservation{ + NodeID: "0123456789abcdef0123456789abcdef", HCL: int64(i + 1), + Fingerprint: "fp:oversized", SourceID: "test", Title: "t", Summary: "s", + CollectedAt: "2026-08-29T00:00:00Z", + } + } + body, _ := json.Marshal(rows) + resp, err := http.Post(sv.URL+"/v1/obs/batch", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatalf("post: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Errorf("oversized batch: want 400, got %d", resp.StatusCode) + } +} + +// TestHandleBatchRejectsHugeBody: an oversized body (beyond the 4 MiB cap) +// must be refused even when the row count is small. +func TestHandleBatchRejectsHugeBody(t *testing.T) { + b := tmpKnoxDB(t) + node := &Node{Kdb: b, Name: "B"} + sv := httptest.NewServer(node.Handler()) + defer sv.Close() + + rows := []db.GossipObservation{{ + NodeID: "0123456789abcdef0123456789abcdef", HCL: 1, + Fingerprint: "fp:huge", SourceID: "test", Title: "t", + Summary: strings.Repeat("x", 5<<20), // 5 MiB summary + CollectedAt: "2026-08-29T00:00:00Z", + }} + body, _ := json.Marshal(rows) + resp, err := http.Post(sv.URL+"/v1/obs/batch", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatalf("post: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Errorf("huge body: want 400, got %d", resp.StatusCode) + } +} + +// TestHandleBatchSkipsSelfRows: rows claiming the receiver's own node_id are +// dropped at the HTTP layer too (the poisoning vector), reported as conflicts. +func TestHandleBatchSkipsSelfRows(t *testing.T) { + b := tmpKnoxDB(t) + node := &Node{Kdb: b, Name: "B"} + sv := httptest.NewServer(node.Handler()) + defer sv.Close() + + rows := []db.GossipObservation{ + {NodeID: b.NodeID(), HCL: 1, Fingerprint: "fp:self1", SourceID: "test", Title: "t", Summary: "s", CollectedAt: "2026-08-29T00:00:00Z"}, + {NodeID: b.NodeID(), HCL: 2, Fingerprint: "fp:self2", SourceID: "test", Title: "t", Summary: "s", CollectedAt: "2026-08-29T00:00:00Z"}, + } + body, _ := json.Marshal(rows) + resp, err := http.Post(sv.URL+"/v1/obs/batch", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatalf("post: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("want 200, got %d", resp.StatusCode) + } + var out struct { + Accepted int `json:"accepted"` + Conflict int `json:"conflict"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + t.Fatalf("decode: %v", err) + } + if out.Accepted != 0 || out.Conflict != 2 { + t.Errorf("self rows: want accepted=0 conflict=2, got accepted=%d conflict=%d", out.Accepted, out.Conflict) } } \ No newline at end of file diff --git a/internal/watch/watch.go b/internal/watch/watch.go index a921344..36b923d 100644 --- a/internal/watch/watch.go +++ b/internal/watch/watch.go @@ -1,11 +1,14 @@ package watch import ( + "io/fs" "log" "net/http" "os" "path/filepath" "strings" + "sync" + "sync/atomic" "time" "github.com/david/knox/internal/db" @@ -28,6 +31,44 @@ type Watcher struct { metrics *metrics.Metrics } +// fileDebouncer schedules one ingest per path a settle-window after the last +// relevant event (trailing edge): writers that emit bursts (multi-write appends, +// atomic save = temp-write + rename) settle before anything is read, so a +// partial file is never recorded as final state. Timers are removed once they +// fire, keeping the map bounded by recently-active paths. +type fileDebouncer struct { + mu sync.Mutex + timers map[string]*time.Timer +} + +func newFileDebouncer() *fileDebouncer { + return &fileDebouncer{timers: make(map[string]*time.Timer)} +} + +func (d *fileDebouncer) schedule(path string, delay time.Duration, fn func(string)) { + d.mu.Lock() + defer d.mu.Unlock() + if t, ok := d.timers[path]; ok { + t.Stop() + } + d.timers[path] = time.AfterFunc(delay, func() { + d.mu.Lock() + delete(d.timers, path) + d.mu.Unlock() + fn(path) + }) +} + +// cancel drops any pending ingest for path (e.g. the file was deleted). +func (d *fileDebouncer) cancel(path string) { + d.mu.Lock() + defer d.mu.Unlock() + if t, ok := d.timers[path]; ok { + t.Stop() + delete(d.timers, path) + } +} + func New(kdb *db.KnoxDB, dirs []string) *Watcher { // Detect Obsidian vault vault, _ := ingest.DetectObsidianVault() @@ -86,17 +127,13 @@ func (w *Watcher) Start() error { log.Printf("[knox] obsidian vault: %s", w.vault) } - debounceMap := make(map[string]time.Time) + debounce := newFileDebouncer() browserTicker := time.NewTicker(browserInterval) giteaTicker := time.NewTicker(10 * time.Minute) gitTicker := time.NewTicker(10 * time.Minute) threadTicker := time.NewTicker(10 * time.Minute) gossipTicker := time.NewTicker(gossipInterval) - browserRunning := false - giteaRunning := false - gitRunning := false - threadRunning := false - gossipRunning := false + var browserRunning, giteaRunning, gitRunning, threadRunning, gossipRunning atomic.Bool for { select { @@ -104,63 +141,76 @@ func (w *Watcher) Start() error { if !ok { return nil } - if !w.isRelevantEvent(event) { - continue + + // fsnotify is non-recursive: files appearing inside newly created + // subdirectories would otherwise be invisible to the daemon. + isDir := false + if event.Has(fsnotify.Create) { + if info, err := os.Stat(event.Name); err == nil && info.IsDir() { + isDir = true + if err := watcher.Add(event.Name); err != nil { + log.Printf("[knox] cannot watch new dir %s: %v", event.Name, err) + } else { + log.Printf("[knox] watching new dir %s", event.Name) + } + } } - now := time.Now() - if last, ok := debounceMap[event.Name]; ok && now.Sub(last) < w.debounce { + if event.Has(fsnotify.Remove) { + // Drop any pending ingest for a deleted file. (No tombstone is + // written yet — the entry lingers until reconcile/prune.) + debounce.cancel(event.Name) + continue + } + if isDir || !w.isRelevantEvent(event) { continue } - debounceMap[event.Name] = now trigger := eventOpName(event.Op) log.Printf("[knox] %s %s", trigger, filepath.Base(event.Name)) - w.ingestFile(event.Name, trigger) + name := event.Name + debounce.schedule(name, w.debounce, func(path string) { + w.ingestFile(path, trigger) + }) case <-browserTicker.C: - if browserRunning { + if !browserRunning.CompareAndSwap(false, true) { continue } - browserRunning = true go func() { - defer func() { browserRunning = false }() + defer browserRunning.Store(false) w.ingestBrowserHistory() }() case <-giteaTicker.C: - if giteaRunning { + if !giteaRunning.CompareAndSwap(false, true) { continue } - giteaRunning = true go func() { - defer func() { giteaRunning = false }() + defer giteaRunning.Store(false) w.ingestGitea() }() case <-gitTicker.C: - if gitRunning { + if !gitRunning.CompareAndSwap(false, true) { continue } - gitRunning = true go func() { - defer func() { gitRunning = false }() + defer gitRunning.Store(false) w.ingestGit() }() case <-threadTicker.C: - if threadRunning { + if !threadRunning.CompareAndSwap(false, true) { continue } - threadRunning = true go func() { - defer func() { threadRunning = false }() + defer threadRunning.Store(false) w.autoThread() }() case <-gossipTicker.C: - if gossipRunning { + if !gossipRunning.CompareAndSwap(false, true) { continue } - gossipRunning = true go func() { - defer func() { gossipRunning = false }() + defer gossipRunning.Store(false) w.syncGossip() }() @@ -175,44 +225,38 @@ func (w *Watcher) Start() error { func (w *Watcher) seed(watcher *fsnotify.Watcher) { for _, dir := range w.dirs { - abs, _ := filepath.Abs(dir) - if err := watcher.Add(abs); err != nil { - log.Printf("[knox] cannot watch %s: %v", abs, err) - continue + if err := w.watchTree(watcher, dir); err != nil { + log.Printf("[knox] cannot watch %s: %v", dir, err) } - log.Printf("[knox] watching %s", abs) } - // Watch Obsidian vault + // Watch Obsidian vault (every subdirectory, non-recursively mirrored) if w.vault != "" { - if err := watcher.Add(w.vault); err != nil { + if err := w.watchTree(watcher, w.vault); err != nil { log.Printf("[knox] cannot watch obsidian vault %s: %v", w.vault, err) - } else { - log.Printf("[knox] watching %s (obsidian)", w.vault) } } - // Seed existing files - for _, ing := range w.fileIngesters { - for _, dir := range w.dirs { - patterns := []string{ - filepath.Join(dir, "*"), - filepath.Join(dir, "*", "SKILL.md"), + // Seed existing files anywhere under the watched dirs. fsnotify watches the + // whole tree, so live events cover any depth; this walk covers startup so + // pre-existing nested files (e.g. skills at two+ levels) are indexed too. + for _, dir := range w.dirs { + _ = filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error { + if err != nil || d.IsDir() { + return nil // skip unreadable entries; dirs are handled by the watch } - for _, pattern := range patterns { - entries, _ := filepath.Glob(pattern) - for _, path := range entries { - if !MatchesIngester(path, ing.SourceID()) { - continue - } + for _, ing := range w.fileIngesters { + if MatchesIngester(path, ing.SourceID()) { w.ingestFileWith(path, ing, "seed") + return nil } } - } + return nil + }) } // Seed Obsidian notes via the full walk: skips dot-dirs (.trash, .obsidian) - // and covers all depths — glob patterns would match dot-dirs and miss depth >2. + // and covers all depths. if w.vault != "" { if notes, err := ingest.NewObsidianIngester(w.vault).IngestAll(); err == nil { for _, r := range notes { @@ -222,6 +266,39 @@ func (w *Watcher) seed(watcher *fsnotify.Watcher) { } } +// watchTree adds a directory and every non-hidden subdirectory to the watcher, +// mirroring fsnotify's non-recursive API with an explicit walk. Hidden +// directories (.git, .obsidian, .trash) are skipped so their churn doesn't burn +// inotify watches. +func (w *Watcher) watchTree(watcher *fsnotify.Watcher, root string) error { + rootAbs, err := filepath.Abs(root) + if err != nil { + return err + } + if err := watcher.Add(rootAbs); err != nil { + return err + } + log.Printf("[knox] watching %s", rootAbs) + return filepath.WalkDir(rootAbs, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return nil + } + if !d.IsDir() { + return nil + } + if path != rootAbs && strings.HasPrefix(d.Name(), ".") { + return filepath.SkipDir + } + if path == rootAbs { + return nil + } + if err := watcher.Add(path); err != nil { + log.Printf("[knox] cannot watch %s: %v", path, err) + } + return nil + }) +} + func (w *Watcher) ingestFile(path, trigger string) { for _, ing := range w.fileIngesters { if MatchesIngester(path, ing.SourceID()) { @@ -456,29 +533,23 @@ func (w *Watcher) syncGossip() { return } - before := 0 - if n, err := w.knoxDB.EntryCount(); err == nil { - before = n - } + pulled := Run(w.knoxDB, w.metrics, peers) - Run(w.knoxDB, w.metrics, peers) - - // If new observations arrived, reconcile to pick up entries/threads they - // imply (deterministic log → derived rebuild). - if after, err := w.knoxDB.EntryCount(); err == nil && after > before { + // A pull only appends to the observation log; the entries cache and + // auto-threads are derived state that reconcile rebuilds. Comparing entry + // counts can never trigger this (pushes/pulls never touch entries directly), + // so reconcile fires on the sweep's newly-inserted observation count. + if pulled > 0 { created, linked, err := Reconcile(w.knoxDB) if err != nil { log.Printf("[knox] gossip reconcile: %v", err) return } - log.Printf("[knox] gossip reconcile done: %d created, %d linked", created, linked) + log.Printf("[knox] gossip reconcile done: %d created, %d linked after pulling %d obs", created, linked, pulled) } } func (w *Watcher) isRelevantEvent(event fsnotify.Event) bool { - if event.Has(fsnotify.Remove) || event.Has(fsnotify.Rename) { - return false - } name := filepath.Base(event.Name) // Opencode session diffs and logs if strings.HasPrefix(name, "ses_") || strings.HasSuffix(name, ".log") { @@ -512,6 +583,8 @@ func eventOpName(op fsnotify.Op) string { return "inotify:WRITE" case op.Has(fsnotify.Chmod): return "inotify:CHMOD" + case op.Has(fsnotify.Rename): + return "inotify:RENAME" default: return "inotify:UNKNOWN" } -- 2.52.0