@@ -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 {} | MsRateLimited {}
132+data MessageStatus = MsDelivered {} | MsFederated {} | MsQueued {} | MsUnauthorized {} | MsRejected {} | MsRateLimited {} | MsDuplicate {}
133133
134134impl ToJSON MessageStatus where
135135 toJson ms = toJson (case ms of
@@ -138,7 +138,8 @@ impl ToJSON MessageStatus where
138138 MsQueued {} -> "queued"
139139 MsUnauthorized {} -> "unauthorized"
140140 MsRejected {} -> "rejected"
141− MsRateLimited {} -> "rate_limited")
141+ MsRateLimited {} -> "rate_limited"
142+ MsDuplicate {} -> "duplicate")
142143
143144-- Internal delivery status (same as MessageStatus but without error variants)
144145data DeliveryStatus = DsDelivered {} | DsFederated {} | DsQueued {}
@@ -258,6 +259,11 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
258259
259260*globalRateWindowStart : Timestamp
260261
262+-- Background task cycle counters (exposed via /stats for health monitoring)
263+*pruneCount : Int
264+
265+*heartbeatCount : Int
266+
261267-- ============================================
262268-- Routes
263269-- ============================================
@@ -312,6 +318,9 @@ route MaintenanceApi where
312318 POST {} /maintenance/prune headers {authorization: Text}
313319 -> {ok: Bool}
314320 = Prune
321+ POST {} /maintenance/drain headers {authorization: Text}
322+ -> {ok: Bool, drained: Int}
323+ = Drain
315324
316325route HealthApi where
317326 GET /health -> {ok: Bool} = Health
@@ -319,7 +328,7 @@ route HealthApi where
319328-- Returns counts of in-memory collections (requires federation or maintenance secret)
320329route StatsApi where
321330 GET /stats headers {authorization: Text}
322− -> {sessions: Int, messages: Int, presence: Int, forwards: Int, challenges: Int, ackedIds: Int, ackOverflow: Int, deliveredIds: Int, forwardFailures: Int, activePolls: Int, ok: Bool}
331+ -> {sessions: Int, messages: Int, presence: Int, forwards: Int, challenges: Int, ackedIds: Int, ackOverflow: Int, deliveredIds: Int, forwardFailures: Int, activePolls: Int, sendRates: Int, challengeRates: Int, federationRates: Int, pruneCount: Int, heartbeatCount: Int, ok: Bool}
323332 = Stats
324333
325334route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi | HealthApi | StatsApi
@@ -901,13 +910,20 @@ withFederationAuth = \authorization t respond fromServer precondition action ->
901910 maybePruneAndBroadcast t
902911 yield (respond {ok: False {}})
903912 else do
904− rateLimited <- checkFederationRate fromServer t
905− if rateLimited
913+ globalLimited <- checkGlobalRate t
914+ if globalLimited
906915 then do
907− println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
916+ println ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)
908917 maybePruneAndBroadcast t
909918 yield (respond {ok: False {}})
910− else action
919+ else do
920+ rateLimited <- checkFederationRate fromServer t
921+ if rateLimited
922+ then do
923+ println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
924+ maybePruneAndBroadcast t
925+ yield (respond {ok: False {}})
926+ else action
911927
912928withSessionAuth = \authorization t onUnauth onAuth -> do
913929 sessions <- *sessions
@@ -1458,16 +1474,13 @@ verifyAuthInner = \pubkey challenge signature respond t -> do
14581474 let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
14591475 validation <- atomic do
14601476 pr <- maybePrune t
1461− if not sigValid
1477+ challenges <- *challenges
1478+ let hasMatch = challengeMatch challenges pubkey challenge t
1479+ -- Consume challenge on any verify attempt (prevents signature brute-force)
1480+ when hasMatch (set *challenges = challengesExcluding challenges pubkey challenge)
1481+ if not hasMatch || not sigValid
14621482 then yield {pr: pr, valid: False {}}
1463− else do
1464− challenges <- *challenges
1465− let hasMatch = challengeMatch challenges pubkey challenge t
1466− if not hasMatch
1467− then yield {pr: pr, valid: False {}}
1468− else do
1469− set *challenges = challengesExcluding challenges pubkey challenge
1470− yield {pr: pr, valid: True {}}
1483+ else yield {pr: pr, valid: True {}}
14711484 if validation.valid
14721485 then do
14731486 token <- genToken
@@ -1561,7 +1574,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
15611574 -- Create forward entries atomically with message storage (prevents race with ack)
15621575 when (federationEnabled && count remoteServers > 0) (do
15631576 forwards <- *forwards
1564− let newFwds = do s <- remoteServers; yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}
1577+ let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}
15651578 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
15661579 set *forwards = union forwards cappedFwds)
15671580 yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers}})
@@ -1585,7 +1598,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
15851598 _ -> forkBroadcastOffline t result.pr
15861599 yield (case result.send of
15871600 SendFailed {status} -> respond {status: status, messageId: ""}
1588− SendDuplicate {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId}
1601+ SendDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId}
15891602 SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})
15901603
15911604-- GET /lookup/:pubkey: Look up the best known server for a pubkey.
@@ -1697,6 +1710,26 @@ handleFederationAck = \messageIds originServer fromServer authorization respond
16971710 forkBroadcastOffline t pruneResult
16981711 yield (respond {ok: True {}}))
16991712
1713+-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
1714+-- Call before stopping the server so peers learn keys went offline immediately.
1715+-- Gossip is sent via fork; allow a few seconds after response before process exit.
1716+handleDrain = \authorization respond -> do
1717+ t <- now
1718+ let authorized = verifyMaintenanceAuth authorization
1719+ if not authorized then yield (respond {ok: False {}, drained: 0})
1720+ else do
1721+ drainResult <- atomic do
1722+ presence <- *presence
1723+ let localEntries = do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p
1724+ set *presence = do p <- presence; where not (isLocalServer p.server) || p.expiresAt <= t; yield p
1725+ yield {localEntries: localEntries, peerServers: collectPeerServers presence (\p -> p.server)}
1726+ let drainedCount = count drainResult.localEntries
1727+ when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
1728+ let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
1729+ sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
1730+ println ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
1731+ yield (respond {ok: True {}, drained: drainedCount})
1732+
17001733-- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).
17011734-- Accepts either federation or maintenance secret.
17021735handlePrune = \authorization respond -> do
@@ -1722,9 +1755,14 @@ handleStats = \authorization respond -> do
17221755 d <- *deliveredIds
17231756 ff <- *forwardFailures
17241757 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 {}})
1758+ sr <- *sendRates
1759+ cr <- *challengeRates
1760+ fr <- *federationRates
1761+ pc <- *pruneCount
1762+ hc <- *heartbeatCount
1763+ 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, sendRates: count sr, challengeRates: count cr, federationRates: count fr, pruneCount: pc, heartbeatCount: hc}
1764+ 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, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, ok: True {}})
1765+ else yield (respond {sessions: 0, messages: 0, presence: 0, forwards: 0, challenges: 0, ackedIds: 0, ackOverflow: 0, deliveredIds: 0, forwardFailures: 0, activePolls: 0, sendRates: 0, challengeRates: 0, federationRates: 0, pruneCount: 0, heartbeatCount: 0, ok: False {}})
17281766
17291767serve = \req -> case req of
17301768 ReqChallenge {pubkey, respond} -> handleChallenge pubkey respond
@@ -1736,6 +1774,7 @@ serve = \req -> case req of
17361774 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer authorization respond
17371775 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> handleFederationAck messageIds originServer fromServer authorization respond
17381776 Prune {authorization, respond} -> handlePrune authorization respond
1777+ Drain {authorization, respond} -> handleDrain authorization respond
17391778 Health {respond} -> handleHealth respond
17401779 Stats {authorization, respond} -> handleStats authorization respond
17411780
@@ -1749,6 +1788,9 @@ backgroundPrune = do
17491788 jitter <- randomInt 5000
17501789 sleep (backgroundPruneInterval + jitter)
17511790 t <- now
1791+ atomic do
1792+ pc <- *pruneCount
1793+ set *pruneCount = pc + 1
17521794 fork (pruneAndRetry t)
17531795 backgroundPrune
17541796
@@ -1760,6 +1802,8 @@ pollHeartbeat = do
17601802 t <- now
17611803 atomic do
17621804 set *pollWakeup = t
1805+ hc <- *heartbeatCount
1806+ set *heartbeatCount = hc + 1
17631807 pollHeartbeat
17641808
17651809main = do