Skip to content

Commit abdd40c

Browse files
Refactor persistency - follow up - review fixes (#73)
removal of persistency of retrievalHints because its never read. The PR also covers some leftovers of #72 - small interface and code style changes.
1 parent 4ccdd12 commit abdd40c

5 files changed

Lines changed: 30 additions & 77 deletions

File tree

AGENTS.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
<!-- gitnexus:start -->
22
# GitNexus — Code Intelligence
33

4+
45
This project is indexed by GitNexus as **nim-sds** (1100 symbols, 1820 relationships, 66 execution flows). Use the GitNexus MCP tools to understand code, assess impact, and navigate safely.
56

67
> If any GitNexus tool warns the index is stale, run `npx gitnexus analyze` in terminal first.

sds.nim

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -106,7 +106,7 @@ proc wrapOutgoingMessage*(
106106
return err(error)
107107

108108
let bfResult = serializeBloomFilter(channel.bloomFilter.filter)
109-
if bfResult.isErr:
109+
if bfResult.isErr():
110110
error "Failed to serialize bloom filter", channelId = channelId
111111
return err(ReliabilityError.reSerializationError)
112112

sds/sds_utils.nim

Lines changed: 19 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -15,22 +15,7 @@ export
1515
proc defaultConfig*(): ReliabilityConfig =
1616
return ReliabilityConfig.init()
1717

18-
proc reliabilityErr*(detail: string): ReliabilityError {.gcsafe, raises: [].} =
19-
## Maps a backend-supplied persistence error string onto the
20-
## `rePersistenceError` enum value. The enum carries no payload, so the
21-
## original detail is logged here — this is the single point where a
22-
## persistence failure is recorded, while the enum value travels up the
23-
## `Result` chain to the public API caller, who decides what to do.
24-
##
25-
## With the snapshot-based Persistence interface, most protocol ops no
26-
## longer propagate persistence errors at all — they log and continue
27-
## (see PLAN_SNAPSHOT_PERSISTENCE.md §8). This helper is still used by
28-
## the durability-intent ops (removeChannel, resetReliabilityManager,
29-
## getOrCreateChannel) that retain err-on-failure semantics.
30-
warn "persistence operation failed", detail = detail
31-
ReliabilityError.rePersistenceError
32-
33-
proc snapshotMeta*(channel: ChannelContext): ChannelMeta {.gcsafe, raises: [].} =
18+
proc snapshotMeta(channel: ChannelContext): ChannelMeta {.gcsafe, raises: [].} =
3419
## Captures the current in-memory state of a `ChannelContext` as a
3520
## `ChannelMeta` blob, suitable for `Persistence.saveChannelMeta`.
3621
##
@@ -39,16 +24,17 @@ proc snapshotMeta*(channel: ChannelContext): ChannelMeta {.gcsafe, raises: [].}
3924
## (see PLAN §6). The bloom filter and message history are intentionally
4025
## excluded — the former is rebuilt from the latter on bootstrap, and
4126
## the latter is persisted separately via `updateHistory`.
42-
result = ChannelMeta.init()
43-
result.lamportTimestamp = channel.lamportTimestamp
27+
var meta = ChannelMeta.init()
28+
meta.lamportTimestamp = channel.lamportTimestamp
4429
for u in channel.outgoingBuffer:
45-
result.outgoingBuffer.add(u)
30+
meta.outgoingBuffer.add(u)
4631
for _, m in channel.incomingBuffer.pairs:
47-
result.incomingBuffer.add(m)
32+
meta.incomingBuffer.add(m)
4833
for id, e in channel.outgoingRepairBuffer.pairs:
49-
result.outgoingRepairBuffer.add(OutgoingRepairKV(messageId: id, entry: e))
34+
meta.outgoingRepairBuffer.add(OutgoingRepairKV(messageId: id, entry: e))
5035
for id, e in channel.incomingRepairBuffer.pairs:
51-
result.incomingRepairBuffer.add(IncomingRepairKV(messageId: id, entry: e))
36+
meta.incomingRepairBuffer.add(IncomingRepairKV(messageId: id, entry: e))
37+
return meta
5238

5339
proc trySaveMeta*(
5440
rm: ReliabilityManager, channelId: SdsChannelID, channel: ChannelContext
@@ -59,12 +45,11 @@ proc trySaveMeta*(
5945
##
6046
## This helper is the single point where snapshot-save failures are
6147
## logged; callers do not need to handle the Result.
62-
let res = await rm.persistence.saveChannelMeta(channelId, snapshotMeta(channel))
63-
if res.isErr:
48+
(await rm.persistence.saveChannelMeta(channelId, snapshotMeta(channel))).isOkOr:
6449
warn "snapshot save failed; in-memory state authoritative, next op will retry",
65-
channelId = channelId, detail = res.error
50+
channelId = channelId, detail = error
6651

67-
proc queueHistoryAppend*(channel: ChannelContext, msgId: SdsMessageID) =
52+
proc queueHistoryAppend(channel: ChannelContext, msgId: SdsMessageID) =
6853
## Push an append onto the pending history queue. Only the id is
6954
## stored — the full SdsMessage is looked up from `messageHistory` at
7055
## flush time (invariant: every queued id is present in messageHistory).
@@ -76,7 +61,7 @@ proc queueHistoryAppend*(channel: ChannelContext, msgId: SdsMessageID) =
7661
channel.pendingHistoryEvicts.excl(msgId)
7762
channel.pendingHistoryAppends.incl(msgId)
7863

79-
proc queueHistoryEvict*(channel: ChannelContext, msgId: SdsMessageID) =
64+
proc queueHistoryEvict(channel: ChannelContext, msgId: SdsMessageID) =
8065
## Push an evict onto the pending history queue. Merge rule symmetric
8166
## with `queueHistoryAppend`: cancels any pending append for the same
8267
## id (the just-evicted message no longer needs to be persisted as an
@@ -119,8 +104,7 @@ proc tryUpdateHistory*(
119104
except KeyError:
120105
return # checked `in` above; unreachable, but tables can raise per spec
121106

122-
if channel.pendingHistoryAppends.len == 0 and
123-
channel.pendingHistoryEvicts.len == 0:
107+
if channel.pendingHistoryAppends.len == 0 and channel.pendingHistoryEvicts.len == 0:
124108
return # nothing to flush — no round-trip cost
125109

126110
var batch = HistoryUpdate.init()
@@ -151,8 +135,7 @@ proc tryUpdateHistory*(
151135
detail = res.error
152136
if channel.pendingHistoryAppends.len > rm.config.maxMessageHistory:
153137
warn "pending history queue exceeds maxMessageHistory; backend may be stuck",
154-
channelId = channelId,
155-
pendingAppends = channel.pendingHistoryAppends.len
138+
channelId = channelId, pendingAppends = channel.pendingHistoryAppends.len
156139

157140
proc dropChannelFromPersistence*(
158141
rm: ReliabilityManager, channelId: SdsChannelID
@@ -165,7 +148,8 @@ proc dropChannelFromPersistence*(
165148
## err on failure (durability is the semantic intent — the caller asked
166149
## us to confirm a disk wipe; we cannot silently lie). See PLAN §8.
167150
(await rm.persistence.dropChannel(channelId)).isOkOr:
168-
return err(reliabilityErr(error))
151+
warn "persistence operation failed", cause = error
152+
return err(ReliabilityError.rePersistenceError)
169153
ok()
170154

171155
proc cleanup*(rm: ReliabilityManager) {.async: (raises: []).} =
@@ -344,16 +328,6 @@ proc getRecentHistoryEntries*(
344328
if not rm.onRetrievalHint.isNil():
345329
{.cast(raises: []).}:
346330
entry.retrievalHint = rm.onRetrievalHint(msgId)
347-
if entry.retrievalHint.len > 0:
348-
# Phase 2B: best-effort hint persistence via V2. Non-fatal —
349-
# hints are an optimisation; a missing hint just means the
350-
# peer falls back to slower retrieval.
351-
let hintRes = await rm.persistence.setRetrievalHint(
352-
msgId, entry.retrievalHint
353-
)
354-
if hintRes.isErr:
355-
warn "retrieval hint save failed; continuing",
356-
msgId = msgId, detail = hintRes.error
357331
entry.senderId = channel.messageHistory[msgId].senderId
358332
entries.add(entry)
359333
ok(entries)
@@ -456,7 +430,9 @@ proc getOrCreateChannel*(
456430
)
457431
)
458432
let data = (await rm.persistence.loadChannel(channelId)).valueOr:
459-
return err(reliabilityErr(error))
433+
warn "persistence operation failed", cause = error
434+
return err(ReliabilityError.rePersistenceError)
435+
460436
channel.lamportTimestamp = data.meta.lamportTimestamp
461437
# Backend contract: messageHistory MUST be ordered oldest-first.
462438
# If a backend violates this, FIFO eviction breaks across restarts.

sds/types/persistence.nim

Lines changed: 6 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@ export results, sds_message_id, channel_meta, history_update
2727
type Persistence* = object
2828
## Pluggable durability backend. Supplied at `newReliabilityManager`
2929
## construction time; defaults to `noOpPersistence()` when not given.
30-
3130
saveChannelMeta*: proc(
3231
channelId: SdsChannelID, meta: ChannelMeta
3332
): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
@@ -42,30 +41,21 @@ type Persistence* = object
4241
## maxMessageHistory cap. Callers SHOULD skip this call entirely when
4342
## `update.isEmpty`.
4443

45-
loadChannel*: proc(
46-
channelId: SdsChannelID
47-
): Future[Result[ChannelData, string]] {.async: (raises: []), gcsafe.}
44+
loadChannel*: proc(channelId: SdsChannelID): Future[Result[ChannelData, string]] {.
45+
async: (raises: []), gcsafe
46+
.}
4847
## Bootstrap on `getOrCreateChannel`. Returns the full prior state, or
4948
## an empty `ChannelData` if the channel is new on disk. Failure
5049
## propagates to the caller — bootstrap is a durability-intent op.
5150

52-
dropChannel*: proc(
53-
channelId: SdsChannelID
54-
): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
51+
dropChannel*: proc(channelId: SdsChannelID): Future[Result[void, string]] {.
52+
async: (raises: []), gcsafe
53+
.}
5554
## Wipe all persisted state for a channel. Called by `removeChannel` /
5655
## `resetReliabilityManager`. Backends SHOULD execute atomically.
5756
## Failure propagates to the caller — the caller asked us to confirm a
5857
## disk wipe and we cannot silently lie.
5958

60-
setRetrievalHint*: proc(
61-
msgId: SdsMessageID, hint: seq[byte]
62-
): Future[Result[void, string]] {.async: (raises: []), gcsafe.}
63-
## Record a retrieval hint for a message id. Called from
64-
## `getRecentHistoryEntries` when an application-supplied hint
65-
## provider returns a non-empty hint. Out-of-band from the
66-
## snapshot/history write path because hints are populated lazily
67-
## during read. Non-fatal on failure.
68-
6959
proc noOpPersistence*(): Persistence =
7060
## Default backend: discards all writes, returns an empty snapshot on
7161
## load. Used when no real backend is supplied (existing tests and
@@ -87,8 +77,4 @@ proc noOpPersistence*(): Persistence =
8777
channelId: SdsChannelID
8878
): Future[Result[void, string]] {.async: (raises: []).} =
8979
ok(),
90-
setRetrievalHint: proc(
91-
msgId: SdsMessageID, hint: seq[byte]
92-
): Future[Result[void, string]] {.async: (raises: []).} =
93-
ok(),
9480
)

tests/in_memory_persistence.nim

Lines changed: 3 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
##
77
## `failingOps` injects backend failures. Op names match the `Persistence`
88
## field names: "saveChannelMeta", "updateHistory", "loadChannel",
9-
## "dropChannel", "setRetrievalHint".
9+
## "dropChannel".
1010

1111
import std/[tables, sets]
1212
import chronos
@@ -15,16 +15,14 @@ import sds
1515
type InMemoryStore* = ref object
1616
lamports*: Table[SdsChannelID, int64]
1717
log*: Table[SdsChannelID, OrderedTable[SdsMessageID, SdsMessage]]
18-
hints*: Table[SdsMessageID, seq[byte]]
1918
outgoing*: Table[SdsChannelID, OrderedTable[SdsMessageID, UnacknowledgedMessage]]
2019
incoming*: Table[SdsChannelID, OrderedTable[SdsMessageID, IncomingMessage]]
2120
outgoingRepair*: Table[SdsChannelID, OrderedTable[SdsMessageID, OutgoingRepairEntry]]
2221
incomingRepair*: Table[SdsChannelID, OrderedTable[SdsMessageID, IncomingRepairEntry]]
2322
dropChannelCalls*: Table[SdsChannelID, int]
2423
## Per-channel counter; lets tests assert dropChannel is invoked
2524
## exactly once per logical drop.
26-
failingOps*: HashSet[string]
27-
## Op names that should return an injected backend error.
25+
failingOps*: HashSet[string] ## Op names that should return an injected backend error.
2826

2927
proc newInMemoryStore*(): InMemoryStore =
3028
InMemoryStore(failingOps: initHashSet[string]())
@@ -48,8 +46,7 @@ proc newInMemoryPersistence*(store: InMemoryStore): Persistence =
4846
store.outgoing[channelId][u.message.messageId] = u
4947

5048
# Incoming buffer.
51-
store.incoming[channelId] =
52-
initOrderedTable[SdsMessageID, IncomingMessage]()
49+
store.incoming[channelId] = initOrderedTable[SdsMessageID, IncomingMessage]()
5350
for m in meta.incomingBuffer:
5451
store.incoming[channelId][m.message.messageId] = m
5552

@@ -120,11 +117,4 @@ proc newInMemoryPersistence*(store: InMemoryStore): Persistence =
120117
store.dropChannelCalls[channelId] =
121118
store.dropChannelCalls.getOrDefault(channelId) + 1
122119
ok(),
123-
setRetrievalHint: proc(
124-
msgId: SdsMessageID, hint: seq[byte]
125-
): Future[Result[void, string]] {.async: (raises: []).} =
126-
if "setRetrievalHint" in store.failingOps:
127-
return err("injected backend failure: setRetrievalHint")
128-
store.hints[msgId] = hint
129-
ok(),
130120
)

0 commit comments

Comments
 (0)