@@ -2967,10 +2967,15 @@ logNonceEviction = \nb lruEvicted oldestRetainedAge ->
29672967 logKV level "nonce_lru_eviction" ["bucket=" ++ show nb, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket, "oldestRetainedMs=" ++ show oldestRetainedAge])
29682968
29692969-- Full record-and-log helper for use outside atomic blocks.
2970+-- Wraps updateForwardNonce in atomic to prevent lost nonces from concurrent calls.
2971+-- Without this, two concurrent sendMessageCore calls (via handleSendMessage) can both
2972+-- read the same nonce bucket, each append their nonce, and the second write silently
2973+-- drops the first nonce — reopening a replay window for that message.
2974+-- Safe when called from inside an existing atomic block (Knot atomic is reentrant).
29702975recordForwardNonce : MessageIdHex -> PubkeyHex -> Timestamp -> IO ()
29712976recordForwardNonce = \messageId toKey t -> do
29722977 let nb = pubkeyBucket toKey
2973− result <- updateForwardNonce nb messageId toKey t
2978+ result <- atomic do updateForwardNonce nb messageId toKey t
29742979 logNonceEviction nb result.lruEvicted result.oldestRetainedAge
29752980
29762981-- ============================================
@@ -3972,6 +3977,13 @@ pruneExpired = \t -> do
39723977 -- Prune old cancel notifications (same retention as delivery receipts)
39733978 cancelNotifs <- *cancelNotifications
39743979 set *cancelNotifications = do c <- cancelNotifs; where t - c.cancelledAt < unackedMessageMaxAge; yield c
3980+ -- Prune stale per-sender drop counters. Without this, drop counters for senders who
3981+ -- disconnect permanently persist until capacity eviction pushes them out. Same retention
3982+ -- as delivery receipts since drop counters track lost receipts/expirations.
3983+ ddr <- *droppedDeliveryReceipts
3984+ set *droppedDeliveryReceipts = do d <- ddr; where t - d.lastUpdated < unackedMessageMaxAge; yield d
3985+ den <- *droppedExpirationNotifs
3986+ set *droppedExpirationNotifs = do d <- den; where t - d.lastUpdated < unackedMessageMaxAge; yield d
39753987 -- Reconcile global federation connection counter with actual in-flight state
39763988 -- (self-heals drift from leaked slots where a forked task died without decrementing)
39773989 inFlightRec <- *federationInFlight
@@ -7155,6 +7167,10 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
71557167
71567168-- Zip two parallel arrays into a list of {messageId, toKey} pairs.
71577169-- Consumes both lists head-first in O(n), replacing the O(n^2) fold-to-index pattern.
7170+-- Uses skipRelation 1 instead of filter-not-equal to correctly handle duplicate values:
7171+-- filter (where m != mid) removes ALL occurrences of mid, corrupting the pairing when
7172+-- the same messageId appears multiple times (e.g., same message sent to different recipients).
7173+-- skipRelation 1 removes only the first element regardless of duplicates.
71587174zipParallelArrays : [MessageIdHex] -> [PubkeyHex] -> [{messageId: MessageIdHex, toKey: PubkeyHex}]
71597175zipParallelArrays = \mids keys ->
71607176 case headOf mids of
@@ -7163,33 +7179,47 @@ zipParallelArrays = \mids keys ->
71637179 case headOf keys of
71647180 Nothing {} -> []
71657181 Just {value: key} ->
7166− let restMids = do m <- mids; where m != mid; yield m in
7167− let restKeys = do k <- keys; where k != key; yield k in
7168− union [{messageId: mid, toKey: key}] (zipParallelArrays restMids restKeys)
7182+ union [{messageId: mid, toKey: key}] (zipParallelArrays (skipRelation 1 mids) (skipRelation 1 keys))
71697183
71707184-- Look up senderKey from messages and record delivery receipts.
7171−-- Consumes pre-zipped pairs head-first (O(n) total, was O(n^2) with index-based access).
7185+-- Groups pairs by pubkeyBucket so each message bucket is read once per group instead
7186+-- of once per receipt. With 10 receipts targeting the same bucket, this reduces bucket
7187+-- reads from 10 to 1 (and senderDeliveries reads/writes from 10 to 1).
71727188receiptsByBucket : [{messageId: MessageIdHex, toKey: PubkeyHex}] -> Timestamp -> IO ()
71737189receiptsByBucket = \pairs t ->
71747190 case headOf pairs of
71757191 Nothing {} -> yield {}
7176− Just {value: pair} -> do
7177− let rest = do p <- pairs; where p.messageId != pair.messageId || p.toKey != pair.toKey; yield p
7178− let b = pubkeyBucket pair.toKey
7192+ Just {value: first} -> do
7193+ let b = pubkeyBucket first.toKey
7194+ let sameBucket = do p <- pairs; where pubkeyBucket p.toKey == b; yield p
7195+ let otherBuckets = do p <- pairs; where pubkeyBucket p.toKey != b; yield p
7196+ -- Process all pairs for this bucket in a single atomic block
71797197 atomic do
71807198 bucket <- readMessageBucket b
7181− let msg = headOf (do m <- bucket; where m.messageId == pair.messageId && m.toKey == pair.toKey && m.senderKey != "" && m.originServer == ""; yield m)
7182− case msg of
7183− Nothing {} -> yield {}
7184− Just {value: m} -> do
7185− sd <- *senderDeliveries
7186− let isDup = any sd (\d -> d.messageId == pair.messageId && d.toKey == pair.toKey)
7187− when (not isDup) (do
7188− let combined = union sd [{senderKey: m.senderKey, messageId: pair.messageId, toKey: pair.toKey, deliveredAt: t}]
7189− set *senderDeliveries = if count combined > maxSenderDeliveries
7190− then evictOneOldest combined (\d -> d.deliveredAt)
7191− else combined)
7192− receiptsByBucket rest t
7199+ sd <- *senderDeliveries
7200+ let updatedSd = receiptsBucketInner sameBucket bucket sd t
7201+ set *senderDeliveries = updatedSd
7202+ receiptsByBucket otherBuckets t
7203+
7204+-- Process all receipt pairs for one bucket within a single atomic read.
7205+-- Threads senderDeliveries through so each new receipt sees previous ones (dedup).
7206+receiptsBucketInner : [{messageId: MessageIdHex, toKey: PubkeyHex}] -> [Message] -> [SenderDelivery] -> Timestamp -> [SenderDelivery]
7207+receiptsBucketInner = \pairs bucket sd t ->
7208+ case headOf pairs of
7209+ Nothing {} -> sd
7210+ Just {value: pair} ->
7211+ let rest = do p <- pairs; where p.messageId != pair.messageId || p.toKey != pair.toKey; yield p in
7212+ let msg = headOf (do m <- bucket; where m.messageId == pair.messageId && m.toKey == pair.toKey && m.senderKey != "" && m.originServer == ""; yield m) in
7213+ case msg of
7214+ Nothing {} -> receiptsBucketInner rest bucket sd t
7215+ Just {value: m} ->
7216+ let isDup = any sd (\d -> d.messageId == pair.messageId && d.toKey == pair.toKey) in
7217+ if isDup then receiptsBucketInner rest bucket sd t
7218+ else let combined = union sd [{senderKey: m.senderKey, messageId: pair.messageId, toKey: pair.toKey, deliveredAt: t}] in
7219+ let updatedSd = if count combined > maxSenderDeliveries
7220+ then evictOneOldest combined (\d -> d.deliveredAt)
7221+ else combined
7222+ in receiptsBucketInner rest bucket updatedSd t
71937223
71947224-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
71957225-- Auth checked once; each cancel processed independently. Partial success possible.
@@ -8877,7 +8907,7 @@ serveOverloaded = \req -> case req of
88778907 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}
88788908 Logout {respond} -> respond {ok: False {}, authorized: False {}}
88798909 -- Client routes
8880− Poll {respond} -> respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: 0, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 1000, cursorReset: False {}, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0}
8910+ Poll {respond} -> respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: 0, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 1000, cursorReset: False {}, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: []}
88818911 SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded", loadPct: 0}
88828912 MessageStatus {respond} -> respond {status: "", authorized: False {}}
88838913 AckMessages {respond} -> respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "overloaded"}
@@ -9005,6 +9035,20 @@ serve = \req -> do
90059035 set *clientAuthLatencyTotal = total + latency
90069036 set *clientAuthLatencyCount = cnt + 1
90079037 set *clientAuthLatencyMax = if latency > mx then latency else mx
9038+ RevokeSession {} -> atomic do
9039+ total <- *clientAuthLatencyTotal
9040+ cnt <- *clientAuthLatencyCount
9041+ mx <- *clientAuthLatencyMax
9042+ set *clientAuthLatencyTotal = total + latency
9043+ set *clientAuthLatencyCount = cnt + 1
9044+ set *clientAuthLatencyMax = if latency > mx then latency else mx
9045+ Logout {} -> atomic do
9046+ total <- *clientAuthLatencyTotal
9047+ cnt <- *clientAuthLatencyCount
9048+ mx <- *clientAuthLatencyMax
9049+ set *clientAuthLatencyTotal = total + latency
9050+ set *clientAuthLatencyCount = cnt + 1
9051+ set *clientAuthLatencyMax = if latency > mx then latency else mx
90089052 _ -> yield {}
90099053 let path = requestPath req
90109054 let ip = requestClientIp req