From a72e17f063a0e6fe7a7daefd473b9479df45bdde Mon Sep 17 00:00:00 2001 From: dimgigov Date: Tue, 12 May 2026 23:30:58 +0300 Subject: [PATCH] =?UTF-8?q?fix:=20=D0=BF=D0=BE=D1=81=D0=BB=D0=B5=D0=B4?= =?UTF-8?q?=D0=BD=D0=B8=D1=82=D0=B5=204=20=D0=BA=D1=80=D0=B8=D1=82=D0=B8?= =?UTF-8?q?=D1=87=D0=BD=D0=B8/=D1=81=D1=80=D0=B5=D0=B4=D0=BD=D0=B8=20?= =?UTF-8?q?=D0=B1=D1=8A=D0=B3=D0=B0=20=E2=80=94=20LSM-Tree,=20MVCC,=20Dist?= =?UTF-8?q?Txn,=20Raft?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Поправени проблеми: - LSM-Tree: WAL write вече е извън db.lock (отделен walLock) → по-добра concurrency - MVCC: добавен compactVersions() който се изпълнява на всеки 100 commits Премахва стари overwritten версии, които не са видими за active транзакции - DistTxn: commit() вече НЕ rollback-ва commit-нали participants Ако някой participant е commit-нал, транзакцията се маркира като committed - Raft: добавен disk persistence за currentTerm, votedFor, log saveState() при всяка промяна; loadState() при стартиране 292 теста, 0 failure-а. --- src/barabadb/core/disttxn.nim | 18 ++++----- src/barabadb/core/mvcc.nim | 34 +++++++++++++++++ src/barabadb/core/raft.nim | 70 ++++++++++++++++++++++++++++++++++- src/barabadb/storage/lsm.nim | 16 ++++++-- 4 files changed, 123 insertions(+), 15 deletions(-) diff --git a/src/barabadb/core/disttxn.nim b/src/barabadb/core/disttxn.nim index 179402c..7fc2537 100644 --- a/src/barabadb/core/disttxn.nim +++ b/src/barabadb/core/disttxn.nim @@ -166,16 +166,16 @@ proc commit*(txn: DistributedTransaction): bool = if allOk: txn.state = dtsCommitted else: - # Rollback committed participants on commit failure - for nodeId in committedNodes: - var p = txn.participants[nodeId] - if p.host.len > 0 and p.port > 0: - discard sendDistTxnRpc(p.host, p.port, txn.id, "ROLLBACK") - p.committed = false - p.aborted = true - txn.state = dtsAborted + if committedNodes.len > 0: + # Some participants already committed — cannot rollback committed nodes + # without violating atomicity. Mark as committed and rely on + # reconciliation/retry for the remaining participants. + txn.state = dtsCommitted + else: + # No participant committed yet — safe to abort + txn.state = dtsAborted release(txn.lock) - return allOk + return allOk or committedNodes.len > 0 proc rollback*(txn: DistributedTransaction): bool = acquire(txn.lock) diff --git a/src/barabadb/core/mvcc.nim b/src/barabadb/core/mvcc.nim index e0d5412..8b270a6 100644 --- a/src/barabadb/core/mvcc.nim +++ b/src/barabadb/core/mvcc.nim @@ -229,6 +229,8 @@ proc write*(tm: TxnManager, txn: Transaction, key: string, value: seq[byte]): bo proc delete*(tm: TxnManager, txn: Transaction, key: string): bool = return tm.write(txn, key, @[]) +proc compactVersions*(tm: TxnManager) + proc commit*(tm: TxnManager, txn: Transaction): bool = acquire(tm.lock) if txn.state != tsActive: @@ -258,9 +260,41 @@ proc commit*(tm: TxnManager, txn: Transaction): bool = tm.committedTxnsSet.incl(txn.id) tm.activeTxns.del(txn.id) tm.deadlockDetector.removeTxn(uint64(txn.id)) + + # Compact version chains periodically (every 100 commits) + if tm.committedTxns.len mod 100 == 0: + tm.compactVersions() + release(tm.lock) return true +proc compactVersions*(tm: TxnManager) = + ## Remove old overwritten versions that are no longer visible to any active transaction. + for key, versions in tm.globalVersions.mpairs: + if versions.len <= 3: + continue + var newVersions: seq[VersionedRecord] = @[] + for i in 0.. 0: + s.writeData(addr entry.data[0], entry.data.len) + s.close() + moveFile(tmpPath, path) + +proc loadState(node: RaftNode) = + if node.dataDir.len == 0: return + let path = node.dataDir / RaftStateFile + if not fileExists(path): return + var s = newFileStream(path, fmRead) + if s == nil: return + try: + node.currentTerm = s.readUint64() + let votedForLen = int(s.readUint32()) + if votedForLen > 0: + node.votedFor = s.readStr(votedForLen) + let logLen = int(s.readUint32()) + node.log = newSeq[LogEntry](logLen) + for i in 0.. 0: + discard s.readData(addr data[0], dataLen) + node.log[i] = LogEntry(term: term, index: index, command: cmd, data: data) + except: + discard + s.close() + +proc newRaftNode*(id: string, peers: seq[string], raftPort: int = 0, + dataDir: string = ""): RaftNode = randomize() - RaftNode( + result = RaftNode( id: id, state: rsFollower, currentTerm: 0, @@ -92,7 +146,9 @@ proc newRaftNode*(id: string, peers: seq[string], raftPort: int = 0): RaftNode = votesReceived: initHashSet[string](), peerAddrs: initTable[string, tuple[host: string, port: int]](), raftPort: raftPort, + dataDir: dataDir, ) + result.loadState() proc newRaftCluster*(): RaftCluster = RaftCluster( @@ -131,6 +187,7 @@ proc becomeFollower*(node: RaftNode, term: uint64) = node.currentTerm = term node.votedFor = "" node.votesReceived.clear() + node.saveState() proc becomeCandidate*(node: RaftNode) = node.state = rsCandidate @@ -138,6 +195,7 @@ proc becomeCandidate*(node: RaftNode) = node.votedFor = node.id node.votesReceived.clear() node.votesReceived.incl(node.id) + node.saveState() proc becomeLeader*(node: RaftNode) = node.state = rsLeader @@ -166,6 +224,7 @@ proc handleRequestVote*(node: RaftNode, msg: RaftMessage): RaftMessage = if canVote and logOk: node.votedFor = msg.senderId + node.saveState() reply.success = true reply.term = node.currentTerm @@ -197,14 +256,20 @@ proc handleAppendEntries*(node: RaftNode, msg: RaftMessage): RaftMessage = return reply # Append new entries + var logChanged = false for entry in msg.entries: let idx = int(entry.index - 1) if idx < node.log.len: if node.log[idx].term != entry.term: node.log.setLen(idx) node.log.add(entry) + logChanged = true else: node.log.add(entry) + logChanged = true + + if logChanged: + node.saveState() # Update commit index if msg.leaderCommit > node.commitIndex: @@ -258,6 +323,7 @@ proc appendLog*(node: RaftNode, command: string, data: seq[byte] = @[]): LogEntr data: data, ) node.log.add(result) + node.saveState() proc handleVoteReply*(node: RaftNode, reply: RaftMessage) = if reply.term > node.currentTerm: diff --git a/src/barabadb/storage/lsm.nim b/src/barabadb/storage/lsm.nim index d713ef6..3206cf5 100644 --- a/src/barabadb/storage/lsm.nim +++ b/src/barabadb/storage/lsm.nim @@ -51,6 +51,7 @@ type currentSeq: uint64 nextSSTableId: int lock*: Lock + walLock*: Lock proc newMemTable(maxSize: int = DefaultMemTableSize): MemTable = MemTable(entries: @[], size: 0, maxSize: maxSize) @@ -322,6 +323,7 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree = new(result) initLock(result.lock) + initLock(result.walLock) result.dir = dir result.memTable = newMemTable(memMaxSize) result.immutableMem = newMemTable(0) @@ -375,10 +377,13 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree = stream.close() proc put*(db: LSMTree, key: string, value: seq[byte]) = + let ts = uint64(getMonoTime().ticks()) + acquire(db.walLock) + db.wal.writePut(cast[seq[byte]](key), value, ts) + release(db.walLock) + acquire(db.lock) defer: release(db.lock) - let ts = uint64(getMonoTime().ticks()) - db.wal.writePut(cast[seq[byte]](key), value, ts) if not db.memTable.put(key, value, ts): if db.immutableMem.len > 0: db.flushUnsafe() @@ -387,10 +392,13 @@ proc put*(db: LSMTree, key: string, value: seq[byte]) = discard db.memTable.put(key, value, ts) proc delete*(db: LSMTree, key: string) = + let ts = uint64(getMonoTime().ticks()) + acquire(db.walLock) + db.wal.writeDelete(cast[seq[byte]](key), ts) + release(db.walLock) + acquire(db.lock) defer: release(db.lock) - let ts = uint64(getMonoTime().ticks()) - db.wal.writeDelete(cast[seq[byte]](key), ts) if not db.memTable.put(key, @[], ts, deleted = true): if db.immutableMem.len > 0: db.flushUnsafe()