logos-messaging/nim-sds
1
0
Fork 0
mirror of https://github.com/logos-messaging/nim-sds.git synced 2026-07-31 03:53:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: make Persistence interface async

Browse Source 
The 14 Persistence proc fields now return Future[...] with
{.async: (raises: []), gcsafe.}, allowing real I/O backends (SQLite,
encrypted file, network) to suspend rather than block the Chronos event
loop the manager runs on.

Propagates through:
- ReliabilityManager.lock: system.Lock -> chronos.AsyncLock. Acquired
  across awaits cleanly; matches the single-threaded Chronos worker the
  FFI uses. Multi-OS-thread use is now explicitly the caller's
  responsibility.
- sds_utils + sds.nim public API procs (wrapOutgoingMessage,
  unwrapReceivedMessage, markDependenciesMet, setCallbacks,
  resetReliabilityManager, cleanup, ensureChannel, removeChannel, the
  getter snapshots, etc.) are now async.
- FFI request handlers in library/sds_thread/... await the new API.
- Tests converted via an asyncTest template that wraps each test body
  in an async proc; setup/teardown use waitFor for their single async
  call (ensureChannel / cleanup).

Lock scope is preserved exactly: the same call sites that held the
kernel Lock today hold AsyncLock now -- no new locking added.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
NagyZoltanPeter 2026-05-20 12:08:29 +02:00
parent 2e9a7683f0
commit 965d67cb01
No known key found for this signature in database
GPG Key ID: 3E1F97CF4A7B6F42
10 changed files with 941 additions and 709 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
library/sds_thread/inter_thread_communication/requests/sds_dependencies_request.nim
View File

