@@ -183,7 +183,7 @@ federationLatencyWindowMs : Int<Ms>
183183federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
184184
185185-- 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
187187
188188-- Server build version: included in /health and /stats responses for deployment verification.
189189-- 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
209209-- captures one valid (snapshot, tag) pair cannot derive the secretHash components
210210-- independently because the final mixing step entangles them non-linearly.
211211-- 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.
212217snapshotIntegrityTag : Int -> Text -> Int
213218snapshotIntegrityTag = \checksum secret ->
214219 if secret == "" then 0
@@ -273,6 +278,14 @@ federationEnabled = True {}
273278-- TODO: If Knot adds a resolved-IP introspection API (e.g., fetchWithResolvedIp),
274279-- validate the resolved IP against isBadServerName before completing the request.
275280federationUrl = \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
276289-- Callback verification: outbound federation requests include a random single-use token;
277290-- the receiver calls back to /federation/verify to confirm the sender actually issued it.
278291maxPendingVerifications = 10000
@@ -329,6 +342,17 @@ maxSenderDeliveries = 10000
329342maxSenderDeliveriesPerKey = 100
330343-- Max pubkeys in a batch lookup request
331344maxBatchLookupKeys = 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
351+
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
332356-- Verify callback caching: cache successful verifications per-server to reduce
333357-- outbound HTTP calls. A peer sending N requests/second forces N verify callbacks;
334358-- caching the success for a short window amortizes the cost.
@@ -352,7 +376,7 @@ healthWatchdogMaxAge = backgroundPruneInterval * 15 -- watchdog runs every ~5 pr
352376-- ============================================
353377
354378-- 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}}
356380
357381-- Gossip events propagate presence changes between federated servers
358382data GossipEventType = GossipOnline {} | GossipOffline {} | GossipUnknown {}
@@ -479,6 +503,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
479503type BatchCancelEntry = {messageId: MessageIdHex, to: PubkeyHex}
480504type BatchCancelResult = {messageId: MessageIdHex, cancelled: Bool}
481505
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}
509+
482510-- Sender expiration notification: tracks when a locally-originated message expired unread.
483511-- Pruned after messageMaxAge. Included in the sender's next poll response.
484512type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}
@@ -857,11 +885,15 @@ maxPeerLatencyEntries = 100
857885-- Populated during ackPolledMessages and handleAckMessages for locally-originated messages.
858886*senderDeliveries : [SenderDelivery]
859887
860−-- Tracks how many sender delivery receipts / expiration notifications were dropped
861−-- due to capacity eviction before the sender could poll them. Reset per-sender on poll
862−-- (drained with the notifications). Lets clients know they may have missed receipts.
863−*droppedDeliveryReceipts : Int
864−*droppedExpirationNotifs : Int
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]
891+
892+-- Per-sender drop counters: tracks how many delivery receipts / expiration notifications
893+-- were evicted before the sender could poll them. Keyed by senderKey so each sender sees
894+-- only their own drops (not a global counter that conflates all senders). Drained on poll.
895+*droppedDeliveryReceipts : [{senderKey: Text, count: Int}]
896+*droppedExpirationNotifs : [{senderKey: Text, count: Int}]
865897
866898-- Runtime maintenance secret: when non-empty, used instead of compile-time maintenanceSecret.
867899-- Set via /maintenance/rotate-secret to enable zero-downtime secret rotation.
@@ -899,6 +931,14 @@ maxPeerLatencyEntries = 100
899931*activeRequests : Int
900932*peakActiveRequests : Int
901933
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>
941+
902942-- Last time the background watchdog loop ran (for health check liveness detection)
903943*lastWatchdogAt : Timestamp
904944
@@ -911,6 +951,12 @@ maxPeerLatencyEntries = 100
911951-- More robust than wall-clock comparison: survives clock jumps without false alarms.
912952*prunesSinceSnapshot : Int
913953
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
959+
914960-- Cached health result: returned when /health is rate-limited instead of false all-healthy.
915961-- Updated on every successful (non-rate-limited) health check.
916962*cachedHealthOk : Bool
@@ -1080,12 +1126,12 @@ route MaintenanceApi where
10801126 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
10811127 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
10821128 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}}
10841130 = Snapshot
10851131 -- Restore: import a previously-exported snapshot. Server must be in drain mode
10861132 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
10871133 -- 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}
10891135 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
10901136 = Restore
10911137 -- Block a federation peer (rejects all inbound/outbound federation with this server)
@@ -1173,7 +1219,7 @@ route HealthApi where
11731219-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
11741220route StatsApi where
11751221 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}}
11771223 = Stats
11781224 GET /stats/peers headers {authorization: Text}
11791225 -> {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
12751321unionNE = \a b -> if count a == 0 then b else if count b == 0 then a else union a b
12761322min = \a b -> if a < b then a else b
12771323max = \a b -> if a > b then a else b
1324+
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}]
1333+
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}
12781338headOf = \items -> fold (\acc x -> case acc of
12791339 Nothing {} -> Just {value: x}
12801340 _ -> acc) Nothing {} items
@@ -2235,14 +2295,30 @@ pruneMessageBucket = \b t -> atomic do
22352295 let localExpired = do m <- removed; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, expiredAt: t}
22362296 when (count localExpired > 0) (do
22372297 se <- *senderExpirations
2238− let combined = union se localExpired
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
2304+ let combined = union se perKeyCapped
22392305 let overCap = count combined > maxSenderExpirations
2240− set *senderExpirations = if overCap
2241− then evictOneOldest combined (\e -> e.expiredAt)
2242− else combined
2243− when overCap (do
2306+ if overCap then do
2307+ let evicted = extractOldest combined (\e -> e.expiredAt)
2308+ set *senderExpirations = evicted.items
22442309 den <- *droppedExpirationNotifs
2245− set *droppedExpirationNotifs = den + 1))
2310+ let denAfterCap = case evicted.candidate of
2311+ Nothing {} -> den
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
2315+ set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1) denAfterCap perKeyDroppedEntries
2316+ else do
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))
22462322 else yield {}
22472323
22482324-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
@@ -3448,6 +3524,9 @@ pruneExpired = \t -> do
34483524 -- Prune stale deferred ack cleanups (older than messageMaxAge are never useful)
34493525 pendingCleanups <- *pendingAckCleanups
34503526 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
34513530 -- Prune old sender expiration notifications. Uses unackedMessageMaxAge (7 days)
34523531 -- instead of sessionExpiry (1 hour) so mobile clients that don't poll for days
34533532 -- still receive expiration notifications on their next poll.
@@ -3969,9 +4048,11 @@ retryForwardsForServer = \entries remaining ->
39694048 -- Process per-message results: remove succeeded, increment retries for failed
39704049 retryForwardsBatchResults batch response.results server
39714050 else do
3972− logWarn ("federation: batch forward to " ++ server ++ " rejected: " ++ response.error)
3973− -- Application-level batch rejection (old peer, invalid request): increment retries
3974− retryForwardsBatchIncrement batch server)
4051+ logWarn ("federation: batch forward to " ++ server ++ " rejected: " ++ response.error ++ ", falling back to single-message forwards")
4052+ -- Application-level batch rejection (old peer, invalid request): fall back to
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)
39754056
39764057-- Increment retry counts for all entries in a failed batch
39774058retryForwardsBatchIncrement = \batch server ->
@@ -3982,6 +4063,34 @@ retryForwardsBatchIncrement = \batch server ->
39824063 incrementForwardRetries f.messageId server f.toKey
39834064 retryForwardsBatchIncrement rest server
39844065
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
4074+ t <- now
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})
4083+ case result of
4084+ Err e -> do
4085+ logWarn ("federation: single forward to " ++ server ++ " failed for " ++ f.messageId ++ ": " ++ show e)
4086+ recordFailure server
4087+ Ok response ->
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
4093+
39854094-- Process per-message results from a batch forward response
39864095retryForwardsBatchResults = \batch results server ->
39874096 case headOf batch of
@@ -4043,15 +4152,28 @@ retryPendingForwardsFromBuckets = do
40434152 activePolls <- *activePolls
40444153 let onlineForwards = do f <- eligibleForwards; where any activePolls (\p -> p.pubkey == f.toKey); yield f
40454154 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
4047− let sortedOnline = sortBy (\f -> 0 - f.priority) onlineForwards
4048− let sortedOffline = sortBy (\f -> 0 - f.priority) offlineForwards
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).
4159+ let sortedOnline = sortBy (\f -> 0 - f.priority + f.retries) onlineForwards
4160+ let sortedOffline = sortBy (\f -> 0 - f.priority + f.retries) offlineForwards
40494161 let prioritizedForwards = union sortedOnline sortedOffline
40504162 -- Prioritize servers that have online-key forwards first
40514163 let onlineServers = collectPeerServers onlineForwards (\x -> x.toServer)
40524164 let offlineOnlyServers = do s <- eligibleServers; where none onlineServers (\os -> os == s); yield s
40534165 let prioritizedServers = union onlineServers offlineOnlyServers
4054− sendForwardsPartitioned prioritizedForwards prioritizedServers maxForwardRetriesPerServer
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)
4176+ sendForwardsPartitioned prioritizedForwards prioritizedServers effectiveMaxRetries
40554177
40564178-- When a remote key comes online, create forward entries for queued messages and send them
40574179forwardQueuedMessages = \targets t -> do
@@ -4261,6 +4383,82 @@ retryOverflowAcks = \overflow -> do
42614383 ackOverflow <- *ackOverflow
42624384 set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))
42634385
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)
4395+ else updated)
4396+
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
4405+
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
4410+ t <- now
4411+ pending <- *pendingFederationCancels
4412+ if count pending == 0 then yield {}
4413+ else do
4414+ failures <- *forwardFailures
4415+ if isFederationCircuitOpen failures t
4416+ then logWarn "federation: circuit breaker open, skipping cancel retries"
4417+ else do
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
4425+ cancelT <- now
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})
4433+ case result of
4434+ Err e -> do
4435+ logWarn ("federation: cancel retry to " ++ server ++ " failed: " ++ show e)
4436+ recordFailure server
4437+ incrementCancelRetries batch server
4438+ Ok r ->
4439+ if r.ok then do
4440+ clearFailure server
4441+ removePendingCancels batch server
4442+ else do
4443+ logWarn ("federation: cancel retry to " ++ server ++ " rejected: " ++ r.error)
4444+ incrementCancelRetries batch server))
4445+
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)
4456+
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
4461+
42644462-- Wait for all in-flight federation requests to complete or timeout.
42654463-- Uses STM retry instead of busy-wait: blocks until *federationInFlight changes
42664464-- (when clearFederationInFlight writes) or *drainTimedOut is set by the timeout fork.
@@ -4418,6 +4616,13 @@ pruneAndRetry = \t -> do
44184616 set *verificationCache = do c <- cache; where t - c.verifiedAt < verificationCacheMs; yield c
44194617 -- Reset per-peer latency tracking with global window
44204618 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
44214626 set *federationLatencyResetAt = t)
44224627 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
44234628 -- These were queued by handleFederationAck instead of fork+sleep.
@@ -4435,7 +4640,8 @@ pruneAndRetry = \t -> do
44354640 when federationEnabled (do
44364641 retryPendingForwardsFromBuckets
44374642 sendAcksAndClean result.pendingAcks
4438− retryOverflowAcks result.pendingOverflow))
4643+ retryOverflowAcks result.pendingOverflow
4644+ retryPendingCancels))
44394645 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
44404646 -- trigger offline gossip when they expire (avoids wasted federation traffic)
44414647 when result.shouldRefresh (do
@@ -4535,14 +4741,19 @@ ackPolledMessages = \pubkey cursor t -> do
45354741 let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
45364742 when (count localDelivered > 0) (do
45374743 sd <- *senderDeliveries
4538− let combined = union sd localDelivered
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
4746+ let combined = union sd deduped
45394747 let overCap = count combined > maxSenderDeliveries
4540− set *senderDeliveries = if overCap
4541− then evictOneOldest combined (\d -> d.deliveredAt)
4542− else combined
4543− when overCap (do
4544− ddr <- *droppedDeliveryReceipts
4545− set *droppedDeliveryReceipts = ddr + 1))
4748+ if overCap then do
4749+ let evicted = extractOldest combined (\d -> d.deliveredAt)
4750+ set *senderDeliveries = evicted.items
4751+ case evicted.candidate of
4752+ Nothing {} -> yield {}
4753+ Just {value: ev} -> do
4754+ ddr <- *droppedDeliveryReceipts
4755+ set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1
4756+ else set *senderDeliveries = combined)
45464757 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
45474758 else yield {pendingAcks: [], evictedEntries: []}
45484759
@@ -4571,7 +4782,8 @@ pollNewEvents = \pubkey clampedCursor t -> do
45714782 yield {id: m.messageId, eventType: "message",
45724783 payload: {toKey: m.toKey,
45734784 encryptedBlob: m.encryptedBlob,
4574− timestamp: m.timestamp}}
4785+ timestamp: m.timestamp,
4786+ expiresAt: m.expiresAt}}
45754787 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
45764788 -- hasMore: tells clients there are additional messages beyond this batch,
45774789 -- so they can immediately re-poll without waiting for the next heartbeat.
@@ -4595,8 +4807,13 @@ pollOnce = \token cursor clientIp -> do
45954807 if not (matchIp clientIp ip)
45964808 then yield (Nothing {})
45974809 else do
4598− -- Refresh session expiry (sliding window keeps active sessions alive)
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))
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)
4815+ when needsRefresh
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)))
46004817 seqCounter <- *seqCounter
46014818 let clampedCursor = clamp 0 (max seqCounter t) cursor
46024819 ackResult <- if clampedCursor > 0
@@ -4733,7 +4950,7 @@ forwardCandidates = \targets messages forwards t ->
47334950-- Case 4 prevents an attacker who intercepts the challenge string from verifying
47344951-- without going through the same reverse proxy that issued the challenge.
47354952challengeMatch = \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)
47374954
47384955challengesExcluding = \challenges pubkey challenge ->
47394956 do c <- challenges; where c.pubkey != pubkey || c.challenge != challenge; yield c
@@ -5059,17 +5276,17 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
50595276 yield capped
50605277 -- Poll slot exhaustion: signal clients to back off for one poll cycle
50615278 let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
5062− -- Read and reset drop counters so clients know if they missed receipts/expirations.
5063− -- Counters are global (not per-key) — a non-zero value means *some* receipts were
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}
5279+ -- Read and drain per-sender drop counters so this sender knows if their receipts were lost.
5280+ -- Each sender sees only their own drops (not a global counter conflating all senders).
5281+ drops <- if not result.authorized || result.pubkey == "" then yield {dr: 0, de: 0}
50675282 else atomic do
5068− dr <- *droppedDeliveryReceipts
5069− de <- *droppedExpirationNotifs
5070− when (dr > 0) (set *droppedDeliveryReceipts = 0)
5071− when (de > 0) (set *droppedExpirationNotifs = 0)
5072− yield {dr: dr, de: de}
5283+ ddr <- *droppedDeliveryReceipts
5284+ let drResult = drainDropCounter ddr result.pubkey
5285+ when (drResult.count > 0) (set *droppedDeliveryReceipts = drResult.remaining)
5286+ den <- *droppedExpirationNotifs
5287+ let deResult = drainDropCounter den result.pubkey
5288+ when (deResult.count > 0) (set *droppedExpirationNotifs = deResult.remaining)
5289+ yield {dr: drResult.count, de: deResult.count}
50735290 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})
50745291
50755292-- Core send logic: dedup, rate limiting, store message, create forwards.
@@ -5194,17 +5411,18 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
51945411
51955412-- Send all batch forwards for one server sequentially (no per-message fork overhead).
51965413-- 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).
51975417sendBatchForwardsSequential = \msgs server ->
51985418 case headOf msgs of
51995419 Nothing {} -> yield {}
52005420 Just {value: r} -> do
52015421 let rest = do m <- msgs; where m.messageId != r.messageId; yield m
5202− batchFwdInfo <- atomic do
5422+ stillPending <- atomic do
52035423 fwdBkt <- readFwdBucket (pubkeyBucket r.toKey)
5204− case headOf (do f <- fwdBkt; where f.messageId == r.messageId && f.toServer == server; yield f) of
5205− Nothing {} -> yield {pending: False {}, fwdExpiresAt: 0, fwdPriority: 0}
5206− Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt, fwdPriority: fe.priority}
5207− when batchFwdInfo.pending (do
5424+ yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))
5425+ when stillPending (do
52085426 fwdT <- now
52095427 fwdAuth <- generateFederationAuth "forward" server
52105428 fwdCorrId <- genTokenN 1
@@ -5213,7 +5431,7 @@ sendBatchForwardsSequential = \msgs server ->
52135431 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
52145432 fromServer: serverHostN, requestTimestamp: fwdT,
52155433 authorization: fwdAuth, correlationId: fwdCorrId,
5216− expiresAt: batchFwdInfo.fwdExpiresAt, priority: batchFwdInfo.fwdPriority, protocolVersion: federationProtocolVersion})
5434+ expiresAt: r.fwdExpiresAt, priority: r.fwdPriority, protocolVersion: federationProtocolVersion})
52175435 case fwdResult of
52185436 Err e -> do
52195437 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
@@ -5765,14 +5983,19 @@ fedDeliveryReceipts = \bucketIds forwards t -> case headOf bucketIds of
57655983 let deliveries = union fromForwards msgDeliveries
57665984 when (count deliveries > 0) (atomic do
57675985 sd <- *senderDeliveries
5768− let combined = union sd deliveries
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
5988+ let combined = union sd deduped
57695989 let overCap = count combined > maxSenderDeliveries
5770− set *senderDeliveries = if overCap
5771− then evictOneOldest combined (\d -> d.deliveredAt)
5772− else combined
5773− when overCap (do
5774− ddr <- *droppedDeliveryReceipts
5775− set *droppedDeliveryReceipts = ddr + 1))
5990+ if overCap then do
5991+ let evicted = extractOldest combined (\d -> d.deliveredAt)
5992+ set *senderDeliveries = evicted.items
5993+ case evicted.candidate of
5994+ Nothing {} -> yield {}
5995+ Just {value: ev} -> do
5996+ ddr <- *droppedDeliveryReceipts
5997+ set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1
5998+ else set *senderDeliveries = combined)
57765999 fedDeliveryReceipts rest forwards t
57776000
57786001-- POST /federation/ack: Peer server confirms delivery of messages we originated.
@@ -6127,7 +6350,8 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
61276350 let capped = takeRelation maxBatchSize cancels
61286351 -- Process each cancel, collecting results and per-server cancel targets
61296352 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.
61316355 fork (when (federationEnabled && count results.cancelServers > 0)
61326356 (sendToEach results.cancelServers (\server -> do
61336357 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
61356359 cancelT <- now
61366360 cancelAuth <- generateFederationAuth "cancel" server
61376361 cancelCorrId <- genTokenN 1
6138− fetchFederation server (FederationBatchCancel {
6362+ cancelResult <- fetchFederation server (FederationBatchCancel {
61396363 cancels: serverCancels, fromServer: serverHostN,
61406364 requestTimestamp: cancelT, correlationId: cancelCorrId,
61416365 authorization: cancelAuth, protocolVersion: federationProtocolVersion})
6142− yield {}))))
6366+ case cancelResult of
6367+ Err e -> do
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)))))
61436373 yield (respond {results: results.results, authorized: True {}, error: ""}))
61446374
61456375-- Process each client cancel entry, returning results and federation cancel targets.
@@ -6245,8 +6475,13 @@ handleDrain = \phase authorization respond -> do
62456475 pac <- *pendingAckCleanups
62466476 se <- *senderExpirations
62476477 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}
62486483 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}
62506485 -- Sentinel-wrapped: sidecar processes should verify [SNAPSHOT_END] follows to detect truncation.
62516486 -- Integrity tag appended when maintenance secret is configured (enables tamper detection on restore).
62526487 runtimeSec <- *runtimeMaintenanceSecret
@@ -6490,7 +6725,7 @@ handleHealth = \authorization respond -> do
64906725 draining: result.isDraining,
64916726 degradedPeers: 0})
64926727
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}}
64946729
64956730handleStats = \authorization respond -> do
64966731 t <- now
@@ -6560,7 +6795,13 @@ handleStats = \authorization respond -> do
65606795 rlf <- *rateLimitedFederation
65616796 arq <- *activeRequests
65626797 parq <- *peakActiveRequests
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}
6798+ cplt <- *clientPollLatencyTotal
6799+ cplc <- *clientPollLatencyCount
6800+ cplm <- *clientPollLatencyMax
6801+ cslt <- *clientSendLatencyTotal
6802+ cslc <- *clientSendLatencyCount
6803+ cslm <- *clientSendLatencyMax
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}
65646805 -- Compute derived metrics
65656806 let sessions = core.sc
65666807 let messages = core.tmc
@@ -6582,11 +6823,13 @@ handleStats = \authorization respond -> do
65826823 -- Top-3 failing peers by failure count (inline in /stats to reduce monitoring queries)
65836824 let topFailing = takeRelation 3 (sortBy (\f -> 0 - f.failures) core.ff)
65846825 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
65856828 yield (respond {ok: True {}, version: serverBuildVersion,
65866829 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},
65876830 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},
65886831 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}})
65906833 else yield (respond zeroStats)
65916834
65926835handleStatsPeers = \authorization respond -> do
@@ -6675,7 +6918,9 @@ handleReady = \respond -> do
66756918 let isReady = lastPrune > 0 && lastPrune >= sa
66766919 yield (respond {ready: isReady})
66776920
6678−emptySnapshot = {ok: False {}, version: 0, checksum: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: [], allowedPeers: [], pendingAckCleanups: [], senderExpirations: [], senderDeliveries: []}
6921+zeroRateLimitOverrides = {globalRequest: 0, ipRate: 0, sendRate: 0, federationRate: 0, authGlobal: 0, recvRate: 0, challengeRate: 0, pairRate: 0, lookupRate: 0, clientWriteRate: 0}
6922+
6923+emptySnapshot = {ok: False {}, version: 0, checksum: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: [], allowedPeers: [], pendingAckCleanups: [], senderExpirations: [], senderDeliveries: [], rateLimitOverrides: zeroRateLimitOverrides}
66796924
66806925-- GET /maintenance/snapshot: Export federation metadata for persistence.
66816926-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
@@ -6699,20 +6944,26 @@ handleSnapshot = \authorization respond -> do
66996944 pac <- *pendingAckCleanups
67006945 se <- *senderExpirations
67016946 sdl <- *senderDeliveries
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}
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}
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}
67036954 atomic do
67046955 set *lastSnapshotExportAt = t
67056956 set *prunesSinceSnapshot = 0
67066957 let cs = snapshotChecksum (count result.presence) (count result.ackedIds) (count result.deliveredIds) (count result.forwards) (count result.forwardNonces)
67076958 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})
67096960
67106961-- POST /maintenance/restore: Import a previously-exported snapshot.
67116962-- Requires drain mode to prevent conflicts with live state. Merges imported data
67126963-- with existing state (dedup entries are unioned, presence is merged by freshness).
67136964-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
67146965-- 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
67166967 authorized <- verifyMaintenanceAuth authorization
67176968 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
67186969 else do
@@ -6819,7 +7070,24 @@ handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData de
68197070 existingSenderDel <- *senderDeliveries
68207071 let mergedSenderDel = union existingSenderDel senderDeliveriesData
68217072 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)
68227086 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)
68237091 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
68247092 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
68257093
@@ -6875,14 +7143,19 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
68757143 let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
68767144 when (count localDelivered > 0) (do
68777145 sd <- *senderDeliveries
6878− let combined = union sd localDelivered
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
7148+ let combined = union sd deduped
68797149 let overCap = count combined > maxSenderDeliveries
6880− set *senderDeliveries = if overCap
6881− then evictOneOldest combined (\d -> d.deliveredAt)
6882− else combined
6883− when overCap (do
6884− ddr <- *droppedDeliveryReceipts
6885− set *droppedDeliveryReceipts = ddr + 1))
7150+ if overCap then do
7151+ let evicted = extractOldest combined (\d -> d.deliveredAt)
7152+ set *senderDeliveries = evicted.items
7153+ case evicted.candidate of
7154+ Nothing {} -> yield {}
7155+ Just {value: ev} -> do
7156+ ddr <- *droppedDeliveryReceipts
7157+ set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1
7158+ else set *senderDeliveries = combined)
68867159 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
68877160 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}
68887161 fork (when federationEnabled (do
@@ -6974,14 +7247,19 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
69747247 let localDeleted = do m <- toDelete; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
69757248 when (count localDeleted > 0) (do
69767249 sd <- *senderDeliveries
6977− let combinedDel = union sd localDeleted
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
7252+ let combinedDel = union sd dedupedDel
69787253 let overCapDel = count combinedDel > maxSenderDeliveries
6979− set *senderDeliveries = if overCapDel
6980− then evictOneOldest combinedDel (\d -> d.deliveredAt)
6981− else combinedDel
6982− when overCapDel (do
6983− ddr <- *droppedDeliveryReceipts
6984− set *droppedDeliveryReceipts = ddr + 1))
7254+ if overCapDel then do
7255+ let evicted = extractOldest combinedDel (\d -> d.deliveredAt)
7256+ set *senderDeliveries = evicted.items
7257+ case evicted.candidate of
7258+ Nothing {} -> yield {}
7259+ Just {value: ev} -> do
7260+ ddr <- *droppedDeliveryReceipts
7261+ set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1
7262+ else set *senderDeliveries = combinedDel)
69857263 if count ackedFederated > 0 then do
69867264 ackedBkt <- readAckedBucket b
69877265 let combinedAcked = union ackedBkt ackedFederated
@@ -7068,19 +7346,26 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
70687346 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
70697347 else updatedNonces)
70707348 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.
70727353 fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)
70737354 (sendToEach cancelResult.cancelServers (\server -> do
70747355 cancelT <- now
70757356 cancelAuth <- generateFederationAuth "cancel" server
70767357 cancelCorrId <- genTokenN 1
7077− cancelFedResult <- fetchFederation server (FederationCancel {
7078− messageId: messageId, toKey: to, fromServer: serverHostN,
7358+ cancelFedResult <- fetchFederation server (FederationBatchCancel {
7359+ cancels: [{messageId: messageId, toKey: to}], fromServer: serverHostN,
70797360 requestTimestamp: cancelT, correlationId: cancelCorrId,
70807361 authorization: cancelAuth, protocolVersion: federationProtocolVersion})
70817362 case cancelFedResult of
7082− Err e -> logWarn ("federation: cancel to " ++ server ++ " failed for " ++ messageId ++ ": " ++ show e)
7083− Ok r -> when (not r.ok) (logWarn ("federation: cancel to " ++ server ++ " rejected for " ++ messageId ++ ": " ++ r.error)))))
7363+ Err e -> do
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
7367+ logWarn ("federation: cancel to " ++ server ++ " rejected for " ++ messageId ++ ": " ++ r.error)
7368+ queuePendingCancel messageId to server cancelT))))
70847369 yield (respond {ok: True {}, authorized: True {}, cancelled: cancelResult.cancelled, error: ""}))
70857370
70867371-- POST /messages/batch: Send multiple messages in one request.
@@ -7169,14 +7454,15 @@ sendBatchBucketGroup = \senderKey prepared t dp acc ->
71697454 Nothing {} -> yield acc
71707455 Just {value: entry} -> do
71717456 let rest = do p <- prepared; where p != entry; yield p
7457+ let clampedPri = clamp 0 maxMessagePriority entry.msg.priority
71727458 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}
71747460 else do
71757461 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}
71777463 else do
7178− sendResult <- sendMessageCore senderKey entry.msg.to entry.msg.encryptedBlob clampedTs entry.effectiveMsgId entry.msg.expiresInMs (clamp 0 maxMessagePriority entry.msg.priority) 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}
7464+ sendResult <- sendMessageCore senderKey entry.msg.to entry.msg.encryptedBlob clampedTs entry.effectiveMsgId entry.msg.expiresInMs clampedPri t
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}
71807466 sendBatchBucketGroup senderKey rest t dp (union acc [result])
71817467
71827468-- Log warnings for batch results outside the atomic block
@@ -7713,7 +7999,7 @@ serveInner = \req -> case req of
77137999 Shutdown {authorization, respond} -> handleShutdown authorization respond
77148000 Undrain {authorization, respond} -> handleUndrain authorization respond
77158001 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
77178003 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
77188004 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
77198005 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
@@ -7830,6 +8116,31 @@ serve = \req -> do
78308116 set *activeRequests = max 0 (ar - 1)
78318117 endTime <- now
78328118 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.
8121+ case req of
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
8143+ _ -> yield {}
78338144 let path = requestPath req
78348145 let ip = requestClientIp req
78358146 let ipSuffix = if ip != "" then " " ++ ip else ""
@@ -8036,7 +8347,12 @@ backgroundSnapshot = do
80368347 pac <- *pendingAckCleanups
80378348 se <- *senderExpirations
80388349 sdl <- *senderDeliveries
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}
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}
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}
80408356 let p = exportSnapshot.p
80418357 let a = exportSnapshot.a
80428358 let ao = exportSnapshot.ao
@@ -8045,7 +8361,7 @@ backgroundSnapshot = do
80458361 let f = exportSnapshot.f
80468362 let fn = exportSnapshot.fn
80478363 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}
80498365 autoRtSec <- *runtimeMaintenanceSecret
80508366 let autoEffSec = if autoRtSec != "" then autoRtSec else maintenanceSecret
80518367 let autoTag = snapshotIntegrityTag exportCs autoEffSec
@@ -8247,6 +8563,68 @@ main = do
82478563 when (autoSnapshotExportInterval > 0) (do
82488564 logInfo (" auto-snapshot export: " ++ show autoSnapshotExportInterval ++ "ms (full [SNAPSHOT_DATA] to stdout)")
82498565 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)")
8574+ Ok snap ->
8575+ if not snap.ok then logWarn ("startup: bootstrap from " ++ bootstrapPeerUrl ++ " returned ok=false (peer may require auth), continuing without snapshot")
8576+ else do
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")
8582+ else do
8583+ atomic do
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
8590+ bt <- now
8591+ atomic do
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)"))
82508628 fork backgroundPrune
82518629 fork pollHeartbeat
82528630 fork backgroundWatchdog