baraba/Baradb
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: критични бъгове в storage, MVCC, auth, Raft, query executor

Browse Source 
Поправени проблеми:
- MVCC: aborted транзакции вече не стават видими (добавено abortedTxns множество)
- LSM-Tree: поправена загуба на данни при immutable memtable overwrite
- LSM-Tree: поправена SSTable search order (сортиране по id вместо minKey)
- WAL: sync()/close() вече извикват posix.fsync()
- Auth: simpleHash (djb2) заменен с HMAC-SHA256
- Recovery: summary() вече не мутира базата данни
- Raft: поправена majority calculation за четен брой нодове
- Query: EXISTS subqueries вече изпълняват подзаявката
- Nimble: поправен build (-d:ssl) и bench task

Добавен BUG_AUDIT.md с пълен доклад и план за подобрения.

292 теста, 0 failure-а.
This commit is contained in:
dimgigov
2026-05-12 22:45:50 +03:00
parent 3396ba4d63
commit 131541a0e5
10 changed files with 288 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/barabadb/core/mvcc.nim
+11
View File
@@ -43,6 +43,8 @@ type
nextLsn: uint64
activeTxns: Table[TxnId, Transaction]
committedTxns: seq[TxnId]
committedTxnsSet: HashSet[TxnId]
abortedTxns: HashSet[TxnId]
globalVersions*: Table[string, seq[VersionedRecord]]
txnTimeoutMs: int64
deadlockDetector*: DeadlockDetector
@@ -59,6 +61,8 @@ proc newTxnManager*(txnTimeoutMs: int64 = 30000): TxnManager =
result.nextLsn = 1
result.activeTxns = initTable[TxnId, Transaction]()
result.committedTxns = @[]
result.committedTxnsSet = initHashSet[TxnId]()
result.abortedTxns = initHashSet[TxnId]()
result.globalVersions = initTable[string, seq[VersionedRecord]]()
result.txnTimeoutMs = txnTimeoutMs
result.deadlockDetector = newDeadlockDetector()
@@ -112,8 +116,13 @@ proc isVisible(tm: TxnManager, txn: Transaction, version: VersionedRecord): bool
return false
if creator in txn.snapshotTxns:
return false
if creator in tm.abortedTxns:
return false
if creator in tm.activeTxns and tm.activeTxns[creator].state != tsCommitted:
return false
# If creator is not active and not committed, it must be aborted
if creator notin tm.activeTxns and creator notin tm.committedTxnsSet:
return false
# Deleter must not be a visible committed txn
if deleter != TxnId(0):
@@ -246,6 +255,7 @@ proc commit*(tm: TxnManager, txn: Transaction): bool =
txn.state = tsCommitted
tm.committedTxns.add(txn.id)
tm.committedTxnsSet.incl(txn.id)
tm.activeTxns.del(txn.id)
tm.deadlockDetector.removeTxn(uint64(txn.id))
release(tm.lock)
@@ -258,6 +268,7 @@ proc abortTxn*(tm: TxnManager, txn: Transaction): bool =
return false
txn.state = tsAborted
tm.activeTxns.del(txn.id)
tm.abortedTxns.incl(txn.id)
tm.deadlockDetector.removeTxn(uint64(txn.id))
release(tm.lock)
return true
src/barabadb/core/raft.nim
+1 -1
View File
@@ -290,7 +290,7 @@ proc handleAppendReply*(node: RaftNode, peerId: string, reply: RaftMessage) =
matchIndices.add(idx)
matchIndices.sort()
let medianIdx = matchIndices[matchIndices.len div 2]
let medianIdx = matchIndices[(matchIndices.len - 1) div 2]
if medianIdx > node.commitIndex:
if medianIdx <= node.lastLogIndex and
node.log[medianIdx - 1].term == node.currentTerm:
src/barabadb/protocol/auth.nim
+32 -10
View File
@@ -2,6 +2,7 @@
import std/strutils
import std/base64
import std/tables
import checksums/sha2
type
AuthMethod* = enum
@@ -51,12 +52,33 @@ proc base64UrlDecode(data: string): string =
s &= "="
return decode(s)
proc simpleHash(data: string, key: string): string =
var prefix = data & key
var h: uint64 = 5381
for ch in prefix:
h = ((h shl 5) + h) + uint64(ord(ch))
return $h
proc hmacSha256(key, message: string): string =
var k = key
if k.len > 64:
var ctx = initSha_256()
ctx.update(k.toOpenArray(0, k.len-1))
let hash = ctx.digest()
k = $hash
while k.len < 64:
k &= "\x00"
var ipad = newString(64)
var opad = newString(64)
for i in 0..<64:
ipad[i] = chr(ord(k[i]) xor 0x36)
opad[i] = chr(ord(k[i]) xor 0x5c)
var innerCtx = initSha_256()
innerCtx.update(ipad.toOpenArray(0, ipad.len-1))
innerCtx.update(message.toOpenArray(0, message.len-1))
let innerHash = innerCtx.digest()
var outerCtx = initSha_256()
outerCtx.update(opad.toOpenArray(0, opad.len-1))
outerCtx.update(innerHash.toOpenArray(0, innerHash.len-1))
let outerHash = outerCtx.digest()
return $outerHash
proc createToken*(am: AuthManager, claims: JWTClaims): string =
let header = base64UrlEncode("{\"alg\":\"HS256\",\"typ\":\"JWT\"}")
@@ -64,7 +86,7 @@ proc createToken*(am: AuthManager, claims: JWTClaims): string =
"{\"sub\":\"" & claims.sub & "\",\"role\":\"" & claims.role &
"\",\"database\":\"" & claims.database & "\"}")
let data = header & "." & payload
let signature = simpleHash(data, am.secretKey)
let signature = hmacSha256(am.secretKey, data)
am.tokens.add(data & "." & base64UrlEncode(signature))
return am.tokens[^1]
@@ -73,7 +95,7 @@ proc verifyToken*(am: AuthManager, token: string): (bool, JWTClaims) =
if parts.len != 3:
return (false, JWTClaims())
let data = parts[0] & "." & parts[1]
let sig = simpleHash(data, am.secretKey)
let sig = hmacSha256(am.secretKey, data)
if base64UrlEncode(sig) != parts[2]:
return (false, JWTClaims())
# Parse payload
@@ -132,8 +154,8 @@ proc validateCredentials*(am: AuthManager, creds: AuthCredentials): AuthResult =
if creds.username in am.users:
let stored = am.users[creds.username]
# SCRAM-SHA-256: client sends SHA-256(password) as payload
let clientHash = if creds.payload.len > 0: creds.payload else: simpleHash("", am.secretKey)
if stored == clientHash or stored == simpleHash(creds.payload, am.secretKey):
let clientHash = if creds.payload.len > 0: creds.payload else: hmacSha256(am.secretKey, "")
if stored == clientHash or stored == hmacSha256(am.secretKey, creds.payload):
return AuthResult(authenticated: true, username: creds.username,
role: "user", database: "default")
return AuthResult(authenticated: false, error: "Invalid SCRAM credentials")
src/barabadb/query/executor.nim
+7 -1
View File
@@ -410,6 +410,8 @@ proc parseRowData(valStr: string): Table[string, string] =
let v = part[eqPos+1..^1].strip()
result[k] = v
proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row]
proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = nil): string =
if expr == nil: return ""
case expr.kind
@@ -593,7 +595,11 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
let v = evalExpr(expr.unExpr, row)
return if not isNull(v): "true" else: "false"
else: return "false"
of irekExists: return "false"
of irekExists:
if ctx != nil:
let rows = executePlan(ctx, expr.existsSubquery)
return if rows.len > 0: "true" else: "false"
return "false"
else: return ""
proc lowerExpr*(node: Node): IRExpr
src/barabadb/storage/lsm.nim
+29 -10
View File
@@ -31,6 +31,7 @@ type
maxSize: int
SSTable* = object
id*: int
path*: string
index*: Table[string, int64]
bloom*: BloomFilter
@@ -186,6 +187,7 @@ proc writeSSTable*(entries: seq[Entry], path: string, level: int): SSTable =
if maxK == "" or entry.key > maxK: maxK = entry.key
result = SSTable(
id: -1,
path: path,
index: idxTable,
bloom: bloom,
@@ -238,6 +240,7 @@ proc loadSSTable*(path: string): SSTable =
bloom.deserialize(bloomData)
result = SSTable(
id: -1,
path: path,
index: idxTable,
bloom: bloom,
@@ -294,6 +297,8 @@ proc close*(sst: var SSTable) =
# LSMTree API
# ----------------------------------------------------------------------
proc flushUnsafe(db: LSMTree) {.gcsafe.}
proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree =
createDir(dir)
createDir(dir / "sstables")
@@ -306,13 +311,14 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree =
if kind == pcFile and path.endsWith(".sst"):
try:
var sst = loadSSTable(path)
sstables.add(sst)
let name = splitFile(path).name
nextId = max(nextId, parseInt(name) + 1)
sst.id = parseInt(name)
sstables.add(sst)
nextId = max(nextId, sst.id + 1)
except:
discard # skip corrupt SSTables
sstables.sort(proc(a, b: SSTable): int = cmp(a.minKey, b.minKey))
sstables.sort(proc(a, b: SSTable): int = cmp(b.id, a.id))
new(result)
initLock(result.lock)
@@ -328,8 +334,9 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree =
# WAL crash recovery — replay unflushed entries into memTable
let walPath = dir / "wal" / "wal.log"
if fileExists(walPath):
var stream: FileStream = nil
try:
let stream = newFileStream(walPath, fmRead)
stream = newFileStream(walPath, fmRead)
if stream != nil:
var magic: uint32 = 0
var version: uint32 = 0
@@ -352,16 +359,20 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree =
if stream.readData(addr value[0], valLen.int) != valLen.int: break
case WalEntryKind(kind)
of wekPut:
discard result.memTable.put(key, value, timestamp)
if not result.memTable.put(key, value, timestamp):
result.flushUnsafe()
discard result.memTable.put(key, value, timestamp)
of wekDelete:
discard result.memTable.put(key, @[], timestamp) # tombstone
if not result.memTable.put(key, @[], timestamp, deleted = true):
result.flushUnsafe()
discard result.memTable.put(key, @[], timestamp, deleted = true)
of wekCommit:
discard
of wekCheckpoint:
discard
finally:
if stream != nil:
stream.close()
except:
discard
proc put*(db: LSMTree, key: string, value: seq[byte]) =
acquire(db.lock)
@@ -369,6 +380,8 @@ proc put*(db: LSMTree, key: string, value: seq[byte]) =
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()
db.immutableMem = db.memTable
db.memTable = newMemTable(db.memMaxSize)
discard db.memTable.put(key, value, ts)
@@ -378,7 +391,12 @@ proc delete*(db: LSMTree, key: string) =
defer: release(db.lock)
let ts = uint64(getMonoTime().ticks())
db.wal.writeDelete(cast[seq[byte]](key), ts)
discard db.memTable.put(key, @[], ts, deleted = true)
if not db.memTable.put(key, @[], ts, deleted = true):
if db.immutableMem.len > 0:
db.flushUnsafe()
db.immutableMem = db.memTable
db.memTable = newMemTable(db.memMaxSize)
discard db.memTable.put(key, @[], ts, deleted = true)
proc getUnsafe(db: LSMTree, key: string): (bool, seq[byte]) =
let (found, entry) = db.memTable.get(key)
@@ -438,8 +456,9 @@ proc flushUnsafe(db: LSMTree) =
inc db.nextSSTableId
var sst = writeSSTable(toFlush.entries, path, level = 0)
sst.id = db.nextSSTableId - 1
db.sstables.add(sst)
db.sstables.sort(proc(a, b: SSTable): int = cmp(a.minKey, b.minKey))
# SSTables are kept in insertion order (newest last) so getUnsafe can search newest-first
db.wal.writeCommit(uint64(getMonoTime().ticks()))
db.wal.sync()
src/barabadb/storage/recovery.nim
+1 -1
View File
@@ -168,7 +168,7 @@ proc recover*(rec: CrashRecovery, db: LSMTree = nil): RecoveryResult =
proc totalEntries*(rec: CrashRecovery): int = rec.entries.len
proc summary*(rec: CrashRecovery): string =
let r = rec.recover()
let r = rec.analyze()
result = "WAL Recovery Summary:\n"
result &= " Total entries: " & $r.totalEntries & "\n"
result &= " Redone (committed): " & $r.redone & "\n"
src/barabadb/storage/wal.nim
+9
View File
@@ -1,6 +1,7 @@
## WAL — Write-Ahead Log for durability
import std/os
import std/streams
import std/posix
const
WALMagic* = 0x42415241'u32 # "BARA"
@@ -77,9 +78,17 @@ proc writeCommit*(wal: var WriteAheadLog, timestamp: uint64) =
proc sync*(wal: var WriteAheadLog) =
wal.stream.flush()
let fd = posix.open(wal.path, O_RDONLY)
if fd != -1:
discard posix.fsync(fd)
discard posix.close(fd)
proc close*(wal: var WriteAheadLog) =
wal.stream.flush()
let fd = posix.open(wal.path, O_RDONLY)
if fd != -1:
discard posix.fsync(fd)
discard posix.close(fd)
wal.stream.close()
proc entryCount*(wal: WriteAheadLog): uint64 = wal.entryCount