@@ -2967,10 +2967,15 @@ logNonceEviction = \nb lruEvicted oldestRetainedAge ->
2967 logKV level "nonce_lru_eviction" ["bucket=" ++ show nb, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket, "oldestRetainedMs=" ++ show oldestRetainedAge])
2967 logKV level "nonce_lru_eviction" ["bucket=" ++ show nb, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket, "oldestRetainedMs=" ++ show oldestRetainedAge])
2969-- Full record-and-log helper for use outside atomic blocks.
2969-- 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).
2970recordForwardNonce : MessageIdHex -> PubkeyHex -> Timestamp -> IO ()
2975recordForwardNonce : MessageIdHex -> PubkeyHex -> Timestamp -> IO ()
2971recordForwardNonce = \messageId toKey t -> do
2976recordForwardNonce = \messageId toKey t -> do
2972 let nb = pubkeyBucket toKey
2977 let nb = pubkeyBucket toKey
2973− result <- updateForwardNonce nb messageId toKey t
2978+ result <- atomic do updateForwardNonce nb messageId toKey t
2974 logNonceEviction nb result.lruEvicted result.oldestRetainedAge
2979 logNonceEviction nb result.lruEvicted result.oldestRetainedAge
2976-- ============================================
2981-- ============================================
@@ -3972,6 +3977,13 @@ pruneExpired = \t -> do
3972 -- Prune old cancel notifications (same retention as delivery receipts)
3977 -- Prune old cancel notifications (same retention as delivery receipts)
3973 cancelNotifs <- *cancelNotifications
3978 cancelNotifs <- *cancelNotifications
3974 set *cancelNotifications = do c <- cancelNotifs; where t - c.cancelledAt < unackedMessageMaxAge; yield c
3979 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
3975 -- Reconcile global federation connection counter with actual in-flight state
3987 -- Reconcile global federation connection counter with actual in-flight state
3976 -- (self-heals drift from leaked slots where a forked task died without decrementing)
3988 -- (self-heals drift from leaked slots where a forked task died without decrementing)
3977 inFlightRec <- *federationInFlight
3989 inFlightRec <- *federationInFlight
@@ -7155,6 +7167,10 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
7156-- Zip two parallel arrays into a list of {messageId, toKey} pairs.
7168-- Zip two parallel arrays into a list of {messageId, toKey} pairs.
7157-- Consumes both lists head-first in O(n), replacing the O(n^2) fold-to-index pattern.
7169-- 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.
7158zipParallelArrays : [MessageIdHex] -> [PubkeyHex] -> [{messageId: MessageIdHex, toKey: PubkeyHex}]
7174zipParallelArrays : [MessageIdHex] -> [PubkeyHex] -> [{messageId: MessageIdHex, toKey: PubkeyHex}]
7159zipParallelArrays = \mids keys ->
7175zipParallelArrays = \mids keys ->
7160 case headOf mids of
7176 case headOf mids of
@@ -7163,33 +7179,47 @@ zipParallelArrays = \mids keys ->
7163 case headOf keys of
7179 case headOf keys of
7164 Nothing {} -> []
7180 Nothing {} -> []
7165 Just {value: key} ->
7181 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))
7170-- Look up senderKey from messages and record delivery receipts.
7184-- 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).
7172receiptsByBucket : [{messageId: MessageIdHex, toKey: PubkeyHex}] -> Timestamp -> IO ()
7188receiptsByBucket : [{messageId: MessageIdHex, toKey: PubkeyHex}] -> Timestamp -> IO ()
7173receiptsByBucket = \pairs t ->
7189receiptsByBucket = \pairs t ->
7174 case headOf pairs of
7190 case headOf pairs of
7175 Nothing {} -> yield {}
7191 Nothing {} -> yield {}
7176− Just {value: pair} -> do
7192+ Just {value: first} -> do
7177− let rest = do p <- pairs; where p.messageId != pair.messageId || p.toKey != pair.toKey; yield p
7193+ let b = pubkeyBucket first.toKey
7178− let b = pubkeyBucket pair.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
7179 atomic do
7197 atomic do
7180 bucket <- readMessageBucket b
7198 bucket <- readMessageBucket b
7199+ sd <- *senderDeliveries
7200+ let updatedSd = receiptsBucketInner sameBucket bucket sd t
7201+ set *senderDeliveries = updatedSd
7202+ receiptsByBucket otherBuckets t
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
7210+ Just {value: pair} ->
7211+ let rest = do p <- pairs; where p.messageId != pair.messageId || p.toKey != pair.toKey; yield p in
7181− let msg = headOf (do m <- bucket; where m.messageId == pair.messageId && m.toKey == pair.toKey && m.senderKey != "" && m.originServer == ""; yield m)
7212+ let msg = headOf (do m <- bucket; where m.messageId == pair.messageId && m.toKey == pair.toKey && m.senderKey != "" && m.originServer == ""; yield m) in
7182− case msg of
7213+ case msg of
7183− Nothing {} -> yield {}
7214+ Nothing {} -> receiptsBucketInner rest bucket sd t
7184− Just {value: m} -> do
7215+ Just {value: m} ->
7185− sd <- *senderDeliveries
7186− let isDup = any sd (\d -> d.messageId == pair.messageId && d.toKey == pair.toKey)
7216+ let isDup = any sd (\d -> d.messageId == pair.messageId && d.toKey == pair.toKey) in
7187− when (not isDup) (do
7217+ if isDup then receiptsBucketInner rest bucket sd t
7188− let combined = union sd [{senderKey: m.senderKey, messageId: pair.messageId, toKey: pair.toKey, deliveredAt: t}]
7218+ else let combined = union sd [{senderKey: m.senderKey, messageId: pair.messageId, toKey: pair.toKey, deliveredAt: t}] in
7189− set *senderDeliveries = if count combined > maxSenderDeliveries
7219+ let updatedSd = if count combined > maxSenderDeliveries
7190− then evictOneOldest combined (\d -> d.deliveredAt)
7220+ then evictOneOldest combined (\d -> d.deliveredAt)
7191− else combined)
7221+ else combined
7192− receiptsByBucket rest t
7222+ in receiptsBucketInner rest bucket updatedSd t
7194-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
7224-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
7195-- Auth checked once; each cancel processed independently. Partial success possible.
7225-- Auth checked once; each cancel processed independently. Partial success possible.
@@ -8877,7 +8907,7 @@ serveOverloaded = \req -> case req of
8877 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}
8907 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}
8878 Logout {respond} -> respond {ok: False {}, authorized: False {}}
8908 Logout {respond} -> respond {ok: False {}, authorized: False {}}
8879 -- Client routes
8909 -- 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: []}
8881 SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded", loadPct: 0}
8911 SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded", loadPct: 0}
8882 MessageStatus {respond} -> respond {status: "", authorized: False {}}
8912 MessageStatus {respond} -> respond {status: "", authorized: False {}}
8883 AckMessages {respond} -> respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "overloaded"}
8913 AckMessages {respond} -> respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "overloaded"}
@@ -9005,6 +9035,20 @@ serve = \req -> do
9005 set *clientAuthLatencyTotal = total + latency
9035 set *clientAuthLatencyTotal = total + latency
9006 set *clientAuthLatencyCount = cnt + 1
9036 set *clientAuthLatencyCount = cnt + 1
9007 set *clientAuthLatencyMax = if latency > mx then latency else mx
9037 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
9008 _ -> yield {}
9052 _ -> yield {}
9009 let path = requestPath req
9053 let path = requestPath req
9010 let ip = requestClientIp req
9054 let ip = requestClientIp req