feat: propagate persistence backend errors via Result

The Persistence contract previously returned `Future[void]` for writes and
`Future[ChannelSnapshot]` for the loader, with `raises: []`. Backends had no
way to report a failure, so a failed write or a failed/partial read was
silently swallowed — and on the read path a mid-scan failure could bootstrap
a *truncated* channel snapshot, corrupting the rebuilt bloom filter and
lamport clock across a restart.

Make every contract field Result-returning:
  * mutating ops  -> Future[Result[void, string]]
  * loadAllForChannel -> Future[Result[ChannelSnapshot, string]]

The backend-supplied error string is mapped to a new
`ReliabilityError.rePersistenceError` (logged once at the boundary via
`reliabilityErr`) and threaded up through every persistence-touching proc to
the public API, where the caller decides what to do. Request-driven paths
(wrap/unwrap/markDependenciesMet/ensureChannel/removeChannel/reset) propagate
the error; background maintenance loops (periodicBufferSweep,
periodicRepairSweep) log and retry on the next tick, since they have no
synchronous caller.

Tests: in-memory backend gains a `failingOps` injection hook; new
"Persistence: error propagation" suite asserts read/write/drop failures
surface as `rePersistenceError`. Full suite passes (90 OK).

BREAKING CHANGE: the `Persistence` contract signature changed; custom
backends must return `Result` and `ok()` on success. Bumped to 0.3.0.

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

Author: NagyZoltanPeter

sds.nim

@@ -1,7 +1,7 @@
 import strutils, os
 
 # Package
-version = "0.2.4"
+version = "0.3.0"
 author = "Logos Messaging Team"
 description = "E2E Scalable Data Sync API"
 license = "MIT"

sds.nimble

@@ -15,13 +15,24 @@ export
 proc defaultConfig*(): ReliabilityConfig =
   return ReliabilityConfig.init()
 
