logos-messaging/nim-sds
1
0
Fork 0
mirror of https://github.com/logos-messaging/nim-sds.git synced 2026-07-25 17:13:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: make Persistence interface async (#69)

Browse Source 
* feat: make Persistence interface async

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>

* refactor: drop asyncSpawn, add asyncSetup/asyncTeardown

Three asyncSpawn usages removed:

- sds.nim startPeriodicTasks: stored the periodic-task futures on
  ReliabilityManager (new field `periodicTasks: seq[FutureBase]`) so
  cleanup can cancel them on shutdown instead of leaking the loops
  against a cleared manager.
- library/sds_thread/sds_thread.nim: fireSync moved BEFORE processing,
  then `await SdsThreadRequest.process(...)` instead of asyncSpawn'ing
  it. Aligns the worker with the SP-channel + lock assumption that
  there are no concurrent requests; caller throughput is unchanged
  because the caller only waits for receipt (fireSync), not processing.
- tests TestBus repair callback: replaced asyncSpawn(deliverExcept...)
  with an explicit pending-delivery queue drained by `bus.drain()`.
  Integration tests no longer rely on `sleepAsync(10ms)` to let
  spawned deliveries finish — they await drain instead.

Tests also pick up an asyncSetup/asyncTeardown pair (tests/async_unittest.nim)
so suite fixtures can `await` directly. All `waitFor` in setup/teardown
blocks is gone; only the top-level asyncTest wrapper still uses waitFor
(once, to drive the async proc to completion).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* Correctly propagate error hidden by new async move

* Correctly handle future cancellation exceptions, +some housekeeping

* Apply suggestion from @Ivansete-status

Co-authored-by: Ivan FB <128452529+Ivansete-status@users.noreply.github.com>

* Stylistics, async default implication addressed, nph style run

* Remove leaking CancelledFuture from public facing + as a consequence it is tuneled into handling CatchableError everywhere

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: Ivan FB <128452529+Ivansete-status@users.noreply.github.com>
This commit is contained in:
NagyZoltanPeter 2026-05-25 22:30:15 +02:00 committed by GitHub
parent 2e9a7683f0
commit 35a33adc98
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
27 changed files with 1450 additions and 961 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.gitignore vendored
View File

