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
This commit is contained in:
2026-08-29 05:41:52 -07:00
parent 48e9c39d7b
commit 25a7112d8a
6 changed files with 204 additions and 20 deletions
+11 -4
View File
@@ -125,14 +125,16 @@ a single node the rowid remains monotonic, so the current dedup query
### 5.1 Transport ### 5.1 Transport
Plain HTTP/JSON on a per-node advertized address (default port `8931`). Nodes Plain HTTP/JSON on a per-node advertized address (default port `8931`). Peers
discover peers via a static list in `settings` (M3). mDNS/rendezvous is future are seeded from a static list (`KNOX_PEERS`), then the swarm discovers itself:
work. 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: 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=<seq>&node=<id> → { cursor, rows: [observation...] } # pull GET /v1/log?after=<seq>&node=<id> → { cursor, rows: [observation...] } # pull
POST /v1/obs/batch → body: [observation...]; reply: { accepted n, conflict n } # push POST /v1/obs/batch → body: [observation...]; reply: { accepted n, conflict n } # push
GET /v1/diff → divergence summary (M4) 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 - **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=...`. 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 - **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 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 already has it and will pull from its origin) — this is the echo/loop
+6
View File
@@ -142,6 +142,12 @@ func Open(path string) (*KnoxDB, error) {
return nil, fmt.Errorf("create locator index: %w", err) 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 return kdb, nil
} }
+67
View File
@@ -141,6 +141,18 @@ func (k *KnoxDB) UpsertPeer(peerID, addr, name string, maxHCL int64) error {
return err 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. // ListPeers returns known peers ordered by first-seen.
func (k *KnoxDB) ListPeers() ([]Peer, error) { 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`) 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 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 // DistinctFingerprints returns the set of all observed fingerprints — the
// ground-truth index of what this node knows. Used by gossip diff. // ground-truth index of what this node knows. Used by gossip diff.
func (k *KnoxDB) DistinctFingerprints() (map[string]bool, error) { func (k *KnoxDB) DistinctFingerprints() (map[string]bool, error) {
-2
View File
@@ -105,8 +105,6 @@ CREATE TABLE IF NOT EXISTS threads (
cluster_key TEXT 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 -- SETTINGS: key-value store for runtime configuration
CREATE TABLE IF NOT EXISTS settings ( CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY, key TEXT PRIMARY KEY,
+49 -12
View File
@@ -31,10 +31,11 @@ type Node struct {
// pingResponse is the anti-entropy summary returned by /v1/ping. // pingResponse is the anti-entropy summary returned by /v1/ping.
type pingResponse struct { type pingResponse struct {
NodeID string `json:"node_id"` NodeID string `json:"node_id"`
Name string `json:"name"` Name string `json:"name"`
Vector map[string]int64 `json:"vector"` // node_id → max hcl Vector map[string]int64 `json:"vector"` // node_id → max hcl
MaxHCL *int64 `json:"max_hcl,omitempty"` Peers []db.PeerInfo `json:"peers"` // swarm membership this node knows
MaxHCL *int64 `json:"max_hcl,omitempty"`
} }
func (n *Node) Handler() http.Handler { 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) http.Error(w, err.Error(), http.StatusInternalServerError)
return return
} }
peers, err := n.Kdb.ShareablePeers()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
writeJSON(w, pingResponse{ writeJSON(w, pingResponse{
NodeID: n.Kdb.NodeID(), NodeID: n.Kdb.NodeID(),
Name: n.Name, Name: n.Name,
Vector: vector, Vector: vector,
Peers: peers,
}) })
} }
@@ -255,17 +262,32 @@ func (c *Client) Diff() (*DiffSummary, error) {
return &d, nil return &d, nil
} }
// Run executes one anti-entropy sweep against the given peer addresses. // Run executes one anti-entropy + membership sweep.
func Run(kdb *db.KnoxDB, peers []string) { //
// 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() myID := kdb.NodeID()
for _, addr := range peers {
addr = strings.TrimSpace(addr) // Seed the work queue with static config plus persisted discoveries.
if addr == "" { persisted, _ := kdb.SwarmPeerAddrs()
continue work := make([]string, 0, len(static)+len(persisted))
} work = append(work, static...)
if strings.HasPrefix(addr, kdb.NodeID()+":") { 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 continue
} }
seen[addr] = true
c := &Client{Addr: addr, Timeout: 10 * time.Second} c := &Client{Addr: addr, Timeout: 10 * time.Second}
p, err := c.Ping() p, err := c.Ping()
if err != nil { if err != nil {
@@ -276,6 +298,21 @@ func Run(kdb *db.KnoxDB, peers []string) {
continue // never talk to ourselves (or an aliased address) 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 pulled := 0
for remoteNode, remoteHCL := range p.Vector { for remoteNode, remoteHCL := range p.Vector {
localHCL := kdbVectorGet(kdb, remoteNode) localHCL := kdbVectorGet(kdb, remoteNode)
+71 -2
View File
@@ -84,8 +84,77 @@ func TestGossipConvergence(t *testing.T) {
} }
} }
// TestGossipDiff ensures /v1/diff reports per-node observation fingerprints and // TestGossipSwarmDiscovery: C only knows A. A knows B. When C sweeps A, it must
// tombstoned thread divergence. // 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) { func TestGossipDiff(t *testing.T) {
a := tmpKnoxDB(t) a := tmpKnoxDB(t)
b := tmpKnoxDB(t) b := tmpKnoxDB(t)