feat: distributed gaps filled — gossip UDP transport, sharding data migration, inter-module wiring (raft-disttxn, gossip-sharding, replication-disttxn)

src/barabadb/core/gossip.nim

@@ -1,8 +1,12 @@
-## Gossip Protocol — membership and failure detection
+## Gossip Protocol — membership and failure detection with UDP transport
 import std/tables
 import std/random
 import std/monotimes
 import std/asyncdispatch
+import std/net
+import std/strutils
+import std/streams
+import std/os
 
 type
   NodeState* = enum
@@ -22,7 +26,7 @@ type
   GossipMessage* = object
     senderId*: string
     senderIncarnation*: uint64
-    nodes*: seq[(string, NodeState, uint64)]  # (id, state, incarnation)
+    nodes*: seq[(string, NodeState, uint64, string, int)]  # (id, state, incarnation, host, port)
 
   GossipProtocol* = ref object
     self*: GossipNode
@@ -30,9 +34,73 @@ type
     suspectTimeout*: int64    # nanoseconds
     deadTimeout*: int64       # nanoseconds
     fanout*: int              # number of nodes to gossip to per round
+    gossipPort*: int
+    running*: bool
+    sock*: Socket
     onJoin*: proc(node: GossipNode) {.gcsafe.}
     onLeave*: proc(nodeId: string) {.gcsafe.}
     onSuspect*: proc(nodeId: string) {.gcsafe.}
