diff --git a/docs/gossip-protocol.md b/docs/gossip-protocol.md index 0e9a6ee..bd6dd08 100644 --- a/docs/gossip-protocol.md +++ b/docs/gossip-protocol.md @@ -225,10 +225,14 @@ UNIQUE index) making auto-creation idempotent; `knox reconcile` rebuilds create/0 link on re-run); a log-only DB reconstructs `entries` and thread `cluster_key`s bit-identical to the original. -**M3 — Peer protocol.** `peers` settings table, `/v1/ping`, `/v1/log` pull, -`/v1/obs/batch` push, handshake + periodic anti-entropy, echo suppression. -Verify: two nodes converge to identical logs after partition (integration test). -`time.Now()`-free fact paths confirmed by grep. +**M3 — Peer protocol.** DONE. `peers` table; HTTP API (`GET /v1/ping` with +knowledge vector, `GET /v1/log?node=&after=` pull with cursor paging, +`POST /v1/obs/batch` push); `watch` daemon serves its log and runs a periodic +anti-entropy sweep (`KNOX_PEERS`, `KNOX_PEER_ADDR`), reconcile-on-pull; echo +suppression by `node_id` ownership; `knox gossip status`. Verify: two nodes +converge to identical logs after a bidirectional sweep (integration test +`internal/watch/gossip_test.go`); second sweep is idempotent; `time.Now()`-free +fact paths confirmed by grep. **M4 — Ops & UX.** `knox gossip` subcommand (status/diff), reconcile-on-pull, tombstoned thread handling in diff output, logging, config (env `KNOX_PEER_ADDR`, diff --git a/internal/cmd/gossip.go b/internal/cmd/gossip.go new file mode 100644 index 0000000..2bb5a26 --- /dev/null +++ b/internal/cmd/gossip.go @@ -0,0 +1,50 @@ +package cmd + +import ( + "fmt" + + "github.com/spf13/cobra" + "github.com/david/knox/internal/db" + "github.com/david/knox/internal/watch" +) + +// NewGossipCmd exposes peer status/diff for the gossip protocol. +func NewGossipCmd(kdb *db.KnoxDB) *cobra.Command { + cmd := &cobra.Command{ + Use: "gossip", + Short: "Inspect peer-to-peer observation sync", + } + cmd.AddCommand( + &cobra.Command{ + Use: "status", + Short: "Show known peers and knowledge vectors", + RunE: func(c *cobra.Command, args []string) error { + peers, err := kdb.ListPeers() + if err != nil { + return err + } + vector, err := kdb.KnowledgeVector() + if err != nil { + return err + } + fmt.Printf("node_id: %s\n", kdb.NodeID()) + fmt.Printf("peers (static: %d configured):\n", len(watch.PeerAddrs())) + for _, p := range peers { + fmt.Printf(" %-24s %-28s last=%s vector=%d\n", p.PeerID, p.Addr, p.LastHandshake, p.Cursor) + } + if len(peers) == 0 { + fmt.Println(" (none — set KNOX_PEERS to sync with other nodes)") + } + fmt.Println("knowledge vector (node_id -> max hcl):") + for nid, hcl := range vector { + if nid == kdb.NodeID() { + continue + } + fmt.Printf(" %-24s -> %d\n", nid, hcl) + } + return nil + }, + }, + ) + return cmd +} \ No newline at end of file diff --git a/internal/cmd/reconcile.go b/internal/cmd/reconcile.go index 82910b9..13b74d6 100644 --- a/internal/cmd/reconcile.go +++ b/internal/cmd/reconcile.go @@ -26,27 +26,18 @@ observations it now holds.`, if dryRun { return dryRunReconcile(kdb) } - return reconcile(kdb) + created, linked, err := watch.Reconcile(kdb) + if err != nil { + return err + } + fmt.Printf("reconciled: entries rebuilt, %d threads created, %d observations linked\n", created, linked) + return nil }, } cmd.Flags().BoolVar(&dryRun, "dry-run", false, "report drift without writing") return cmd } -func reconcile(kdb *db.KnoxDB) error { - before, after, err := kdb.RebuildEntriesFromObservations() - if err != nil { - return fmt.Errorf("rebuild entries: %w", err) - } - threader := watch.NewAutoThreader(kdb) - created, linked, err := threader.AutoThread() - if err != nil { - return fmt.Errorf("auto-thread: %w", err) - } - fmt.Printf("reconciled: entries %d -> %d, %d threads created, %d observations linked\n", before, after, created, linked) - return nil -} - func dryRunReconcile(kdb *db.KnoxDB) error { // Ground truth from the log (computed in a throwaway way via a count of // what rebuild would produce) vs the current materialized cache. diff --git a/internal/db/gossip.go b/internal/db/gossip.go new file mode 100644 index 0000000..6c0f795 --- /dev/null +++ b/internal/db/gossip.go @@ -0,0 +1,182 @@ +package db + +import ( + "database/sql" + "fmt" +) + +// GossipObservation is the serializable wire form of an observation exchanged +// between nodes. It carries the locator (NodeID, HCL) so the receiver can +// idempotently INSERT OR IGNORE on the unique index. +type GossipObservation struct { + NodeID string + HCL int64 + Fingerprint string + SourceID string + SourcePath string + Project string + ContentType string + Title string + Summary string + CollectedAt string + CreatedAt string + LineStart int + LineEnd int + Confidence float64 + IngesterVersion string + Trigger string + Provenance string +} + +// PushObservations inserts gossip rows idempotently. Rows whose (node_id, hcl) +// already exist are ignored (a node's HCL is locally monotonic, so a given +// locator is immutable). Returns the number newly inserted. +func (k *KnoxDB) PushObservations(rows []GossipObservation) (int, error) { + tx, err := k.db.Begin() + if err != nil { + return 0, err + } + defer tx.Rollback() + + inserted := 0 + for _, o := range rows { + res, err := tx.Exec( + `INSERT OR IGNORE INTO observations + (fingerprint, source_id, source_path, project, content_type, title, summary, + collected_at, created_at, line_start, line_end, confidence, ingester_version, trigger, provenance, + node_id, hcl) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + o.Fingerprint, o.SourceID, o.SourcePath, o.Project, o.ContentType, o.Title, o.Summary, + o.CollectedAt, o.CreatedAt, o.LineStart, o.LineEnd, o.Confidence, o.IngesterVersion, o.Trigger, o.Provenance, + o.NodeID, o.HCL, + ) + if err != nil { + return inserted, fmt.Errorf("push observation: %w", err) + } + if n, _ := res.RowsAffected(); n > 0 { + inserted++ + } + } + return inserted, tx.Commit() +} + +// ObservationsAfter returns gossip rows for one node after a given HCL cursor, +// in HCL order. after <= 0 means "everything" (initial handshake). +func (k *KnoxDB) ObservationsAfter(nodeID string, after int64, limit int) ([]GossipObservation, error) { + if limit <= 0 { + limit = 500 + } + query := `SELECT node_id, hcl, fingerprint, source_id, COALESCE(source_path,''), COALESCE(project,''), + COALESCE(content_type,''), COALESCE(title,''), COALESCE(summary,''), + COALESCE(collected_at,''), COALESCE(created_at,''), + COALESCE(line_start,0), COALESCE(line_end,0), COALESCE(confidence,0.5), + COALESCE(ingester_version,''), COALESCE(trigger,''), COALESCE(provenance,'{}') + FROM observations WHERE node_id=?` + args := []any{nodeID} + if after > 0 { + query += ` AND hcl > ?` + args = append(args, after) + } + query += ` ORDER BY hcl ASC LIMIT ?` + args = append(args, limit) + + rows, err := k.db.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []GossipObservation + for rows.Next() { + var o GossipObservation + if err := rows.Scan(&o.NodeID, &o.HCL, &o.Fingerprint, &o.SourceID, &o.SourcePath, + &o.Project, &o.ContentType, &o.Title, &o.Summary, &o.CollectedAt, &o.CreatedAt, + &o.LineStart, &o.LineEnd, &o.Confidence, &o.IngesterVersion, &o.Trigger, &o.Provenance); err != nil { + return nil, err + } + out = append(out, o) + } + return out, rows.Err() +} + +// KnowledgeVector returns node_id → max HCL held locally. It is the anti-entropy +// summary exchanged at handshake. +func (k *KnoxDB) KnowledgeVector() (map[string]int64, error) { + rows, err := k.db.Query(`SELECT node_id, MAX(hcl) FROM observations WHERE node_id<>'' AND hcl IS NOT NULL GROUP BY node_id`) + if err != nil { + return nil, err + } + defer rows.Close() + + kv := make(map[string]int64) + for rows.Next() { + var nid string + var hcl int64 + if err := rows.Scan(&nid, &hcl); err != nil { + return nil, err + } + kv[nid] = hcl + } + return kv, rows.Err() +} + +// ObservePairs returns all observations from other nodes (for consume/echo +// suppression checks). +func (k *KnoxDB) ObserveForeign(nodeID string) (int, error) { + var n int + err := k.db.QueryRow(`SELECT COUNT(*) FROM observations WHERE node_id<>? AND node_id<>''`, nodeID).Scan(&n) + return n, err +} + +// UpsertPeer records a peer's last-handshake metadata. +func (k *KnoxDB) UpsertPeer(peerID, addr, name string, maxHCL int64) error { + _, err := k.db.Exec( + `INSERT INTO peers (peer_id, addr, name, last_handshake, cursor) VALUES (?, ?, ?, datetime('now'), ?) + ON CONFLICT(peer_id) DO UPDATE SET + addr=?, name=?, + last_handshake=datetime('now'), + cursor=?`, + peerID, addr, name, maxHCL, addr, name, maxHCL, + ) + return err +} + +// ListPeers returns known peers ordered by first-seen. +func (k *KnoxDB) ListPeers() ([]Peer, error) { + rows, err := k.db.Query(`SELECT peer_id, COALESCE(addr,''), COALESCE(name,''), COALESCE(last_handshake,''), COALESCE(cursor,0), COALESCE(created_at,'') FROM peers ORDER BY created_at`) + if err != nil { + return nil, err + } + defer rows.Close() + + var peers []Peer + for rows.Next() { + var p Peer + if err := rows.Scan(&p.PeerID, &p.Addr, &p.Name, &p.LastHandshake, &p.Cursor, &p.CreatedAt); err != nil { + return nil, err + } + peers = append(peers, p) + } + return peers, rows.Err() +} + +// GetPeerByAddr finds a peer whose address matches (statically configured). +func (k *KnoxDB) GetPeerByAddr(addr string) (*Peer, error) { + row := k.db.QueryRow(`SELECT peer_id, COALESCE(addr,''), COALESCE(name,''), COALESCE(last_handshake,''), COALESCE(cursor,0), COALESCE(created_at,'') FROM peers WHERE addr=?`, addr) + p := &Peer{} + err := row.Scan(&p.PeerID, &p.Addr, &p.Name, &p.LastHandshake, &p.Cursor, &p.CreatedAt) + if err == sql.ErrNoRows { + return nil, nil + } + return p, err +} + +// Peer models a row in the peers table. +type Peer struct { + PeerID string + Addr string + Name string + LastHandshake string + Cursor int64 + CreatedAt string +} \ No newline at end of file diff --git a/internal/db/schema.go b/internal/db/schema.go index 278dd43..75c2771 100644 --- a/internal/db/schema.go +++ b/internal/db/schema.go @@ -113,6 +113,16 @@ CREATE TABLE IF NOT EXISTS settings ( value TEXT ); +-- PEERS: other knox nodes known to this one (gossip ring) +CREATE TABLE IF NOT EXISTS peers ( + peer_id TEXT PRIMARY KEY, -- remote node_id + addr TEXT, -- e.g. http://192.168.1.20:8931 + name TEXT, + last_handshake TEXT, + cursor INTEGER DEFAULT 0, -- remote's max_hcl we've seen (knowledge) + created_at TEXT DEFAULT (datetime('now')) +); + CREATE INDEX IF NOT EXISTS idx_threads_status ON threads(status); CREATE TABLE IF NOT EXISTS thread_observations ( diff --git a/internal/watch/gossip.go b/internal/watch/gossip.go new file mode 100644 index 0000000..b4e5676 --- /dev/null +++ b/internal/watch/gossip.go @@ -0,0 +1,304 @@ +// Package watch implements the knox peer-to-peer observation sync. +// +// A node advertises a small HTTP/JSON API (ping, log pull, batch push). Peers +// periodically exchange knowledge vectors, pull what they lack, and push their +// own observations. Observations are immutable and merged idempotently on the +// (node_id, hcl) locator; materialized state is rebuilt by reconcile after a +// pull. +package watch + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "os" + "strings" + "time" + + "github.com/david/knox/internal/db" +) + +const defaultPort = "8931" + +type Node struct { + Kdb *db.KnoxDB + Name string + Addr string // advertised base URL, e.g. http://192.168.1.20:8931 +} + +// pingResponse is the anti-entropy summary returned by /v1/ping. +type pingResponse struct { + NodeID string `json:"node_id"` + Name string `json:"name"` + Vector map[string]int64 `json:"vector"` // node_id → max hcl + MaxHCL *int64 `json:"max_hcl,omitempty"` +} + +func (n *Node) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("GET /v1/ping", n.handlePing) + mux.HandleFunc("GET /v1/log", n.handleLog) + mux.HandleFunc("POST /v1/obs/batch", n.handleBatch) + return mux +} + +func (n *Node) handlePing(w http.ResponseWriter, r *http.Request) { + vector, err := n.Kdb.KnowledgeVector() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + writeJSON(w, pingResponse{ + NodeID: n.Kdb.NodeID(), + Name: n.Name, + Vector: vector, + }) +} + +func (n *Node) handleLog(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + nodeID := q.Get("node") + var after int64 + fmt.Sscanf(q.Get("after"), "%d", &after) + + if nodeID == "" { + http.Error(w, "node param required", http.StatusBadRequest) + return + } + rows, err := n.Kdb.ObservationsAfter(nodeID, after, 500) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + writeJSON(w, map[string]any{ + "node": nodeID, + "after": after, + "cursor": nextCursor(rows), + "rows": rows, + }) +} + +func nextCursor(rows []db.GossipObservation) int64 { + if len(rows) == 0 { + return 0 + } + return rows[len(rows)-1].HCL +} + +func (n *Node) handleBatch(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + var rows []db.GossipObservation + if err := json.Unmarshal(body, &rows); err != nil { + http.Error(w, "bad json: "+err.Error(), http.StatusBadRequest) + return + } + inserted, err := n.Kdb.PushObservations(rows) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + writeJSON(w, map[string]any{ + "accepted": inserted, + "conflict": len(rows) - inserted, + }) +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + _ = enc.Encode(v) +} + +// Client is the pull/push half used by the exchange loop. +type Client struct { + NodeID string + Addr string + Timeout time.Duration +} + +func (c *Client) Ping() (*pingResponse, error) { + var p pingResponse + if err := c.get(c.Addr+"/v1/ping", &p); err != nil { + return nil, err + } + return &p, nil +} + +// Pull fetches a peer node's observations newer than the local cursor, looping +// through cursor pages until caught up. Returns newly inserted count. +func (c *Client) Pull(remoteNodeID string, after int64, kdb *db.KnoxDB) (int, error) { + total := 0 + for { + var resp struct { + Rows []db.GossipObservation `json:"rows"` + } + url := fmt.Sprintf("%s/v1/log?node=%s&after=%d", c.Addr, remoteNodeID, after) + if err := c.get(url, &resp); err != nil { + return total, err + } + if len(resp.Rows) == 0 { + break + } + n, err := kdb.PushObservations(resp.Rows) + if err != nil { + return total, err + } + total += n + last := resp.Rows[len(resp.Rows)-1].HCL + if last <= after { + break // no progress — bail to avoid an infinite loop + } + after = last + if len(resp.Rows) < 500 { + break + } + } + return total, nil +} + +// Push sends this node's own observations newer than what the peer reported. +// Ownership by node_id prevents echo: we never resend observations we merely +// received from another node. +func (c *Client) Push(kdb *db.KnoxDB, peerVector map[string]int64) (int, error) { + local := kdb.NodeID() + after := peerVector[local] // what the peer already has from us + rows, err := kdb.ObservationsAfter(local, after, 500) + if err != nil { + return 0, err + } + if len(rows) == 0 { + return 0, nil + } + body, _ := json.Marshal(rows) + req, err := http.NewRequest(http.MethodPost, c.Addr+"/v1/obs/batch", bytes.NewReader(body)) + if err != nil { + return 0, err + } + req.Header.Set("Content-Type", "application/json") + resp, err := (&http.Client{Timeout: c.Timeout}).Do(req) + if err != nil { + return 0, err + } + defer resp.Body.Close() + var out struct { + Accepted int `json:"accepted"` + } + _ = json.NewDecoder(resp.Body).Decode(&out) + return out.Accepted, nil +} + +func (c *Client) get(url string, v any) error { + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + return err + } + resp, err := (&http.Client{Timeout: c.Timeout}).Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("%s %s: %s", resp.Request.Method, resp.Request.URL.String(), resp.Status) + } + return json.NewDecoder(resp.Body).Decode(v) +} + +// Run executes one anti-entropy sweep against the given peer addresses. +func Run(kdb *db.KnoxDB, peers []string) { + myID := kdb.NodeID() + for _, addr := range peers { + addr = strings.TrimSpace(addr) + if addr == "" { + continue + } + if strings.HasPrefix(addr, kdb.NodeID()+":") { + continue + } + c := &Client{Addr: addr, Timeout: 10 * time.Second} + p, err := c.Ping() + if err != nil { + log.Printf("[gossip] ping %s: %v", addr, err) + continue + } + if p.NodeID == myID { + continue // never talk to ourselves (or an aliased address) + } + + pulled := 0 + for remoteNode, remoteHCL := range p.Vector { + localHCL := kdbVectorGet(kdb, remoteNode) + if remoteNode == myID { + continue // the peer cannot tell us about ourselves + } + if remoteHCL > localHCL { + n, err := c.Pull(remoteNode, localHCL, kdb) + if err != nil { + log.Printf("[gossip] pull %s@%s: %v", remoteNode, addr, err) + continue + } + pulled += n + } + } + + pushed, _ := c.Push(kdb, p.Vector) + + if err := kdb.UpsertPeer(p.NodeID, addr, p.Name, vectorMax(p.Vector)); err != nil { + log.Printf("[gossip] peer upsert: %v", err) + } + if pulled > 0 { + log.Printf("[gossip] synced with %s (%s): pulled %d obs, pushed %d", p.NodeID, addr, pulled, pushed) + } else { + log.Printf("[gossip] synced with %s (%s): in sync (pushed %d)", p.NodeID, addr, pushed) + } + } +} + +func kdbVectorGet(kdb *db.KnoxDB, nodeID string) int64 { + kv, err := kdb.KnowledgeVector() + if err != nil { + return 0 + } + return kv[nodeID] +} + +func vectorMax(v map[string]int64) int64 { + var m int64 + for _, h := range v { + if h > m { + m = h + } + } + return m +} + +// ListenAddr returns the advertised HTTP address (KNOX_PEER_ADDR or default). +func ListenAddr() (addr string) { + if addr = os.Getenv("KNOX_PEER_ADDR"); addr != "" { + return addr + } + return "localhost:" + defaultPort +} + +// PeerAddrs returns the configured peer list (KNOX_PEERS, comma-separated). +func PeerAddrs() []string { + raw := os.Getenv("KNOX_PEERS") + if raw == "" { + return nil + } + var out []string + for _, p := range strings.Split(raw, ",") { + p = strings.TrimSpace(p) + if p != "" { + out = append(out, p) + } + } + return out +} \ No newline at end of file diff --git a/internal/watch/gossip_test.go b/internal/watch/gossip_test.go new file mode 100644 index 0000000..0eab2c1 --- /dev/null +++ b/internal/watch/gossip_test.go @@ -0,0 +1,114 @@ +package watch + +import ( + "net/http/httptest" + "path/filepath" + "testing" + + "github.com/david/knox/internal/db" +) + +func tmpKnoxDB(t *testing.T) *db.KnoxDB { + t.Helper() + kdb, err := db.Open(filepath.Join(t.TempDir(), "index.db")) + if err != nil { + t.Fatalf("open db: %v", err) + } + t.Cleanup(func() { kdb.Close() }) + return kdb +} + +func seedObs(k *db.KnoxDB, prefix string) { + // Insert distinct observations only via RecordObservation so they get this + // node's node_id and HCL. + for i := 0; i < 3; i++ { + _, _, err := k.RecordObservation(db.ObservationRecord{ + Fingerprint: prefix + ":" + string(rune('a'+i)), + SourceID: "test", + SourcePath: prefix, + Project: "itest", + ContentType: "test", + Title: prefix + string(rune('a'+i)), + Summary: "summary " + prefix + string(rune('a'+i)), + CreatedAt: "2026-08-29T00:00:00Z", + LineEnd: 0, + Confidence: 0.9, + IngesterVersion: "itest/v1", + }) + if err != nil { + panic(err) + } + } +} + +// TestGossipConvergence starts two nodes, seeds disjoint observations, and runs +// a bidirectional handshake. Both must end with the concatenation of both logs. +func TestGossipConvergence(t *testing.T) { + a := tmpKnoxDB(t) + b := tmpKnoxDB(t) + + seedObs(a, "AAA") + seedObs(b, "BBB") + + // Wire a and b as peer HTTP servers. + nodeA := &Node{Kdb: a, Name: "A"} + sa := httptest.NewServer(nodeA.Handler()) + defer sa.Close() + + nodeB := &Node{Kdb: b, Name: "B"} + sb := httptest.NewServer(nodeB.Handler()) + defer sb.Close() + + // A pulls from B, then B pulls from A (bidirectional sweep). + Run(a, []string{sb.URL}) + Run(b, []string{sa.URL}) + + av, err := a.KnowledgeVector() + if err != nil { + t.Fatalf("a vector: %v", err) + } + bv, err := b.KnowledgeVector() + if err != nil { + t.Fatalf("b vector: %v", err) + } + + if len(av) != 2 || len(bv) != 2 { + t.Fatalf("expected both nodes to hold 2 source logs, got a=%v b=%v", av, bv) + } + + // Same maximum HCL per origin node on both sides. + for nid, hcl := range av { + if bv[nid] != hcl { + t.Errorf("node %s max hcl mismatch: a=%d b=%d", nid, hcl, bv[nid]) + } + } +} + +// TestGossipIdempotent ensures a repeated sweep does not duplicate observations. +func TestGossipIdempotent(t *testing.T) { + a := tmpKnoxDB(t) + b := tmpKnoxDB(t) + + seedObs(a, "AAA") + + nodeA := &Node{Kdb: a, Name: "A"} + sa := httptest.NewServer(nodeA.Handler()) + defer sa.Close() + nodeB := &Node{Kdb: b, Name: "B"} + sb := httptest.NewServer(nodeB.Handler()) + defer sb.Close() + + Run(b, []string{sa.URL}) + before, err := b.EntryCount() + if err != nil { + t.Fatal(err) + } + Run(b, []string{sa.URL}) + after, err := b.EntryCount() + if err != nil { + t.Fatal(err) + } + if before != after { + t.Errorf("second sweep changed entry count: %d -> %d", before, after) + } +} \ No newline at end of file diff --git a/internal/watch/reconcile.go b/internal/watch/reconcile.go new file mode 100644 index 0000000..e64eac2 --- /dev/null +++ b/internal/watch/reconcile.go @@ -0,0 +1,23 @@ +package watch + +import ( + "fmt" + + "github.com/david/knox/internal/db" +) + +// Reconcile rebuilds all derived state from the observation log: entries are +// reconstituted via deterministic SQL aggregation, then the auto-threader +// re-links/re-creates threads idempotently by cluster_key. Returns the threads +// created and observations linked. +func Reconcile(kdb *db.KnoxDB) (int, int, error) { + if _, _, err := kdb.RebuildEntriesFromObservations(); err != nil { + return 0, 0, fmt.Errorf("rebuild entries: %w", err) + } + threader := NewAutoThreader(kdb) + created, linked, err := threader.AutoThread() + if err != nil { + return 0, 0, fmt.Errorf("auto-thread: %w", err) + } + return created, linked, nil +} \ No newline at end of file diff --git a/internal/watch/watch.go b/internal/watch/watch.go index 0c79d91..333197e 100644 --- a/internal/watch/watch.go +++ b/internal/watch/watch.go @@ -2,6 +2,7 @@ package watch import ( "log" + "net/http" "os" "path/filepath" "strings" @@ -12,7 +13,10 @@ import ( "github.com/fsnotify/fsnotify" ) -const browserInterval = 5 * time.Minute +const ( + browserInterval = 5 * time.Minute + gossipInterval = 1 * time.Minute +) type Watcher struct { knoxDB *db.KnoxDB @@ -53,15 +57,31 @@ func (w *Watcher) Start() error { log.Printf("[knox] obsidian vault: %s", w.vault) } + // Gossip: serve our observation log to peers and periodically converge. + gossipAddr := ListenAddr() + if peers := PeerAddrs(); len(peers) > 0 { + node := &Node{Kdb: w.knoxDB, Name: "knox", Addr: gossipAddr} + srv := &http.Server{Addr: gossipAddr, Handler: node.Handler()} + go func() { + log.Printf("[knox] gossip listening on %s", gossipAddr) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Printf("[knox] gossip server: %v", err) + } + }() + log.Printf("[knox] gossip peers: %v", peers) + } + debounceMap := make(map[string]time.Time) 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 for { select { @@ -119,6 +139,15 @@ func (w *Watcher) Start() error { defer func() { threadRunning = false }() w.autoThread() }() + case <-gossipTicker.C: + if gossipRunning { + continue + } + gossipRunning = true + go func() { + defer func() { gossipRunning = false }() + w.syncGossip() + }() case err, ok := <-watcher.Errors: if !ok { @@ -404,6 +433,33 @@ func (w *Watcher) autoThread() { } } +// syncGossip runs one anti-entropy sweep against configured peers and rebuilds +// derived state if anything new arrived. +func (w *Watcher) syncGossip() { + peers := PeerAddrs() + if len(peers) == 0 { + return + } + + before := 0 + if n, err := w.knoxDB.EntryCount(); err == nil { + before = n + } + + Run(w.knoxDB, 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 { + 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) + } +} + func (w *Watcher) isRelevantEvent(event fsnotify.Event) bool { if event.Has(fsnotify.Remove) || event.Has(fsnotify.Rename) { return false diff --git a/main.go b/main.go index a1b1228..3ad9313 100644 --- a/main.go +++ b/main.go @@ -64,6 +64,7 @@ and maintains a searchable index. Use 'knox watch' for daemon mode.`, root.AddCommand(knoxcmd.NewGiteaCmd(kdb)) root.AddCommand(knoxcmd.NewGitCmd(kdb)) root.AddCommand(knoxcmd.NewReconcileCmd(kdb)) + root.AddCommand(knoxcmd.NewGossipCmd(kdb)) root.AddCommand(newServeCmd(kdb)) // MCP server subcommand