@ -1,11 +1,9 @@
.DS_Store
tests/test_reliability
tests/bloom
nph
docs
for_reference
do_not_commit
build/*
/build/
result
sds.nims
/.update.timestamp
@ -15,3 +13,4 @@ sds.nims
nimble.develop
nimble.paths
nimbledeps
.claude/**

43
AGENTS.md Normal file
View File

@ -0,0 +1,43 @@
<!-- gitnexus:start -->
# GitNexus — Code Intelligence
This project is indexed by GitNexus as **nim-sds** (889 symbols, 1437 relationships, 45 execution flows). Use the GitNexus MCP tools to understand code, assess impact, and navigate safely.
> If any GitNexus tool warns the index is stale, run `npx gitnexus analyze` in terminal first.
## Always Do
- **MUST run impact analysis before editing any symbol.** Before modifying a function, class, or method, run `gitnexus_impact({target: "symbolName", direction: "upstream"})` and report the blast radius (direct callers, affected processes, risk level) to the user.
- **MUST run `gitnexus_detect_changes()` before committing** to verify your changes only affect expected symbols and execution flows.
- **MUST warn the user** if impact analysis returns HIGH or CRITICAL risk before proceeding with edits.
- When exploring unfamiliar code, use `gitnexus_query({query: "concept"})` to find execution flows instead of grepping. It returns process-grouped results ranked by relevance.
- When you need full context on a specific symbol — callers, callees, which execution flows it participates in — use `gitnexus_context({name: "symbolName"})`.
## Never Do
- NEVER edit a function, class, or method without first running `gitnexus_impact` on it.
- NEVER ignore HIGH or CRITICAL risk warnings from impact analysis.
- NEVER rename symbols with find-and-replace — use `gitnexus_rename` which understands the call graph.
- NEVER commit changes without running `gitnexus_detect_changes()` to check affected scope.
## Resources
| Resource | Use for |
|----------|---------|
| `gitnexus://repo/nim-sds/context` | Codebase overview, check index freshness |
| `gitnexus://repo/nim-sds/clusters` | All functional areas |
| `gitnexus://repo/nim-sds/processes` | All execution flows |
| `gitnexus://repo/nim-sds/process/{name}` | Step-by-step execution trace |
## CLI
| Task | Read this skill file |
|------|---------------------|
| Understand architecture / "How does X work?" | `.claude/skills/gitnexus/gitnexus-exploring/SKILL.md` |
| Blast radius / "What breaks if I change X?" | `.claude/skills/gitnexus/gitnexus-impact-analysis/SKILL.md` |
| Trace bugs / "Why is X failing?" | `.claude/skills/gitnexus/gitnexus-debugging/SKILL.md` |
| Rename / extract / split / refactor | `.claude/skills/gitnexus/gitnexus-refactoring/SKILL.md` |
| Tools, resources, schema reference | `.claude/skills/gitnexus/gitnexus-guide/SKILL.md` |
| Index, status, clean, wiki CLI commands | `.claude/skills/gitnexus/gitnexus-cli/SKILL.md` |
<!-- gitnexus:end -->

44
CLAUDE.md
View File

@ -162,3 +162,47 @@ nimble lock # update lock after changing sds.nimble
```
If using Nix, also recalculate the fixed-output hash in `nix/deps.nix` after updating `nimble.lock` (run `nix build`, copy the expected hash from the error, paste into `outputHash`).
<!-- gitnexus:start -->
# GitNexus — Code Intelligence
This project is indexed by GitNexus as **nim-sds** (889 symbols, 1437 relationships, 45 execution flows). Use the GitNexus MCP tools to understand code, assess impact, and navigate safely.
> If any GitNexus tool warns the index is stale, run `npx gitnexus analyze` in terminal first.
## Always Do
- **MUST run impact analysis before editing any symbol.** Before modifying a function, class, or method, run `gitnexus_impact({target: "symbolName", direction: "upstream"})` and report the blast radius (direct callers, affected processes, risk level) to the user.
- **MUST run `gitnexus_detect_changes()` before committing** to verify your changes only affect expected symbols and execution flows.
- **MUST warn the user** if impact analysis returns HIGH or CRITICAL risk before proceeding with edits.
- When exploring unfamiliar code, use `gitnexus_query({query: "concept"})` to find execution flows instead of grepping. It returns process-grouped results ranked by relevance.
- When you need full context on a specific symbol — callers, callees, which execution flows it participates in — use `gitnexus_context({name: "symbolName"})`.
## Never Do
- NEVER edit a function, class, or method without first running `gitnexus_impact` on it.
- NEVER ignore HIGH or CRITICAL risk warnings from impact analysis.
- NEVER rename symbols with find-and-replace — use `gitnexus_rename` which understands the call graph.
- NEVER commit changes without running `gitnexus_detect_changes()` to check affected scope.
## Resources
| Resource | Use for |
|----------|---------|
| `gitnexus://repo/nim-sds/context` | Codebase overview, check index freshness |
| `gitnexus://repo/nim-sds/clusters` | All functional areas |
| `gitnexus://repo/nim-sds/processes` | All execution flows |
| `gitnexus://repo/nim-sds/process/{name}` | Step-by-step execution trace |
## CLI
| Task | Read this skill file |
|------|---------------------|
| Understand architecture / "How does X work?" | `.claude/skills/gitnexus/gitnexus-exploring/SKILL.md` |
| Blast radius / "What breaks if I change X?" | `.claude/skills/gitnexus/gitnexus-impact-analysis/SKILL.md` |
| Trace bugs / "Why is X failing?" | `.claude/skills/gitnexus/gitnexus-debugging/SKILL.md` |
| Rename / extract / split / refactor | `.claude/skills/gitnexus/gitnexus-refactoring/SKILL.md` |
| Tools, resources, schema reference | `.claude/skills/gitnexus/gitnexus-guide/SKILL.md` |
| Index, status, clean, wiki CLI commands | `.claude/skills/gitnexus/gitnexus-cli/SKILL.md` |
<!-- gitnexus:end -->

8
library/events/json_message_ready_event.nim
View File

@ -5,8 +5,12 @@ type JsonMessageReadyEvent* = ref object of JsonEvent
messageId*: SdsMessageID
channelId*: SdsChannelID
proc new*(T: type JsonMessageReadyEvent, messageId: SdsMessageID, channelId: SdsChannelID): T =
return JsonMessageReadyEvent(eventType: "message_ready", messageId: messageId, channelId: channelId)
proc new*(
T: type JsonMessageReadyEvent, messageId: SdsMessageID, channelId: SdsChannelID
): T =
return JsonMessageReadyEvent(
eventType: "message_ready", messageId: messageId, channelId: channelId
)
method `$`*(jsonMessageReady: JsonMessageReadyEvent): string =
$(%*jsonMessageReady)

8
library/events/json_message_sent_event.nim
View File

@ -5,8 +5,12 @@ type JsonMessageSentEvent* = ref object of JsonEvent
messageId*: SdsMessageID
channelId*: SdsChannelID
proc new*(T: type JsonMessageSentEvent, messageId: SdsMessageID, channelId: SdsChannelID): T =
return JsonMessageSentEvent(eventType: "message_sent", messageId: messageId, channelId: channelId)
proc new*(
T: type JsonMessageSentEvent, messageId: SdsMessageID, channelId: SdsChannelID
): T =
return JsonMessageSentEvent(
eventType: "message_sent", messageId: messageId, channelId: channelId
)
method `$`*(jsonMessageSent: JsonMessageSentEvent): string =
$(%*jsonMessageSent)

5
library/events/json_missing_dependencies_event.nim
View File

@ -13,7 +13,10 @@ proc new*(
channelId: SdsChannelID,
): T =
return JsonMissingDependenciesEvent(
eventType: "missing_dependencies", messageId: messageId, missingDeps: missingDeps, channelId: channelId
eventType: "missing_dependencies",
messageId: messageId,
missingDeps: missingDeps,
channelId: channelId,
)
method `$`*(jsonMissingDependencies: JsonMissingDependenciesEvent): string =

12
library/libsds.nim
View File

@ -63,7 +63,8 @@ var
proc acquireCtx(callback: SdsCallBack, userData: pointer): ptr SdsContext =
ctxPoolLock.acquire()
defer: ctxPoolLock.release()
defer:
ctxPoolLock.release()
if ctxPool.len > 0:
result = ctxPool.pop()
else:
@ -74,7 +75,8 @@ proc acquireCtx(callback: SdsCallBack, userData: pointer): ptr SdsContext =
proc releaseCtx(ctx: ptr SdsContext) =
ctxPoolLock.acquire()
defer: ctxPoolLock.release()
defer:
ctxPoolLock.release()
ctx.userData = nil
ctx.eventCallback = nil
ctx.eventUserData = nil
@ -105,7 +107,9 @@ proc onMessageSent(ctx: ptr SdsContext): MessageSentCallback =
$JsonMessageSentEvent.new(messageId, channelId)
proc onMissingDependencies(ctx: ptr SdsContext): MissingDependenciesCallback =
return proc(messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID) {.gcsafe.} =
return proc(
messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID
) {.gcsafe.} =
callEventCallback(ctx, "onMissingDependencies"):
$JsonMissingDependenciesEvent.new(messageId, missingDeps, channelId)
@ -135,7 +139,7 @@ proc onRetrievalHint(ctx: ptr SdsContext): RetrievalHintProvider =
copyMem(addr hintBytes[0], hint, hintLen)
deallocShared(hint)
return hintBytes
return @[]
### End of not-exported components

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:

12
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,10 +64,14 @@ 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)
let res = SdsUnwrapResponse(
message: unwrappedMessage, missingDeps: missingDeps, channelId: extractedChannelId
)
# return the result as a json string
var node = newJObject()

16
library/sds_thread/sds_thread.nim
View File

@ -4,10 +4,7 @@
import std/[options, atomics, os, net, locks]
import chronicles, chronos, chronos/threadsync, taskpools/channels_spsc_single, results
import
../ffi_types,
./inter_thread_communication/sds_thread_request,
sds/sds_utils
import ../ffi_types, ./inter_thread_communication/sds_thread_request, sds/sds_utils
type SdsContext* = object
thread: Thread[(ptr SdsContext)]
@ -43,13 +40,18 @@ proc runSds(ctx: ptr SdsContext) {.async.} =
error "sds thread could not receive a request"
continue
## Handle the request
asyncSpawn SdsThreadRequest.process(request, addr rm)
## Ack receipt to the requester thread BEFORE processing — it only
## waits for "received", not "processed", so the caller's throughput
## doesn't change. Processing is then awaited (was: asyncSpawn'd),
## which serializes requests on this worker. The SP channel + lock
## above already assume no concurrent requests, so awaiting here
## aligns the processing side with that assumption.
let fireRes = ctx.reqReceivedSignal.fireSync()
if fireRes.isErr():
error "could not fireSync back to requester thread", error = fireRes.error
await SdsThreadRequest.process(request, addr rm)
proc run(ctx: ptr SdsContext) {.thread.} =
## Launch sds worker
waitFor runSds(ctx)

550
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,197 @@ 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
) {.async: (raises: [CatchableError]).} =
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 as e:
error "Failed to review ack status", msg = getCurrentExceptionMsg()
raise e
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
) {.async: (raises: [CatchableError]).} =
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 as e:
error "Failed to process incoming buffer",
channelId = channelId, msg = getCurrentExceptionMsg()
raise e
proc unwrapReceivedMessage*(
rm: ReliabilityManager, message: seq[byte]
): Result[
tuple[message: seq[byte], missingDeps: seq[HistoryEntry], channelId: SdsChannelID],
ReliabilityError,
] =
): Future[
Result[
tuple[message: seq[byte], missingDeps: seq[HistoryEntry], channelId: SdsChannelID],
ReliabilityError,
]
] {.async: (raises: []).} =
## Unwraps a received message and processes its reliability metadata.
try:
let channelId = extractChannelId(message).valueOr:
@ -200,22 +232,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,18 +257,19 @@ 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)
if repairEntry.messageId in channel.messageHistory and
rm.participantId.len > 0 and repairEntry.senderId.len > 0:
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(
rm.participantId, repairEntry.senderId,
repairEntry.messageId, rm.config.numResponseGroups
rm.participantId, repairEntry.senderId, repairEntry.messageId,
rm.config.numResponseGroups,
):
let serialized = serializeMessage(channel.messageHistory[repairEntry.messageId])
let serialized =
serializeMessage(channel.messageHistory[repairEntry.messageId])
if serialized.isOk():
let tResp = computeTResp(
rm.participantId, repairEntry.senderId,
repairEntry.messageId, rm.config.repairTMax
rm.participantId, repairEntry.senderId, repairEntry.messageId,
rm.config.repairTMax,
)
let inEntry = IncomingRepairEntry(
inHistEntry: repairEntry,
@ -244,7 +277,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 +293,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,43 +303,44 @@ 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(),
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:
for dep in missingDeps:
if dep.messageId notin channel.outgoingRepairBuffer:
let tReq = computeTReq(
rm.participantId, dep.messageId,
rm.config.repairTMin, rm.config.repairTMax
)
let outEntry = OutgoingRepairEntry(
outHistEntry: dep,
minTimeRepairReq: now + tReq,
rm.participantId, dep.messageId, rm.config.repairTMin,
rm.config.repairTMax,
)
let outEntry =
OutgoingRepairEntry(outHistEntry: dep, 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: []).} =
## Marks the specified message dependencies as met.
try:
if channelId notin rm.channels:
@ -322,17 +358,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,82 +381,89 @@ proc setCallbacks*(
onPeriodicSync: PeriodicSyncCallback = nil,
onRetrievalHint: RetrievalHintProvider = nil,
onRepairReady: RepairReadyCallback = nil,
) =
) {.async: (raises: []).} =
## 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
) {.async: (raises: []).} =
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
) {.async: (raises: [CancelledError]), gcsafe.} =
proc periodicBufferSweep(rm: ReliabilityManager) {.async: (raises: [CancelledError]).} =
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 CatchableError:
error "Error in periodic buffer sweep", msg = getCurrentExceptionMsg()
await sleepAsync(chronos.milliseconds(rm.config.bufferSweepInterval.inMilliseconds))
proc periodicSyncMessage(
rm: ReliabilityManager
) {.async: (raises: [CancelledError]), gcsafe.} =
proc periodicSyncMessage(rm: ReliabilityManager) {.async: (raises: [CancelledError]).} =
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) {.async: (raises: []).} =
## 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 +477,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,43 +490,59 @@ 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.} =
proc periodicRepairSweep(rm: ReliabilityManager) {.async: (raises: [CancelledError]).} =
## SDS-R: Periodically checks repair buffers for expired entries.
while true:
rm.runRepairSweep()
try:
await rm.runRepairSweep()
except CatchableError:
error "Error in periodic repair sweep", msg = getCurrentExceptionMsg()
await sleepAsync(chronos.milliseconds(rm.config.repairSweepInterval.inMilliseconds))
proc startPeriodicTasks*(rm: ReliabilityManager) =
## Starts the periodic tasks for buffer sweeping and sync message sending.
asyncSpawn rm.periodicBufferSweep()
asyncSpawn rm.periodicSyncMessage()
asyncSpawn rm.periodicRepairSweep()
## Starts the periodic background tasks (buffer sweep, sync message,
## SDS-R repair sweep). The futures are kept on the manager so `cleanup`
## can cancel them — without that, the loops would outlive a cleaned-up
## manager and keep firing against cleared state.
rm.periodicTasks.add(FutureBase(rm.periodicBufferSweep()))
rm.periodicTasks.add(FutureBase(rm.periodicSyncMessage()))
rm.periodicTasks.add(FutureBase(rm.periodicRepairSweep()))
proc resetReliabilityManager*(rm: ReliabilityManager): Result[void, ReliabilityError] =
proc resetReliabilityManager*(
rm: ReliabilityManager
): Future[Result[void, ReliabilityError]] {.async: (raises: []).} =
## 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)

6
sds.nimble
View File

@ -65,9 +65,9 @@ proc getMyCpu(): string =
# Tasks
task test, "Run the test suite":
exec "nim c -r tests/test_bloom.nim"
exec "nim c -r tests/test_reliability.nim"
exec "nim c -r tests/test_persistence.nim"
exec "nim c -r --outdir:build tests/test_bloom.nim"
exec "nim c -r --outdir:build tests/test_reliability.nim"
exec "nim c -r --outdir:build tests/test_persistence.nim"
task libsdsDynamicWindows, "Generate bindings":
let outLibNameAndExt = "libsds.dll"

3
sds/bloom.nim
View File

@ -80,7 +80,8 @@ proc initializeBloomFilter*(
proc `$`*(bf: BloomFilter): string =
## Prints the configuration of the Bloom filter.
return "Bloom filter with $1 capacity, $2 error rate, $3 hash functions, and requiring $4 bits of memory." %
return
"Bloom filter with $1 capacity, $2 error rate, $3 hash functions, and requiring $4 bits of memory." %
[
$bf.capacity,
formatFloat(bf.errorRate, format = ffScientific, precision = 1),

9
sds/message.nim
View File

@ -7,10 +7,5 @@ import ./types/repair_entry
import ./types/reliability_config
export
sds_message_id,
history_entry,
sds_message,
unacknowledged_message,
incoming_message,
repair_entry,
reliability_config
sds_message_id, history_entry, sds_message, unacknowledged_message, incoming_message,
repair_entry, reliability_config

231
sds/sds_utils.nim
View File

@ -1,11 +1,12 @@
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,
reliability_error, callbacks, app_callbacks, reliability_config, repair_entry,
channel_context, reliability_manager,
]
import
./types/[
sds_message_id, history_entry, sds_message, unacknowledged_message,
incoming_message, reliability_error, callbacks, app_callbacks, reliability_config,
repair_entry, channel_context, reliability_manager,
]
export
sds_message_id, history_entry, sds_message, unacknowledged_message, incoming_message,
reliability_error, callbacks, app_callbacks, reliability_config, repair_entry,
@ -16,44 +17,59 @@ proc defaultConfig*(): ReliabilityConfig =
proc dropChannelFromPersistence*(
rm: ReliabilityManager, channelId: SdsChannelID
) {.gcsafe, raises: [].} =
) {.async: (raises: []).} =
## 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) {.async: (raises: []).} =
## 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():
##
## Periodic tasks are cancelled BEFORE acquiring the lock so that a task
## currently blocked on `lock.acquire()` can unwind via CancelledError
## without deadlocking against cleanup itself.
if rm.isNil():
return
for task in rm.periodicTasks:
if not task.finished:
await task.cancelAndWait()
rm.periodicTasks.setLen(0)
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:
) {.async: (raises: []).} =
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: [].} =
) {.async: (raises: []).} =
## 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,31 +78,33 @@ 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: [].} =
) {.async: (raises: []).} =
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()
proc newHistoryEntry*(messageId: SdsMessageID, retrievalHint: seq[byte] = @[]): HistoryEntry =
proc newHistoryEntry*(
messageId: SdsMessageID, retrievalHint: seq[byte] = @[]
): HistoryEntry =
return HistoryEntry.init(messageId, retrievalHint)
proc toCausalHistory*(messageIds: seq[SdsMessageID]): seq[HistoryEntry] =
@ -100,9 +118,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 +134,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
@ -142,14 +160,14 @@ proc isInResponseGroup*(
## Determines if this participant is in the response group for a given message per SDS-R spec:
## hash(participant_id, message_id) % num_groups == hash(sender_id, message_id) % num_groups
if numResponseGroups <= 1:
return true # All participants in the same group
return true # All participants in the same group
let myGroup = abs(hash(participantId.string & messageId)) mod numResponseGroups
let senderGroup = abs(hash(senderId.string & messageId)) mod numResponseGroups
myGroup == senderGroup
proc getRecentHistoryEntries*(
rm: ReliabilityManager, n: int, channelId: SdsChannelID
): seq[HistoryEntry] =
): Future[seq[HistoryEntry]] {.async: (raises: []).} =
## Get recent history entries for sending in causal history.
## Populates retrieval hints and senderId (SDS-R) for each entry.
try:
@ -158,21 +176,21 @@ proc getRecentHistoryEntries*(
var orderedIds: seq[SdsMessageID] = @[]
for msgId in channel.messageHistory.keys:
orderedIds.add(msgId)
let recentMessageIds =
orderedIds[max(0, orderedIds.len - n) .. ^1]
let recentMessageIds = orderedIds[max(0, orderedIds.len - n) .. ^1]
var entries: seq[HistoryEntry] = @[]
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 +215,9 @@ proc checkDependencies*(
proc getMessageHistory*(
rm: ReliabilityManager, channelId: SdsChannelID
): seq[SdsMessageID] =
withLock rm.lock:
): Future[seq[SdsMessageID]] {.async: (raises: []).} =
try:
await rm.lock.acquire()
try:
if channelId in rm.channels:
var ids: seq[SdsMessageID] = @[]
@ -207,42 +226,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: []).} =
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]).} =
## 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
@ -250,9 +277,11 @@ proc getOrCreateChannel*(
try:
if channelId notin rm.channels:
let channel = ChannelContext.new(
RollingBloomFilter.init(rm.config.bloomFilterCapacity, rm.config.bloomFilterErrorRate)
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 +296,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: []).} =
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: []).} =
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)

18
sds/types.nim
View File

@ -16,19 +16,7 @@ import sds/types/reliability_manager
import sds/types/protobuf_error
export
sds_message_id,
history_entry,
sds_message,
unacknowledged_message,
incoming_message,
bloom_filter,
rolling_bloom_filter,
reliability_error,
callbacks,
app_callbacks,
reliability_config,
repair_entry,
channel_context,
persistence,
reliability_manager,
sds_message_id, history_entry, sds_message, unacknowledged_message, incoming_message,
bloom_filter, rolling_bloom_filter, reliability_error, callbacks, app_callbacks,
reliability_config, repair_entry, channel_context, persistence, reliability_manager,
protobuf_error

3
sds/types/history_entry.nim
View File

@ -3,7 +3,8 @@ export sds_message_id
type HistoryEntry* = object
messageId*: SdsMessageID
retrievalHint*: seq[byte] ## Optional hint for efficient retrieval (e.g., Waku message hash)
retrievalHint*: seq[byte]
## Optional hint for efficient retrieval (e.g., Waku message hash)
senderId*: SdsParticipantID ## Original message sender's participant ID (SDS-R)
proc init*(

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,124 @@ 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: [].}
removeLogEntry*:
proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].}
setRetrievalHint*:
proc(msgId: SdsMessageID, hint: seq[byte]) {.gcsafe, raises: [].}
appendLogEntry*: proc(channelId: SdsChannelID, msg: SdsMessage): Future[void] {.
async: (raises: []), gcsafe
.}
removeLogEntry*: 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: [].}
removeOutgoing*:
proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].}
saveOutgoing*: proc(
channelId: SdsChannelID, msg: UnacknowledgedMessage
): Future[void] {.async: (raises: []), gcsafe.}
removeOutgoing*: proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {.
async: (raises: []), gcsafe
.}
# Incoming dependency-waiting buffer
saveIncoming*:
proc(channelId: SdsChannelID, msg: IncomingMessage) {.gcsafe, raises: [].}
removeIncoming*:
proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].}
saveIncoming*: proc(channelId: SdsChannelID, msg: IncomingMessage): Future[void] {.
async: (raises: []), gcsafe
.}
removeIncoming*: 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: [].}
removeOutgoingRepair*:
proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].}
) {.async: (raises: []).}
removeOutgoingRepair*: 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: [].}
removeIncomingRepair*:
proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].}
) {.async: (raises: []).}
removeIncomingRepair*: 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: [].}
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) {.async: (raises: []).} =
discard,
appendLogEntry: proc(channelId: SdsChannelID, msg: SdsMessage) =
appendLogEntry: proc(
channelId: SdsChannelID, msg: SdsMessage
) {.async: (raises: []).} =
discard,
removeLogEntry: proc(channelId: SdsChannelID, msgId: SdsMessageID) =
removeLogEntry: proc(
channelId: SdsChannelID, msgId: SdsMessageID
) {.async: (raises: []).} =
discard,
setRetrievalHint: proc(msgId: SdsMessageID, hint: seq[byte]) =
setRetrievalHint: proc(
msgId: SdsMessageID, hint: seq[byte]
) {.async: (raises: []).} =
discard,
saveOutgoing: proc(channelId: SdsChannelID, msg: UnacknowledgedMessage) =
saveOutgoing: proc(
channelId: SdsChannelID, msg: UnacknowledgedMessage
) {.async: (raises: []).} =
discard,
removeOutgoing: proc(channelId: SdsChannelID, msgId: SdsMessageID) =
removeOutgoing: proc(
channelId: SdsChannelID, msgId: SdsMessageID
) {.async: (raises: []).} =
discard,
saveIncoming: proc(channelId: SdsChannelID, msg: IncomingMessage) =
saveIncoming: proc(
channelId: SdsChannelID, msg: IncomingMessage
) {.async: (raises: []).} =
discard,
removeIncoming: proc(channelId: SdsChannelID, msgId: SdsMessageID) =
removeIncoming: proc(
channelId: SdsChannelID, msgId: SdsMessageID
) {.async: (raises: []).} =
discard,
saveOutgoingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID, entry: OutgoingRepairEntry
) =
) {.async: (raises: []).} =
discard,
removeOutgoingRepair: proc(channelId: SdsChannelID, msgId: SdsMessageID) =
removeOutgoingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID
) {.async: (raises: []).} =
discard,
saveIncomingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID, entry: IncomingRepairEntry
) =
) {.async: (raises: []).} =
discard,
removeIncomingRepair: proc(channelId: SdsChannelID, msgId: SdsMessageID) =
removeIncomingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID
) {.async: (raises: []).} =
discard,
dropChannel: proc(channelId: SdsChannelID) =
dropChannel: proc(channelId: SdsChannelID) {.async: (raises: []).} =
discard,
loadAllForChannel: proc(channelId: SdsChannelID): ChannelSnapshot =
ChannelSnapshot(),
loadAllForChannel: proc(
channelId: SdsChannelID
): Future[ChannelSnapshot] {.async: (raises: []).} =
return ChannelSnapshot(),
)

3
sds/types/reliability_config.nim
View File

@ -55,8 +55,7 @@ proc init*(
# clean() churn and incomplete summaries to peers, with no compensating gain.
if maxMessageHistory > bloomFilterCapacity:
warn "maxMessageHistory > bloomFilterCapacity will cause continuous bloom rebuilds and incomplete summaries to peers; reduce maxMessageHistory or increase bloomFilterCapacity unless you have a specific reason",
maxMessageHistory = maxMessageHistory,
bloomFilterCapacity = bloomFilterCapacity
maxMessageHistory = maxMessageHistory, bloomFilterCapacity = bloomFilterCapacity
return T(
bloomFilterCapacity: bloomFilterCapacity,
bloomFilterErrorRate: bloomFilterErrorRate,

15
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,14 @@ 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.
periodicTasks*: seq[FutureBase]
## Handles to the background loops started by `startPeriodicTasks` so
## `cleanup` can cancel them on shutdown instead of leaking them.
onMessageReady*: proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.}
onMessageSent*: proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.}
onMissingDependencies*: proc(
@ -42,6 +50,7 @@ proc new*(
config: config,
participantId: participantId,
persistence: persistence,
lock: newAsyncLock(),
periodicTasks: @[],
)
rm.lock.initLock()
return rm

5
sds/types/unacknowledged_message.nim
View File

@ -8,6 +8,9 @@ type UnacknowledgedMessage* = object
resendAttempts*: int
proc init*(
T: type UnacknowledgedMessage, message: SdsMessage, sendTime: Time, resendAttempts: int
T: type UnacknowledgedMessage,
message: SdsMessage,
sendTime: Time,
resendAttempts: int,
): T =
return T(message: message, sendTime: sendTime, resendAttempts: resendAttempts)

70
tests/async_unittest.nim Normal file
View File

@ -0,0 +1,70 @@
## Shared async-aware wrappers around `unittest` so tests in this repo can
## `await` directly in setup/test/teardown blocks instead of sprinkling
## `waitFor` at each call site.
##
## Usage:
##
## ```nim
## import ./async_unittest
##
## suite "X":
## var rm: ReliabilityManager
##
## asyncSetup:
## rm = newReliabilityManager(...).get()
## check (await rm.ensureChannel("ch")).isOk()
##
## asyncTeardown:
## if not rm.isNil:
## await rm.cleanup()
##
## asyncTest "Y":
## await rm.wrapOutgoingMessage(...)
## ```
##
## All three blocks run inside the same async proc (per test). unittest's
## own `setup:`/`teardown:` still work for purely synchronous fixtures.
import unittest, chronos
export unittest, chronos
template asyncSetup*(body: untyped) {.dirty.} =
## Async counterpart to unittest's `setup:`. Runs inside each asyncTest's
## async proc, so `await` works.
template asyncTestSetupIMPL(): untyped {.dirty.} =
body
template asyncTeardown*(body: untyped) {.dirty.} =
## Async counterpart to unittest's `teardown:`. Runs in a `finally` so it
## executes even when the test body (or setup) raises.
template asyncTestTeardownIMPL(): untyped {.dirty.} =
body
template asyncTest*(name: string, body: untyped) =
## Wraps a unittest `test` body in an async proc so `await` works on the
## now-async ReliabilityManager API. unittest's `check` raises Exception,
## which is wider than chronos's default CatchableError; the exception is
## caught inside the async body, stashed, and re-raised after waitFor so
## unittest's normal failure handling sees it.
##
## `cast(gcsafe)` is needed because suite-level vars (e.g. `var rm`) look
## like globals to the async closure, but the FFI runtime is single-thread
## so the "not gcsafe" warning isn't a real hazard here.
test name:
var asyncTestErr {.inject.}: ref Exception = nil
proc inner() {.async.} =
{.cast(gcsafe).}:
try:
when declared(asyncTestSetupIMPL):
asyncTestSetupIMPL()
try:
body
finally:
when declared(asyncTestTeardownIMPL):
asyncTestTeardownIMPL()
except Exception as e:
asyncTestErr = e
waitFor inner()
if asyncTestErr != nil:
raise asyncTestErr

82
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,86 @@ proc newInMemoryStore*(): InMemoryStore =
proc newInMemoryPersistence*(store: InMemoryStore): Persistence =
Persistence(
saveLamport: proc(channelId: SdsChannelID, lamport: int64) {.gcsafe, raises: [].} =
saveLamport: proc(channelId: SdsChannelID, lamport: int64) {.async: (raises: []).} =
store.lamports[channelId] = lamport,
appendLogEntry: proc(channelId: SdsChannelID, msg: SdsMessage) {.gcsafe, raises: [].} =
appendLogEntry: proc(
channelId: SdsChannelID, msg: SdsMessage
) {.async: (raises: []).} =
{.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
) {.async: (raises: []).} =
{.cast(raises: []).}:
if channelId in store.log:
store.log[channelId].del(msgId),
setRetrievalHint: proc(msgId: SdsMessageID, hint: seq[byte]) {.gcsafe, raises: [].} =
store.log[channelId].del(msgId)
,
setRetrievalHint: proc(
msgId: SdsMessageID, hint: seq[byte]
) {.async: (raises: []).} =
store.hints[msgId] = hint,
saveOutgoing: proc(channelId: SdsChannelID, msg: UnacknowledgedMessage) {.gcsafe, raises: [].} =
saveOutgoing: proc(
channelId: SdsChannelID, msg: UnacknowledgedMessage
) {.async: (raises: []).} =
{.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
) {.async: (raises: []).} =
{.cast(raises: []).}:
if channelId in store.outgoing:
store.outgoing[channelId].del(msgId),
saveIncoming: proc(channelId: SdsChannelID, msg: IncomingMessage) {.gcsafe, raises: [].} =
store.outgoing[channelId].del(msgId)
,
saveIncoming: proc(
channelId: SdsChannelID, msg: IncomingMessage
) {.async: (raises: []).} =
{.cast(raises: []).}:
if channelId notin store.incoming:
store.incoming[channelId] =
initOrderedTable[SdsMessageID, IncomingMessage]()
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
) {.async: (raises: []).} =
{.cast(raises: []).}:
if channelId in store.incoming:
store.incoming[channelId].del(msgId),
store.incoming[channelId].del(msgId)
,
saveOutgoingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID, entry: OutgoingRepairEntry
) {.gcsafe, raises: [].} =
) {.async: (raises: []).} =
{.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
) {.async: (raises: []).} =
{.cast(raises: []).}:
if channelId in store.outgoingRepair:
store.outgoingRepair[channelId].del(msgId),
store.outgoingRepair[channelId].del(msgId)
,
saveIncomingRepair: proc(
channelId: SdsChannelID, msgId: SdsMessageID, entry: IncomingRepairEntry
) {.gcsafe, raises: [].} =
) {.async: (raises: []).} =
{.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
) {.async: (raises: []).} =
{.cast(raises: []).}:
if channelId in store.incomingRepair:
store.incomingRepair[channelId].del(msgId),
dropChannel: proc(channelId: SdsChannelID) {.gcsafe, raises: [].} =
store.incomingRepair[channelId].del(msgId)
,
dropChannel: proc(channelId: SdsChannelID) {.async: (raises: []).} =
{.cast(raises: []).}:
store.lamports.del(channelId)
store.log.del(channelId)
@ -102,8 +113,9 @@ proc newInMemoryPersistence*(store: InMemoryStore): Persistence =
store.incomingRepair.del(channelId)
store.dropChannelCalls[channelId] =
store.dropChannelCalls.getOrDefault(channelId) + 1,
loadAllForChannel: proc(channelId: SdsChannelID): ChannelSnapshot {.gcsafe, raises: [].} =
loadAllForChannel: proc(
channelId: SdsChannelID
): Future[ChannelSnapshot] {.async: (raises: []).} =
{.cast(raises: []).}:
var snap = ChannelSnapshot()
if channelId in store.lamports:
@ -123,5 +135,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,
)

15
tests/test_bloom.nim
View File

@ -102,14 +102,13 @@ suite "bloom filter":
suite "bloom filter special cases":
test "different patterns of strings":
const testSize = 10_000
let patterns =
@[
"shortstr",
repeat("a", 1000), # Very long string
"special@#$%^&*()", # Special characters
"unicode→★∑≈", # Unicode characters
repeat("pattern", 10), # Repeating pattern
]
let patterns = @[
"shortstr",
repeat("a", 1000), # Very long string
"special@#$%^&*()", # Special characters
"unicode→★∑≈", # Unicode characters
repeat("pattern", 10), # Repeating pattern
]
let bfResult = initializeBloomFilter(testSize, 0.01)
check bfResult.isOk

281
tests/test_persistence.nim
View File

@ -1,51 +1,53 @@
import unittest, results, std/[tables, sets, times]
import results, std/[tables, sets, times]
import sds
import ./async_unittest
import ./in_memory_persistence
converter toParticipantID(s: string): SdsParticipantID = s.SdsParticipantID
converter toParticipantID(s: string): SdsParticipantID =
s.SdsParticipantID
const testChannel = "testChannel"
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)
check store.lamports[testChannel] == 43 # max(42, 0) + 1
rm1.cleanup()
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
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 +57,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 +89,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 +112,54 @@ 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":
let rm = newReliabilityManager(participantId = "alice").get() # default no-op persistence
check rm.ensureChannel(testChannel).isOk()
let wrapped = rm.wrapOutgoingMessage(@[1.byte], "msg-n", testChannel)
asyncTest "noOpPersistence keeps existing manager working":
let rm = newReliabilityManager(participantId = "alice").get()
# default no-op persistence
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 +167,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 +186,46 @@ 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()
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
check (await rm1.ensureChannel(testChannel)).isOk()
# Receive a message that references an unknown dep — triggers SDS-R repair.
let depMsg = SdsMessage.init(
@ -232,32 +237,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
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()
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
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 +276,19 @@ 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
@ -287,59 +297,74 @@ suite "Persistence: write → restart → read-back":
check "m5" in history
# FIFO continues correctly after restart: adding m6 evicts m3, not a stale entry.
let m6 = SdsMessage.init(
messageId = "m6", lamportTimestamp = 6, causalHistory = @[],
channelId = testChannel, content = @[6.byte],
bloomFilter = @[], senderId = "alice",
messageId = "m6",
lamportTimestamp = 6,
causalHistory = @[],
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(
messageId = "c", lamportTimestamp = 30,
messageId = "c",
lamportTimestamp = 30,
causalHistory = @[HistoryEntry.init("b", @[])],
channelId = testChannel, content = @[3.byte],
bloomFilter = @[], senderId = "carol",
channelId = testChannel,
content = @[3.byte],
bloomFilter = @[],
senderId = "carol",
)
let msgB = SdsMessage.init(
messageId = "b", lamportTimestamp = 20,
messageId = "b",
lamportTimestamp = 20,
causalHistory = @[HistoryEntry.init("a", @[])],
channelId = testChannel, content = @[2.byte],
bloomFilter = @[], senderId = "bob",
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
# Now receive a (root) — should cascade-deliver a, b, c.
let msgA = SdsMessage.init(
messageId = "a", lamportTimestamp = 10, causalHistory = @[],
channelId = testChannel, content = @[1.byte],
bloomFilter = @[], senderId = "alice",
messageId = "a",
lamportTimestamp = 10,
causalHistory = @[],
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()

844
tests/test_reliability.nim
View File

File diff suppressed because it is too large Load Diff