@@ -505,6 +505,8 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
505*fwdBucket14 : [ForwardEntry]
505*fwdBucket14 : [ForwardEntry]
506*fwdBucket15 : [ForwardEntry]
506*fwdBucket15 : [ForwardEntry]
508+*totalForwardCount : Int
508-- Messages acked by local recipients, pending federation ack to origin server.
510-- Messages acked by local recipients, pending federation ack to origin server.
509-- Sharded into 16 buckets by pubkeyBucket(toKey) for O(n/16) dedup in sendMessageCore.
511-- Sharded into 16 buckets by pubkeyBucket(toKey) for O(n/16) dedup in sendMessageCore.
510*ackedBucket0 : [AckedEntry]
512*ackedBucket0 : [AckedEntry]
@@ -2339,8 +2341,10 @@ removeForwardsByServer = \server messageIds b acc ->
2339 removed <- atomic do
2341 removed <- atomic do
2340 bucket <- readFwdBucket b
2342 bucket <- readFwdBucket b
2341 let matched = do f <- bucket; where f.toServer == server && any messageIds (\mid -> f.messageId == mid); yield f
2343 let matched = do f <- bucket; where f.toServer == server && any messageIds (\mid -> f.messageId == mid); yield f
2342− when (count matched > 0)
2344+ when (count matched > 0) (do
2343− (writeFwdBucket b (do f <- bucket; where not (f.toServer == server && any messageIds (\mid -> f.messageId == mid)); yield f))
2345+ writeFwdBucket b (do f <- bucket; where not (f.toServer == server && any messageIds (\mid -> f.messageId == mid)); yield f)
2346+ tfc <- *totalForwardCount
2347+ set *totalForwardCount = max 0 (tfc - count matched))
2344 yield matched
2348 yield matched
2345 removeForwardsByServer server messageIds (b + 1) (if count removed > 0 then union acc removed else acc)
2349 removeForwardsByServer server messageIds (b + 1) (if count removed > 0 then union acc removed else acc)
@@ -2622,11 +2626,11 @@ dedupMessageIds = \messageIds ->
2622 else union [mid] acc) [] capped
2626 else union [mid] acc) [] capped
2625−-- Constant-time comparison iteration count. Derived from tokenHexLen (48) + 16
2629+-- Constant-time comparison iteration count. Must be >= the longest secret
2626−-- headroom. Must be >= the longest secret compared (tokens = tokenHexLen hex
2630+-- compared: session tokens are tokenHexLen (48) hex chars, but maintenance
2627−-- chars, maintenance secret = variable). If tokenChunks changes, this adjusts
2631+-- secrets can be up to maxTokenLen (256) chars (validated by extractToken).
2628−-- automatically.
2632+-- Using maxTokenLen + 16 ensures full comparison for all accepted inputs.
2629−safeCompareLen = tokenHexLen + 16
2633+safeCompareLen = maxTokenLen + 16
2631-- Constant-time string comparison to prevent timing side-channels.
2635-- Constant-time string comparison to prevent timing side-channels.
2632-- Always iterates safeCompareLen characters regardless of input length;
2636-- Always iterates safeCompareLen characters regardless of input length;
@@ -2985,8 +2989,10 @@ pruneExpired = \t -> do
2985 atomic do
2989 atomic do
2986 forwards <- readAllForwards
2990 forwards <- readAllForwards
2987 -- Clear all forwards when federation is disabled (they can never be retried)
2991 -- Clear all forwards when federation is disabled (they can never be retried)
2988− writeAllForwards (if not federationEnabled then []
2992+ let prunedForwards = if not federationEnabled then []
2989− else do f <- forwards; where f.createdAt > retainCutoff; yield f)
2993+ else do f <- forwards; where f.createdAt > retainCutoff; yield f
2994+ writeAllForwards prunedForwards
2995+ set *totalForwardCount = count prunedForwards
2990 -- AckedIds: per-bucket pruning (each bucket in its own atomic block to reduce lock
2996 -- AckedIds: per-bucket pruning (each bucket in its own atomic block to reduce lock
2991 -- contention with poll-ack paths that write to ackedIds concurrently).
2997 -- contention with poll-ack paths that write to ackedIds concurrently).
2992 pruneAllAckedBuckets 0 retainCutoff
2998 pruneAllAckedBuckets 0 retainCutoff
@@ -3342,7 +3348,14 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
3342 Authorized {pubkey, ip} ->
3348 Authorized {pubkey, ip} ->
3343 if not (matchIp clientIp ip)
3349 if not (matchIp clientIp ip)
3344 then onUnauth
3350 then onUnauth
3352+ -- Sliding window: refresh session expiry on every authenticated operation.
3353+ -- Without this, users who send/lookup/ack but don't poll would have their
3354+ -- session expire after sessionExpiry even though they're actively using it.
3355+ -- Pre-filter by pubkey before safeCompare to avoid O(bucket_size) constant-time
3356+ -- comparisons (same optimization as reauthenticateSession).
3357+ writeSessionBucket bucket (do s <- sessions; yield (if s.pubkey == pubkey && safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip, createdAt: s.createdAt} else s))
3345− else onAuth pubkey
3358+ onAuth pubkey
3347-- Check if a pubkey has no remaining sessions and, if so, clean up local presence.
3360-- Check if a pubkey has no remaining sessions and, if so, clean up local presence.
3348-- Must be called inside an atomic block after the session has already been removed.
3361-- Must be called inside an atomic block after the session has already been removed.
@@ -3381,7 +3394,12 @@ gossipOfflineIfLastSession = \r label -> do
3381removeForward = \mid srv toKey -> atomic do
3394removeForward = \mid srv toKey -> atomic do
3382 let b = pubkeyBucket toKey
3395 let b = pubkeyBucket toKey
3383 fwdBkt <- readFwdBucket b
3396 fwdBkt <- readFwdBucket b
3384− writeFwdBucket b (forwardsExcluding fwdBkt srv [mid])
3397+ let kept = forwardsExcluding fwdBkt srv [mid]
3398+ writeFwdBucket b kept
3399+ let removed = count fwdBkt - count kept
3400+ when (removed > 0) (do
3401+ tfc <- *totalForwardCount
3402+ set *totalForwardCount = max 0 (tfc - removed))
3386-- Reclaim origin messages after federation ack confirms delivery.
3404-- Reclaim origin messages after federation ack confirms delivery.
3387-- Batched by message bucket: all entries for the same bucket are processed in one
3405-- Batched by message bucket: all entries for the same bucket are processed in one
@@ -3476,6 +3494,8 @@ incrementForwardRetries = \messageId toServer toKey -> atomic do
3476 if shouldDrop
3494 if shouldDrop
3478 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toServer == toServer); yield f)
3496 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toServer == toServer); yield f)
3497+ tfc <- *totalForwardCount
3498+ set *totalForwardCount = max 0 (tfc - 1)
3479 logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
3499 logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
3480 else writeFwdBucket b (do f <- fwdBkt; yield (if f.messageId == messageId && f.toServer == toServer then {messageId: f.messageId, toServer: f.toServer, createdAt: f.createdAt, toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, retries: f.retries + 1} else f))
3500 else writeFwdBucket b (do f <- fwdBkt; yield (if f.messageId == messageId && f.toServer == toServer then {messageId: f.messageId, toServer: f.toServer, createdAt: f.createdAt, toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, retries: f.retries + 1} else f))
@@ -3553,7 +3573,7 @@ forwardQueuedMessages = \targets t -> do
3553 let neededBuckets = uniqueBucketsFor remoteTargets (\target -> target.pubkey)
3573 let neededBuckets = uniqueBucketsFor remoteTargets (\target -> target.pubkey)
3554 relevantMessages <- readSelectedBuckets neededBuckets
3574 relevantMessages <- readSelectedBuckets neededBuckets
3555 let candidates = forwardCandidates remoteTargets relevantMessages forwards t
3575 let candidates = forwardCandidates remoteTargets relevantMessages forwards t
3556− let totalFwdCount <- countAllForwards 0
3576+ totalFwdCount <- *totalForwardCount
3557 -- Per-key cap: filter candidates to respect maxForwardsPerKey per recipient
3577 -- Per-key cap: filter candidates to respect maxForwardsPerKey per recipient
3558 let perKeyCapped = fold (\acc c ->
3578 let perKeyCapped = fold (\acc c ->
3559 let existingForKey = count (do f <- forwards; where f.toKey == c.toKey; yield f)
3579 let existingForKey = count (do f <- forwards; where f.toKey == c.toKey; yield f)
@@ -3563,6 +3583,7 @@ forwardQueuedMessages = \targets t -> do
3563 -- Write back only the affected forward buckets (not all 16)
3583 -- Write back only the affected forward buckets (not all 16)
3564 let combined = union forwards capped
3584 let combined = union forwards capped
3565 writeSelectedFwdBuckets neededFwdBuckets combined
3585 writeSelectedFwdBuckets neededFwdBuckets combined
3586+ set *totalForwardCount = totalFwdCount + count capped
3566 yield capped
3587 yield capped
3567 let servers = collectPeerServers newForwards (\x -> x.toServer)
3588 let servers = collectPeerServers newForwards (\x -> x.toServer)
3568 failures <- *forwardFailures
3589 failures <- *forwardFailures
@@ -3585,6 +3606,41 @@ fetchFederationAck = \origin messageIds -> do
3585 originServer: origin, fromServer: serverHostN,
3606 originServer: origin, fromServer: serverHostN,
3586 requestTimestamp: t, correlationId: corrId})
3607 requestTimestamp: t, correlationId: corrId})
3609+-- Per-bucket ack retry increment: scans only the buckets containing entries from the batch.
3610+-- Each bucket is processed in its own atomic block to reduce lock contention.
3611+-- Drops entries exceeding maxAckRetries to deliveredIds. Returns total dropped count.
3612+incrementAckRetriesPerBucket = \bucketIds server batch droppedAcc ->
3613+ case headOf bucketIds of
3614+ Nothing {} -> yield droppedAcc
3615+ Just {value: b} -> do
3616+ let remainingBuckets = do x <- bucketIds; where x != b; yield x
3617+ bucketDropped <- atomic do
3618+ bucket <- readAckedBucket b
3619+ let isTarget = \a -> a.originServer == server && hasMessage a.messageId a.toKey batch
3620+ let toDrop = do a <- bucket; where isTarget a && a.retries + 1 >= maxAckRetries; yield a
3621+ let dropCount = count toDrop
3622+ writeAckedBucket b (do a <- bucket
3623+ where not (isTarget a && a.retries + 1 >= maxAckRetries)
3624+ yield (if isTarget a then {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1} else a))
3625+ when (dropCount > 0) (do
3626+ delivBkt <- readDelivBucket b
3627+ let newDelivered = do d <- toDrop; yield {messageId: d.messageId, toKey: d.toKey, deliveredAt: d.ackedAt}
3628+ let combined = union delivBkt newDelivered
3629+ writeDelivBucket b (if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined))
3631+ incrementAckRetriesPerBucket remainingBuckets server batch (droppedAcc + bucketDropped)
3633+-- Per-bucket ack removal on success: only rewrites buckets containing matched entries.
3634+removeAckedPerBucket = \bucketIds server batch ->
3635+ case headOf bucketIds of
3636+ Nothing {} -> yield {}
3637+ Just {value: b} -> do
3638+ let remainingBuckets = do x <- bucketIds; where x != b; yield x
3640+ bucket <- readAckedBucket b
3641+ writeAckedBucket b (do a <- bucket; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a)
3642+ removeAckedPerBucket remainingBuckets server batch
3588sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
3644sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
3589sendBatchedAcksToServer = \serverAcks server ->
3645sendBatchedAcksToServer = \serverAcks server ->
3590 forEachBatch serverAcks maxMessageIds (\batch -> do
3646 forEachBatch serverAcks maxMessageIds (\batch -> do
@@ -3593,29 +3649,17 @@ sendBatchedAcksToServer = \serverAcks server ->
3593 Err e -> do
3649 Err e -> do
3594 logWarn ("federation: ack to " ++ server ++ " failed: " ++ show e)
3650 logWarn ("federation: ack to " ++ server ++ " failed: " ++ show e)
3595 recordFailure server
3651 recordFailure server
3596− -- Increment retry counters; drop entries exceeding maxAckRetries to deliveredIds
3652+ -- Increment retry counters per-bucket; drop entries exceeding maxAckRetries to deliveredIds.
3597− dropped <- atomic do
3653+ -- Only touches buckets containing entries from this batch (vs readAllAcked scanning all 16).
3598− ackedIds <- readAllAcked
3654+ let affectedBuckets = uniqueBucketsFor batch (\a -> a.toKey)
3599− let isTarget = \a -> a.originServer == server && hasMessage a.messageId a.toKey batch
3655+ dropped <- incrementAckRetriesPerBucket affectedBuckets server batch 0
3600− let toDrop = do a <- ackedIds; where isTarget a && a.retries + 1 >= maxAckRetries; yield a
3601− let dropCount = count toDrop
3602− writeAllAcked (do a <- ackedIds
3603− where not (isTarget a && a.retries + 1 >= maxAckRetries)
3604− yield (if isTarget a then {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1} else a))
3605− when (dropCount > 0) (do
3606− deliveredIds <- readAllDelivered
3607− let newDelivered = do d <- toDrop; yield {messageId: d.messageId, toKey: d.toKey, deliveredAt: d.ackedAt}
3608− let combined = union deliveredIds newDelivered
3609− writeAllDelivered (if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined))
3611 when (dropped > 0)
3656 when (dropped > 0)
3612 (logWarn ("federation: dropped " ++ show dropped ++ " ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
3657 (logWarn ("federation: dropped " ++ show dropped ++ " ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
3613 Ok _ -> do
3658 Ok _ -> do
3614 clearFailure server
3659 clearFailure server
3615− atomic do
3660+ -- Remove acked entries per-bucket (vs readAllAcked/writeAllAcked repartitioning all 16)
3616− ackedIds <- readAllAcked
3661+ let affectedBuckets = uniqueBucketsFor batch (\a -> a.toKey)
3617− let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch
3662+ removeAckedPerBucket affectedBuckets server batch)
3618− writeAllAcked updatedAckedIds)
3620-- Send delivery acknowledgments back to origin servers and remove acked entries on success
3664-- Send delivery acknowledgments back to origin servers and remove acked entries on success
3621sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
3665sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
@@ -3771,8 +3815,13 @@ pruneAndRetry = \t -> do
3771 stored <- *totalSessionCount
3815 stored <- *totalSessionCount
3772 set *totalSessionCount = actual
3816 set *totalSessionCount = actual
3773 yield (actual - stored)
3817 yield (actual - stored)
3818+ fwdDrift <- atomic do
3819+ actual <- countAllForwards 0
3820+ stored <- *totalForwardCount
3821+ set *totalForwardCount = actual
3822+ yield (actual - stored)
3774− when (presDrift != 0 || sessDrift != 0)
3823+ when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0)
3775− (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))
3824+ (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift))
3776 -- Message reconciliation: cheap check first (sum rcBucket .count fields vs totalMessageCount).
3825 -- Message reconciliation: cheap check first (sum rcBucket .count fields vs totalMessageCount).
3777 -- Full scan only when drift detected or every Nth cycle (reconciliationFullCycleInterval).
3826 -- Full scan only when drift detected or every Nth cycle (reconciliationFullCycleInterval).
3778 pc <- *pruneCount
3827 pc <- *pruneCount
@@ -4489,13 +4538,14 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
4489 let fbi = pubkeyBucket to
4538 let fbi = pubkeyBucket to
4490 fwdBkt <- readFwdBucket fbi
4539 fwdBkt <- readFwdBucket fbi
4491 let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}
4540 let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}
4492− totalFwdCount <- countAllForwards 0
4541+ totalFwdCount <- *totalForwardCount
4493 let globalCap = maxForwardsEntries - totalFwdCount
4542 let globalCap = maxForwardsEntries - totalFwdCount
4494 -- Per-key cap: prevents one key from consuming disproportionate forward slots
4543 -- Per-key cap: prevents one key from consuming disproportionate forward slots
4495 let keyFwdCount = count (do f <- fwdBkt; where f.toKey == to; yield f)
4544 let keyFwdCount = count (do f <- fwdBkt; where f.toKey == to; yield f)
4496 let keyCap = maxForwardsPerKey - keyFwdCount
4545 let keyCap = maxForwardsPerKey - keyFwdCount
4497 let cappedFwds = takeRelation (min globalCap keyCap) newFwds
4546 let cappedFwds = takeRelation (min globalCap keyCap) newFwds
4498 writeFwdBucket fbi (union fwdBkt cappedFwds)
4547 writeFwdBucket fbi (union fwdBkt cappedFwds)
4548+ set *totalForwardCount = totalFwdCount + count cappedFwds
4499 -- Return whether any forwards were actually enqueued (vs dropped by capacity cap)
4549 -- Return whether any forwards were actually enqueued (vs dropped by capacity cap)
4500 yield (count cappedFwds > 0)
4550 yield (count cappedFwds > 0)
4501 else yield (False {})
4551 else yield (False {})
@@ -5300,9 +5350,14 @@ handleReconcile = \authorization respond -> do
5300 stored <- *totalSessionCount
5350 stored <- *totalSessionCount
5301 set *totalSessionCount = actual
5351 set *totalSessionCount = actual
5302 yield (actual - stored)
5352 yield (actual - stored)
5353+ fwdDrift <- atomic do
5354+ actual <- countAllForwards 0
5355+ stored <- *totalForwardCount
5356+ set *totalForwardCount = actual
5357+ yield (actual - stored)
5303 msgDrift <- reconcileMessageCounts 0 0
5358 msgDrift <- reconcileMessageCounts 0 0
5304− when (presDrift != 0 || sessDrift != 0 || msgDrift != 0)
5359+ when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0 || msgDrift != 0)
5305− (logInfo ("reconcile: corrected drift — messages: " ++ show msgDrift ++ ", presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))
5360+ (logInfo ("reconcile: corrected drift — messages: " ++ show msgDrift ++ ", presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift))
5306 yield (respond {ok: True {}, messageDrift: msgDrift, presenceDrift: presDrift, sessionDrift: sessDrift})
5361 yield (respond {ok: True {}, messageDrift: msgDrift, presenceDrift: presDrift, sessionDrift: sessDrift})
5308-- Compute health state and update the cached health values.
5363-- Compute health state and update the cached health values.
@@ -5667,6 +5722,7 @@ handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData de
5667 let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) existingForwards validForwards
5722 let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) existingForwards validForwards
5668 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
5723 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
5669 writeAllForwards cappedForwards
5724 writeAllForwards cappedForwards
5725+ set *totalForwardCount = count cappedForwards
5670 -- Restore forward nonces (dedup by messageId + toKey + originServer, keep fresher on conflict)
5726 -- Restore forward nonces (dedup by messageId + toKey + originServer, keep fresher on conflict)
5671 existingNonces <- readAllNonces
5727 existingNonces <- readAllNonces
5672 let validNonces = do n <- forwardNoncesData; where n.seenAt <= t && t - n.seenAt < forwardNonceRetention; yield n
5728 let validNonces = do n <- forwardNoncesData; where n.seenAt <= t && t - n.seenAt < forwardNonceRetention; yield n