From 63cb05afe298a55de3fa790f02f168eb4cf483bb Mon Sep 17 00:00:00 2001 From: dimgigov Date: Fri, 31 Jul 2026 00:35:55 +0300 Subject: [PATCH] test(raft): failover under sustained write load e2e --- .gitignore | 1 + baradadb.nimble | 1 + tests/raft_failover_load_e2e_test.nim | 441 ++++++++++++++++++++++++++ 3 files changed, 443 insertions(+) create mode 100644 tests/raft_failover_load_e2e_test.nim diff --git a/.gitignore b/.gitignore index 232eefc..94e3ac1 100644 --- a/.gitignore +++ b/.gitignore @@ -16,6 +16,7 @@ tests/bugfix_test tests/nimforum_smoke_test tests/raft_e2e_test tests/raft_writes_e2e_test +tests/raft_failover_load_e2e_test benchmarks/bench_all benchmarks/compare clients/nim/tests/test_client diff --git a/baradadb.nimble b/baradadb.nimble index 300557a..644867f 100644 --- a/baradadb.nimble +++ b/baradadb.nimble @@ -30,6 +30,7 @@ task test, "Run all tests": for t in ["test_minimal", "test_all", "bugfix_test", "join_tests", "test_lock", "test_schema_persist", "test_storage_hardening", "tla_faithfulness", "nimforum_smoke_test", "raft_e2e_test", "raft_writes_e2e_test", + "raft_failover_load_e2e_test", "fuzz_test", "prop_test", "test_wire_insert_stress", "stress_test"]: exec "nim c -r tests/" & t & ".nim" diff --git a/tests/raft_failover_load_e2e_test.nim b/tests/raft_failover_load_e2e_test.nim new file mode 100644 index 0000000..fc19299 --- /dev/null +++ b/tests/raft_failover_load_e2e_test.nim @@ -0,0 +1,441 @@ +## Raft failover-under-load E2E — real 3-node cluster over the TCP transport. +## A writer thread hammers the current leader with INSERTs; once 50 writes +## have been acknowledged the leader is killed mid-load. Asserts: +## A (availability): a survivor accepts an INSERT within 10s of the kill. +## B (durability): every acknowledged id is present on BOTH survivors +## once the new leader is stable and the remaining +## follower has caught up. +## Process-management conventions follow tests/raft_writes_e2e_test.nim; +## client access follows tests/nimforum_smoke_test.nim. +## +## NOTE: load_test deliberately has a non-PK column. A table whose only +## column is the PK stores an EMPTY LSM value per row (execInsert drops PK +## columns from the value), and the raft write path (appendWriteToRaft in +## src/barabadb/core/server.nim) encodes empty values as "delete" entries — +## so on commit every node applies a delete over the just-inserted row and +## it vanishes everywhere. That is a v1.2.0 product bug in the raft entry +## encoding (put/delete must not be inferred from value emptiness); until it +## is fixed in src/, this test exercises the two-column row shape that the +## current encoding handles correctly. +import std/unittest +import std/osproc +import std/os +import std/strtabs +import std/strutils +import std/sequtils +import std/times +import std/net +import std/posix +import std/sets +import std/locks +import std/typedthreads + +import ../adaptors/nim/baradb_sqlite as sqlite + +const + BinaryPath = "./build/baradadb" + LeaderMarker = "became leader" + AckTarget = 50 # kill the leader once this many writes are acked + ProbeId = 1000001 # availability-probe row id (clear of writer's 1..N) + +type + NodeProc = object + id: string + clientPort: int + p: Process + dataDir: string + output: string + alive: bool + +# Shared writer-thread state, passed by pointer into the thread (same +# convention as tests/test_storage_hardening.nim); all access goes through +# the lock. +type + WriterArgs = object + lock: ptr Lock + acked: ptr seq[int] # ids whose INSERT was acknowledged by the cluster + stop: ptr bool # main thread sets this to end the writer loop + ports: array[3, int] # candidate client ports (leader first, then survivors) + +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 countRows(port: int): int = + ## Row count of load_test on `port`, or -1 on any error (not ready yet). + result = -1 + try: + let db = openClient(port) + try: + result = parseInt(db.getValue(sql"SELECT count(*) FROM load_test")) + finally: + db.close() + except CatchableError: + discard + +proc allIds(port: int): HashSet[int] = + ## Every id in load_test on `port`. Raises on error. + let db = openClient(port) + try: + for row in db.getAllRows(sql"SELECT id FROM load_test"): + if row.len >= 1 and row[0].len > 0: + result.incl(parseInt(row[0])) + finally: + db.close() + +proc writerLoop(args: WriterArgs) {.thread.} = + ## INSERTs n = 1, 2, ... against the cluster. Every acknowledged n is + ## appended to args.acked. On any error the client is dropped and reopened + ## against the next candidate port — the documented client retry contract. + ## Writes against followers are forwarded to the leader by the server. + var n = 1 + var portIdx = 0 + var db: DbConn + while true: + withLock args.lock[]: + if args.stop[]: break + if cast[pointer](db) == nil: + let port = args.ports[portIdx] + try: + db = open("127.0.0.1:" & $port, "", "", "default") + except CatchableError: + portIdx = (portIdx + 1) mod args.ports.len + sleep(50) + continue + try: + db.exec(sql("INSERT INTO load_test (id, n) VALUES (" & $n & ", " & $n & ")")) + withLock args.lock[]: + args.acked[].add n + inc n + except CatchableError: + if cast[pointer](db) != nil: + try: db.close() + except CatchableError: discard + db = default(DbConn) + portIdx = (portIdx + 1) mod args.ports.len + sleep(50) + if cast[pointer](db) != nil: + try: db.close() + except CatchableError: discard + +proc runFailoverLoadScenario() = + ## 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 bases: distinct from nimforum_smoke_test (35000+mod10000), + # raft_e2e_test (41000+mod5000) and raft_writes_e2e_test (46000+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 = 50000 + (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] + for i in 1 .. 3: + let id = "n" & $i + let dataDir = getTempDir() / "baradb_raft_failover_load_e2e_" & $tstamp & "_" & id + createDir(dataDir) + var env = newStringTable() + for key, val in envPairs(): + env[key] = val + env["BARADB_PORT"] = $(cbase + i * 10) + env["BARADB_RAFT_ENABLED"] = "true" + env["BARADB_RAFT_PORT"] = $(rbase + i) + env["BARADB_RAFT_NODE_ID"] = id + env["BARADB_RAFT_PEERS"] = peers + env["BARADB_RAFT_CLIENT_PEERS"] = clientPeers + 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) + nodes.add NodeProc(id: id, clientPort: cbase + i * 10, p: p, + dataDir: dataDir, alive: true) + + var + writer: Thread[WriterArgs] + writerStarted = false + wLock: Lock + wAcked: seq[int] + wStop = false + initLock(wLock) + try: + # Readiness: all three client ports accept TCP connections (10s each). + for i in 0 ..< nodes.len: + let readyStart = getTime() + var ok = false + while getTime() - readyStart < initDuration(seconds = 10): + if portOpen(nodes[i].clientPort): + ok = true + break + sleep(100) + if not ok: + 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, ")" + + # Schema goes through the raft ddl log on the leader (C3c). + block: + let db = openClient(nodes[leaderIdx].clientPort) + try: + db.exec(sql"CREATE TABLE load_test (id INT PRIMARY KEY, n INT)") + except CatchableError as e: + echo "leader CREATE TABLE failed: ", e.msg + dumpAll(nodes) + fail() + return + db.close() + echo "leader schema committed via raft ddl" + + # Load phase: sustained INSERTs from a background writer thread. + let wArgs = WriterArgs( + lock: addr wLock, acked: addr wAcked, stop: addr wStop, + ports: [nodes[leaderIdx].clientPort, + nodes[(leaderIdx + 1) mod 3].clientPort, + nodes[(leaderIdx + 2) mod 3].clientPort]) + createThread(writer, writerLoop, wArgs) + writerStarted = true + + # Wait until AckTarget writes are acknowledged (30s deadline). + var ackedAtKill = 0 + let loadStart = getTime() + while getTime() - loadStart < initDuration(seconds = 30): + withLock wLock: + ackedAtKill = wAcked.len + if ackedAtKill >= AckTarget: break + sleep(20) + if ackedAtKill < AckTarget: + echo "only ", ackedAtKill, " writes acked within 30s (need ", AckTarget, ")" + dumpAll(nodes) + fail() + return + echo "load phase: ", ackedAtKill, " writes acknowledged, killing leader ", + nodes[leaderIdx].id + + # Kill the leader mid-load. + killNode(nodes[leaderIdx]) + let killTime = getTime() + + # Assert A (availability): a survivor accepts an INSERT within 10s of + # the kill. Probe both survivors; keep the one that answers. Each attempt + # uses a fresh id: an attempt may commit but lose its response during the + # failover, and retrying the same id would then loop on UNIQUE violations. + var writerSurvivor = -1 + var writeErr = "" + var probeAttempt = 0 + while getTime() - killTime < initDuration(seconds = 10): + for i in 0 ..< nodes.len: + if i == leaderIdx: continue + inc probeAttempt + let probeId = ProbeId + probeAttempt + try: + let db = openClient(nodes[i].clientPort) + try: + db.exec(sql("INSERT INTO load_test (id, n) VALUES (" & $probeId & ", " & $probeId & ")")) + writerSurvivor = i + finally: + db.close() + if writerSurvivor >= 0: break + except CatchableError as e: + writeErr = e.msg + # "not leader" / commit timeout / connection blips — keep probing. + if writerSurvivor >= 0: break + sleep(100) + if writerSurvivor < 0: + echo "ASSERT A FAILED: no survivor accepted a write within 10s of the kill", + (if writeErr.len > 0: " (last error: " & writeErr & ")" else: "") + dumpAll(nodes) + fail() + return + let availMs = inMilliseconds(getTime() - killTime) + echo "availability: ", nodes[writerSurvivor].id, + " accepted a write ", availMs, "ms after the kill" + + # Stop the writer thread and snapshot what was acknowledged. + withLock wLock: + wStop = true + joinThreads(writer) + writerStarted = false + withLock wLock: + ackedAtKill = wAcked.len + echo "writer stopped; ", ackedAtKill, " total acknowledged writes" + + # Let the new leader stabilize and the remaining follower catch up: + # poll until both survivors agree on a row count that covers every + # acknowledged write (10s deadline). The >= ackedAtKill guard prevents + # a trivial 0 == 0 pass before any raft entries have been applied. + let survivorIdx = [0, 1, 2].filterIt(it != leaderIdx) + var caughtUp = false + let cuStart = getTime() + while getTime() - cuStart < initDuration(seconds = 10): + let c0 = countRows(nodes[survivorIdx[0]].clientPort) + let c1 = countRows(nodes[survivorIdx[1]].clientPort) + if c0 >= ackedAtKill and c0 == c1: + caughtUp = true + break + sleep(100) + if not caughtUp: + echo "survivors never reached equal row counts within 10s (", + countRows(nodes[survivorIdx[0]].clientPort), " vs ", + countRows(nodes[survivorIdx[1]].clientPort), ")" + dumpAll(nodes) + fail() + return + echo "survivors caught up: equal row counts" + + # Assert B (durability): every acknowledged id must be present on BOTH + # survivors. + var acked: HashSet[int] + withLock wLock: + acked = toHashSet(wAcked) + for i in survivorIdx: + var ids: HashSet[int] + try: + ids = allIds(nodes[i].clientPort) + except CatchableError as e: + echo "ASSERT B FAILED: could not read ids from ", nodes[i].id, + ": ", e.msg + dumpAll(nodes) + fail() + return + let missing = acked - ids + if missing.len > 0: + echo "ASSERT B FAILED: ", nodes[i].id, " is missing ", + missing.len, " acknowledged ids" + dumpAll(nodes) + fail() + return + echo "durability: ", nodes[i].id, " contains all ", + acked.len, " acknowledged ids" + + check acked.len >= AckTarget + finally: + if writerStarted: + withLock wLock: + wStop = true + joinThreads(writer) + deinitLock(wLock) + for n in nodes.mitems: + n.killNode() + if n.p != nil: n.p.close() + removeDir(n.dataDir) + +suite "Raft failover under load E2E": + test "committed writes survive a leader kill under sustained write load": + if not fileExists(BinaryPath): + echo "[SKIP] ", BinaryPath, " missing — run `nimble test` (builds the server first)" + skip() + else: + runFailoverLoadScenario()