@@ -129,7 +129,7 @@ type ValidGossipEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Times
129129-- Status returned to the sender after SendMessage
130130-- Delivered: recipient is online locally; Federated: forwarded to remote server;
131131-- Queued: stored but recipient location unknown; Unauthorized/Rejected: not accepted
132−data MessageStatus = MsDelivered {} | MsFederated {} | MsQueued {} | MsUnauthorized {} | MsRejected {}
132+data MessageStatus = MsDelivered {} | MsFederated {} | MsQueued {} | MsUnauthorized {} | MsRejected {} | MsRateLimited {}
133133
134134impl ToJSON MessageStatus where
135135 toJson ms = toJson (case ms of
@@ -137,7 +137,8 @@ impl ToJSON MessageStatus where
137137 MsFederated {} -> "federated"
138138 MsQueued {} -> "queued"
139139 MsUnauthorized {} -> "unauthorized"
140− MsRejected {} -> "rejected")
140+ MsRejected {} -> "rejected"
141+ MsRateLimited {} -> "rate_limited")
141142
142143-- Internal delivery status (same as MessageStatus but without error variants)
143144data DeliveryStatus = DsDelivered {} | DsFederated {} | DsQueued {}
@@ -318,7 +319,7 @@ route HealthApi where
318319-- Returns counts of in-memory collections (requires federation or maintenance secret)
319320route StatsApi where
320321 GET /stats headers {authorization: Text}
321− -> {sessions: Int, messages: Int, presence: Int, forwards: Int, challenges: Int, ackedIds: Int, ok: Bool}
322+ -> {sessions: Int, messages: Int, presence: Int, forwards: Int, challenges: Int, ackedIds: Int, ackOverflow: Int, deliveredIds: Int, forwardFailures: Int, activePolls: Int, ok: Bool}
322323 = Stats
323324
324325route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi | HealthApi | StatsApi
@@ -627,6 +628,20 @@ safeCompare = \a b ->
627628 + (if a == "" then 1 else 0)
628629 + (if b == "" then 1 else 0) == 0
629630
631+-- Generic sliding-window rate limiter. Returns whether the request is rate-limited
632+-- and the updated entries list. Callers provide match/create/update functions for their entry type.
633+checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntries t ->
634+ let entry = headOf (do r <- entries; where matchKey r; yield r) in
635+ let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= windowMs in
636+ let rateCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count in
637+ if rateCount >= maxRate
638+ then {limited: True {}, entries: entries}
639+ else
640+ let newEntries = if isNewWindow
641+ then union (do r <- entries; where not (matchKey r); yield r) [makeNew t]
642+ else do r <- entries; yield (if matchKey r then updateCount r else r)
643+ in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs; yield r) else newEntries}
644+
630645-- Remove the single item with the smallest key value (LRU-style eviction)
631646evictOneOldest = \items getKey ->
632647 let result = fold (\acc x ->
@@ -859,17 +874,12 @@ maybePruneAndBroadcast = \t -> do
859874
860875checkFederationRate = \server t -> atomic do
861876 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 {})
877+ let result = checkRateWindow federationRates (\r -> r.server == server)
878+ (\ts -> {server: server, windowStart: ts, count: 1})
879+ (\r -> {server: r.server, windowStart: r.windowStart, count: r.count + 1})
880+ maxFederationRatePerServer federationRateWindowMs maxFederationRateEntries t
881+ set *federationRates = result.entries
882+ yield result.limited
873883
874884checkGlobalRate = \t -> atomic do
875885 windowStart <- *globalRateWindowStart
@@ -1000,17 +1010,24 @@ sendBatchedAcksToServer = \serverAcks server ->
10001010 forEachBatch serverAcks maxMessageIds (\batch -> do
10011011 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
10021012 case result of
1003− Err _ -> println ("federation: ack to " ++ server ++ " failed")
1004− Ok _ -> atomic do
1005− ackedIds <- *ackedIds
1006− let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch
1007− set *ackedIds = updatedAckedIds)
1013+ Err _ -> do
1014+ println ("federation: ack to " ++ server ++ " failed")
1015+ recordFailure server
1016+ Ok _ -> do
1017+ clearFailure server
1018+ atomic do
1019+ ackedIds <- *ackedIds
1020+ let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch
1021+ set *ackedIds = updatedAckedIds)
10081022
10091023-- Send delivery acknowledgments back to origin servers and remove acked entries on success
10101024sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
1011−sendAcksAndClean = \pendingAcks ->
1012− let servers = collectPeerServers pendingAcks (\a -> a.originServer) in
1013− sendToEach servers (\server ->
1025+sendAcksAndClean = \pendingAcks -> do
1026+ let servers = collectPeerServers pendingAcks (\a -> a.originServer)
1027+ t <- now
1028+ failures <- *forwardFailures
1029+ let eligibleServers = do s <- servers; where shouldRetryServer s t failures; yield s
1030+ sendToEach eligibleServers (\server ->
10141031 sendBatchedAcksToServer (acksForServer pendingAcks server) server)
10151032
10161033-- Retry sending acks that were evicted from ackedIds due to capacity limits.
@@ -1057,16 +1074,23 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
10571074 Ok _ -> yield {})
10581075
10591076-- Retry acks sitting in the overflow buffer (last chance before they become deliveredIds)
1060−retryOverflowAcks = \overflow ->
1061− let servers = collectPeerServers overflow (\a -> a.originServer) in
1062− sendToEach servers (\server ->
1077+retryOverflowAcks = \overflow -> do
1078+ let servers = collectPeerServers overflow (\a -> a.originServer)
1079+ t <- now
1080+ failures <- *forwardFailures
1081+ let eligibleServers = do s <- servers; where shouldRetryServer s t failures; yield s
1082+ sendToEach eligibleServers (\server ->
10631083 forEachBatch (acksForServer overflow server) maxMessageIds (\batch -> do
10641084 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
10651085 case result of
1066− Err _ -> println ("federation: overflow ack to " ++ server ++ " failed")
1067− Ok _ -> atomic do
1068− ackOverflow <- *ackOverflow
1069− set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))
1086+ Err _ -> do
1087+ println ("federation: overflow ack to " ++ server ++ " failed")
1088+ recordFailure server
1089+ Ok _ -> do
1090+ clearFailure server
1091+ atomic do
1092+ ackOverflow <- *ackOverflow
1093+ set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))
10701094
10711095-- Full maintenance cycle: prune expired state, retry pending forwards and acks,
10721096-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
@@ -1391,16 +1415,14 @@ handleChallenge = \pubkey respond -> do
13911415 then yield {pr: pr, accepted: False {}, rateLimited: False {}}
13921416 else do
13931417 challengeRates <- *challengeRates
1394− let senderEntry = headOf (do r <- challengeRates; where r.pubkey == pubkey; yield r)
1395− let isNewWindow = case senderEntry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= challengeRateWindowMs
1396− let rateCount = case senderEntry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count
1397− if rateCount >= maxChallengeRatePerKey
1418+ let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
1419+ (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
1420+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1421+ maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
1422+ if rateResult.limited
13981423 then yield {pr: pr, accepted: False {}, rateLimited: True {}}
13991424 else do
1400− let newRates = if isNewWindow
1401− then union (do r <- challengeRates; where r.pubkey != pubkey; yield r) [{pubkey: pubkey, windowStart: t, count: 1}]
1402− else do r <- challengeRates; yield (if r.pubkey == pubkey then {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1} else r)
1403− set *challengeRates = if count newRates > maxChallengeRateEntries then (do r <- newRates; where t - r.windowStart < challengeRateWindowMs; yield r) else newRates
1425+ set *challengeRates = rateResult.entries
14041426 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
14051427 yield {pr: pr, accepted: True {}, rateLimited: False {}}
14061428 when result.rateLimited
@@ -1412,8 +1434,26 @@ handleChallenge = \pubkey respond -> do
14121434
14131435-- POST /auth/verify: Verify Ed25519 signature over challenge string.
14141436-- On success: creates a session, registers local presence, and gossips online status.
1437+-- Rate-limited globally and per-key (shares auth rate window with challenge requests).
14151438handleVerifyAuth = \pubkey challenge signature respond -> do
14161439 t <- now
1440+ globalRateLimited <- checkGlobalRate t
1441+ if globalRateLimited then yield (respond {token: "", expiresAt: 0})
1442+ else do
1443+ authRateLimited <- atomic do
1444+ challengeRates <- *challengeRates
1445+ let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
1446+ (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
1447+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1448+ maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
1449+ set *challengeRates = rateResult.entries
1450+ yield rateResult.limited
1451+ if authRateLimited then do
1452+ println ("verify: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey)
1453+ yield (respond {token: "", expiresAt: 0})
1454+ else verifyAuthInner pubkey challenge signature respond t
1455+
1456+verifyAuthInner = \pubkey challenge signature respond t -> do
14171457 let expiry = t + sessionExpiry
14181458 let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
14191459 validation <- atomic do
@@ -1457,15 +1497,19 @@ handleVerifyAuth = \pubkey challenge signature respond -> do
14571497-- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.
14581498-- Acks for federated messages are sent to origin servers in the background.
14591499handlePoll = \cursor authorization respond -> do
1460− result <- case extractToken authorization of
1461− Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
1462− Just {value: token} -> pollOnce token cursor
1463− tAfterPoll <- now
1464− maybePruneAndBroadcast tAfterPoll
1465− fork (when federationEnabled (do
1466− sendAcksAndClean result.pendingAcks
1467− retryEvictedAcks result.evictedEntries result.pendingAcks))
1468− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
1500+ t <- now
1501+ globalRateLimited <- checkGlobalRate t
1502+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}})
1503+ else do
1504+ result <- case extractToken authorization of
1505+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
1506+ Just {value: token} -> pollOnce token cursor
1507+ tAfterPoll <- now
1508+ maybePruneAndBroadcast tAfterPoll
1509+ fork (when federationEnabled (do
1510+ sendAcksAndClean result.pendingAcks
1511+ retryEvictedAcks result.evictedEntries result.pendingAcks))
1512+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
14691513
14701514-- POST /messages: Send an encrypted message to a recipient.
14711515-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
@@ -1473,7 +1517,7 @@ handlePoll = \cursor authorization respond -> do
14731517handleSendMessage = \to encryptedBlob timestamp messageId authorization respond -> do
14741518 t <- now
14751519 globalRateLimited <- checkGlobalRate t
1476− if globalRateLimited then yield (respond {status: MsRejected {}, messageId: ""})
1520+ if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: ""})
14771521 else do
14781522 effectiveMsgId <- if messageId != "" then yield messageId else genToken
14791523 let clampedTs = clampTimestamp timestamp t
@@ -1489,16 +1533,14 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
14891533 then yield {pr: pr, send: SendDuplicate {}}
14901534 else do
14911535 sendRates <- *sendRates
1492− let senderEntry = headOf (do r <- sendRates; where r.pubkey == senderKey; yield r)
1493− let isNewWindow = case senderEntry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= sendRateWindowMs
1494− let rateCount = case senderEntry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count
1495− if rateCount >= maxSendRatePerKey
1496− then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
1536+ let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
1537+ (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})
1538+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1539+ maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
1540+ if rateResult.limited
1541+ then yield {pr: pr, send: SendFailed {status: MsRateLimited {}}}
14971542 else do
1498− let newRates = if isNewWindow
1499− then union (do r <- sendRates; where r.pubkey != senderKey; yield r) [{pubkey: senderKey, windowStart: t, count: 1}]
1500− else do r <- sendRates; yield (if r.pubkey == senderKey then {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1} else r)
1501− set *sendRates = if count newRates > maxSendRateEntries then (do r <- newRates; where t - r.windowStart < sendRateWindowMs; yield r) else newRates
1543+ set *sendRates = rateResult.entries
15021544 if countMessagesFor messages to t >= maxMessagesPerRecipient
15031545 then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
15041546 else do
@@ -1516,24 +1558,29 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
15161558 let delivery = if localOnline then DsDelivered {}
15171559 else if federationEnabled && count remoteServers > 0 then DsFederated {}
15181560 else DsQueued {}
1561+ -- Create forward entries atomically with message storage (prevents race with ack)
1562+ when (federationEnabled && count remoteServers > 0) (do
1563+ forwards <- *forwards
1564+ let newFwds = do s <- remoteServers; yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}
1565+ let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
1566+ set *forwards = union forwards cappedFwds)
15191567 yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers}})
15201568 case result.send of
15211569 SendDone {delivery, remoteServers} ->
15221570 fork (do
15231571 when (count remoteServers > 0 && federationEnabled) (
15241572 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}])
15301573 fwdResult <- fetch (federationUrl server) (ForwardMessage {
15311574 toKey: to, encryptedBlob: encryptedBlob,
15321575 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
15331576 fromServer: serverHostN, authorization: federationAuth})
15341577 case fwdResult of
1535− Err _ -> println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)
1536− Ok _ -> removeForward effectiveMsgId server))
1578+ Err _ -> do
1579+ println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)
1580+ recordFailure server
1581+ Ok _ -> do
1582+ removeForward effectiveMsgId server
1583+ clearFailure server))
15371584 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
15381585 _ -> forkBroadcastOffline t result.pr
15391586 yield (case result.send of
@@ -1545,16 +1592,19 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
15451592-- Returns local server if key is online locally, otherwise the remote with latest expiry.
15461593handleLookupPresence = \pubkey authorization respond -> do
15471594 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})
1595+ globalRateLimited <- checkGlobalRate t
1596+ if globalRateLimited then yield (respond {server: "", authorized: False {}})
1597+ else do
1598+ result <- atomic do
1599+ pr <- maybePrune t
1600+ withSessionAuth authorization t
1601+ (yield {pr: pr, server: "", authorized: False {}})
1602+ (\_ -> do
1603+ presence <- *presence
1604+ let foundServer = bestServerForKey (presenceForKey presence pubkey t)
1605+ yield {pr: pr, server: foundServer, authorized: True {}})
1606+ forkBroadcastOffline t result.pr
1607+ yield (respond {server: result.server, authorized: result.authorized})
15581608
15591609-- POST /federation/gossip: Receive presence gossip from a peer server.
15601610-- Validates events, merges into presence table (last-write-wins), and re-gossips
@@ -1668,9 +1718,13 @@ handleStats = \authorization respond -> do
16681718 f <- *forwards
16691719 c <- *challenges
16701720 a <- *ackedIds
1671− yield {sessions: count s, messages: count m, presence: count p, forwards: count f, challenges: count c, ackedIds: count a}
1672− yield (respond {sessions: result.sessions, messages: result.messages, presence: result.presence, forwards: result.forwards, challenges: result.challenges, ackedIds: result.ackedIds, ok: True {}})
1673− else yield (respond {sessions: 0, messages: 0, presence: 0, forwards: 0, challenges: 0, ackedIds: 0, ok: False {}})
1721+ ao <- *ackOverflow
1722+ d <- *deliveredIds
1723+ ff <- *forwardFailures
1724+ ap <- *activePolls
1725+ yield {sessions: count s, messages: count m, presence: count p, forwards: count f, challenges: count c, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff, activePolls: count ap}
1726+ yield (respond {sessions: result.sessions, messages: result.messages, presence: result.presence, forwards: result.forwards, challenges: result.challenges, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, ok: True {}})
1727+ else yield (respond {sessions: 0, messages: 0, presence: 0, forwards: 0, challenges: 0, ackedIds: 0, ackOverflow: 0, deliveredIds: 0, forwardFailures: 0, activePolls: 0, ok: False {}})
16741728
16751729serve = \req -> case req of
16761730 ReqChallenge {pubkey, respond} -> handleChallenge pubkey respond
@@ -1710,6 +1764,9 @@ pollHeartbeat = do
17101764
17111765main = do
17121766 println "Skrepka relay server starting on port 8080..."
1767+ println (" federation: " ++ if federationEnabled then "enabled" else "disabled")
1768+ println (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
1769+ println (" poll timeout: " ++ show pollTimeoutMs ++ "ms")
17131770 fork backgroundPrune
17141771 fork pollHeartbeat
17151772 listen 8080 serve