@@ -262,8 +262,15 @@ type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}
262262-- Per-peer failure counters; drives `shouldRetryServer`.
263263*forwardFailures : [ForwardFailure]
264264
265−-- Strictly-monotonic sequence used as `Message.receivedAt`; guarantees poll-cursor ordering.
266−*seqCounter : Timestamp
265+-- Strictly-monotonic sequence used as `Message.receivedAt`; guarantees
266+-- poll-cursor ordering. Held as a 0-or-1-row wrapped relation, not a bare
267+-- `Timestamp` scalar source: the runtime now panics when a scalar source's
268+-- backing table is empty (it treats a missing row as corruption rather than
269+-- defaulting to 0), which would 500 the very first /poll or /messages on a
270+-- fresh database. Read through `currentSeq` (0 default). The `{value: ...}`
271+-- record wrapper is required because a relation of bare scalars (`[Timestamp]`)
272+-- also can't be persisted — same reason `Message.deliveredTo` is wrapped.
273+*seqCounter : [{value: Timestamp}]
267274
268275-- ============================================
269276-- Rate limiting
@@ -610,15 +617,22 @@ hasLocalPresence = \pres pk t -> any (\p -> p.pubkey == pk && isLocalServer p.se
610617collectPeers : [a] -> (a -> Text) -> [Text]
611618collectPeers = \items getServer -> filter (\s -> not isLocalServer s) (map getServer items)
612619
620+-- Current sequence value, or 0 before the first message seeds it. Reads the
621+-- 0-or-1-row `*seqCounter` relation and folds out the value with a 0 default,
622+-- rather than unwrapping a (possibly empty) scalar source — which the runtime
623+-- panics on.
624+currentSeq : [{value: Timestamp}] -> Timestamp
625+currentSeq = \rows -> fold (\acc r -> max acc r.value) (0 : Timestamp) rows
626+
613627-- Append one ciphertext to the local mailbox under a fresh monotonic
614628-- sequence. Row-level STM retry on `*messages` wakes any parked /poll
615629-- transaction whose filter matches the new row. Caller wraps in `atomic`.
616630appendMessage : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *seqCounter} {}
617631appendMessage = \toKey encryptedBlob t -> do
618632 msgs <- *messages
619− seq <- *seqCounter
620− let seqTs = max t (seq + 1)
621− *seqCounter = seqTs
633+ seqRows <- *seqCounter
634+ let seqTs = max t (currentSeq seqRows + 1)
635+ *seqCounter = [{value: seqTs}]
622636 *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs, deliveredTo: []}]
623637
624638-- True when `f` is the queued forward of `blob` to `server`. Used to dedupe
@@ -709,7 +723,14 @@ retryPendingForwards = do
709723 then atomic (do
710724 cur <- *forwards
711725 *forwards = filter (\x -> x.toServer != server) cur)
712− else when (shouldRetryServer server t failures) (forEach (take maxBatchSize (filter (\f -> f.toServer == server) fwds)) (\f -> do
726+ -- Order the per-peer backlog deterministically before `take maxBatchSize`:
727+ -- `[a]` is an unordered set, so an unsorted `take` returns an
728+ -- implementation-defined subset that could keep retrying the same entries
729+ -- while starving others past the cap. A stable key drains a fixed prefix as
730+ -- entries succeed (removed) or hit `maxForwardRetries` (dropped). (Same class
731+ -- of fix as `forwardQueuedToServer`'s `sortBy receivedAt`; `ForwardEntry`
732+ -- carries no timestamp, so key on the unique blob.)
733+ else when (shouldRetryServer server t failures) (forEach (take maxBatchSize (sortBy (\f -> f.encryptedBlob) (filter (\f -> f.toServer == server) fwds))) (\f -> do
713734 result <- fetch (federationUrl server) (ForwardMessage {f.toKey, f.encryptedBlob})
714735 success <- case result of
715736 Err _ -> do
@@ -839,7 +860,8 @@ handlePoll = \cursor authorization clientIp -> do
839860 -- in-between messages. `max t curSeq` keeps the old behaviour when the
840861 -- counter is idle/behind, while still refusing to ack beyond what's
841862 -- been issued.
842− curSeq <- *seqCounter
863+ curSeqRows <- *seqCounter
864+ let curSeq = currentSeq curSeqRows
843865 let clampedCursor = max 0 (min (max t curSeq) cursor)
844866 when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)
845867 pres <- *presence
@@ -928,6 +950,13 @@ handleSendMessages = \messages authorization clientIp -> do
928950 Just {value: senderKey} -> do
929951 if count messages > maxBatchSize
930952 then yield (Err {error: {status: 413, message: "batch_too_large"}})
953+ -- Refined-type validation at the route boundary only fires for top-level
954+ -- scalar body fields; `to`/`encryptedBlob` are nested inside the `messages`
955+ -- list, so the runtime does not check them. Enforce the `PubkeyHex` /
956+ -- `BlobHex` contracts here so a malformed key or over-`maxBlobLen` / non-hex
957+ -- blob is rejected instead of stored.
958+ else if any (\m -> not (isValidExactHex 64 m.to && isValidBoundedHex maxBlobLen m.encryptedBlob)) messages
959+ then yield (Err {error: {status: 400, message: "invalid_message"}})
931960 else if any (\m -> m.to == senderKey) messages
932961 then yield (Err {error: {status: 400, message: "self_send"}})
933962 else do
@@ -947,13 +976,20 @@ handleSendMessages = \messages authorization clientIp -> do
947976 -- appear twice within one batch.
948977 existingMsgs <- *messages
949978 when (not any (\m -> m.toKey == c.to && m.encryptedBlob == c.encryptedBlob) existingMsgs) (appendMessage c.to c.encryptedBlob t)
979+ -- Peers that already hold this blob — from an earlier send of the
980+ -- identical ciphertext that was forwarded then pruned from
981+ -- `*forwards`. A brand-new message has empty `deliveredTo`, so this
982+ -- is `[]` on the normal path; on a client resend it carries the real
983+ -- set. Mirrors `forwardQueuedToServer`'s guard so we don't re-stage a
984+ -- forward to a peer that already accepted this blob.
985+ let delivered = fold (\acc m -> if m.toKey == c.to && m.encryptedBlob == c.encryptedBlob then union acc (map (\d -> d.server) m.deliveredTo) else acc) [] existingMsgs
950986 -- Stage outbox rows for peers under their `maxForwardsPerServer`
951987 -- ceiling. Peers at the cap still get the immediate post-commit
952988 -- delivery attempt (`forkForwardsTo`) — only the retry-queue row
953989 -- is skipped.
954990 when federationEnabled (do
955991 fwds <- *forwards
956− let newFwds = map (\s -> {toServer: s, toKey: c.to, c.encryptedBlob, retries: 0}) (filter (\s -> not (any (forwardMatches c.encryptedBlob s) fwds) && countWhere (\f -> f.toServer == s) fwds < maxForwardsPerServer) c.remoteServers)
992+ let newFwds = map (\s -> {toServer: s, toKey: c.to, c.encryptedBlob, retries: 0}) (filter (\s -> not (any (forwardMatches c.encryptedBlob s) fwds) && not elem s delivered && countWhere (\f -> f.toServer == s) fwds < maxForwardsPerServer) c.remoteServers)
957993 *forwards = union fwds newFwds))
958994 yield staged)
959995 when federationEnabled (forEach committed (\c -> fork (forkForwardsTo c.to c.encryptedBlob c.remoteServers)))