@@ -189,6 +189,15 @@ maxPresencePerServer = 5000
189-- the same argument for sessions on `maxSessions`, and PROTOCOL.md §10.
189-- the same argument for sessions on `maxSessions`, and PROTOCOL.md §10.
190maxMessagesPerRecipient : Int
190maxMessagesPerRecipient : Int
191maxMessagesPerRecipient = 10000
191maxMessagesPerRecipient = 10000
192+-- Global ceiling on total stored messages across all recipients (S6). Every
193+-- other state ref has a global cap (*sessions, *challenges, *presence,
194+-- *forwards); *messages had only a per-recipient cap, so total storage was
195+-- bounded only by (distinct recipient keys x maxMessagesPerRecipient), and an
196+-- attacker minting fresh invalid pubkeys could fill the disk (bounded only by
197+-- rate limits and the 30-day TTL). On overflow this evicts the globally-oldest
198+-- rows, symmetric with the other refs. See PROTOCOL.md §10.
199+maxMessagesTotal : Int
200+maxMessagesTotal = 1000000
192-- Validity window for an auth challenge before it must be re-issued (1 min).
201-- Validity window for an auth challenge before it must be re-issued (1 min).
193challengeExpiry : Int<Ms>
202challengeExpiry : Int<Ms>
194challengeExpiry = 60000
203challengeExpiry = 60000
@@ -957,7 +966,15 @@ appendMessage = \toKey encryptedBlob t -> do
957 let excess = max 0 (count forRecipient - (maxMessagesPerRecipient - 1))
966 let excess = max 0 (count forRecipient - (maxMessagesPerRecipient - 1))
958 let evicted = map (\m -> m.encryptedBlob) (take excess (sortBy (\m -> m.receivedAt) forRecipient))
967 let evicted = map (\m -> m.encryptedBlob) (take excess (sortBy (\m -> m.receivedAt) forRecipient))
959 let kept = filter (\m -> not (m.toKey == toKey && elem m.encryptedBlob evicted)) msgs
968 let kept = filter (\m -> not (m.toKey == toKey && elem m.encryptedBlob evicted)) msgs
969+ -- Global cap (S6): after the per-recipient eviction, bound the total mailbox
970+ -- across all recipients by dropping the globally-oldest rows. Only runs on
971+ -- overflow, so the O(n log n) sort is off the hot path. Dropped rows may
972+ -- belong to other recipients; their orphaned *forwards are reclaimed by
973+ -- `pruneExpired`, and their lost `deliveredTo` is harmless (the blob is gone).
974+ let globallyCapped = if count kept > maxMessagesTotal
975+ then drop (count kept - maxMessagesTotal) (sortBy (\m -> m.receivedAt) kept)
960− *messages = union kept [{toKey, encryptedBlob, receivedAt: seqTs, deliveredTo: []}])
977+ *messages = union globallyCapped [{toKey, encryptedBlob, receivedAt: seqTs, deliveredTo: []}])
962-- Record that `server` has accepted a federated forward of this message, so a
979-- Record that `server` has accepted a federated forward of this message, so a
963-- later presence re-announce of `toKey` doesn't re-forward the same blob to a peer
980-- later presence re-announce of `toKey` doesn't re-forward the same blob to a peer
@@ -1073,56 +1090,70 @@ sendBatchedGossip = \peers events -> case localServerName of
1073-- comes online: any messages already buffered for that pubkey are enqueued
1090-- comes online: any messages already buffered for that pubkey are enqueued
1074-- as forwards to the newly-known server.
1091-- as forwards to the newly-known server.
1076−-- Attempt every forward in `batch` to one peer, in order, and report whether any
1093+-- Attempt every forward in `batch` to one peer, in order, and report both whether
1094+-- any attempt was a genuine peer failure (`anyFailed`) and whether any attempt
1077−-- attempt was a genuine peer failure. Drops successful entries from `*forwards` and
1095+-- genuinely succeeded (`anySucceeded`). Drops successful entries from `*forwards`
1078−-- records the peer on the message's `deliveredTo`; on failure bumps the retry count
1096+-- and records the peer on the message's `deliveredTo`; every attempt (success,
1079−-- and gives up after `maxForwardRetries`.
1097+-- failure, or 404) bumps the retry count and gives up after `maxForwardRetries`.
1081-- INVARIANT — this must not touch `*forwardFailures`. The caller records at most one
1099-- INVARIANT — this must not touch `*forwardFailures`. The caller records at most one
1082−-- failure per peer per sweep, from the Bool returned here; recording per failing
1100+-- failure per peer per sweep, from the `anyFailed` flag returned here; recording per
1083−-- *message* would let one unreachable peer with N buffered messages take N backoff
1101+-- failing *message* would let one unreachable peer with N buffered messages take N
1084−-- steps in a single sweep and pin itself at `retryBackoffMax` for `forwardFailureMaxAge`.
1102+-- backoff steps in a single sweep and pin itself at `retryBackoffMax` for
1085−-- The counter counts consecutive failed *sweeps*.
1103+-- `forwardFailureMaxAge`. The counter counts consecutive failed *sweeps*.
1087-- A 404 is NOT a peer failure. `handleForwardMessage` answers 404 `no_presence`
1105-- A 404 is NOT a peer failure. `handleForwardMessage` answers 404 `no_presence`
1088-- whenever the recipient is not *currently* online at that peer, which is an
1106-- whenever the recipient is not *currently* online at that peer, which is an
1089-- expected, retryable outcome (PROTOCOL.md §7) — the peer is healthy and answering.
1107-- expected, retryable outcome (PROTOCOL.md §7) — the peer is healthy and answering.
1090−-- But `fetch` returns `Err` for every status >= 400, so treating any `Err` as a
1108+-- `fetch` returns `Err` for every status >= 400, so treating any `Err` as a failure
1091−-- failure would feed a routine 404 into `recordFailure`, and `*forwardFailures` is
1109+-- would feed a routine 404 into `recordFailure`, and `*forwardFailures` is shared
1092−-- shared with gossip: one stale-presence 404 would suppress all forwards *and* all
1110+-- with gossip: one stale-presence 404 would suppress all forwards *and* all gossip to
1093−-- gossip to that peer for a full backoff step (30 s–8 min). So a 404 leaves the peer
1111+-- that peer for a full backoff step (30 s–8 min). So a 404 leaves the peer's backoff
1094−-- unpenalized and the forward row untouched — no retry bump either, since the row is
1112+-- untouched (`anyFailed` stays false for it) and never clears it (`anySucceeded`
1095−-- waiting on the recipient's presence, not on a delivery that went wrong.
1113+-- stays false too).
1115+-- But a 404 DOES bump the forward's retry count now (S1). It previously left the row
1116+-- entirely untouched, so a forward whose recipient never came online at the peer was
1117+-- retried every sweep for the full 30-day message TTL — the counter never advanced,
1118+-- so `maxForwardRetries` was never reached. Counting the attempt (without charging
1119+-- the peer's backoff) lets the forward hit the cap and be dropped after
1120+-- `maxForwardRetries` sweeps while still not penalizing a healthy peer.
1097-- Recursive rather than `forEach` because the result has to be accumulated across
1122-- Recursive rather than `forEach` because the result has to be accumulated across
1098-- iterations; `sortBy` fixes the iteration order that `head`/`drop 1` then walk.
1123-- iterations; `sortBy` fixes the iteration order that `head`/`drop 1` then walk.
1099−retryForwardsTo : Text -> [ForwardEntry] -> IO {rw *forwards, rw *messages, network} Bool
1124+retryForwardsTo : Text -> [ForwardEntry] -> IO {rw *forwards, rw *messages, network} {anyFailed: Bool, anySucceeded: Bool}
1100retryForwardsTo = \server batch -> case head batch of
1125retryForwardsTo = \server batch -> case head batch of
1101− Nothing {} -> yield (False {})
1126+ Nothing {} -> yield {anyFailed: False {}, anySucceeded: False {}}
1102 Just {value: f} -> do
1127 Just {value: f} -> do
1103 result <- fetch (federationUrl server) (ForwardMessage {f.toKey, f.encryptedBlob})
1128 result <- fetch (federationUrl server) (ForwardMessage {f.toKey, f.encryptedBlob})
1104 let status = case result of
1129 let status = case result of
1105 Err {error: e} -> e.status
1130 Err {error: e} -> e.status
1106 Ok _ -> 200
1131 Ok _ -> 200
1107 let success = status < 400
1132 let success = status < 400
1108− -- Recipient simply isn't online at the peer yet: hold the row, blame no one.
1133+ -- Recipient simply isn't online at the peer yet: hold the peer blameless, but
1134+ -- still count the attempt so the row can't outlive the message TTL (S1).
1109 let deferred = status == 404
1135 let deferred = status == 404
1110 let isMatch = forwardMatches f.toKey f.encryptedBlob server
1136 let isMatch = forwardMatches f.toKey f.encryptedBlob server
1111 let isDone = \x -> isMatch x && (success || x.retries + 1 >= maxForwardRetries)
1137 let isDone = \x -> isMatch x && (success || x.retries + 1 >= maxForwardRetries)
1112 atomic (do
1138 atomic (do
1113− when (not deferred) (do
1114− cur <- *forwards
1139+ cur <- *forwards
1140+ -- Count the attempt regardless of outcome: a 404 still consumes a retry slot
1141+ -- so a forward that never finds the recipient online is eventually dropped
1142+ -- (S1). On success or after `maxForwardRetries`, `isDone` filters the row out
1143+ -- first, so the increment then no-ops on an absent row.
1115− *forwards = map (\x -> if isMatch x
1144+ *forwards = map (\x -> if isMatch x
1116− then {x.toServer, x.toKey, x.encryptedBlob, retries: x.retries + 1}
1145+ then {x.toServer, x.toKey, x.encryptedBlob, retries: x.retries + 1}
1117− else x) (filter (\x -> not isDone x) cur))
1146+ else x) (filter (\x -> not isDone x) cur)
1118 when success (markDelivered f.toKey f.encryptedBlob server))
1147 when success (markDelivered f.toKey f.encryptedBlob server))
1119− restFailed <- retryForwardsTo server (drop 1 batch)
1148+ rest <- retryForwardsTo server (drop 1 batch)
1120− yield ((not success && not deferred) || restFailed)
1149+ yield {anyFailed: (not success && not deferred) || rest.anyFailed, anySucceeded: success || rest.anySucceeded}
1122-- Walk pending forwards grouped by destination and retry each peer's backlog (up
1151-- Walk pending forwards grouped by destination and retry each peer's backlog (up
1123-- to maxBatchSize per peer) via `/federation/forward`. One failure record per peer
1152-- to maxBatchSize per peer) via `/federation/forward`. One failure record per peer
1124−-- per sweep: a sweep in which every attempt succeeded clears the peer's backoff, a
1125−-- sweep with any failure takes exactly one backoff step (see `retryForwardsTo`).
1153+-- per sweep: a sweep with any genuine failure takes exactly one backoff step; a
1154+-- sweep with at least one genuine success clears the peer's backoff; an all-404
1155+-- sweep leaves the backoff untouched, neither charging nor clearing it (S1). See
1156+-- `retryForwardsTo`.
1126retryPendingForwards : IO {rw *forwardFailures, rw *forwards, rw *messages, network, clock} {}
1157retryPendingForwards : IO {rw *forwardFailures, rw *forwards, rw *messages, network, clock} {}
1127retryPendingForwards = do
1158retryPendingForwards = do
1128 t <- now
1159 t <- now
@@ -1144,13 +1175,17 @@ retryPendingForwards = do
1144 -- `maxForwardRetries` (dropped). `ForwardEntry` carries no timestamp, so key on
1175 -- `maxForwardRetries` (dropped). `ForwardEntry` carries no timestamp, so key on
1145 -- the unique blob.
1176 -- the unique blob.
1146 let batch = take maxBatchSize (sortBy (\f -> f.encryptedBlob) (filter (\f -> f.toServer == server) fwds))
1177 let batch = take maxBatchSize (sortBy (\f -> f.encryptedBlob) (filter (\f -> f.toServer == server) fwds))
1147− anyFailed <- retryForwardsTo server batch
1178+ result <- retryForwardsTo server batch
1148− -- `count batch > 0` guards the clear: `maxBatchSize` is startup-overridable,
1179+ -- Clear the peer's backoff only on a sweep with at least one genuine success
1149− -- and at 0 every batch is empty — an empty sweep must not be read as a
1180+ -- (S1). An all-404 sweep (recipient not online at the peer) is neither a
1150− -- success and reset a failing peer's backoff.
1181+ -- failure nor a success: clearing backoff on it would reset the peer to
1182+ -- "fresh" every sweep, firing a retry every `backgroundPruneInterval` for the
1183+ -- full 30-day message TTL. `count batch > 0` also guards against reading an
1184+ -- empty sweep (possible with a startup-overridden `maxBatchSize` of 0) as
1151− if anyFailed
1186+ if result.anyFailed
1152 then recordFailure server t
1187 then recordFailure server t
1153− else when (count batch > 0) (clearFailure server t)))
1188+ else when (count batch > 0 && result.anySucceeded) (clearFailure server t)))
1155-- Forward queued messages to a server when a remote key comes online.
1190-- Forward queued messages to a server when a remote key comes online.
1156-- Skips messages already delivered to `server` (tracked via Message.deliveredTo)
1191-- Skips messages already delivered to `server` (tracked via Message.deliveredTo)
@@ -1173,6 +1208,32 @@ forwardQueuedToServer = \pubkey serverName -> atomic (do
1173 let newFwds = map (\m -> {toServer: server, m.toKey, m.encryptedBlob, retries: 0}) (take budget (sortBy (\m -> m.receivedAt) candidates))
1208 let newFwds = map (\m -> {toServer: server, m.toKey, m.encryptedBlob, retries: 0}) (take budget (sortBy (\m -> m.receivedAt) candidates))
1174 *forwards = capForwards (union fwds newFwds))
1209 *forwards = capForwards (union fwds newFwds))
1211+-- Batched version of `forwardQueuedToServer`: stage forwards for *all* newly-online
1212+-- keys in a single mailbox scan (S4). Calling `forwardQueuedToServer` per key scans
1213+-- `*messages` once per key, which is why `handleRecvGossip` previously capped the
1214+-- number of keys at `maxNewlyOnlineForwards` (N13) — but that cap silently deferred
1215+-- delivery past the 10th key, and a re-announce within the TTL never recovered it
1216+-- (the `deliveredTo`/presence state made the key look already-handled). A single
1217+-- pass bounds the cost to O(|messages|) regardless of how many keys came online.
1219+-- SECURITY: shares `forwardQueuedToServer`'s exposure — it routes on gossip-written
1220+-- presence. The same `deliveredTo` / `notYetForwarded` / per-peer budget guards
1221+-- apply; do not widen them without revisiting the gossip-redirect note above.
1222+forwardQueuedToAll : [PubkeyHex] -> ServerName -> IO {rw *forwards, r *messages} {}
1223+forwardQueuedToAll = \pubkeys serverName -> atomic (do
1224+ let server = (serverName : Text)
1227+ let existing = countWhere (\f -> f.toServer == server) fwds
1228+ let budget = max 0 (maxForwardsPerServer - existing)
1229+ let notYetForwarded = \m -> not any (forwardMatches m.toKey m.encryptedBlob server) fwds
1230+ let notYetDelivered = \m -> not any (\d -> d.server == server) m.deliveredTo
1231+ -- `elem m.toKey pubkeys` is O(|pubkeys|), but `pubkeys` is bounded by
1232+ -- `maxGossipEvents` (100), so the scan stays O(|messages|) overall.
1233+ let candidates = filter (\m -> elem m.toKey pubkeys && notYetForwarded m && notYetDelivered m) msgs
1234+ let newFwds = map (\m -> {toServer: server, m.toKey, m.encryptedBlob, retries: 0}) (take budget (sortBy (\m -> m.receivedAt) candidates))
1235+ *forwards = capForwards (union fwds newFwds))
1176-- ============================================
1237-- ============================================
1177-- Handler: challenge / verify
1238-- Handler: challenge / verify
1178-- ============================================
1239-- ============================================
@@ -1223,68 +1284,82 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
1223 -- relay server cannot replay the signature to a different server. We verify
1284 -- relay server cannot replay the signature to a different server. We verify
1224 -- against our own configured hostname (port stripped), which a relay cannot forge.
1285 -- against our own configured hostname (port stripped), which a relay cannot forge.
1226− -- Signature verification is deferred until *after* the challenge lookup confirms
1287+ -- Signature verification is split out of the session-install transaction (S7).
1227− -- a matching challenge exists (inside the atomic below). This avoids spending
1288+ -- Phase 1 is a cheap read-only atomic that confirms a matching, unexpired,
1289+ -- IP-bound challenge exists — the original CPU-DoS guard, which still spends no
1228− -- expensive Ed25519 CPU on /auth/verify calls that have no matching challenge,
1290+ -- Ed25519 CPU on /auth/verify calls that have no matching challenge. Phase 2 runs
1229− -- which would otherwise be a cheap CPU-DoS vector. The trade-off is that
1291+ -- the expensive `verify` *outside* any atomic, so an STM retry on the phase-3
1230− -- `verify` runs inside the STM transaction and may re-run on retries — but
1231− -- retries only happen on concurrent writes to *challenges/*sessions, which
1292+ -- install transaction (triggered by concurrent writes to *challenges/*sessions)
1232− -- are rare, so the expected retry count is ~1.
1293+ -- never re-executes it — previously `verify` sat inside that transaction and every
1294+ -- retry re-ran the Ed25519 check. Phase 3 re-checks the challenge before consuming
1295+ -- it, closing the TOCTOU window opened between phases 1 and 3.
1233 let signedText = "skrepka-auth-v1:" ++ authHost serverHost ++ ":" ++ challenge
1296 let signedText = "skrepka-auth-v1:" ++ authHost serverHost ++ ":" ++ challenge
1234− result <- atomic (do
1297+ challengeOk <- atomic (do
1235 ch <- *challenges
1298 ch <- *challenges
1236 let same = \c -> c.pubkey == pubkey && c.challenge == challenge
1299 let same = \c -> c.pubkey == pubkey && c.challenge == challenge
1237− let matchingChallenge = findFirst same ch
1300+ case findFirst ch same of
1238− case matchingChallenge of
1239− Nothing {} -> yield (Nothing {})
1301+ Nothing {} -> yield (False {})
1240− Just {value: c} -> do
1241− if not (c.expiresAt > t && matchIp clientIp c.ip)
1302+ Just {value: c} -> yield (c.expiresAt > t && matchIp clientIp c.ip))
1242− then yield (Nothing {})
1303+ if not challengeOk
1304+ then yield (Err {error: {status: 401, message: "invalid"}})
1243− else do
1305+ else do
1244− -- Verify the signature only now that a matching, unexpired, IP-bound
1306+ -- Phase 2: verify OUTSIDE any atomic. Expensive, and must not be re-executed on
1245− -- challenge exists.
1307+ -- an STM retry of the install transaction below (S7).
1246− let sigValid = case bytesFromHex pubkey of
1308+ let sigValid = case bytesFromHex pubkey of
1247− Just {value: pk} -> case bytesFromHex signature of
1309+ Just {value: pk} -> case bytesFromHex signature of
1248− Just {value: sigBytes} -> verify pk (textToBytes signedText) sigBytes
1310+ Just {value: sigBytes} -> verify pk (textToBytes signedText) sigBytes
1249− _ -> False {}
1311+ _ -> False {}
1250− _ -> False {}
1312+ _ -> False {}
1251− if not sigValid
1313+ if not sigValid
1314+ then yield (Err {error: {status: 401, message: "invalid"}})
1316+ -- Phase 3: re-check the challenge (it may have expired or been consumed by a
1317+ -- concurrent verify since phase 1), consume it, and install the session. The
1318+ -- body is cheap, so an STM retry here costs almost nothing.
1319+ result <- atomic (do
1321+ let same = \c -> c.pubkey == pubkey && c.challenge == challenge
1322+ let matchingChallenge = findFirst ch same
1323+ case matchingChallenge of
1252− then yield (Nothing {})
1324+ Nothing {} -> yield (Nothing {})
1253− else do
1325+ Just {value: c} -> do
1326+ if not (c.expiresAt > t && matchIp clientIp c.ip)
1327+ then yield (Nothing {})
1254− -- Consume the challenge and optionally revoke other sessions + challenges.
1329+ -- Consume the challenge and optionally revoke other sessions + challenges.
1255− -- When `revokeOthers` is set, drop all other sessions AND outstanding
1330+ -- When `revokeOthers` is set, drop all other sessions AND outstanding
1256− -- challenges for this pubkey, so a captured challenge cannot be
1331+ -- challenges for this pubkey, so a captured challenge cannot be
1257− -- redeemed for a new session after revocation (Q35).
1332+ -- redeemed for a new session after revocation (Q35).
1258− *challenges = if revokeOthers
1333+ *challenges = if revokeOthers
1259− then filter (\x -> not (x.pubkey == pubkey)) ch
1334+ then filter (\x -> not (x.pubkey == pubkey)) ch
1260− else filter (\x -> not same x) ch
1335+ else filter (\x -> not same x) ch
1261− ses <- *sessions
1336+ ses <- *sessions
1262− let pruned = if revokeOthers
1337+ let pruned = if revokeOthers
1263− then filter (\s -> s.pubkey != pubkey) ses
1338+ then filter (\s -> s.pubkey != pubkey) ses
1264− else capPerKey ses pubkey maxSessionsPerKey
1339+ else capPerKey ses pubkey maxSessionsPerKey
1265− let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp}
1340+ let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp}
1266− -- LRU-evicts the soonest-to-expire session when the pool is full; this global
1341+ -- LRU-evicts the soonest-to-expire session when the pool is full; this global
1267− -- eviction is Sybil-floodable, see the LIMITATION note on `maxSessions`.
1342+ -- eviction is Sybil-floodable, see the LIMITATION note on `maxSessions`.
1268− let pooled = union pruned [newSession]
1343+ let pooled = union pruned [newSession]
1269− -- Count what `capList` is about to shed, so the handler can warn about it
1344+ -- Count what `capList` is about to shed, so the handler can warn about it
1270− -- below. That flood has no other observable signature: evicted users just
1345+ -- below. That flood has no other observable signature: evicted users just
1271− -- start seeing 401s and silently re-authenticate, which looks like normal
1346+ -- start seeing 401s and silently re-authenticate, which looks like normal
1272− -- traffic in the access log. A sustained nonzero rate here is the alert.
1347+ -- traffic in the access log. A sustained nonzero rate here is the alert.
1273− let evicted = max 0 (count pooled - maxSessions)
1348+ let evicted = max 0 (count pooled - maxSessions)
1274− *sessions = capList pooled maxSessions
1349+ *sessions = capList pooled maxSessions
1275− peers <- refreshLocalPresence pubkey expiry
1350+ peers <- refreshLocalPresence pubkey expiry
1276− yield (Just {value: {peers, evicted}}))
1351+ yield (Just {value: {peers, evicted}}))
1277− case result of
1352+ case result of
1278− Nothing {} -> yield (Err {error: {status: 401, message: "invalid"}})
1353+ Nothing {} -> yield (Err {error: {status: 401, message: "invalid"}})
1279− Just {value: r} -> do
1354+ Just {value: r} -> do
1280− -- Warn outside the `atomic` block, never inside: an STM transaction can
1355+ -- Warn outside the `atomic` block, never inside: an STM transaction can
1281− -- retry, and a console write is not rolled back with it — logging inside
1356+ -- retry, and a console write is not rolled back with it — logging inside
1282− -- would emit one line per retry attempt rather than one per eviction.
1357+ -- would emit one line per retry attempt rather than one per eviction.
1283− -- Include the requesting pubkey and IP for audit (Q38): an operator
1358+ -- Include the requesting pubkey and IP for audit (Q38): an operator
1284− -- investigating a Sybil flood needs to know which identities are responsible.
1359+ -- investigating a Sybil flood needs to know which identities are responsible.
1285− when (r.evicted > 0) (logWarn ("session pool full: evicted " ++ show r.evicted ++ " session(s) to seat a new login (pubkey=" ++ pubkey ++ ", ip=" ++ clientIp ++ ", maxSessions=" ++ show maxSessions ++ ")"))
1360+ when (r.evicted > 0) (logWarn ("session pool full: evicted " ++ show r.evicted ++ " session(s) to seat a new login (pubkey=" ++ pubkey ++ ", ip=" ++ clientIp ++ ", maxSessions=" ++ show maxSessions ++ ")"))
1286− when federationEnabled (fork (sendBatchedGossip r.peers [{eventType: "online", pubkey: Just {value: pubkey}}]))
1361+ when federationEnabled (fork (sendBatchedGossip r.peers [{eventType: "online", pubkey: Just {value: pubkey}}]))
1287− yield (Ok {value: {token, expiresAt: expiry}})
1362+ yield (Ok {value: {token, expiresAt: expiry}})
1289-- ============================================
1364-- ============================================
1290-- Handler: poll
1365-- Handler: poll
@@ -1438,10 +1513,15 @@ handleSendMessages = \messages authorization clientIp -> do
1438 Nothing {} -> do
1513 Nothing {} -> do
1439 committed <- atomic (do
1514 committed <- atomic (do
1440 pres <- *presence
1515 pres <- *presence
1516+ -- S2: cap the immediate fan-out at `maxForwardsPerServer` servers per
1517+ -- recipient. The post-commit `forkForwardsTo` fires against this list
1518+ -- directly (bypassing the `*forwards` queue and its per-peer cap), so
1519+ -- without a bound here a recipient gossiped as present on thousands of
1520+ -- servers would spawn one immediate HTTP forward per server per message.
1441 let staged = map (\m -> {
1521 let staged = map (\m -> {
1443 m.encryptedBlob,
1523 m.encryptedBlob,
1444− remoteServers: collectPeers (filter (\p -> p.pubkey == m.to && p.expiresAt > t) pres) (\p -> p.server)
1524+ remoteServers: take maxForwardsPerServer (collectPeers (filter (\p -> p.pubkey == m.to && p.expiresAt > t) pres) (\p -> p.server))
1445 }) messages
1525 }) messages
1446 forEach staged (\c -> do
1526 forEach staged (\c -> do
1447 -- Read `*messages` *before* the append: `delivered` below must see the
1527 -- Read `*messages` *before* the append: `delivered` below must see the
@@ -1546,7 +1626,7 @@ forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> when (
1546-- fromServer) row touched by this batch, then re-adds fresh-TTL rows for
1626-- fromServer) row touched by this batch, then re-adds fresh-TTL rows for
1547-- pubkeys with an `online` event (so Online wins on collision). For keys
1627-- pubkeys with an `online` event (so Online wins on collision). For keys
1548-- that transition offline → online, flushes any locally-buffered messages via
1628-- that transition offline → online, flushes any locally-buffered messages via
1549−-- `forwardQueuedToServer`. Events with an unrecognized `eventType` or no
1629+-- `forwardQueuedToAll`. Events with an unrecognized `eventType` or no
1550-- `pubkey` are skipped (forward-compat) rather than rejecting the batch.
1630-- `pubkey` are skipped (forward-compat) rather than rejecting the batch.
1552-- SECURITY: this is the gossip-redirect ingress described in the section header
1632-- SECURITY: this is the gossip-redirect ingress described in the section header
@@ -1557,7 +1637,7 @@ handleRecvGossip : [GossipEvent] -> ServerName -> IO {r *messages, rw *forwards,
1557handleRecvGossip = \events fromServer -> do
1637handleRecvGossip = \events fromServer -> do
1558 t <- now
1638 t <- now
1559 -- Widen the peer-supplied `ServerName` once (see `type ServerName`);
1639 -- Widen the peer-supplied `ServerName` once (see `type ServerName`);
1560− -- `forwardQueuedToServer` below still takes the refined name.
1640+ -- `forwardQueuedToAll` below still takes the refined name.
1561 let from = (fromServer : Text)
1641 let from = (fromServer : Text)
1562 if not federationEnabled
1642 if not federationEnabled
1563 then yield (Err {error: {status: 403, message: "federation_disabled"}})
1643 then yield (Err {error: {status: 403, message: "federation_disabled"}})
@@ -1598,21 +1678,28 @@ handleRecvGossip = \events fromServer -> do
1598 let localRows = filter (\p -> isLocalServer p.server) merged
1678 let localRows = filter (\p -> isLocalServer p.server) merged
1599 let remoteRows = filter (\p -> not isLocalServer p.server) merged
1679 let remoteRows = filter (\p -> not isLocalServer p.server) merged
1600 *presence = union localRows (capList (capPerServer remoteRows from maxPresencePerServer) maxPresence)
1680 *presence = union localRows (capList (capPerServer remoteRows from maxPresencePerServer) maxPresence)
1681+ -- S8: Make online+offline for the same key mutually exclusive within one
1682+ -- batch. If a batch carries both events for a key, online wins — the presence
1683+ -- merge above already added the online row — so that key must NOT also travel
1684+ -- the offline path and cancel its own freshly-staged forwards.
1685+ let onlineKeys = map (\p -> p.pubkey) collected.added
1686+ let offlineOnly = filter (\pk -> not elem pk onlineKeys) collected.offline
1601 -- N14: Cancel pending forwards to `from` for offline keys. Without this,
1687 -- N14: Cancel pending forwards to `from` for offline keys. Without this,
1602 -- forwards to a peer that just went offline retry for up to 80 minutes,
1688 -- forwards to a peer that just went offline retry for up to 80 minutes,
1603 -- wasting maxForwardsPerServer slots and generating failed HTTP requests.
1689 -- wasting maxForwardsPerServer slots and generating failed HTTP requests.
1604 -- The message stays in *messages, so it's re-forwarded if the key
1690 -- The message stays in *messages, so it's re-forwarded if the key
1605 -- re-appears on another peer.
1691 -- re-appears on another peer.
1606− when (count collected.offline > 0) (do
1692+ when (count offlineOnly > 0) (do
1607 fwds <- *forwards
1693 fwds <- *forwards
1608− *forwards = filter (\f -> not (f.toServer == from && elem f.toKey collected.offline)) fwds)
1694+ *forwards = filter (\f -> not (f.toServer == from && elem f.toKey offlineOnly)) fwds)
1609 yield (filter (\pk -> not any (\p -> p.pubkey == pk && p.server == from && p.expiresAt > t) pres) (map (\p -> p.pubkey) collected.added)))
1695 yield (filter (\pk -> not any (\p -> p.pubkey == pk && p.server == from && p.expiresAt > t) pres) (map (\p -> p.pubkey) collected.added)))
1610− -- N13: Cap the number of newly-online keys that trigger forward sweeps.
1696+ -- S4: forward queued mail for ALL newly-online keys in one mailbox pass. The
1611− -- Without this, 100 online events = 100 mailbox scans. The remaining keys
1697+ -- old per-key cap (maxNewlyOnlineForwards, N13) bounded amplification but
1612− -- are still in `*presence` — their forwards will be staged when a poll
1698+ -- silently deferred delivery past the 10th key, and a re-announce within the
1613− -- refreshes presence or the next gossip re-announces them.
1699+ -- TTL never recovered it (the deliveredTo/presence state made the key look
1614− let toForward = take maxNewlyOnlineForwards newlyOnline
1700+ -- already-handled). `forwardQueuedToAll` bounds the cost to O(|messages|)
1615− fork (forEach toForward (\pk -> forwardQueuedToServer pk fromServer))
1701+ -- regardless of how many keys came online.
1702+ fork (forwardQueuedToAll newlyOnline fromServer)
1616 yield (Ok {value: {}})
1703 yield (Ok {value: {}})
1618-- POST /federation/forward: receive a single federated message destined for
1705-- POST /federation/forward: receive a single federated message destined for