@ -43,7 +43,7 @@ proc process*(
let messageIdsC = self.messageIds.toSeq()
let messageIds = messageIdsC.mapIt($it)
markDependenciesMet(rm[], messageIds, $self.channelId).isOkOr:
(await markDependenciesMet(rm[], messageIds, $self.channelId)).isOkOr:
error "MARK_DEPENDENCIES_MET failed", error = error
return err("error processing MARK_DEPENDENCIES_MET request: " & $error)

4
library/sds_thread/inter_thread_communication/requests/sds_lifecycle_request.nim
View File

@ -41,7 +41,7 @@ proc createReliabilityManager(
error "Failed creating reliability manager", error = error
return err("Failed creating reliability manager: " & $error)
rm.setCallbacks(
await rm.setCallbacks(
appCallbacks.messageReadyCb, appCallbacks.messageSentCb,
appCallbacks.missingDependenciesCb, appCallbacks.periodicSyncCb,
appCallbacks.retrievalHintProvider, appCallbacks.repairReadyCb,
@ -61,7 +61,7 @@ proc process*(
error "CREATE_RELIABILITY_MANAGER failed", error = error
return err("error processing CREATE_RELIABILITY_MANAGER request: " & $error)
of RESET_RELIABILITY_MANAGER:
resetReliabilityManager(rm[]).isOkOr:
(await resetReliabilityManager(rm[])).isOkOr:
error "RESET_RELIABILITY_MANAGER failed", error = error
return err("error processing RESET_RELIABILITY_MANAGER request: " & $error)
of START_PERIODIC_TASKS:

8
library/sds_thread/inter_thread_communication/requests/sds_message_request.nim
View File

@ -53,7 +53,9 @@ proc process*(
of WRAP_MESSAGE:
let messageBytes = self.message.toSeq()
let wrappedMessage = wrapOutgoingMessage(rm[], messageBytes, $self.messageId, $self.channelId).valueOr:
let wrappedMessage = (
await wrapOutgoingMessage(rm[], messageBytes, $self.messageId, $self.channelId)
).valueOr:
error "WRAP_MESSAGE failed", error = error
return err("error processing WRAP_MESSAGE request: " & $error)
@ -62,7 +64,9 @@ proc process*(
of UNWRAP_MESSAGE:
let messageBytes = self.message.toSeq()
let (unwrappedMessage, missingDeps, extractedChannelId) = unwrapReceivedMessage(rm[], messageBytes).valueOr:
let (unwrappedMessage, missingDeps, extractedChannelId) = (
await unwrapReceivedMessage(rm[], messageBytes)
).valueOr:
return err("error processing UNWRAP_MESSAGE request: " & $error)
let res = SdsUnwrapResponse(message: unwrappedMessage, missingDeps: missingDeps, channelId: extractedChannelId)

487
sds.nim
View File

@ -1,4 +1,4 @@
import std/[algorithm, times, locks, tables, sets, options]
import std/[algorithm, times, tables, sets, options]
import chronos, results, chronicles
import sds/[types, protobuf, sds_utils, rolling_bloom_filter]
@ -33,165 +33,191 @@ proc isAcknowledged*(
return false
proc reviewAckStatus(rm: ReliabilityManager, msg: SdsMessage) {.gcsafe.} =
var rbf: Option[RollingBloomFilter]
if msg.bloomFilter.len > 0:
let bfResult = deserializeBloomFilter(msg.bloomFilter)
if bfResult.isOk():
let bf = bfResult.get()
rbf = some(
RollingBloomFilter.init(
filter = bf,
capacity = bf.capacity,
minCapacity = (bf.capacity.float * (100 - CapacityFlexPercent).float / 100.0).int,
maxCapacity = (bf.capacity.float * (100 + CapacityFlexPercent).float / 100.0).int,
proc reviewAckStatus(
rm: ReliabilityManager, msg: SdsMessage
): Future[void] {.async: (raises: []), gcsafe.} =
try:
var rbf: Option[RollingBloomFilter]
if msg.bloomFilter.len > 0:
let bfResult = deserializeBloomFilter(msg.bloomFilter)
if bfResult.isOk():
let bf = bfResult.get()
rbf = some(
RollingBloomFilter.init(
filter = bf,
capacity = bf.capacity,
minCapacity = (bf.capacity.float * (100 - CapacityFlexPercent).float / 100.0).int,
maxCapacity = (bf.capacity.float * (100 + CapacityFlexPercent).float / 100.0).int,
)
)
)
else:
error "Failed to deserialize bloom filter", error = bfResult.error
rbf = none[RollingBloomFilter]()
else:
error "Failed to deserialize bloom filter", error = bfResult.error
rbf = none[RollingBloomFilter]()
else:
rbf = none[RollingBloomFilter]()
if msg.channelId notin rm.channels:
return
if msg.channelId notin rm.channels:
return
let channel = rm.channels[msg.channelId]
var toDelete: seq[(int, SdsMessageID)] = @[]
var i = 0
let channel = rm.channels[msg.channelId]
var toDelete: seq[(int, SdsMessageID)] = @[]
var i = 0
while i < channel.outgoingBuffer.len:
let outMsg = channel.outgoingBuffer[i]
if outMsg.isAcknowledged(msg.causalHistory, rbf):
if not rm.onMessageSent.isNil():
rm.onMessageSent(outMsg.message.messageId, outMsg.message.channelId)
toDelete.add((i, outMsg.message.messageId))
inc i
while i < channel.outgoingBuffer.len:
let outMsg = channel.outgoingBuffer[i]
if outMsg.isAcknowledged(msg.causalHistory, rbf):
if not rm.onMessageSent.isNil():
{.cast(raises: []).}:
rm.onMessageSent(outMsg.message.messageId, outMsg.message.channelId)
toDelete.add((i, outMsg.message.messageId))
inc i
for k in countdown(toDelete.high, 0):
let (idx, ackedId) = toDelete[k]
channel.outgoingBuffer.delete(idx)
rm.persistence.removeOutgoing(msg.channelId, ackedId)
for k in countdown(toDelete.high, 0):
let (idx, ackedId) = toDelete[k]
channel.outgoingBuffer.delete(idx)
await rm.persistence.removeOutgoing(msg.channelId, ackedId)
except CatchableError:
error "Failed to review ack status", msg = getCurrentExceptionMsg()
proc wrapOutgoingMessage*(
rm: ReliabilityManager,
message: seq[byte],
messageId: SdsMessageID,
channelId: SdsChannelID,
): Result[seq[byte], ReliabilityError] =
): Future[Result[seq[byte], ReliabilityError]] {.async: (raises: []), gcsafe.} =
## Wraps an outgoing message with reliability metadata.
if message.len == 0:
return err(ReliabilityError.reInvalidArgument)
if message.len > MaxMessageSize:
return err(ReliabilityError.reMessageTooLarge)
withLock rm.lock:
try:
await rm.lock.acquire()
try:
let channel = rm.getOrCreateChannel(channelId)
rm.updateLamportTimestamp(getTime().toUnix, channelId)
try:
let channel = await rm.getOrCreateChannel(channelId)
await rm.updateLamportTimestamp(getTime().toUnix, channelId)
let bfResult = serializeBloomFilter(channel.bloomFilter.filter)
if bfResult.isErr:
error "Failed to serialize bloom filter", channelId = channelId
let bfResult = serializeBloomFilter(channel.bloomFilter.filter)
if bfResult.isErr:
error "Failed to serialize bloom filter", channelId = channelId
return err(ReliabilityError.reSerializationError)
# SDS-R: collect eligible expired repair requests to attach. Per
# spec (sds-r-send-message, RECOMMENDED), prioritise the entries with
# the smallest minTimeRepairReq — they are the most overdue and the
# ones the network most needs us to ask about.
var repairReqs: seq[HistoryEntry] = @[]
let now = getTime()
var expiredKeys: seq[SdsMessageID] = @[]
var eligible: seq[(SdsMessageID, OutgoingRepairEntry)] = @[]
for msgId, repairEntry in channel.outgoingRepairBuffer:
if now >= repairEntry.minTimeRepairReq:
eligible.add((msgId, repairEntry))
eligible.sort do(a, b: (SdsMessageID, OutgoingRepairEntry)) -> int:
cmp(a[1].minTimeRepairReq, b[1].minTimeRepairReq)
let take = min(eligible.len, rm.config.maxRepairRequests)
for i in 0 ..< take:
repairReqs.add(eligible[i][1].outHistEntry)
expiredKeys.add(eligible[i][0])
for key in expiredKeys:
channel.outgoingRepairBuffer.del(key)
await rm.persistence.removeOutgoingRepair(channelId, key)
let msg = SdsMessage.init(
messageId = messageId,
lamportTimestamp = channel.lamportTimestamp,
causalHistory =
await rm.getRecentHistoryEntries(rm.config.maxCausalHistory, channelId),
channelId = channelId,
content = message,
bloomFilter = bfResult.get(),
senderId = rm.participantId,
repairRequest = repairReqs,
)
let unackMsg = UnacknowledgedMessage.init(
message = msg, sendTime = getTime(), resendAttempts = 0
)
channel.outgoingBuffer.add(unackMsg)
await rm.persistence.saveOutgoing(channelId, unackMsg)
channel.bloomFilter.add(msg.messageId)
# The full SdsMessage carries senderId and content, so a single
# addToHistory replaces the old triple-write to messageHistory,
# messageCache, and messageSenders.
await rm.addToHistory(msg, channelId)
return serializeMessage(msg)
except CatchableError:
error "Failed to wrap message",
channelId = channelId, msg = getCurrentExceptionMsg()
return err(ReliabilityError.reSerializationError)
finally:
rm.lock.release()
except CatchableError:
error "Failed to wrap message (lock)",
channelId = channelId, msg = getCurrentExceptionMsg()
return err(ReliabilityError.reSerializationError)
# SDS-R: collect eligible expired repair requests to attach. Per
# spec (sds-r-send-message, RECOMMENDED), prioritise the entries with
# the smallest minTimeRepairReq — they are the most overdue and the
# ones the network most needs us to ask about.
var repairReqs: seq[HistoryEntry] = @[]
let now = getTime()
var expiredKeys: seq[SdsMessageID] = @[]
var eligible: seq[(SdsMessageID, OutgoingRepairEntry)] = @[]
for msgId, repairEntry in channel.outgoingRepairBuffer:
if now >= repairEntry.minTimeRepairReq:
eligible.add((msgId, repairEntry))
eligible.sort do(a, b: (SdsMessageID, OutgoingRepairEntry)) -> int:
cmp(a[1].minTimeRepairReq, b[1].minTimeRepairReq)
let take = min(eligible.len, rm.config.maxRepairRequests)
for i in 0 ..< take:
repairReqs.add(eligible[i][1].outHistEntry)
expiredKeys.add(eligible[i][0])
for key in expiredKeys:
channel.outgoingRepairBuffer.del(key)
rm.persistence.removeOutgoingRepair(channelId, key)
proc processIncomingBuffer(
rm: ReliabilityManager, channelId: SdsChannelID
): Future[void] {.async: (raises: []), gcsafe.} =
try:
await rm.lock.acquire()
try:
if channelId notin rm.channels:
error "Channel does not exist", channelId = channelId
return
let msg = SdsMessage.init(
messageId = messageId,
lamportTimestamp = channel.lamportTimestamp,
causalHistory = rm.getRecentHistoryEntries(rm.config.maxCausalHistory, channelId),
channelId = channelId,
content = message,
bloomFilter = bfResult.get(),
senderId = rm.participantId,
repairRequest = repairReqs,
)
let channel = rm.channels[channelId]
if channel.incomingBuffer.len == 0:
return
let unackMsg =
UnacknowledgedMessage.init(message = msg, sendTime = getTime(), resendAttempts = 0)
channel.outgoingBuffer.add(unackMsg)
rm.persistence.saveOutgoing(channelId, unackMsg)
var processed = initHashSet[SdsMessageID]()
var readyToProcess = newSeq[SdsMessageID]()
channel.bloomFilter.add(msg.messageId)
# The full SdsMessage carries senderId and content, so a single
# addToHistory replaces the old triple-write to messageHistory,
# messageCache, and messageSenders.
rm.addToHistory(msg, channelId)
for msgId, entry in channel.incomingBuffer:
if entry.missingDeps.len == 0:
readyToProcess.add(msgId)
return serializeMessage(msg)
except Exception:
error "Failed to wrap message",
channelId = channelId, msg = getCurrentExceptionMsg()
return err(ReliabilityError.reSerializationError)
while readyToProcess.len > 0:
let msgId = readyToProcess.pop()
if msgId in processed:
continue
proc processIncomingBuffer(rm: ReliabilityManager, channelId: SdsChannelID) {.gcsafe.} =
withLock rm.lock:
if channelId notin rm.channels:
error "Channel does not exist", channelId = channelId
return
if msgId in channel.incomingBuffer:
await rm.addToHistory(channel.incomingBuffer[msgId].message, channelId)
if not rm.onMessageReady.isNil():
{.cast(raises: []).}:
rm.onMessageReady(msgId, channelId)
processed.incl(msgId)
let channel = rm.channels[channelId]
if channel.incomingBuffer.len == 0:
return
for remainingId, entry in channel.incomingBuffer:
if remainingId notin processed:
if msgId in entry.missingDeps:
channel.incomingBuffer[remainingId].missingDeps.excl(msgId)
await rm.persistence.saveIncoming(
channelId, channel.incomingBuffer[remainingId]
)
if channel.incomingBuffer[remainingId].missingDeps.len == 0:
readyToProcess.add(remainingId)
var processed = initHashSet[SdsMessageID]()
var readyToProcess = newSeq[SdsMessageID]()
for msgId, entry in channel.incomingBuffer:
if entry.missingDeps.len == 0:
readyToProcess.add(msgId)
while readyToProcess.len > 0:
let msgId = readyToProcess.pop()
if msgId in processed:
continue
if msgId in channel.incomingBuffer:
rm.addToHistory(channel.incomingBuffer[msgId].message, channelId)
if not rm.onMessageReady.isNil():
rm.onMessageReady(msgId, channelId)
processed.incl(msgId)
for remainingId, entry in channel.incomingBuffer:
if remainingId notin processed:
if msgId in entry.missingDeps:
channel.incomingBuffer[remainingId].missingDeps.excl(msgId)
rm.persistence.saveIncoming(
channelId, channel.incomingBuffer[remainingId]
)
if channel.incomingBuffer[remainingId].missingDeps.len == 0:
readyToProcess.add(remainingId)
for msgId in processed:
channel.incomingBuffer.del(msgId)
rm.persistence.removeIncoming(channelId, msgId)
for msgId in processed:
channel.incomingBuffer.del(msgId)
await rm.persistence.removeIncoming(channelId, msgId)
finally:
rm.lock.release()
except CatchableError:
error "Failed to process incoming buffer",
channelId = channelId, msg = getCurrentExceptionMsg()
proc unwrapReceivedMessage*(
rm: ReliabilityManager, message: seq[byte]
): Result[
): Future[Result[
tuple[message: seq[byte], missingDeps: seq[HistoryEntry], channelId: SdsChannelID],
ReliabilityError,
] =
]] {.async: (raises: []), gcsafe.} =
## Unwraps a received message and processes its reliability metadata.
try:
let channelId = extractChannelId(message).valueOr:
@ -200,22 +226,22 @@ proc unwrapReceivedMessage*(
let msg = deserializeMessage(message).valueOr:
return err(ReliabilityError.reDeserializationError)
let channel = rm.getOrCreateChannel(channelId)
let channel = await rm.getOrCreateChannel(channelId)
# SDS-R: opportunistic repair-buffer cleanup — applies to duplicates too,
# so rebroadcasts cancel redundant responses on peers that already have the message.
channel.outgoingRepairBuffer.del(msg.messageId)
rm.persistence.removeOutgoingRepair(channelId, msg.messageId)
await rm.persistence.removeOutgoingRepair(channelId, msg.messageId)
channel.incomingRepairBuffer.del(msg.messageId)
rm.persistence.removeIncomingRepair(channelId, msg.messageId)
await rm.persistence.removeIncomingRepair(channelId, msg.messageId)
if msg.messageId in channel.messageHistory:
return ok((msg.content, @[], channelId))
channel.bloomFilter.add(msg.messageId)
rm.updateLamportTimestamp(msg.lamportTimestamp, channelId)
rm.reviewAckStatus(msg)
await rm.updateLamportTimestamp(msg.lamportTimestamp, channelId)
await rm.reviewAckStatus(msg)
# SDS-R: process incoming repair requests from this message. We can only
# answer for messages we have actually delivered (i.e. that live in
@ -225,7 +251,7 @@ proc unwrapReceivedMessage*(
for repairEntry in msg.repairRequest:
# Remove from our own outgoing repair buffer (someone else is also requesting)
channel.outgoingRepairBuffer.del(repairEntry.messageId)
rm.persistence.removeOutgoingRepair(channelId, repairEntry.messageId)
await rm.persistence.removeOutgoingRepair(channelId, repairEntry.messageId)
if repairEntry.messageId in channel.messageHistory and
rm.participantId.len > 0 and repairEntry.senderId.len > 0:
if isInResponseGroup(
@ -244,7 +270,9 @@ proc unwrapReceivedMessage*(
minTimeRepairResp: now + tResp,
)
channel.incomingRepairBuffer[repairEntry.messageId] = inEntry
rm.persistence.saveIncomingRepair(channelId, repairEntry.messageId, inEntry)
await rm.persistence.saveIncomingRepair(
channelId, repairEntry.messageId, inEntry
)
var missingDeps = rm.checkDependencies(msg.causalHistory, channelId)
@ -258,9 +286,9 @@ proc unwrapReceivedMessage*(
let entry =
IncomingMessage.init(message = msg, missingDeps = initHashSet[SdsMessageID]())
channel.incomingBuffer[msg.messageId] = entry
rm.persistence.saveIncoming(channelId, entry)
await rm.persistence.saveIncoming(channelId, entry)
else:
rm.addToHistory(msg, channelId)
await rm.addToHistory(msg, channelId)
# Unblock any buffered messages that were waiting on this one.
var unblocked: seq[SdsMessageID] = @[]
for pendingId, entry in channel.incomingBuffer:
@ -268,19 +296,21 @@ proc unwrapReceivedMessage*(
channel.incomingBuffer[pendingId].missingDeps.excl(msg.messageId)
unblocked.add(pendingId)
for pendingId in unblocked:
rm.persistence.saveIncoming(channelId, channel.incomingBuffer[pendingId])
rm.processIncomingBuffer(channelId)
await rm.persistence.saveIncoming(channelId, channel.incomingBuffer[pendingId])
await rm.processIncomingBuffer(channelId)
if not rm.onMessageReady.isNil():
rm.onMessageReady(msg.messageId, channelId)
{.cast(raises: []).}:
rm.onMessageReady(msg.messageId, channelId)
else:
let entry = IncomingMessage.init(
message = msg,
missingDeps = missingDeps.getMessageIds().toHashSet(),
)
channel.incomingBuffer[msg.messageId] = entry
rm.persistence.saveIncoming(channelId, entry)
await rm.persistence.saveIncoming(channelId, entry)
if not rm.onMissingDependencies.isNil():
rm.onMissingDependencies(msg.messageId, missingDeps, channelId)
{.cast(raises: []).}:
rm.onMissingDependencies(msg.messageId, missingDeps, channelId)
# SDS-R: add missing deps to outgoing repair buffer
if rm.participantId.len > 0:
@ -295,16 +325,16 @@ proc unwrapReceivedMessage*(
minTimeRepairReq: now + tReq,
)
channel.outgoingRepairBuffer[dep.messageId] = outEntry
rm.persistence.saveOutgoingRepair(channelId, dep.messageId, outEntry)
await rm.persistence.saveOutgoingRepair(channelId, dep.messageId, outEntry)
return ok((msg.content, missingDeps, channelId))
except Exception:
except CatchableError:
error "Failed to unwrap message", msg = getCurrentExceptionMsg()
return err(ReliabilityError.reDeserializationError)
proc markDependenciesMet*(
rm: ReliabilityManager, messageIds: seq[SdsMessageID], channelId: SdsChannelID
): Result[void, ReliabilityError] =
): Future[Result[void, ReliabilityError]] {.async: (raises: []), gcsafe.} =
## Marks the specified message dependencies as met.
try:
if channelId notin rm.channels:
@ -322,17 +352,17 @@ proc markDependenciesMet*(
channel.incomingBuffer[pendingId].missingDeps.excl(msgId)
unblocked.add(pendingId)
for pendingId in unblocked:
rm.persistence.saveIncoming(channelId, channel.incomingBuffer[pendingId])
await rm.persistence.saveIncoming(channelId, channel.incomingBuffer[pendingId])
# SDS-R: clear from repair buffers (dependency now met)
channel.outgoingRepairBuffer.del(msgId)
rm.persistence.removeOutgoingRepair(channelId, msgId)
await rm.persistence.removeOutgoingRepair(channelId, msgId)
channel.incomingRepairBuffer.del(msgId)
rm.persistence.removeIncomingRepair(channelId, msgId)
await rm.persistence.removeIncomingRepair(channelId, msgId)
rm.processIncomingBuffer(channelId)
await rm.processIncomingBuffer(channelId)
return ok()
except Exception:
except CatchableError:
error "Failed to mark dependencies as met",
channelId = channelId, msg = getCurrentExceptionMsg()
return err(ReliabilityError.reInternalError)
@ -345,45 +375,59 @@ proc setCallbacks*(
onPeriodicSync: PeriodicSyncCallback = nil,
onRetrievalHint: RetrievalHintProvider = nil,
onRepairReady: RepairReadyCallback = nil,
) =
): Future[void] {.async: (raises: []), gcsafe.} =
## Sets the callback functions for various events in the ReliabilityManager.
withLock rm.lock:
rm.onMessageReady = onMessageReady
rm.onMessageSent = onMessageSent
rm.onMissingDependencies = onMissingDependencies
rm.onPeriodicSync = onPeriodicSync
rm.onRetrievalHint = onRetrievalHint
rm.onRepairReady = onRepairReady
try:
await rm.lock.acquire()
try:
rm.onMessageReady = onMessageReady
rm.onMessageSent = onMessageSent
rm.onMissingDependencies = onMissingDependencies
rm.onPeriodicSync = onPeriodicSync
rm.onRetrievalHint = onRetrievalHint
rm.onRepairReady = onRepairReady
finally:
rm.lock.release()
except CatchableError:
error "Failed to set callbacks", msg = getCurrentExceptionMsg()
proc checkUnacknowledgedMessages(
rm: ReliabilityManager, channelId: SdsChannelID
) {.gcsafe.} =
withLock rm.lock:
if channelId notin rm.channels:
error "Channel does not exist", channelId = channelId
return
): Future[void] {.async: (raises: []), gcsafe.} =
try:
await rm.lock.acquire()
try:
if channelId notin rm.channels:
error "Channel does not exist", channelId = channelId
return
let channel = rm.channels[channelId]
let now = getTime()
var newOutgoingBuffer: seq[UnacknowledgedMessage] = @[]
let channel = rm.channels[channelId]
let now = getTime()
var newOutgoingBuffer: seq[UnacknowledgedMessage] = @[]
for unackMsg in channel.outgoingBuffer:
let elapsed = now - unackMsg.sendTime
if elapsed > rm.config.resendInterval:
if unackMsg.resendAttempts < rm.config.maxResendAttempts:
var updatedMsg = unackMsg
updatedMsg.resendAttempts += 1
updatedMsg.sendTime = now
newOutgoingBuffer.add(updatedMsg)
rm.persistence.saveOutgoing(channelId, updatedMsg)
for unackMsg in channel.outgoingBuffer:
let elapsed = now - unackMsg.sendTime
if elapsed > rm.config.resendInterval:
if unackMsg.resendAttempts < rm.config.maxResendAttempts:
var updatedMsg = unackMsg
updatedMsg.resendAttempts += 1
updatedMsg.sendTime = now
newOutgoingBuffer.add(updatedMsg)
await rm.persistence.saveOutgoing(channelId, updatedMsg)
else:
if not rm.onMessageSent.isNil():
{.cast(raises: []).}:
rm.onMessageSent(unackMsg.message.messageId, channelId)
await rm.persistence.removeOutgoing(channelId, unackMsg.message.messageId)
else:
if not rm.onMessageSent.isNil():
rm.onMessageSent(unackMsg.message.messageId, channelId)
rm.persistence.removeOutgoing(channelId, unackMsg.message.messageId)
else:
newOutgoingBuffer.add(unackMsg)
newOutgoingBuffer.add(unackMsg)
channel.outgoingBuffer = newOutgoingBuffer
channel.outgoingBuffer = newOutgoingBuffer
finally:
rm.lock.release()
except CatchableError:
error "Failed to check unacknowledged messages",
channelId = channelId, msg = getCurrentExceptionMsg()
proc periodicBufferSweep(
rm: ReliabilityManager
@ -391,13 +435,11 @@ proc periodicBufferSweep(
while true:
try:
for channelId, channel in rm.channels:
try:
rm.checkUnacknowledgedMessages(channelId)
rm.cleanBloomFilter(channelId)
except Exception:
error "Error in buffer sweep for channel",
channelId = channelId, msg = getCurrentExceptionMsg()
except Exception:
await rm.checkUnacknowledgedMessages(channelId)
await rm.cleanBloomFilter(channelId)
except CancelledError as e:
raise e
except CatchableError:
error "Error in periodic buffer sweep", msg = getCurrentExceptionMsg()
await sleepAsync(chronos.milliseconds(rm.config.bufferSweepInterval.inMilliseconds))
@ -408,19 +450,23 @@ proc periodicSyncMessage(
while true:
try:
if not rm.onPeriodicSync.isNil():
rm.onPeriodicSync()
except Exception:
{.cast(raises: []).}:
rm.onPeriodicSync()
except CatchableError:
error "Error in periodic sync", msg = getCurrentExceptionMsg()
await sleepAsync(chronos.seconds(rm.config.syncMessageInterval.inSeconds))
proc runRepairSweep*(rm: ReliabilityManager) {.gcsafe, raises: [].} =
proc runRepairSweep*(
rm: ReliabilityManager
): Future[void] {.async: (raises: []), gcsafe.} =
## SDS-R: Runs a single pass of the repair sweep.
## - Incoming: fires onRepairReady for expired T_resp entries and removes them
## - Outgoing: drops entries past T_max window
## Exposed so it can be driven directly in tests; also invoked by periodicRepairSweep.
## Acquires rm.lock so the repair buffers cannot be observed mid-mutation by
## a concurrent wrapOutgoingMessage / unwrapReceivedMessage on another thread.
withLock rm.lock:
## a concurrent wrapOutgoingMessage / unwrapReceivedMessage on another task.
try:
await rm.lock.acquire()
try:
let now = getTime()
for channelId, channel in rm.channels:
@ -434,9 +480,10 @@ proc runRepairSweep*(rm: ReliabilityManager) {.gcsafe, raises: [].} =
for msgId in toRebroadcast:
let entry = channel.incomingRepairBuffer[msgId]
channel.incomingRepairBuffer.del(msgId)
rm.persistence.removeIncomingRepair(channelId, msgId)
await rm.persistence.removeIncomingRepair(channelId, msgId)
if not rm.onRepairReady.isNil():
rm.onRepairReady(entry.cachedMessage, channelId)
{.cast(raises: []).}:
rm.onRepairReady(entry.cachedMessage, channelId)
# Drop expired outgoing repair entries past T_max
var toRemove: seq[SdsMessageID] = @[]
@ -446,19 +493,21 @@ proc runRepairSweep*(rm: ReliabilityManager) {.gcsafe, raises: [].} =
toRemove.add(msgId)
for msgId in toRemove:
channel.outgoingRepairBuffer.del(msgId)
rm.persistence.removeOutgoingRepair(channelId, msgId)
except Exception:
await rm.persistence.removeOutgoingRepair(channelId, msgId)
except CatchableError:
error "Error in repair sweep for channel",
channelId = channelId, msg = getCurrentExceptionMsg()
except Exception:
error "Error in repair sweep", msg = getCurrentExceptionMsg()
finally:
rm.lock.release()
except CatchableError:
error "Error in repair sweep", msg = getCurrentExceptionMsg()
proc periodicRepairSweep(
rm: ReliabilityManager
) {.async: (raises: [CancelledError]), gcsafe.} =
## SDS-R: Periodically checks repair buffers for expired entries.
while true:
rm.runRepairSweep()
await rm.runRepairSweep()
await sleepAsync(chronos.milliseconds(rm.config.repairSweepInterval.inMilliseconds))
proc startPeriodicTasks*(rm: ReliabilityManager) =
@ -467,22 +516,32 @@ proc startPeriodicTasks*(rm: ReliabilityManager) =
asyncSpawn rm.periodicSyncMessage()
asyncSpawn rm.periodicRepairSweep()
proc resetReliabilityManager*(rm: ReliabilityManager): Result[void, ReliabilityError] =
proc resetReliabilityManager*(
rm: ReliabilityManager
): Future[Result[void, ReliabilityError]] {.async: (raises: []), gcsafe.} =
## Resets the ReliabilityManager to its initial state.
withLock rm.lock:
try:
await rm.lock.acquire()
try:
for channelId, channel in rm.channels:
rm.dropChannelFromPersistence(channelId)
channel.lamportTimestamp = 0
channel.messageHistory.clear()
channel.outgoingBuffer.setLen(0)
channel.incomingBuffer.clear()
channel.outgoingRepairBuffer.clear()
channel.incomingRepairBuffer.clear()
channel.bloomFilter =
RollingBloomFilter.init(rm.config.bloomFilterCapacity, rm.config.bloomFilterErrorRate)
rm.channels.clear()
return ok()
except Exception:
error "Failed to reset ReliabilityManager", msg = getCurrentExceptionMsg()
return err(ReliabilityError.reInternalError)
try:
for channelId, channel in rm.channels:
await rm.dropChannelFromPersistence(channelId)
channel.lamportTimestamp = 0
channel.messageHistory.clear()
channel.outgoingBuffer.setLen(0)
channel.incomingBuffer.clear()
channel.outgoingRepairBuffer.clear()
channel.incomingRepairBuffer.clear()
channel.bloomFilter = RollingBloomFilter.init(
rm.config.bloomFilterCapacity, rm.config.bloomFilterErrorRate
)
rm.channels.clear()
return ok()
except CatchableError:
error "Failed to reset ReliabilityManager", msg = getCurrentExceptionMsg()
return err(ReliabilityError.reInternalError)
finally:
rm.lock.release()
except CatchableError:
error "Failed to reset ReliabilityManager (lock)", msg = getCurrentExceptionMsg()
return err(ReliabilityError.reInternalError)

201
sds/sds_utils.nim
View File

@ -1,5 +1,5 @@
import std/[times, locks, tables, sequtils, hashes]
import chronicles, results
import std/[times, tables, sequtils, hashes]
import chronos, chronicles, results
import ./rolling_bloom_filter
import ./types/[
sds_message_id, history_entry, sds_message, unacknowledged_message, incoming_message,
@ -16,44 +16,53 @@ proc defaultConfig*(): ReliabilityConfig =
proc dropChannelFromPersistence*(
rm: ReliabilityManager, channelId: SdsChannelID
) {.gcsafe, raises: [].} =
): Future[void] {.async: (raises: []), gcsafe.} =
## Wipes all persisted state for a channel via a single backend call.
## Called by removeChannel / resetReliabilityManager before they clear
## in-memory state. Backend executes the wipe in one transaction.
rm.persistence.dropChannel(channelId)
await rm.persistence.dropChannel(channelId)
proc cleanup*(rm: ReliabilityManager) {.raises: [].} =
proc cleanup*(
rm: ReliabilityManager
): Future[void] {.async: (raises: []), gcsafe.} =
## Releases in-memory state. Does NOT wipe persistence — the manager may be
## reconstructed against the same backend after cleanup, so disk state must
## survive. For deliberate disk wipe, use `removeChannel` or
## `resetReliabilityManager`.
if not rm.isNil():
if rm.isNil():
return
try:
await rm.lock.acquire()
try:
withLock rm.lock:
for channelId, channel in rm.channels:
channel.outgoingBuffer.setLen(0)
channel.incomingBuffer.clear()
channel.messageHistory.clear()
channel.outgoingRepairBuffer.clear()
channel.incomingRepairBuffer.clear()
rm.channels.clear()
except Exception:
error "Error during cleanup", error = getCurrentExceptionMsg()
for channelId, channel in rm.channels:
channel.outgoingBuffer.setLen(0)
channel.incomingBuffer.clear()
channel.messageHistory.clear()
channel.outgoingRepairBuffer.clear()
channel.incomingRepairBuffer.clear()
rm.channels.clear()
finally:
rm.lock.release()
except CatchableError:
error "Error during cleanup", error = getCurrentExceptionMsg()
proc cleanBloomFilter*(
rm: ReliabilityManager, channelId: SdsChannelID
) {.gcsafe, raises: [].} =
withLock rm.lock:
): Future[void] {.async: (raises: []), gcsafe.} =
try:
await rm.lock.acquire()
try:
if channelId in rm.channels:
rm.channels[channelId].bloomFilter.clean()
except Exception:
error "Failed to clean bloom filter",
error = getCurrentExceptionMsg(), channelId = channelId
finally:
rm.lock.release()
except CatchableError:
error "Failed to clean bloom filter",
error = getCurrentExceptionMsg(), channelId = channelId
proc addToHistory*(
rm: ReliabilityManager, msg: SdsMessage, channelId: SdsChannelID
) {.gcsafe, raises: [].} =
): Future[void] {.async: (raises: []), gcsafe.} =
## Inserts a delivered message into the channel's history map and evicts the
## eldest entries when the bound is exceeded. The full SdsMessage is kept so
## senderId is available for downstream causal-history population and the
@ -62,27 +71,27 @@ proc addToHistory*(
if channelId in rm.channels:
let channel = rm.channels[channelId]
channel.messageHistory[msg.messageId] = msg
rm.persistence.appendLogEntry(channelId, msg)
await rm.persistence.appendLogEntry(channelId, msg)
while channel.messageHistory.len > rm.config.maxMessageHistory:
var firstKey: SdsMessageID
for k in channel.messageHistory.keys:
firstKey = k
break
channel.messageHistory.del(firstKey)
rm.persistence.removeLogEntry(channelId, firstKey)
except Exception:
await rm.persistence.removeLogEntry(channelId, firstKey)
except CatchableError:
error "Failed to add to history",
channelId = channelId, msgId = msg.messageId, error = getCurrentExceptionMsg()
proc updateLamportTimestamp*(
rm: ReliabilityManager, msgTs: int64, channelId: SdsChannelID
) {.gcsafe, raises: [].} =
): Future[void] {.async: (raises: []), gcsafe.} =
try:
if channelId in rm.channels:
let channel = rm.channels[channelId]
channel.lamportTimestamp = max(msgTs, channel.lamportTimestamp) + 1
rm.persistence.saveLamport(channelId, channel.lamportTimestamp)
except Exception:
await rm.persistence.saveLamport(channelId, channel.lamportTimestamp)
except CatchableError:
error "Failed to update lamport timestamp",
channelId = channelId, msgTs = msgTs, error = getCurrentExceptionMsg()
@ -100,9 +109,9 @@ proc getMessageIds*(causalHistory: seq[HistoryEntry]): seq[SdsMessageID] =
proc computeTReq*(
participantId: SdsParticipantID,
messageId: SdsMessageID,
tMin: Duration,
tMax: Duration,
): Duration =
tMin: times.Duration,
tMax: times.Duration,
): times.Duration =
## Computes the repair request backoff duration per SDS-R spec:
## T_req = hash(participant_id, message_id) % (T_max - T_min) + T_min
let h = abs(hash(participantId.string & messageId))
@ -116,8 +125,8 @@ proc computeTResp*(
participantId: SdsParticipantID,
senderId: SdsParticipantID,
messageId: SdsMessageID,
tMax: Duration,
): Duration =
tMax: times.Duration,
): times.Duration =
## Computes the repair response backoff duration per SDS-R spec:
## distance = hash(participant_id) XOR hash(sender_id)
## T_resp = distance * hash(message_id) % T_max
@ -149,7 +158,7 @@ proc isInResponseGroup*(
proc getRecentHistoryEntries*(
rm: ReliabilityManager, n: int, channelId: SdsChannelID
): seq[HistoryEntry] =
): Future[seq[HistoryEntry]] {.async: (raises: []), gcsafe.} =
## Get recent history entries for sending in causal history.
## Populates retrieval hints and senderId (SDS-R) for each entry.
try:
@ -164,15 +173,16 @@ proc getRecentHistoryEntries*(
for msgId in recentMessageIds:
var entry = HistoryEntry(messageId: msgId)
if not rm.onRetrievalHint.isNil():
entry.retrievalHint = rm.onRetrievalHint(msgId)
{.cast(raises: []).}:
entry.retrievalHint = rm.onRetrievalHint(msgId)
if entry.retrievalHint.len > 0:
rm.persistence.setRetrievalHint(msgId, entry.retrievalHint)
await rm.persistence.setRetrievalHint(msgId, entry.retrievalHint)
entry.senderId = channel.messageHistory[msgId].senderId
entries.add(entry)
return entries
else:
return @[]
except Exception:
except CatchableError:
error "Failed to get recent history entries",
channelId = channelId, n = n, error = getCurrentExceptionMsg()
return @[]
@ -197,8 +207,9 @@ proc checkDependencies*(
proc getMessageHistory*(
rm: ReliabilityManager, channelId: SdsChannelID
): seq[SdsMessageID] =
withLock rm.lock:
): Future[seq[SdsMessageID]] {.async: (raises: []), gcsafe.} =
try:
await rm.lock.acquire()
try:
if channelId in rm.channels:
var ids: seq[SdsMessageID] = @[]
@ -207,42 +218,50 @@ proc getMessageHistory*(
return ids
else:
return @[]
except Exception:
error "Failed to get message history",
channelId = channelId, error = getCurrentExceptionMsg()
return @[]
finally:
rm.lock.release()
except CatchableError:
error "Failed to get message history",
channelId = channelId, error = getCurrentExceptionMsg()
return @[]
proc getOutgoingBuffer*(
rm: ReliabilityManager, channelId: SdsChannelID
): seq[UnacknowledgedMessage] =
withLock rm.lock:
): Future[seq[UnacknowledgedMessage]] {.async: (raises: []), gcsafe.} =
try:
await rm.lock.acquire()
try:
if channelId in rm.channels:
return rm.channels[channelId].outgoingBuffer
else:
return @[]
except Exception:
error "Failed to get outgoing buffer",
channelId = channelId, error = getCurrentExceptionMsg()
return @[]
finally:
rm.lock.release()
except CatchableError:
error "Failed to get outgoing buffer",
channelId = channelId, error = getCurrentExceptionMsg()
return @[]
proc getIncomingBuffer*(
rm: ReliabilityManager, channelId: SdsChannelID
): Table[SdsMessageID, IncomingMessage] =
withLock rm.lock:
): Future[Table[SdsMessageID, IncomingMessage]] {.async: (raises: []), gcsafe.} =
try:
await rm.lock.acquire()
try:
if channelId in rm.channels:
return rm.channels[channelId].incomingBuffer
else:
return initTable[SdsMessageID, IncomingMessage]()
except Exception:
error "Failed to get incoming buffer",
channelId = channelId, error = getCurrentExceptionMsg()
return initTable[SdsMessageID, IncomingMessage]()
finally:
rm.lock.release()
except CatchableError:
error "Failed to get incoming buffer",
channelId = channelId, error = getCurrentExceptionMsg()
return initTable[SdsMessageID, IncomingMessage]()
proc getOrCreateChannel*(
rm: ReliabilityManager, channelId: SdsChannelID
): ChannelContext =
): Future[ChannelContext] {.async: (raises: [CatchableError]), gcsafe.} =
## Returns the channel context, creating and bootstrapping it from the
## persistence backend if it does not yet exist in memory. The bloom filter
## is rebuilt deterministically from the loaded message history rather than
@ -252,7 +271,7 @@ proc getOrCreateChannel*(
let channel = ChannelContext.new(
RollingBloomFilter.init(rm.config.bloomFilterCapacity, rm.config.bloomFilterErrorRate)
)
let snapshot = rm.persistence.loadAllForChannel(channelId)
let snapshot = await rm.persistence.loadAllForChannel(channelId)
channel.lamportTimestamp = snapshot.lamportTimestamp
for msg in snapshot.messageHistory:
channel.messageHistory[msg.messageId] = msg
@ -267,39 +286,55 @@ proc getOrCreateChannel*(
channel.incomingRepairBuffer[msgId] = entry
rm.channels[channelId] = channel
return rm.channels[channelId]
except Exception:
except CatchableError as e:
error "Failed to get or create channel",
channelId = channelId, error = getCurrentExceptionMsg()
raise
raise e
proc ensureChannel*(
rm: ReliabilityManager, channelId: SdsChannelID
): Result[void, ReliabilityError] =
withLock rm.lock:
): Future[Result[void, ReliabilityError]] {.async: (raises: []), gcsafe.} =
try:
await rm.lock.acquire()
try:
discard rm.getOrCreateChannel(channelId)
return ok()
except Exception:
error "Failed to ensure channel",
channelId = channelId, msg = getCurrentExceptionMsg()
return err(ReliabilityError.reInternalError)
try:
discard await rm.getOrCreateChannel(channelId)
return ok()
except CatchableError:
error "Failed to ensure channel",
channelId = channelId, msg = getCurrentExceptionMsg()
return err(ReliabilityError.reInternalError)
finally:
rm.lock.release()
except CatchableError:
error "Failed to ensure channel (lock)",
channelId = channelId, msg = getCurrentExceptionMsg()
return err(ReliabilityError.reInternalError)
proc removeChannel*(
rm: ReliabilityManager, channelId: SdsChannelID
): Result[void, ReliabilityError] =
withLock rm.lock:
): Future[Result[void, ReliabilityError]] {.async: (raises: []), gcsafe.} =
try:
await rm.lock.acquire()
try:
if channelId in rm.channels:
let channel = rm.channels[channelId]
rm.dropChannelFromPersistence(channelId)
channel.outgoingBuffer.setLen(0)
channel.incomingBuffer.clear()
channel.messageHistory.clear()
channel.outgoingRepairBuffer.clear()
channel.incomingRepairBuffer.clear()
rm.channels.del(channelId)
return ok()
except Exception:
error "Failed to remove channel",
channelId = channelId, msg = getCurrentExceptionMsg()
return err(ReliabilityError.reInternalError)
try:
if channelId in rm.channels:
let channel = rm.channels[channelId]
await rm.dropChannelFromPersistence(channelId)
channel.outgoingBuffer.setLen(0)
channel.incomingBuffer.clear()
channel.messageHistory.clear()
channel.outgoingRepairBuffer.clear()
channel.incomingRepairBuffer.clear()
rm.channels.del(channelId)
return ok()
except CatchableError:
error "Failed to remove channel",
channelId = channelId, msg = getCurrentExceptionMsg()
return err(ReliabilityError.reInternalError)
finally:
rm.lock.release()
except CatchableError:
error "Failed to remove channel (lock)",
channelId = channelId, msg = getCurrentExceptionMsg()
return err(ReliabilityError.reInternalError)

117
sds/types/persistence.nim
View File

@ -1,3 +1,4 @@
import chronos
import ./sds_message_id
import ./sds_message
import ./unacknowledged_message
@ -14,6 +15,9 @@ export
## protocol calls into one of these procs immediately after the in-memory
## change, so on-disk state stays in lockstep with in-memory state.
##
## All proc fields are async (return `Future`) so backends can do real I/O
## without blocking the Chronos event loop the manager runs on.
##
## Bloom filter is intentionally not persisted: it is rebuilt from the local
## history log on bootstrap. Async timers are likewise recomputed from the
## absolute timestamps stored in the repair buffer entries.
@ -36,91 +40,136 @@ type
## Pluggable persistence contract. The caller supplies an instance of this
## type at `newReliabilityManager` construction time. Each proc field is
## invoked by nim-sds at the corresponding state-mutation point.
## All fields are async; nim-sds awaits each call to keep on-disk and
## in-memory state in lockstep without blocking the event loop.
# Per-channel lamport clock
saveLamport*:
proc(channelId: SdsChannelID, lamport: int64) {.gcsafe, raises: [].}
saveLamport*: proc(channelId: SdsChannelID, lamport: int64): Future[void] {.
async: (raises: []), gcsafe
.}
# Local log (delivered messages)
appendLogEntry*:
proc(channelId: SdsChannelID, msg: SdsMessage) {.gcsafe, raises: [].}
appendLogEntry*: proc(channelId: SdsChannelID, msg: SdsMessage): Future[void] {.
async: (raises: []), gcsafe
.}
removeLogEntry*:
proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].}
setRetrievalHint*:
proc(msgId: SdsMessageID, hint: seq[byte]) {.gcsafe, raises: [].}
proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {.
async: (raises: []), gcsafe
.}
setRetrievalHint*: proc(msgId: SdsMessageID, hint: seq[byte]): Future[void] {.
async: (raises: []), gcsafe
.}
# Outgoing unacknowledged buffer
saveOutgoing*:
proc(channelId: SdsChannelID, msg: UnacknowledgedMessage) {.gcsafe, raises: [].}
proc(channelId: SdsChannelID, msg: UnacknowledgedMessage): Future[void] {.
async: (raises: []), gcsafe
.}
removeOutgoing*:
proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].}
proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {.
async: (raises: []), gcsafe
.}
# Incoming dependency-waiting buffer
saveIncoming*:
proc(channelId: SdsChannelID, msg: IncomingMessage) {.gcsafe, raises: [].}
proc(channelId: SdsChannelID, msg: IncomingMessage): Future[void] {.
async: (raises: []), gcsafe
.}
removeIncoming*:
proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].}
proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {.
async: (raises: []), gcsafe
.}
# SDS-R outgoing repair buffer
saveOutgoingRepair*: proc(
channelId: SdsChannelID, msgId: SdsMessageID, entry: OutgoingRepairEntry
) {.gcsafe, raises: [].}
): Future[void] {.async: (raises: []), gcsafe.}
removeOutgoingRepair*:
proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].}
proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {.
async: (raises: []), gcsafe
.}
# SDS-R incoming repair buffer
saveIncomingRepair*: proc(
channelId: SdsChannelID, msgId: SdsMessageID, entry: IncomingRepairEntry
) {.gcsafe, raises: [].}
): Future[void] {.async: (raises: []), gcsafe.}
removeIncomingRepair*:
proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].}
proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {.
async: (raises: []), gcsafe
.}
# Wipe all persisted state for a channel in one transactional call.
# Called by removeChannel / resetReliabilityManager. Backends should
# implement this atomically (e.g. one BEGIN/COMMIT) — a per-row loop on
# the nim-sds side would mean N fsyncs per drop.
dropChannel*:
proc(channelId: SdsChannelID) {.gcsafe, raises: [].}
dropChannel*: proc(channelId: SdsChannelID): Future[void] {.
async: (raises: []), gcsafe
.}
# Bootstrap on `addChannel` / `getOrCreateChannel`.
loadAllForChannel*:
proc(channelId: SdsChannelID): ChannelSnapshot {.gcsafe, raises: [].}
loadAllForChannel*: proc(channelId: SdsChannelID): Future[ChannelSnapshot] {.
async: (raises: []), gcsafe
.}
proc noOpPersistence*(): Persistence =
## Default backend that discards every write and returns an empty snapshot.
## Used so existing callers (and tests) that don't care about durability
## keep working without supplying a real backend.
Persistence(
saveLamport: proc(channelId: SdsChannelID, lamport: int64) =
saveLamport: proc(
channelId: SdsChannelID, lamport: int64
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
appendLogEntry: proc(channelId: SdsChannelID, msg: SdsMessage) =
appendLogEntry: proc(
channelId: SdsChannelID, msg: SdsMessage
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
removeLogEntry: proc(channelId: SdsChannelID, msgId: SdsMessageID) =
removeLogEntry: proc(
channelId: SdsChannelID, msgId: SdsMessageID
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
setRetrievalHint: proc(msgId: SdsMessageID, hint: seq[byte]) =
setRetrievalHint: proc(
msgId: SdsMessageID, hint: seq[byte]
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
saveOutgoing: proc(channelId: SdsChannelID, msg: UnacknowledgedMessage) =
saveOutgoing: proc(
channelId: SdsChannelID, msg: UnacknowledgedMessage
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
removeOutgoing: proc(channelId: SdsChannelID, msgId: SdsMessageID) =
removeOutgoing: proc(
channelId: SdsChannelID, msgId: SdsMessageID
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
saveIncoming: proc(channelId: SdsChannelID, msg: IncomingMessage) =
saveIncoming: proc(
channelId: SdsChannelID, msg: IncomingMessage
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
removeIncoming: proc(channelId: SdsChannelID, msgId: SdsMessageID) =
removeIncoming: proc(
channelId: SdsChannelID, msgId: SdsMessageID
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
saveOutgoingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID, entry: OutgoingRepairEntry
) =
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
removeOutgoingRepair: proc(channelId: SdsChannelID, msgId: SdsMessageID) =
removeOutgoingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
saveIncomingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID, entry: IncomingRepairEntry
) =
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
removeIncomingRepair: proc(channelId: SdsChannelID, msgId: SdsMessageID) =
removeIncomingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
dropChannel: proc(channelId: SdsChannelID) =
dropChannel: proc(
channelId: SdsChannelID
): Future[void] {.async: (raises: []), gcsafe.} =
discard,
loadAllForChannel: proc(channelId: SdsChannelID): ChannelSnapshot =
ChannelSnapshot(),
loadAllForChannel: proc(
channelId: SdsChannelID
): Future[ChannelSnapshot] {.async: (raises: []), gcsafe.} =
return ChannelSnapshot(),
)

11
sds/types/reliability_manager.nim
View File

@ -1,4 +1,5 @@
import std/[tables, locks]
import std/tables
import chronos
import ./sds_message_id
import ./history_entry
import ./callbacks
@ -15,7 +16,11 @@ type ReliabilityManager* = ref object
participantId*: SdsParticipantID
persistence*: Persistence
## Pluggable durability backend; defaults to a no-op when not supplied.
lock*: Lock
lock*: AsyncLock
## Single-threaded Chronos cooperative lock. Serializes mutators against
## one another at await points; the manager assumes all calls come from
## the same Chronos event loop (the FFI worker thread). Multi-OS-thread
## use is the caller's responsibility.
onMessageReady*: proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.}
onMessageSent*: proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.}
onMissingDependencies*: proc(
@ -42,6 +47,6 @@ proc new*(
config: config,
participantId: participantId,
persistence: persistence,
lock: newAsyncLock(),
)
rm.lock.initLock()
return rm

55
tests/in_memory_persistence.nim
View File

@ -1,4 +1,5 @@
import std/tables
import chronos
import sds
## Test-only Persistence backend backed by Nim tables. Lets tests verify the
@ -23,76 +24,98 @@ proc newInMemoryStore*(): InMemoryStore =
proc newInMemoryPersistence*(store: InMemoryStore): Persistence =
Persistence(
saveLamport: proc(channelId: SdsChannelID, lamport: int64) {.gcsafe, raises: [].} =
saveLamport: proc(
channelId: SdsChannelID, lamport: int64
): Future[void] {.async: (raises: []), gcsafe.} =
store.lamports[channelId] = lamport,
appendLogEntry: proc(channelId: SdsChannelID, msg: SdsMessage) {.gcsafe, raises: [].} =
appendLogEntry: proc(
channelId: SdsChannelID, msg: SdsMessage
): Future[void] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
if channelId notin store.log:
store.log[channelId] = initOrderedTable[SdsMessageID, SdsMessage]()
store.log[channelId][msg.messageId] = msg,
removeLogEntry: proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} =
removeLogEntry: proc(
channelId: SdsChannelID, msgId: SdsMessageID
): Future[void] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
if channelId in store.log:
store.log[channelId].del(msgId),
setRetrievalHint: proc(msgId: SdsMessageID, hint: seq[byte]) {.gcsafe, raises: [].} =
setRetrievalHint: proc(
msgId: SdsMessageID, hint: seq[byte]
): Future[void] {.async: (raises: []), gcsafe.} =
store.hints[msgId] = hint,
saveOutgoing: proc(channelId: SdsChannelID, msg: UnacknowledgedMessage) {.gcsafe, raises: [].} =
saveOutgoing: proc(
channelId: SdsChannelID, msg: UnacknowledgedMessage
): Future[void] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
if channelId notin store.outgoing:
store.outgoing[channelId] =
initOrderedTable[SdsMessageID, UnacknowledgedMessage]()
store.outgoing[channelId][msg.message.messageId] = msg,
removeOutgoing: proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} =
removeOutgoing: proc(
channelId: SdsChannelID, msgId: SdsMessageID
): Future[void] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
if channelId in store.outgoing:
store.outgoing[channelId].del(msgId),
saveIncoming: proc(channelId: SdsChannelID, msg: IncomingMessage) {.gcsafe, raises: [].} =
saveIncoming: proc(
channelId: SdsChannelID, msg: IncomingMessage
): Future[void] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
if channelId notin store.incoming:
store.incoming[channelId] =
initOrderedTable[SdsMessageID, IncomingMessage]()
store.incoming[channelId][msg.message.messageId] = msg,
removeIncoming: proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} =
removeIncoming: proc(
channelId: SdsChannelID, msgId: SdsMessageID
): Future[void] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
if channelId in store.incoming:
store.incoming[channelId].del(msgId),
saveOutgoingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID, entry: OutgoingRepairEntry
) {.gcsafe, raises: [].} =
): Future[void] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
if channelId notin store.outgoingRepair:
store.outgoingRepair[channelId] =
initOrderedTable[SdsMessageID, OutgoingRepairEntry]()
store.outgoingRepair[channelId][msgId] = entry,
removeOutgoingRepair: proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} =
removeOutgoingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID
): Future[void] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
if channelId in store.outgoingRepair:
store.outgoingRepair[channelId].del(msgId),
saveIncomingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID, entry: IncomingRepairEntry
) {.gcsafe, raises: [].} =
): Future[void] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
if channelId notin store.incomingRepair:
store.incomingRepair[channelId] =
initOrderedTable[SdsMessageID, IncomingRepairEntry]()
store.incomingRepair[channelId][msgId] = entry,
removeIncomingRepair: proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} =
removeIncomingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID
): Future[void] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
if channelId in store.incomingRepair:
store.incomingRepair[channelId].del(msgId),
dropChannel: proc(channelId: SdsChannelID) {.gcsafe, raises: [].} =
dropChannel: proc(
channelId: SdsChannelID
): Future[void] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
store.lamports.del(channelId)
store.log.del(channelId)
@ -103,7 +126,9 @@ proc newInMemoryPersistence*(store: InMemoryStore): Persistence =
store.dropChannelCalls[channelId] =
store.dropChannelCalls.getOrDefault(channelId) + 1,
loadAllForChannel: proc(channelId: SdsChannelID): ChannelSnapshot {.gcsafe, raises: [].} =
loadAllForChannel: proc(
channelId: SdsChannelID
): Future[ChannelSnapshot] {.async: (raises: []), gcsafe.} =
{.cast(raises: []).}:
var snap = ChannelSnapshot()
if channelId in store.lamports:
@ -123,5 +148,5 @@ proc newInMemoryPersistence*(store: InMemoryStore): Persistence =
if channelId in store.incomingRepair:
for msgId, entry in store.incomingRepair[channelId]:
snap.incomingRepairBuffer.add((msgId, entry))
snap,
return snap,
)

