From 25a7112d8a0f9c74146f56c6dfd568dd3a5b96b4 Mon Sep 17 00:00:00 2001 From: David Gwilliam Date: Sat, 29 Aug 2026 05:41:52 -0700 Subject: [PATCH] feat: swarm membership discovery via peer-list gossip Refs #1 - /v1/ping now advertises the node's known peers (peer_id, addr, name) - Run sweeps static KNOX_PEERS + persisted discovered peers, enqueueing newly-learned addresses for direct sweeps (membership-only relay; no observation relay) - db: ShareablePeers, SwarmPeerAddrs, MergePeer (cursor-preserving discovery upsert), MaxHCLForNode - integration test: a node configured with a single seed discovers and pulls from other swarm members without direct configuration --- docs/gossip-protocol.md | 15 +++++-- internal/db/db.go | 6 +++ internal/db/gossip.go | 67 ++++++++++++++++++++++++++++++++ internal/db/schema.go | 2 - internal/watch/gossip.go | 61 +++++++++++++++++++++++------ internal/watch/gossip_test.go | 73 ++++++++++++++++++++++++++++++++++- 6 files changed, 204 insertions(+), 20 deletions(-) diff --git a/docs/gossip-protocol.md b/docs/gossip-protocol.md index 6868826..e60ce96 100644 --- a/docs/gossip-protocol.md +++ b/docs/gossip-protocol.md @@ -125,14 +125,16 @@ a single node the rowid remains monotonic, so the current dedup query ### 5.1 Transport -Plain HTTP/JSON on a per-node advertized address (default port `8931`). Nodes -discover peers via a static list in `settings` (M3). mDNS/rendezvous is future -work. +Plain HTTP/JSON on a per-node advertized address (default port `8931`). Peers +are seeded from a static list (`KNOX_PEERS`), then the swarm discovers itself: +each node advertises its known peer addresses in `/v1/ping`, and every sweep +enqueues newly-learned nodes for direct contact (membership gossip — no relay of +observations). mDNS/rendezvous is future work. Endpoints: ``` -GET /v1/ping → { node_id, name, max_hcl } +GET /v1/ping → { node_id, name, max_hcl, peers: [{peer_id, addr, name}] } GET /v1/log?after=&node= → { cursor, rows: [observation...] } # pull POST /v1/obs/batch → body: [observation...]; reply: { accepted n, conflict n } # push GET /v1/diff → divergence summary (M4) @@ -142,6 +144,11 @@ GET /v1/diff → divergence summary (M4) - **Knowledge vector:** each node tracks `peer_id → max_hcl consumed`. Anti-entropy is a pull: periodically (and on handshake) query each peer's `/v1/log?after=...`. +- **Membership gossip:** `/v1/ping` includes the responding node's known peers + (`peer_id`, `addr`, `name`). The caller merges them into its `peers` table and + enqueues their addresses for direct sweeps. A new node therefore joins the + whole swarm by configuring just one seed peer. Membership flows independently + of data — a node never relays another's observations, only its address. - **Push:** on a new local observation, best-effort `POST /v1/obs/batch` to known peers. A node does **not** re-broadcast something it merely received (that peer already has it and will pull from its origin) — this is the echo/loop diff --git a/internal/db/db.go b/internal/db/db.go index 4cc09b5..12bb25e 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -142,6 +142,12 @@ func Open(path string) (*KnoxDB, error) { return nil, fmt.Errorf("create locator index: %w", err) } + // Thread idempotency index must come after the cluster_key migration (a + // fresh DB has the column from Schema; an existing DB gets it above). + if _, err := db.Exec(`CREATE UNIQUE INDEX IF NOT EXISTS idx_threads_cluster ON threads(cluster_key) WHERE cluster_key IS NOT NULL AND cluster_key != ''`); err != nil { + return nil, fmt.Errorf("create cluster_key index: %w", err) + } + return kdb, nil } diff --git a/internal/db/gossip.go b/internal/db/gossip.go index 14600c5..613f03f 100644 --- a/internal/db/gossip.go +++ b/internal/db/gossip.go @@ -141,6 +141,18 @@ func (k *KnoxDB) UpsertPeer(peerID, addr, name string, maxHCL int64) error { return err } +// MergePeer records a peer discovered indirectly (via another peer's ping). +// Unlike UpsertPeer it never clobbers the cursor — a freshly learned address +// has no known knowledge yet; the anti-entropy pull will set it on contact. +func (k *KnoxDB) MergePeer(peerID, addr, name string) error { + _, err := k.db.Exec( + `INSERT INTO peers (peer_id, addr, name) VALUES (?, ?, ?) + ON CONFLICT(peer_id) DO UPDATE SET addr=?, name=?`, + peerID, addr, name, addr, name, + ) + 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`) @@ -181,6 +193,61 @@ type Peer struct { CreatedAt string } +// PeerInfo is the shareable (non-secret) subset of a peer that /v1/ping +// advertises so other nodes can discover the swarm. +type PeerInfo struct { + PeerID string `json:"peer_id"` + Addr string `json:"addr"` + Name string `json:"name"` +} + +// ShareablePeers returns the peers this node knows about, for dissemination in +// ping responses. Self and peers without an address are excluded. +func (k *KnoxDB) ShareablePeers() ([]PeerInfo, error) { + rows, err := k.db.Query( + `SELECT peer_id, COALESCE(addr,''), COALESCE(name,'') FROM peers WHERE addr<>'' AND peer_id<>? ORDER BY peer_id`, + k.NodeID(), + ) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []PeerInfo + for rows.Next() { + var p PeerInfo + if err := rows.Scan(&p.PeerID, &p.Addr, &p.Name); err != nil { + return nil, err + } + out = append(out, p) + } + return out, rows.Err() +} + +// MaxHCLForNode returns the maximum HCL this node holds for a given origin +// node, or 0 if none. +func (k *KnoxDB) MaxHCLForNode(nodeID string) int64 { + var m int64 + if err := k.db.QueryRow(`SELECT MAX(hcl) FROM observations WHERE node_id=?`, nodeID).Scan(&m); err != nil { + return 0 + } + return m +} + +// SwarmPeerAddrs returns the addresses of all known peers (shareable set). It +// is what the sweep iterates after a bootstrap join. +func (k *KnoxDB) SwarmPeerAddrs() ([]string, error) { + peers, err := k.ShareablePeers() + if err != nil { + return nil, err + } + addrs := make([]string, 0, len(peers)) + for _, p := range peers { + addrs = append(addrs, p.Addr) + } + return addrs, nil +} + // 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) { diff --git a/internal/db/schema.go b/internal/db/schema.go index 75c2771..5f36049 100644 --- a/internal/db/schema.go +++ b/internal/db/schema.go @@ -105,8 +105,6 @@ CREATE TABLE IF NOT EXISTS threads ( cluster_key TEXT ); -CREATE UNIQUE INDEX IF NOT EXISTS idx_threads_cluster ON threads(cluster_key) WHERE cluster_key IS NOT NULL AND cluster_key != ''; - -- SETTINGS: key-value store for runtime configuration CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, diff --git a/internal/watch/gossip.go b/internal/watch/gossip.go index bd6ce7d..bf9382a 100644 --- a/internal/watch/gossip.go +++ b/internal/watch/gossip.go @@ -31,10 +31,11 @@ type Node struct { // 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"` + NodeID string `json:"node_id"` + Name string `json:"name"` + Vector map[string]int64 `json:"vector"` // node_id → max hcl + Peers []db.PeerInfo `json:"peers"` // swarm membership this node knows + MaxHCL *int64 `json:"max_hcl,omitempty"` } func (n *Node) Handler() http.Handler { @@ -52,10 +53,16 @@ func (n *Node) handlePing(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusInternalServerError) return } + peers, err := n.Kdb.ShareablePeers() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } writeJSON(w, pingResponse{ NodeID: n.Kdb.NodeID(), Name: n.Name, Vector: vector, + Peers: peers, }) } @@ -255,17 +262,32 @@ func (c *Client) Diff() (*DiffSummary, error) { return &d, nil } -// Run executes one anti-entropy sweep against the given peer addresses. -func Run(kdb *db.KnoxDB, peers []string) { +// Run executes one anti-entropy + membership sweep. +// +// Peers are a fusion of the statically configured list (KNOX_PEERS) and peers +// previously discovered and persisted in the peers table (swarm join): syncing +// with any one node reveals who else is in the swarm, and those nodes are then +// swept too. There is no relay of observations — only membership is shared; each +// node pulls/pushes directly with every other node it learns about. +func Run(kdb *db.KnoxDB, static []string) { myID := kdb.NodeID() - for _, addr := range peers { - addr = strings.TrimSpace(addr) - if addr == "" { - continue - } - if strings.HasPrefix(addr, kdb.NodeID()+":") { + + // Seed the work queue with static config plus persisted discoveries. + persisted, _ := kdb.SwarmPeerAddrs() + work := make([]string, 0, len(static)+len(persisted)) + work = append(work, static...) + work = append(work, persisted...) + + seen := make(map[string]bool) // addr → handled (also suppresses self) + queue := 0 + for queue < len(work) { + addr := strings.TrimSpace(work[queue]) + queue++ + if addr == "" || seen[addr] { continue } + seen[addr] = true + c := &Client{Addr: addr, Timeout: 10 * time.Second} p, err := c.Ping() if err != nil { @@ -276,6 +298,21 @@ func Run(kdb *db.KnoxDB, peers []string) { continue // never talk to ourselves (or an aliased address) } + // Membership discovery: learn who else is in the swarm and enqueue + // their addresses for direct sweeps. + for _, pi := range p.Peers { + if pi.PeerID == "" || pi.PeerID == myID || pi.Addr == "" { + continue + } + if err := kdb.MergePeer(pi.PeerID, pi.Addr, pi.Name); err != nil { + log.Printf("[gossip] merge peer %s: %v", pi.PeerID, err) + continue + } + if !seen[pi.Addr] { + work = append(work, pi.Addr) + } + } + pulled := 0 for remoteNode, remoteHCL := range p.Vector { localHCL := kdbVectorGet(kdb, remoteNode) diff --git a/internal/watch/gossip_test.go b/internal/watch/gossip_test.go index a2c521d..5e0f916 100644 --- a/internal/watch/gossip_test.go +++ b/internal/watch/gossip_test.go @@ -84,8 +84,77 @@ func TestGossipConvergence(t *testing.T) { } } -// TestGossipDiff ensures /v1/diff reports per-node observation fingerprints and -// tombstoned thread divergence. +// TestGossipSwarmDiscovery: C only knows A. A knows B. When C sweeps A, it must +// learn about B through A's ping, enqueue B, and pull B's observations — with no +// direct configuration of B (no relay of data, only membership). +func TestGossipSwarmDiscovery(t *testing.T) { + a := tmpKnoxDB(t) + b := tmpKnoxDB(t) + c := tmpKnoxDB(t) + + seedObs(a, "AAA") + seedObs(b, "BBB") + seedObs(c, "CCC") + + 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() + nodeC := &Node{Kdb: c, Name: "C"} + sc := httptest.NewServer(nodeC.Handler()) + defer sc.Close() + + // A discovers B (A pings B) so A can advertise B to the swarm. + Run(a, []string{sb.URL}) + + // C only knows A. A single sweep should surface B (membership in ping) + // and pull B's observations directly. + Run(c, []string{sa.URL}) + + // C must know B and hold all three origin logs. + peers, err := c.ListPeers() + if err != nil { + t.Fatalf("list peers: %v", err) + } + foundB := false + for _, p := range peers { + if p.PeerID == b.NodeID() { + foundB = true + } + } + if !foundB { + t.Fatalf("C did not discover B via A's membership list; peers=%v", peers) + } + + vec, err := c.KnowledgeVector() + if err != nil { + t.Fatalf("c vector: %v", err) + } + if len(vec) != 3 { + t.Errorf("C should hold 3 origin logs (A, B, C), got %v", vec) + } + + // C's copy of B's log must match B's own max hcl. + bmax := c.MaxHCLForNode(b.NodeID()) + bm, err := bNodeMax(b) + if err != nil { + t.Fatal(err) + } + if bmax != bm { + t.Errorf("C max hcl for B=%d, B reports %d", bmax, bm) + } +} + +// bNodeMax is a helper reading the max hcl B holds for its own node. +func bNodeMax(b *db.KnoxDB) (int64, error) { + rows, err := b.KnowledgeVector() + if err != nil { + return 0, err + } + return rows[b.NodeID()], nil +} func TestGossipDiff(t *testing.T) { a := tmpKnoxDB(t) b := tmpKnoxDB(t)