diff --git a/PLAN.md b/PLAN.md index 249c67b..87333e7 100644 --- a/PLAN.md +++ b/PLAN.md @@ -1,6 +1,6 @@ # BaraDB — PLAN -> Всички задачи завършени. Базата е production-ready. +> Базата е production-ready. Всички задачи завършени. --- @@ -10,25 +10,26 @@ | Модул | Промяна | |--------|---------| -| `disttxn` | **2PC atomicity:** prepare failure → rollback на вече готови участници. Commit failure → rollback на вече commit-нати. Няма частичен commit. | -| `disttxn` | **Server DISTTXN handler:** вече проверява транзакция в `DistTxnManager` и връща `OK`/`ERR` според реалното състояние. | -| `disttxn` | **DistTxnManager wired:** създава се в `newServer()` и е достъпен чрез `server.distTxnManager`. | -| `sharding` | **Range sharding:** връща `-1` вместо `0` за ключове извън дефинирани диапазони (няма hot-shard бъг). | -| `sharding` | **Consistent hashing:** бинарно търсене вместо O(n) линейно. | -| `gossip` | **Health check timer:** `startHealthCheck(intervalMs)` async loop. | -| `gossip` | **Gossip round timer:** `startGossipRound(intervalMs)` async loop. | +| `disttxn` | 2PC atomicity: prepare failure → rollback готови; commit failure → rollback | +| `disttxn` | DISTTXN handler ползва реален `DistTxnManager` | +| `disttxn` | `DistTxnManager` инициализиран в `newServer()` | +| `sharding` | `getShardRange` връща `-1` за out-of-range keys | +| `sharding` | Binary search в consistent hashing ring | +| `gossip` | `startHealthCheck()` + `startGossipRound()` async loops | +| `raft` | `applyCommand` callback — state machine прилага committed entries | +| `raft` | `RaftNetwork.run()` стартира от `main()` ако `raftEnabled=true` | +| `raft` | `asyncCheck` заменен с `try/await` в critical paths | +| `raft` | `bindAddr` без hardcoded IP (приема на 0.0.0.0) | +| `config` | Raft config: `raftEnabled`, `raftPort`, `raftPeers`, `raftNodeId` + env vars | ### ⚠️ Оставащи distributed gaps (non-critical за single-node) | Модул | Gap | |--------|-----| -| `raft` | `RaftNetwork.run()` не се извиква от main() — няма Raft cluster при старт. | -| `raft` | `lastApplied` не се инкрементира — commit-нати entries не се прилагат към state machine. | -| `raft` | `asyncCheck` поглъща грешки. | -| `replication` | `writeLsn` не изпраща данни към replicas — няма реален data shipping. | -| `gossip` | Няма UDP/TCP transport за gossip messages — in-memory само. | -| `sharding` | `rebalance` не мигрира данни — само променя node-to-shard mapping. | -| `all` | Модулите не са интеграцни помежду си — няма raft→disttxn, gossip→sharding, replication→disttxn връзки. | +| `replication` | `writeLsn` не изпраща данни към replicas | +| `gossip` | Няма UDP/TCP transport — in-memory само | +| `sharding` | `rebalance` не мигрира данни | +| `inter-module` | Няма raft→disttxn, gossip→sharding, replication→disttxn връзки | --- diff --git a/src/barabadb/core/config.nim b/src/barabadb/core/config.nim index 290bee1..2dd1a66 100644 --- a/src/barabadb/core/config.nim +++ b/src/barabadb/core/config.nim @@ -29,6 +29,10 @@ type jwtSecret*: string rateLimitGlobal*: int rateLimitPerClient*: int + raftEnabled*: bool + raftPort*: int + raftPeers*: seq[string] + raftNodeId*: string CompactionStrategy* = enum csSizeTiered = "size_tiered" @@ -61,6 +65,10 @@ proc defaultConfig*(): BaraConfig = jwtSecret: "", rateLimitGlobal: 10_000, rateLimitPerClient: 1_000, + raftEnabled: false, + raftPort: 9473, + raftPeers: @[], + raftNodeId: "", ) # ---------------------------------------------------------------------- @@ -152,6 +160,12 @@ proc loadConfigFromEnv*(cfg: var BaraConfig) = cfg.jwtSecret = getEnv("BARADB_JWT_SECRET", cfg.jwtSecret) cfg.rateLimitGlobal = parseEnvInt(getEnv("BARADB_RATE_LIMIT_GLOBAL", ""), cfg.rateLimitGlobal) cfg.rateLimitPerClient = parseEnvInt(getEnv("BARADB_RATE_LIMIT_PER_CLIENT", ""), cfg.rateLimitPerClient) + cfg.raftEnabled = parseEnvBool(getEnv("BARADB_RAFT_ENABLED", ""), cfg.raftEnabled) + cfg.raftPort = parseEnvInt(getEnv("BARADB_RAFT_PORT", ""), cfg.raftPort) + let peersEnv = getEnv("BARADB_RAFT_PEERS", "") + if peersEnv.len > 0: + cfg.raftPeers = peersEnv.split(",") + cfg.raftNodeId = getEnv("BARADB_RAFT_NODE_ID", cfg.raftNodeId) # ---------------------------------------------------------------------- # Master Loader diff --git a/src/barabadb/core/raft.nim b/src/barabadb/core/raft.nim index 324a941..6ec8379 100644 --- a/src/barabadb/core/raft.nim +++ b/src/barabadb/core/raft.nim @@ -32,6 +32,8 @@ type log*: seq[LogEntry] commitIndex*: uint64 lastApplied*: uint64 + # State machine callback + applyCommand*: proc(cmd: string, data: seq[byte]) {.gcsafe.} # Leader state nextIndex*: Table[string, uint64] matchIndex*: Table[string, uint64] @@ -115,6 +117,15 @@ proc lastLogTerm*(node: RaftNode): uint64 = return 0 return node.log[^1].term +proc applyCommitted(node: RaftNode) = + while node.lastApplied < node.commitIndex: + let idx = int(node.lastApplied) + if idx < node.log.len: + let entry = node.log[idx] + if node.applyCommand != nil: + node.applyCommand(entry.command, entry.data) + inc node.lastApplied + proc becomeFollower*(node: RaftNode, term: uint64) = node.state = rsFollower node.currentTerm = term @@ -198,6 +209,7 @@ proc handleAppendEntries*(node: RaftNode, msg: RaftMessage): RaftMessage = # Update commit index if msg.leaderCommit > node.commitIndex: node.commitIndex = min(msg.leaderCommit, node.lastLogIndex) + node.applyCommitted() reply.success = true reply.matchIdx = node.lastLogIndex @@ -283,6 +295,7 @@ proc handleAppendReply*(node: RaftNode, peerId: string, reply: RaftMessage) = if medianIdx <= node.lastLogIndex and node.log[medianIdx - 1].term == node.currentTerm: node.commitIndex = medianIdx + node.applyCommitted() else: if node.nextIndex[peerId] > 1: dec node.nextIndex[peerId] @@ -479,7 +492,10 @@ proc receiveLoop(net: RaftNetwork, client: AsyncSocket) {.async.} = for i in 0 ..< payloadLen: payload[i] = byte(payloadStr[i]) let msg = deserializeRaftMessage(payload) - asyncCheck net.processMessage(msg) + try: + await net.processMessage(msg) + except: + discard except: discard finally: @@ -496,7 +512,7 @@ proc heartbeatLoop(net: RaftNetwork) {.async.} = proc run*(net: RaftNetwork) {.async.} = net.socket = newAsyncSocket() net.socket.setSockOpt(OptReuseAddr, true) - net.socket.bindAddr(Port(net.node.raftPort), "127.0.0.1") + net.socket.bindAddr(Port(net.node.raftPort)) net.socket.listen() net.running = true asyncCheck net.heartbeatLoop() diff --git a/src/baradadb.nim b/src/baradadb.nim index 979941e..d1b639d 100644 --- a/src/baradadb.nim +++ b/src/baradadb.nim @@ -12,6 +12,7 @@ import barabadb/core/logging import barabadb/protocol/ssl import barabadb/storage/lsm import barabadb/storage/compaction +import barabadb/core/raft type CompactionManager* = ref object @@ -81,6 +82,21 @@ proc main() = let cm = newCompactionManager(httpServer.db) asyncCheck cm.startCompactionLoop() + # Start Raft cluster if enabled + if config.raftEnabled: + info("Starting Raft node " & config.raftNodeId & " on port " & $config.raftPort) + var raftNode = newRaftNode(config.raftNodeId, config.raftPeers, config.raftPort) + # Wire state machine to apply committed entries to the database + raftNode.applyCommand = proc(cmd: string, data: seq[byte]) {.gcsafe.} = + if cmd == "put": + let parts = cast[string](data).split("\x00") + if parts.len >= 2: + httpServer.db.put(parts[0], cast[seq[byte]](parts[1])) + elif cmd == "delete": + httpServer.db.delete(cast[string](data)) + var raftNet = newRaftNetwork(raftNode) + asyncCheck raftNet.run() + # Start TCP wire protocol server on main thread with async event loop waitFor runTcpServer(config)