237
tests/test_persistence.nim
View File

@ -1,4 +1,4 @@
import unittest, results, std/[tables, sets, times]
import unittest, results, chronos, std/[tables, sets, times]
import sds
import ./in_memory_persistence
@ -6,46 +6,63 @@ converter toParticipantID(s: string): SdsParticipantID = s.SdsParticipantID
const testChannel = "testChannel"
template asyncTest(name: string, body: untyped) =
## Wraps a unittest `test` body in an async proc and runs it to completion.
## Tests can now `await` rm.* and rm.persistence.* calls directly.
## unittest's `check` raises Exception (wider than chronos's CatchableError),
## so catch inside the async body and re-raise after waitFor.
test name:
var asyncTestErr {.inject.}: ref Exception = nil
proc inner() {.async.} =
{.cast(gcsafe).}:
try:
body
except Exception as e:
asyncTestErr = e
waitFor inner()
if asyncTestErr != nil:
raise asyncTestErr
suite "Persistence: write → restart → read-back":
test "outgoing buffer survives restart":
asyncTest "outgoing buffer survives restart":
let store = newInMemoryStore()
let p1 = newInMemoryPersistence(store)
let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get()
check rm1.ensureChannel(testChannel).isOk()
let wrapped = rm1.wrapOutgoingMessage(@[1.byte, 2, 3], "msg-1", testChannel)
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
check (await rm1.ensureChannel(testChannel)).isOk()
let wrapped = await rm1.wrapOutgoingMessage(@[1.byte, 2, 3], "msg-1", testChannel)
check wrapped.isOk()
check store.outgoing[testChannel].len == 1
check "msg-1" in store.outgoing[testChannel]
rm1.cleanup()
await rm1.cleanup()
# Simulate restart: fresh manager, same backend
let p2 = newInMemoryPersistence(store)
let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get()
check rm2.ensureChannel(testChannel).isOk()
let buf = rm2.getOutgoingBuffer(testChannel)
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
check (await rm2.ensureChannel(testChannel)).isOk()
let buf = await rm2.getOutgoingBuffer(testChannel)
check buf.len == 1
check buf[0].message.messageId == "msg-1"
rm2.cleanup()
await rm2.cleanup()
test "lamport clock survives restart":
asyncTest "lamport clock survives restart":
let store = newInMemoryStore()
let p1 = newInMemoryPersistence(store)
let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get()
check rm1.ensureChannel(testChannel).isOk()
rm1.updateLamportTimestamp(42, testChannel)
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
check (await rm1.ensureChannel(testChannel)).isOk()
await rm1.updateLamportTimestamp(42, testChannel)
check store.lamports[testChannel] == 43 # max(42, 0) + 1
rm1.cleanup()
await rm1.cleanup()
let p2 = newInMemoryPersistence(store)
let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get()
check rm2.ensureChannel(testChannel).isOk()
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
check (await rm2.ensureChannel(testChannel)).isOk()
check rm2.channels[testChannel].lamportTimestamp == 43
test "delivered messages survive restart and rebuild bloom":
asyncTest "delivered messages survive restart and rebuild bloom":
let store = newInMemoryStore()
let p1 = newInMemoryPersistence(store)
let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get()
check rm1.ensureChannel(testChannel).isOk()
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
check (await rm1.ensureChannel(testChannel)).isOk()
let msg = SdsMessage.init(
messageId = "delivered-1",
lamportTimestamp = 1,
@ -55,25 +72,25 @@ suite "Persistence: write → restart → read-back":
bloomFilter = @[],
senderId = "alice",
)
rm1.addToHistory(msg, testChannel)
await rm1.addToHistory(msg, testChannel)
check store.log[testChannel].len == 1
rm1.cleanup()
await rm1.cleanup()
let p2 = newInMemoryPersistence(store)
let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get()
check rm2.ensureChannel(testChannel).isOk()
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
check (await rm2.ensureChannel(testChannel)).isOk()
let ch = rm2.channels[testChannel]
check ch.messageHistory.len == 1
check "delivered-1" in ch.messageHistory
# Bloom filter rebuilt from log on bootstrap
check ch.bloomFilter.contains("delivered-1")
test "ack removes outgoing entry from persistence":
asyncTest "ack removes outgoing entry from persistence":
let store = newInMemoryStore()
let p = newInMemoryPersistence(store)
let rm = newReliabilityManager(participantId = "alice", persistence =p).get()
check rm.ensureChannel(testChannel).isOk()
discard rm.wrapOutgoingMessage(@[1.byte], "msg-x", testChannel)
let rm = newReliabilityManager(participantId = "alice", persistence = p).get()
check (await rm.ensureChannel(testChannel)).isOk()
discard await rm.wrapOutgoingMessage(@[1.byte], "msg-x", testChannel)
check "msg-x" in store.outgoing[testChannel]
# Synthesize an incoming message that ACKs msg-x via causal history
@ -87,22 +104,22 @@ suite "Persistence: write → restart → read-back":
senderId = "bob",
)
let serialized = serializeMessage(ackMsg).get()
discard rm.unwrapReceivedMessage(serialized)
discard await rm.unwrapReceivedMessage(serialized)
check "msg-x" notin store.outgoing[testChannel]
rm.cleanup()
await rm.cleanup()
test "removeChannel issues exactly one dropChannel call and wipes all state":
asyncTest "removeChannel issues exactly one dropChannel call and wipes all state":
# Regression for PR #66 review: removal must be a single transactional
# drop, not N per-row removes — otherwise SQLite eats N fsyncs per drop.
let store = newInMemoryStore()
let p = newInMemoryPersistence(store)
let rm = newReliabilityManager(participantId = "alice", persistence =p).get()
check rm.ensureChannel(testChannel).isOk()
discard rm.wrapOutgoingMessage(@[1.byte], "msg-r", testChannel)
let rm = newReliabilityManager(participantId = "alice", persistence = p).get()
check (await rm.ensureChannel(testChannel)).isOk()
discard await rm.wrapOutgoingMessage(@[1.byte], "msg-r", testChannel)
check store.outgoing[testChannel].len == 1
check store.lamports[testChannel] > 0
check rm.removeChannel(testChannel).isOk()
check (await rm.removeChannel(testChannel)).isOk()
check store.dropChannelCalls.getOrDefault(testChannel) == 1
check testChannel notin store.outgoing
check testChannel notin store.lamports
@ -110,50 +127,52 @@ suite "Persistence: write → restart → read-back":
check testChannel notin store.incoming
check testChannel notin store.outgoingRepair
check testChannel notin store.incomingRepair
rm.cleanup()
await rm.cleanup()
test "noOpPersistence keeps existing manager working":
asyncTest "noOpPersistence keeps existing manager working":
let rm = newReliabilityManager(participantId = "alice").get() # default no-op persistence
check rm.ensureChannel(testChannel).isOk()
let wrapped = rm.wrapOutgoingMessage(@[1.byte], "msg-n", testChannel)
check (await rm.ensureChannel(testChannel)).isOk()
let wrapped = await rm.wrapOutgoingMessage(@[1.byte], "msg-n", testChannel)
check wrapped.isOk()
check rm.getOutgoingBuffer(testChannel).len == 1
rm.cleanup()
let buf = await rm.getOutgoingBuffer(testChannel)
check buf.len == 1
await rm.cleanup()
test "continue operating after restart: lamport stays monotonic":
asyncTest "continue operating after restart: lamport stays monotonic":
let store = newInMemoryStore()
let p1 = newInMemoryPersistence(store)
let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get()
check rm1.ensureChannel(testChannel).isOk()
discard rm1.wrapOutgoingMessage(@[1.byte], "m1", testChannel)
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
check (await rm1.ensureChannel(testChannel)).isOk()
discard await rm1.wrapOutgoingMessage(@[1.byte], "m1", testChannel)
let lamportAfterSession1 = store.lamports[testChannel]
check lamportAfterSession1 > 0
rm1.cleanup()
await rm1.cleanup()
# Restart and send another message — lamport must not regress.
let p2 = newInMemoryPersistence(store)
let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get()
check rm2.ensureChannel(testChannel).isOk()
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
check (await rm2.ensureChannel(testChannel)).isOk()
check rm2.channels[testChannel].lamportTimestamp == lamportAfterSession1
discard rm2.wrapOutgoingMessage(@[2.byte], "m2", testChannel)
discard await rm2.wrapOutgoingMessage(@[2.byte], "m2", testChannel)
check store.lamports[testChannel] > lamportAfterSession1
check rm2.getOutgoingBuffer(testChannel).len == 2
rm2.cleanup()
let buf = await rm2.getOutgoingBuffer(testChannel)
check buf.len == 2
await rm2.cleanup()
test "multiple restart cycles preserve state":
asyncTest "multiple restart cycles preserve state":
let store = newInMemoryStore()
for i in 1 .. 3:
let p = newInMemoryPersistence(store)
let rm = newReliabilityManager(participantId = "alice", persistence =p).get()
check rm.ensureChannel(testChannel).isOk()
discard rm.wrapOutgoingMessage(@[byte(i)], "m" & $i, testChannel)
rm.cleanup()
let rm = newReliabilityManager(participantId = "alice", persistence = p).get()
check (await rm.ensureChannel(testChannel)).isOk()
discard await rm.wrapOutgoingMessage(@[byte(i)], "m" & $i, testChannel)
await rm.cleanup()
# Final session: all three messages must be in the buffer.
let pFinal = newInMemoryPersistence(store)
let rmFinal = newReliabilityManager(participantId = "alice", persistence =pFinal).get()
check rmFinal.ensureChannel(testChannel).isOk()
let buf = rmFinal.getOutgoingBuffer(testChannel)
let rmFinal = newReliabilityManager(participantId = "alice", persistence = pFinal).get()
check (await rmFinal.ensureChannel(testChannel)).isOk()
let buf = await rmFinal.getOutgoingBuffer(testChannel)
check buf.len == 3
var ids = newSeq[string]()
for unack in buf:
@ -161,13 +180,13 @@ suite "Persistence: write → restart → read-back":
check "m1" in ids
check "m2" in ids
check "m3" in ids
rmFinal.cleanup()
await rmFinal.cleanup()
test "incoming dep-waiting buffer survives restart with missingDeps intact":
asyncTest "incoming dep-waiting buffer survives restart with missingDeps intact":
let store = newInMemoryStore()
let p1 = newInMemoryPersistence(store)
let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get()
check rm1.ensureChannel(testChannel).isOk()
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
check (await rm1.ensureChannel(testChannel)).isOk()
# Receive a message whose causal-history references an unknown predecessor.
let depMsg = SdsMessage.init(
@ -180,47 +199,48 @@ suite "Persistence: write → restart → read-back":
senderId = "carol",
)
let serialized = serializeMessage(depMsg).get()
discard rm1.unwrapReceivedMessage(serialized)
discard await rm1.unwrapReceivedMessage(serialized)
check "msg-with-deps" in store.incoming[testChannel]
rm1.cleanup()
await rm1.cleanup()
# Restart — buffered message and its missing-deps set must be back.
let p2 = newInMemoryPersistence(store)
let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get()
check rm2.ensureChannel(testChannel).isOk()
let inbuf = rm2.getIncomingBuffer(testChannel)
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
check (await rm2.ensureChannel(testChannel)).isOk()
let inbuf = await rm2.getIncomingBuffer(testChannel)
check "msg-with-deps" in inbuf
check "missing-dep" in inbuf["msg-with-deps"].missingDeps
rm2.cleanup()
await rm2.cleanup()
test "removeChannel + recreate does not inherit stale lamport":
asyncTest "removeChannel + recreate does not inherit stale lamport":
# Regression: dropChannel must wipe the lamport row; otherwise a recreate
# of the same channelId after restart picks up the old timestamp.
let store = newInMemoryStore()
let p1 = newInMemoryPersistence(store)
let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get()
check rm1.ensureChannel(testChannel).isOk()
discard rm1.wrapOutgoingMessage(@[1.byte], "m-old", testChannel)
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
check (await rm1.ensureChannel(testChannel)).isOk()
discard await rm1.wrapOutgoingMessage(@[1.byte], "m-old", testChannel)
check store.lamports[testChannel] > 0
check rm1.removeChannel(testChannel).isOk()
check (await rm1.removeChannel(testChannel)).isOk()
check testChannel notin store.lamports
rm1.cleanup()
await rm1.cleanup()
# Recreate the same channelId after a restart — must start fresh.
let p2 = newInMemoryPersistence(store)
let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get()
check rm2.ensureChannel(testChannel).isOk()
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
check (await rm2.ensureChannel(testChannel)).isOk()
check rm2.channels[testChannel].lamportTimestamp == 0
check rm2.getOutgoingBuffer(testChannel).len == 0
rm2.cleanup()
let buf = await rm2.getOutgoingBuffer(testChannel)
check buf.len == 0
await rm2.cleanup()
test "SDS-R outgoing repair buffer survives restart with absolute t_req_at":
asyncTest "SDS-R outgoing repair buffer survives restart with absolute t_req_at":
let store = newInMemoryStore()
let p1 = newInMemoryPersistence(store)
let rm1 = newReliabilityManager(
participantId = "alice", persistence = p1
).get()
check rm1.ensureChannel(testChannel).isOk()
check (await rm1.ensureChannel(testChannel)).isOk()
# Receive a message that references an unknown dep — triggers SDS-R repair.
let depMsg = SdsMessage.init(
@ -232,32 +252,34 @@ suite "Persistence: write → restart → read-back":
bloomFilter = @[],
senderId = "bob",
)
discard rm1.unwrapReceivedMessage(serializeMessage(depMsg).get())
discard await rm1.unwrapReceivedMessage(serializeMessage(depMsg).get())
check "missing-dep" in store.outgoingRepair[testChannel]
let originalTReqAt = store.outgoingRepair[testChannel]["missing-dep"].minTimeRepairReq
check originalTReqAt.toUnix > 0
rm1.cleanup()
await rm1.cleanup()
# Restart — repair entry must be back with the SAME absolute time, not "now".
let p2 = newInMemoryPersistence(store)
let rm2 = newReliabilityManager(
participantId = "alice", persistence = p2
).get()
check rm2.ensureChannel(testChannel).isOk()
check (await rm2.ensureChannel(testChannel)).isOk()
let buf = rm2.channels[testChannel].outgoingRepairBuffer
check "missing-dep" in buf
check buf["missing-dep"].minTimeRepairReq == originalTReqAt
rm2.cleanup()
await rm2.cleanup()
test "FIFO eviction state survives restart":
asyncTest "FIFO eviction state survives restart":
let store = newInMemoryStore()
var smallCfg = defaultConfig()
smallCfg.maxMessageHistory = 3
smallCfg.bloomFilterCapacity = 3
let p1 = newInMemoryPersistence(store)
let rm1 = newReliabilityManager(participantId = "alice", config = smallCfg, persistence =p1).get()
check rm1.ensureChannel(testChannel).isOk()
let rm1 = newReliabilityManager(
participantId = "alice", config = smallCfg, persistence = p1
).get()
check (await rm1.ensureChannel(testChannel)).isOk()
# Add 5 delivered messages — first 2 should be evicted by FIFO.
for i in 1 .. 5:
let m = SdsMessage.init(
@ -269,16 +291,18 @@ suite "Persistence: write → restart → read-back":
bloomFilter = @[],
senderId = "alice",
)
rm1.addToHistory(m, testChannel)
await rm1.addToHistory(m, testChannel)
check store.log[testChannel].len == 3
check "m1" notin store.log[testChannel]
check "m2" notin store.log[testChannel]
rm1.cleanup()
await rm1.cleanup()
# Restart — evicted entries must NOT come back; survivors keep order.
let p2 = newInMemoryPersistence(store)
let rm2 = newReliabilityManager(participantId = "alice", config = smallCfg, persistence =p2).get()
check rm2.ensureChannel(testChannel).isOk()
let rm2 = newReliabilityManager(
participantId = "alice", config = smallCfg, persistence = p2
).get()
check (await rm2.ensureChannel(testChannel)).isOk()
let history = rm2.channels[testChannel].messageHistory
check history.len == 3
check "m1" notin history
@ -291,16 +315,16 @@ suite "Persistence: write → restart → read-back":
channelId = testChannel, content = @[6.byte],
bloomFilter = @[], senderId = "alice",
)
rm2.addToHistory(m6, testChannel)
await rm2.addToHistory(m6, testChannel)
check "m3" notin store.log[testChannel]
check "m6" in store.log[testChannel]
rm2.cleanup()
await rm2.cleanup()
test "dep-clear cascade resumes correctly across a restart":
asyncTest "dep-clear cascade resumes correctly across a restart":
let store = newInMemoryStore()
let p1 = newInMemoryPersistence(store)
let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get()
check rm1.ensureChannel(testChannel).isOk()
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
check (await rm1.ensureChannel(testChannel)).isOk()
# Receive c (deps on b), then b (deps on a). Both must buffer.
let msgC = SdsMessage.init(
@ -315,17 +339,17 @@ suite "Persistence: write → restart → read-back":
channelId = testChannel, content = @[2.byte],
bloomFilter = @[], senderId = "bob",
)
discard rm1.unwrapReceivedMessage(serializeMessage(msgC).get())
discard rm1.unwrapReceivedMessage(serializeMessage(msgB).get())
discard await rm1.unwrapReceivedMessage(serializeMessage(msgC).get())
discard await rm1.unwrapReceivedMessage(serializeMessage(msgB).get())
check "c" in store.incoming[testChannel]
check "b" in store.incoming[testChannel]
rm1.cleanup()
await rm1.cleanup()
# Restart — both still buffered, with intact missingDeps.
let p2 = newInMemoryPersistence(store)
let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get()
check rm2.ensureChannel(testChannel).isOk()
let inbuf = rm2.getIncomingBuffer(testChannel)
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
check (await rm2.ensureChannel(testChannel)).isOk()
let inbuf = await rm2.getIncomingBuffer(testChannel)
check "c" in inbuf
check "b" in inbuf
@ -335,11 +359,12 @@ suite "Persistence: write → restart → read-back":
channelId = testChannel, content = @[1.byte],
bloomFilter = @[], senderId = "alice",
)
discard rm2.unwrapReceivedMessage(serializeMessage(msgA).get())
discard await rm2.unwrapReceivedMessage(serializeMessage(msgA).get())
let history = rm2.channels[testChannel].messageHistory
check "a" in history
check "b" in history
check "c" in history
# Buffer should be drained.
check rm2.getIncomingBuffer(testChannel).len == 0
rm2.cleanup()
let inbufFinal = await rm2.getIncomingBuffer(testChannel)
check inbufFinal.len == 0
await rm2.cleanup()

528
tests/test_reliability.nim
View File

File diff suppressed because it is too large Load Diff