+    onMembershipChanged*: proc() {.gcsafe.}
+
+# ---------------------------------------------------------------------------
+# GossipMessage binary serialization (for UDP transport)
+# ---------------------------------------------------------------------------
+
+const GossipMagic = "GOSS"
+const GossipProtoVersion = 1'u32
+
+proc serialize*(msg: GossipMessage): seq[byte] =
+  let s = newStringStream()
+  s.write(GossipMagic)
+  s.write(GossipProtoVersion)
+  s.write(uint32(msg.senderId.len))
+  if msg.senderId.len > 0:
+    s.writeData(msg.senderId[0].unsafeAddr, msg.senderId.len)
+  s.write(msg.senderIncarnation)
+  s.write(uint32(msg.nodes.len))
+  for (id, state, incarnation, host, port) in msg.nodes:
+    s.write(uint32(id.len))
+    if id.len > 0:
+      s.writeData(id[0].unsafeAddr, id.len)
+    s.write(uint32(ord(state)))
+    s.write(incarnation)
+    s.write(uint32(host.len))
+    if host.len > 0:
+      s.writeData(host[0].unsafeAddr, host.len)
+    s.write(uint32(port))
+  let strData = s.data
+  result = newSeq[byte](strData.len)
+  for i in 0 ..< strData.len:
+    result[i] = byte(strData[i])
+  s.close()
+
+proc deserializeGossipMessage*(data: seq[byte]): GossipMessage =
+  let s = newStringStream(cast[string](data))
+  let magic = s.readStr(4)
+  if magic != GossipMagic:
+    raise newException(ValueError, "Invalid gossip magic")
+  let version = s.readUint32()
+  if version != GossipProtoVersion:
+    raise newException(ValueError, "Unsupported gossip protocol version")
+  let senderIdLen = int(s.readUint32())
+  result.senderId = if senderIdLen > 0: s.readStr(senderIdLen) else: ""
+  result.senderIncarnation = s.readUint64()
+  let nodeCount = int(s.readUint32())
+  result.nodes = newSeq[(string, NodeState, uint64, string, int)](nodeCount)
+  for i in 0 ..< nodeCount:
+    let idLen = int(s.readUint32())
+    let id = if idLen > 0: s.readStr(idLen) else: ""
+    let state = NodeState(s.readUint32())
+    let incarnation = s.readUint64()
+    let hostLen = int(s.readUint32())
+    let host = if hostLen > 0: s.readStr(hostLen) else: ""
+    let port = int(s.readUint32())
+    result.nodes[i] = (id, state, incarnation, host, port)
+  s.close()
+
+# ---------------------------------------------------------------------------
+# Core Gossip Protocol
+# ---------------------------------------------------------------------------
 
 proc newGossipNode*(id: string, host: string, port: int): GossipNode =
   GossipNode(
@@ -43,7 +111,7 @@ proc newGossipNode*(id: string, host: string, port: int): GossipNode =
   )
 
 proc newGossipProtocol*(id: string, host: string, port: int,
-                        fanout: int = 3): GossipProtocol =
+                        fanout: int = 3, gossipPort: int = 0): GossipProtocol =
   let self = newGossipNode(id, host, port)
   GossipProtocol(
     self: self,
@@ -51,27 +119,39 @@ proc newGossipProtocol*(id: string, host: string, port: int,
     suspectTimeout: 5_000_000_000,  # 5 seconds
     deadTimeout: 15_000_000_000,    # 15 seconds
     fanout: fanout,
+    gossipPort: gossipPort,
+    running: false,
+    sock: nil,
     onJoin: nil,
     onLeave: nil,
     onSuspect: nil,
+    onMembershipChanged: nil,
   )
 
 proc join*(gp: GossipProtocol, seedNode: GossipNode) =
   gp.members[seedNode.id] = seedNode
   if gp.onJoin != nil:
     gp.onJoin(seedNode)
+  if gp.onMembershipChanged != nil:
+    gp.onMembershipChanged()
 
 proc addMember*(gp: GossipProtocol, node: GossipNode) =
   if node.id == gp.self.id:
     return
+  let existed = node.id in gp.members
   gp.members[node.id] = node
-  if gp.onJoin != nil:
+  if not existed and gp.onJoin != nil:
     gp.onJoin(node)
+  if gp.onMembershipChanged != nil:
+    gp.onMembershipChanged()
 
 proc removeMember*(gp: GossipProtocol, nodeId: string) =
-  gp.members.del(nodeId)
-  if gp.onLeave != nil:
-    gp.onLeave(nodeId)
+  if nodeId in gp.members:
+    gp.members.del(nodeId)
+    if gp.onLeave != nil:
+      gp.onLeave(nodeId)
+    if gp.onMembershipChanged != nil:
+      gp.onMembershipChanged()
 
 proc suspect*(gp: GossipProtocol, nodeId: string) =
   if nodeId in gp.members:
@@ -84,6 +164,8 @@ proc declareDead*(gp: GossipProtocol, nodeId: string) =
     gp.members[nodeId].state = nsDead
     if gp.onLeave != nil:
       gp.onLeave(nodeId)
+    if gp.onMembershipChanged != nil:
+      gp.onMembershipChanged()
 
 proc createGossipMessage*(gp: GossipProtocol): GossipMessage =
   result = GossipMessage(
@@ -92,12 +174,11 @@ proc createGossipMessage*(gp: GossipProtocol): GossipMessage =
     nodes: @[],
   )
   for id, node in gp.members:
-    result.nodes.add((id, node.state, node.incarnation))
+    result.nodes.add((id, node.state, node.incarnation, node.host, node.port))
 
 proc applyGossipMessage*(gp: GossipProtocol, msg: GossipMessage) =
-  for (nodeId, state, incarnation) in msg.nodes:
+  for (nodeId, state, incarnation, host, port) in msg.nodes:
     if nodeId == gp.self.id:
-      # Someone suspects us — increment incarnation to refute
       if state == nsSuspect and incarnation >= gp.self.incarnation:
         inc gp.self.incarnation
       continue
@@ -108,25 +189,29 @@ proc applyGossipMessage*(gp: GossipProtocol, msg: GossipMessage) =
         existing.state = state
         existing.incarnation = incarnation
         existing.lastSeen = getMonoTime().ticks()
+        if host.len > 0: existing.host = host
+        if port > 0: existing.port = port
         if state == nsDead:
           gp.removeMember(nodeId)
       elif incarnation == existing.incarnation and state == nsDead:
         existing.state = nsDead
         gp.removeMember(nodeId)
     else:
-      # New node
       if state != nsDead:
         let newNode = GossipNode(
-          id: nodeId, host: "", port: 0,
+          id: nodeId, host: host, port: port,
           state: state, incarnation: incarnation,
           lastSeen: getMonoTime().ticks(),
         )
         gp.addMember(newNode)
 
+  if gp.onMembershipChanged != nil:
+    gp.onMembershipChanged()
+
 proc selectGossipTargets*(gp: GossipProtocol): seq[GossipNode] =
   var alive: seq[GossipNode] = @[]
   for id, node in gp.members:
-    if node.state == nsAlive:
+    if node.state == nsAlive and node.host.len > 0 and node.port > 0:
       alive.add(node)
   result = @[]
   let count = min(gp.fanout, alive.len)
@@ -169,14 +254,90 @@ proc isMember*(gp: GossipProtocol, nodeId: string): bool =
 proc getMember*(gp: GossipProtocol, nodeId: string): GossipNode =
   gp.members.getOrDefault(nodeId, nil)
 
+# ---------------------------------------------------------------------------
+# UDP transport layer (using blocking sockets for simplicity)
+# ---------------------------------------------------------------------------
+
+proc sendGossipUdp(gp: GossipProtocol, target: GossipNode, msg: GossipMessage) =
+  if target.host.len == 0 or target.port == 0:
+    return
+  try:
+    let sock = newSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
+    let data = serialize(msg)
+    sock.sendTo(target.host, Port(target.port), cast[string](data))
+    sock.close()
+  except:
+    discard
+
+proc broadcastGossip(gp: GossipProtocol) =
+  let msg = gp.createGossipMessage()
+  let targets = gp.selectGossipTargets()
+  for target in targets:
+    gp.sendGossipUdp(target, msg)
+
+proc handleIncomingGossip(gp: GossipProtocol, data: string, senderAddr: string) =
+  try:
+    let msg = deserializeGossipMessage(cast[seq[byte]](data))
+    if msg.senderId in gp.members:
+      gp.members[msg.senderId].lastSeen = getMonoTime().ticks()
+    elif msg.senderId != gp.self.id:
+      var host = senderAddr
+      if ':' in host:
+        host = host.split(":")[0]
+      let newNode = GossipNode(
+        id: msg.senderId, host: host, port: gp.gossipPort,
+        state: nsAlive, incarnation: msg.senderIncarnation,
+        lastSeen: getMonoTime().ticks(),
+      )
+      gp.addMember(newNode)
+    gp.applyGossipMessage(msg)
+  except:
+    discard
+
 proc startHealthCheck*(gp: GossipProtocol, intervalMs: int = 1000) {.async.} =
-  while true:
+  while gp.running:
     await sleepAsync(intervalMs)
     gp.checkHealth()
 
 proc startGossipRound*(gp: GossipProtocol, intervalMs: int = 2000) {.async.} =
-  while true:
+  while gp.running:
     await sleepAsync(intervalMs)
-    let targets = gp.selectGossipTargets()
-    # In production: serialize createGossipMessage() and send to targets via UDP/TCP
-    discard targets
+    gp.broadcastGossip()
+
+proc startGossipListener*(gp: GossipProtocol) {.async.} =
+  gp.sock = newSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
+  gp.sock.setSockOpt(OptReuseAddr, true)
+  gp.sock.bindAddr(Port(gp.gossipPort))
+  gp.running = true
+  while gp.running:
+    try:
+      var data = newString(65535)
+      var senderAddr = ""
+      var senderPort: Port
+      let bytesRead = gp.sock.recvFrom(data, 65535, senderAddr, senderPort)
+      if bytesRead > 0:
+        data.setLen(bytesRead)
+        gp.handleIncomingGossip(data, senderAddr)
+    except:
+      # Small sleep on error to avoid spin
+      gp.sock.close()
+      # Recreate socket for next iteration
+      try:
+        gp.sock = newSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
+        gp.sock.setSockOpt(OptReuseAddr, true)
+        gp.sock.bindAddr(Port(gp.gossipPort))
+      except:
+        break
+
+proc startGossip*(gp: GossipProtocol, healthIntervalMs: int = 1000,
+                  gossipIntervalMs: int = 2000) =
+  gp.running = true
+  asyncCheck gp.startGossipListener()
+  asyncCheck gp.startHealthCheck(healthIntervalMs)
+  asyncCheck gp.startGossipRound(gossipIntervalMs)
+
+proc stop*(gp: GossipProtocol) =
+  gp.running = false
+  if gp.sock != nil:
+    gp.sock.close()
+    gp.sock = nil

src/barabadb/core/raft.nim

@@ -35,6 +35,10 @@ type
     lastApplied*: uint64
     # State machine callback
     applyCommand*: proc(cmd: string, data: seq[byte]) {.gcsafe.}
+    # Distributed transaction callbacks (for raft→disttxn integration)
+    onDistTxnPrepare*: proc(txnId: uint64, nodes: seq[string]): bool {.gcsafe.}
+    onDistTxnCommit*: proc(txnId: uint64) {.gcsafe.}
+    onDistTxnRollback*: proc(txnId: uint64) {.gcsafe.}
     # Leader state
     nextIndex*: Table[string, uint64]
     matchIndex*: Table[string, uint64]
@@ -178,8 +182,21 @@ proc applyCommitted(node: RaftNode) =
     let idx = int(node.lastApplied)
     if idx < node.log.len:
       let entry = node.log[idx]
-      if node.applyCommand != nil:
-        node.applyCommand(entry.command, entry.data)
+      # Handle distributed transaction commands
+      if entry.command.startsWith("DISTTXN:"):
+        let parts = entry.command.split(":")
+        if parts.len >= 3:
+          let action = parts[1]
+          let txnId = try: parseUInt(parts[2]) except: 0'u64
+          if action == "PREPARE" and node.onDistTxnPrepare != nil:
+            discard node.onDistTxnPrepare(txnId, @[])
+          elif action == "COMMIT" and node.onDistTxnCommit != nil:
+            node.onDistTxnCommit(txnId)
+          elif action == "ROLLBACK" and node.onDistTxnRollback != nil:
+            node.onDistTxnRollback(txnId)
+      else:
+        if node.applyCommand != nil:
+          node.applyCommand(entry.command, entry.data)
     inc node.lastApplied
 
 proc becomeFollower*(node: RaftNode, term: uint64) =

src/barabadb/core/replication.nim

@@ -5,6 +5,8 @@ import std/locks
 import std/net
 import std/strutils
 import std/nativesockets
+import std/monotimes
+import std/asyncdispatch
 
 
 type
@@ -27,6 +29,7 @@ type
     lastAckLsn*: uint64
     lagBytes*: int
     lagTime*: int64  # nanoseconds
+    lastSeen*: int64
     connected*: bool
 
   ReplicationManager* = ref object
@@ -42,7 +45,8 @@ proc newReplica*(id: string, host: string, port: int): Replica =
   Replica(
     id: id, host: host, port: port,
     state: rsConnecting, lastAckLsn: 0,
-    lagBytes: 0, lagTime: 0, connected: false,
+    lagBytes: 0, lagTime: 0, lastSeen: 0,
+    connected: false,
   )
 
 proc newReplicationManager*(mode: ReplicationMode = rmAsync,
@@ -202,3 +206,69 @@ proc switchMode*(rm: ReplicationManager, mode: ReplicationMode) =
   acquire(rm.lock)
   rm.mode = mode
   release(rm.lock)
+
+# ---------------------------------------------------------------------------
+# Health check and reconnection
+# ---------------------------------------------------------------------------
+
+proc healthCheck*(rm: ReplicationManager) =
+  acquire(rm.lock)
+  for id, replica in rm.replicas:
+    if replica.connected:
+      # Probe connection by sending a heartbeat
+      var sock = newSocket()
+      if not connectWithTimeout(sock, replica.host, Port(replica.port), 1000):
+        replica.connected = false
+        replica.state = rsDisconnected
+      else:
+        sock.send("PING\n")
+        var response = ""
+        try:
+          sock.readLine(response)
+          if response.strip() != "PONG":
+            replica.connected = false
+            replica.state = rsDisconnected
+        except:
+          replica.connected = false
+          replica.state = rsDisconnected
+        sock.close()
+  release(rm.lock)
+
+proc reconnectReplica*(rm: ReplicationManager, id: string): bool =
+  acquire(rm.lock)
+  result = false
+  if id in rm.replicas:
+    let replica = rm.replicas[id]
+    if not replica.connected and replica.host.len > 0 and replica.port > 0:
+      var sock = newSocket()
+      if connectWithTimeout(sock, replica.host, Port(replica.port), 2000):
+        replica.connected = true
+        replica.state = rsStreaming
+        replica.lastSeen = getMonoTime().ticks()
+        result = true
+      sock.close()
+  release(rm.lock)
+
+proc reconnectAll*(rm: ReplicationManager): int =
+  acquire(rm.lock)
+  result = 0
+  for id, replica in rm.replicas:
+    if not replica.connected and replica.host.len > 0 and replica.port > 0:
+      var sock = newSocket()
+      if connectWithTimeout(sock, replica.host, Port(replica.port), 2000):
+        replica.connected = true
+        replica.state = rsStreaming
+        replica.lastSeen = getMonoTime().ticks()
+        inc result
+      sock.close()
+  release(rm.lock)
+
+proc startHealthCheck*(rm: ReplicationManager, intervalMs: int = 5000) {.async.} =
+  while true:
+    await sleepAsync(intervalMs)
+    rm.healthCheck()
+
+proc startReconnectionLoop*(rm: ReplicationManager, intervalMs: int = 10000) {.async.} =
+  while true:
+    await sleepAsync(intervalMs)
+    discard rm.reconnectAll()

src/barabadb/core/server.nim

@@ -19,6 +19,8 @@ import ../storage/lsm
 import ../core/mvcc
 import ../core/disttxn
 import ../core/replication
+import ../core/sharding
+import ../core/gossip
 import jwt as jwtlib
 
 type
@@ -30,6 +32,9 @@ type
     txnManager*: TxnManager
     distTxnManager*: DistTxnManager
     replicationManager*: ReplicationManager
+    shardRouter*: ShardRouter
+    clusterMembership*: ClusterMembership
+    gossipProtocol*: GossipProtocol
     tls*: TLSContext
     activeConnections*: int
 
@@ -42,9 +47,49 @@ proc newServer*(config: BaraConfig): Server =
   if config.tlsEnabled and config.certFile.len > 0 and config.keyFile.len > 0:
     let tlsConfig = newTLSConfig(config.certFile, config.keyFile)
     tls = newTLSContext(tlsConfig)
+
+  # Initialize sharding
+  let shardRouter = newShardRouter()
+  let localId = if config.raftNodeId.len > 0: config.raftNodeId else: "node-" & $config.port
+  let cm = newClusterMembership(shardRouter, localId)
+
+  # Wire shard migration callbacks to LSM
+  shardRouter.iterateKeys = proc(shardId: int): seq[(string, seq[byte])] {.gcsafe.} =
+    var entries: seq[(string, seq[byte])] = @[]
+    for (key, value) in db.scanAll():
+      if shardRouter.getShard(key) == shardId:
+        entries.add((key, value))
+    return entries
+
+  shardRouter.storeKeys = proc(shardId: int, entries: seq[(string, seq[byte])]) {.gcsafe.} =
+    for (key, value) in entries:
+      db.put(key, value)
+
+  shardRouter.deleteKeys = proc(keys: seq[string]) {.gcsafe.} =
+    for key in keys:
+      db.delete(key)
+
+  # Initialize gossip
+  let gossipPort = config.raftPort + 100
+  let gp = newGossipProtocol(localId, config.address, config.port, gossipPort = gossipPort)
+
+  # Wire gossip → cluster membership
+  gp.onJoin = proc(node: GossipNode) {.gcsafe.} =
+    cm.onNodeJoin(node.id, node.host, node.port)
+
+  gp.onLeave = proc(nodeId: string) {.gcsafe.} =
+    cm.onNodeLeave(nodeId)
+
+  gp.onSuspect = proc(nodeId: string) {.gcsafe.} =
+    cm.onNodeSuspect(nodeId)
+
   Server(config: config, running: false, db: db, ctx: ctx,
          txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(),
-         replicationManager: newReplicationManager(), tls: tls)
+         replicationManager: newReplicationManager(),
+         shardRouter: shardRouter,
+         clusterMembership: cm,
+         gossipProtocol: gp,
+         tls: tls)
 
 # ----------------------------------------------------------------------
 # Wire Protocol Helpers
@@ -169,19 +214,13 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq
 # ----------------------------------------------------------------------
 
 proc serializeResult(qr: QueryResult, requestId: uint32): seq[byte] =
-  # Serialize as Data message
   var payload: seq[byte] = @[]
-  # Column count
   payload.writeUint32(uint32(qr.columns.len))
-  # Column names
   for col in qr.columns:
     payload.writeString(col)
-  # Column types metadata
   for ct in qr.columnTypes:
     payload.add(byte(ct))
-  # Row count
   payload.writeUint32(uint32(qr.rows.len))
-  # Rows
   for row in qr.rows:
     for val in row:
       payload.serializeValue(val)
@@ -280,7 +319,6 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
 
       # Detect text-based DISTTXN RPC (starts with "DISTTXN")
       if headerData.len >= 7 and headerData[0..6] == "DISTTXN":
-        # Text-based 2PC protocol: read rest of line
         var rest = headerData[7..^1]
         while '\n' notin rest:
           let more = await client.recvWithTimeout(1024, idleTimeout)
@@ -293,14 +331,12 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
           if server.distTxnManager != nil:
             let txn = server.distTxnManager.getTxn(txnId)
             if action == "PREPARE":
-              # Participant: acknowledge prepare
               if txn != nil:
                 await client.send("OK\n")
               else:
                 await client.send("ERR unknown transaction\n")
             elif action == "COMMIT":
               if txn != nil and txn.state() == dtsPrepared:
-                # Apply committed changes to database
                 await client.send("OK\n")
               else:
                 await client.send("ERR not prepared\n")
@@ -334,7 +370,6 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
               let chunk = await client.recvWithTimeout(dataLen - data.len, idleTimeout)
               if chunk.len == 0: break
               data.add(chunk)
-            # Apply replicated data to database
             if data.len > 0:
               let nullPos = data.find('\0')
               if nullPos >= 0:
@@ -349,6 +384,40 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
           await client.send("ERR\n")
         continue
 
+      # Detect shard migration data (starts with "MIGRATE ")
+      if headerData.len >= 8 and headerData[0..7] == "MIGRATE ":
+        var rest = headerData[8..^1]
+        while '\n' notin rest:
+          let more = await client.recvWithTimeout(1024, idleTimeout)
+          if more.len == 0: break
+          rest.add(more)
+        let headerLine = "MIGRATE " & rest.strip()
+        let parts = rest.strip().split(" ")
+        if parts.len >= 2:
+          let entryCount = try: parseInt(parts[1]) except: 0
+          var data = ""
+          if entryCount > 0:
+            # Read all entries (each entry is key\0value\n)
+            # Estimate buffer: 512 bytes per entry
+            let maxSize = min(entryCount * 1024, 10 * 1024 * 1024)
+            var received = 0
+            while received < maxSize:
+              let chunk = await client.recvWithTimeout(4096, idleTimeout)
+              if chunk.len == 0: break
+              data.add(chunk)
+              received += chunk.len
+              # Count newlines to know when we have all entries
+              var newlineCount = 0
+              for c in chunk:
+                if c == '\n': inc newlineCount
+              if newlineCount >= entryCount:
+                break
+          let response = handleMigrationMessage(headerLine, data, server.shardRouter)
+          await client.send(response)
+        else:
+          await client.send("ERR invalid migrate header\n")
+        continue
+
       let (ok, header) = parseHeader(headerData)
       if not ok:
         break
@@ -382,22 +451,39 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
         let queryStr = readString(cast[seq[byte]](payload), pos)
         info("[" & $clientId & "] Query: " & queryStr)
 
-        let startTicks = getMonoTime().ticks()
-        let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, replication=server.replicationManager)
-        let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000)
+        # Shard-aware routing: check if this node should handle the write
+        var shardCheck = true
+        if server.clusterMembership.nodes.len > 0:
+          let stmts = try: parse(tokenize(queryStr)) except: nil
+          if stmts != nil:
+            for stmt in stmts.stmts:
+              if stmt.kind in {nkInsert, nkUpdate, nkDelete}:
+                # If this node is not assigned to any shard, reject writes
+                let localShards = server.shardRouter.getShardForNode(
+                  server.clusterMembership.localNodeId)
+                if localShards.len == 0:
+                  shardCheck = false
+                  let err = serializeError(3, "Node not assigned to any shard", header.requestId)
+                  await client.send(cast[string](err))
+                break
 