+proc reliabilityErr*(detail: string): ReliabilityError {.gcsafe, raises: [].} =
+  ## Maps a backend-supplied persistence error string onto the
+  ## `rePersistenceError` enum value. The enum carries no payload, so the
+  ## original detail is logged here — this is the single point where a
+  ## persistence failure is recorded, while the enum value travels up the
+  ## `Result` chain to the public API caller, who decides what to do.
+  warn "persistence operation failed", detail = detail
+  ReliabilityError.rePersistenceError
+
 proc dropChannelFromPersistence*(
     rm: ReliabilityManager, channelId: SdsChannelID
-) {.async: (raises: []).} =
+): Future[Result[void, ReliabilityError]] {.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.
-  await rm.persistence.dropChannel(channelId)
+  (await rm.persistence.dropChannel(channelId)).isOkOr:
+    return err(reliabilityErr(error))
+  ok()
 
 proc cleanup*(rm: ReliabilityManager) {.async: (raises: []).} =
   ## Releases in-memory state. Does NOT wipe persistence — the manager may be
@@ -69,7 +80,7 @@ proc cleanBloomFilter*(
 
 proc addToHistory*(
     rm: ReliabilityManager, msg: SdsMessage, channelId: SdsChannelID
-) {.async: (raises: []).} =
+): Future[Result[void, ReliabilityError]] {.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
@@ -78,29 +89,36 @@ proc addToHistory*(
     if channelId in rm.channels:
       let channel = rm.channels[channelId]
       channel.messageHistory[msg.messageId] = msg
-      await rm.persistence.appendLogEntry(channelId, msg)
+      (await rm.persistence.appendLogEntry(channelId, msg)).isOkOr:
+        return err(reliabilityErr(error))
       while channel.messageHistory.len > rm.config.maxMessageHistory:
         var firstKey: SdsMessageID
         for k in channel.messageHistory.keys:
           firstKey = k
           break
         channel.messageHistory.del(firstKey)
-        await rm.persistence.removeLogEntry(channelId, firstKey)
+        (await rm.persistence.removeLogEntry(channelId, firstKey)).isOkOr:
+          return err(reliabilityErr(error))
+    ok()
   except CatchableError:
     error "Failed to add to history",
       channelId = channelId, msgId = msg.messageId, error = getCurrentExceptionMsg()
+    err(ReliabilityError.reInternalError)
 
 proc updateLamportTimestamp*(
     rm: ReliabilityManager, msgTs: int64, channelId: SdsChannelID
-) {.async: (raises: []).} =
+): Future[Result[void, ReliabilityError]] {.async: (raises: []).} =
   try:
     if channelId in rm.channels:
       let channel = rm.channels[channelId]
       channel.lamportTimestamp = max(msgTs, channel.lamportTimestamp) + 1
-      await rm.persistence.saveLamport(channelId, channel.lamportTimestamp)
+      (await rm.persistence.saveLamport(channelId, channel.lamportTimestamp)).isOkOr:
+        return err(reliabilityErr(error))
+    ok()
   except CatchableError:
     error "Failed to update lamport timestamp",
       channelId = channelId, msgTs = msgTs, error = getCurrentExceptionMsg()
+    err(ReliabilityError.reInternalError)
 
 proc newHistoryEntry*(
     messageId: SdsMessageID, retrievalHint: seq[byte] = @[]
@@ -167,7 +185,7 @@ proc isInResponseGroup*(
 
 proc getRecentHistoryEntries*(
     rm: ReliabilityManager, n: int, channelId: SdsChannelID
-): Future[seq[HistoryEntry]] {.async: (raises: []).} =
+): Future[Result[seq[HistoryEntry], ReliabilityError]] {.async: (raises: []).} =
   ## Get recent history entries for sending in causal history.
   ## Populates retrieval hints and senderId (SDS-R) for each entry.
   try:
@@ -184,16 +202,17 @@ proc getRecentHistoryEntries*(
           {.cast(raises: []).}:
             entry.retrievalHint = rm.onRetrievalHint(msgId)
           if entry.retrievalHint.len > 0:
-            await rm.persistence.setRetrievalHint(msgId, entry.retrievalHint)
+            (await rm.persistence.setRetrievalHint(msgId, entry.retrievalHint)).isOkOr:
+              return err(reliabilityErr(error))
         entry.senderId = channel.messageHistory[msgId].senderId
         entries.add(entry)
-      return entries
+      ok(entries)
     else:
-      return @[]
+      ok(newSeq[HistoryEntry]())
   except CatchableError:
     error "Failed to get recent history entries",
       channelId = channelId, n = n, error = getCurrentExceptionMsg()
-    return @[]
+    err(ReliabilityError.reInternalError)
 
 proc checkDependencies*(
     rm: ReliabilityManager, deps: seq[HistoryEntry], channelId: SdsChannelID
@@ -269,7 +288,7 @@ proc getIncomingBuffer*(
 
 proc getOrCreateChannel*(
     rm: ReliabilityManager, channelId: SdsChannelID
-): Future[ChannelContext] {.async: (raises: [CatchableError]).} =
+): Future[Result[ChannelContext, ReliabilityError]] {.async: (raises: []).} =
   ## 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
@@ -281,7 +300,8 @@ proc getOrCreateChannel*(
           rm.config.bloomFilterCapacity, rm.config.bloomFilterErrorRate
         )
       )
-      let snapshot = await rm.persistence.loadAllForChannel(channelId)
+      let snapshot = (await rm.persistence.loadAllForChannel(channelId)).valueOr:
+        return err(reliabilityErr(error))
       channel.lamportTimestamp = snapshot.lamportTimestamp
       for msg in snapshot.messageHistory:
         channel.messageHistory[msg.messageId] = msg
@@ -295,25 +315,21 @@ proc getOrCreateChannel*(
       for (msgId, entry) in snapshot.incomingRepairBuffer:
         channel.incomingRepairBuffer[msgId] = entry
       rm.channels[channelId] = channel
-    return rm.channels[channelId]
-  except CatchableError as e:
+    ok(rm.channels[channelId])
+  except CatchableError:
     error "Failed to get or create channel",
       channelId = channelId, error = getCurrentExceptionMsg()
-    raise e
+    err(ReliabilityError.reInternalError)
 
 proc ensureChannel*(
     rm: ReliabilityManager, channelId: SdsChannelID
 ): Future[Result[void, ReliabilityError]] {.async: (raises: []).} =
   try:
     await rm.lock.acquire()
     try:
-      try:
-        discard await rm.getOrCreateChannel(channelId)
-        return ok()
-      except CatchableError:
-        error "Failed to ensure channel",
-          channelId = channelId, msg = getCurrentExceptionMsg()
-        return err(ReliabilityError.reInternalError)
+      (await rm.getOrCreateChannel(channelId)).isOkOr:
+        return err(error)
+      return ok()
     finally:
       rm.lock.release()
   except CatchableError:
@@ -330,7 +346,8 @@ proc removeChannel*(
       try:
         if channelId in rm.channels:
           let channel = rm.channels[channelId]
-          await rm.dropChannelFromPersistence(channelId)
+          (await rm.dropChannelFromPersistence(channelId)).isOkOr:
+            return err(error)
           channel.outgoingBuffer.setLen(0)
           channel.incomingBuffer.clear()
           channel.messageHistory.clear()

sds/sds_utils.nim

@@ -1,11 +1,12 @@
-import chronos
+import chronos, results
 import ./sds_message_id
 import ./sds_message
 import ./unacknowledged_message
 import ./incoming_message
 import ./repair_entry
 export
-  sds_message_id, sds_message, unacknowledged_message, incoming_message, repair_entry
+  results, sds_message_id, sds_message, unacknowledged_message, incoming_message,
+  repair_entry
 
 ## SDS state persistence interface (issue #64).
 ##
@@ -18,6 +19,14 @@ export
 ## All proc fields are async (return `Future`) so backends can do real I/O
 ## without blocking the Chronos event loop the manager runs on.
 ##
+## Every field returns a `Result` so backend failures are propagated to nim-sds
+## rather than swallowed by the backend. Mutating ops return
+## `Result[void, string]`; the getter (`loadAllForChannel`) returns
+## `Result[ChannelSnapshot, string]`. The error is a backend-supplied message;
+## nim-sds maps it to `ReliabilityError.rePersistenceError` and surfaces it on
+## the corresponding public API call. The contract still forbids raising
+## (`raises: []`): failure must travel through the `Result`, not an exception.
+##
 ## 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.
@@ -44,120 +53,125 @@ type
     ## in-memory state in lockstep without blocking the event loop.
 
     # Per-channel lamport clock
-    saveLamport*: proc(channelId: SdsChannelID, lamport: int64): Future[void] {.
-      async: (raises: []), gcsafe
-    .}
+    saveLamport*: proc(
+      channelId: SdsChannelID, lamport: int64
+    ): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
 
     # Local log (delivered messages)
-    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
-    .}
+    appendLogEntry*: proc(
+      channelId: SdsChannelID, msg: SdsMessage
+    ): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
+    removeLogEntry*: proc(
+      channelId: SdsChannelID, msgId: SdsMessageID
+    ): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
+    setRetrievalHint*: proc(
+      msgId: SdsMessageID, hint: seq[byte]
+    ): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
 
     # Outgoing unacknowledged buffer
     saveOutgoing*: proc(
       channelId: SdsChannelID, msg: UnacknowledgedMessage
-    ): Future[void] {.async: (raises: []), gcsafe.}
-    removeOutgoing*: proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {.
-      async: (raises: []), gcsafe
-    .}
+    ): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
+    removeOutgoing*: proc(
+      channelId: SdsChannelID, msgId: SdsMessageID
+    ): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
 
     # Incoming dependency-waiting buffer
-    saveIncoming*: proc(channelId: SdsChannelID, msg: IncomingMessage): Future[void] {.
-      async: (raises: []), gcsafe
-    .}
-    removeIncoming*: proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {.
-      async: (raises: []), gcsafe
-    .}
+    saveIncoming*: proc(
+      channelId: SdsChannelID, msg: IncomingMessage
+    ): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
+    removeIncoming*: proc(
+      channelId: SdsChannelID, msgId: SdsMessageID
+    ): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
 
     # SDS-R outgoing repair buffer
     saveOutgoingRepair*: proc(
       channelId: SdsChannelID, msgId: SdsMessageID, entry: OutgoingRepairEntry
-    ) {.async: (raises: []).}
+    ): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
     removeOutgoingRepair*: proc(
       channelId: SdsChannelID, msgId: SdsMessageID
-    ): Future[void] {.async: (raises: []), gcsafe.}
+    ): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
 
     # SDS-R incoming repair buffer
     saveIncomingRepair*: proc(
       channelId: SdsChannelID, msgId: SdsMessageID, entry: IncomingRepairEntry
-    ) {.async: (raises: []).}
+    ): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
     removeIncomingRepair*: proc(
       channelId: SdsChannelID, msgId: SdsMessageID
-    ): Future[void] {.async: (raises: []), gcsafe.}
+    ): Future[Result[void, string]] {.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): Future[void] {.async: (raises: []), gcsafe.}
-
-    # Bootstrap on `addChannel` / `getOrCreateChannel`.
-    loadAllForChannel*: proc(channelId: SdsChannelID): Future[ChannelSnapshot] {.
+    dropChannel*: proc(channelId: SdsChannelID): Future[Result[void, string]] {.
       async: (raises: []), gcsafe
     .}
 
+    # Bootstrap on `addChannel` / `getOrCreateChannel`.
+    loadAllForChannel*: proc(
+      channelId: SdsChannelID
+    ): Future[Result[ChannelSnapshot, string]] {.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) {.async: (raises: []).} =
-      discard,
+    saveLamport: proc(
+        channelId: SdsChannelID, lamport: int64
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     appendLogEntry: proc(
         channelId: SdsChannelID, msg: SdsMessage
-    ) {.async: (raises: []).} =
-      discard,
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     removeLogEntry: proc(
         channelId: SdsChannelID, msgId: SdsMessageID
-    ) {.async: (raises: []).} =
-      discard,
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     setRetrievalHint: proc(
         msgId: SdsMessageID, hint: seq[byte]
-    ) {.async: (raises: []).} =
-      discard,
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     saveOutgoing: proc(
         channelId: SdsChannelID, msg: UnacknowledgedMessage
-    ) {.async: (raises: []).} =
-      discard,
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     removeOutgoing: proc(
         channelId: SdsChannelID, msgId: SdsMessageID
-    ) {.async: (raises: []).} =
-      discard,
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     saveIncoming: proc(
         channelId: SdsChannelID, msg: IncomingMessage
-    ) {.async: (raises: []).} =
-      discard,
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     removeIncoming: proc(
         channelId: SdsChannelID, msgId: SdsMessageID
-    ) {.async: (raises: []).} =
-      discard,
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     saveOutgoingRepair: proc(
         channelId: SdsChannelID, msgId: SdsMessageID, entry: OutgoingRepairEntry
-    ) {.async: (raises: []).} =
-      discard,
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     removeOutgoingRepair: proc(
         channelId: SdsChannelID, msgId: SdsMessageID
-    ) {.async: (raises: []).} =
-      discard,
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     saveIncomingRepair: proc(
         channelId: SdsChannelID, msgId: SdsMessageID, entry: IncomingRepairEntry
-    ) {.async: (raises: []).} =
-      discard,
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     removeIncomingRepair: proc(
         channelId: SdsChannelID, msgId: SdsMessageID
-    ) {.async: (raises: []).} =
-      discard,
-    dropChannel: proc(channelId: SdsChannelID) {.async: (raises: []).} =
-      discard,
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
+    dropChannel: proc(
+        channelId: SdsChannelID
+    ): Future[Result[void, string]] {.async: (raises: []).} =
+      ok(),
     loadAllForChannel: proc(
         channelId: SdsChannelID
-    ): Future[ChannelSnapshot] {.async: (raises: []).} =
-      return ChannelSnapshot(),
+    ): Future[Result[ChannelSnapshot, string]] {.async: (raises: []).} =
+      ok(ChannelSnapshot()),
   )

sds/types/persistence.nim

@@ -5,3 +5,4 @@ type ReliabilityError* {.pure.} = enum
   reSerializationError
   reDeserializationError
   reMessageTooLarge
+  rePersistenceError ## A persistence backend operation (read or write) failed.