fix(raft): graph apply + reject non-default DB writes
CI / test (push) Has been cancelled
CI / verify (push) Has been cancelled
Clients CI / build-server (push) Has been cancelled
Clients CI / test-python (push) Has been cancelled
Clients CI / test-javascript (push) Has been cancelled
Clients CI / test-nim (push) Has been cancelled
Clients CI / test-rust (push) Has been cancelled
CI / test (push) Has been cancelled
CI / verify (push) Has been cancelled
Clients CI / build-server (push) Has been cancelled
Clients CI / test-python (push) Has been cancelled
Clients CI / test-javascript (push) Has been cancelled
Clients CI / test-nim (push) Has been cancelled
Clients CI / test-rust (push) Has been cancelled
- applyReplicatedPut/Delete keep in-memory graphs in sync with node/edge table rows (idempotent edges via addEdgeWithIdIfAbsent). - When Raft is enabled, DML is refused on any database other than 'default' (the only DB the state machine is wired to). - Docs updated (en/bg); unit test for graph apply.
This commit is contained in:
@@ -17,7 +17,7 @@ Leader election и log репликация през TCP. Включване:
|
|||||||
| `BARADB_RAFT_PEERS` | Списък `id@host:port` (вкл. себе си) |
|
| `BARADB_RAFT_PEERS` | Списък `id@host:port` (вкл. себе си) |
|
||||||
| `BARADB_RAFT_WRITE_TIMEOUT_MS` | Макс. изчакване за majority commit при SQL записи (по подразбиране 5000) |
|
| `BARADB_RAFT_WRITE_TIMEOUT_MS` | Макс. изчакване за majority commit при SQL записи (по подразбиране 5000) |
|
||||||
|
|
||||||
Когато Raft е активен, SQL DML (`INSERT`/`UPDATE`/`DELETE`/`MERGE` и транзакционен `COMMIT`) се приема само от лидера: KV двойките се добавят в Raft лога и клиентът чака majority commit. Followers отказват записи с `not leader; leader is '…'`. Приложените записи отиват в **default** базата. DDL (напр. `CREATE TABLE`) още не се репликира — схемата трябва да се създаде на всеки възел.
|
Когато Raft е активен, SQL DML (`INSERT`/`UPDATE`/`DELETE`/`MERGE` и транзакционен `COMMIT`) се приема само от лидера на **`default`** базата: KV двойките се добавят в Raft лога и клиентът чака majority commit. Followers отказват записи с `not leader; leader is '…'`. Записи към друга database name се отказват (`raft writes only supported on the 'default' database`). Приложените записи обновяват LSM + secondary B-tree/FTS/HNSW индекси и in-memory графи на всеки възел. DDL (напр. `CREATE TABLE`) още не се репликира — схемата трябва да се създаде на всеки възел.
|
||||||
|
|
||||||
```nim
|
```nim
|
||||||
import barabadb/core/raft
|
import barabadb/core/raft
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ Leader election and log replication over TCP. Enable with:
|
|||||||
| `BARADB_RAFT_PEERS` | Comma-separated `id@host:port` (include self) |
|
| `BARADB_RAFT_PEERS` | Comma-separated `id@host:port` (include self) |
|
||||||
| `BARADB_RAFT_WRITE_TIMEOUT_MS` | Max wait for majority commit on SQL writes (default 5000) |
|
| `BARADB_RAFT_WRITE_TIMEOUT_MS` | Max wait for majority commit on SQL writes (default 5000) |
|
||||||
|
|
||||||
When Raft is enabled, SQL DML (`INSERT`/`UPDATE`/`DELETE`/`MERGE` and transactional `COMMIT`) is accepted only on the leader: each write's KV pairs are appended to the Raft log and the client waits until the entry is majority-committed. Followers reject writes with `not leader; leader is '…'`. Followers apply committed entries via `applyCommand` into the **default** database. DDL (e.g. `CREATE TABLE`) is not replicated yet — apply schema on every node.
|
When Raft is enabled, SQL DML (`INSERT`/`UPDATE`/`DELETE`/`MERGE` and transactional `COMMIT`) is accepted only on the leader of the **`default`** database: each write's KV pairs are appended to the Raft log and the client waits until the entry is majority-committed. Followers reject writes with `not leader; leader is '…'`. Writes against any other database name are rejected (`raft writes only supported on the 'default' database`). Committed entries update LSM plus secondary B-tree/FTS/HNSW indexes and in-memory graphs on every node. DDL (e.g. `CREATE TABLE`) is not replicated yet — apply schema on every node.
|
||||||
|
|
||||||
```nim
|
```nim
|
||||||
import barabadb/core/raft
|
import barabadb/core/raft
|
||||||
|
|||||||
@@ -254,15 +254,22 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq
|
|||||||
|
|
||||||
# C3b: writes go through the Raft log — only the leader may accept them.
|
# C3b: writes go through the Raft log — only the leader may accept them.
|
||||||
# Inspect every statement so "SELECT 1; INSERT ..." cannot bypass the gate.
|
# Inspect every statement so "SELECT 1; INSERT ..." cannot bypass the gate.
|
||||||
|
# Raft state machine is wired only to the default database (v1).
|
||||||
if raftNode != nil:
|
if raftNode != nil:
|
||||||
var hasWrite = false
|
var hasWrite = false
|
||||||
for stmt in astNode.stmts:
|
for stmt in astNode.stmts:
|
||||||
if isWrite(stmt):
|
if isWrite(stmt):
|
||||||
hasWrite = true
|
hasWrite = true
|
||||||
break
|
break
|
||||||
if hasWrite and raftNode.state != rsLeader:
|
if hasWrite:
|
||||||
let who = if raftNode.leaderId.len > 0: raftNode.leaderId else: "none elected"
|
let dbName = if ctx.currentDatabase.len > 0: ctx.currentDatabase else: "default"
|
||||||
return (false, QueryResult(), "not leader; leader is '" & who & "'")
|
if dbName != "default":
|
||||||
|
return (false, QueryResult(),
|
||||||
|
"raft writes only supported on the 'default' database; current is '" &
|
||||||
|
dbName & "'")
|
||||||
|
if raftNode.state != rsLeader:
|
||||||
|
let who = if raftNode.leaderId.len > 0: raftNode.leaderId else: "none elected"
|
||||||
|
return (false, QueryResult(), "not leader; leader is '" & who & "'")
|
||||||
|
|
||||||
let res = executor.executeQuery(ctx, astNode, params)
|
let res = executor.executeQuery(ctx, astNode, params)
|
||||||
if res.success:
|
if res.success:
|
||||||
|
|||||||
@@ -111,6 +111,40 @@ proc addEdgeWithId*(g: Graph, src, dst: NodeId, label: string = "",
|
|||||||
g.adjacency[src].add(AdjacencyEntry(edgeId: id, neighbor: dst, weight: weight, label: label))
|
g.adjacency[src].add(AdjacencyEntry(edgeId: id, neighbor: dst, weight: weight, label: label))
|
||||||
g.reverseAdj[dst].add(AdjacencyEntry(edgeId: id, neighbor: src, weight: weight, label: label))
|
g.reverseAdj[dst].add(AdjacencyEntry(edgeId: id, neighbor: src, weight: weight, label: label))
|
||||||
|
|
||||||
|
proc hasEdgeBetween*(g: Graph, src, dst: NodeId, label: string = ""): bool =
|
||||||
|
## True if an edge with the same endpoints and label already exists.
|
||||||
|
acquire(g.lock)
|
||||||
|
defer: release(g.lock)
|
||||||
|
for entry in g.adjacency.getOrDefault(src, @[]):
|
||||||
|
if entry.neighbor == dst and entry.label == label:
|
||||||
|
return true
|
||||||
|
return false
|
||||||
|
|
||||||
|
proc addEdgeWithIdIfAbsent*(g: Graph, src, dst: NodeId, label: string = "",
|
||||||
|
weight: float64 = 1.0) =
|
||||||
|
## Idempotent edge insert for raft apply / leader double-apply.
|
||||||
|
if hasEdgeBetween(g, src, dst, label):
|
||||||
|
return
|
||||||
|
addEdgeWithId(g, src, dst, label, weight)
|
||||||
|
|
||||||
|
proc removeEdgesBetween*(g: Graph, src, dst: NodeId, label: string = "") =
|
||||||
|
## Drop edges matching endpoints (and label if non-empty). Used by raft apply.
|
||||||
|
acquire(g.lock)
|
||||||
|
defer: release(g.lock)
|
||||||
|
if src notin g.adjacency: return
|
||||||
|
var keep: seq[AdjacencyEntry] = @[]
|
||||||
|
for entry in g.adjacency[src]:
|
||||||
|
if entry.neighbor == dst and (label.len == 0 or entry.label == label):
|
||||||
|
g.edges.del(entry.edgeId)
|
||||||
|
var newRev: seq[AdjacencyEntry] = @[]
|
||||||
|
for rev in g.reverseAdj.getOrDefault(dst, @[]):
|
||||||
|
if rev.edgeId != entry.edgeId:
|
||||||
|
newRev.add(rev)
|
||||||
|
g.reverseAdj[dst] = newRev
|
||||||
|
else:
|
||||||
|
keep.add(entry)
|
||||||
|
g.adjacency[src] = keep
|
||||||
|
|
||||||
proc getNode*(g: Graph, id: NodeId): GraphNode =
|
proc getNode*(g: Graph, id: NodeId): GraphNode =
|
||||||
acquire(g.lock)
|
acquire(g.lock)
|
||||||
defer: release(g.lock)
|
defer: release(g.lock)
|
||||||
|
|||||||
@@ -378,6 +378,26 @@ proc removeIndexesForRow(ctx: ExecutionContext, table: string, fullKey: string,
|
|||||||
for ftsKey, ftsIdx in ctx.ftsIndexes:
|
for ftsKey, ftsIdx in ctx.ftsIndexes:
|
||||||
if ftsKey.startsWith(table & "."):
|
if ftsKey.startsWith(table & "."):
|
||||||
ftsIdx.removeDocument(docId)
|
ftsIdx.removeDocument(docId)
|
||||||
|
# In-memory graphs: drop node/edge when the backing row is removed.
|
||||||
|
for graphName, graph in ctx.graphs:
|
||||||
|
if table == graphName & "_nodes":
|
||||||
|
if "id" in oldRow:
|
||||||
|
try:
|
||||||
|
let nid = gengine.NodeId(parseUInt(valueToString(oldRow["id"])))
|
||||||
|
gengine.removeNode(graph, nid)
|
||||||
|
except CatchableError:
|
||||||
|
discard
|
||||||
|
elif table == graphName & "_edges":
|
||||||
|
let srcStr = if "source_id" in oldRow: valueToString(oldRow["source_id"]) else: ""
|
||||||
|
let dstStr = if "dest_id" in oldRow: valueToString(oldRow["dest_id"]) else: ""
|
||||||
|
let label = if "edge_label" in oldRow: valueToString(oldRow["edge_label"]) else: ""
|
||||||
|
if srcStr.len > 0 and dstStr.len > 0:
|
||||||
|
try:
|
||||||
|
gengine.removeEdgesBetween(graph,
|
||||||
|
gengine.NodeId(parseUInt(srcStr)),
|
||||||
|
gengine.NodeId(parseUInt(dstStr)), label)
|
||||||
|
except CatchableError:
|
||||||
|
discard
|
||||||
|
|
||||||
proc insertIndexesForRow(ctx: ExecutionContext, table: string, fullKey: string,
|
proc insertIndexesForRow(ctx: ExecutionContext, table: string, fullKey: string,
|
||||||
valStr: string) =
|
valStr: string) =
|
||||||
@@ -416,9 +436,40 @@ proc insertIndexesForRow(ctx: ExecutionContext, table: string, fullKey: string,
|
|||||||
for col, val in newRow:
|
for col, val in newRow:
|
||||||
meta[col] = valueToString(val)
|
meta[col] = valueToString(val)
|
||||||
vengine.insert(vecIdx, docId, vec, meta)
|
vengine.insert(vecIdx, docId, vec, meta)
|
||||||
|
# In-memory graphs: mirror insert/update of backing node/edge tables.
|
||||||
|
for graphName, graph in ctx.graphs:
|
||||||
|
if table == graphName & "_nodes":
|
||||||
|
if "id" notin newRow: continue
|
||||||
|
try:
|
||||||
|
let nid = gengine.NodeId(parseUInt(valueToString(newRow["id"])))
|
||||||
|
var label = if "node_label" in newRow: valueToString(newRow["node_label"]) else: ""
|
||||||
|
var props = initTable[string, string]()
|
||||||
|
for col, val in newRow:
|
||||||
|
if col notin ["id", "node_label", "properties"]:
|
||||||
|
props[col] = valueToString(val)
|
||||||
|
# remove+add so property updates replace the in-memory node
|
||||||
|
gengine.removeNode(graph, nid)
|
||||||
|
gengine.addNodeWithId(graph, nid, label, props)
|
||||||
|
except CatchableError:
|
||||||
|
discard
|
||||||
|
elif table == graphName & "_edges":
|
||||||
|
let srcStr = if "source_id" in newRow: valueToString(newRow["source_id"]) else: ""
|
||||||
|
let dstStr = if "dest_id" in newRow: valueToString(newRow["dest_id"]) else: ""
|
||||||
|
let label = if "edge_label" in newRow: valueToString(newRow["edge_label"]) else: ""
|
||||||
|
var weight = 1.0
|
||||||
|
if "weight" in newRow:
|
||||||
|
try: weight = parseFloat(valueToString(newRow["weight"]))
|
||||||
|
except CatchableError: discard
|
||||||
|
if srcStr.len > 0 and dstStr.len > 0:
|
||||||
|
try:
|
||||||
|
gengine.addEdgeWithIdIfAbsent(graph,
|
||||||
|
gengine.NodeId(parseUInt(srcStr)),
|
||||||
|
gengine.NodeId(parseUInt(dstStr)), label, weight)
|
||||||
|
except CatchableError:
|
||||||
|
discard
|
||||||
|
|
||||||
proc applyReplicatedPut*(ctx: ExecutionContext, fullKey: string, value: seq[byte]) =
|
proc applyReplicatedPut*(ctx: ExecutionContext, fullKey: string, value: seq[byte]) =
|
||||||
## Apply a raft/replication put: LSM write + secondary B-tree/FTS/HNSW.
|
## Apply a raft/replication put: LSM write + secondary B-tree/FTS/HNSW/graphs.
|
||||||
## Idempotent on the leader (local DML already applied the same engines).
|
## Idempotent on the leader (local DML already applied the same engines).
|
||||||
let dot = fullKey.find('.')
|
let dot = fullKey.find('.')
|
||||||
let table = if dot > 0: fullKey[0..<dot] else: ""
|
let table = if dot > 0: fullKey[0..<dot] else: ""
|
||||||
|
|||||||
@@ -2744,6 +2744,36 @@ suite "Raft SQL Write Path":
|
|||||||
let (found, _) = db.get("t.id=1")
|
let (found, _) = db.get("t.id=1")
|
||||||
check not found
|
check not found
|
||||||
|
|
||||||
|
test "applyReplicatedPut updates in-memory graphs":
|
||||||
|
var testDir = getTempDir() / "baradb_raft_apply_g_" & $getCurrentProcessId() & "_" & $getMonoTime().ticks
|
||||||
|
createDir(testDir)
|
||||||
|
var db = newLSMTree(testDir)
|
||||||
|
var ctx = qexec.newExecutionContext(db)
|
||||||
|
discard qexec.executeQuery(ctx, parse("CREATE GRAPH g"))
|
||||||
|
check "g" in ctx.graphs
|
||||||
|
# Follower-style apply of node rows (no execInsert path).
|
||||||
|
applyReplicatedPut(ctx, "g_nodes.id=1",
|
||||||
|
cast[seq[byte]]("node_label=person,name=alice"))
|
||||||
|
applyReplicatedPut(ctx, "g_nodes.id=2",
|
||||||
|
cast[seq[byte]]("node_label=person,name=bob"))
|
||||||
|
applyReplicatedPut(ctx, "g_edges.source_id=1",
|
||||||
|
cast[seq[byte]]("source_id=1,dest_id=2,edge_label=knows,weight=1.0"))
|
||||||
|
check ctx.graphs["g"].getNode(gengine.NodeId(1)).label == "person"
|
||||||
|
check gengine.hasEdgeBetween(ctx.graphs["g"], gengine.NodeId(1),
|
||||||
|
gengine.NodeId(2), "knows")
|
||||||
|
# Second apply must not duplicate the edge (leader double-apply / re-apply).
|
||||||
|
applyReplicatedPut(ctx, "g_edges.source_id=1",
|
||||||
|
cast[seq[byte]]("source_id=1,dest_id=2,edge_label=knows,weight=1.0"))
|
||||||
|
check gengine.hasEdgeBetween(ctx.graphs["g"], gengine.NodeId(1),
|
||||||
|
gengine.NodeId(2), "knows")
|
||||||
|
applyReplicatedDelete(ctx, "g_nodes.id=1")
|
||||||
|
# Node gone; edges incident are cleaned by removeNode.
|
||||||
|
try:
|
||||||
|
discard ctx.graphs["g"].getNode(gengine.NodeId(1))
|
||||||
|
check false # should have raised
|
||||||
|
except CatchableError:
|
||||||
|
check true
|
||||||
|
|
||||||
suite "CLI Autocomplete":
|
suite "CLI Autocomplete":
|
||||||
test "Autocomplete commands":
|
test "Autocomplete commands":
|
||||||
let res = autocomplete("he")
|
let res = autocomplete("he")
|
||||||
|
|||||||
Reference in New Issue
Block a user