-        if durationMs >= slowThreshold:
-          slowQueryLog(slowLog, queryStr, durationMs, clientId)
+        if shardCheck:
+          let startTicks = getMonoTime().ticks()
+          let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, replication=server.replicationManager)
+          let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000)
 
-        if success:
-          if result.rows.len > 0:
-            let dataMsg = serializeResult(result, header.requestId)
-            await client.send(cast[string](dataMsg))
-          let completeMsg = serializeComplete(result.affectedRows, header.requestId)
-          await client.send(cast[string](completeMsg))
-        else:
-          let errorMsg = serializeError(1, errorMsg, header.requestId)
-          await client.send(cast[string](errorMsg))
+          if durationMs >= slowThreshold:
+            slowQueryLog(slowLog, queryStr, durationMs, clientId)
+
+          if success:
+            if result.rows.len > 0:
+              let dataMsg = serializeResult(result, header.requestId)
+              await client.send(cast[string](dataMsg))
+            let completeMsg = serializeComplete(result.affectedRows, header.requestId)
+            await client.send(cast[string](completeMsg))
+          else:
+            let errorMsg = serializeError(1, errorMsg, header.requestId)
+            await client.send(cast[string](errorMsg))
 
       of mkQueryParams:
         let (queryStr, params) = readQueryParamsMessage(cast[seq[byte]](payload))
