From ad90ebcd5eb3817ead9110d194530410a1041f03 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Fri, 31 Jul 2026 03:53:48 +0300 Subject: [PATCH] test(raft): cold-node rejoin and wiped-node join e2e Includes two raft fixes the e2e exposed: - heartbeatLoop drops cached peer sockets on leadership acquisition (half-dead sockets to a restarted peer never errored, so no redial ever happened and the cluster livelocked without heartbeats) - restoreSnapshot repoints tcpServer ctx/db at the reopened database (queries kept reading the closed pre-restore LSM, serving 0 rows) --- .github/workflows/ci.yml | 1 + .gitignore | 1 + baradadb.nimble | 1 + src/barabadb/core/raft.nim | 15 + src/baradadb.nim | 16 +- tests/raft_coldnode_e2e_test.nim | 480 +++++++++++++++++++++++++++++++ 6 files changed, 513 insertions(+), 1 deletion(-) create mode 100644 tests/raft_coldnode_e2e_test.nim diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f5fd35e..14aa15b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -83,6 +83,7 @@ jobs: nim c -d:ssl --threads:on --path:src -r tests/raft_writes_e2e_test.nim nim c -d:ssl --threads:on --path:src -r tests/raft_failover_load_e2e_test.nim nim c -d:ssl --threads:on --path:src -r tests/raft_tls_e2e_test.nim + nim c -d:ssl --threads:on --path:src -r tests/raft_coldnode_e2e_test.nim verify: runs-on: ubuntu-latest diff --git a/.gitignore b/.gitignore index 4eca121..6dac124 100644 --- a/.gitignore +++ b/.gitignore @@ -18,6 +18,7 @@ tests/raft_e2e_test tests/raft_writes_e2e_test tests/raft_failover_load_e2e_test tests/raft_tls_e2e_test +tests/raft_coldnode_e2e_test benchmarks/bench_all benchmarks/compare clients/nim/tests/test_client diff --git a/baradadb.nimble b/baradadb.nimble index 9cb6d27..a4c3713 100644 --- a/baradadb.nimble +++ b/baradadb.nimble @@ -31,6 +31,7 @@ task test, "Run all tests": "test_schema_persist", "test_storage_hardening", "tla_faithfulness", "nimforum_smoke_test", "raft_e2e_test", "raft_writes_e2e_test", "raft_failover_load_e2e_test", "raft_tls_e2e_test", + "raft_coldnode_e2e_test", "fuzz_test", "prop_test", "test_wire_insert_stress", "stress_test"]: exec "nim c -r tests/" & t & ".nim" diff --git a/src/barabadb/core/raft.nim b/src/barabadb/core/raft.nim index 755f3d7..bf00b3c 100644 --- a/src/barabadb/core/raft.nim +++ b/src/barabadb/core/raft.nim @@ -1131,8 +1131,21 @@ proc receiveLoop(net: RaftNetwork, client: AsyncSocket) {.async.} = proc heartbeatLoop(net: RaftNetwork) {.async.} = ## Fan out heartbeats in parallel so a slow/dead peer cannot delay ## AppendEntries to the rest of the cluster. + var wasLeader = false while net.running: if net.node.state == rsLeader: + if not wasLeader: + # Fresh term, fresh connections. A peer that restarted while we were + # partitioned leaves a half-dead cached socket whose writes can keep + # "succeeding" into the void (the TCP error surfaces late or never, + # so the error-triggered redial in send() may never fire) — the + # restarted peer then never sees AppendEntries, keeps candidating, + # and the cluster livelocks. Drop all cached peer connections on + # leadership acquisition so the heartbeat fan-out redials. + for peerId, sock in net.peerSockets: + try: sock.close() except CatchableError: discard + net.peerSockets.clear() + wasLeader = true var futs: seq[Future[void]] = @[] for peer in net.node.peers: let msg = net.node.appendEntries(peer) @@ -1142,6 +1155,8 @@ proc heartbeatLoop(net: RaftNetwork) {.async.} = await f except CatchableError: discard + else: + wasLeader = false await sleepAsync(net.node.heartbeatTimeout) proc timerLoop*(net: RaftNetwork) {.async.} diff --git a/src/baradadb.nim b/src/baradadb.nim index 209ab3f..80f4b2d 100644 --- a/src/baradadb.nim +++ b/src/baradadb.nim @@ -410,7 +410,21 @@ proc main() = # extraction failure it rolls back automatically. Reopen whatever is # on disk either way so the node is not left with a closed DB. let restored = restoreDataDir(archivePath, defaultDbDir) - result = registry.reopenDatabase("default") and restored + let reopened = registry.reopenDatabase("default") + if reopened: + # Serve the (re)opened data. Client connections clone tcpServer.ctx + # on accept (cloneForConnection), and reopenDatabase installs a NEW + # ctx object in the registry slot — without repointing, queries + # keep reading the closed pre-restore LSM (empty results, no + # error). The websocket change hook was installed on the previous + # ctx object; carry it over. + let oldCtx = tcpServer.ctx + let newCtx = cast[ExecutionContext](cast[pointer](defaultDbInfo.ctx)) + newCtx.onChange = oldCtx.onChange + tcpServer.db = defaultDbInfo.db + tcpServer.ctx = newCtx + tcpServer.txnManager = newCtx.txnManager + result = reopened and restored except CatchableError as e: echo "[raft] Snapshot restore failed: ", e.msg result = false diff --git a/tests/raft_coldnode_e2e_test.nim b/tests/raft_coldnode_e2e_test.nim new file mode 100644 index 0000000..ec40daa --- /dev/null +++ b/tests/raft_coldnode_e2e_test.nim @@ -0,0 +1,480 @@ +## Raft cold-node E2E — real 3-node cluster; end-to-end proof of the +## InstallSnapshot work (compaction, snapshot build/send, follower restore). +## Starts three actual build/baradadb processes with a tiny raft log +## (BARADB_RAFT_LOG_MAX_ENTRIES=16) and a short stale window +## (BARADB_RAFT_PEER_STALE_MS=3000) so the leader compacts quickly past a +## downed peer's matchIndex. +## +## Scenario A: a node is killed, 100 rows are written through the leader +## (forcing compaction past its matchIndex), the node restarts with its +## intact data dir and must catch up via InstallSnapshot within 15 s. +## Scenario B: the node is stopped, its data dir is WIPED, it rejoins with +## the same node id and must serve the full row set within 20 s. +## +## Process-management conventions follow tests/raft_writes_e2e_test.nim +## (copying is the repo's e2e convention — do not factor out). +import std/unittest +import std/osproc +import std/os +import std/strtabs +import std/strutils +import std/times +import std/net +import std/posix +import std/httpclient + +import ../adaptors/nim/baradb_sqlite as sqlite + +const + BinaryPath = "./build/baradadb" + LeaderMarker = "became leader" + SnapshotMarker = "Installing snapshot" + TableName = "cold_test" + RowCount = 100 + +type + NodeProc = object + id: string + clientPort: int + raftPort: int + peers: string + clientPeers: string + p: Process + dataDir: string + output: string + alive: bool + +proc drainOutput(n: var NodeProc) = + ## Reads whatever the child has written so far. The pipe was set O_NONBLOCK + ## at start, so this never blocks — a hung read is impossible here. + var tmp: array[8192, char] + while true: + let count = posix.read(n.p.outputHandle.cint, tmp[0].addr, tmp.len) + if count <= 0: break + for i in 0 ..< count: n.output.add tmp[i] + +proc drainAll(nodes: var seq[NodeProc]) = + for n in nodes.mitems: + if n.p != nil: n.drainOutput() + +proc dumpAll(nodes: var seq[NodeProc]) = + ## Debuggability: on failure, everything the nodes said. + nodes.drainAll() + for n in nodes: + echo "===== output of ", n.id, " (port ", n.clientPort, ") =====" + echo n.output + +proc portOpen(port: int): bool = + var s: Socket + try: + s = newSocket() + s.connect("127.0.0.1", Port(port), timeout = 250) + s.close() + result = true + except CatchableError: + if s != nil: s.close() + result = false + +proc killNode(n: var NodeProc) = + if n.p != nil and n.alive: + try: + n.p.terminate() + discard n.p.waitForExit() + except CatchableError: + discard + n.alive = false + +proc leaderTerms(output: string): seq[int] = + ## All terms this node logged leadership for ("became leader for term T"). + var pos = 0 + while true: + let idx = output.find(LeaderMarker, pos) + if idx < 0: break + let tIdx = output.find("term ", idx) + if tIdx < 0: break + let numStart = tIdx + 5 + var numEnd = numStart + while numEnd < output.len and output[numEnd] in Digits: inc numEnd + if numEnd > numStart: + result.add(parseInt(output[numStart ..< numEnd])) + pos = numEnd + +proc maxLeader(nodes: seq[NodeProc]): tuple[idx, term: int] = + ## Node that logged leadership for the highest term seen so far. + result = (-1, 0) + for i in 0 ..< nodes.len: + for t in leaderTerms(nodes[i].output): + if t > result.term: result = (i, t) + +proc drainFor(nodes: var seq[NodeProc], ms: int) = + let start = getTime() + while getTime() - start < initDuration(milliseconds = ms): + nodes.drainAll() + sleep(50) + +proc openClient(port: int): DbConn = + ## Connect with retries — the port may accept TCP before the DB is usable. + for i in 0 ..< 50: + try: + return open("127.0.0.1:" & $port, "", "", "default") + except CatchableError: + sleep(100) + raise newException(IOError, "cannot connect to port " & $port) + +proc rowCountOn(port: int): int = + ## SELECT COUNT(*) — raises while the table is not there yet (e.g. before + ## the snapshot restore has landed); callers poll and tolerate that. + let db = openClient(port) + defer: db.close() + parseInt(db.getValue(sql("SELECT COUNT(*) FROM " & TableName))) + +proc fetchMetric(port: int, name: string): int = + ## GET /metrics on the node's HTTP port (clientPort + 440) and parse + ## `name{...} `. Returns -1 on any error or when absent. + let client = newHttpClient(timeout = 1500) + defer: client.close() + try: + let body = client.getContent("http://127.0.0.1:" & $(port + 440) & "/metrics") + for line in body.splitLines(): + if line.startsWith(name & "{"): + let parts = line.splitWhitespace() + if parts.len >= 2: + return parseInt(parts[^1]) + except CatchableError: + discard + return -1 + +proc startNode(id, dataDir: string, clientPort, raftPort: int, + peers, clientPeers: string): NodeProc = + ## Boots one build/baradadb process. Takes the data dir explicitly so a + ## node can be restarted with the same dir (scenario A) or a wiped dir + ## (scenario B). Compaction/stale-window env goes on EVERY node. + createDir(dataDir) + var env = newStringTable() + for key, val in envPairs(): + env[key] = val + env["BARADB_PORT"] = $clientPort + env["BARADB_RAFT_ENABLED"] = "true" + env["BARADB_RAFT_PORT"] = $raftPort + env["BARADB_RAFT_NODE_ID"] = id + env["BARADB_RAFT_PEERS"] = peers + env["BARADB_RAFT_CLIENT_PEERS"] = clientPeers + env["BARADB_RAFT_LOG_MAX_ENTRIES"] = "16" + env["BARADB_RAFT_PEER_STALE_MS"] = "3000" + env["BARADB_DATA_DIR"] = dataDir + env["BARADB_LOG_LEVEL"] = "info" + let p = startProcess(BinaryPath, env = env, + options = {poStdErrToStdOut, poDaemon}) + discard fcntl(p.outputHandle.cint, F_SETFL, + fcntl(p.outputHandle.cint, F_GETFL) or O_NONBLOCK) + NodeProc(id: id, clientPort: clientPort, raftPort: raftPort, + peers: peers, clientPeers: clientPeers, p: p, + dataDir: dataDir, alive: true) + +proc restartNode(n: var NodeProc, wipe: bool) = + ## Kill, then boot the same node id again. wipe=false keeps the data dir + ## (cold-node return); wipe=true deletes it first (fresh node rejoin). + n.killNode() + if n.p != nil: n.p.close() + if wipe: + removeDir(n.dataDir) + let fresh = startNode(n.id, n.dataDir, n.clientPort, n.raftPort, + n.peers, n.clientPeers) + n.p = fresh.p + n.alive = true + +proc waitReady(n: NodeProc, deadlineSec: int): bool = + let start = getTime() + while getTime() - start < initDuration(seconds = deadlineSec): + if portOpen(n.clientPort): + return true + sleep(100) + return false + +proc runColdNodeScenario() = + ## Fatal phase failures dump all captured node output, record a test + ## failure, and return; cleanup happens in the finally below either way. + let tstamp = getTime().toUnix.int + # Port base per the T11 brief: distinct from raft_writes_e2e_test + # (46000+mod4000) and raft_tls_e2e_test (54000+mod4000). Client ports are + # spaced by 10 because the server derives HTTP (port+440), WS (port+441) + # and gossip (raftPort+100) ports — consecutive client ports collide. + let cbase = 58000 + (tstamp mod 4000) + let rbase = cbase + 100 + let peers = "n1@127.0.0.1:" & $(rbase + 1) & + ",n2@127.0.0.1:" & $(rbase + 2) & + ",n3@127.0.0.1:" & $(rbase + 3) + # SQL client ports for transparent leader write forwarding. + let clientPeers = "n1@127.0.0.1:" & $(cbase + 10) & + ",n2@127.0.0.1:" & $(cbase + 20) & + ",n3@127.0.0.1:" & $(cbase + 30) + + var nodes: seq[NodeProc] + + try: + # Node starts live inside the try so a raise from start #2/#3 still + # reaches the cleanup in the finally below. + for i in 1 .. 3: + let id = "n" & $i + let dataDir = getTempDir() / "baradb_raft_coldnode_e2e_" & $tstamp & "_" & id + nodes.add startNode(id, dataDir, cbase + i * 10, rbase + i, + peers, clientPeers) + + # Readiness: all three client ports accept TCP connections (10s each). + for i in 0 ..< nodes.len: + if not waitReady(nodes[i], 10): + echo "node ", nodes[i].id, " never became ready" + dumpAll(nodes) + fail() + return + + # Election: timeouts are 150-300ms, heartbeat 50ms — a leader should + # emerge within ~2s; 10s deadline for margin. + var elected = false + let electStart = getTime() + while getTime() - electStart < initDuration(seconds = 10): + nodes.drainAll() + if maxLeader(nodes).idx >= 0: + elected = true + break + sleep(50) + if not elected: + echo "no leader elected within 10s" + dumpAll(nodes) + fail() + return + + # Settle and require stability, same as raft_writes_e2e_test. + nodes.drainFor(2000) + let (leaderIdx, leaderTerm) = maxLeader(nodes) + nodes.drainFor(1000) + let (stableIdx, stableTerm) = maxLeader(nodes) + if stableIdx != leaderIdx or stableTerm != leaderTerm: + echo "cluster unstable: leadership moved from ", nodes[leaderIdx].id, + " (term ", leaderTerm, ") to ", nodes[stableIdx].id, + " (term ", stableTerm, ")" + dumpAll(nodes) + fail() + return + echo "leader elected: ", nodes[leaderIdx].id, " (term ", leaderTerm, ")" + + # The cold node: n3 unless n3 is the leader (then n1) — it must be a + # follower so killing it never forces a re-election. + let coldIdx = (if leaderIdx == 2: 0 else: 2) + let otherIdx = 3 - leaderIdx - coldIdx + echo "cold node: ", nodes[coldIdx].id, "; surviving follower: ", + nodes[otherIdx].id + + # Schema: CREATE TABLE goes through the raft "ddl" log (C3c). Create on + # the leader; wait until the cold node has applied it before killing it. + block: + let db = openClient(nodes[leaderIdx].clientPort) + try: + db.exec(sql("CREATE TABLE " & TableName & " (id INT PRIMARY KEY, name STRING)")) + except CatchableError as e: + echo "leader CREATE TABLE failed: ", e.msg + dumpAll(nodes) + fail() + return + db.close() + block: + let start = getTime() + var ready = false + while getTime() - start < initDuration(seconds = 5): + try: + discard rowCountOn(nodes[coldIdx].clientPort) + ready = true + break + except CatchableError: + sleep(100) + if not ready: + echo "cold node ", nodes[coldIdx].id, + " never applied CREATE TABLE within 5s" + dumpAll(nodes) + fail() + return + echo "schema replicated to cold node ", nodes[coldIdx].id + + # ---- Kill the cold node; write 100 rows through the leader. ---- + # With logMaxEntries=16 and peerStaleMs=3000 the leader compacts past the + # dead node's matchIndex partway through the writes, so the node can only + # catch up via InstallSnapshot on return. + killNode(nodes[coldIdx]) + echo "cold node ", nodes[coldIdx].id, " killed; writing ", RowCount, " rows" + block: + let db = openClient(nodes[leaderIdx].clientPort) + var atRow = 0 + try: + for i in 1 .. RowCount: + atRow = i + db.exec(sql("INSERT INTO " & TableName & " (id, name) VALUES (" & + $i & ", 'row-" & $i & "')")) + except CatchableError as e: + echo "leader INSERT failed at row ", atRow, ": ", e.msg + dumpAll(nodes) + fail() + return + db.close() + echo RowCount, " rows committed with ", nodes[coldIdx].id, " down" + + # Compaction evidence on the leader: the in-memory log stayed bounded + # (<= 64, far below the 100 entries written) and the snapshot base moved. + nodes.drainAll() + let leaderLogLen = fetchMetric(nodes[leaderIdx].clientPort, + "baradb_raft_log_entries") + let leaderSnapIdx = fetchMetric(nodes[leaderIdx].clientPort, + "baradb_raft_snapshot_index") + echo "leader after writes: log_entries=", leaderLogLen, + " snapshot_index=", leaderSnapIdx + if leaderLogLen < 0 or leaderLogLen > 64: + echo "leader raft log not bounded: baradb_raft_log_entries=", + leaderLogLen, " (want <= 64)" + dumpAll(nodes) + fail() + return + if leaderSnapIdx <= 0: + echo "leader never compacted: baradb_raft_snapshot_index=", + leaderSnapIdx, " (want > 0)" + dumpAll(nodes) + fail() + return + echo "leader log stayed bounded and compaction advanced" + + # Sanity: leader and surviving follower agree on the row count. + let wantCount = rowCountOn(nodes[leaderIdx].clientPort) + if wantCount != RowCount: + echo "leader count=", wantCount, " want ", RowCount + dumpAll(nodes) + fail() + return + + # ---- Scenario A: cold node returns with its intact data dir. ---- + nodes[coldIdx].output.setLen(0) # fresh log for the snapshot marker scan + restartNode(nodes[coldIdx], wipe = false) + if not waitReady(nodes[coldIdx], 10): + echo "cold node ", nodes[coldIdx].id, " never became ready after restart" + dumpAll(nodes) + fail() + return + echo "scenario A: ", nodes[coldIdx].id, " restarted with intact data dir" + + # Within 15s: snapshot evidence (metric on the cold node, or its restore + # log line) AND data convergence with the leader. + block: + let start = getTime() + var snapSeen = false + var converged = false + while getTime() - start < initDuration(seconds = 15): + nodes.drainAll() + if not snapSeen: + snapSeen = fetchMetric(nodes[coldIdx].clientPort, + "baradb_raft_snapshot_index") > 0 or + nodes[coldIdx].output.contains(SnapshotMarker) + if not converged: + try: + converged = rowCountOn(nodes[coldIdx].clientPort) == wantCount + except CatchableError: + discard + if snapSeen and converged: break + sleep(200) + if not snapSeen: + echo "scenario A: no snapshot evidence on ", nodes[coldIdx].id, + " (baradb_raft_snapshot_index stayed 0, no restore log line)" + dumpAll(nodes) + fail() + return + if not converged: + echo "scenario A: ", nodes[coldIdx].id, + " count did not converge to ", wantCount, " within 15s" + dumpAll(nodes) + fail() + return + echo "scenario A: snapshot installed, count converged to ", wantCount + + # Spot-check a few ids survived the snapshot round-trip. + block: + let db = openClient(nodes[coldIdx].clientPort) + defer: db.close() + for i in [1, 42, RowCount]: + let v = db.getValue(sql("SELECT name FROM " & TableName & + " WHERE id = " & $i)) + if v != "row-" & $i: + echo "scenario A: spot-check id=", i, " got '", v, + "' want 'row-", i, "'" + dumpAll(nodes) + fail() + return + echo "scenario A: spot-checks passed" + + # ---- Scenario B: wiped node rejoins with the same node id. ---- + nodes[coldIdx].output.setLen(0) + restartNode(nodes[coldIdx], wipe = true) + if not waitReady(nodes[coldIdx], 10): + echo "wiped node ", nodes[coldIdx].id, " never became ready after restart" + dumpAll(nodes) + fail() + return + echo "scenario B: ", nodes[coldIdx].id, " restarted with a wiped data dir" + + # Within 20s: snapshot → catch-up, then the full row set. + block: + let start = getTime() + var snapSeen = false + var converged = false + while getTime() - start < initDuration(seconds = 20): + nodes.drainAll() + if not snapSeen: + snapSeen = fetchMetric(nodes[coldIdx].clientPort, + "baradb_raft_snapshot_index") > 0 or + nodes[coldIdx].output.contains(SnapshotMarker) + if not converged: + try: + converged = rowCountOn(nodes[coldIdx].clientPort) == wantCount + except CatchableError: + discard + if snapSeen and converged: break + sleep(200) + if not snapSeen: + echo "scenario B: no snapshot evidence on wiped ", nodes[coldIdx].id + dumpAll(nodes) + fail() + return + if not converged: + echo "scenario B: wiped ", nodes[coldIdx].id, + " count did not converge to ", wantCount, " within 20s" + dumpAll(nodes) + fail() + return + echo "scenario B: wiped node converged to ", wantCount, " rows" + + # Spot-check again on the wiped node. + block: + let db = openClient(nodes[coldIdx].clientPort) + defer: db.close() + for i in [1, 42, RowCount]: + let v = db.getValue(sql("SELECT name FROM " & TableName & + " WHERE id = " & $i)) + if v != "row-" & $i: + echo "scenario B: spot-check id=", i, " got '", v, + "' want 'row-", i, "'" + dumpAll(nodes) + fail() + return + echo "scenario B: spot-checks passed" + finally: + for n in nodes.mitems: + n.killNode() + if n.p != nil: n.p.close() + removeDir(n.dataDir) + +suite "Raft cold-node E2E": + test "compacted-away node and wiped node rejoin and converge": + if not fileExists(BinaryPath): + if getEnv("CI").len > 0: + echo "[FAIL] ", BinaryPath, " missing under CI — build step broken?" + fail() + else: + echo "[SKIP] ", BinaryPath, " missing — run `nimble test` (builds the server first)" + skip() + else: + runColdNodeScenario()