@@ -110,6 +110,10 @@ maxServerDowntime = 86400000 -- 24 hours; stop retrying a server after this dura
110federationFetchTimeoutMs : Int<Ms>
110federationFetchTimeoutMs : Int<Ms>
111federationFetchTimeoutMs = 30000 -- 30 seconds; consider server unresponsive if fetch hasn't returned
111federationFetchTimeoutMs = 30000 -- 30 seconds; consider server unresponsive if fetch hasn't returned
112maxFederationInFlight = 100
112maxFederationInFlight = 100
113+-- Federation circuit breaker: pause all outbound federation when this many unique peer
114+-- servers have active (non-expired) failure records. Prevents wasted retries and log spam
115+-- when the local network is down. Set to 0 to disable.
116+federationCircuitBreakerThreshold = 3
114-- Federation latency metric window
118-- Federation latency metric window
115federationLatencyWindowMs : Int<Ms>
119federationLatencyWindowMs : Int<Ms>
@@ -416,6 +420,9 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
416-- Max concurrent background prune forks (prevents resource exhaustion from stuck recovery)
420-- Max concurrent background prune forks (prevents resource exhaustion from stuck recovery)
417*activePruneForks : Int
421*activePruneForks : Int
423+-- Signal for drain timeout (set by forked timer, watched by STM retry in waitForInFlightDrain)
424+*drainTimedOut : Bool
419-- ============================================
426-- ============================================
421-- ============================================
428-- ============================================
@@ -464,15 +471,15 @@ route ClientApi where
464route FederationApi where
471route FederationApi where
465 /federation
472 /federation
466 -- Receive presence gossip events from a peer server
473 -- Receive presence gossip events from a peer server
467− POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName} /gossip headers {authorization: Text}
474+ POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp} /gossip headers {authorization: Text}
468 -> {ok: Bool, error: Text}
475 -> {ok: Bool, error: Text}
469 = RecvGossip
476 = RecvGossip
470 -- Receive a forwarded message from a peer server on behalf of a sender
477 -- Receive a forwarded message from a peer server on behalf of a sender
471− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName} /forward headers {authorization: Text}
478+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp} /forward headers {authorization: Text}
472 -> {ok: Bool, error: Text}
479 -> {ok: Bool, error: Text}
473 = ForwardMessage
480 = ForwardMessage
474 -- Acknowledge delivery of messages back to the origin server
481 -- Acknowledge delivery of messages back to the origin server
475− POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName} /ack headers {authorization: Text}
482+ POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp} /ack headers {authorization: Text}
476 -> {ok: Bool, error: Text}
483 -> {ok: Bool, error: Text}
477 = FederationAck
484 = FederationAck
@@ -487,6 +494,17 @@ route MaintenanceApi where
487 POST {} /maintenance/undrain headers {authorization: Text}
494 POST {} /maintenance/undrain headers {authorization: Text}
488 -> {ok: Bool}
495 -> {ok: Bool}
489 = Undrain
496 = Undrain
497+ -- Snapshot: export federation metadata for persistence across restarts.
498+ -- Messages are excluded (ephemeral by design; clients handle re-delivery).
499+ -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
500+ GET /maintenance/snapshot headers {authorization: Text}
501+ -> {ok: Bool, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure]}
503+ -- Restore: import a previously-exported snapshot. Server must be in drain mode
504+ -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
505+ POST {presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure]} /maintenance/restore headers {authorization: Text}
506+ -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
491route HealthApi where
509route HealthApi where
492 GET /health -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool} = Health
510 GET /health -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool} = Health
@@ -875,6 +893,17 @@ countAllSessions = do
875 s15 <- *sessionBucket15
893 s15 <- *sessionBucket15
876 yield (count s0 + count s1 + count s2 + count s3 + count s4 + count s5 + count s6 + count s7 + count s8 + count s9 + count s10 + count s11 + count s12 + count s13 + count s14 + count s15)
894 yield (count s0 + count s1 + count s2 + count s3 + count s4 + count s5 + count s6 + count s7 + count s8 + count s9 + count s10 + count s11 + count s12 + count s13 + count s14 + count s15)
896+-- Short-circuit scan: check if any session bucket contains a session for the given pubkey.
897+-- Stops reading buckets as soon as one matching session is found (O(n/16) best case
898+-- vs O(n) for readAllSessions when the user has sessions in early buckets).
899+hasSessionForKey : PubkeyHex -> Int -> IO Bool
900+hasSessionForKey = \pubkey b ->
901+ if b > 15 then yield (False {})
903+ bucket <- readSessionBucket b
904+ if any bucket (\s -> s.pubkey == pubkey) then yield (True {})
905+ else hasSessionForKey pubkey (b + 1)
878-- Apply a filter/transform to every session bucket
907-- Apply a filter/transform to every session bucket
879transformAllSessionBuckets : ([Session] -> [Session]) -> IO ()
908transformAllSessionBuckets : ([Session] -> [Session]) -> IO ()
880transformAllSessionBuckets = \f -> do
909transformAllSessionBuckets = \f -> do
@@ -1394,9 +1423,13 @@ pruneExpired = \t -> do
1394 set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
1423 set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
1395 -- Clean up stale in-flight entries: remove zeroed entries and reset leaked slots
1424 -- Clean up stale in-flight entries: remove zeroed entries and reset leaked slots
1396 -- where the forked task died without calling clearFederationInFlight.
1425 -- where the forked task died without calling clearFederationInFlight.
1397− -- Uses 3x timeout threshold to allow the skip-hung-server logic to take effect first.
1426+ -- Uses 2x timeout threshold (tighter than the skip-hung-server logic at 1x)
1427+ -- so leaked slots are reclaimed promptly after the skip logic has already taken effect.
1398 inFlight <- *federationInFlight
1428 inFlight <- *federationInFlight
1429+ let staleInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs * 2; yield r
1430+ when (count staleInFlight > 0)
1431+ (println ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))
1399− set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < federationFetchTimeoutMs * 3); yield r
1432+ set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < federationFetchTimeoutMs * 2); yield r
1400 yield presenceResult
1433 yield presenceResult
@@ -1416,9 +1449,11 @@ forEachBatch = \items batchSize process ->
1416sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {}
1449sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {}
1417sendBatchedGossipToServer = \server allEvents hops ->
1450sendBatchedGossipToServer = \server allEvents hops ->
1418 forEachBatch allEvents maxGossipEvents (\batch -> do
1451 forEachBatch allEvents maxGossipEvents (\batch -> do
1419 result <- fetchFederation server (RecvGossip {
1453 result <- fetchFederation server (RecvGossip {
1420 authorization: federationAuthForServer server,
1454 authorization: federationAuthForServer server,
1421− events: batch, hopsRemaining: hops, fromServer: serverHostN})
1455+ events: batch, hopsRemaining: hops, fromServer: serverHostN,
1456+ requestTimestamp: t})
1422 case result of
1457 case result of
1423 Err e -> println ("federation: gossip to " ++ server ++ " failed: " ++ show e)
1458 Err e -> println ("federation: gossip to " ++ server ++ " failed: " ++ show e)
1424 Ok _ -> yield {})
1459 Ok _ -> yield {})
@@ -1429,6 +1464,10 @@ sendBatchedGossip = \peerServers allEvents hops ->
1430broadcastOfflineGossip = \t peerServers expiredLocal ->
1465broadcastOfflineGossip = \t peerServers expiredLocal ->
1431 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do
1466 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do
1467+ failures <- *forwardFailures
1468+ if isFederationCircuitOpen failures t
1469+ then println ("federation: circuit breaker open, skipping offline gossip broadcast")
1432 let cappedExpired = takeRelation maxPresenceEntries expiredLocal
1471 let cappedExpired = takeRelation maxPresenceEntries expiredLocal
1433 when (count expiredLocal > maxPresenceEntries)
1472 when (count expiredLocal > maxPresenceEntries)
1434 (println ("gossip: capped offline broadcast to " ++ show maxPresenceEntries ++ " entries (had " ++ show (count expiredLocal) ++ ")"))
1473 (println ("gossip: capped offline broadcast to " ++ show maxPresenceEntries ++ " entries (had " ++ show (count expiredLocal) ++ ")"))
@@ -1516,9 +1555,15 @@ checkPreAuth = \clientIp t -> do
1516 if ipLimited then yield "ip_rate_limited"
1555 if ipLimited then yield "ip_rate_limited"
1517 else yield ""
1556 else yield ""
1519−withFederationAuth = \authorization t respond fromServer precondition action ->
1558+withFederationAuth = \authorization t respond fromServer requestTimestamp precondition action ->
1520 if not (verifyFederationAuthForServer fromServer authorization)
1559 if not (verifyFederationAuthForServer fromServer authorization)
1521 then yield (respond {ok: False {}, error: "unauthorized"})
1560 then yield (respond {ok: False {}, error: "unauthorized"})
1561+ -- Replay protection: reject requests with timestamps too far from local clock.
1562+ -- requestTimestamp == 0 means the peer doesn't support this field yet (accept).
1563+ else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance)
1565+ println ("federation: clock skew " ++ show (t - requestTimestamp) ++ "ms from " ++ fromServer ++ ", rejecting (tolerance: " ++ show clockSkewTolerance ++ "ms)")
1566+ yield (respond {ok: False {}, error: "clock_skew"})
1522 else if not precondition
1567 else if not precondition
1523 then yield (respond {ok: False {}, error: "invalid_request"})
1568 then yield (respond {ok: False {}, error: "invalid_request"})
@@ -1558,6 +1603,31 @@ removeForward = \mid srv -> atomic do
1558 forwards <- *forwards
1603 forwards <- *forwards
1559 set *forwards = forwardsExcluding forwards srv [mid]
1604 set *forwards = forwardsExcluding forwards srv [mid]
1606+-- Reclaim origin messages after federation ack confirms delivery.
1607+-- Iterates removedForwards with one atomic block per message (short locks).
1608+ackCleanupForwards = \entries t -> case headOf entries of
1609+ Nothing {} -> yield {}
1610+ Just {value: rf} -> do
1611+ let rest = do x <- entries; where x.messageId != rf.messageId || x.toKey != rf.toKey; yield x
1613+ presence <- *presence
1614+ forwards <- *forwards
1615+ let hasLocal = any presence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t)
1616+ let hasOtherForwards = any forwards (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey)
1617+ when (not hasLocal && not hasOtherForwards) (do
1618+ let b = pubkeyBucket rf.toKey
1619+ bucket <- readMessageBucket b
1620+ let msg = headOf (do m <- bucket; where m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod; yield m)
1622+ Nothing {} -> yield {}
1624+ writeMessageBucket b (do m <- bucket; where not (m.messageId == rf.messageId && m.toKey == rf.toKey); yield m)
1625+ tc <- *totalMessageCount
1626+ set *totalMessageCount = max 0 (tc - 1)
1627+ rc <- *recipientMsgCounts
1628+ set *recipientMsgCounts = adjustRecipientCount rc rf.toKey (0 - 1))
1629+ ackCleanupForwards rest t
1561pow2 : Int -> Int
1631pow2 : Int -> Int
1562pow2 = \n -> if n <= 0 then 1 else 2 * pow2 (n - 1)
1632pow2 = \n -> if n <= 0 then 1 else 2 * pow2 (n - 1)
@@ -1577,6 +1647,14 @@ shouldRetryServerEx : Text -> Timestamp -> [ForwardFailure] -> [FederationInFlig
1577shouldRetryServerEx = \server t failures inFlight ->
1647shouldRetryServerEx = \server t failures inFlight ->
1578 shouldRetryServer server t failures && not (hasHungFederationRequest server t inFlight)
1648 shouldRetryServer server t failures && not (hasHungFederationRequest server t inFlight)
1650+-- Global circuit breaker: returns True when too many peers are failing simultaneously,
1651+-- indicating a likely local network issue rather than individual peer problems.
1652+-- When tripped, all outbound federation is paused until the next prune cycle.
1653+isFederationCircuitOpen : [ForwardFailure] -> Timestamp -> Bool
1654+isFederationCircuitOpen = \failures t ->
1655+ federationCircuitBreakerThreshold > 0
1656+ && count (do f <- failures; where t - f.firstFailedAt < maxServerDowntime; yield f) >= federationCircuitBreakerThreshold
1580recordFailure = \server -> do
1658recordFailure = \server -> do
1581 t <- now
1659 t <- now
1582 atomic do
1660 atomic do
@@ -1602,10 +1680,12 @@ retryForwardsForServer = \entries remaining ->
1602 Nothing {} -> yield {}
1680 Nothing {} -> yield {}
1603 Just {value: f} -> do
1681 Just {value: f} -> do
1604 let rest = do x <- entries; where x.messageId != f.messageId; yield x
1682 let rest = do x <- entries; where x.messageId != f.messageId; yield x
1605 result <- fetchFederation f.toServer (ForwardMessage {
1684 result <- fetchFederation f.toServer (ForwardMessage {
1606 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
1685 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
1607 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
1686 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
1687+ fromServer: serverHostN, requestTimestamp: t,
1608− fromServer: serverHostN, authorization: federationAuthForServer f.toServer})
1688+ authorization: federationAuthForServer f.toServer})
1609 case result of
1689 case result of
1610 Err e -> do
1690 Err e -> do
1611 println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
1691 println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
@@ -1620,11 +1700,15 @@ retryForwardsForServer = \entries remaining ->
1620 retryForwardsForServer rest (remaining - 1)
1700 retryForwardsForServer rest (remaining - 1)
1622-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
1702-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
1703+-- Circuit breaker: skips entirely when too many peers are failing (likely local network issue).
1623retryPendingForwards : [ForwardEntry] -> IO {network} {}
1704retryPendingForwards : [ForwardEntry] -> IO {network} {}
1624retryPendingForwards = \pending -> do
1705retryPendingForwards = \pending -> do
1625 let servers = collectPeerServers pending (\x -> x.toServer)
1706 let servers = collectPeerServers pending (\x -> x.toServer)
1626 t <- now
1707 t <- now
1627 failures <- *forwardFailures
1708 failures <- *forwardFailures
1709+ if isFederationCircuitOpen failures t
1710+ then println ("federation: circuit breaker open (" ++ show (count failures) ++ " peers failing), skipping forward retries")
1628 inFlight <- *federationInFlight
1712 inFlight <- *federationInFlight
1629 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
1713 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
1630 sendToEach eligibleServers (\server ->
1714 sendToEach eligibleServers (\server ->
@@ -1648,16 +1732,21 @@ forwardQueuedMessages = \targets t -> do
1648 yield capped
1732 yield capped
1649 let servers = collectPeerServers newForwards (\x -> x.toServer)
1733 let servers = collectPeerServers newForwards (\x -> x.toServer)
1650 failures <- *forwardFailures
1734 failures <- *forwardFailures
1735+ if isFederationCircuitOpen failures t
1736+ then println ("federation: circuit breaker open, skipping queued message forwarding")
1651 inFlight <- *federationInFlight
1738 inFlight <- *federationInFlight
1652 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
1739 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
1653 sendToEach eligibleServers (\server ->
1740 sendToEach eligibleServers (\server ->
1654 retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)
1741 retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)
1656−fetchFederationAck = \origin messageIds ->
1743+fetchFederationAck = \origin messageIds -> do
1657 fetchFederation origin (FederationAck {
1745 fetchFederation origin (FederationAck {
1658 authorization: federationAuthForServer origin,
1746 authorization: federationAuthForServer origin,
1659 messageIds: messageIds,
1747 messageIds: messageIds,
1660− originServer: origin, fromServer: serverHostN})
1748+ originServer: origin, fromServer: serverHostN,
1749+ requestTimestamp: t})
1662sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
1751sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
1663sendBatchedAcksToServer = \serverAcks server ->
1752sendBatchedAcksToServer = \serverAcks server ->
@@ -1680,6 +1769,9 @@ sendAcksAndClean = \pendingAcks -> do
1680 let servers = collectPeerServers pendingAcks (\a -> a.originServer)
1769 let servers = collectPeerServers pendingAcks (\a -> a.originServer)
1681 t <- now
1770 t <- now
1682 failures <- *forwardFailures
1771 failures <- *forwardFailures
1772+ if isFederationCircuitOpen failures t
1773+ then println ("federation: circuit breaker open, skipping ack retries")
1683 inFlight <- *federationInFlight
1775 inFlight <- *federationInFlight
1684 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
1776 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
1685 sendToEach eligibleServers (\server ->
1777 sendToEach eligibleServers (\server ->
@@ -1736,6 +1828,9 @@ retryOverflowAcks = \overflow -> do
1736 let servers = collectPeerServers overflow (\a -> a.originServer)
1828 let servers = collectPeerServers overflow (\a -> a.originServer)
1737 t <- now
1829 t <- now
1738 failures <- *forwardFailures
1830 failures <- *forwardFailures
1831+ if isFederationCircuitOpen failures t
1832+ then println ("federation: circuit breaker open, skipping overflow ack retries")
1739 inFlight <- *federationInFlight
1834 inFlight <- *federationInFlight
1740 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
1835 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
1741 sendToEach eligibleServers (\server ->
1836 sendToEach eligibleServers (\server ->
@@ -1751,18 +1846,29 @@ retryOverflowAcks = \overflow -> do
1751 ackOverflow <- *ackOverflow
1846 ackOverflow <- *ackOverflow
1752 set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))
1847 set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))
1754−-- Poll federationInFlight until all in-flight requests complete or deadline passes.
1849+-- Wait for all in-flight federation requests to complete or timeout.
1755−-- Used during drain to wait for outstanding federation work before signaling completion.
1850+-- Uses STM retry instead of busy-wait: blocks until *federationInFlight changes
1851+-- (when clearFederationInFlight writes) or *drainTimedOut is set by the timeout fork.
1852+-- This avoids the 1-second polling loop and wakes immediately when work completes.
1756waitForInFlightDrain = \deadline -> do
1853waitForInFlightDrain = \deadline -> do
1757 t <- now
1854 t <- now
1758− if t >= deadline then println "drain: timed out waiting for in-flight federation tasks"
1855+ let remaining = deadline - t
1759− else do
1856+ atomic do set *drainTimedOut = False {}
1760− inFlight <- *federationInFlight
1857+ -- Fork a timer that sets the timeout signal after the deadline
1761− let activeCount = fold (\acc r -> acc + r.count) 0 inFlight
1858+ fork (do
1859+ when (remaining > 0) (sleep remaining)
1860+ atomic do set *drainTimedOut = True {})
1861+ -- Block until either all in-flight complete or timeout fires
1863+ timedOut <- *drainTimedOut
1762− if activeCount <= 0 then yield {}
1864+ if timedOut then yield {}
1764− sleep 1000
1866+ inFlight <- *federationInFlight
1765− waitForInFlightDrain deadline
1867+ let activeCount = fold (\acc r -> acc + r.count) 0 inFlight
1868+ if activeCount <= 0 then yield {}
1870+ timedOut <- *drainTimedOut
1871+ when timedOut (println "drain: timed out waiting for in-flight federation tasks")
1767-- Full maintenance cycle: prune expired state, retry pending forwards and acks,
1873-- Full maintenance cycle: prune expired state, retry pending forwards and acks,
1768-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
1874-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
@@ -2393,10 +2499,12 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2393 forwards <- *forwards
2499 forwards <- *forwards
2394 yield (any forwards (\f -> f.messageId == effectiveMsgId && f.toServer == server))
2500 yield (any forwards (\f -> f.messageId == effectiveMsgId && f.toServer == server))
2395 when stillPending (do
2501 when stillPending (do
2396 fwdResult <- fetchFederation server (ForwardMessage {
2503 fwdResult <- fetchFederation server (ForwardMessage {
2397 toKey: to, encryptedBlob: encryptedBlob,
2504 toKey: to, encryptedBlob: encryptedBlob,
2398 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
2505 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
2506+ fromServer: serverHostN, requestTimestamp: fwdT,
2399− fromServer: serverHostN, authorization: federationAuthForServer server})
2507+ authorization: federationAuthForServer server})
2400 case fwdResult of
2508 case fwdResult of
2401 Err e -> do
2509 Err e -> do
2402 println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
2510 println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
@@ -2535,10 +2643,12 @@ handleLogout = \authorization clientIp respond -> do
2535 writeSessionBucket bucket (do s <- sessions; where not (safeCompare s.token token); yield s)
2643 writeSessionBucket bucket (do s <- sessions; where not (safeCompare s.token token); yield s)
2536 sc <- *totalSessionCount
2644 sc <- *totalSessionCount
2537 set *totalSessionCount = max 0 (sc - 1)
2645 set *totalSessionCount = max 0 (sc - 1)
2538− -- Check if this was the last session for this key
2646+ -- Check if this was the last session for this key.
2539− allSessions <- readAllSessions
2647+ -- Short-circuit scan: reads buckets one at a time, stopping as soon as
2540− let remaining = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
2648+ -- a matching session is found. Avoids reading all 16 buckets when the
2541− if remaining == 0
2649+ -- user has sessions in other buckets (common case: multi-device).
2650+ hasRemaining <- hasSessionForKey pubkey 0
2651+ if not hasRemaining
2543 -- Remove local presence immediately
2653 -- Remove local presence immediately
2544 presence <- *presence
2654 presence <- *presence
@@ -2561,10 +2671,10 @@ handleLogout = \authorization clientIp respond -> do
2561-- Validates events, merges into presence table (last-write-wins), and re-gossips
2671-- Validates events, merges into presence table (last-write-wins), and re-gossips
2562-- to other known peers with decremented hop count. Triggers message forwarding
2672-- to other known peers with decremented hop count. Triggers message forwarding
2563-- when a remote key comes online.
2673-- when a remote key comes online.
2564−handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do
2674+handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorization respond -> do
2565 t <- now
2675 t <- now
2566 let cappedHops = clamp 0 maxGossipHops hopsRemaining
2676 let cappedHops = clamp 0 maxGossipHops hopsRemaining
2567− withFederationAuth authorization t respond fromServer (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
2677+ withFederationAuth authorization t respond fromServer requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
2568 let validated = validGossipEvents events t
2678 let validated = validGossipEvents events t
2569 let onlineEntries = validated.online
2679 let onlineEntries = validated.online
2570 let trustedOffline = trustedOfflineEvents validated.offline fromServer
2680 let trustedOffline = trustedOfflineEvents validated.offline fromServer
@@ -2608,9 +2718,9 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do
2608-- (fromServer != originServer) where an intermediate server re-forwards because
2718-- (fromServer != originServer) where an intermediate server re-forwards because
2609-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored
2719-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored
2610-- twice) and forwardCandidates (won't re-forward to the same server).
2720-- twice) and forwardCandidates (won't re-forward to the same server).
2611−handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer authorization respond -> do
2721+handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond -> do
2612 t <- now
2722 t <- now
2613− withFederationAuth authorization t respond fromServer (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
2723+ withFederationAuth authorization t respond fromServer requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
2614 when (fromServer != originServer)
2724 when (fromServer != originServer)
2615 (println ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
2725 (println ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
2616 -- Replay protection: reject forwards with timestamps older than messageMaxAge.
2726 -- Replay protection: reject forwards with timestamps older than messageMaxAge.
@@ -2672,16 +2782,27 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
2672-- POST /federation/ack: Peer server confirms delivery of messages we originated.
2782-- POST /federation/ack: Peer server confirms delivery of messages we originated.
2673-- Removes corresponding forward entries. Also cleans up the original message if the
2783-- Removes corresponding forward entries. Also cleans up the original message if the
2674-- recipient has no local presence and no other pending forwards for that message.
2784-- recipient has no local presence and no other pending forwards for that message.
2675−handleFederationAck = \messageIds originServer fromServer authorization respond -> do
2785+handleFederationAck = \messageIds originServer fromServer requestTimestamp authorization respond -> do
2676 t <- now
2786 t <- now
2677− withFederationAuth authorization t respond fromServer (isLocalServer originServer && not (isLocalServer fromServer)) (do
2787+ withFederationAuth authorization t respond fromServer requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (do
2678 let dedupedIds = dedupMessageIds messageIds
2788 let dedupedIds = dedupMessageIds messageIds
2679− -- Only clear forward entries; origin messages are retained until the recipient
2789+ -- Clear forward entries for the acking server. Extract removed forwards' toKey
2680− -- polls from this server or they age out via unackedMessageMaxAge. This prevents
2790+ -- so we can reclaim origin messages that are no longer needed.
2681− -- a compromised peer from deleting messages by sending unverified acks.
2682− atomic do
2791+ removedForwards <- atomic do
2683 forwards <- *forwards
2792 forwards <- *forwards
2793+ let removed = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
2684 set *forwards = forwardsExcluding forwards fromServer dedupedIds
2794 set *forwards = forwardsExcluding forwards fromServer dedupedIds
2796+ -- Best-effort origin message cleanup (forked to avoid slowing the ack response).
2797+ -- Reclaims messages past the ack grace period when the recipient has no local
2798+ -- presence and no other pending forwards remain. Without this, confirmed-delivered
2799+ -- messages sit for up to unackedMessageMaxAge (7 days).
2800+ when (count removedForwards > 0) (fork (do
2801+ sleep ackGracePeriod
2803+ -- Each forward is cleaned up in its own atomic block (short locks, one per message).
2804+ -- Uses explicit recursion since fold is pure and can't sequence atomic IO.
2805+ ackCleanupForwards removedForwards t2))
2685 yield (respond {ok: True {}, error: ""}))
2806 yield (respond {ok: True {}, error: ""}))
2687-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
2808-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
@@ -2883,6 +3004,67 @@ handleStatsKey = \pubkey authorization respond -> do
2883 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})
3004 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})
2884 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})
3005 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})
3007+emptySnapshot = {ok: False {}, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: []}
3009+-- GET /maintenance/snapshot: Export federation metadata for persistence.
3010+-- Reads all dedup state (ackedIds, deliveredIds), presence, and peer failure records
3011+-- in a single atomic snapshot. Messages and forwards are excluded — messages are
3012+-- ephemeral (clients re-send unconfirmed), and forwards are re-created from
3013+-- messages + presence after restore by running a prune cycle.
3014+handleSnapshot = \authorization respond -> do
3015+ let authorized = verifyMaintenanceAuth authorization
3016+ if not authorized then yield (respond emptySnapshot)
3018+ result <- atomic do
3023+ ff <- *forwardFailures
3024+ yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff}
3025+ println ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds")
3026+ yield (respond {ok: True {}, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures})
3028+-- POST /maintenance/restore: Import a previously-exported snapshot.
3029+-- Requires drain mode to prevent conflicts with live state. Merges imported data
3030+-- with existing state (dedup entries are unioned, presence is merged by freshness).
3031+-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
3032+-- trigger federation forward creation from messages + restored presence.
3033+handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData authorization respond -> do
3034+ let authorized = verifyMaintenanceAuth authorization
3035+ if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
3038+ isDraining <- *draining
3039+ if not isDraining then yield (respond {ok: False {}, error: "server_must_be_draining", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
3041+ result <- atomic do
3042+ -- Merge presence (existing entries win on conflict via freshness)
3043+ presence <- *presence
3044+ let mergedPresence = mergeByPubkeyServer presence (do p <- presenceData; where p.expiresAt > t; yield p)
3045+ let cappedPresence = takeRelation maxPresenceEntries mergedPresence
3046+ set *presence = cappedPresence
3047+ -- Union dedup state, respecting capacity limits
3048+ ackedIds <- *ackedIds
3049+ let mergedAcked = union ackedIds ackedIdsData
3050+ let cappedAcked = takeRelation maxAckedIds mergedAcked
3051+ set *ackedIds = cappedAcked
3052+ ackOverflow <- *ackOverflow
3053+ let mergedOverflow = union ackOverflow ackOverflowData
3054+ set *ackOverflow = takeRelation maxAckOverflow mergedOverflow
3055+ deliveredIds <- *deliveredIds
3056+ let mergedDelivered = union deliveredIds deliveredIdsData
3057+ let cappedDelivered = takeRelation maxDeliveredIds mergedDelivered
3058+ set *deliveredIds = cappedDelivered
3059+ -- Restore peer failure records
3060+ forwardFailures <- *forwardFailures
3061+ let mergedFailures = mergeByKey (\a b -> a.server == b.server) (\x -> x.failedAt) forwardFailures forwardFailuresData
3062+ let cappedFailures = takeRelation maxForwardFailures mergedFailures
3063+ set *forwardFailures = cappedFailures
3064+ yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
3065+ println ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
3066+ yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
2886serve = \req -> case req of
3068serve = \req -> case req of
2887 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
3069 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
2888 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
3070 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
@@ -2892,12 +3074,14 @@ serve = \req -> case req of
2892 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond
3074 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond
2893 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
3075 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
2894 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
3076 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
2895− RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer authorization respond
3077+ RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
2896− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer authorization respond
3078+ ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond
2897− FederationAck {messageIds, originServer, fromServer, authorization, respond} -> handleFederationAck messageIds originServer fromServer authorization respond
3079+ FederationAck {messageIds, originServer, fromServer, requestTimestamp, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp authorization respond
2898 Prune {authorization, respond} -> handlePrune authorization respond
3080 Prune {authorization, respond} -> handlePrune authorization respond
2899 Drain {authorization, respond} -> handleDrain authorization respond
3081 Drain {authorization, respond} -> handleDrain authorization respond
2900 Undrain {authorization, respond} -> handleUndrain authorization respond
3082 Undrain {authorization, respond} -> handleUndrain authorization respond
3083+ Snapshot {authorization, respond} -> handleSnapshot authorization respond
3084+ Restore {presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, authorization, respond} -> handleRestore presence ackedIds ackOverflow deliveredIds forwardFailures authorization respond
2901 Health {respond} -> handleHealth respond
3085 Health {respond} -> handleHealth respond
2902 Stats {authorization, respond} -> handleStats authorization respond
3086 Stats {authorization, respond} -> handleStats authorization respond
2903 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
3087 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
@@ -2988,9 +3172,23 @@ main = do
2988 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
3172 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
2989 && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0
3173 && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0
2990 && maxPresenceTtl > 0 && maxForwardsEntries > 0 && maxAckedIds > 0
3174 && maxPresenceTtl > 0 && maxForwardsEntries > 0 && maxAckedIds > 0
3175+ -- Relationship invariants: these misconfigurations cause subtle bugs
3176+ -- (e.g., messages pruned before their stated max age, backoff that never increases)
3177+ let configRelationsOk = unackedMessageMaxAge <= messageMaxAge
3178+ && challengeExpiry < sessionExpiry
3179+ && retryBackoffBase <= retryBackoffMax
3180+ && backgroundPruneInterval < healthPruneMaxAge
3181+ && ackGracePeriod < unackedMessageMaxAge
3182+ && clockSkewTolerance < challengeExpiry
3183+ && maxMessagesPerRecipient <= maxMessages
3184+ && maxSessionsPerKey <= maxSessions
3185+ && maxChallengesPerKey <= maxChallenges
3186+ && maxPollsPerKey <= maxActivePolls
2991 let peerSecretsOk = none federationPeerSecrets (\p -> isBadServerName (normalizeServerName p.server) || p.secret == "")
3187 let peerSecretsOk = none federationPeerSecrets (\p -> isBadServerName (normalizeServerName p.server) || p.secret == "")
2992 if not configOk
3188 if not configOk
2993 then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
3189 then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
3190+ else if not configRelationsOk
3191+ then println "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, per-key limits <= global limits), server not started"
2994 else if not peerSecretsOk
3192 else if not peerSecretsOk
2995 then println "FATAL: invalid federationPeerSecrets — server names must pass validation and secrets must be non-empty"
3193 then println "FATAL: invalid federationPeerSecrets — server names must pass validation and secrets must be non-empty"
@@ -3012,6 +3210,8 @@ main = do
3012 println (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
3210 println (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
3013 when (count federationPeerSecrets > 0)
3211 when (count federationPeerSecrets > 0)
3014 (println (" per-peer secrets: " ++ show (count federationPeerSecrets) ++ " configured"))
3212 (println (" per-peer secrets: " ++ show (count federationPeerSecrets) ++ " configured"))
3213+ when (federationCircuitBreakerThreshold > 0)
3214+ (println (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
3015 fork backgroundPrune
3215 fork backgroundPrune
3016 fork pollHeartbeat
3216 fork pollHeartbeat
3017 listen 8080 serve
3217 listen 8080 serve