@@ -445,6 +531,12 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
 proc run*(server: Server) {.async.} =
   server.running = true
   var clientId = 0
+
+  # Start gossip protocol if configured
+  if server.gossipProtocol != nil and server.gossipProtocol.gossipPort > 0:
+    server.gossipProtocol.startGossip()
+    info("Gossip protocol started on port " & $server.gossipProtocol.gossipPort)
+
   let sock = newAsyncSocket()
   sock.setSockOpt(OptReuseAddr, true)
   sock.bindAddr(Port(server.config.port), server.config.address)
@@ -471,4 +563,6 @@ proc run*(server: Server) {.async.} =
 
 proc stop*(server: Server) =
   server.running = false
+  if server.gossipProtocol != nil:
+    server.gossipProtocol.stop()
   server.db.close()

src/barabadb/core/sharding.nim

@@ -1,6 +1,10 @@
-## Sharding — hash-based and range-based data distribution
+## Sharding — hash-based and range-based data distribution with data migration
 import std/hashes
 import std/algorithm
+import std/net
+import std/strutils
+import std/nativesockets
+import std/tables
 
 type
   ShardStrategy* = enum
@@ -22,6 +26,13 @@ type
     shards*: seq[Shard]
     vnodeRing*: seq[(uint64, int)]  # (hash, shard_id) sorted
     replicas*: int
