@@ -183,7 +183,7 @@ federationLatencyWindowMs : Int<Ms>
183federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
183federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
185-- Snapshot format version (increment on schema changes to prevent corrupt restores)
185-- Snapshot format version (increment on schema changes to prevent corrupt restores)
186−snapshotVersion = 13 -- bumped for double-hash snapshot integrity tag
186+snapshotVersion = 14 -- bumped for rate limit override persistence in snapshots
188-- Server build version: included in /health and /stats responses for deployment verification.
188-- Server build version: included in /health and /stats responses for deployment verification.
189-- Update this string when deploying a new build so operators can confirm which version is running.
189-- Update this string when deploying a new build so operators can confirm which version is running.
@@ -209,6 +209,11 @@ snapshotChecksum = \presenceCount ackedCount deliveredCount forwardCount nonceCo
209-- captures one valid (snapshot, tag) pair cannot derive the secretHash components
209-- captures one valid (snapshot, tag) pair cannot derive the secretHash components
210-- independently because the final mixing step entangles them non-linearly.
210-- independently because the final mixing step entangles them non-linearly.
211-- Returns 0 when no maintenance secret is configured (no integrity protection).
211-- Returns 0 when no maintenance secret is configured (no integrity protection).
212+-- SECURITY NOTE: This is NOT a substitute for cryptographic signing. The hash constants
213+-- are public (in this source), so an attacker with access to the maintenance secret can
214+-- forge valid tags. For high-security deployments, GPG-sign or HMAC snapshot lines
215+-- externally (e.g., via a sidecar that signs [SNAPSHOT_DATA] lines before archiving).
216+-- TODO: Replace with HMAC when Knot adds a cryptographic hash primitive.
212snapshotIntegrityTag : Int -> Text -> Int
217snapshotIntegrityTag : Int -> Text -> Int
213snapshotIntegrityTag = \checksum secret ->
218snapshotIntegrityTag = \checksum secret ->
214 if secret == "" then 0
219 if secret == "" then 0
@@ -273,6 +278,14 @@ federationEnabled = True {}
273-- TODO: If Knot adds a resolved-IP introspection API (e.g., fetchWithResolvedIp),
278-- TODO: If Knot adds a resolved-IP introspection API (e.g., fetchWithResolvedIp),
274-- validate the resolved IP against isBadServerName before completing the request.
279-- validate the resolved IP against isBadServerName before completing the request.
275federationUrl = \server -> federationScheme ++ server
280federationUrl = \server -> federationScheme ++ server
281+-- Bootstrap peer: on startup, fetch a snapshot from this server and auto-restore.
282+-- Set to a trusted peer's hostname (e.g., "peer1.example.com:8080") to bootstrap
283+-- federation state from an existing server. Empty = no bootstrap (default).
284+-- Requires the bootstrap peer to accept unauthenticated /maintenance/snapshot (or
285+-- set bootstrapPeerSecret to the peer's maintenance secret).
286+-- TODO: If Knot adds file I/O, add local file-based auto-load as an alternative.
287+bootstrapPeerUrl = "" -- set to a peer hostname to bootstrap state on startup
288+bootstrapPeerSecret = "" -- maintenance secret of the bootstrap peer
276-- Callback verification: outbound federation requests include a random single-use token;
289-- Callback verification: outbound federation requests include a random single-use token;
277-- the receiver calls back to /federation/verify to confirm the sender actually issued it.
290-- the receiver calls back to /federation/verify to confirm the sender actually issued it.
278maxPendingVerifications = 10000
291maxPendingVerifications = 10000
@@ -329,6 +342,17 @@ maxSenderDeliveries = 10000
329maxSenderDeliveriesPerKey = 100
342maxSenderDeliveriesPerKey = 100
330-- Max pubkeys in a batch lookup request
343-- Max pubkeys in a batch lookup request
331maxBatchLookupKeys = 50
344maxBatchLookupKeys = 50
345+-- Max per-sender drop counter entries (bounds memory for per-sender receipt/expiration drop tracking)
346+maxDroppedCountEntries = 1000
347+-- Max pending federation cancel entries (bounds memory for cancel retry queue)
348+maxPendingCancels = 10000
349+-- Max retries per cancel entry before dropping (uses same backoff as forward retries)
350+maxCancelRetries = 10
352+-- Forward storm prevention: after snapshot restore, ramp up forward retries over this many
353+-- prune cycles (each cycle retries 1/N of maxForwardRetriesPerServer, increasing to full rate).
354+-- Prevents overwhelming peers when thousands of queued forwards are restored at once.
355+restoreForwardRampCycles = 5
332-- Verify callback caching: cache successful verifications per-server to reduce
356-- Verify callback caching: cache successful verifications per-server to reduce
333-- outbound HTTP calls. A peer sending N requests/second forces N verify callbacks;
357-- outbound HTTP calls. A peer sending N requests/second forces N verify callbacks;
334-- caching the success for a short window amortizes the cost.
358-- caching the success for a short window amortizes the cost.
@@ -352,7 +376,7 @@ healthWatchdogMaxAge = backgroundPruneInterval * 15 -- watchdog runs every ~5 pr
352-- ============================================
376-- ============================================
354-- Event delivered to a polling client: an encrypted message blob
378-- Event delivered to a polling client: an encrypted message blob
355−type PollEvent = {id: MessageIdHex, eventType: Text, payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex}}
379+type PollEvent = {id: MessageIdHex, eventType: Text, payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex, expiresAt: Timestamp}}
357-- Gossip events propagate presence changes between federated servers
381-- Gossip events propagate presence changes between federated servers
358data GossipEventType = GossipOnline {} | GossipOffline {} | GossipUnknown {}
382data GossipEventType = GossipOnline {} | GossipOffline {} | GossipUnknown {}
@@ -479,6 +503,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
479type BatchCancelEntry = {messageId: MessageIdHex, to: PubkeyHex}
503type BatchCancelEntry = {messageId: MessageIdHex, to: PubkeyHex}
480type BatchCancelResult = {messageId: MessageIdHex, cancelled: Bool}
504type BatchCancelResult = {messageId: MessageIdHex, cancelled: Bool}
506+-- Pending federation cancel: queued when /messages/cancel or /messages/cancel-batch
507+-- fails to deliver a cancel to a remote server. Retried during the prune cycle.
508+type PendingFederationCancel = {messageId: MessageIdHex, toKey: PubkeyHex, toServer: ServerName, createdAt: Timestamp, retries: Int}
482-- Sender expiration notification: tracks when a locally-originated message expired unread.
510-- Sender expiration notification: tracks when a locally-originated message expired unread.
483-- Pruned after messageMaxAge. Included in the sender's next poll response.
511-- Pruned after messageMaxAge. Included in the sender's next poll response.
484type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}
512type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}
@@ -857,11 +885,15 @@ maxPeerLatencyEntries = 100
857-- Populated during ackPolledMessages and handleAckMessages for locally-originated messages.
885-- Populated during ackPolledMessages and handleAckMessages for locally-originated messages.
858*senderDeliveries : [SenderDelivery]
886*senderDeliveries : [SenderDelivery]
888+-- Pending federation cancels: queued when cancel delivery to a remote server fails.
889+-- Retried during the prune cycle alongside forward retries and ack retries.
890+*pendingFederationCancels : [PendingFederationCancel]
860−-- Tracks how many sender delivery receipts / expiration notifications were dropped
892+-- Per-sender drop counters: tracks how many delivery receipts / expiration notifications
861−-- due to capacity eviction before the sender could poll them. Reset per-sender on poll
893+-- were evicted before the sender could poll them. Keyed by senderKey so each sender sees
862−-- (drained with the notifications). Lets clients know they may have missed receipts.
894+-- only their own drops (not a global counter that conflates all senders). Drained on poll.
863−*droppedDeliveryReceipts : Int
895+*droppedDeliveryReceipts : [{senderKey: Text, count: Int}]
864−*droppedExpirationNotifs : Int
896+*droppedExpirationNotifs : [{senderKey: Text, count: Int}]
866-- Runtime maintenance secret: when non-empty, used instead of compile-time maintenanceSecret.
898-- Runtime maintenance secret: when non-empty, used instead of compile-time maintenanceSecret.
867-- Set via /maintenance/rotate-secret to enable zero-downtime secret rotation.
899-- Set via /maintenance/rotate-secret to enable zero-downtime secret rotation.
@@ -899,6 +931,14 @@ maxPeerLatencyEntries = 100
899*activeRequests : Int
931*activeRequests : Int
900*peakActiveRequests : Int
932*peakActiveRequests : Int
934+-- Client operation latency tracking (for /stats, reset with federation latency window)
935+*clientPollLatencyTotal : Int<Ms>
936+*clientPollLatencyCount : Int
937+*clientPollLatencyMax : Int<Ms>
938+*clientSendLatencyTotal : Int<Ms>
939+*clientSendLatencyCount : Int
940+*clientSendLatencyMax : Int<Ms>
902-- Last time the background watchdog loop ran (for health check liveness detection)
942-- Last time the background watchdog loop ran (for health check liveness detection)
903*lastWatchdogAt : Timestamp
943*lastWatchdogAt : Timestamp
@@ -911,6 +951,12 @@ maxPeerLatencyEntries = 100
911-- More robust than wall-clock comparison: survives clock jumps without false alarms.
951-- More robust than wall-clock comparison: survives clock jumps without false alarms.
912*prunesSinceSnapshot : Int
952*prunesSinceSnapshot : Int
954+-- Prune cycles since last snapshot restore (forward storm prevention).
955+-- After restore, forward retries are throttled for restoreForwardRampCycles prune cycles
956+-- to avoid overwhelming peers with thousands of queued forwards simultaneously.
957+-- 0 = not in ramp-up phase. Decremented each prune cycle; forwards are throttled while > 0.
958+*restoreForwardRampRemaining : Int
914-- Cached health result: returned when /health is rate-limited instead of false all-healthy.
960-- Cached health result: returned when /health is rate-limited instead of false all-healthy.
915-- Updated on every successful (non-rate-limited) health check.
961-- Updated on every successful (non-rate-limited) health check.
916*cachedHealthOk : Bool
962*cachedHealthOk : Bool
@@ -1080,12 +1126,12 @@ route MaintenanceApi where
1080 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
1126 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
1081 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
1127 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
1082 GET /maintenance/snapshot headers {authorization: Text}
1128 GET /maintenance/snapshot headers {authorization: Text}
1083− -> {ok: Bool, version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}], senderExpirations: [SenderExpiration], senderDeliveries: [SenderDelivery]}
1129+ -> {ok: Bool, version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}], senderExpirations: [SenderExpiration], senderDeliveries: [SenderDelivery], rateLimitOverrides: {globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}}
1084 = Snapshot
1130 = Snapshot
1085 -- Restore: import a previously-exported snapshot. Server must be in drain mode
1131 -- Restore: import a previously-exported snapshot. Server must be in drain mode
1086 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
1132 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
1087 -- checksum: integrity check (0 = skip verification, for pre-v6 snapshots)
1133 -- checksum: integrity check (0 = skip verification, for pre-v6 snapshots)
1088− POST {version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}], senderExpirations: [SenderExpiration], senderDeliveries: [SenderDelivery]} /maintenance/restore headers {authorization: Text}
1134+ POST {version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}], senderExpirations: [SenderExpiration], senderDeliveries: [SenderDelivery], rateLimitOverrides: {globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}} /maintenance/restore headers {authorization: Text}
1089 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
1135 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
1090 = Restore
1136 = Restore
1091 -- Block a federation peer (rejects all inbound/outbound federation with this server)
1137 -- Block a federation peer (rejects all inbound/outbound federation with this server)
@@ -1173,7 +1219,7 @@ route HealthApi where
1173-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1219-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1174route StatsApi where
1220route StatsApi where
1175 GET /stats headers {authorization: Text}
1221 GET /stats headers {authorization: Text}
1176− -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, activePolls: Int, activePollsMax: Int, recipientCounts: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}]}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, senderExpirations: Int, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, prunesSinceSnapshot: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int}}
1222+ -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, activePolls: Int, activePollsMax: Int, recipientCounts: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}]}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, senderExpirations: Int, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, prunesSinceSnapshot: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int}}
1178 GET /stats/peers headers {authorization: Text}
1224 GET /stats/peers headers {authorization: Text}
1179 -> {peers: [PeerStatus], ok: Bool}
1225 -> {peers: [PeerStatus], ok: Bool}
@@ -1275,6 +1321,20 @@ clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x
1275unionNE = \a b -> if count a == 0 then b else if count b == 0 then a else union a b
1321unionNE = \a b -> if count a == 0 then b else if count b == 0 then a else union a b
1276min = \a b -> if a < b then a else b
1322min = \a b -> if a < b then a else b
1277max = \a b -> if a > b then a else b
1323max = \a b -> if a > b then a else b
1325+-- Per-sender drop counter helpers: increment a per-key counter and drain it for a specific key.
1326+-- Used to track per-sender delivery receipt / expiration notification drops.
1327+incrementDropCounter = \entries key n ->
1328+ if any entries (\e -> e.senderKey == key)
1329+ then do e <- entries; yield (if e.senderKey == key then {senderKey: e.senderKey, count: e.count + n} else e)
1330+ else if count entries >= maxDroppedCountEntries
1331+ then union (evictOneOldest entries (\e -> e.count)) [{senderKey: key, count: n}]
1332+ else union entries [{senderKey: key, count: n}]
1334+drainDropCounter = \entries key ->
1335+ case headOf (do e <- entries; where e.senderKey == key; yield e) of
1336+ Nothing {} -> {count: 0, remaining: entries}
1337+ Just {value: e} -> {count: e.count, remaining: do x <- entries; where x.senderKey != key; yield x}
1278headOf = \items -> fold (\acc x -> case acc of
1338headOf = \items -> fold (\acc x -> case acc of
1279 Nothing {} -> Just {value: x}
1339 Nothing {} -> Just {value: x}
1280 _ -> acc) Nothing {} items
1340 _ -> acc) Nothing {} items
@@ -2235,14 +2295,30 @@ pruneMessageBucket = \b t -> atomic do
2235 let localExpired = do m <- removed; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, expiredAt: t}
2295 let localExpired = do m <- removed; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, expiredAt: t}
2236 when (count localExpired > 0) (do
2296 when (count localExpired > 0) (do
2237 se <- *senderExpirations
2297 se <- *senderExpirations
2298+ -- Per-key cap: limit expirations per sender to prevent one noisy key from dominating the queue
2299+ let perKeyCapped = fold (\acc e ->
2300+ let existingForKey = count (do x <- se; where x.senderKey == e.senderKey; yield x)
2301+ + count (do x <- acc; where x.senderKey == e.senderKey; yield x)
2302+ in if existingForKey >= maxSenderExpirationsPerKey then acc else union acc [e]) [] localExpired
2303+ let perKeyDropped = count localExpired - count perKeyCapped
2238− let combined = union se localExpired
2304+ let combined = union se perKeyCapped
2239 let overCap = count combined > maxSenderExpirations
2305 let overCap = count combined > maxSenderExpirations
2240− set *senderExpirations = if overCap
2306+ if overCap then do
2241− then evictOneOldest combined (\e -> e.expiredAt)
2307+ let evicted = extractOldest combined (\e -> e.expiredAt)
2242− else combined
2308+ set *senderExpirations = evicted.items
2244 den <- *droppedExpirationNotifs
2309 den <- *droppedExpirationNotifs
2310+ let denAfterCap = case evicted.candidate of
2312+ Just {value: ev} -> incrementDropCounter den ev.senderKey 1
2313+ -- Also record per-key drops for senders that hit maxSenderExpirationsPerKey
2314+ let perKeyDroppedEntries = do e <- localExpired; where none perKeyCapped (\c -> c.messageId == e.messageId && c.toKey == e.toKey); yield e
2245− set *droppedExpirationNotifs = den + 1))
2315+ set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1) denAfterCap perKeyDroppedEntries
2317+ set *senderExpirations = combined
2318+ when (perKeyDropped > 0) (do
2319+ den <- *droppedExpirationNotifs
2320+ let perKeyDroppedEntries = do e <- localExpired; where none perKeyCapped (\c -> c.messageId == e.messageId && c.toKey == e.toKey); yield e
2321+ set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1) den perKeyDroppedEntries))
2246 else yield {}
2322 else yield {}
2248-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
2324-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
@@ -3448,6 +3524,9 @@ pruneExpired = \t -> do
3448 -- Prune stale deferred ack cleanups (older than messageMaxAge are never useful)
3524 -- Prune stale deferred ack cleanups (older than messageMaxAge are never useful)
3449 pendingCleanups <- *pendingAckCleanups
3525 pendingCleanups <- *pendingAckCleanups
3450 set *pendingAckCleanups = do c <- pendingCleanups; where t - c.cleanupAfter < messageMaxAge; yield c
3526 set *pendingAckCleanups = do c <- pendingCleanups; where t - c.cleanupAfter < messageMaxAge; yield c
3527+ -- Prune old pending federation cancel entries (same retention as other dedup state)
3528+ pendingCancels <- *pendingFederationCancels
3529+ set *pendingFederationCancels = do c <- pendingCancels; where t - c.createdAt < messageMaxAge; yield c
3451 -- Prune old sender expiration notifications. Uses unackedMessageMaxAge (7 days)
3530 -- Prune old sender expiration notifications. Uses unackedMessageMaxAge (7 days)
3452 -- instead of sessionExpiry (1 hour) so mobile clients that don't poll for days
3531 -- instead of sessionExpiry (1 hour) so mobile clients that don't poll for days
3453 -- still receive expiration notifications on their next poll.
3532 -- still receive expiration notifications on their next poll.
@@ -3969,9 +4048,11 @@ retryForwardsForServer = \entries remaining ->
3969 -- Process per-message results: remove succeeded, increment retries for failed
4048 -- Process per-message results: remove succeeded, increment retries for failed
3970 retryForwardsBatchResults batch response.results server
4049 retryForwardsBatchResults batch response.results server
3972− logWarn ("federation: batch forward to " ++ server ++ " rejected: " ++ response.error)
4051+ logWarn ("federation: batch forward to " ++ server ++ " rejected: " ++ response.error ++ ", falling back to single-message forwards")
3973− -- Application-level batch rejection (old peer, invalid request): increment retries
4052+ -- Application-level batch rejection (old peer, invalid request): fall back to
3974− retryForwardsBatchIncrement batch server)
4053+ -- sequential single-message forwards instead of burning retries. The peer likely
4054+ -- doesn't support /federation/forward-batch, so individual forwards may succeed.
4055+ retryForwardsSingle batch server)
3976-- Increment retry counts for all entries in a failed batch
4057-- Increment retry counts for all entries in a failed batch
3977retryForwardsBatchIncrement = \batch server ->
4058retryForwardsBatchIncrement = \batch server ->
@@ -3982,6 +4063,34 @@ retryForwardsBatchIncrement = \batch server ->
3982 incrementForwardRetries f.messageId server f.toKey
4063 incrementForwardRetries f.messageId server f.toKey
3983 retryForwardsBatchIncrement rest server
4064 retryForwardsBatchIncrement rest server
4066+-- Fallback: send forwards individually when batch forward is rejected (old peer).
4067+-- Each message gets its own auth token and HTTP round-trip. On success, removes the
4068+-- forward entry; on failure, increments retries (same as batch per-message failure).
4069+retryForwardsSingle = \batch server ->
4070+ case headOf batch of
4071+ Nothing {} -> yield {}
4072+ Just {value: f} -> do
4073+ let rest = do x <- batch; where x.messageId != f.messageId; yield x
4075+ auth <- generateFederationAuth "forward" server
4076+ corrId <- genTokenN 1
4077+ result <- fetchFederation server (ForwardMessage {
4078+ toKey: f.toKey, encryptedBlob: f.encryptedBlob,
4079+ timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
4080+ fromServer: serverHostN, requestTimestamp: t,
4081+ authorization: auth, correlationId: corrId,
4082+ expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})
4085+ logWarn ("federation: single forward to " ++ server ++ " failed for " ++ f.messageId ++ ": " ++ show e)
4086+ recordFailure server
4088+ if response.ok then do
4089+ removeForward f.messageId server f.toKey
4090+ clearFailure server
4091+ else incrementForwardRetries f.messageId server f.toKey
4092+ retryForwardsSingle rest server
3985-- Process per-message results from a batch forward response
4094-- Process per-message results from a batch forward response
3986retryForwardsBatchResults = \batch results server ->
4095retryForwardsBatchResults = \batch results server ->
3987 case headOf batch of
4096 case headOf batch of
@@ -4043,15 +4152,28 @@ retryPendingForwardsFromBuckets = do
4043 activePolls <- *activePolls
4152 activePolls <- *activePolls
4044 let onlineForwards = do f <- eligibleForwards; where any activePolls (\p -> p.pubkey == f.toKey); yield f
4153 let onlineForwards = do f <- eligibleForwards; where any activePolls (\p -> p.pubkey == f.toKey); yield f
4045 let offlineForwards = do f <- eligibleForwards; where none activePolls (\p -> p.pubkey == f.toKey); yield f
4154 let offlineForwards = do f <- eligibleForwards; where none activePolls (\p -> p.pubkey == f.toKey); yield f
4046− -- Sort by priority descending (higher priority first) within each group
4155+ -- Sort by effective priority descending (higher priority first) within each group.
4156+ -- Penalty by retry count: messages with many retries are deprioritized so fresh
4157+ -- messages aren't blocked behind likely-failing ones. A fresh priority-5 (score -5)
4158+ -- beats a retry-6 priority-5 (score 1).
4047− let sortedOnline = sortBy (\f -> 0 - f.priority) onlineForwards
4159+ let sortedOnline = sortBy (\f -> 0 - f.priority + f.retries) onlineForwards
4048− let sortedOffline = sortBy (\f -> 0 - f.priority) offlineForwards
4160+ let sortedOffline = sortBy (\f -> 0 - f.priority + f.retries) offlineForwards
4049 let prioritizedForwards = union sortedOnline sortedOffline
4161 let prioritizedForwards = union sortedOnline sortedOffline
4050 -- Prioritize servers that have online-key forwards first
4162 -- Prioritize servers that have online-key forwards first
4051 let onlineServers = collectPeerServers onlineForwards (\x -> x.toServer)
4163 let onlineServers = collectPeerServers onlineForwards (\x -> x.toServer)
4052 let offlineOnlyServers = do s <- eligibleServers; where none onlineServers (\os -> os == s); yield s
4164 let offlineOnlyServers = do s <- eligibleServers; where none onlineServers (\os -> os == s); yield s
4053 let prioritizedServers = union onlineServers offlineOnlyServers
4165 let prioritizedServers = union onlineServers offlineOnlyServers
4166+ -- Forward storm prevention: after restore, limit per-server retries to a fraction that
4167+ -- increases each prune cycle. Prevents thousands of restored forwards from overwhelming
4168+ -- peers in the first cycle. rampRemaining counts down from restoreForwardRampCycles to 0.
4169+ rampRemaining <- *restoreForwardRampRemaining
4170+ let effectiveMaxRetries = if rampRemaining > 0
4171+ then max 1 (maxForwardRetriesPerServer * (restoreForwardRampCycles - rampRemaining + 1) / restoreForwardRampCycles)
4172+ else maxForwardRetriesPerServer
4173+ when (rampRemaining > 0) (do
4174+ logInfo ("federation: post-restore ramp " ++ show (restoreForwardRampCycles - rampRemaining + 1) ++ "/" ++ show restoreForwardRampCycles ++ " (max " ++ show effectiveMaxRetries ++ " retries/server)")
4175+ atomic do set *restoreForwardRampRemaining = rampRemaining - 1)
4054− sendForwardsPartitioned prioritizedForwards prioritizedServers maxForwardRetriesPerServer
4176+ sendForwardsPartitioned prioritizedForwards prioritizedServers effectiveMaxRetries
4056-- When a remote key comes online, create forward entries for queued messages and send them
4178-- When a remote key comes online, create forward entries for queued messages and send them
4057forwardQueuedMessages = \targets t -> do
4179forwardQueuedMessages = \targets t -> do
@@ -4261,6 +4383,82 @@ retryOverflowAcks = \overflow -> do
4261 ackOverflow <- *ackOverflow
4383 ackOverflow <- *ackOverflow
4262 set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))
4384 set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))
4386+-- Queue a single failed cancel for retry during the next prune cycle.
4387+queuePendingCancel = \messageId toKey server t -> atomic do
4388+ pcs <- *pendingFederationCancels
4389+ -- Dedup: don't queue if already pending for the same (messageId, toKey, server)
4390+ let isDup = any pcs (\c -> c.messageId == messageId && c.toKey == toKey && c.toServer == server)
4391+ when (not isDup) (do
4392+ let updated = union pcs [{messageId: messageId, toKey: toKey, toServer: server, createdAt: t, retries: 0}]
4393+ set *pendingFederationCancels = if count updated > maxPendingCancels
4394+ then evictOneOldest updated (\c -> c.createdAt)
4397+-- Queue multiple failed cancels for retry (batch cancel failure).
4398+queuePendingCancels = \cancels server t ->
4399+ case headOf cancels of
4400+ Nothing {} -> yield {}
4401+ Just {value: c} -> do
4402+ let rest = do x <- cancels; where x.messageId != c.messageId || x.toKey != c.toKey; yield x
4403+ queuePendingCancel c.messageId c.toKey server t
4404+ queuePendingCancels rest server t
4406+-- Retry pending federation cancels. Called during the prune cycle.
4407+-- Groups cancels by server and sends batch cancels, respecting backoff.
4408+retryPendingCancels : IO {network} {}
4409+retryPendingCancels = do
4411+ pending <- *pendingFederationCancels
4412+ if count pending == 0 then yield {}
4414+ failures <- *forwardFailures
4415+ if isFederationCircuitOpen failures t
4416+ then logWarn "federation: circuit breaker open, skipping cancel retries"
4418+ inFlight <- *federationInFlight
4419+ blocked <- *blockedPeers
4420+ let servers = collectPeerServers pending (\c -> c.toServer)
4421+ let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
4422+ sendToEach eligibleServers (\server -> do
4423+ let serverCancels = do c <- pending; where c.toServer == server; yield c
4424+ forEachBatch serverCancels maxBatchSize (\batch -> do
4426+ auth <- generateFederationAuth "cancel" server
4427+ corrId <- genTokenN 1
4428+ let payload = do c <- batch; yield {messageId: c.messageId, toKey: c.toKey}
4429+ result <- fetchFederation server (FederationBatchCancel {
4430+ cancels: payload, fromServer: serverHostN,
4431+ requestTimestamp: cancelT, correlationId: corrId,
4432+ authorization: auth, protocolVersion: federationProtocolVersion})
4435+ logWarn ("federation: cancel retry to " ++ server ++ " failed: " ++ show e)
4436+ recordFailure server
4437+ incrementCancelRetries batch server
4440+ clearFailure server
4441+ removePendingCancels batch server
4443+ logWarn ("federation: cancel retry to " ++ server ++ " rejected: " ++ r.error)
4444+ incrementCancelRetries batch server))
4446+-- Increment retry counters for failed cancel entries; drop entries exceeding maxCancelRetries.
4447+incrementCancelRetries = \batch server -> atomic do
4448+ pcs <- *pendingFederationCancels
4449+ let isTarget = \c -> c.toServer == server && any batch (\b -> b.messageId == c.messageId && c.toKey == b.toKey)
4450+ let toDrop = do c <- pcs; where isTarget c && c.retries + 1 >= maxCancelRetries; yield c
4451+ when (count toDrop > 0)
4452+ (logWarn ("federation: dropped " ++ show (count toDrop) ++ " cancel entries to " ++ server ++ " after " ++ show maxCancelRetries ++ " retries"))
4453+ set *pendingFederationCancels = do c <- pcs
4454+ where not (isTarget c && c.retries + 1 >= maxCancelRetries)
4455+ yield (if isTarget c then {messageId: c.messageId, toKey: c.toKey, toServer: c.toServer, createdAt: c.createdAt, retries: c.retries + 1} else c)
4457+-- Remove successfully delivered cancel entries.
4458+removePendingCancels = \batch server -> atomic do
4459+ pcs <- *pendingFederationCancels
4460+ set *pendingFederationCancels = do c <- pcs; where not (c.toServer == server && any batch (\b -> b.messageId == c.messageId && c.toKey == b.toKey)); yield c
4264-- Wait for all in-flight federation requests to complete or timeout.
4462-- Wait for all in-flight federation requests to complete or timeout.
4265-- Uses STM retry instead of busy-wait: blocks until *federationInFlight changes
4463-- Uses STM retry instead of busy-wait: blocks until *federationInFlight changes
4266-- (when clearFederationInFlight writes) or *drainTimedOut is set by the timeout fork.
4464-- (when clearFederationInFlight writes) or *drainTimedOut is set by the timeout fork.
@@ -4418,6 +4616,13 @@ pruneAndRetry = \t -> do
4418 set *verificationCache = do c <- cache; where t - c.verifiedAt < verificationCacheMs; yield c
4616 set *verificationCache = do c <- cache; where t - c.verifiedAt < verificationCacheMs; yield c
4419 -- Reset per-peer latency tracking with global window
4617 -- Reset per-peer latency tracking with global window
4420 set *peerLatencies = []
4618 set *peerLatencies = []
4619+ -- Reset client operation latency metrics (windowed with federation metrics)
4620+ set *clientPollLatencyTotal = 0
4621+ set *clientPollLatencyCount = 0
4622+ set *clientPollLatencyMax = 0
4623+ set *clientSendLatencyTotal = 0
4624+ set *clientSendLatencyCount = 0
4625+ set *clientSendLatencyMax = 0
4421 set *federationLatencyResetAt = t)
4626 set *federationLatencyResetAt = t)
4422 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
4627 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
4423 -- These were queued by handleFederationAck instead of fork+sleep.
4628 -- These were queued by handleFederationAck instead of fork+sleep.
@@ -4435,7 +4640,8 @@ pruneAndRetry = \t -> do
4435 when federationEnabled (do
4640 when federationEnabled (do
4436 retryPendingForwardsFromBuckets
4641 retryPendingForwardsFromBuckets
4437 sendAcksAndClean result.pendingAcks
4642 sendAcksAndClean result.pendingAcks
4438− retryOverflowAcks result.pendingOverflow))
4643+ retryOverflowAcks result.pendingOverflow
4644+ retryPendingCancels))
4439 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
4645 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
4440 -- trigger offline gossip when they expire (avoids wasted federation traffic)
4646 -- trigger offline gossip when they expire (avoids wasted federation traffic)
4441 when result.shouldRefresh (do
4647 when result.shouldRefresh (do
@@ -4535,14 +4741,19 @@ ackPolledMessages = \pubkey cursor t -> do
4535 let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
4741 let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
4536 when (count localDelivered > 0) (do
4742 when (count localDelivered > 0) (do
4537 sd <- *senderDeliveries
4743 sd <- *senderDeliveries
4744+ -- Dedup: skip receipts already recorded (e.g., cursor ack + explicit ack for same message)
4745+ let deduped = do d <- localDelivered; where none sd (\e -> e.messageId == d.messageId && e.toKey == d.toKey); yield d
4538− let combined = union sd localDelivered
4746+ let combined = union sd deduped
4539 let overCap = count combined > maxSenderDeliveries
4747 let overCap = count combined > maxSenderDeliveries
4540− set *senderDeliveries = if overCap
4748+ if overCap then do
4541− then evictOneOldest combined (\d -> d.deliveredAt)
4749+ let evicted = extractOldest combined (\d -> d.deliveredAt)
4542− else combined
4750+ set *senderDeliveries = evicted.items
4543− when overCap (do
4751+ case evicted.candidate of
4752+ Nothing {} -> yield {}
4753+ Just {value: ev} -> do
4544− ddr <- *droppedDeliveryReceipts
4754+ ddr <- *droppedDeliveryReceipts
4545− set *droppedDeliveryReceipts = ddr + 1))
4755+ set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1
4756+ else set *senderDeliveries = combined)
4546 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
4757 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
4547 else yield {pendingAcks: [], evictedEntries: []}
4758 else yield {pendingAcks: [], evictedEntries: []}
@@ -4571,7 +4782,8 @@ pollNewEvents = \pubkey clampedCursor t -> do
4571 yield {id: m.messageId, eventType: "message",
4782 yield {id: m.messageId, eventType: "message",
4572 payload: {toKey: m.toKey,
4783 payload: {toKey: m.toKey,
4573 encryptedBlob: m.encryptedBlob,
4784 encryptedBlob: m.encryptedBlob,
4574− timestamp: m.timestamp}}
4785+ timestamp: m.timestamp,
4786+ expiresAt: m.expiresAt}}
4575 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
4787 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
4576 -- hasMore: tells clients there are additional messages beyond this batch,
4788 -- hasMore: tells clients there are additional messages beyond this batch,
4577 -- so they can immediately re-poll without waiting for the next heartbeat.
4789 -- so they can immediately re-poll without waiting for the next heartbeat.
@@ -4595,8 +4807,13 @@ pollOnce = \token cursor clientIp -> do
4595 if not (matchIp clientIp ip)
4807 if not (matchIp clientIp ip)
4596 then yield (Nothing {})
4808 then yield (Nothing {})
4598− -- Refresh session expiry (sliding window keeps active sessions alive)
4810+ -- Sliding window: refresh session expiry when within half of sessionExpiry.
4811+ -- Avoids rewriting the session bucket on every poll (reduces write contention
4812+ -- on the hottest path). Uses pubkey pre-filter so safeCompare only runs on
4813+ -- sessions matching the authenticated pubkey (~1-2 per bucket vs ~600).
4814+ let needsRefresh = any sessions (\s -> s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 2)
4599− writeSessionBucket bucket (do s <- sessions; yield (if safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip, createdAt: s.createdAt} else s))
4816+ (writeSessionBucket bucket (do s <- sessions; yield (if s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 2 && safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip, createdAt: s.createdAt} else s)))
4600 seqCounter <- *seqCounter
4817 seqCounter <- *seqCounter
4601 let clampedCursor = clamp 0 (max seqCounter t) cursor
4818 let clampedCursor = clamp 0 (max seqCounter t) cursor
4602 ackResult <- if clampedCursor > 0
4819 ackResult <- if clampedCursor > 0
@@ -4733,7 +4950,7 @@ forwardCandidates = \targets messages forwards t ->
4733-- Case 4 prevents an attacker who intercepts the challenge string from verifying
4950-- Case 4 prevents an attacker who intercepts the challenge string from verifying
4734-- without going through the same reverse proxy that issued the challenge.
4951-- without going through the same reverse proxy that issued the challenge.
4735challengeMatch = \challenges pubkey challenge ip t ->
4952challengeMatch = \challenges pubkey challenge ip t ->
4736− any challenges (\c -> c.pubkey == pubkey && safeCompare c.challenge challenge && c.expiresAt > t && (if c.ip == "" then True {} else if ip == "" then False {} else c.ip == ip))
4953+ any challenges (\c -> c.pubkey == pubkey && safeCompare c.challenge challenge && c.expiresAt > t && matchIp ip c.ip)
4738challengesExcluding = \challenges pubkey challenge ->
4955challengesExcluding = \challenges pubkey challenge ->
4739 do c <- challenges; where c.pubkey != pubkey || c.challenge != challenge; yield c
4956 do c <- challenges; where c.pubkey != pubkey || c.challenge != challenge; yield c
@@ -5059,17 +5276,17 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
5059 yield capped
5276 yield capped
5060 -- Poll slot exhaustion: signal clients to back off for one poll cycle
5277 -- Poll slot exhaustion: signal clients to back off for one poll cycle
5061 let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
5278 let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
5062− -- Read and reset drop counters so clients know if they missed receipts/expirations.
5279+ -- Read and drain per-sender drop counters so this sender knows if their receipts were lost.
5063− -- Counters are global (not per-key) — a non-zero value means *some* receipts were
5280+ -- Each sender sees only their own drops (not a global counter conflating all senders).
5064− -- dropped due to capacity, though not necessarily for this sender. This is acceptable:
5065− -- the client treats it as advisory ("some receipts may have been lost") rather than precise.
5066− drops <- if not result.authorized then yield {dr: 0, de: 0}
5281+ drops <- if not result.authorized || result.pubkey == "" then yield {dr: 0, de: 0}
5067 else atomic do
5282 else atomic do
5068− dr <- *droppedDeliveryReceipts
5283+ ddr <- *droppedDeliveryReceipts
5069− de <- *droppedExpirationNotifs
5284+ let drResult = drainDropCounter ddr result.pubkey
5070− when (dr > 0) (set *droppedDeliveryReceipts = 0)
5285+ when (drResult.count > 0) (set *droppedDeliveryReceipts = drResult.remaining)
5286+ den <- *droppedExpirationNotifs
5287+ let deResult = drainDropCounter den result.pubkey
5071− when (de > 0) (set *droppedExpirationNotifs = 0)
5288+ when (deResult.count > 0) (set *droppedExpirationNotifs = deResult.remaining)
5072− yield {dr: dr, de: de}
5289+ yield {dr: drResult.count, de: deResult.count}
5073 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, hasMore: result.hasMore, cursorStale: cursorStale, messageExpirations: expirations, messageDeliveries: deliveries, droppedDeliveryReceipts: drops.dr, droppedExpirationNotifs: drops.de})
5290 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, hasMore: result.hasMore, cursorStale: cursorStale, messageExpirations: expirations, messageDeliveries: deliveries, droppedDeliveryReceipts: drops.dr, droppedExpirationNotifs: drops.de})
5075-- Core send logic: dedup, rate limiting, store message, create forwards.
5292-- Core send logic: dedup, rate limiting, store message, create forwards.
@@ -5194,17 +5411,18 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
5195-- Send all batch forwards for one server sequentially (no per-message fork overhead).
5412-- Send all batch forwards for one server sequentially (no per-message fork overhead).
5196-- Used by handleSendBatch; reduces fork count from O(messages * servers) to O(servers).
5413-- Used by handleSendBatch; reduces fork count from O(messages * servers) to O(servers).
5414+-- Uses pre-computed fwdExpiresAt/fwdPriority from the batch result to avoid re-reading
5415+-- the forward bucket for those values; only checks if the forward entry still exists
5416+-- (it could be removed by a concurrent poll/ack between creation and this forked send).
5197sendBatchForwardsSequential = \msgs server ->
5417sendBatchForwardsSequential = \msgs server ->
5198 case headOf msgs of
5418 case headOf msgs of
5199 Nothing {} -> yield {}
5419 Nothing {} -> yield {}
5200 Just {value: r} -> do
5420 Just {value: r} -> do
5201 let rest = do m <- msgs; where m.messageId != r.messageId; yield m
5421 let rest = do m <- msgs; where m.messageId != r.messageId; yield m
5202− batchFwdInfo <- atomic do
5422+ stillPending <- atomic do
5203 fwdBkt <- readFwdBucket (pubkeyBucket r.toKey)
5423 fwdBkt <- readFwdBucket (pubkeyBucket r.toKey)
5204− case headOf (do f <- fwdBkt; where f.messageId == r.messageId && f.toServer == server; yield f) of
5424+ yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))
5205− Nothing {} -> yield {pending: False {}, fwdExpiresAt: 0, fwdPriority: 0}
5425+ when stillPending (do
5206− Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt, fwdPriority: fe.priority}
5207− when batchFwdInfo.pending (do
5208 fwdT <- now
5426 fwdT <- now
5209 fwdAuth <- generateFederationAuth "forward" server
5427 fwdAuth <- generateFederationAuth "forward" server
5210 fwdCorrId <- genTokenN 1
5428 fwdCorrId <- genTokenN 1
@@ -5213,7 +5431,7 @@ sendBatchForwardsSequential = \msgs server ->
5213 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
5431 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
5214 fromServer: serverHostN, requestTimestamp: fwdT,
5432 fromServer: serverHostN, requestTimestamp: fwdT,
5215 authorization: fwdAuth, correlationId: fwdCorrId,
5433 authorization: fwdAuth, correlationId: fwdCorrId,
5216− expiresAt: batchFwdInfo.fwdExpiresAt, priority: batchFwdInfo.fwdPriority, protocolVersion: federationProtocolVersion})
5434+ expiresAt: r.fwdExpiresAt, priority: r.fwdPriority, protocolVersion: federationProtocolVersion})
5217 case fwdResult of
5435 case fwdResult of
5218 Err e -> do
5436 Err e -> do
5219 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
5437 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
@@ -5765,14 +5983,19 @@ fedDeliveryReceipts = \bucketIds forwards t -> case headOf bucketIds of
5765 let deliveries = union fromForwards msgDeliveries
5983 let deliveries = union fromForwards msgDeliveries
5766 when (count deliveries > 0) (atomic do
5984 when (count deliveries > 0) (atomic do
5767 sd <- *senderDeliveries
5985 sd <- *senderDeliveries
5986+ -- Dedup: skip receipts already recorded (avoids duplicates from concurrent ack paths)
5987+ let deduped = do d <- deliveries; where none sd (\e -> e.messageId == d.messageId && e.toKey == d.toKey); yield d
5768− let combined = union sd deliveries
5988+ let combined = union sd deduped
5769 let overCap = count combined > maxSenderDeliveries
5989 let overCap = count combined > maxSenderDeliveries
5770− set *senderDeliveries = if overCap
5990+ if overCap then do
5771− then evictOneOldest combined (\d -> d.deliveredAt)
5991+ let evicted = extractOldest combined (\d -> d.deliveredAt)
5772− else combined
5992+ set *senderDeliveries = evicted.items
5773− when overCap (do
5993+ case evicted.candidate of
5994+ Nothing {} -> yield {}
5995+ Just {value: ev} -> do
5774− ddr <- *droppedDeliveryReceipts
5996+ ddr <- *droppedDeliveryReceipts
5775− set *droppedDeliveryReceipts = ddr + 1))
5997+ set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1
5998+ else set *senderDeliveries = combined)
5776 fedDeliveryReceipts rest forwards t
5999 fedDeliveryReceipts rest forwards t
5778-- POST /federation/ack: Peer server confirms delivery of messages we originated.
6001-- POST /federation/ack: Peer server confirms delivery of messages we originated.
@@ -6127,7 +6350,8 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
6127 let capped = takeRelation maxBatchSize cancels
6350 let capped = takeRelation maxBatchSize cancels
6128 -- Process each cancel, collecting results and per-server cancel targets
6351 -- Process each cancel, collecting results and per-server cancel targets
6129 results <- clientCancelBatchInner senderKey capped t [] []
6352 results <- clientCancelBatchInner senderKey capped t [] []
6130− -- Send federation batch cancels grouped by server
6353+ -- Send federation batch cancels grouped by server.
6354+ -- Failed cancels are queued for retry during the next prune cycle.
6131 fork (when (federationEnabled && count results.cancelServers > 0)
6355 fork (when (federationEnabled && count results.cancelServers > 0)
6132 (sendToEach results.cancelServers (\server -> do
6356 (sendToEach results.cancelServers (\server -> do
6133 let serverCancels = do c <- results.fedCancels; where any c.servers (\s -> s == server); yield {messageId: c.messageId, toKey: c.toKey}
6357 let serverCancels = do c <- results.fedCancels; where any c.servers (\s -> s == server); yield {messageId: c.messageId, toKey: c.toKey}
@@ -6135,11 +6359,17 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
6135 cancelT <- now
6359 cancelT <- now
6136 cancelAuth <- generateFederationAuth "cancel" server
6360 cancelAuth <- generateFederationAuth "cancel" server
6137 cancelCorrId <- genTokenN 1
6361 cancelCorrId <- genTokenN 1
6138− fetchFederation server (FederationBatchCancel {
6362+ cancelResult <- fetchFederation server (FederationBatchCancel {
6139 cancels: serverCancels, fromServer: serverHostN,
6363 cancels: serverCancels, fromServer: serverHostN,
6140 requestTimestamp: cancelT, correlationId: cancelCorrId,
6364 requestTimestamp: cancelT, correlationId: cancelCorrId,
6141 authorization: cancelAuth, protocolVersion: federationProtocolVersion})
6365 authorization: cancelAuth, protocolVersion: federationProtocolVersion})
6142− yield {}))))
6366+ case cancelResult of
6368+ logWarn ("federation: batch cancel to " ++ server ++ " failed: " ++ show e)
6369+ queuePendingCancels serverCancels server cancelT
6370+ Ok r -> when (not r.ok) (do
6371+ logWarn ("federation: batch cancel to " ++ server ++ " rejected: " ++ r.error)
6372+ queuePendingCancels serverCancels server cancelT)))))
6143 yield (respond {results: results.results, authorized: True {}, error: ""}))
6373 yield (respond {results: results.results, authorized: True {}, error: ""}))
6145-- Process each client cancel entry, returning results and federation cancel targets.
6375-- Process each client cancel entry, returning results and federation cancel targets.
@@ -6245,8 +6475,13 @@ handleDrain = \phase authorization respond -> do
6245 pac <- *pendingAckCleanups
6475 pac <- *pendingAckCleanups
6246 se <- *senderExpirations
6476 se <- *senderExpirations
6247 sdl <- *senderDeliveries
6477 sdl <- *senderDeliveries
6478+ rlGlobal <- *rateOverrideGlobalRequest; rlIp <- *rateOverrideIpRate; rlSend <- *rateOverrideSendRate
6479+ rlFed <- *rateOverrideFederationRate; rlAuth <- *rateOverrideAuthGlobal; rlRecv <- *rateOverrideRecvRate
6480+ rlChal <- *rateOverrideChallengeRate; rlPair <- *rateOverridePairRate; rlLookup <- *rateOverrideLookupRate
6481+ rlWrite <- *rateOverrideClientWriteRate
6482+ let drainRlo = {globalRequest: rlGlobal, ipRate: rlIp, sendRate: rlSend, federationRate: rlFed, authGlobal: rlAuth, recvRate: rlRecv, challengeRate: rlChal, pairRate: rlPair, lookupRate: rlLookup, clientWriteRate: rlWrite}
6248 let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
6483 let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
6249− yield {version: snapshotVersion, checksum: drainCs, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se, senderDeliveries: sdl}
6484+ yield {version: snapshotVersion, checksum: drainCs, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se, senderDeliveries: sdl, rateLimitOverrides: drainRlo}
6250 -- Sentinel-wrapped: sidecar processes should verify [SNAPSHOT_END] follows to detect truncation.
6485 -- Sentinel-wrapped: sidecar processes should verify [SNAPSHOT_END] follows to detect truncation.
6251 -- Integrity tag appended when maintenance secret is configured (enables tamper detection on restore).
6486 -- Integrity tag appended when maintenance secret is configured (enables tamper detection on restore).
6252 runtimeSec <- *runtimeMaintenanceSecret
6487 runtimeSec <- *runtimeMaintenanceSecret
@@ -6490,7 +6725,7 @@ handleHealth = \authorization respond -> do
6490 draining: result.isDraining,
6725 draining: result.isDraining,
6491 degradedPeers: 0})
6726 degradedPeers: 0})
6493−zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: []}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, prunesSinceSnapshot: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0}}
6728+zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: []}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, prunesSinceSnapshot: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0}}
6495handleStats = \authorization respond -> do
6730handleStats = \authorization respond -> do
6496 t <- now
6731 t <- now
@@ -6560,7 +6795,13 @@ handleStats = \authorization respond -> do
6560 rlf <- *rateLimitedFederation
6795 rlf <- *rateLimitedFederation
6561 arq <- *activeRequests
6796 arq <- *activeRequests
6562 parq <- *peakActiveRequests
6797 parq <- *peakActiveRequests
6798+ cplt <- *clientPollLatencyTotal
6799+ cplc <- *clientPollLatencyCount
6800+ cplm <- *clientPollLatencyMax
6801+ cslt <- *clientSendLatencyTotal
6802+ cslc <- *clientSendLatencyCount
6803+ cslm <- *clientSendLatencyMax
6563− yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, pss: pss, flt: flt, flc: flc, flm: flm, svc: svc, vfc: vfc, dca: dca, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf, arq: arq, parq: parq}
6804+ yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, pss: pss, flt: flt, flc: flc, flm: flm, svc: svc, vfc: vfc, dca: dca, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf, arq: arq, parq: parq, cplt: cplt, cplc: cplc, cplm: cplm, cslt: cslt, cslc: cslc, cslm: cslm}
6564 -- Compute derived metrics
6805 -- Compute derived metrics
6565 let sessions = core.sc
6806 let sessions = core.sc
6566 let messages = core.tmc
6807 let messages = core.tmc
@@ -6582,11 +6823,13 @@ handleStats = \authorization respond -> do
6582 -- Top-3 failing peers by failure count (inline in /stats to reduce monitoring queries)
6823 -- Top-3 failing peers by failure count (inline in /stats to reduce monitoring queries)
6583 let topFailing = takeRelation 3 (sortBy (\f -> 0 - f.failures) core.ff)
6824 let topFailing = takeRelation 3 (sortBy (\f -> 0 - f.failures) core.ff)
6584 let topFailingPeers = do f <- topFailing; yield {server: f.server, failures: f.failures}
6825 let topFailingPeers = do f <- topFailing; yield {server: f.server, failures: f.failures}
6826+ let pollAvg = if counters.cplc > 0 then counters.cplt / counters.cplc else 0
6827+ let sendAvg = if counters.cslc > 0 then counters.cslt / counters.cslc else 0
6585 yield (respond {ok: True {}, version: serverBuildVersion,
6828 yield (respond {ok: True {}, version: serverBuildVersion,
6586 capacity: {sessions: sessions, sessionsMax: maxSessions, sessionsPct: sessPct, messages: messages, messagesMax: maxMessages, messagesPct: msgPct, totalMessageCount: messages, presence: presence, presenceMax: maxPresenceEntries, presencePct: presPct, forwards: fwds, forwardsMax: maxForwardsEntries, forwardsPct: fwdPct, challenges: count core.c, challengesMax: maxChallenges, ackedIds: core.aCount, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: core.dCount, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
6829 capacity: {sessions: sessions, sessionsMax: maxSessions, sessionsPct: sessPct, messages: messages, messagesMax: maxMessages, messagesPct: msgPct, totalMessageCount: messages, presence: presence, presenceMax: maxPresenceEntries, presencePct: presPct, forwards: fwds, forwardsMax: maxForwardsEntries, forwardsPct: fwdPct, challenges: count core.c, challengesMax: maxChallenges, ackedIds: core.aCount, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: core.dCount, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
6587 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, slowVerifies: counters.svc, verifyFailures: counters.vfc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca, forwardServers: fwdServerCount, maxForwardQueueDepth: maxFwdDepth, topFailingPeers: topFailingPeers},
6830 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, slowVerifies: counters.svc, verifyFailures: counters.vfc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca, forwardServers: fwdServerCount, maxForwardQueueDepth: maxFwdDepth, topFailingPeers: topFailingPeers},
6588 rates: {send: count counters.sr, challenge: count counters.cr, federation: count counters.fr, federationPair: count counters.fpr, global: counters.grc, ip: count counters.ir, lookup: count counters.lr, recv: count counters.rr, pair: count counters.prr, gossipEvent: count core.ger, verifyIp: count core.vir, uniqueKeyIp: count core.ukir, pendingAckCleanups: count core.pac, senderExpirations: count core.se, senderDeliveries: count core.sdl, rejectedGlobal: counters.rlg, rejectedIp: counters.rli, rejectedAuth: counters.rla, rejectedPerKey: counters.rlk, rejectedFederation: counters.rlf},
6831 rates: {send: count counters.sr, challenge: count counters.cr, federation: count counters.fr, federationPair: count counters.fpr, global: counters.grc, ip: count counters.ir, lookup: count counters.lr, recv: count counters.rr, pair: count counters.prr, gossipEvent: count core.ger, verifyIp: count core.vir, uniqueKeyIp: count core.ukir, pendingAckCleanups: count core.pac, senderExpirations: count core.se, senderDeliveries: count core.sdl, rejectedGlobal: counters.rlg, rejectedIp: counters.rli, rejectedAuth: counters.rla, rejectedPerKey: counters.rlk, rejectedFederation: counters.rlf},
6589− timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0, drainPhase: counters.dr, prunesSinceSnapshot: counters.pss, totalEntries: sessions + messages + presence + fwds + count core.c + core.aCount + count core.ao + core.dCount + core.fnCount + count core.ap + count core.fi, activeRequests: counters.arq, peakActiveRequests: counters.parq}})
6832+ timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0, drainPhase: counters.dr, prunesSinceSnapshot: counters.pss, totalEntries: sessions + messages + presence + fwds + count core.c + core.aCount + count core.ao + core.dCount + core.fnCount + count core.ap + count core.fi, activeRequests: counters.arq, peakActiveRequests: counters.parq, pollAvgMs: pollAvg, pollMaxMs: counters.cplm, pollCount: counters.cplc, sendAvgMs: sendAvg, sendMaxMs: counters.cslm, sendCount: counters.cslc}})
6590 else yield (respond zeroStats)
6833 else yield (respond zeroStats)
6592handleStatsPeers = \authorization respond -> do
6835handleStatsPeers = \authorization respond -> do
@@ -6675,7 +6918,9 @@ handleReady = \respond -> do
6675 let isReady = lastPrune > 0 && lastPrune >= sa
6918 let isReady = lastPrune > 0 && lastPrune >= sa
6676 yield (respond {ready: isReady})
6919 yield (respond {ready: isReady})
6921+zeroRateLimitOverrides = {globalRequest: 0, ipRate: 0, sendRate: 0, federationRate: 0, authGlobal: 0, recvRate: 0, challengeRate: 0, pairRate: 0, lookupRate: 0, clientWriteRate: 0}
6678−emptySnapshot = {ok: False {}, version: 0, checksum: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: [], allowedPeers: [], pendingAckCleanups: [], senderExpirations: [], senderDeliveries: []}
6923+emptySnapshot = {ok: False {}, version: 0, checksum: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: [], allowedPeers: [], pendingAckCleanups: [], senderExpirations: [], senderDeliveries: [], rateLimitOverrides: zeroRateLimitOverrides}
6680-- GET /maintenance/snapshot: Export federation metadata for persistence.
6925-- GET /maintenance/snapshot: Export federation metadata for persistence.
6681-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
6926-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
@@ -6699,20 +6944,26 @@ handleSnapshot = \authorization respond -> do
6699 pac <- *pendingAckCleanups
6944 pac <- *pendingAckCleanups
6700 se <- *senderExpirations
6945 se <- *senderExpirations
6701 sdl <- *senderDeliveries
6946 sdl <- *senderDeliveries
6947+ -- Read rate limit overrides for snapshot persistence (v14+)
6948+ rlGlobal <- *rateOverrideGlobalRequest; rlIp <- *rateOverrideIpRate; rlSend <- *rateOverrideSendRate
6949+ rlFed <- *rateOverrideFederationRate; rlAuth <- *rateOverrideAuthGlobal; rlRecv <- *rateOverrideRecvRate
6950+ rlChal <- *rateOverrideChallengeRate; rlPair <- *rateOverridePairRate; rlLookup <- *rateOverrideLookupRate
6951+ rlWrite <- *rateOverrideClientWriteRate
6952+ let rlo = {globalRequest: rlGlobal, ipRate: rlIp, sendRate: rlSend, federationRate: rlFed, authGlobal: rlAuth, recvRate: rlRecv, challengeRate: rlChal, pairRate: rlPair, lookupRate: rlLookup, clientWriteRate: rlWrite}
6702− yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se, senderDeliveries: sdl}
6953+ yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se, senderDeliveries: sdl, rateLimitOverrides: rlo}
6703 atomic do
6954 atomic do
6704 set *lastSnapshotExportAt = t
6955 set *lastSnapshotExportAt = t
6705 set *prunesSinceSnapshot = 0
6956 set *prunesSinceSnapshot = 0
6706 let cs = snapshotChecksum (count result.presence) (count result.ackedIds) (count result.deliveredIds) (count result.forwards) (count result.forwardNonces)
6957 let cs = snapshotChecksum (count result.presence) (count result.ackedIds) (count result.deliveredIds) (count result.forwards) (count result.forwardNonces)
6707 logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards, " ++ show (count result.forwardNonces) ++ " forwardNonces, " ++ show (count result.blockedPeers) ++ " blockedPeers, " ++ show (count result.allowedPeers) ++ " allowedPeers, " ++ show (count result.pendingAckCleanups) ++ " pendingAckCleanups, " ++ show (count result.senderExpirations) ++ " senderExpirations, " ++ show (count result.senderDeliveries) ++ " senderDeliveries, checksum=" ++ show cs)
6958 logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards, " ++ show (count result.forwardNonces) ++ " forwardNonces, " ++ show (count result.blockedPeers) ++ " blockedPeers, " ++ show (count result.allowedPeers) ++ " allowedPeers, " ++ show (count result.pendingAckCleanups) ++ " pendingAckCleanups, " ++ show (count result.senderExpirations) ++ " senderExpirations, " ++ show (count result.senderDeliveries) ++ " senderDeliveries, checksum=" ++ show cs)
6708− yield (respond {ok: True {}, version: snapshotVersion, checksum: cs, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces, blockedPeers: result.blockedPeers, allowedPeers: result.allowedPeers, pendingAckCleanups: result.pendingAckCleanups, senderExpirations: result.senderExpirations, senderDeliveries: result.senderDeliveries})
6959+ yield (respond {ok: True {}, version: snapshotVersion, checksum: cs, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces, blockedPeers: result.blockedPeers, allowedPeers: result.allowedPeers, pendingAckCleanups: result.pendingAckCleanups, senderExpirations: result.senderExpirations, senderDeliveries: result.senderDeliveries, rateLimitOverrides: result.rateLimitOverrides})
6710-- POST /maintenance/restore: Import a previously-exported snapshot.
6961-- POST /maintenance/restore: Import a previously-exported snapshot.
6711-- Requires drain mode to prevent conflicts with live state. Merges imported data
6962-- Requires drain mode to prevent conflicts with live state. Merges imported data
6712-- with existing state (dedup entries are unioned, presence is merged by freshness).
6963-- with existing state (dedup entries are unioned, presence is merged by freshness).
6713-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
6964-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
6714-- trigger federation forward creation from messages + restored presence.
6965-- trigger federation forward creation from messages + restored presence.
6715−handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData blockedPeersData allowedPeersData pendingAckCleanupsData senderExpirationsData senderDeliveriesData authorization respond -> do
6966+handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData blockedPeersData allowedPeersData pendingAckCleanupsData senderExpirationsData senderDeliveriesData rateLimitOverridesData authorization respond -> do
6716 authorized <- verifyMaintenanceAuth authorization
6967 authorized <- verifyMaintenanceAuth authorization
6717 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6968 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
@@ -6819,7 +7070,24 @@ handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData de
6819 existingSenderDel <- *senderDeliveries
7070 existingSenderDel <- *senderDeliveries
6820 let mergedSenderDel = union existingSenderDel senderDeliveriesData
7071 let mergedSenderDel = union existingSenderDel senderDeliveriesData
6821 set *senderDeliveries = takeRelation maxSenderDeliveries mergedSenderDel
7072 set *senderDeliveries = takeRelation maxSenderDeliveries mergedSenderDel
7073+ -- Restore rate limit overrides (v14+; pre-v14 snapshots have all-zero fields via Knot defaults)
7074+ -- Non-zero values from the snapshot override compile-time defaults; 0 = keep current.
7075+ let rlo = rateLimitOverridesData
7076+ when (rlo.globalRequest > 0) (set *rateOverrideGlobalRequest = rlo.globalRequest)
7077+ when (rlo.ipRate > 0) (set *rateOverrideIpRate = rlo.ipRate)
7078+ when (rlo.sendRate > 0) (set *rateOverrideSendRate = rlo.sendRate)
7079+ when (rlo.federationRate > 0) (set *rateOverrideFederationRate = rlo.federationRate)
7080+ when (rlo.authGlobal > 0) (set *rateOverrideAuthGlobal = rlo.authGlobal)
7081+ when (rlo.recvRate > 0) (set *rateOverrideRecvRate = rlo.recvRate)
7082+ when (rlo.challengeRate > 0) (set *rateOverrideChallengeRate = rlo.challengeRate)
7083+ when (rlo.pairRate > 0) (set *rateOverridePairRate = rlo.pairRate)
7084+ when (rlo.lookupRate > 0) (set *rateOverrideLookupRate = rlo.lookupRate)
7085+ when (rlo.clientWriteRate > 0) (set *rateOverrideClientWriteRate = rlo.clientWriteRate)
6822 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
7086 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
7087+ -- Activate forward storm prevention: ramp up retries over several prune cycles
7088+ -- so restored forwards don't overwhelm peers in the first cycle.
7089+ when (count validForwards > 0) (atomic do
7090+ set *restoreForwardRampRemaining = restoreForwardRampCycles)
6823 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
7091 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
6824 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
7092 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
@@ -6875,14 +7143,19 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
6875 let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
7143 let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
6876 when (count localDelivered > 0) (do
7144 when (count localDelivered > 0) (do
6877 sd <- *senderDeliveries
7145 sd <- *senderDeliveries
7146+ -- Dedup: skip receipts already recorded (e.g., cursor ack + explicit ack for same message)
7147+ let deduped = do d <- localDelivered; where none sd (\e -> e.messageId == d.messageId && e.toKey == d.toKey); yield d
6878− let combined = union sd localDelivered
7148+ let combined = union sd deduped
6879 let overCap = count combined > maxSenderDeliveries
7149 let overCap = count combined > maxSenderDeliveries
6880− set *senderDeliveries = if overCap
7150+ if overCap then do
6881− then evictOneOldest combined (\d -> d.deliveredAt)
7151+ let evicted = extractOldest combined (\d -> d.deliveredAt)
6882− else combined
7152+ set *senderDeliveries = evicted.items
6883− when overCap (do
7153+ case evicted.candidate of
7154+ Nothing {} -> yield {}
7155+ Just {value: ev} -> do
6884− ddr <- *droppedDeliveryReceipts
7156+ ddr <- *droppedDeliveryReceipts
6885− set *droppedDeliveryReceipts = ddr + 1))
7157+ set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1
7158+ else set *senderDeliveries = combined)
6886 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
7159 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
6887 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}
7160 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}
6888 fork (when federationEnabled (do
7161 fork (when federationEnabled (do
@@ -6974,14 +7247,19 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
6974 let localDeleted = do m <- toDelete; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
7247 let localDeleted = do m <- toDelete; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
6975 when (count localDeleted > 0) (do
7248 when (count localDeleted > 0) (do
6976 sd <- *senderDeliveries
7249 sd <- *senderDeliveries
7250+ -- Dedup: skip receipts already recorded (avoids duplicates from concurrent ack paths)
7251+ let dedupedDel = do d <- localDeleted; where none sd (\e -> e.messageId == d.messageId && e.toKey == d.toKey); yield d
6977− let combinedDel = union sd localDeleted
7252+ let combinedDel = union sd dedupedDel
6978 let overCapDel = count combinedDel > maxSenderDeliveries
7253 let overCapDel = count combinedDel > maxSenderDeliveries
6979− set *senderDeliveries = if overCapDel
7254+ if overCapDel then do
6980− then evictOneOldest combinedDel (\d -> d.deliveredAt)
7255+ let evicted = extractOldest combinedDel (\d -> d.deliveredAt)
6981− else combinedDel
7256+ set *senderDeliveries = evicted.items
6982− when overCapDel (do
7257+ case evicted.candidate of
7258+ Nothing {} -> yield {}
7259+ Just {value: ev} -> do
6983− ddr <- *droppedDeliveryReceipts
7260+ ddr <- *droppedDeliveryReceipts
6984− set *droppedDeliveryReceipts = ddr + 1))
7261+ set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1
7262+ else set *senderDeliveries = combinedDel)
6985 if count ackedFederated > 0 then do
7263 if count ackedFederated > 0 then do
6986 ackedBkt <- readAckedBucket b
7264 ackedBkt <- readAckedBucket b
6987 let combinedAcked = union ackedBkt ackedFederated
7265 let combinedAcked = union ackedBkt ackedFederated
@@ -7068,19 +7346,26 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
7068 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
7346 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
7069 else updatedNonces)
7347 else updatedNonces)
7070 yield {cancelled: True {}, cancelServers: allCancelServers}
7348 yield {cancelled: True {}, cancelServers: allCancelServers}
7071− -- Send federation cancel to remote servers that may have the message
7349+ -- Send federation batch cancel to remote servers that may have the message.
7350+ -- Uses FederationBatchCancel (single-entry batch) instead of FederationCancel
7351+ -- for code path consistency with handleCancelBatch.
7352+ -- Failed cancels are queued for retry during the next prune cycle.
7072 fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)
7353 fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)
7073 (sendToEach cancelResult.cancelServers (\server -> do
7354 (sendToEach cancelResult.cancelServers (\server -> do
7074 cancelT <- now
7355 cancelT <- now
7075 cancelAuth <- generateFederationAuth "cancel" server
7356 cancelAuth <- generateFederationAuth "cancel" server
7076 cancelCorrId <- genTokenN 1
7357 cancelCorrId <- genTokenN 1
7077− cancelFedResult <- fetchFederation server (FederationCancel {
7358+ cancelFedResult <- fetchFederation server (FederationBatchCancel {
7078− messageId: messageId, toKey: to, fromServer: serverHostN,
7359+ cancels: [{messageId: messageId, toKey: to}], fromServer: serverHostN,
7079 requestTimestamp: cancelT, correlationId: cancelCorrId,
7360 requestTimestamp: cancelT, correlationId: cancelCorrId,
7080 authorization: cancelAuth, protocolVersion: federationProtocolVersion})
7361 authorization: cancelAuth, protocolVersion: federationProtocolVersion})
7081 case cancelFedResult of
7362 case cancelFedResult of
7082− Err e -> logWarn ("federation: cancel to " ++ server ++ " failed for " ++ messageId ++ ": " ++ show e)
7364+ logWarn ("federation: cancel to " ++ server ++ " failed for " ++ messageId ++ ": " ++ show e)
7365+ queuePendingCancel messageId to server cancelT
7366+ Ok r -> when (not r.ok) (do
7083− Ok r -> when (not r.ok) (logWarn ("federation: cancel to " ++ server ++ " rejected for " ++ messageId ++ ": " ++ r.error)))))
7367+ logWarn ("federation: cancel to " ++ server ++ " rejected for " ++ messageId ++ ": " ++ r.error)
7368+ queuePendingCancel messageId to server cancelT))))
7084 yield (respond {ok: True {}, authorized: True {}, cancelled: cancelResult.cancelled, error: ""}))
7369 yield (respond {ok: True {}, authorized: True {}, cancelled: cancelResult.cancelled, error: ""}))
7086-- POST /messages/batch: Send multiple messages in one request.
7371-- POST /messages/batch: Send multiple messages in one request.
@@ -7169,14 +7454,15 @@ sendBatchBucketGroup = \senderKey prepared t dp acc ->
7169 Nothing {} -> yield acc
7454 Nothing {} -> yield acc
7170 Just {value: entry} -> do
7455 Just {value: entry} -> do
7171 let rest = do p <- prepared; where p != entry; yield p
7456 let rest = do p <- prepared; where p != entry; yield p
7457+ let clampedPri = clamp 0 maxMessagePriority entry.msg.priority
7172 result <- if entry.stale
7458 result <- if entry.stale
7173− then yield {messageId: entry.effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, retryAfterMs: 0}
7459+ then yield {messageId: entry.effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, retryAfterMs: 0, fwdExpiresAt: 0, fwdPriority: 0}
7175 let clampedTs = clampTimestamp entry.msg.timestamp t
7461 let clampedTs = clampTimestamp entry.msg.timestamp t
7176− if dp >= 2 then yield {messageId: entry.effectiveMsgId, status: MsDraining {}, remoteServers: [], toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: 0}
7462+ if dp >= 2 then yield {messageId: entry.effectiveMsgId, status: MsDraining {}, remoteServers: [], toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: 0, fwdExpiresAt: 0, fwdPriority: 0}
7178− sendResult <- sendMessageCore senderKey entry.msg.to entry.msg.encryptedBlob clampedTs entry.effectiveMsgId entry.msg.expiresInMs (clamp 0 maxMessagePriority entry.msg.priority) t
7464+ sendResult <- sendMessageCore senderKey entry.msg.to entry.msg.encryptedBlob clampedTs entry.effectiveMsgId entry.msg.expiresInMs clampedPri t
7179− yield {messageId: entry.effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryAfterMs}
7465+ yield {messageId: entry.effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryAfterMs, fwdExpiresAt: t + effectiveMessageTtl entry.msg.expiresInMs, fwdPriority: clampedPri}
7180 sendBatchBucketGroup senderKey rest t dp (union acc [result])
7466 sendBatchBucketGroup senderKey rest t dp (union acc [result])
7182-- Log warnings for batch results outside the atomic block
7468-- Log warnings for batch results outside the atomic block
@@ -7713,7 +7999,7 @@ serveInner = \req -> case req of
7713 Shutdown {authorization, respond} -> handleShutdown authorization respond
7999 Shutdown {authorization, respond} -> handleShutdown authorization respond
7714 Undrain {authorization, respond} -> handleUndrain authorization respond
8000 Undrain {authorization, respond} -> handleUndrain authorization respond
7715 Snapshot {authorization, respond} -> handleSnapshot authorization respond
8001 Snapshot {authorization, respond} -> handleSnapshot authorization respond
7716− Restore {version, checksum, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, blockedPeers, allowedPeers, pendingAckCleanups, senderExpirations, senderDeliveries, authorization, respond} -> handleRestore version checksum presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces blockedPeers allowedPeers pendingAckCleanups senderExpirations senderDeliveries authorization respond
8002+ Restore {version, checksum, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, blockedPeers, allowedPeers, pendingAckCleanups, senderExpirations, senderDeliveries, rateLimitOverrides, authorization, respond} -> handleRestore version checksum presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces blockedPeers allowedPeers pendingAckCleanups senderExpirations senderDeliveries rateLimitOverrides authorization respond
7717 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
8003 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
7718 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
8004 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
7719 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
8005 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
@@ -7830,6 +8116,31 @@ serve = \req -> do
7830 set *activeRequests = max 0 (ar - 1)
8116 set *activeRequests = max 0 (ar - 1)
7831 endTime <- now
8117 endTime <- now
7832 let latency = endTime - startTime
8118 let latency = endTime - startTime
8119+ -- Record client operation latency for /stats monitoring (windowed avg/max).
8120+ -- Only tracks poll and send — the two highest-volume client operations.
8122+ Poll {} -> atomic do
8123+ total <- *clientPollLatencyTotal
8124+ cnt <- *clientPollLatencyCount
8125+ mx <- *clientPollLatencyMax
8126+ set *clientPollLatencyTotal = total + latency
8127+ set *clientPollLatencyCount = cnt + 1
8128+ set *clientPollLatencyMax = if latency > mx then latency else mx
8129+ SendMessage {} -> atomic do
8130+ total <- *clientSendLatencyTotal
8131+ cnt <- *clientSendLatencyCount
8132+ mx <- *clientSendLatencyMax
8133+ set *clientSendLatencyTotal = total + latency
8134+ set *clientSendLatencyCount = cnt + 1
8135+ set *clientSendLatencyMax = if latency > mx then latency else mx
8136+ SendBatch {} -> atomic do
8137+ total <- *clientSendLatencyTotal
8138+ cnt <- *clientSendLatencyCount
8139+ mx <- *clientSendLatencyMax
8140+ set *clientSendLatencyTotal = total + latency
8141+ set *clientSendLatencyCount = cnt + 1
8142+ set *clientSendLatencyMax = if latency > mx then latency else mx
7833 let path = requestPath req
8144 let path = requestPath req
7834 let ip = requestClientIp req
8145 let ip = requestClientIp req
7835 let ipSuffix = if ip != "" then " " ++ ip else ""
8146 let ipSuffix = if ip != "" then " " ++ ip else ""
@@ -8036,7 +8347,12 @@ backgroundSnapshot = do
8036 pac <- *pendingAckCleanups
8347 pac <- *pendingAckCleanups
8037 se <- *senderExpirations
8348 se <- *senderExpirations
8038 sdl <- *senderDeliveries
8349 sdl <- *senderDeliveries
8350+ rlGlobal <- *rateOverrideGlobalRequest; rlIp <- *rateOverrideIpRate; rlSend <- *rateOverrideSendRate
8351+ rlFed <- *rateOverrideFederationRate; rlAuth <- *rateOverrideAuthGlobal; rlRecv <- *rateOverrideRecvRate
8352+ rlChal <- *rateOverrideChallengeRate; rlPair <- *rateOverridePairRate; rlLookup <- *rateOverrideLookupRate
8353+ rlWrite <- *rateOverrideClientWriteRate
8354+ let autoRlo = {globalRequest: rlGlobal, ipRate: rlIp, sendRate: rlSend, federationRate: rlFed, authGlobal: rlAuth, recvRate: rlRecv, challengeRate: rlChal, pairRate: rlPair, lookupRate: rlLookup, clientWriteRate: rlWrite}
8039− yield {p: p, a: a, ao: ao, d: d, ff: ff, f: f, fn: fn, bp: bp, ap: ap, pac: pac, se: se, sdl: sdl}
8355+ yield {p: p, a: a, ao: ao, d: d, ff: ff, f: f, fn: fn, bp: bp, ap: ap, pac: pac, se: se, sdl: sdl, rlo: autoRlo}
8040 let p = exportSnapshot.p
8356 let p = exportSnapshot.p
8041 let a = exportSnapshot.a
8357 let a = exportSnapshot.a
8042 let ao = exportSnapshot.ao
8358 let ao = exportSnapshot.ao
@@ -8045,7 +8361,7 @@ backgroundSnapshot = do
8045 let f = exportSnapshot.f
8361 let f = exportSnapshot.f
8046 let fn = exportSnapshot.fn
8362 let fn = exportSnapshot.fn
8047 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
8363 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
8048− let exportData = {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: exportSnapshot.bp, allowedPeers: exportSnapshot.ap, pendingAckCleanups: exportSnapshot.pac, senderExpirations: exportSnapshot.se, senderDeliveries: exportSnapshot.sdl}
8364+ let exportData = {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: exportSnapshot.bp, allowedPeers: exportSnapshot.ap, pendingAckCleanups: exportSnapshot.pac, senderExpirations: exportSnapshot.se, senderDeliveries: exportSnapshot.sdl, rateLimitOverrides: exportSnapshot.rlo}
8049 autoRtSec <- *runtimeMaintenanceSecret
8365 autoRtSec <- *runtimeMaintenanceSecret
8050 let autoEffSec = if autoRtSec != "" then autoRtSec else maintenanceSecret
8366 let autoEffSec = if autoRtSec != "" then autoRtSec else maintenanceSecret
8051 let autoTag = snapshotIntegrityTag exportCs autoEffSec
8367 let autoTag = snapshotIntegrityTag exportCs autoEffSec
@@ -8247,6 +8563,68 @@ main = do
8247 when (autoSnapshotExportInterval > 0) (do
8563 when (autoSnapshotExportInterval > 0) (do
8248 logInfo (" auto-snapshot export: " ++ show autoSnapshotExportInterval ++ "ms (full [SNAPSHOT_DATA] to stdout)")
8564 logInfo (" auto-snapshot export: " ++ show autoSnapshotExportInterval ++ "ms (full [SNAPSHOT_DATA] to stdout)")
8249 logWarn "SECURITY: auto-snapshot exports unencrypted federation state to stdout — ensure log pipeline is secured and access-restricted")
8565 logWarn "SECURITY: auto-snapshot exports unencrypted federation state to stdout — ensure log pipeline is secured and access-restricted")
8566+ -- Bootstrap from peer: fetch snapshot from a trusted peer and auto-restore.
8567+ -- Runs before listen so the server starts with federation state.
8568+ when (bootstrapPeerUrl != "") (do
8569+ logInfo ("startup: bootstrapping from peer " ++ bootstrapPeerUrl ++ "...")
8570+ let authHeader = if bootstrapPeerSecret != "" then "Bearer " ++ bootstrapPeerSecret else ""
8571+ bootstrapResult <- fetch (federationUrl bootstrapPeerUrl) (Snapshot {authorization: authHeader})
8572+ case bootstrapResult of
8573+ Err e -> logWarn ("startup: bootstrap from " ++ bootstrapPeerUrl ++ " failed: " ++ show e ++ " (continuing without snapshot)")
8575+ if not snap.ok then logWarn ("startup: bootstrap from " ++ bootstrapPeerUrl ++ " returned ok=false (peer may require auth), continuing without snapshot")
8577+ logInfo ("startup: restoring bootstrap snapshot (version " ++ show snap.version ++ ", " ++ show (count snap.presence) ++ " presence, " ++ show (count snap.forwards) ++ " forwards)")
8578+ -- Inline restore logic (can't call handleRestore since we're not listening yet).
8579+ -- Only restore if version is compatible; skip checksum verification for simplicity.
8580+ if snap.version < 6 || snap.version > snapshotVersion
8581+ then logWarn ("startup: bootstrap snapshot version " ++ show snap.version ++ " incompatible (need 6-" ++ show snapshotVersion ++ "), skipping")
8584+ set *drainPhase = 2 -- temporarily drain for restore safety
8585+ -- Validate and merge (simplified version of handleRestore)
8586+ let validPresence = do p <- snap.presence; where not (isLocalServer p.server) && not (isBadServerName p.server); yield p
8587+ let validForwards = do f <- snap.forwards; where not (isLocalServer f.toServer) && (f.originServer == "" || not (isBadServerName f.originServer)) && not (isLocalServer f.originServer); yield f
8588+ let validAcked = do a <- snap.ackedIds; where not (isLocalServer a.originServer); yield a
8589+ let validOverflow = do a <- snap.ackOverflow; where not (isLocalServer a.originServer); yield a
8592+ let cappedPresence = takeRelation maxPresenceEntries (do p <- validPresence; where p.expiresAt > bt; yield p)
8593+ writeAllPresence cappedPresence
8594+ set *totalPresenceCount = count cappedPresence
8595+ set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)
8596+ set *presenceVersion = bt
8597+ writeAllAcked (takeRelation maxAckedIds validAcked)
8598+ set *ackOverflow = takeRelation maxAckOverflow validOverflow
8599+ writeAllDelivered (takeRelation maxDeliveredIds snap.deliveredIds)
8600+ let validFailures = do f <- snap.forwardFailures; where not (isBadServerName f.server) && not (isLocalServer f.server); yield f
8601+ set *forwardFailures = takeRelation maxForwardFailures validFailures
8602+ let cappedForwards = takeRelation maxForwardsEntries validForwards
8603+ writeAllForwards cappedForwards
8604+ set *totalForwardCount = count cappedForwards
8605+ let validNonces = do n <- snap.forwardNonces; where n.seenAt <= bt && bt - n.seenAt < forwardNonceRetention; yield n
8606+ writeAllNonces (takeRelation maxForwardNonces validNonces)
8607+ set *blockedPeers = takeRelation maxBlockedPeers snap.blockedPeers
8608+ set *allowedPeers = takeRelation maxAllowedPeers snap.allowedPeers
8609+ set *pendingAckCleanups = takeRelation maxPendingAckCleanups snap.pendingAckCleanups
8610+ set *senderExpirations = takeRelation maxSenderExpirations snap.senderExpirations
8611+ set *senderDeliveries = takeRelation maxSenderDeliveries snap.senderDeliveries
8612+ -- Restore rate limit overrides (v14+; pre-v14 defaults to all zeros)
8613+ let rlo = snap.rateLimitOverrides
8614+ when (rlo.globalRequest > 0) (set *rateOverrideGlobalRequest = rlo.globalRequest)
8615+ when (rlo.ipRate > 0) (set *rateOverrideIpRate = rlo.ipRate)
8616+ when (rlo.sendRate > 0) (set *rateOverrideSendRate = rlo.sendRate)
8617+ when (rlo.federationRate > 0) (set *rateOverrideFederationRate = rlo.federationRate)
8618+ when (rlo.authGlobal > 0) (set *rateOverrideAuthGlobal = rlo.authGlobal)
8619+ when (rlo.recvRate > 0) (set *rateOverrideRecvRate = rlo.recvRate)
8620+ when (rlo.challengeRate > 0) (set *rateOverrideChallengeRate = rlo.challengeRate)
8621+ when (rlo.pairRate > 0) (set *rateOverridePairRate = rlo.pairRate)
8622+ when (rlo.lookupRate > 0) (set *rateOverrideLookupRate = rlo.lookupRate)
8623+ when (rlo.clientWriteRate > 0) (set *rateOverrideClientWriteRate = rlo.clientWriteRate)
8624+ set *drainPhase = 0 -- undrain after restore
8625+ when (count cappedForwards > 0)
8626+ (atomic do set *restoreForwardRampRemaining = restoreForwardRampCycles)
8627+ logInfo ("startup: bootstrap complete (" ++ show (count validPresence) ++ " presence, " ++ show (count validForwards) ++ " forwards)"))
8250 fork backgroundPrune
8628 fork backgroundPrune
8251 fork pollHeartbeat
8629 fork pollHeartbeat
8252 fork backgroundWatchdog
8630 fork backgroundWatchdog