logos-messaging
diff --git a/‎AGENTS.md‎
Lines changed: 2 additions & 2 deletions b/‎AGENTS.md‎
Lines changed: 2 additions & 2 deletions
diff --git a/‎CLAUDE.md‎
Lines changed: 1 addition & 1 deletion b/‎CLAUDE.md‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎doc/PLAN_SNAPSHOT_PERSISTENCE.md‎
Lines changed: 502 additions & 0 deletions b/‎doc/PLAN_SNAPSHOT_PERSISTENCE.md‎
Lines changed: 502 additions & 0 deletions
diff --git a/‎sds.nim‎
Lines changed: 139 additions & 66 deletions b/‎sds.nim‎
Lines changed: 139 additions & 66 deletions
diff --git a/‎sds.nimble‎
Lines changed: 2 additions & 1 deletion b/‎sds.nimble‎
Lines changed: 2 additions & 1 deletion
diff --git a/‎sds/sds_utils.nim‎
Lines changed: 214 additions & 38 deletions b/‎sds/sds_utils.nim‎
Lines changed: 214 additions & 38 deletions
diff --git a/‎sds/snapshot_codec.nim‎
Lines changed: 326 additions & 0 deletions b/‎sds/snapshot_codec.nim‎
Lines changed: 326 additions & 0 deletions
@@ -1,7 +1,7 @@
 <!-- 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.
+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.
 
 > If any GitNexus tool warns the index is stale, run `npx gitnexus analyze` in terminal first.
 