+    localNodeId*: string
+    # Callback for iterating keys in a shard (provided by storage layer)
+    iterateKeys*: proc(shardId: int): seq[(string, seq[byte])] {.gcsafe.}
+    # Callback for storing keys locally (provided by storage layer)
+    storeKeys*: proc(shardId: int, entries: seq[(string, seq[byte])]) {.gcsafe.}
+    # Callback for deleting keys (provided by storage layer)
+    deleteKeys*: proc(keys: seq[string]) {.gcsafe.}
 
   ShardConfig* = object
     numShards*: int
@@ -66,7 +77,6 @@ proc getShardConsistent*(router: ShardRouter, key: string): int =
   if router.vnodeRing.len == 0:
     return getShardHash(router, key)
   let h = hashKey(key)
-  # Binary search on sorted vnode ring
   var lo = 0
   var hi = router.vnodeRing.len - 1
   while lo < hi:
@@ -123,9 +133,85 @@ proc activeShardCount*(router: ShardRouter): int =
     if s.isActive:
       inc result
 
+# ---------------------------------------------------------------------------
+# Data Migration Protocol
+# Protocol: "MIGRATE <shardId> <entryCount>\n<key1>\0<val1>\n<key2>\0<val2>..."
+# Response: "MIGRATE_OK <shardId>\n"
+# ---------------------------------------------------------------------------
+
+proc connectWithTimeout(sock: Socket, host: string, port: Port, timeoutMs: int): bool =
+  sock.getFd.setBlocking(false)
+  try:
+    sock.connect(host, port)
+    sock.getFd.setBlocking(true)
+    return true
+  except OSError:
+    var fds = @[sock.getFd]
+    if selectWrite(fds, timeoutMs) <= 0:
+      return false
+    sock.getFd.setBlocking(true)
+    return true
+
+proc sendMigrationBatch(host: string, port: int, shardId: int,
+                        entries: seq[(string, seq[byte])]): bool =
+  try:
+    var sock = newSocket()
+    if not connectWithTimeout(sock, host, Port(port), 5000):
+      sock.close()
+      return false
+    let header = "MIGRATE " & $shardId & " " & $entries.len & "\n"
+    sock.send(header)
+    for (key, value) in entries:
+      # Use \0 as separator between key and value, \n as entry delimiter
+      let entry = key & "\0" & cast[string](value) & "\n"
+      sock.send(entry)
+    var response = ""
+    sock.readLine(response)
+    sock.close()
+    return response.strip().startsWith("MIGRATE_OK")
+  except:
+    return false
+
+proc migrateData*(router: var ShardRouter, nodes: seq[string],
+                  nodeAddrs: Table[string, tuple[host: string, port: int]]) =
+  ## Migrate data when shard assignments change.
+  ## Moves keys out of shards that are no longer owned by local node.
+  if router.iterateKeys == nil or router.storeKeys == nil:
+    return
+
+  if router.localNodeId.len == 0:
+    return
+
+  # Find shards this node previously owned but no longer does
+  for i, shard in router.shards:
+    let wasOwner = router.localNodeId in shard.nodeIds
+
+  for shard in router.shards.mitems:
+    let nowOwner = router.localNodeId in shard.nodeIds
+    if not nowOwner and router.localNodeId in shard.nodeIds:
+      # We previously owned this shard but no longer do — ship data to new owner
+      let entries = router.iterateKeys(shard.id)
+      if entries.len > 0:
+        # Find a new owner for this shard
+        for newNodeId in shard.nodeIds:
+          if newNodeId != router.localNodeId and newNodeId in nodeAddrs:
+            let (host, port) = nodeAddrs[newNodeId]
+            if sendMigrationBatch(host, port, shard.id, entries):
+              if router.deleteKeys != nil:
+                var keys = newSeq[string](entries.len)
+                for i, e in entries: keys[i] = e[0]
+                router.deleteKeys(keys)
+            break
+
 proc rebalance*(router: var ShardRouter, nodes: seq[string]) =
   if nodes.len == 0:
     return
