From d3b38251c3825162ac04612bb202b259c1423d58 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Thu, 7 May 2026 15:16:40 +0300 Subject: [PATCH] update core modules and project config --- baradadb.nimble | 2 ++ src/barabadb/core/httpserver.nim | 8 ++--- src/barabadb/core/replication.nim | 52 ++++++++++++++++++++++--------- src/barabadb/core/server.nim | 52 ++++++++++++++++++++++++++++--- src/barabadb/query/executor.nim | 38 +++++++++++++++------- 5 files changed, 117 insertions(+), 35 deletions(-) diff --git a/baradadb.nimble b/baradadb.nimble index e35b2b7..db4f5b9 100644 --- a/baradadb.nimble +++ b/baradadb.nimble @@ -7,6 +7,8 @@ srcDir = "src" bin = @["baradadb"] binDir = "build" +switch("define", "ssl") + # Dependencies requires "nim >= 2.2.0" requires "https://github.com/katehonz/hunos >= 1.2.0" diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index 7e9be94..d8c0e8d 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -58,10 +58,10 @@ proc newHttpServer*(config: BaraConfig): HttpServer = proc createToken*(server: HttpServer, userId, role: string): string = let header = %*{"alg": "HS256", "typ": "JWT"} var claims = newTable[string, Claim]() - claims["sub"] = newStringClaim(userId) - claims["role"] = newStringClaim(role) - claims["iat"] = newTimeClaim(getTime()) - claims["exp"] = newTimeClaim(getTime() + 24.hours) + claims["sub"] = newSUB(%userId) + claims["role"] = newClaim(GENERAL, %role) + claims["iat"] = newIAT(getTime().toUnix()) + claims["exp"] = newEXP((getTime() + 24.hours).toUnix()) var token = initJWT(header, claims) token.sign(server.secretKey) return $token diff --git a/src/barabadb/core/replication.nim b/src/barabadb/core/replication.nim index 6eb9a50..2a0eb7c 100644 --- a/src/barabadb/core/replication.nim +++ b/src/barabadb/core/replication.nim @@ -2,6 +2,9 @@ import std/tables import std/sets import std/locks +import std/net +import std/strutils + type ReplicationMode* = enum @@ -69,40 +72,59 @@ proc connectReplica*(rm: ReplicationManager, id: string) = rm.replicas[id].connected = true release(rm.lock) -proc disconnectReplica*(rm: ReplicationManager, id: string) = - acquire(rm.lock) - if id in rm.replicas: - rm.replicas[id].state = rsDisconnected - rm.replicas[id].connected = false - release(rm.lock) +proc shipToReplica(replica: Replica, lsn: uint64, data: seq[byte]): bool = + ## Send replication data to a replica via TCP. + ## Protocol: "REP \n" + ## Response: "ACK \n" on success + try: + var sock = newSocket() + sock.connect(replica.host, Port(replica.port)) + let header = "REP " & $lsn & " " & $data.len & "\n" + sock.send(header) + if data.len > 0: + sock.send(cast[string](data)) + var response = "" + sock.readLine(response) + sock.close() + let parts = response.strip().split(" ") + return parts.len >= 2 and parts[0] == "ACK" + except: + return false proc writeLsn*(rm: ReplicationManager, data: seq[byte]): uint64 = acquire(rm.lock) inc rm.currentLsn let lsn = rm.currentLsn + var replicasToShip: seq[Replica] + for id, replica in rm.replicas: + if replica.connected and replica.host.len > 0 and replica.port > 0: + replicasToShip.add(replica) + case rm.mode of rmAsync: - # Fire and forget — don't wait release(rm.lock) + for replica in replicasToShip: + discard shipToReplica(replica, lsn, data) return lsn of rmSync: - # Wait for all replicas rm.pendingAcks[lsn] = initHashSet[string]() - for id, replica in rm.replicas: - if replica.connected: - rm.pendingAcks[lsn].incl(id) + for replica in replicasToShip: + rm.pendingAcks[lsn].incl(replica.id) release(rm.lock) + for replica in replicasToShip: + discard shipToReplica(replica, lsn, data) return lsn of rmSemiSync: - # Wait for N replicas rm.pendingAcks[lsn] = initHashSet[string]() var count = 0 - for id, replica in rm.replicas: - if replica.connected and count < rm.syncReplicaCount: - rm.pendingAcks[lsn].incl(id) + for replica in replicasToShip: + if count < rm.syncReplicaCount: + rm.pendingAcks[lsn].incl(replica.id) inc count release(rm.lock) + for replica in replicasToShip: + discard shipToReplica(replica, lsn, data) return lsn proc ackLsn*(rm: ReplicationManager, replicaId: string, lsn: uint64) = diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index a528e65..0d7cf14 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -18,6 +18,7 @@ import ../query/executor import ../storage/lsm import ../core/mvcc import ../core/disttxn +import ../core/replication import jwt as jwtlib type @@ -28,6 +29,7 @@ type ctx*: ExecutionContext txnManager*: TxnManager distTxnManager*: DistTxnManager + replicationManager*: ReplicationManager tls*: TLSContext activeConnections*: int @@ -46,7 +48,8 @@ proc newServer*(config: BaraConfig): Server = let tlsConfig = newTLSConfig(config.certFile, config.keyFile) tls = newTLSContext(tlsConfig) Server(config: config, running: false, db: db, ctx: ctx, - txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(), tls: tls) + txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(), + replicationManager: newReplicationManager(), tls: tls) # ---------------------------------------------------------------------- # Wire Protocol Helpers @@ -111,7 +114,8 @@ proc valueToWire(val: string, colType: string): WireValue = return WireValue(kind: fkJson, jsonVal: val) return WireValue(kind: fkString, strVal: val) -proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq[WireValue] = @[]): (bool, QueryResult, string) = +proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq[WireValue] = @[], + replication: ReplicationManager = nil): (bool, QueryResult, string) = try: let tokens = tokenize(query) let astNode = parse(tokens) @@ -121,6 +125,14 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq let result = executor.executeQuery(ctx, astNode, params) if result.success: + # Ship written key-value pairs to replicas + if replication != nil and result.keyValuePairs.len > 0: + for (key, value) in result.keyValuePairs: + var data = newSeq[byte](key.len + 1 + value.len) + for i, c in key: data[i] = byte(c) + data[key.len] = byte(0) + for i, c in value: data[key.len + 1 + i] = c + discard replication.writeLsn(data) var qr = QueryResult(affectedRows: result.affectedRows, rowCount: result.rows.len) qr.columns = result.columns @@ -304,6 +316,38 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} await client.send("ERR invalid message\n") continue + # Detect replication data (starts with "REP ") + if headerData.len >= 4 and headerData[0..3] == "REP ": + var rest = headerData[4..^1] + while '\n' notin rest: + let more = await client.recv(1024) + if more.len == 0: break + rest.add(more) + let parts = rest.strip().split(" ") + if parts.len >= 2: + let lsn = try: parseUInt(parts[0]) except: 0'u64 + let dataLen = try: parseInt(parts[1]) except: 0 + if dataLen > 0: + var data = "" + while data.len < dataLen: + let chunk = await client.recv(dataLen - data.len) + 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: + let key = data[0.. 0: + server.db.put(key, cast[seq[byte]](value)) + else: + server.db.delete(key) + await client.send("ACK " & $lsn & "\n") + else: + await client.send("ERR\n") + continue + let (ok, header) = parseHeader(headerData) if not ok: break @@ -338,7 +382,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} info("[" & $clientId & "] Query: " & queryStr) let startTicks = getMonoTime().ticks() - let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr) + let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, replication=server.replicationManager) let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000) if durationMs >= slowThreshold: @@ -359,7 +403,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} info("[" & $clientId & "] QueryParams: " & queryStr & " (" & $params.len & " params)") let startTicks = getMonoTime().ticks() - let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, params) + let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, params, replication=server.replicationManager) let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000) if durationMs >= slowThreshold: diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index d61b86d..d528c67 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -118,9 +118,12 @@ type rows*: seq[Row] affectedRows*: int message*: string + keyValuePairs*: seq[(string, seq[byte])] -proc okResult*(rows: seq[Row] = @[], cols: seq[string] = @[], affected: int = 0, msg: string = ""): ExecResult = - ExecResult(success: true, columns: cols, rows: rows, affectedRows: affected, message: msg) +proc okResult*(rows: seq[Row] = @[], cols: seq[string] = @[], affected: int = 0, msg: string = "", + kvPairs: seq[(string, seq[byte])] = @[]): ExecResult = + ExecResult(success: true, columns: cols, rows: rows, affectedRows: affected, message: msg, + keyValuePairs: kvPairs) proc errResult*(msg: string): ExecResult = ExecResult(success: false, columns: @[], rows: @[], affectedRows: 0, message: msg) @@ -688,7 +691,8 @@ proc execPointRead(ctx: ExecutionContext, table: string, key: string): seq[Row] return @[row] return @[] -proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], values: seq[seq[string]]): int = +proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], values: seq[seq[string]], + kvPairs: var seq[(string, seq[byte])]): int = if not hasPrivilege(ctx, table, "INSERT"): return 0 var count = 0 @@ -720,6 +724,7 @@ proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], valu discard ctx.txnManager.write(ctx.pendingTxn, fullKey, cast[seq[byte]](valStr)) else: ctx.db.put(fullKey, cast[seq[byte]](valStr)) + kvPairs.add((fullKey, cast[seq[byte]](valStr))) for colName in ctx.btrees.keys.toSeq(): if colName.startsWith(table & "."): @@ -746,7 +751,8 @@ proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], valu inc count return count -proc execDelete*(ctx: ExecutionContext, table: string, key: string): int = +proc execDelete*(ctx: ExecutionContext, table: string, key: string, + kvPairs: var seq[(string, seq[byte])]): int = if not hasPrivilege(ctx, table, "DELETE"): return 0 let fullKey = table & "." & key @@ -763,6 +769,7 @@ proc execDelete*(ctx: ExecutionContext, table: string, key: string): int = discard ctx.txnManager.delete(ctx.pendingTxn, fullKey) else: ctx.db.delete(fullKey) + kvPairs.add((fullKey, @[])) # Update FTS indexes for ftsKey, ftsIdx in ctx.ftsIndexes: if ftsKey.startsWith(table & "."): @@ -773,7 +780,8 @@ proc execDelete*(ctx: ExecutionContext, table: string, key: string): int = return 1 return 0 -proc execUpdateRow*(ctx: ExecutionContext, table: string, key: string, sets: Table[string, string]): int = +proc execUpdateRow*(ctx: ExecutionContext, table: string, key: string, sets: Table[string, string], + kvPairs: var seq[(string, seq[byte])]): int = if not hasPrivilege(ctx, table, "UPDATE"): return 0 let fullKey = table & "." & key @@ -819,6 +827,7 @@ proc execUpdateRow*(ctx: ExecutionContext, table: string, key: string, sets: Tab discard ctx.txnManager.write(ctx.pendingTxn, fullKey, cast[seq[byte]](newVal)) else: ctx.db.put(fullKey, cast[seq[byte]](newVal)) + kvPairs.add((fullKey, cast[seq[byte]](newVal))) # Update FTS indexes: remove old doc, add new for ftsKey, ftsIdx in ctx.ftsIndexes: if ftsKey.startsWith(table & "."): @@ -1953,7 +1962,8 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] row[f] = mutableValues[0][i] fireTriggers(ctx, stmt.insTarget, "before", "insert", row) - let count = execInsert(ctx, stmt.insTarget, mutableFields, mutableValues) + var kvPairs: seq[(string, seq[byte])] + let count = execInsert(ctx, stmt.insTarget, mutableFields, mutableValues, kvPairs) # Fire AFTER INSERT triggers fireTriggers(ctx, stmt.insTarget, "after", "insert", row) @@ -1961,7 +1971,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] if ctx.onChange != nil: for i in 0..