Apply suggestions from code review

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

tests/persistency/test_blob_codec.nim

@@ -87,3 +87,8 @@ suite "Persistency blob codec":
       discard fromBlob(bytes[0 ..< bytes.len - 5], SdsMessage)
     expect ValueError:
       discard fromBlob(@[0xFF'u8, 0xFF, 0xFF, 0xFF], string) # claims 4 GiB
+
+  test "trailing input raises ValueError":
+    let bytes = toBlob(sampleMessage())
+    expect ValueError:
+      discard fromBlob(bytes & @[0x00'u8], SdsMessage)

waku/persistency/payload_codec.nim

@@ -82,7 +82,11 @@ proc writePart*(buf: var seq[byte], v: int) =
   writePart(buf, int64(v))
 
 proc readPart*(r: var ReadCtx, _: typedesc[int]): int {.raises: [ValueError].} =
-  int(readPart(r, int64))
+  let x = readPart(r, int64)
+  when sizeof(int) < sizeof(int64):
+    if x < int64(low(int)) or x > int64(high(int)):
+      raise newException(ValueError, "int out of range: " & $x)
+  result = int(x)
 
 # ── Small scalars ───────────────────────────────────────────────────────
 
@@ -114,7 +118,12 @@ proc writePart*[E: enum](buf: var seq[byte], v: E) =
   writePart(buf, int64(ord(v)))
 
 proc readPart*[E: enum](r: var ReadCtx, _: typedesc[E]): E {.raises: [ValueError].} =
-  E(readPart(r, int64))
+  let x = readPart(r, int64)
+  let lo = int64(ord(low(E)))
+  let hi = int64(ord(high(E)))
+  if x < lo or x > hi:
+    raise newException(ValueError, "enum value out of range: " & $x)
+  result = E(x)
 
 # ── string / seq[byte] (4-byte length) ──────────────────────────────────
 
@@ -124,7 +133,10 @@ proc writePart*(buf: var seq[byte], s: string) =
     buf.add(byte(c))
 
 proc readPart*(r: var ReadCtx, _: typedesc[string]): string {.raises: [ValueError].} =
-  let n = int(readPart(r, uint32))
+  let nU = readPart(r, uint32)
+  if nU > uint32(high(int)):
+    raise newException(ValueError, "string length out of range: " & $nU)
+  let n = int(nU)
   r.need(n)
   result = newString(n)
   for i in 0 ..< n:
@@ -139,7 +151,10 @@ proc writePart*(buf: var seq[byte], b: seq[byte]) =
 proc readPart*(
     r: var ReadCtx, _: typedesc[seq[byte]]
 ): seq[byte] {.raises: [ValueError].} =
-  let n = int(readPart(r, uint32))
+  let nU = readPart(r, uint32)
+  if nU > uint32(high(int)):
+    raise newException(ValueError, "blob length out of range: " & $nU)
+  let n = int(nU)
   r.need(n)
   result = newSeq[byte](n)
   for i in 0 ..< n:
@@ -183,11 +198,13 @@ proc readPart*[T](
     r: var ReadCtx, _: typedesc[seq[T]]
 ): seq[T] {.raises: [ValueError].} =
   mixin readPart
-  let n = int(readPart(r, uint32))
+  let nU = readPart(r, uint32)
+  if nU > uint32(high(int)):
+    raise newException(ValueError, "sequence length out of range: " & $nU)
+  let n = int(nU)
   result = newSeqOfCap[T](n)
   for _ in 0 ..< n:
     result.add(readPart(r, T))
-
 proc writePart*[T](buf: var seq[byte], s: HashSet[T]) =
   mixin writePart
   writePart(buf, uint32(s.len))
@@ -198,11 +215,13 @@ proc readPart*[T](
     r: var ReadCtx, _: typedesc[HashSet[T]]
 ): HashSet[T] {.raises: [ValueError].} =
   mixin readPart
-  let n = int(readPart(r, uint32))
+  let nU = readPart(r, uint32)
+  if nU > uint32(high(int)):
+    raise newException(ValueError, "set length out of range: " & $nU)
+  let n = int(nU)
   result = initHashSet[T](max(n, 2))
   for _ in 0 ..< n:
     result.incl(readPart(r, T))
-
 proc writePart*[T: tuple](buf: var seq[byte], v: T) =
   mixin writePart
   for f in fields(v):
@@ -305,6 +324,8 @@ proc toBlob*[T](v: T): seq[byte] =
 proc fromBlob*[T](bytes: openArray[byte], _: typedesc[T]): T {.raises: [ValueError].} =
   mixin readPart
   var r = initReadCtx(bytes)
-  readPart(r, T)
+  result = readPart(r, T)
+  if r.pos != r.buf.len:
+    raise newException(ValueError, "trailing payload bytes: " & $(r.buf.len - r.pos))
 
 {.pop.}

waku/persistency/sds_persistency.nim

@@ -10,12 +10,13 @@
 ## {.async: (raises: []), gcsafe.}`) — SDS awaits them on its own chronos
 ## loop. We map each onto the matching async `Job` op:
 ##
-##  * **Writes (save*/remove*)** — `await` the Job op through the `safePut`/
-##    `safeDelete` helpers, which trap any backend error and log it rather
-##    than raising (the contract forbids raising).
+##  * **Writes (save*/remove*)** — call the fire-and-forget `Job.persist*` ops
+##    through the `safePut`/`safeDelete` helpers, which trap any backend error
+##    and log it rather than raising (the contract forbids raising). Note that
+##    Persistency v1 only guarantees the event has been queued when the Future
+##    resolves — reads immediately after an awaited write can still be racy.
 ##  * **`dropChannel`** — awaits `doDropChannel`, which batches every row of
-##    the channel into one transactional `persist` so the wipe is atomic
-##    (called from removeChannel / resetReliabilityManager).
+##    the channel into one transactional `persist` (atomic when applied).
 ##  * **`loadAllForChannel`** — awaits `doLoadAll` and returns the snapshot
 ##    the SDS bootstrap path needs.
 ##