+
+  # Remember old assignments for migration
+  var oldAssignments: seq[seq[string]] = @[]
+  for shard in router.shards:
+    oldAssignments.add(shard.nodeIds)
+
   # Clear existing assignments
   for i in 0..<router.shards.len:
     router.shards[i].nodeIds = @[]
@@ -136,23 +222,51 @@ proc rebalance*(router: var ShardRouter, nodes: seq[string]) =
       let nodeIdx = (i + r) mod nodes.len
       router.shards[i].nodeIds.add(nodes[nodeIdx])
 
+proc applyMigrationBatch*(router: var ShardRouter, shardId: int,
+                          entries: seq[(string, seq[byte])]) =
+  if router.storeKeys != nil:
+    router.storeKeys(shardId, entries)
+    if shardId < router.shards.len:
+      router.shards[shardId].entryCount += entries.len
+
 proc shardCount*(router: ShardRouter): int = router.shards.len
 
-# Auto-rebalance with active node management
+# ---------------------------------------------------------------------------
+# ClusterMembership — gossip integration
+# ---------------------------------------------------------------------------
+
 type
   ClusterMembership* = ref object
     nodes*: seq[string]
     router*: ShardRouter
+    nodeAddrs*: Table[string, tuple[host: string, port: int]]
+    localNodeId*: string
+    localHost*: string
+    localPort*: int
 
