git.iko.soy/skrepka
on master help sign in
Improve relay security and reliability: federation/global rate limits, forward dedup, atomic poll check, exponential backoff, handler extraction
9aca0cciko
server.knot+296−218
compare with master
unifiedsplit
@@ -56,6 +56,15 @@ maxSendRatePerKey = 20 -- max messages sent per key per window
56sendRateWindowMs : Int<Ms>
56sendRateWindowMs : Int<Ms>
57sendRateWindowMs = 1000
57sendRateWindowMs = 1000
58maxSendRateEntries = 1000
58maxSendRateEntries = 1000
59+-- Federation rate limiting (per-server)
60+maxFederationRatePerServer = 50 -- max federation requests per server per window
61+federationRateWindowMs : Int<Ms>
62+federationRateWindowMs = 1000
63+maxFederationRateEntries = 100
64+-- Global request rate limiting
65+maxGlobalRequestRate = 1000 -- max total requests per window
66+globalRateWindowMs : Int<Ms>
67+globalRateWindowMs = 1000
59-- Session and gossip timing
68-- Session and gossip timing
60sessionExpiry : Int<Ms>
69sessionExpiry : Int<Ms>
61sessionExpiry = 3600000 -- 1 hour
70sessionExpiry = 3600000 -- 1 hour
@@ -171,6 +180,8 @@ type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}
171-- Sliding-window rate limit state per pubkey
180-- Sliding-window rate limit state per pubkey
172type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
181type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
173type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
182type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
183+-- Sliding-window rate limit state per federation peer server
184+type FederationRateEntry = {server: Text, windowStart: Timestamp, count: Int}
174-- Tracks active long-poll connections to enforce per-key concurrency limits
185-- Tracks active long-poll connections to enforce per-key concurrency limits
175type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
186type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
176data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
187data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
@@ -240,6 +251,12 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
240-- Final dedup set: messages confirmed delivered (ack sent or overflow drained)
251-- Final dedup set: messages confirmed delivered (ack sent or overflow drained)
241*deliveredIds : [DeliveredEntry]
252*deliveredIds : [DeliveredEntry]
242
253
254+*federationRates : [FederationRateEntry]
255+
256+*globalRateCount : Int
257+
258+*globalRateWindowStart : Timestamp
259+
243-- ============================================
260-- ============================================
244-- Routes
261-- Routes
245-- ============================================
262-- ============================================
@@ -603,7 +620,12 @@ safeCompareLoop = \s1 s2 n result ->
603
620
604safeCompare : Text -> Text -> Bool
621safeCompare : Text -> Text -> Bool
605safeCompare = \a b ->
622safeCompare = \a b ->
623+ let lenA = count (chars a) in
624+ let lenB = count (chars b) in
606− safeCompareLoop a b maxTokenLen 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0
625+ safeCompareLoop a b maxTokenLen 0
626+ + (if lenA != lenB then 1 else 0)
627+ + (if a == "" then 1 else 0)
628+ + (if b == "" then 1 else 0) == 0
607
629
608-- Remove the single item with the smallest key value (LRU-style eviction)
630-- Remove the single item with the smallest key value (LRU-style eviction)
609evictOneOldest = \items getKey ->
631evictOneOldest = \items getKey ->
@@ -760,6 +782,8 @@ pruneExpired = \t -> do
760 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < retryBackoffMax * 2; yield f
782 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < retryBackoffMax * 2; yield f
761 challengeRates <- *challengeRates
783 challengeRates <- *challengeRates
762 set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r
784 set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r
785+ federationRates <- *federationRates
786+ set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs; yield r
763 activePolls <- *activePolls
787 activePolls <- *activePolls
764 set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p
788 set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p
765 ackOverflow <- *ackOverflow
789 ackOverflow <- *ackOverflow
@@ -833,12 +857,47 @@ maybePruneAndBroadcast = \t -> do
833 pruneResult <- atomic do maybePrune t
857 pruneResult <- atomic do maybePrune t
834 forkBroadcastOffline t pruneResult
858 forkBroadcastOffline t pruneResult
835
859
860+checkFederationRate = \server t -> atomic do
861+ federationRates <- *federationRates
862+ let entry = headOf (do r <- federationRates; where r.server == server; yield r)
863+ let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= federationRateWindowMs
864+ let rateCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count
865+ if rateCount >= maxFederationRatePerServer
866+ then yield (True {})
867+ else do
868+ let newRates = if isNewWindow
869+ then union (do r <- federationRates; where r.server != server; yield r) [{server: server, windowStart: t, count: 1}]
870+ else do r <- federationRates; yield (if r.server == server then {server: r.server, windowStart: r.windowStart, count: r.count + 1} else r)
871+ set *federationRates = if count newRates > maxFederationRateEntries then (do r <- newRates; where t - r.windowStart < federationRateWindowMs; yield r) else newRates
872+ yield (False {})
873+
874+checkGlobalRate = \t -> atomic do
875+ windowStart <- *globalRateWindowStart
876+ rateCount <- *globalRateCount
877+ if t - windowStart >= globalRateWindowMs
878+ then do
879+ set *globalRateWindowStart = t
880+ set *globalRateCount = 1
881+ yield (False {})
882+ else if rateCount >= maxGlobalRequestRate
883+ then yield (True {})
884+ else do
885+ set *globalRateCount = rateCount + 1
886+ yield (False {})
887+
836−withFederationAuth = \authorization t respond precondition action ->
888+withFederationAuth = \authorization t respond fromServer precondition action ->
837 if not (verifyFederationAuth authorization) || not precondition
889 if not (verifyFederationAuth authorization) || not precondition
838 then do
890 then do
839 maybePruneAndBroadcast t
891 maybePruneAndBroadcast t
840 yield (respond {ok: False {}})
892 yield (respond {ok: False {}})
893+ else do
894+ rateLimited <- checkFederationRate fromServer t
895+ if rateLimited
896+ then do
897+ println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
898+ maybePruneAndBroadcast t
899+ yield (respond {ok: False {}})
841− else action
900+ else action
842
901
843withSessionAuth = \authorization t onUnauth onAuth -> do
902withSessionAuth = \authorization t onUnauth onAuth -> do
844 sessions <- *sessions
903 sessions <- *sessions
@@ -853,14 +912,13 @@ removeForward = \mid srv -> atomic do
853 forwards <- *forwards
912 forwards <- *forwards
854 set *forwards = forwardsExcluding forwards srv [mid]
913 set *forwards = forwardsExcluding forwards srv [mid]
855
914
915+pow2 : Int -> Int
916+pow2 = \n -> if n <= 0 then 1 else 2 * pow2 (n - 1)
917+
856−-- Exponential backoff: 30s -> 60s -> 120s -> 240s -> 480s (capped)
918+-- Exponential backoff: base * 2^(failures-1), capped at retryBackoffMax
857retryBackoff : Int -> Int<Ms>
919retryBackoff : Int -> Int<Ms>
858retryBackoff = \failures ->
920retryBackoff = \failures ->
859− if failures <= 1 then retryBackoffBase
860− else if failures <= 2 then retryBackoffBase * 2
861− else if failures <= 3 then retryBackoffBase * 4
862− else if failures <= 4 then retryBackoffBase * 8
863− else retryBackoffMax
921+ min retryBackoffMax (retryBackoffBase * pow2 (max 0 (failures - 1)))
864
922
865shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
923shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
866shouldRetryServer = \server t failures ->
924shouldRetryServer = \server t failures ->
@@ -1163,15 +1221,19 @@ pollOnce = \token cursor -> do
1163 then yield {authorized: True {}, events: [], cursor: cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
1221 then yield {authorized: True {}, events: [], cursor: cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
1164 else do
1222 else do
1165 pollResult <- do
1223 pollResult <- do
1166− -- Phase 1: check for existing messages (no blocking)
1224+ -- Phase 1: check for existing messages (atomic read)
1225+ initialResult <- atomic do
1167− initialMessages <- *messages
1226+ messages <- *messages
1168− let initial = pollableMessages initialMessages s.pubkey s.clampedCursor t
1227+ if count (pollableMessages messages s.pubkey s.clampedCursor t) > 0
1169− if count initial > 0
1170− then do
1228+ then do
1171− result <- atomic do pollNewEvents s.pubkey s.clampedCursor t
1229+ r <- pollNewEvents s.pubkey s.clampedCursor t
1230+ yield (Just {value: {events: r.events, cursor: r.cursor}})
1231+ else yield (Nothing {})
1232+ case initialResult of
1233+ Just {value: result} ->
1172 yield {authorized: True {}, events: result.events, cursor: result.cursor}
1234 yield {authorized: True {}, events: result.events, cursor: result.cursor}
1173− else do
1235+ Nothing {} -> do
1174− -- Phase 2: wait for signal or heartbeat (no *messages dependency)
1236+ -- Phase 2: wait for signal or heartbeat
1175 atomic do
1237 atomic do
1176 wakeup <- *pollWakeup
1238 wakeup <- *pollWakeup
1177 signal <- readMessageSignal s.pubkey
1239 signal <- readMessageSignal s.pubkey
@@ -1307,13 +1369,14 @@ remoteOnlyTargets = \targets localOnline ->
1307-- Request Handler
1369-- Request Handler
1308-- ============================================
1370-- ============================================
1309
1371
1310−serve = \req -> case req of
1311−
1312− -- POST /auth/challenge: Generate a random challenge for Ed25519 authentication.
1372+-- POST /auth/challenge: Generate a random challenge for Ed25519 authentication.
1313− -- Rate-limited per key; globally capped. Challenge expires in 1 minute.
1373+-- Rate-limited per key; globally capped. Challenge expires in 1 minute.
1314− ReqChallenge {pubkey, respond} -> do
1374+handleChallenge = \pubkey respond -> do
1375+ t <- now
1376+ globalRateLimited <- checkGlobalRate t
1377+ if globalRateLimited then yield (respond {challenge: "", ok: False {}, expiresAt: 0})
1378+ else do
1315 challenge <- genToken
1379 challenge <- genToken
1316− t <- now
1317 result <- atomic do
1380 result <- atomic do
1318 pr <- maybePrune t
1381 pr <- maybePrune t
1319 challenges <- *challenges
1382 challenges <- *challenges
@@ -1347,68 +1410,71 @@ serve = \req -> case req of
1347 forkBroadcastOffline t result.pr
1410 forkBroadcastOffline t result.pr
1348 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
1411 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
1349
1412
1350− -- POST /auth/verify: Verify Ed25519 signature over challenge string.
1413+-- POST /auth/verify: Verify Ed25519 signature over challenge string.
1351− -- On success: creates a session, registers local presence, and gossips online status.
1414+-- On success: creates a session, registers local presence, and gossips online status.
1352− VerifyAuth {pubkey, challenge, signature, respond} -> do
1415+handleVerifyAuth = \pubkey challenge signature respond -> do
1353− t <- now
1416+ t <- now
1354− let expiry = t + sessionExpiry
1417+ let expiry = t + sessionExpiry
1355− let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
1418+ let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
1356− validation <- atomic do
1419+ validation <- atomic do
1357− pr <- maybePrune t
1420+ pr <- maybePrune t
1358− if not sigValid
1421+ if not sigValid
1359− then yield {pr: pr, valid: False {}}
1360− else do
1361− challenges <- *challenges
1362− let hasMatch = challengeMatch challenges pubkey challenge t
1363− if not hasMatch
1364− then yield {pr: pr, valid: False {}}
1422+ then yield {pr: pr, valid: False {}}
1365− else do
1366− set *challenges = challengesExcluding challenges pubkey challenge
1367− yield {pr: pr, valid: True {}}
1368− if validation.valid
1369− then do
1370− token <- genToken
1371− peerServers <- atomic do
1372− sessions <- *sessions
1373− presence <- *presence
1374− let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
1375− set *sessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
1376− let presenceWithout = presenceWithoutLocalKey presence pubkey
1377− let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
1378− let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
1379− set *presence = updatedPresence
1380− yield (if federationEnabled
1381− then collectPeerServers updatedPresence (\p -> p.server)
1382− else [])
1383− let expiredLocal = do p <- validation.pr.expiredLocal; where p.pubkey != pubkey; yield p
1384− fork (do
1385− when federationEnabled
1386− (sendBatchedGossip peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops)
1387− broadcastOfflineGossip t validation.pr.peerServers expiredLocal)
1388− yield (respond {token: token, expiresAt: expiry})
1389 else do
1423 else do
1424+ challenges <- *challenges
1425+ let hasMatch = challengeMatch challenges pubkey challenge t
1426+ if not hasMatch
1427+ then yield {pr: pr, valid: False {}}
1428+ else do
1429+ set *challenges = challengesExcluding challenges pubkey challenge
1430+ yield {pr: pr, valid: True {}}
1431+ if validation.valid
1432+ then do
1433+ token <- genToken
1434+ peerServers <- atomic do
1435+ sessions <- *sessions
1436+ presence <- *presence
1437+ let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
1438+ set *sessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
1439+ let presenceWithout = presenceWithoutLocalKey presence pubkey
1440+ let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
1441+ let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
1442+ set *presence = updatedPresence
1443+ yield (if federationEnabled
1444+ then collectPeerServers updatedPresence (\p -> p.server)
1445+ else [])
1446+ let expiredLocal = do p <- validation.pr.expiredLocal; where p.pubkey != pubkey; yield p
1447+ fork (do
1448+ when federationEnabled
1449+ (sendBatchedGossip peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops)
1450+ broadcastOfflineGossip t validation.pr.peerServers expiredLocal)
1451+ yield (respond {token: token, expiresAt: expiry})
1452+ else do
1390− forkBroadcastOffline t validation.pr
1453+ forkBroadcastOffline t validation.pr
1391− yield (respond {token: "", expiresAt: 0})
1454+ yield (respond {token: "", expiresAt: 0})
1392−
1455+
1393− -- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor
1456+-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor
1394− -- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.
1457+-- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.
1395− -- Acks for federated messages are sent to origin servers in the background.
1458+-- Acks for federated messages are sent to origin servers in the background.
1396− Poll {cursor, authorization, respond} -> do
1459+handlePoll = \cursor authorization respond -> do
1397− result <- case extractToken authorization of
1460+ result <- case extractToken authorization of
1398− Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
1461+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
1399− Just {value: token} -> pollOnce token cursor
1462+ Just {value: token} -> pollOnce token cursor
1400− tAfterPoll <- now
1463+ tAfterPoll <- now
1401− maybePruneAndBroadcast tAfterPoll
1464+ maybePruneAndBroadcast tAfterPoll
1402− fork (when federationEnabled (do
1465+ fork (when federationEnabled (do
1403− sendAcksAndClean result.pendingAcks
1466+ sendAcksAndClean result.pendingAcks
1404− retryEvictedAcks result.evictedEntries result.pendingAcks))
1467+ retryEvictedAcks result.evictedEntries result.pendingAcks))
1405− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
1468+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
1406−
1469+
1407− -- POST /messages: Send an encrypted message to a recipient.
1470+-- POST /messages: Send an encrypted message to a recipient.
1408− -- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
1471+-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
1409− -- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
1472+-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
1410− SendMessage {to, encryptedBlob, timestamp, messageId, authorization, respond} -> do
1473+handleSendMessage = \to encryptedBlob timestamp messageId authorization respond -> do
1411− t <- now
1474+ t <- now
1475+ globalRateLimited <- checkGlobalRate t
1476+ if globalRateLimited then yield (respond {status: MsRejected {}, messageId: ""})
1477+ else do
1412 effectiveMsgId <- if messageId != "" then yield messageId else genToken
1478 effectiveMsgId <- if messageId != "" then yield messageId else genToken
1413 let clampedTs = clampTimestamp timestamp t
1479 let clampedTs = clampTimestamp timestamp t
1414 result <- atomic do
1480 result <- atomic do
@@ -1456,18 +1522,18 @@ serve = \req -> case req of
1456 fork (do
1522 fork (do
1457 when (count remoteServers > 0 && federationEnabled) (
1523 when (count remoteServers > 0 && federationEnabled) (
1458 sendToEach remoteServers (\server -> do
1524 sendToEach remoteServers (\server -> do
1525+ -- Pre-register forward entry for dedup and retry on failure
1526+ atomic do
1527+ forwards <- *forwards
1528+ when (count forwards < maxForwardsEntries)
1529+ (set *forwards = union forwards [{messageId: effectiveMsgId, toServer: server, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}])
1459 fwdResult <- fetch (federationUrl server) (ForwardMessage {
1530 fwdResult <- fetch (federationUrl server) (ForwardMessage {
1460 toKey: to, encryptedBlob: encryptedBlob,
1531 toKey: to, encryptedBlob: encryptedBlob,
1461 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
1532 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
1462 fromServer: serverHostN, authorization: federationAuth})
1533 fromServer: serverHostN, authorization: federationAuth})
1463 case fwdResult of
1534 case fwdResult of
1464− Err _ -> do
1465− println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)
1535+ Err _ -> println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)
1466− atomic do
1536+ Ok _ -> removeForward effectiveMsgId server))
1467− forwards <- *forwards
1468− when (count forwards < maxForwardsEntries)
1469− (set *forwards = union forwards [{messageId: effectiveMsgId, toServer: server, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}])
1470− Ok _ -> yield {}))
1471 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
1537 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
1472 _ -> forkBroadcastOffline t result.pr
1538 _ -> forkBroadcastOffline t result.pr
1473 yield (case result.send of
1539 yield (case result.send of
@@ -1475,137 +1541,149 @@ serve = \req -> case req of
1475 SendDuplicate {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId}
1541 SendDuplicate {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId}
1476 SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})
1542 SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})
1477
1543
1478−
1479− -- GET /lookup/:pubkey: Look up the best known server for a pubkey.
1544+-- GET /lookup/:pubkey: Look up the best known server for a pubkey.
1480− -- Returns local server if key is online locally, otherwise the remote with latest expiry.
1545+-- Returns local server if key is online locally, otherwise the remote with latest expiry.
1481− LookupPresence {pubkey, authorization, respond} -> do
1546+handleLookupPresence = \pubkey authorization respond -> do
1547+ t <- now
1548+ result <- atomic do
1549+ pr <- maybePrune t
1550+ withSessionAuth authorization t
1551+ (yield {pr: pr, server: "", authorized: False {}})
1552+ (\_ -> do
1553+ presence <- *presence
1554+ let foundServer = bestServerForKey (presenceForKey presence pubkey t)
1555+ yield {pr: pr, server: foundServer, authorized: True {}})
1556+ forkBroadcastOffline t result.pr
1557+ yield (respond {server: result.server, authorized: result.authorized})
1558+
1559+-- POST /federation/gossip: Receive presence gossip from a peer server.
1560+-- Validates events, merges into presence table (last-write-wins), and re-gossips
1561+-- to other known peers with decremented hop count. Triggers message forwarding
1562+-- when a remote key comes online.
1563+handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do
1564+ t <- now
1565+ let cappedHops = clamp 0 maxGossipHops hopsRemaining
1566+ withFederationAuth authorization t respond fromServer (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
1567+ let validated = validGossipEvents events t
1568+ let onlineEntries = validated.online
1569+ let trustedOffline = trustedOfflineEvents validated.offline fromServer
1570+ let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
1571+ let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
1572+ let deduped = mergeByPubkeyServer [] filteredOnline
1573+ gossipResult <- atomic do
1574+ pr <- maybePrune t
1575+ presence <- *presence
1576+ let gm = mergeGossipPresence presence deduped trustedOffline fromServer t
1577+ set *presence = gm.finalPresence
1578+ yield {gm: gm, pr: pr}
1579+ let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
1580+ when (not gossipResult.gm.withinCapacity)
1581+ (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
1582+ let newlyOnlineRemote = newlyOnlineRemoteKeys filteredOnline gossipResult.gm.appliedKeys
1583+ fork (do
1584+ when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (
1585+ sendBatchedGossip (gossipTargets gossipResult.gm.gossipPeerServers fromServer) effectiveEvents (cappedHops - 1))
1586+ broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal
1587+ when (federationEnabled && count newlyOnlineRemote > 0)
1588+ (forwardQueuedMessages newlyOnlineRemote t))
1589+ yield (respond {ok: True {}}))
1590+
1591+-- POST /federation/forward: Receive a message forwarded from a peer server.
1592+-- Only accepts messages where fromServer == originServer (no cascade forwarding).
1593+-- Deduplicates and stores; rejects if recipient mailbox is full.
1594+handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer authorization respond -> do
1482− t <- now
1595+ t <- now
1596+ withFederationAuth authorization t respond fromServer (not (isLocalServer originServer) && fromServer == originServer) (do
1597+ let clampedTs = clampTimestamp timestamp t
1483 result <- atomic do
1598 result <- atomic do
1484 pr <- maybePrune t
1599 pr <- maybePrune t
1485− withSessionAuth authorization t
1600+ messages <- *messages
1486− (yield {pr: pr, server: "", authorized: False {}})
1601+ ackedIds <- *ackedIds
1487− (\_ -> do
1602+ deliveredIds <- *deliveredIds
1488− presence <- *presence
1603+ if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds
1489− let foundServer = bestServerForKey (presenceForKey presence pubkey t)
1604+ then yield {pr: pr, forward: ForwardDuplicate {}}
1490− yield {pr: pr, server: foundServer, authorized: True {}})
1605+ else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1606+ then yield {pr: pr, forward: ForwardRejected {}}
1607+ else do
1608+ seqCounter <- *seqCounter
1609+ let seqTs = nextSeqTs t seqCounter
1610+ set *seqCounter = seqTs
1611+ let updatedMessages = union messages
1612+ [{messageId: messageId, toKey: toKey,
1613+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1614+ set *messages = updatedMessages
1615+ signalMessageForKey toKey seqTs
1616+ yield {pr: pr, forward: ForwardStored {}}
1617+ case result.forward of
1618+ ForwardRejected {} -> println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
1619+ _ -> yield {}
1491 forkBroadcastOffline t result.pr
1620 forkBroadcastOffline t result.pr
1492− yield (respond {server: result.server, authorized: result.authorized})
1621+ yield (respond {ok: case result.forward of
1493−
1622+ ForwardRejected {} -> False {}
1494− -- POST /federation/gossip: Receive presence gossip from a peer server.
1623+ _ -> True {}}))
1495− -- Validates events, merges into presence table (last-write-wins), and re-gossips
1624+
1496− -- to other known peers with decremented hop count. Triggers message forwarding
1625+-- POST /federation/ack: Peer server confirms delivery of messages we originated.
1497− -- when a remote key comes online.
1626+-- Removes corresponding forward entries. Also cleans up the original message if the
1627+-- recipient has no local presence and no other pending forwards for that message.
1498− RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
1628+handleFederationAck = \messageIds originServer fromServer authorization respond -> do
1499− t <- now
1629+ t <- now
1500− let cappedHops = clamp 0 maxGossipHops hopsRemaining
1501− withFederationAuth authorization t respond (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
1630+ withFederationAuth authorization t respond fromServer (isLocalServer originServer && not (isLocalServer fromServer)) (do
1502− let validated = validGossipEvents events t
1631+ let dedupedIds = dedupMessageIds messageIds
1503− let onlineEntries = validated.online
1504− let trustedOffline = trustedOfflineEvents validated.offline fromServer
1505− let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
1506− let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
1507− let deduped = mergeByPubkeyServer [] filteredOnline
1508− gossipResult <- atomic do
1632+ pruneResult <- atomic do
1509− pr <- maybePrune t
1633+ pr <- maybePrune t
1510− presence <- *presence
1634+ forwards <- *forwards
1511− let gm = mergeGossipPresence presence deduped trustedOffline fromServer t
1635+ let ackedForwards = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
1512− set *presence = gm.finalPresence
1636+ let updatedForwards = forwardsExcluding forwards fromServer dedupedIds
1513− yield {gm: gm, pr: pr}
1637+ set *forwards = updatedForwards
1514− let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
1638+ messages <- *messages
1515− when (not gossipResult.gm.withinCapacity)
1639+ presence <- *presence
1516− (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
1640+ let removable = do
1517− let newlyOnlineRemote = newlyOnlineRemoteKeys filteredOnline gossipResult.gm.appliedKeys
1641+ f <- ackedForwards
1518− fork (do
1642+ where none updatedForwards (\rf -> rf.messageId == f.messageId && rf.toKey == f.toKey)
1519− when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (
1643+ where none presence (\p -> p.pubkey == f.toKey && isLocalServer p.server && p.expiresAt > t)
1520− sendBatchedGossip (gossipTargets gossipResult.gm.gossipPeerServers fromServer) effectiveEvents (cappedHops - 1))
1644+ yield f
1521− broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal
1645+ set *messages = do m <- messages; where none removable (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m
1522− when (federationEnabled && count newlyOnlineRemote > 0)
1646+ yield pr
1523− (forwardQueuedMessages newlyOnlineRemote t))
1647+ forkBroadcastOffline t pruneResult
1648+ yield (respond {ok: True {}}))
1649+
1650+-- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).
1651+-- Accepts either federation or maintenance secret.
1652+handlePrune = \authorization respond -> do
1653+ t <- now
1654+ let authorized = verifyMaintenanceAuth authorization
1655+ when authorized (fork (pruneAndRetry t))
1656+ yield (respond {ok: authorized})
1657+
1524− yield (respond {ok: True {}}))
1658+handleHealth = \respond -> yield (respond {ok: True {}})
1525−
1659+
1526− -- POST /federation/forward: Receive a message forwarded from a peer server.
1660+handleStats = \authorization respond -> do
1527− -- Only accepts messages where fromServer == originServer (no cascade forwarding).
1661+ let authorized = verifyMaintenanceAuth authorization
1528− -- Deduplicates and stores; rejects if recipient mailbox is full.
1662+ if authorized
1529− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, authorization, respond} -> do
1663+ then do
1530− t <- now
1531− withFederationAuth authorization t respond (not (isLocalServer originServer) && fromServer == originServer) (do
1532− let clampedTs = clampTimestamp timestamp t
1533 result <- atomic do
1664 result <- atomic do
1534− pr <- maybePrune t
1665+ s <- *sessions
1535− messages <- *messages
1666+ m <- *messages
1536− ackedIds <- *ackedIds
1667+ p <- *presence
1537− deliveredIds <- *deliveredIds
1668+ f <- *forwards
1538− if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds
1669+ c <- *challenges
1539− then yield {pr: pr, forward: ForwardDuplicate {}}
1670+ a <- *ackedIds
1540− else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1671+ yield {sessions: count s, messages: count m, presence: count p, forwards: count f, challenges: count c, ackedIds: count a}
1541− then yield {pr: pr, forward: ForwardRejected {}}
1672+ yield (respond {sessions: result.sessions, messages: result.messages, presence: result.presence, forwards: result.forwards, challenges: result.challenges, ackedIds: result.ackedIds, ok: True {}})
1542− else do
1673+ else yield (respond {sessions: 0, messages: 0, presence: 0, forwards: 0, challenges: 0, ackedIds: 0, ok: False {}})
1543− seqCounter <- *seqCounter
1674+
1544− let seqTs = nextSeqTs t seqCounter
1675+serve = \req -> case req of
1545− set *seqCounter = seqTs
1676+ ReqChallenge {pubkey, respond} -> handleChallenge pubkey respond
1546− let updatedMessages = union messages
1677+ VerifyAuth {pubkey, challenge, signature, respond} -> handleVerifyAuth pubkey challenge signature respond
1547− [{messageId: messageId, toKey: toKey,
1678+ Poll {cursor, authorization, respond} -> handlePoll cursor authorization respond
1548− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1679+ SendMessage {to, encryptedBlob, timestamp, messageId, authorization, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization respond
1549− set *messages = updatedMessages
1680+ LookupPresence {pubkey, authorization, respond} -> handleLookupPresence pubkey authorization respond
1550− signalMessageForKey toKey seqTs
1681+ RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer authorization respond
1551− yield {pr: pr, forward: ForwardStored {}}
1682+ ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer authorization respond
1552− case result.forward of
1553− ForwardRejected {} -> println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
1554− _ -> yield {}
1555− forkBroadcastOffline t result.pr
1556− yield (respond {ok: case result.forward of
1557− ForwardRejected {} -> False {}
1558− _ -> True {}}))
1559−
1560− -- POST /federation/ack: Peer server confirms delivery of messages we originated.
1561− -- Removes corresponding forward entries. Also cleans up the original message if the
1562− -- recipient has no local presence and no other pending forwards for that message.
1563− FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
1683+ FederationAck {messageIds, originServer, fromServer, authorization, respond} -> handleFederationAck messageIds originServer fromServer authorization respond
1564− t <- now
1565− withFederationAuth authorization t respond (isLocalServer originServer && not (isLocalServer fromServer)) (do
1566− let dedupedIds = dedupMessageIds messageIds
1567− pruneResult <- atomic do
1568− pr <- maybePrune t
1569− forwards <- *forwards
1570− let ackedForwards = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
1571− let updatedForwards = forwardsExcluding forwards fromServer dedupedIds
1572− set *forwards = updatedForwards
1573− messages <- *messages
1574− presence <- *presence
1575− let removable = do
1576− f <- ackedForwards
1577− where none updatedForwards (\rf -> rf.messageId == f.messageId && rf.toKey == f.toKey)
1578− where none presence (\p -> p.pubkey == f.toKey && isLocalServer p.server && p.expiresAt > t)
1579− yield f
1580− set *messages = do m <- messages; where none removable (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m
1581− yield pr
1582− forkBroadcastOffline t pruneResult
1583− yield (respond {ok: True {}}))
1584−
1585− -- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).
1586− -- Accepts either federation or maintenance secret.
1587− Prune {authorization, respond} -> do
1684+ Prune {authorization, respond} -> handlePrune authorization respond
1588− t <- now
1589− let authorized = verifyMaintenanceAuth authorization
1590− when authorized (fork (pruneAndRetry t))
1591− yield (respond {ok: authorized})
1592−
1593− Health {respond} -> yield (respond {ok: True {}})
1685+ Health {respond} -> handleHealth respond
1594−
1595− Stats {authorization, respond} -> do
1686+ Stats {authorization, respond} -> handleStats authorization respond
1596− let authorized = verifyMaintenanceAuth authorization
1597− if authorized
1598− then do
1599− result <- atomic do
1600− s <- *sessions
1601− m <- *messages
1602− p <- *presence
1603− f <- *forwards
1604− c <- *challenges
1605− a <- *ackedIds
1606− yield {sessions: count s, messages: count m, presence: count p, forwards: count f, challenges: count c, ackedIds: count a}
1607− yield (respond {sessions: result.sessions, messages: result.messages, presence: result.presence, forwards: result.forwards, challenges: result.challenges, ackedIds: result.ackedIds, ok: True {}})
1608− else yield (respond {sessions: 0, messages: 0, presence: 0, forwards: 0, challenges: 0, ackedIds: 0, ok: False {}})
1609
1687
1610-- ============================================
1688-- ============================================
1611-- Entry Point
1689-- Entry Point