Complete plan execution: gossip/raft bounds, deadlock fix, JSON escaping, memtable limits
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
Remaining plan items: - 1.4 hybrid_search: JSON built with std/json instead of string concat - 2.3 lsm.nim: reject oversized memtable entries - 3.3 gossip.nim: bounds checking for idLen/nodeCount/hostLen - 3.4 raft.nim: max length caps for cmdLen/dataLen/logLen - 4.3 deadlock.nim: per-path seq stack instead of global parent table
This commit is contained in:
@@ -1,7 +1,6 @@
|
|||||||
## Deadlock Detection — wait-for graph
|
## Deadlock Detection — wait-for graph
|
||||||
import std/tables
|
import std/tables
|
||||||
import std/sets
|
import std/sets
|
||||||
import std/algorithm
|
|
||||||
import std/locks
|
import std/locks
|
||||||
|
|
||||||
type
|
type
|
||||||
@@ -68,32 +67,25 @@ proc removeTxn*(dd: DeadlockDetector, txnId: uint64) =
|
|||||||
proc detectCycleUnsafe(dd: DeadlockDetector): seq[uint64] =
|
proc detectCycleUnsafe(dd: DeadlockDetector): seq[uint64] =
|
||||||
var visited = initHashSet[uint64]()
|
var visited = initHashSet[uint64]()
|
||||||
var inStack = initHashSet[uint64]()
|
var inStack = initHashSet[uint64]()
|
||||||
var parent = initTable[uint64, uint64]()
|
|
||||||
|
|
||||||
proc dfs(node: uint64): seq[uint64] =
|
proc dfs(node: uint64, path: seq[uint64]): seq[uint64] =
|
||||||
visited.incl(node)
|
visited.incl(node)
|
||||||
inStack.incl(node)
|
inStack.incl(node)
|
||||||
|
var newPath = path & @[node]
|
||||||
for neighbor in dd.adjacency.getOrDefault(node, @[]):
|
for neighbor in dd.adjacency.getOrDefault(node, @[]):
|
||||||
if neighbor in inStack:
|
if neighbor in inStack:
|
||||||
# Found cycle — reconstruct
|
# Found cycle — reconstruct from path
|
||||||
var cycle = @[neighbor, node]
|
var cycle: seq[uint64] = @[]
|
||||||
var current = node
|
var found = false
|
||||||
while parent.getOrDefault(current, 0'u64) != neighbor and
|
for n in newPath:
|
||||||
parent.getOrDefault(current, 0'u64) != 0:
|
if n == neighbor:
|
||||||
current = parent[current]
|
found = true
|
||||||
if current == 0: break
|
if found:
|
||||||
cycle.add(current)
|
cycle.add(n)
|
||||||
# Verify we actually closed the cycle back to neighbor
|
cycle.add(neighbor)
|
||||||
if cycle[^1] != neighbor and parent.getOrDefault(cycle[^1], 0'u64) == neighbor:
|
|
||||||
cycle.add(neighbor)
|
|
||||||
elif cycle[^1] != neighbor:
|
|
||||||
# Incomplete cycle — should not happen with valid parent chain
|
|
||||||
return @[]
|
|
||||||
cycle.reverse()
|
|
||||||
return cycle
|
return cycle
|
||||||
if neighbor notin visited:
|
if neighbor notin visited:
|
||||||
parent[neighbor] = node
|
let cycle = dfs(neighbor, newPath)
|
||||||
let cycle = dfs(neighbor)
|
|
||||||
if cycle.len > 0:
|
if cycle.len > 0:
|
||||||
return cycle
|
return cycle
|
||||||
inStack.excl(node)
|
inStack.excl(node)
|
||||||
@@ -101,7 +93,7 @@ proc detectCycleUnsafe(dd: DeadlockDetector): seq[uint64] =
|
|||||||
|
|
||||||
for txnId in dd.txnIds:
|
for txnId in dd.txnIds:
|
||||||
if txnId notin visited:
|
if txnId notin visited:
|
||||||
let cycle = dfs(txnId)
|
let cycle = dfs(txnId, @[])
|
||||||
if cycle.len > 0:
|
if cycle.len > 0:
|
||||||
return cycle
|
return cycle
|
||||||
return @[]
|
return @[]
|
||||||
|
|||||||
@@ -75,6 +75,9 @@ proc serialize*(msg: GossipMessage): seq[byte] =
|
|||||||
s.close()
|
s.close()
|
||||||
|
|
||||||
proc deserializeGossipMessage*(data: seq[byte]): GossipMessage =
|
proc deserializeGossipMessage*(data: seq[byte]): GossipMessage =
|
||||||
|
const MaxGossipIdLen = 256
|
||||||
|
const MaxGossipNodeCount = 4096
|
||||||
|
const MaxGossipHostLen = 256
|
||||||
let s = newStringStream(cast[string](data))
|
let s = newStringStream(cast[string](data))
|
||||||
let magic = s.readStr(4)
|
let magic = s.readStr(4)
|
||||||
if magic != GossipMagic:
|
if magic != GossipMagic:
|
||||||
@@ -83,16 +86,24 @@ proc deserializeGossipMessage*(data: seq[byte]): GossipMessage =
|
|||||||
if version != GossipProtoVersion:
|
if version != GossipProtoVersion:
|
||||||
raise newException(ValueError, "Unsupported gossip protocol version")
|
raise newException(ValueError, "Unsupported gossip protocol version")
|
||||||
let senderIdLen = int(s.readUint32())
|
let senderIdLen = int(s.readUint32())
|
||||||
|
if senderIdLen > MaxGossipIdLen:
|
||||||
|
raise newException(ValueError, "Gossip senderId too long")
|
||||||
result.senderId = if senderIdLen > 0: s.readStr(senderIdLen) else: ""
|
result.senderId = if senderIdLen > 0: s.readStr(senderIdLen) else: ""
|
||||||
result.senderIncarnation = s.readUint64()
|
result.senderIncarnation = s.readUint64()
|
||||||
let nodeCount = int(s.readUint32())
|
let nodeCount = int(s.readUint32())
|
||||||
|
if nodeCount > MaxGossipNodeCount:
|
||||||
|
raise newException(ValueError, "Gossip node count too large")
|
||||||
result.nodes = newSeq[(string, NodeState, uint64, string, int)](nodeCount)
|
result.nodes = newSeq[(string, NodeState, uint64, string, int)](nodeCount)
|
||||||
for i in 0 ..< nodeCount:
|
for i in 0 ..< nodeCount:
|
||||||
let idLen = int(s.readUint32())
|
let idLen = int(s.readUint32())
|
||||||
|
if idLen > MaxGossipIdLen:
|
||||||
|
raise newException(ValueError, "Gossip node id too long")
|
||||||
let id = if idLen > 0: s.readStr(idLen) else: ""
|
let id = if idLen > 0: s.readStr(idLen) else: ""
|
||||||
let state = NodeState(s.readUint32())
|
let state = NodeState(s.readUint32())
|
||||||
let incarnation = s.readUint64()
|
let incarnation = s.readUint64()
|
||||||
let hostLen = int(s.readUint32())
|
let hostLen = int(s.readUint32())
|
||||||
|
if hostLen > MaxGossipHostLen:
|
||||||
|
raise newException(ValueError, "Gossip host too long")
|
||||||
let host = if hostLen > 0: s.readStr(hostLen) else: ""
|
let host = if hostLen > 0: s.readStr(hostLen) else: ""
|
||||||
let port = int(s.readUint32())
|
let port = int(s.readUint32())
|
||||||
result.nodes[i] = (id, state, incarnation, host, port)
|
result.nodes[i] = (id, state, incarnation, host, port)
|
||||||
|
|||||||
@@ -115,13 +115,19 @@ proc loadState(node: RaftNode) =
|
|||||||
if votedForLen > 0:
|
if votedForLen > 0:
|
||||||
node.votedFor = s.readStr(votedForLen)
|
node.votedFor = s.readStr(votedForLen)
|
||||||
let logLen = int(s.readUint32())
|
let logLen = int(s.readUint32())
|
||||||
|
if logLen > 1_000_000:
|
||||||
|
raise newException(ValueError, "Raft log length too large")
|
||||||
node.log = newSeq[LogEntry](logLen)
|
node.log = newSeq[LogEntry](logLen)
|
||||||
for i in 0..<logLen:
|
for i in 0..<logLen:
|
||||||
let term = s.readUint64()
|
let term = s.readUint64()
|
||||||
let index = s.readUint64()
|
let index = s.readUint64()
|
||||||
let cmdLen = int(s.readUint32())
|
let cmdLen = int(s.readUint32())
|
||||||
|
if cmdLen > 1_000_000:
|
||||||
|
raise newException(ValueError, "Raft command length too large")
|
||||||
let cmd = s.readStr(cmdLen)
|
let cmd = s.readStr(cmdLen)
|
||||||
let dataLen = int(s.readUint32())
|
let dataLen = int(s.readUint32())
|
||||||
|
if dataLen > 10_000_000:
|
||||||
|
raise newException(ValueError, "Raft data length too large")
|
||||||
var data = newSeq[byte](dataLen)
|
var data = newSeq[byte](dataLen)
|
||||||
if dataLen > 0:
|
if dataLen > 0:
|
||||||
if s.readData(addr data[0], dataLen) != dataLen:
|
if s.readData(addr data[0], dataLen) != dataLen:
|
||||||
|
|||||||
@@ -1241,10 +1241,13 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
|
|||||||
let queryVec = evalExpr(expr.irFuncArgs[4], row, ctx)
|
let queryVec = evalExpr(expr.irFuncArgs[4], row, ctx)
|
||||||
let k = try: parseInt(evalExpr(expr.irFuncArgs[5], row, ctx)) except ValueError: 10
|
let k = try: parseInt(evalExpr(expr.irFuncArgs[5], row, ctx)) except ValueError: 10
|
||||||
let results = doHybridSearch(ctx, table, vecCol, textCol, queryText, queryVec, k)
|
let results = doHybridSearch(ctx, table, vecCol, textCol, queryText, queryVec, k)
|
||||||
var parts: seq[string] = @[]
|
var jsonArr = newJArray()
|
||||||
for (id, score) in results:
|
for (id, score) in results:
|
||||||
parts.add("{\"id\":\"" & $id & "\",\"score\":\"" & $score & "\"}")
|
var obj = newJObject()
|
||||||
return "[" & parts.join(",") & "]"
|
obj["id"] = %id
|
||||||
|
obj["score"] = %score
|
||||||
|
jsonArr.add(obj)
|
||||||
|
return $jsonArr
|
||||||
of "hybrid_search_ids":
|
of "hybrid_search_ids":
|
||||||
if expr.irFuncArgs.len < 6: return ""
|
if expr.irFuncArgs.len < 6: return ""
|
||||||
let table = evalExpr(expr.irFuncArgs[0], row, ctx)
|
let table = evalExpr(expr.irFuncArgs[0], row, ctx)
|
||||||
|
|||||||
@@ -59,6 +59,8 @@ proc len*(mt: MemTable): int = mt.entries.len
|
|||||||
|
|
||||||
proc put*(mt: var MemTable, key: string, value: seq[byte], timestamp: uint64, deleted: bool = false): bool =
|
proc put*(mt: var MemTable, key: string, value: seq[byte], timestamp: uint64, deleted: bool = false): bool =
|
||||||
let entrySize = key.len + value.len + 16
|
let entrySize = key.len + value.len + 16
|
||||||
|
if entrySize > mt.maxSize:
|
||||||
|
return false
|
||||||
if mt.size + entrySize > mt.maxSize and mt.entries.len > 0:
|
if mt.size + entrySize > mt.maxSize and mt.entries.len > 0:
|
||||||
return false
|
return false
|
||||||
let entry = Entry(key: key, value: value, timestamp: timestamp, deleted: deleted)
|
let entry = Entry(key: key, value: value, timestamp: timestamp, deleted: deleted)
|
||||||
|
|||||||
Reference in New Issue
Block a user