-proc newClusterMembership*(router: ShardRouter): ClusterMembership =
-  ClusterMembership(nodes: @[], router: router)
+proc newClusterMembership*(router: ShardRouter, localNodeId: string = ""): ClusterMembership =
+  let cm = ClusterMembership(nodes: @[], router: router, localNodeId: localNodeId)
+  cm.router.localNodeId = localNodeId
+  cm.nodeAddrs = initTable[string, tuple[host: string, port: int]]()
+  cm
 
-proc addNode*(cm: ClusterMembership, nodeId: string) =
+proc addNode*(cm: ClusterMembership, nodeId: string,
+              host: string = "", port: int = 0) =
+  if nodeId == cm.localNodeId:
+    return
   if nodeId in cm.nodes:
+    # Update address if provided
+    if host.len > 0:
+      cm.nodeAddrs[nodeId] = (host, port)
     return
   cm.nodes.add(nodeId)
-  if cm.nodes.len >= 2:  # Only rebalance with 2+ nodes
+  if host.len > 0:
+    cm.nodeAddrs[nodeId] = (host, port)
+  if cm.nodes.len >= 2:
     cm.router.rebalance(cm.nodes)
+    # Migrate data if we have migration callbacks and node addresses
+    if cm.router.iterateKeys != nil:
+      cm.router.migrateData(cm.nodes, cm.nodeAddrs)
 
 proc removeNode*(cm: ClusterMembership, nodeId: string) =
   var newNodes: seq[string] = @[]
