Distributed Cluster

The distributed layer sits on top of the single-node engine. Every shard is a full, independent copy of the single-node write path (WAL, then HNSW, then mmap snapshot), coordinated across nodes by the packages below.

The packages

• internal/cluster/raft: wraps hashicorp/raft, and governs cluster metadata only, meaning which nodes exist and which node leads each shard. It never touches a vector write directly. Heartbeat, election, lease, commit and propose timeouts are configurable.
• internal/cluster/ring: consistent hashing over a fixed number of shards, chosen once at cluster creation. Each shard is a stable unit with its own Raft-coordinated leader and replica set. It doesn't hash vector ids directly onto nodes, since that would make shard identity (and therefore replication) a moving target.
• internal/cluster/cluster.go: ties Raft and the ring together. Every node rebuilds the same shard assignment independently from the same replicated Raft state, with no separate gossip or broadcast step.
• internal/cluster/router: scatter-gather query routing. Insert/Delete hash a vector id (FNV-1a) to exactly one shard. Search fans out to every shard concurrently and merges each shard's own top-K into one globally-ranked top-K. A failing shard is retried once against its current owner (so a retry after failover reaches the new leader); if some shards still fail, the router returns the others' results along with a PartialError naming the failed shards. It takes gRPC dial options, so it can use TLS and send an API key.
• internal/cluster/replication: streams a shard leader's WAL to follower replicas over a plain TCP connection, kept outside Raft for write latency. The wire protocol is minimal: a follower sends its sequence number and the checksum of its record at that position, and the leader replies 'S' (a snapshot follows) or 'N' (stream from the live WAL). The leader sends a snapshot when the follower is too far behind, and also when it has diverged: it's ahead of the leader, or its record at that position doesn't match. Snapshots stream from and to disk with 64-bit lengths, and the stream wakes on each new write instead of polling.
• internal/cluster/health: only the current Raft leader probes node liveness, so different nodes can't reach different verdicts about who is down. It probes all nodes at once, so a dead node costs one timeout per round. There are two stages: a fast per-shard failover after FailureThreshold missed probes, then a full node eviction and shard rebalance after EvictThreshold. Failover promotes the replica that has applied the most WAL, when the caller supplies a way to ask replicas their position; otherwise the first healthy one.

Cost of sharding, measured

10,000 base vectors, 100 queries, dim=128, SIFT-small, single-node vs. a 4-shard cluster:

Both build with batches of 500; the router splits each batch by shard and sends every shard its part in parallel. RSS is summed across the four shard processes.

Cluster QPS drops 38% (at ef=100) to 45% (at ef=50) compared to single-node, from scatter-gather network hops and thinner per-shard candidate lists. Recall tracks closely: sharding by id doesn't remove vectors, it just changes which process holds them. At ef=10 the cluster's 0.978 even beats single-node's 0.910.

Testing the cluster

• Chaos tests: TestFullPartitionLosesQuorumThenHeals (a real toxiproxy network partition, not a process kill), TestRepeatedKillAndRestart, TestOpenFailsCleanlyOnUnwritableDataDir.
• Jepsen-style linearizability testing, using the porcupine checker (the same library Jepsen analyses use): TestLinearizableUnderNormalOperation and TestFailoverLinearizability. The failover test reports whichever outcome actually occurs under a real induced failover, rather than asserting one result, since both a linearizable and a non-linearizable outcome are possible depending on timing.

More detail in What Raft gave and cost.