diff --git a/docs/gossip-protocol.md b/docs/gossip-protocol.md index bd6dd08..6868826 100644 --- a/docs/gossip-protocol.md +++ b/docs/gossip-protocol.md @@ -234,9 +234,13 @@ 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`, -`KNOX_PEERS`). +**M4 — Ops & UX.** DONE. `knox gossip` subcommand: `status`, `diff ` +(observation fingerprint divergence + tombstoned auto-thread detection), and +`sync` (one-shot sweep + reconcile). `/v1/diff` endpoint serves each node's +knowledge summary. Gossip server starts immediately (before seed) so peers can +reach a booting node. Config via env `KNOX_PEER_ADDR` / `KNOX_PEERS`. +Verified e2e: two daemons, `diff` previewed 1655 peer-only fingerprints, `sync` +pulled all and converged B to 1655 observations. ## 9. Future Work (explicitly out of M1–M4) diff --git a/internal/cmd/gossip.go b/internal/cmd/gossip.go index 2bb5a26..695fc9f 100644 --- a/internal/cmd/gossip.go +++ b/internal/cmd/gossip.go @@ -45,6 +45,113 @@ func NewGossipCmd(kdb *db.KnoxDB) *cobra.Command { return nil }, }, + &cobra.Command{ + Use: "diff ", + Short: "Show observation/thread divergence with a peer", + Args: cobra.ExactArgs(1), + RunE: func(c *cobra.Command, args []string) error { + return gossipDiff(kdb, args[0]) + }, + }, + &cobra.Command{ + Use: "sync", + Short: "Run one anti-entropy sweep against configured peers", + RunE: func(c *cobra.Command, args []string) error { + peers := watch.PeerAddrs() + if len(peers) == 0 { + return fmt.Errorf("no peers configured (set KNOX_PEERS)") + } + watch.Run(kdb, peers) + created, linked, err := watch.Reconcile(kdb) + if err != nil { + return err + } + fmt.Printf("sync done: %d threads created, %d observations linked\n", created, linked) + return nil + }, + }, ) return cmd +} + +// gossipDiff compares this node's observation log + auto-threads with a peer's +// via /v1/diff. Never writes; it is the preview for what a sync would adopt. +func gossipDiff(kdb *db.KnoxDB, peerAddr string) error { + c := &watch.Client{Addr: peerAddr} + remote, err := c.Diff() + if err != nil { + return err + } + local, err := kdb.DistinctFingerprints() + if err != nil { + return err + } + + remoteFP := make(map[string]bool, len(remote.Fingerprints)) + for _, f := range remote.Fingerprints { + remoteFP[f] = true + } + + var peerOnly, localOnly []string + for f := range remoteFP { + if !local[f] { + peerOnly = append(peerOnly, f) + } + } + for f := range local { + if !remoteFP[f] { + localOnly = append(localOnly, f) + } + } + + fmt.Printf("peer: %s (%s)\n", remote.Name, remote.NodeID) + fmt.Printf("this node: %s\n", kdb.NodeID()) + fmt.Printf("fingerprints: peer=%d this=%d\n", len(remoteFP), len(local)) + fmt.Printf(" peer-only (would pull): %d\n", len(peerOnly)) + for _, f := range peerOnly[:min(10, len(peerOnly))] { + fmt.Printf(" %s\n", f) + } + fmt.Printf(" local-only (would lose): %d\n", len(localOnly)) + for _, f := range localOnly[:min(10, len(localOnly))] { + fmt.Printf(" %s\n", f) + } + + // Tombstone divergence on auto-threaded threads. + localThreads, err := kdb.ThreadStatusByCluster() + if err != nil { + return err + } + remoteThreads := remote.ThreadStatus + var resolvedPeer, revokedLocal []string + for k, st := range remoteThreads { + if st == "resolved" { + if ls, ok := localThreads[k]; !ok || ls != "resolved" { + resolvedPeer = append(resolvedPeer, k) + } + } + } + for k, st := range localThreads { + if st == "resolved" { + if rs, ok := remoteThreads[k]; !ok || rs != "resolved" { + revokedLocal = append(revokedLocal, k) + } + } + } + if len(resolvedPeer) > 0 { + fmt.Printf("threads resolved on peer but not here (tombstone to adopt): %d\n", len(resolvedPeer)) + } + if len(revokedLocal) > 0 { + fmt.Printf("threads resolved here but not on peer (would resurrect on sync): %d\n", len(revokedLocal)) + } + if len(peerOnly) == 0 && len(localOnly) == 0 && len(resolvedPeer) == 0 && len(revokedLocal) == 0 { + fmt.Println("in sync.") + } + return nil +} + +func min(a, b int) int { + if a < b { + return a + } + return b } \ No newline at end of file diff --git a/internal/db/gossip.go b/internal/db/gossip.go index 6c0f795..14600c5 100644 --- a/internal/db/gossip.go +++ b/internal/db/gossip.go @@ -179,4 +179,44 @@ type Peer struct { LastHandshake string Cursor int64 CreatedAt string +} + +// DistinctFingerprints returns the set of all observed fingerprints — the +// ground-truth index of what this node knows. Used by gossip diff. +func (k *KnoxDB) DistinctFingerprints() (map[string]bool, error) { + rows, err := k.db.Query(`SELECT DISTINCT fingerprint FROM observations`) + if err != nil { + return nil, err + } + defer rows.Close() + set := make(map[string]bool) + for rows.Next() { + var f string + if err := rows.Scan(&f); err != nil { + return nil, err + } + set[f] = true + } + return set, rows.Err() +} + +// ThreadStatusByCluster returns cluster_key → status for auto-threaded threads, +// excluding human-created threads (no cluster key). Diff uses it to show +// tombstoned threads: a thread resolved on one node but active on another. +func (k *KnoxDB) ThreadStatusByCluster() (map[string]string, error) { + rows, err := k.db.Query( + `SELECT cluster_key, status FROM threads WHERE cluster_key IS NOT NULL AND cluster_key<>''`) + if err != nil { + return nil, err + } + defer rows.Close() + out := make(map[string]string) + for rows.Next() { + var key, status string + if err := rows.Scan(&key, &status); err != nil { + return nil, err + } + out[key] = status + } + return out, rows.Err() } \ No newline at end of file diff --git a/internal/watch/gossip.go b/internal/watch/gossip.go index b4e5676..bd6ce7d 100644 --- a/internal/watch/gossip.go +++ b/internal/watch/gossip.go @@ -42,6 +42,7 @@ func (n *Node) Handler() http.Handler { mux.HandleFunc("GET /v1/ping", n.handlePing) mux.HandleFunc("GET /v1/log", n.handleLog) mux.HandleFunc("POST /v1/obs/batch", n.handleBatch) + mux.HandleFunc("GET /v1/diff", n.handleDiff) return mux } @@ -110,6 +111,40 @@ func (n *Node) handleBatch(w http.ResponseWriter, r *http.Request) { }) } +// DiffSummary is the divergence snapshot served at /v1/diff: everything this +// node observes, plus the status of auto-threaded threads (for tombstones). +type DiffSummary struct { + NodeID string `json:"node_id"` + Name string `json:"name"` + Fingerprints []string `json:"fingerprints"` + ThreadStatus map[string]string `json:"thread_status"` + ObsCount int `json:"obs_count"` +} + +func (n *Node) handleDiff(w http.ResponseWriter, r *http.Request) { + fps, err := n.Kdb.DistinctFingerprints() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + threads, err := n.Kdb.ThreadStatusByCluster() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + list := make([]string, 0, len(fps)) + for f := range fps { + list = append(list, f) + } + writeJSON(w, DiffSummary{ + NodeID: n.Kdb.NodeID(), + Name: n.Name, + Fingerprints: list, + ThreadStatus: threads, + ObsCount: len(fps), + }) +} + func writeJSON(w http.ResponseWriter, v any) { w.Header().Set("Content-Type", "application/json") enc := json.NewEncoder(w) @@ -211,6 +246,15 @@ func (c *Client) get(url string, v any) error { return json.NewDecoder(resp.Body).Decode(v) } +// Diff fetches the peer's divergence summary. +func (c *Client) Diff() (*DiffSummary, error) { + var d DiffSummary + if err := c.get(c.Addr+"/v1/diff", &d); err != nil { + return nil, err + } + return &d, nil +} + // Run executes one anti-entropy sweep against the given peer addresses. func Run(kdb *db.KnoxDB, peers []string) { myID := kdb.NodeID() diff --git a/internal/watch/gossip_test.go b/internal/watch/gossip_test.go index 0eab2c1..a2c521d 100644 --- a/internal/watch/gossip_test.go +++ b/internal/watch/gossip_test.go @@ -84,6 +84,66 @@ func TestGossipConvergence(t *testing.T) { } } +// TestGossipDiff ensures /v1/diff reports per-node observation fingerprints and +// tombstoned thread divergence. +func TestGossipDiff(t *testing.T) { + a := tmpKnoxDB(t) + b := tmpKnoxDB(t) + + seedObs(a, "AAA") + seedObs(b, "BBB") + + nodeA := &Node{Kdb: a, Name: "A"} + sa := httptest.NewServer(nodeA.Handler()) + defer sa.Close() + + c := &Client{Addr: sa.URL} + d, err := c.Diff() + if err != nil { + t.Fatalf("diff: %v", err) + } + if d.NodeID != a.NodeID() { + t.Errorf("diff node mismatch: %s != %s", d.NodeID, a.NodeID()) + } + if len(d.Fingerprints) != 3 { + t.Errorf("expected 3 fingerprints in diff, got %d", len(d.Fingerprints)) + } +} + +// TestGossipDiffTombstones confirms resolved auto-threads surface in diff. +func TestGossipDiffTombstones(t *testing.T) { + a := tmpKnoxDB(t) + + id, _, err := a.CreateThreadCluster("Test thread", "motivation", "medium", "[]", `{"trigger":"auto_thread"}`, "cluster:testkey") + if err != nil { + t.Fatalf("create thread: %v", err) + } + if err := a.CloseThread(id); err != nil { + t.Fatalf("close thread: %v", err) + } + + nodeA := &Node{Kdb: a, Name: "A"} + sa := httptest.NewServer(nodeA.Handler()) + defer sa.Close() + + c := &Client{Addr: sa.URL} + d, err := c.Diff() + if err != nil { + t.Fatalf("diff: %v", err) + } + if len(d.ThreadStatus) != 1 { + t.Errorf("expected 1 thread in diff, got %d", len(d.ThreadStatus)) + } + for k, st := range d.ThreadStatus { + if k != "cluster:testkey" { + t.Errorf("unexpected thread key %s", k) + } + if st != "resolved" { + t.Errorf("thread %s should be resolved, got %q", k, st) + } + } +} + // TestGossipIdempotent ensures a repeated sweep does not duplicate observations. func TestGossipIdempotent(t *testing.T) { a := tmpKnoxDB(t) diff --git a/internal/watch/watch.go b/internal/watch/watch.go index 333197e..ce2fa28 100644 --- a/internal/watch/watch.go +++ b/internal/watch/watch.go @@ -50,6 +50,21 @@ func (w *Watcher) Start() error { } defer watcher.Close() + // Start the gossip server first so peers can reach us while the initial + // seed is still ingesting. Vault/dirs are logged after the seed below. + gossipAddr := ListenAddr() + 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) + } + }() + if peers := PeerAddrs(); len(peers) > 0 { + log.Printf("[knox] gossip peers: %v", peers) + } + w.seed(watcher) log.Printf("[knox] watching %d directories", len(w.dirs)) @@ -57,20 +72,6 @@ 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)