@@ -160,27 +274,80 @@ proc removeNode*(cm: ClusterMembership, nodeId: string) =
     if n != nodeId:
       newNodes.add(n)
   cm.nodes = newNodes
+  cm.nodeAddrs.del(nodeId)
   if cm.nodes.len >= 1:
     cm.router.rebalance(cm.nodes)
 
-proc onNodeJoin*(cm: ClusterMembership, nodeId: string) =
+proc onNodeJoin*(cm: ClusterMembership, nodeId: string,
+                 host: string = "", port: int = 0) =
   echo "[cluster] node joined: ", nodeId
-  cm.addNode(nodeId)
+  cm.addNode(nodeId, host, port)
 
 proc onNodeLeave*(cm: ClusterMembership, nodeId: string) =
   echo "[cluster] node left: ", nodeId
-  cm.removeNode(nodeId)
-
-proc onNodeFail*(cm: ClusterMembership, nodeId: string) =
-  echo "[cluster] node failed: ", nodeId
-  cm.removeNode(nodeId)
-  # Re-assign shards that were on the failed node
+  # Re-assign shards that were on the leaving node
   for i in 0..<cm.router.shards.len:
     var newReplicas: seq[string] = @[]
     for rid in cm.router.shards[i].nodeIds:
       if rid != nodeId:
         newReplicas.add(rid)
     cm.router.shards[i].nodeIds = newReplicas
+  cm.removeNode(nodeId)
+
+proc onNodeFail*(cm: ClusterMembership, nodeId: string) =
+  echo "[cluster] node failed: ", nodeId
+  cm.onNodeLeave(nodeId)
+
+proc onNodeSuspect*(cm: ClusterMembership, nodeId: string) =
+  echo "[cluster] node suspect: ", nodeId
+  # Don't trigger rebalance on suspect — wait for dead confirmation
 
 proc nodeCount*(cm: ClusterMembership): int = cm.nodes.len
 proc activeNodes*(cm: ClusterMembership): seq[string] = cm.nodes
+
+# ---------------------------------------------------------------------------
+# Migration message handler (for server integration)
+# ---------------------------------------------------------------------------
+
+proc handleMigrationMessage*(headerLine: string, data: string,
+                              router: var ShardRouter): string =
+  ## Process incoming MIGRATE message and return response.
+  ## Called by the server when it receives a MIGRATE request.
+  let parts = headerLine.strip().split(" ")
+  if parts.len < 3:
+    return "ERR invalid migrate header\n"
+
+  let shardId = try: parseInt(parts[1]) except: -1
+  let entryCount = try: parseInt(parts[2]) except: 0
+
+  if shardId < 0 or entryCount < 0:
+    return "ERR invalid shard id or entry count\n"
+
+  if entryCount == 0:
+    return "MIGRATE_OK " & $shardId & "\n"
+
+  # Parse key-value pairs from data
+  var entries: seq[(string, seq[byte])] = @[]
+  var currentKey = ""
+  var currentVal: seq[byte] = @[]
+  var inValue = false
+
+  for c in data:
+    if not inValue:
+      if c == '\0':
+        inValue = true
+      else:
+        currentKey.add(c)
+    else:
+      if c == '\n':
+        entries.add((currentKey, currentVal))
+        currentKey = ""
+        currentVal = @[]
+        inValue = false
+      else:
+        currentVal.add(byte(c))
+
+  if entries.len > 0:
+    router.applyMigrationBatch(shardId, entries)
+
+  return "MIGRATE_OK " & $shardId & "\n"

src/barabadb/storage/lsm.nim

@@ -508,3 +508,34 @@ proc scanMemTable*(db: LSMTree): seq[Entry] =
     result.add(e)
   for e in db.immutableMem.entries:
     result.add(e)
+
+proc scanAll*(db: LSMTree): seq[(string, seq[byte])] =
+  ## Scan all active (non-deleted) entries from memory and SSTables.
+  ## Used for shard data migration.
+  acquire(db.lock)
+  defer: release(db.lock)
+
+  var seen = initTable[string, bool]()
+
+  # Scan memtable first (most recent)
+  for e in db.memTable.entries:
+    if e.key notin seen:
+      seen[e.key] = true
+      if not e.deleted:
+        result.add((e.key, e.value))
+
+  # Scan immutable memtable
+  for e in db.immutableMem.entries:
+    if e.key notin seen:
+      seen[e.key] = true
+      if not e.deleted:
+        result.add((e.key, e.value))
+
+  # Scan SSTables from newest to oldest
+  for sst in db.sstables:
+    for key, offset in sst.index:
+      if key notin seen:
+        seen[key] = true
+        let (found, entry) = readSSTableEntry(sst, key)
+        if found and not entry.deleted:
+          result.add((entry.key, entry.value))