@@ -40,4 +40,4 @@ This project is indexed by GitNexus as **nim-sds** (889 symbols, 1437 relationsh
 | 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 -->
+<!-- gitnexus:end -->
@@ -166,7 +166,7 @@ If using Nix, also recalculate the fixed-output hash in `nix/deps.nix` after upd
 <!-- 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.
+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.
 
 > If any GitNexus tool warns the index is stale, run `npx gitnexus analyze` in terminal first.
 
 
@@ -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"
@@ -68,6 +68,7 @@ task test, "Run the test suite":
   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"
+  exec "nim c -r --outdir:build tests/test_snapshot_codec.nim"
 
 task libsdsDynamicWindows, "Generate bindings":
   let outLibNameAndExt = "libsds.dll"
 
@@ -0,0 +1,326 @@
+## Storage encoding for the snapshot persistence types.
+##
+## This is the codec nim-sds runs on its side of the persistence boundary
+## to turn a `ChannelMeta` (or `ChannelData`, or `HistoryUpdate`) into the
+## opaque `seq[byte]` blob the KV persistence backend stores. The KV
+## backend treats the blob as fully opaque. See PLAN_SNAPSHOT_PERSISTENCE.md
+## §1.5 for why this codec exists at all and §6 for the choice of protobuf.
+##
+## This is NOT the SDS network wire format — that lives in `sds/protobuf.nim`
+## and is unchanged. Encoders for `SdsMessage` and `HistoryEntry` are reused
+## from there to avoid maintaining two codecs for the same shape.
+
+{.push raises: [].}
+
+import std/[sets, times]
+import libp2p/protobuf/minprotobuf
+import ./types/[
+  channel_meta, history_update, sds_message, sds_message_id, history_entry,
+  unacknowledged_message, incoming_message, repair_entry, reliability_error,
+]
+import ./protobufutil
+import ./protobuf as wire
+
+export channel_meta, history_update
+
+# ---------------------------------------------------------------------------
+# Time <-> int64 unix milliseconds
+# ---------------------------------------------------------------------------
+# The protocol uses `getTime()` (wall clock). For wire stability we encode
+# as unix milliseconds in int64 (zigzag not needed — pre-1970 values do not
+# occur in practice). Sub-millisecond precision is intentionally dropped:
+# the protocol's repair backoff windows are seconds-scale.
+
+proc toUnixMs(t: Time): int64 =
+  t.toUnix * 1000'i64 + int64(t.nanosecond div 1_000_000)
+
+proc fromUnixMs(ms: int64): Time =
+  let secs = ms div 1000
+  let nanos = (ms mod 1000).int * 1_000_000
+  initTime(secs, nanos)
+
+# ---------------------------------------------------------------------------
+# UnacknowledgedMessage
+# ---------------------------------------------------------------------------
+
+proc encodeUnacked(u: UnacknowledgedMessage): ProtoBuffer =
+  var pb = initProtoBuffer()
+  let msgPb = wire.encode(u.message)
+  pb.write(1, msgPb.buffer)
+  pb.write(2, uint64(u.sendTime.toUnixMs))
+  pb.write(3, uint32(u.resendAttempts))
+  pb.finish()
+  pb
+
+proc decodeUnacked(buf: seq[byte]): ProtobufResult[UnacknowledgedMessage] =
+  let pb = initProtoBuffer(buf)
+  var msgBytes: seq[byte]
+  if not ?pb.getField(1, msgBytes):
+    return err(ProtobufError.missingRequiredField("UnacknowledgedMessage.message"))
+  let msg = SdsMessage.decode(msgBytes).valueOr:
+    return err(ProtobufError.missingRequiredField("UnacknowledgedMessage.message"))
+  var sendMs: uint64
+  if not ?pb.getField(2, sendMs):
+    return err(ProtobufError.missingRequiredField("UnacknowledgedMessage.sendTime"))
+  var attempts: uint32
+  discard pb.getField(3, attempts)
+  ok(
+    UnacknowledgedMessage.init(
+      message = msg,
+      sendTime = fromUnixMs(int64(sendMs)),
+      resendAttempts = int(attempts),
+    )
+  )
+
+# ---------------------------------------------------------------------------
+# IncomingMessage
+# ---------------------------------------------------------------------------
+
+proc encodeIncoming(m: IncomingMessage): ProtoBuffer =
+  var pb = initProtoBuffer()
+  let msgPb = wire.encode(m.message)
+  pb.write(1, msgPb.buffer)
+  for dep in m.missingDeps:
+    pb.write(2, dep) # SdsMessageID is string
+  pb.finish()
+  pb
+
+proc decodeIncoming(buf: seq[byte]): ProtobufResult[IncomingMessage] =
+  let pb = initProtoBuffer(buf)
+  var msgBytes: seq[byte]
+  if not ?pb.getField(1, msgBytes):
+    return err(ProtobufError.missingRequiredField("IncomingMessage.message"))
+  let msg = SdsMessage.decode(msgBytes).valueOr:
+    return err(ProtobufError.missingRequiredField("IncomingMessage.message"))
+  var deps: seq[SdsMessageID]
+  discard pb.getRepeatedField(2, deps)
+  var depSet = initHashSet[SdsMessageID]()
+  for d in deps:
+    depSet.incl(d)
+  ok(IncomingMessage.init(message = msg, missingDeps = depSet))
+
+# ---------------------------------------------------------------------------
+# OutgoingRepairEntry / OutgoingRepairKV
+# ---------------------------------------------------------------------------
+
+proc encodeOutRepairEntry(e: OutgoingRepairEntry): ProtoBuffer =
+  var pb = initProtoBuffer()
+  let histPb = wire.encodeHistoryEntry(e.outHistEntry)
+  pb.write(1, histPb.buffer)
+  pb.write(2, uint64(e.minTimeRepairReq.toUnixMs))
+  pb.finish()
+  pb
+
+proc decodeOutRepairEntry(buf: seq[byte]): ProtobufResult[OutgoingRepairEntry] =
+  let pb = initProtoBuffer(buf)
+  var histBytes: seq[byte]
+  if not ?pb.getField(1, histBytes):
+    return err(ProtobufError.missingRequiredField("OutgoingRepairEntry.outHistEntry"))
+  let histPb = initProtoBuffer(histBytes)
+  let entry = ?wire.decodeHistoryEntry(histPb)
+  var ms: uint64
+  if not ?pb.getField(2, ms):
+    return err(ProtobufError.missingRequiredField("OutgoingRepairEntry.minTimeRepairReq"))
+  ok(
+    OutgoingRepairEntry.init(
+      outHistEntry = entry, minTimeRepairReq = fromUnixMs(int64(ms))
+    )
+  )
+
+proc encodeOutRepairKV(kv: OutgoingRepairKV): ProtoBuffer =
+  var pb = initProtoBuffer()
+  pb.write(1, kv.messageId)
+  let entryPb = encodeOutRepairEntry(kv.entry)
+  pb.write(2, entryPb.buffer)
+  pb.finish()
+  pb
+
+proc decodeOutRepairKV(buf: seq[byte]): ProtobufResult[OutgoingRepairKV] =
+  let pb = initProtoBuffer(buf)
+  var msgId: SdsMessageID
+  if not ?pb.getField(1, msgId):
+    return err(ProtobufError.missingRequiredField("OutgoingRepairKV.messageId"))
+  var entryBytes: seq[byte]
+  if not ?pb.getField(2, entryBytes):
+    return err(ProtobufError.missingRequiredField("OutgoingRepairKV.entry"))
+  let entry = ?decodeOutRepairEntry(entryBytes)
+  ok(OutgoingRepairKV(messageId: msgId, entry: entry))
+
+# ---------------------------------------------------------------------------
+# IncomingRepairEntry / IncomingRepairKV
+# ---------------------------------------------------------------------------
+
+proc encodeInRepairEntry(e: IncomingRepairEntry): ProtoBuffer =
+  var pb = initProtoBuffer()
+  let histPb = wire.encodeHistoryEntry(e.inHistEntry)
+  pb.write(1, histPb.buffer)
+  pb.write(2, e.cachedMessage)
+  pb.write(3, uint64(e.minTimeRepairResp.toUnixMs))
+  pb.finish()
+  pb
+
+proc decodeInRepairEntry(buf: seq[byte]): ProtobufResult[IncomingRepairEntry] =
+  let pb = initProtoBuffer(buf)
+  var histBytes: seq[byte]
+  if not ?pb.getField(1, histBytes):
+    return err(ProtobufError.missingRequiredField("IncomingRepairEntry.inHistEntry"))
+  let histPb = initProtoBuffer(histBytes)
+  let entry = ?wire.decodeHistoryEntry(histPb)
+  var cached: seq[byte]
+  if not ?pb.getField(2, cached):
+    return err(ProtobufError.missingRequiredField("IncomingRepairEntry.cachedMessage"))
+  var ms: uint64
+  if not ?pb.getField(3, ms):
+    return err(ProtobufError.missingRequiredField("IncomingRepairEntry.minTimeRepairResp"))
+  ok(
+    IncomingRepairEntry.init(
+      inHistEntry = entry,
+      cachedMessage = cached,
+      minTimeRepairResp = fromUnixMs(int64(ms)),
+    )
+  )
+
+proc encodeInRepairKV(kv: IncomingRepairKV): ProtoBuffer =
+  var pb = initProtoBuffer()
+  pb.write(1, kv.messageId)
+  let entryPb = encodeInRepairEntry(kv.entry)
+  pb.write(2, entryPb.buffer)
+  pb.finish()
+  pb
+
+proc decodeInRepairKV(buf: seq[byte]): ProtobufResult[IncomingRepairKV] =
+  let pb = initProtoBuffer(buf)
+  var msgId: SdsMessageID
+  if not ?pb.getField(1, msgId):
+    return err(ProtobufError.missingRequiredField("IncomingRepairKV.messageId"))
+  var entryBytes: seq[byte]
+  if not ?pb.getField(2, entryBytes):
+    return err(ProtobufError.missingRequiredField("IncomingRepairKV.entry"))
+  let entry = ?decodeInRepairEntry(entryBytes)
+  ok(IncomingRepairKV(messageId: msgId, entry: entry))
+
+# ---------------------------------------------------------------------------
+# ChannelMeta (top-level snapshot)
+# ---------------------------------------------------------------------------
+
+proc encode*(meta: ChannelMeta): ProtoBuffer =
+  var pb = initProtoBuffer()
+  pb.write(1, meta.schemaVersion)
+  pb.write(2, uint64(meta.lamportTimestamp))
+  for u in meta.outgoingBuffer:
+    let entryPb = encodeUnacked(u)
+    pb.write(3, entryPb.buffer)
+  for m in meta.incomingBuffer:
+    let entryPb = encodeIncoming(m)
+    pb.write(4, entryPb.buffer)
+  for kv in meta.outgoingRepairBuffer:
+    let entryPb = encodeOutRepairKV(kv)
+    pb.write(5, entryPb.buffer)
+  for kv in meta.incomingRepairBuffer:
+    let entryPb = encodeInRepairKV(kv)
+    pb.write(6, entryPb.buffer)
+  pb.finish()
+  pb
+
+proc decode*(T: type ChannelMeta, buf: seq[byte]): ProtobufResult[T] =
+  let pb = initProtoBuffer(buf)
+  var meta = ChannelMeta.init()
+
+  var ver: uint32
+  if not ?pb.getField(1, ver):
+    return err(ProtobufError.missingRequiredField("ChannelMeta.schemaVersion"))
+  if ver != ChannelMetaSchemaVersion:
+    # Per the contract: refuse loudly rather than silently truncating.
+    return err(ProtobufError.missingRequiredField(
+      "ChannelMeta.schemaVersion(unsupported)"
+    ))
+  meta.schemaVersion = ver
+
+  var lts: uint64
+  if not ?pb.getField(2, lts):
+    return err(ProtobufError.missingRequiredField("ChannelMeta.lamportTimestamp"))
+  meta.lamportTimestamp = int64(lts)
+
+  var outBufs, inBufs, outRepBufs, inRepBufs: seq[seq[byte]]
+  discard pb.getRepeatedField(3, outBufs)
+  for b in outBufs:
+    meta.outgoingBuffer.add(?decodeUnacked(b))
+  discard pb.getRepeatedField(4, inBufs)
+  for b in inBufs:
+    meta.incomingBuffer.add(?decodeIncoming(b))
+  discard pb.getRepeatedField(5, outRepBufs)
+  for b in outRepBufs:
+    meta.outgoingRepairBuffer.add(?decodeOutRepairKV(b))
+  discard pb.getRepeatedField(6, inRepBufs)
+  for b in inRepBufs:
+    meta.incomingRepairBuffer.add(?decodeInRepairKV(b))
+  ok(meta)
+
+proc serialize*(meta: ChannelMeta): Result[seq[byte], ReliabilityError] =
+  ok(encode(meta).buffer)
+
+proc deserializeChannelMeta*(
+    data: seq[byte]
+): Result[ChannelMeta, ReliabilityError] =
+  let m = ChannelMeta.decode(data).valueOr:
+    return err(ReliabilityError.reDeserializationError)
+  ok(m)
+
+# ---------------------------------------------------------------------------
+# ChannelData (bootstrap payload)
+# ---------------------------------------------------------------------------
+
+proc encode*(d: ChannelData): ProtoBuffer =
+  var pb = initProtoBuffer()
+  let metaPb = encode(d.meta)
+  pb.write(1, metaPb.buffer)
+  for m in d.messageHistory:
+    let msgPb = wire.encode(m)
+    pb.write(2, msgPb.buffer)
+  pb.finish()
+  pb
+
+proc decode*(T: type ChannelData, buf: seq[byte]): ProtobufResult[T] =
+  let pb = initProtoBuffer(buf)
+  var d = ChannelData.init()
+  var metaBytes: seq[byte]
+  if not ?pb.getField(1, metaBytes):
+    return err(ProtobufError.missingRequiredField("ChannelData.meta"))
+  d.meta = ?ChannelMeta.decode(metaBytes)
+  var histBufs: seq[seq[byte]]
+  discard pb.getRepeatedField(2, histBufs)
+  for b in histBufs:
+    let m = SdsMessage.decode(b).valueOr:
+      return err(ProtobufError.missingRequiredField("ChannelData.messageHistory[i]"))
+    d.messageHistory.add(m)
+  ok(d)
+
+# ---------------------------------------------------------------------------
+# HistoryUpdate
+# ---------------------------------------------------------------------------
+
+proc encode*(u: HistoryUpdate): ProtoBuffer =
+  var pb = initProtoBuffer()
+  for m in u.append:
+    let msgPb = wire.encode(m)
+    pb.write(1, msgPb.buffer)
+  for id in u.evict:
+    pb.write(2, id)
+  pb.finish()
+  pb
+
+proc decode*(T: type HistoryUpdate, buf: seq[byte]): ProtobufResult[T] =
+  let pb = initProtoBuffer(buf)
+  var u = HistoryUpdate.init()
+  var appBufs: seq[seq[byte]]
+  discard pb.getRepeatedField(1, appBufs)
+  for b in appBufs:
+    let m = SdsMessage.decode(b).valueOr:
+      return err(ProtobufError.missingRequiredField("HistoryUpdate.append[i]"))
+    u.append.add(m)
+  var ev: seq[SdsMessageID]
+  discard pb.getRepeatedField(2, ev)
+  u.evict = ev
+  ok(u)
+
+{.pop.}