mirror of
https://github.com/logos-messaging/nim-sds.git
synced 2026-08-03 05:23:15 +00:00
refactor(persistence): strip legacy interface from protocol path; migrate tests to V2 (phase 2B+2C+2D)
End-state of phase 2: the protocol code no longer issues any legacy
fine-grained Persistence calls. All state survives via the snapshot-based
PersistenceV2 interface — one trySaveMeta per op end, plus tryUpdateHistory
batched inside addToHistory. The legacy Persistence field on
ReliabilityManager remains for backwards compatibility; phase 3 deletes it.
Protocol changes (sds.nim, sds/sds_utils.nim):
- reviewAckStatus, processIncomingBuffer, updateLamportTimestamp →
pure in-memory; no per-mutation persistence.
- addToHistory: replaces appendLogEntry+removeLogEntry with a single
tryUpdateHistory call carrying (append, evict) atomically.
- getRecentHistoryEntries: setRetrievalHint switched to V2; non-fatal.
- wrapOutgoingMessage, unwrapReceivedMessage, markDependenciesMet:
all per-row saveOutgoing / removeOutgoing / saveIncoming /
removeIncoming / saveOutgoingRepair / removeOutgoingRepair /
saveIncomingRepair / removeIncomingRepair calls removed (16 call
sites in total). State is captured by the op-end trySaveMeta added
in phase 2A.
- getOrCreateChannel: bootstraps from persistenceV2.loadChannel.
- dropChannelFromPersistence: uses persistenceV2.dropChannel.
Failure policy (PLAN_SNAPSHOT_PERSISTENCE.md §8):
- Foreground ops (wrap, unwrap, markDeps, sweeps): non-fatal —
trySaveMeta / tryUpdateHistory log and continue; the protocol op
returns ok regardless of disk failure. In-memory state is the source
of truth; the next op re-issues a complete snapshot and disk catches
up automatically.
- Durability-intent ops (removeChannel, resetReliabilityManager via
dropChannelFromPersistence; getOrCreateChannel via loadChannel):
still propagate rePersistenceError, because the caller asked us to
confirm a disk operation and we cannot silently lie.
Test infrastructure:
- tests/in_memory_persistence_v2.nim: new V2 adapter mock that
decomposes the meta blob into the existing InMemoryStore shape so
test assertions on store.outgoing / store.incoming / etc. continue to
work without change.
- tests/test_persistence.nim: 17 tests, all rewritten against V2.
- 13 state-survival tests carry over with identical assertions.
- "loadChannel failure surfaces as err on bootstrap" — bootstrap
keeps durability-intent semantics.
- "saveChannelMeta failure during send does NOT surface" — deliberate
inversion of the legacy "write failure surfaces as err" test. Asserts
the new non-fatal policy: op returns ok, in-memory state correct,
disk re-syncs on the next op.
- "updateHistory failure during send does NOT surface" — same policy
applied to the history path.
- "dropChannel failure during removeChannel surfaces as err" — kept.
- All 17 tests pass.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
NagyZoltanPeter
parent
6019066ccc
commit
a1147c96af
87
sds.nim
87
sds.nim
@ -79,10 +79,10 @@ proc reviewAckStatus(
|
||||
inc i
|
||||
|
||||
for k in countdown(toDelete.high, 0):
|
||||
let (idx, ackedId) = toDelete[k]
|
||||
channel.outgoingBuffer.delete(idx)
|
||||
(await rm.persistence.removeOutgoing(msg.channelId, ackedId)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
# Phase 2B: in-memory deletion only; the caller's op-end trySaveMeta
|
||||
# captures the new outgoingBuffer state. The msgId half of the
|
||||
# tuple is unused now that there is no per-row persistence call.
|
||||
channel.outgoingBuffer.delete(toDelete[k][0])
|
||||
ok()
|
||||
except CatchableError:
|
||||
error "Failed to review ack status", msg = getCurrentExceptionMsg()
|
||||
@ -133,8 +133,7 @@ proc wrapOutgoingMessage*(
|
||||
expiredKeys.add(eligible[i][0])
|
||||
for key in expiredKeys:
|
||||
channel.outgoingRepairBuffer.del(key)
|
||||
(await rm.persistence.removeOutgoingRepair(channelId, key)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
# Phase 2B: in-memory deletion only; op-end trySaveMeta covers it.
|
||||
|
||||
let causalHistory = (
|
||||
await rm.getRecentHistoryEntries(rm.config.maxCausalHistory, channelId)
|
||||
@ -155,8 +154,7 @@ proc wrapOutgoingMessage*(
|
||||
message = msg, sendTime = getTime(), resendAttempts = 0
|
||||
)
|
||||
channel.outgoingBuffer.add(unackMsg)
|
||||
(await rm.persistence.saveOutgoing(channelId, unackMsg)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
# Phase 2B: in-memory append only; op-end trySaveMeta covers it.
|
||||
|
||||
channel.bloomFilter.add(msg.messageId)
|
||||
# The full SdsMessage carries senderId and content, so a single
|
||||
@ -222,20 +220,17 @@ proc processIncomingBuffer(
|
||||
for remainingId, entry in channel.incomingBuffer:
|
||||
if remainingId notin processed:
|
||||
if msgId in entry.missingDeps:
|
||||
# Phase 2B: in-memory dep-set shrink only; the parent op
|
||||
# (unwrap / markDeps) issues a single trySaveMeta at its
|
||||
# end that captures the final incomingBuffer state.
|
||||
channel.incomingBuffer[remainingId].missingDeps.excl(msgId)
|
||||
(
|
||||
await rm.persistence.saveIncoming(
|
||||
channelId, channel.incomingBuffer[remainingId]
|
||||
)
|
||||
).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
if channel.incomingBuffer[remainingId].missingDeps.len == 0:
|
||||
readyToProcess.add(remainingId)
|
||||
|
||||
for msgId in processed:
|
||||
# Phase 2B: in-memory deletion only; parent op's trySaveMeta covers
|
||||
# the drained buffer state.
|
||||
channel.incomingBuffer.del(msgId)
|
||||
(await rm.persistence.removeIncoming(channelId, msgId)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
ok()
|
||||
finally:
|
||||
rm.lock.release()
|
||||
@ -265,12 +260,9 @@ proc unwrapReceivedMessage*(
|
||||
|
||||
# SDS-R: opportunistic repair-buffer cleanup — applies to duplicates too,
|
||||
# so rebroadcasts cancel redundant responses on peers that already have the message.
|
||||
# Phase 2B: in-memory deletes only; op-end trySaveMeta covers it.
|
||||
channel.outgoingRepairBuffer.del(msg.messageId)
|
||||
(await rm.persistence.removeOutgoingRepair(channelId, msg.messageId)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
channel.incomingRepairBuffer.del(msg.messageId)
|
||||
(await rm.persistence.removeIncomingRepair(channelId, msg.messageId)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
|
||||
if msg.messageId in channel.messageHistory:
|
||||
# Duplicate: no state change beyond the repair-buffer cleanup above.
|
||||
@ -292,10 +284,9 @@ proc unwrapReceivedMessage*(
|
||||
# to confidently rebroadcast.
|
||||
let now = getTime()
|
||||
for repairEntry in msg.repairRequest:
|
||||
# Remove from our own outgoing repair buffer (someone else is also requesting)
|
||||
# Remove from our own outgoing repair buffer (someone else is also requesting).
|
||||
# Phase 2B: in-memory delete only; op-end trySaveMeta covers it.
|
||||
channel.outgoingRepairBuffer.del(repairEntry.messageId)
|
||||
(await rm.persistence.removeOutgoingRepair(channelId, repairEntry.messageId)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
if repairEntry.messageId in channel.messageHistory and rm.participantId.len > 0 and
|
||||
repairEntry.senderId.len > 0:
|
||||
if isInResponseGroup(
|
||||
@ -314,13 +305,8 @@ proc unwrapReceivedMessage*(
|
||||
cachedMessage: serialized.get(),
|
||||
minTimeRepairResp: now + tResp,
|
||||
)
|
||||
# Phase 2B: in-memory insert only; op-end trySaveMeta covers it.
|
||||
channel.incomingRepairBuffer[repairEntry.messageId] = inEntry
|
||||
(
|
||||
await rm.persistence.saveIncomingRepair(
|
||||
channelId, repairEntry.messageId, inEntry
|
||||
)
|
||||
).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
|
||||
var missingDeps = rm.checkDependencies(msg.causalHistory, channelId)
|
||||
|
||||
@ -333,25 +319,19 @@ proc unwrapReceivedMessage*(
|
||||
if depsInBuffer:
|
||||
let entry =
|
||||
IncomingMessage.init(message = msg, missingDeps = initHashSet[SdsMessageID]())
|
||||
# Phase 2B: in-memory insert only; op-end trySaveMeta covers it.
|
||||
channel.incomingBuffer[msg.messageId] = entry
|
||||
(await rm.persistence.saveIncoming(channelId, entry)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
else:
|
||||
(await rm.addToHistory(msg, channelId)).isOkOr:
|
||||
return err(error)
|
||||
# Unblock any buffered messages that were waiting on this one.
|
||||
var unblocked: seq[SdsMessageID] = @[]
|
||||
# Phase 2B: in-memory dep-set shrink only; op-end trySaveMeta and
|
||||
# the subsequent processIncomingBuffer cascade (which is also
|
||||
# in-memory only) leave the final state on the in-memory side, and
|
||||
# the op-end trySaveMeta snapshots it.
|
||||
for pendingId, entry in channel.incomingBuffer:
|
||||
if msg.messageId in entry.missingDeps:
|
||||
channel.incomingBuffer[pendingId].missingDeps.excl(msg.messageId)
|
||||
unblocked.add(pendingId)
|
||||
for pendingId in unblocked:
|
||||
(
|
||||
await rm.persistence.saveIncoming(
|
||||
channelId, channel.incomingBuffer[pendingId]
|
||||
)
|
||||
).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
(await rm.processIncomingBuffer(channelId)).isOkOr:
|
||||
return err(error)
|
||||
if not rm.onMessageReady.isNil():
|
||||
@ -361,9 +341,8 @@ proc unwrapReceivedMessage*(
|
||||
let entry = IncomingMessage.init(
|
||||
message = msg, missingDeps = missingDeps.getMessageIds().toHashSet()
|
||||
)
|
||||
# Phase 2B: in-memory insert only; op-end trySaveMeta covers it.
|
||||
channel.incomingBuffer[msg.messageId] = entry
|
||||
(await rm.persistence.saveIncoming(channelId, entry)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
if not rm.onMissingDependencies.isNil():
|
||||
{.cast(raises: []).}:
|
||||
rm.onMissingDependencies(msg.messageId, missingDeps, channelId)
|
||||
@ -378,13 +357,8 @@ proc unwrapReceivedMessage*(
|
||||
)
|
||||
let outEntry =
|
||||
OutgoingRepairEntry(outHistEntry: dep, minTimeRepairReq: now + tReq)
|
||||
# Phase 2B: in-memory insert only; op-end trySaveMeta covers it.
|
||||
channel.outgoingRepairBuffer[dep.messageId] = outEntry
|
||||
(
|
||||
await rm.persistence.saveOutgoingRepair(
|
||||
channelId, dep.messageId, outEntry
|
||||
)
|
||||
).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
|
||||
# Phase 2.5: single V2 meta snapshot covers ALL three paths
|
||||
# (deps-met-fresh, deps-met-buffered, missing-deps). Buffer mutations,
|
||||
@ -411,26 +385,15 @@ proc markDependenciesMet*(
|
||||
if not channel.bloomFilter.contains(msgId):
|
||||
channel.bloomFilter.add(msgId)
|
||||
|
||||
var unblocked: seq[SdsMessageID] = @[]
|
||||
# Phase 2B: in-memory dep-set shrink + repair-buffer dels only; the
|
||||
# op-end trySaveMeta below covers all mutations atomically.
|
||||
for pendingId, entry in channel.incomingBuffer:
|
||||
if msgId in entry.missingDeps:
|
||||
channel.incomingBuffer[pendingId].missingDeps.excl(msgId)
|
||||
unblocked.add(pendingId)
|
||||
for pendingId in unblocked:
|
||||
(
|
||||
await rm.persistence.saveIncoming(
|
||||
channelId, channel.incomingBuffer[pendingId]
|
||||
)
|
||||
).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
|
||||
# SDS-R: clear from repair buffers (dependency now met)
|
||||
# SDS-R: clear from repair buffers (dependency now met).
|
||||
channel.outgoingRepairBuffer.del(msgId)
|
||||
(await rm.persistence.removeOutgoingRepair(channelId, msgId)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
channel.incomingRepairBuffer.del(msgId)
|
||||
(await rm.persistence.removeIncomingRepair(channelId, msgId)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
|
||||
(await rm.processIncomingBuffer(channelId)).isOkOr:
|
||||
return err(error)
|
||||
|
||||
@ -88,7 +88,11 @@ proc dropChannelFromPersistence*(
|
||||
## 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)).isOkOr:
|
||||
##
|
||||
## Phase 2D: uses `persistenceV2.dropChannel`. This op DOES propagate
|
||||
## err on failure (durability is the semantic intent — the caller asked
|
||||
## us to confirm a disk wipe; we cannot silently lie). See PLAN §8.
|
||||
(await rm.persistenceV2.dropChannel(channelId)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
ok()
|
||||
|
||||
@ -143,20 +147,25 @@ proc addToHistory*(
|
||||
## eldest entries when the bound is exceeded. The full SdsMessage is kept so
|
||||
## senderId is available for downstream causal-history population and the
|
||||
## bytes can be re-serialized on demand to answer SDS-R repair requests.
|
||||
## Persistence (phase 2B): mutations are batched into ONE V2
|
||||
## `tryUpdateHistory` call at the end of this proc (append the new
|
||||
## message + evict whatever rolled past `maxMessageHistory`). Failure is
|
||||
## non-fatal: in-memory state is the source of truth, the next op's
|
||||
## history update re-synchronises disk. Legacy per-row `appendLogEntry`
|
||||
## / `removeLogEntry` calls are removed.
|
||||
try:
|
||||
if channelId in rm.channels:
|
||||
let channel = rm.channels[channelId]
|
||||
channel.messageHistory[msg.messageId] = msg
|
||||
(await rm.persistence.appendLogEntry(channelId, msg)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
var evicted: seq[SdsMessageID] = @[]
|
||||
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)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
evicted.add(firstKey)
|
||||
await rm.tryUpdateHistory(channelId, @[msg], evicted)
|
||||
ok()
|
||||
except CatchableError:
|
||||
error "Failed to add to history",
|
||||
@ -166,12 +175,13 @@ proc addToHistory*(
|
||||
proc updateLamportTimestamp*(
|
||||
rm: ReliabilityManager, msgTs: int64, channelId: SdsChannelID
|
||||
): Future[Result[void, ReliabilityError]] {.async: (raises: []).} =
|
||||
## Pure in-memory update (phase 2B). The new lamport value is captured
|
||||
## by the op-end `trySaveMeta` issued by the calling protocol op; no
|
||||
## per-mutation persistence call here.
|
||||
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)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
ok()
|
||||
except CatchableError:
|
||||
error "Failed to update lamport timestamp",
|
||||
@ -260,8 +270,15 @@ proc getRecentHistoryEntries*(
|
||||
{.cast(raises: []).}:
|
||||
entry.retrievalHint = rm.onRetrievalHint(msgId)
|
||||
if entry.retrievalHint.len > 0:
|
||||
(await rm.persistence.setRetrievalHint(msgId, entry.retrievalHint)).isOkOr:
|
||||
return err(reliabilityErr(error))
|
||||
# Phase 2B: best-effort hint persistence via V2. Non-fatal —
|
||||
# hints are an optimisation; a missing hint just means the
|
||||
# peer falls back to slower retrieval.
|
||||
let hintRes = await rm.persistenceV2.setRetrievalHint(
|
||||
msgId, entry.retrievalHint
|
||||
)
|
||||
if hintRes.isErr:
|
||||
warn "retrieval hint save failed; continuing",
|
||||
msgId = msgId, detail = hintRes.error
|
||||
entry.senderId = channel.messageHistory[msgId].senderId
|
||||
entries.add(entry)
|
||||
ok(entries)
|
||||
@ -351,6 +368,11 @@ proc getOrCreateChannel*(
|
||||
## persistence backend if it does not yet exist in memory. The bloom filter
|
||||
## is rebuilt deterministically from the loaded message history rather than
|
||||
## persisted directly. Caller is expected to hold rm.lock.
|
||||
##
|
||||
## Phase 2C: bootstrap via `persistenceV2.loadChannel`. Bootstrap DOES
|
||||
## propagate err on load failure — the caller asked us to materialise a
|
||||
## channel and we cannot do that without knowing the prior state. See
|
||||
## PLAN §8.
|
||||
try:
|
||||
if channelId notin rm.channels:
|
||||
let channel = ChannelContext.new(
|
||||
@ -358,20 +380,22 @@ proc getOrCreateChannel*(
|
||||
rm.config.bloomFilterCapacity, rm.config.bloomFilterErrorRate
|
||||
)
|
||||
)
|
||||
let snapshot = (await rm.persistence.loadAllForChannel(channelId)).valueOr:
|
||||
let data = (await rm.persistenceV2.loadChannel(channelId)).valueOr:
|
||||
return err(reliabilityErr(error))
|
||||
channel.lamportTimestamp = snapshot.lamportTimestamp
|
||||
for msg in snapshot.messageHistory:
|
||||
channel.lamportTimestamp = data.meta.lamportTimestamp
|
||||
# Backend contract: messageHistory MUST be ordered oldest-first.
|
||||
# If a backend violates this, FIFO eviction breaks across restarts.
|
||||
for msg in data.messageHistory:
|
||||
channel.messageHistory[msg.messageId] = msg
|
||||
channel.bloomFilter.add(msg.messageId)
|
||||
for unack in snapshot.outgoingBuffer:
|
||||
for unack in data.meta.outgoingBuffer:
|
||||
channel.outgoingBuffer.add(unack)
|
||||
for incoming in snapshot.incomingBuffer:
|
||||
for incoming in data.meta.incomingBuffer:
|
||||
channel.incomingBuffer[incoming.message.messageId] = incoming
|
||||
for (msgId, entry) in snapshot.outgoingRepairBuffer:
|
||||
channel.outgoingRepairBuffer[msgId] = entry
|
||||
for (msgId, entry) in snapshot.incomingRepairBuffer:
|
||||
channel.incomingRepairBuffer[msgId] = entry
|
||||
for kv in data.meta.outgoingRepairBuffer:
|
||||
channel.outgoingRepairBuffer[kv.messageId] = kv.entry
|
||||
for kv in data.meta.incomingRepairBuffer:
|
||||
channel.incomingRepairBuffer[kv.messageId] = kv.entry
|
||||
rm.channels[channelId] = channel
|
||||
ok(rm.channels[channelId])
|
||||
except CatchableError:
|
||||
|
||||
119
tests/in_memory_persistence_v2.nim
Normal file
119
tests/in_memory_persistence_v2.nim
Normal file
@ -0,0 +1,119 @@
|
||||
## V2 test-only backend that adapts the snapshot-based `PersistenceV2`
|
||||
## interface onto the same `InMemoryStore` shape used by the legacy mock.
|
||||
##
|
||||
## Tests can assert against `store.outgoing`, `store.log`, etc. exactly as
|
||||
## they did against the legacy backend; this adapter decomposes the
|
||||
## snapshot blob into the same denormalised tables. That keeps the
|
||||
## state-survival tests in tests/test_persistence.nim portable across the
|
||||
## migration without rewriting every assertion.
|
||||
##
|
||||
## `failingOps` injects backend failures. Op names match the `PersistenceV2`
|
||||
## field names: "saveChannelMeta", "updateHistory", "loadChannel",
|
||||
## "dropChannel", "setRetrievalHint".
|
||||
|
||||
import std/[tables, sets]
|
||||
import chronos
|
||||
import sds
|
||||
import ./in_memory_persistence
|
||||
|
||||
export in_memory_persistence
|
||||
|
||||
proc newInMemoryPersistenceV2*(store: InMemoryStore): PersistenceV2 =
|
||||
PersistenceV2(
|
||||
saveChannelMeta: proc(
|
||||
channelId: SdsChannelID, meta: ChannelMeta
|
||||
): Future[Result[void, string]] {.async: (raises: []).} =
|
||||
if "saveChannelMeta" in store.failingOps:
|
||||
return err("injected backend failure: saveChannelMeta")
|
||||
{.cast(raises: []).}:
|
||||
# Lamport.
|
||||
store.lamports[channelId] = meta.lamportTimestamp
|
||||
|
||||
# Outgoing buffer — replace existing rows wholesale (snapshot is
|
||||
# the complete state, not a delta).
|
||||
store.outgoing[channelId] =
|
||||
initOrderedTable[SdsMessageID, UnacknowledgedMessage]()
|
||||
for u in meta.outgoingBuffer:
|
||||
store.outgoing[channelId][u.message.messageId] = u
|
||||
|
||||
# Incoming buffer.
|
||||
store.incoming[channelId] =
|
||||
initOrderedTable[SdsMessageID, IncomingMessage]()
|
||||
for m in meta.incomingBuffer:
|
||||
store.incoming[channelId][m.message.messageId] = m
|
||||
|
||||
# Repair buffers.
|
||||
store.outgoingRepair[channelId] =
|
||||
initOrderedTable[SdsMessageID, OutgoingRepairEntry]()
|
||||
for kv in meta.outgoingRepairBuffer:
|
||||
store.outgoingRepair[channelId][kv.messageId] = kv.entry
|
||||
store.incomingRepair[channelId] =
|
||||
initOrderedTable[SdsMessageID, IncomingRepairEntry]()
|
||||
for kv in meta.incomingRepairBuffer:
|
||||
store.incomingRepair[channelId][kv.messageId] = kv.entry
|
||||
ok(),
|
||||
updateHistory: proc(
|
||||
channelId: SdsChannelID, update: HistoryUpdate
|
||||
): Future[Result[void, string]] {.async: (raises: []).} =
|
||||
if "updateHistory" in store.failingOps:
|
||||
return err("injected backend failure: updateHistory")
|
||||
{.cast(raises: []).}:
|
||||
if channelId notin store.log:
|
||||
store.log[channelId] = initOrderedTable[SdsMessageID, SdsMessage]()
|
||||
for m in update.append:
|
||||
store.log[channelId][m.messageId] = m
|
||||
for id in update.evict:
|
||||
store.log[channelId].del(id)
|
||||
ok(),
|
||||
loadChannel: proc(
|
||||
channelId: SdsChannelID
|
||||
): Future[Result[ChannelData, string]] {.async: (raises: []).} =
|
||||
if "loadChannel" in store.failingOps:
|
||||
return err("injected backend failure: loadChannel")
|
||||
{.cast(raises: []).}:
|
||||
var data = ChannelData.init()
|
||||
if channelId in store.lamports:
|
||||
data.meta.lamportTimestamp = store.lamports[channelId]
|
||||
if channelId in store.outgoing:
|
||||
for u in store.outgoing[channelId].values:
|
||||
data.meta.outgoingBuffer.add(u)
|
||||
if channelId in store.incoming:
|
||||
for m in store.incoming[channelId].values:
|
||||
data.meta.incomingBuffer.add(m)
|
||||
if channelId in store.outgoingRepair:
|
||||
for id, e in store.outgoingRepair[channelId].pairs:
|
||||
data.meta.outgoingRepairBuffer.add(
|
||||
OutgoingRepairKV(messageId: id, entry: e)
|
||||
)
|
||||
if channelId in store.incomingRepair:
|
||||
for id, e in store.incomingRepair[channelId].pairs:
|
||||
data.meta.incomingRepairBuffer.add(
|
||||
IncomingRepairKV(messageId: id, entry: e)
|
||||
)
|
||||
if channelId in store.log:
|
||||
for m in store.log[channelId].values:
|
||||
data.messageHistory.add(m)
|
||||
return ok(data),
|
||||
dropChannel: proc(
|
||||
channelId: SdsChannelID
|
||||
): Future[Result[void, string]] {.async: (raises: []).} =
|
||||
if "dropChannel" in store.failingOps:
|
||||
return err("injected backend failure: dropChannel")
|
||||
{.cast(raises: []).}:
|
||||
store.lamports.del(channelId)
|
||||
store.log.del(channelId)
|
||||
store.outgoing.del(channelId)
|
||||
store.incoming.del(channelId)
|
||||
store.outgoingRepair.del(channelId)
|
||||
store.incomingRepair.del(channelId)
|
||||
store.dropChannelCalls[channelId] =
|
||||
store.dropChannelCalls.getOrDefault(channelId) + 1
|
||||
ok(),
|
||||
setRetrievalHint: proc(
|
||||
msgId: SdsMessageID, hint: seq[byte]
|
||||
): Future[Result[void, string]] {.async: (raises: []).} =
|
||||
if "setRetrievalHint" in store.failingOps:
|
||||
return err("injected backend failure: setRetrievalHint")
|
||||
store.hints[msgId] = hint
|
||||
ok(),
|
||||
)
|
||||
@ -1,18 +1,30 @@
|
||||
import results, std/[tables, sets, times]
|
||||
import sds
|
||||
import ./async_unittest
|
||||
import ./in_memory_persistence
|
||||
import ./in_memory_persistence_v2
|
||||
|
||||
converter toParticipantID(s: string): SdsParticipantID =
|
||||
s.SdsParticipantID
|
||||
|
||||
const testChannel = "testChannel"
|
||||
|
||||
suite "Persistence: write → restart → read-back":
|
||||
# Helper: build a ReliabilityManager wired only to the V2 in-memory
|
||||
# persistence (no legacy backend). Mirrors how production callers will
|
||||
# construct the manager once phase 3 deletes the legacy field.
|
||||
proc newV2Manager(
|
||||
store: InMemoryStore, config = defaultConfig()
|
||||
): ReliabilityManager =
|
||||
newReliabilityManager(
|
||||
participantId = "alice",
|
||||
config = config,
|
||||
persistenceV2 = newInMemoryPersistenceV2(store),
|
||||
)
|
||||
.get()
|
||||
|
||||
suite "Persistence (V2): write → restart → read-back":
|
||||
asyncTest "outgoing buffer survives restart":
|
||||
let store = newInMemoryStore()
|
||||
let p1 = newInMemoryPersistence(store)
|
||||
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
|
||||
let rm1 = newV2Manager(store)
|
||||
check (await rm1.ensureChannel(testChannel)).isOk()
|
||||
let wrapped = await rm1.wrapOutgoingMessage(@[1.byte, 2, 3], "msg-1", testChannel)
|
||||
check wrapped.isOk()
|
||||
@ -20,9 +32,8 @@ suite "Persistence: write → restart → read-back":
|
||||
check "msg-1" in store.outgoing[testChannel]
|
||||
await rm1.cleanup()
|
||||
|
||||
# Simulate restart: fresh manager, same backend
|
||||
let p2 = newInMemoryPersistence(store)
|
||||
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
|
||||
# Simulate restart: fresh manager, same backend.
|
||||
let rm2 = newV2Manager(store)
|
||||
check (await rm2.ensureChannel(testChannel)).isOk()
|
||||
let buf = await rm2.getOutgoingBuffer(testChannel)
|
||||
check buf.len == 1
|
||||
@ -31,22 +42,25 @@ suite "Persistence: write → restart → read-back":
|
||||
|
||||
asyncTest "lamport clock survives restart":
|
||||
let store = newInMemoryStore()
|
||||
let p1 = newInMemoryPersistence(store)
|
||||
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
|
||||
let rm1 = newV2Manager(store)
|
||||
check (await rm1.ensureChannel(testChannel)).isOk()
|
||||
check (await rm1.updateLamportTimestamp(42, testChannel)).isOk()
|
||||
check store.lamports[testChannel] == 43 # max(42, 0) + 1
|
||||
# updateLamportTimestamp is now pure; the mutation is persisted by the
|
||||
# next op-end save. Drive a wrap to force a trySaveMeta.
|
||||
discard await rm1.wrapOutgoingMessage(@[byte(1)], "tick", testChannel)
|
||||
# max(42,0)+1 then max(getTime().toUnix, 43)+1; whatever wrap sets is
|
||||
# what we'll see. We just assert it stayed monotonic.
|
||||
check store.lamports[testChannel] >= 43
|
||||
let savedLamport = store.lamports[testChannel]
|
||||
await rm1.cleanup()
|
||||
|
||||
let p2 = newInMemoryPersistence(store)
|
||||
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
|
||||
let rm2 = newV2Manager(store)
|
||||
check (await rm2.ensureChannel(testChannel)).isOk()
|
||||
check rm2.channels[testChannel].lamportTimestamp == 43
|
||||
check rm2.channels[testChannel].lamportTimestamp == savedLamport
|
||||
|
||||
asyncTest "delivered messages survive restart and rebuild bloom":
|
||||
let store = newInMemoryStore()
|
||||
let p1 = newInMemoryPersistence(store)
|
||||
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
|
||||
let rm1 = newV2Manager(store)
|
||||
check (await rm1.ensureChannel(testChannel)).isOk()
|
||||
let msg = SdsMessage.init(
|
||||
messageId = "delivered-1",
|
||||
@ -61,24 +75,22 @@ suite "Persistence: write → restart → read-back":
|
||||
check store.log[testChannel].len == 1
|
||||
await rm1.cleanup()
|
||||
|
||||
let p2 = newInMemoryPersistence(store)
|
||||
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
|
||||
let rm2 = newV2Manager(store)
|
||||
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
|
||||
# Bloom filter rebuilt from log on bootstrap.
|
||||
check ch.bloomFilter.contains("delivered-1")
|
||||
|
||||
asyncTest "ack removes outgoing entry from persistence":
|
||||
let store = newInMemoryStore()
|
||||
let p = newInMemoryPersistence(store)
|
||||
let rm = newReliabilityManager(participantId = "alice", persistence = p).get()
|
||||
let rm = newV2Manager(store)
|
||||
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
|
||||
# Synthesize an incoming message that ACKs msg-x via causal history.
|
||||
let ackMsg = SdsMessage.init(
|
||||
messageId = "ack-bearer",
|
||||
lamportTimestamp = 5,
|
||||
@ -95,10 +107,9 @@ suite "Persistence: write → restart → read-back":
|
||||
|
||||
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.
|
||||
# drop, not N per-row removes.
|
||||
let store = newInMemoryStore()
|
||||
let p = newInMemoryPersistence(store)
|
||||
let rm = newReliabilityManager(participantId = "alice", persistence = p).get()
|
||||
let rm = newV2Manager(store)
|
||||
check (await rm.ensureChannel(testChannel)).isOk()
|
||||
discard await rm.wrapOutgoingMessage(@[1.byte], "msg-r", testChannel)
|
||||
check store.outgoing[testChannel].len == 1
|
||||
@ -114,9 +125,9 @@ suite "Persistence: write → restart → read-back":
|
||||
check testChannel notin store.incomingRepair
|
||||
await rm.cleanup()
|
||||
|
||||
asyncTest "noOpPersistence keeps existing manager working":
|
||||
asyncTest "noOpPersistenceV2 keeps existing manager working":
|
||||
let rm = newReliabilityManager(participantId = "alice").get()
|
||||
# default no-op persistence
|
||||
# default no-op persistence (both legacy and V2)
|
||||
check (await rm.ensureChannel(testChannel)).isOk()
|
||||
let wrapped = await rm.wrapOutgoingMessage(@[1.byte], "msg-n", testChannel)
|
||||
check wrapped.isOk()
|
||||
@ -126,8 +137,7 @@ suite "Persistence: write → restart → read-back":
|
||||
|
||||
asyncTest "continue operating after restart: lamport stays monotonic":
|
||||
let store = newInMemoryStore()
|
||||
let p1 = newInMemoryPersistence(store)
|
||||
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
|
||||
let rm1 = newV2Manager(store)
|
||||
check (await rm1.ensureChannel(testChannel)).isOk()
|
||||
discard await rm1.wrapOutgoingMessage(@[1.byte], "m1", testChannel)
|
||||
let lamportAfterSession1 = store.lamports[testChannel]
|
||||
@ -135,8 +145,7 @@ suite "Persistence: write → restart → read-back":
|
||||
await rm1.cleanup()
|
||||
|
||||
# Restart and send another message — lamport must not regress.
|
||||
let p2 = newInMemoryPersistence(store)
|
||||
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
|
||||
let rm2 = newV2Manager(store)
|
||||
check (await rm2.ensureChannel(testChannel)).isOk()
|
||||
check rm2.channels[testChannel].lamportTimestamp == lamportAfterSession1
|
||||
discard await rm2.wrapOutgoingMessage(@[2.byte], "m2", testChannel)
|
||||
@ -148,16 +157,13 @@ suite "Persistence: write → restart → read-back":
|
||||
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()
|
||||
let rm = newV2Manager(store)
|
||||
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()
|
||||
let rmFinal = newV2Manager(store)
|
||||
check (await rmFinal.ensureChannel(testChannel)).isOk()
|
||||
let buf = await rmFinal.getOutgoingBuffer(testChannel)
|
||||
check buf.len == 3
|
||||
@ -171,8 +177,7 @@ suite "Persistence: write → restart → read-back":
|
||||
|
||||
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()
|
||||
let rm1 = newV2Manager(store)
|
||||
check (await rm1.ensureChannel(testChannel)).isOk()
|
||||
|
||||
# Receive a message whose causal-history references an unknown predecessor.
|
||||
@ -191,8 +196,7 @@ suite "Persistence: write → restart → read-back":
|
||||
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()
|
||||
let rm2 = newV2Manager(store)
|
||||
check (await rm2.ensureChannel(testChannel)).isOk()
|
||||
let inbuf = await rm2.getIncomingBuffer(testChannel)
|
||||
check "msg-with-deps" in inbuf
|
||||
@ -200,11 +204,8 @@ suite "Persistence: write → restart → read-back":
|
||||
await rm2.cleanup()
|
||||
|
||||
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()
|
||||
let rm1 = newV2Manager(store)
|
||||
check (await rm1.ensureChannel(testChannel)).isOk()
|
||||
discard await rm1.wrapOutgoingMessage(@[1.byte], "m-old", testChannel)
|
||||
check store.lamports[testChannel] > 0
|
||||
@ -213,8 +214,7 @@ suite "Persistence: write → restart → read-back":
|
||||
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()
|
||||
let rm2 = newV2Manager(store)
|
||||
check (await rm2.ensureChannel(testChannel)).isOk()
|
||||
check rm2.channels[testChannel].lamportTimestamp == 0
|
||||
let buf = await rm2.getOutgoingBuffer(testChannel)
|
||||
@ -223,11 +223,9 @@ suite "Persistence: write → restart → read-back":
|
||||
|
||||
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()
|
||||
let rm1 = newV2Manager(store)
|
||||
check (await rm1.ensureChannel(testChannel)).isOk()
|
||||
|
||||
# Receive a message that references an unknown dep — triggers SDS-R repair.
|
||||
let depMsg = SdsMessage.init(
|
||||
messageId = "msg-needs-repair",
|
||||
lamportTimestamp = 5,
|
||||
@ -244,13 +242,16 @@ suite "Persistence: write → restart → read-back":
|
||||
check originalTReqAt.toUnix > 0
|
||||
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()
|
||||
# Restart — repair entry must be back with the SAME absolute time.
|
||||
# Codec serialises Time as int64 unix milliseconds (PLAN §1.5), so the
|
||||
# restored Time may differ by sub-millisecond precision from the
|
||||
# original. Compare at second resolution which is what the protocol
|
||||
# actually relies on.
|
||||
let rm2 = newV2Manager(store)
|
||||
check (await rm2.ensureChannel(testChannel)).isOk()
|
||||
let buf = rm2.channels[testChannel].outgoingRepairBuffer
|
||||
check "missing-dep" in buf
|
||||
check buf["missing-dep"].minTimeRepairReq == originalTReqAt
|
||||
check buf["missing-dep"].minTimeRepairReq.toUnix == originalTReqAt.toUnix
|
||||
await rm2.cleanup()
|
||||
|
||||
asyncTest "FIFO eviction state survives restart":
|
||||
@ -259,11 +260,7 @@ suite "Persistence: write → restart → read-back":
|
||||
smallCfg.maxMessageHistory = 3
|
||||
smallCfg.bloomFilterCapacity = 3
|
||||
|
||||
let p1 = newInMemoryPersistence(store)
|
||||
let rm1 = newReliabilityManager(
|
||||
participantId = "alice", config = smallCfg, persistence = p1
|
||||
)
|
||||
.get()
|
||||
let rm1 = newV2Manager(store, smallCfg)
|
||||
check (await rm1.ensureChannel(testChannel)).isOk()
|
||||
# Add 5 delivered messages — first 2 should be evicted by FIFO.
|
||||
for i in 1 .. 5:
|
||||
@ -283,11 +280,7 @@ suite "Persistence: write → restart → read-back":
|
||||
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()
|
||||
let rm2 = newV2Manager(store, smallCfg)
|
||||
check (await rm2.ensureChannel(testChannel)).isOk()
|
||||
let history = rm2.channels[testChannel].messageHistory
|
||||
check history.len == 3
|
||||
@ -295,7 +288,7 @@ suite "Persistence: write → restart → read-back":
|
||||
check "m2" notin history
|
||||
check "m3" in history
|
||||
check "m5" in history
|
||||
# FIFO continues correctly after restart: adding m6 evicts m3, not a stale entry.
|
||||
# FIFO continues correctly after restart: adding m6 evicts m3.
|
||||
let m6 = SdsMessage.init(
|
||||
messageId = "m6",
|
||||
lamportTimestamp = 6,
|
||||
@ -312,8 +305,7 @@ suite "Persistence: write → restart → read-back":
|
||||
|
||||
asyncTest "dep-clear cascade resumes correctly across a restart":
|
||||
let store = newInMemoryStore()
|
||||
let p1 = newInMemoryPersistence(store)
|
||||
let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get()
|
||||
let rm1 = newV2Manager(store)
|
||||
check (await rm1.ensureChannel(testChannel)).isOk()
|
||||
|
||||
# Receive c (deps on b), then b (deps on a). Both must buffer.
|
||||
@ -341,9 +333,8 @@ suite "Persistence: write → restart → read-back":
|
||||
check "b" in store.incoming[testChannel]
|
||||
await rm1.cleanup()
|
||||
|
||||
# Restart — both still buffered, with intact missingDeps.
|
||||
let p2 = newInMemoryPersistence(store)
|
||||
let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get()
|
||||
# Restart — both still buffered with intact missingDeps.
|
||||
let rm2 = newV2Manager(store)
|
||||
check (await rm2.ensureChannel(testChannel)).isOk()
|
||||
let inbuf = await rm2.getIncomingBuffer(testChannel)
|
||||
check "c" in inbuf
|
||||
@ -364,40 +355,74 @@ suite "Persistence: write → restart → read-back":
|
||||
check "a" in history
|
||||
check "b" in history
|
||||
check "c" in history
|
||||
# Buffer should be drained.
|
||||
let inbufFinal = await rm2.getIncomingBuffer(testChannel)
|
||||
check inbufFinal.len == 0
|
||||
await rm2.cleanup()
|
||||
|
||||
suite "Persistence: error propagation":
|
||||
asyncTest "loadAllForChannel failure surfaces as rePersistenceError":
|
||||
suite "Persistence (V2): failure policy":
|
||||
asyncTest "loadChannel failure surfaces as rePersistenceError on bootstrap":
|
||||
# Bootstrap durability is the semantic intent of getOrCreateChannel —
|
||||
# the caller asked us to materialise a channel and we can't do that
|
||||
# without knowing prior state. So this op DOES propagate err on load
|
||||
# failure (PLAN §8).
|
||||
let store = newInMemoryStore()
|
||||
store.failingOps.incl("loadAllForChannel")
|
||||
store.failingOps.incl("loadChannel")
|
||||
let rm = newReliabilityManager(
|
||||
participantId = "alice", persistence = newInMemoryPersistence(store)
|
||||
participantId = "alice", persistenceV2 = newInMemoryPersistenceV2(store)
|
||||
)
|
||||
.get()
|
||||
let res = await rm.ensureChannel(testChannel)
|
||||
check res.isErr()
|
||||
check res.error == ReliabilityError.rePersistenceError
|
||||
|
||||
asyncTest "write failure during send surfaces as rePersistenceError":
|
||||
asyncTest "saveChannelMeta failure during send does NOT surface — non-fatal policy":
|
||||
# PLAN §8: persistence failures during foreground ops are logged but
|
||||
# MUST NOT abort the op. The in-memory state is the source of truth;
|
||||
# the next op's snapshot will re-synchronise on-disk state. This test
|
||||
# is the inversion of the legacy "write failure surfaces as err" —
|
||||
# the new policy is deliberate.
|
||||
let store = newInMemoryStore()
|
||||
let rm = newReliabilityManager(
|
||||
participantId = "alice", persistence = newInMemoryPersistence(store)
|
||||
participantId = "alice", persistenceV2 = newInMemoryPersistenceV2(store)
|
||||
)
|
||||
.get()
|
||||
check (await rm.ensureChannel(testChannel)).isOk()
|
||||
# Make the outgoing-buffer write fail; wrapOutgoingMessage must not swallow it.
|
||||
store.failingOps.incl("saveOutgoing")
|
||||
store.failingOps.incl("saveChannelMeta")
|
||||
let res = await rm.wrapOutgoingMessage(@[byte(1)], "m1", testChannel)
|
||||
check res.isErr()
|
||||
check res.error == ReliabilityError.rePersistenceError
|
||||
# Op succeeds: bytes were produced, protocol state is correct in
|
||||
# memory, the FFI caller is unaffected.
|
||||
check res.isOk()
|
||||
# In-memory state is correct even though disk save was rejected.
|
||||
let buf = await rm.getOutgoingBuffer(testChannel)
|
||||
check buf.len == 1
|
||||
check buf[0].message.messageId == "m1"
|
||||
# Recovery: clear the failure, drive another op, disk catches up.
|
||||
store.failingOps.excl("saveChannelMeta")
|
||||
let res2 = await rm.wrapOutgoingMessage(@[byte(2)], "m2", testChannel)
|
||||
check res2.isOk()
|
||||
check "m1" in store.outgoing[testChannel]
|
||||
check "m2" in store.outgoing[testChannel]
|
||||
|
||||
asyncTest "dropChannel failure during removeChannel surfaces as rePersistenceError":
|
||||
asyncTest "updateHistory failure during send does NOT surface — non-fatal policy":
|
||||
# Same policy applied to the history-update path.
|
||||
let store = newInMemoryStore()
|
||||
let rm = newReliabilityManager(
|
||||
participantId = "alice", persistence = newInMemoryPersistence(store)
|
||||
participantId = "alice", persistenceV2 = newInMemoryPersistenceV2(store)
|
||||
)
|
||||
.get()
|
||||
check (await rm.ensureChannel(testChannel)).isOk()
|
||||
store.failingOps.incl("updateHistory")
|
||||
let res = await rm.wrapOutgoingMessage(@[byte(1)], "m1", testChannel)
|
||||
check res.isOk()
|
||||
check rm.channels[testChannel].messageHistory.len == 1
|
||||
|
||||
asyncTest "dropChannel failure during removeChannel surfaces as rePersistenceError":
|
||||
# Durability is the semantic intent of removeChannel — the caller
|
||||
# asked us to confirm a disk wipe. We cannot silently lie. So this op
|
||||
# DOES propagate err on failure (PLAN §8).
|
||||
let store = newInMemoryStore()
|
||||
let rm = newReliabilityManager(
|
||||
participantId = "alice", persistenceV2 = newInMemoryPersistenceV2(store)
|
||||
)
|
||||
.get()
|
||||
check (await rm.ensureChannel(testChannel)).isOk()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user