@@ -198,7 +198,7 @@ data DeliveryStatus = DsDelivered {} | DsFederated {} | DsQueued {}
198198
199199-- Validated hex-encoded types with maximum length constraints
200200isValidBoundedHex : Int -> Text -> Bool
201−isValidBoundedHex = \maxLen s -> isValidHex s && not (tooLong s maxLen) && s == normalizeHex s
201+isValidBoundedHex = \maxLen s -> not (tooLong s maxLen) && isValidHex s && s == normalizeHex s
202202type PubkeyHex = Text where isValidBoundedHex maxPubkeyLen
203203type BlobHex = Text where isValidBoundedHex maxBlobLen
204204type SignatureHex = Text where isValidBoundedHex maxSignatureLen
@@ -472,7 +472,7 @@ route HealthApi where
472472-- Returns counts of in-memory collections (requires federation or maintenance secret)
473473route StatsApi where
474474 GET /stats headers {authorization: Text}
475− -> {sessions: Int, sessionsMax: Int, messages: Int, messagesMax: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pruneCount: Int, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, federationEnabled: Bool, recentGossip: Int, totalMessageCount: Int, recipientCounts: Int, federationInFlight: Int, messagesPct: Int, presencePct: Int, forwardsPct: Int, sessionsPct: Int, federationAvgLatencyMs: Int, federationMaxLatencyMs: Int, federationCallCount: Int, drainCompletedAt: Timestamp, ok: Bool}
475+ -> {sessions: Int, sessionsMax: Int, messages: Int, messagesMax: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, federationEnabled: Bool, recentGossip: Int, totalMessageCount: Int, recipientCounts: Int, federationInFlight: Int, messagesPct: Int, presencePct: Int, forwardsPct: Int, sessionsPct: Int, federationAvgLatencyMs: Int, federationMaxLatencyMs: Int, federationCallCount: Int, drainCompletedAt: Timestamp, ok: Bool}
476476 = Stats
477477 GET /stats/peers headers {authorization: Text}
478478 -> {peers: [PeerStatus], ok: Bool}
@@ -1011,13 +1011,13 @@ adjustRecipientCount = \counts pubkey delta ->
10111011 else if delta > 0 then union counts [{pubkey: pubkey, count: delta}]
10121012 else counts
10131013
1014−-- Count items per toKey in a single pass: collect unique keys, then count each.
1015−-- Avoids rebuilding the accumulator list on every matched item.
1014+-- Count items per toKey in a single pass: fold accumulates {pubkey, count} pairs,
1015+-- incrementing existing entries or inserting new ones.
10161016countByRecipient = \items ->
1017− let uniqueKeys = fold (\acc x ->
1018− if any acc (\k -> k == x.toKey) then acc
1019− else union acc [x.toKey]) [] items
1020− in do k <- uniqueKeys; yield {pubkey: k, count: count (do x <- items; where x.toKey == k; yield x)}
1017+ fold (\acc x ->
1018+ if any acc (\c -> c.pubkey == x.toKey)
1019+ then do c <- acc; yield (if c.pubkey == x.toKey then {pubkey: c.pubkey, count: c.count + 1} else c)
1020+ else union acc [{pubkey: x.toKey, count: 1}]) [] items
10211021
10221022recomputeRecipientCounts : [Message] -> [RecipientCount]
10231023recomputeRecipientCounts = countByRecipient
@@ -1502,11 +1502,9 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
15021502 sessions <- readSessionBucket bucket
15031503 case authenticateSession sessions token t of
15041504 Unauthorized {} -> onUnauth
1505− Authorized {pubkey, ip} -> do
1505+ Authorized {pubkey, ip} ->
15061506 if clientIp != "" && ip != "" && clientIp != ip
1507− then do
1508− println ("session: IP mismatch for " ++ pubkey ++ " (session: " ++ ip ++ ", request: " ++ clientIp ++ "), rejecting")
1509− onUnauth
1507+ then onUnauth
15101508 else onAuth pubkey
15111509
15121510removeForward = \mid srv -> atomic do
@@ -1565,10 +1563,13 @@ retryForwardsForServer = \entries remaining ->
15651563 Err e -> do
15661564 println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
15671565 recordFailure f.toServer
1568− -- Stop retrying: server is unreachable, remaining forwards will fail too
1569− Ok _ -> do
1570− removeForward f.messageId f.toServer
1571− clearFailure f.toServer
1566+ -- Network error: stop retrying, server is unreachable
1567+ Ok response -> do
1568+ if response.ok then do
1569+ removeForward f.messageId f.toServer
1570+ clearFailure f.toServer
1571+ else
1572+ println ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)
15721573 retryForwardsForServer rest (remaining - 1)
15731574
15741575-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
@@ -1870,11 +1871,9 @@ pollOnce = \token cursor clientIp -> do
18701871 sessions <- readSessionBucket bucket
18711872 case authenticateSession sessions token t of
18721873 Unauthorized {} -> yield (Nothing {})
1873− Authorized {pubkey, ip} -> do
1874+ Authorized {pubkey, ip} ->
18741875 if clientIp != "" && ip != "" && clientIp != ip
1875− then do
1876− println ("poll: session IP mismatch for " ++ pubkey ++ " (session: " ++ ip ++ ", request: " ++ clientIp ++ "), rejecting")
1877− yield (Nothing {})
1876+ then yield (Nothing {})
18781877 else do
18791878 -- Refresh session expiry (sliding window keeps active sessions alive)
18801879 writeSessionBucket bucket (do s <- sessions; yield (if safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip} else s))
@@ -2069,7 +2068,8 @@ handleChallenge = \pubkey clientIp respond -> do
20692068 when (preAuth == "ip_rate_limited") (println ("challenge: IP rate limited for " ++ clientIp))
20702069 yield (respond {challenge: "", ok: False {}, expiresAt: 0})
20712070 else do
2072− challenge <- genToken
2071+ rawChallenge <- genToken
2072+ let challenge = serverHostN ++ ":" ++ rawChallenge
20732073 result <- atomic do
20742074 challenges <- *challenges
20752075 let active = do c <- challenges; where c.expiresAt > t; yield c
@@ -2244,6 +2244,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
22442244 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: ""})
22452245 else do
22462246 effectiveMsgId <- if messageId != "" then yield messageId else genToken
2247+ if t - timestamp > messageMaxAge
2248+ then yield (respond {status: MsRejected {}, messageId: ""})
2249+ else do
22472250 let clampedTs = clampTimestamp timestamp t
22482251 result <- atomic do
22492252 draining <- *draining
@@ -2335,9 +2338,12 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
23352338 Err e -> do
23362339 println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
23372340 recordFailure server
2338− Ok _ -> do
2339− removeForward effectiveMsgId server
2340− clearFailure server))))
2341+ Ok response ->
2342+ if response.ok then do
2343+ removeForward effectiveMsgId server
2344+ clearFailure server
2345+ else
2346+ println ("federation: send forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))
23412347 _ -> yield {}
23422348 yield (case result.send of
23432349 SendFailed {status} -> respond {status: status, messageId: ""}
@@ -2418,6 +2424,10 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do
24182424 let validated = validGossipEvents events t
24192425 let onlineEntries = validated.online
24202426 let trustedOffline = trustedOfflineEvents validated.offline fromServer
2427+ -- Warn if a peer sends both online and offline for the same key in one batch (suspicious)
2428+ let conflictingKeys = do on <- onlineEntries; off <- trustedOffline; where on.pubkey == off.pubkey && on.server == off.server; yield on.pubkey
2429+ when (count conflictingKeys > 0)
2430+ (println ("gossip: warning: " ++ fromServer ++ " sent conflicting online+offline for same key(s)"))
24212431 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
24222432 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
24232433 let deduped = mergeByPubkeyServer [] filteredOnline
@@ -2600,6 +2610,7 @@ handleUndrain = \authorization respond -> do
26002610 wasDraining <- atomic do
26012611 d <- *draining
26022612 set *draining = False {}
2613+ set *drainCompletedAt = 0
26032614 yield d
26042615 when wasDraining (println "undrain: server resumed, accepting new requests")
26052616 when (not wasDraining) (println "undrain: server was not draining")
@@ -2628,7 +2639,9 @@ handleHealth = \respond -> do
26282639 tmc <- *totalMessageCount
26292640 lastFedSuccess <- *lastFederationSuccessAt
26302641 forwardFailures <- *forwardFailures
2642+ pruneRun <- *pruneRunning
26312643 let warmup = t - sa < healthPruneMaxAge
2644+ let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
26322645 -- Capacity pressure: flag when any capped collection exceeds 90% full
26332646 let capacityOk = tmc * 10 <= maxMessages * 9
26342647 && count presence * 10 <= maxPresenceEntries * 9
@@ -2639,7 +2652,7 @@ handleHealth = \respond -> do
26392652 let fedOk = not federationEnabled || warmup
26402653 || count forwardFailures == 0
26412654 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
2642− yield {pruneOk: warmup || t - lastPrune < healthPruneMaxAge,
2655+ yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
26432656 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
26442657 capacityOk: capacityOk,
26452658 federationOk: fedOk}
@@ -2649,17 +2662,16 @@ handleHealth = \respond -> do
26492662 capacityHealthy: result.capacityOk,
26502663 federationHealthy: result.federationOk})
26512664
2652−zeroStats = {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pruneCount: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, drainCompletedAt: 0, ok: False {}}
2665+zeroStats = {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, drainCompletedAt: 0, ok: False {}}
26532666
26542667handleStats = \authorization respond -> do
26552668 let authorized = verifyMaintenanceAuth authorization
26562669 if authorized
26572670 then do
26582671 t <- now
2659− -- Read all refs atomically for a consistent snapshot, count outside to minimize lock time
2660− -- Read all refs atomically for a consistent snapshot. Messages use totalMessageCount
2661− -- (maintained counter) instead of scanning all 16 message buckets.
2662− snapshot <- atomic do
2672+ -- Split into two atomic reads: core capacity state (consistent) and counters/timing
2673+ -- (slight inconsistency acceptable for monitoring). This narrows the lock window.
2674+ core <- atomic do
26632675 sc <- countAllSessions
26642676 p <- *presence
26652677 f <- *forwards
@@ -2669,6 +2681,12 @@ handleStats = \authorization respond -> do
26692681 d <- *deliveredIds
26702682 ff <- *forwardFailures
26712683 ap <- *activePolls
2684+ tmc <- *totalMessageCount
2685+ rc <- *recipientMsgCounts
2686+ fi <- *federationInFlight
2687+ rg <- *recentGossip
2688+ yield {sc: sc, p: p, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg}
2689+ counters <- atomic do
26722690 sr <- *sendRates
26732691 cr <- *challengeRates
26742692 fr <- *federationRates
@@ -2677,27 +2695,24 @@ handleStats = \authorization respond -> do
26772695 lr <- *lookupRates
26782696 rr <- *recvRates
26792697 pc <- *pruneCount
2698+ pr <- *pruneRunning
26802699 hc <- *heartbeatCount
26812700 sa <- *startedAt
26822701 lp <- *lastPrunedAt
26832702 lh <- *lastHeartbeatAt
26842703 dr <- *draining
2685− rg <- *recentGossip
2686− tmc <- *totalMessageCount
2687− rc <- *recipientMsgCounts
2688− fi <- *federationInFlight
26892704 flt <- *federationLatencyTotal
26902705 flc <- *federationLatencyCount
26912706 flm <- *federationLatencyMax
26922707 dca <- *drainCompletedAt
2693− yield {sc: sc, p: p, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, sr: sr, cr: cr, fr: fr, grc: grc, ir: ir, lr: lr, rr: rr, pc: pc, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, rg: rg, tmc: tmc, rc: rc, fi: fi, flt: flt, flc: flc, flm: flm, dca: dca}
2694− let result = {sessions: snapshot.sc, messages: snapshot.tmc, presence: count snapshot.p, forwards: count snapshot.f, challenges: count snapshot.c, ackedIds: count snapshot.a, ackOverflow: count snapshot.ao, deliveredIds: count snapshot.d, forwardFailures: count snapshot.ff, activePolls: count snapshot.ap, sendRates: count snapshot.sr, challengeRates: count snapshot.cr, federationRates: count snapshot.fr, globalRateCount: snapshot.grc, ipRates: count snapshot.ir, lookupRates: count snapshot.lr, recvRates: count snapshot.rr, pruneCount: snapshot.pc, heartbeatCount: snapshot.hc, uptime: t - snapshot.sa, lastPruneAt: snapshot.lp, lastHeartbeatAt: snapshot.lh, draining: snapshot.dr, recentGossip: count snapshot.rg, totalMessageCount: snapshot.tmc, recipientCounts: count snapshot.rc, federationInFlight: count snapshot.fi, federationLatencyTotal: snapshot.flt, federationLatencyCount: snapshot.flc, federationLatencyMax: snapshot.flm, drainCompletedAt: snapshot.dca}
2708+ yield {sr: sr, cr: cr, fr: fr, grc: grc, ir: ir, lr: lr, rr: rr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, dca: dca}
2709+ let result = {sessions: core.sc, messages: core.tmc, presence: count core.p, forwards: count core.f, challenges: count core.c, ackedIds: count core.a, ackOverflow: count core.ao, deliveredIds: count core.d, forwardFailures: count core.ff, activePolls: count core.ap, sendRates: count counters.sr, challengeRates: count counters.cr, federationRates: count counters.fr, globalRateCount: counters.grc, ipRates: count counters.ir, lookupRates: count counters.lr, recvRates: count counters.rr, pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr, recentGossip: count core.rg, totalMessageCount: core.tmc, recipientCounts: count core.rc, federationInFlight: count core.fi, federationLatencyTotal: counters.flt, federationLatencyCount: counters.flc, federationLatencyMax: counters.flm, drainCompletedAt: counters.dca}
26952710 let msgPct = if maxMessages > 0 then result.messages * 100 / maxMessages else 0
26962711 let presPct = if maxPresenceEntries > 0 then result.presence * 100 / maxPresenceEntries else 0
26972712 let fwdPct = if maxForwardsEntries > 0 then result.forwards * 100 / maxForwardsEntries else 0
26982713 let sessPct = if maxSessions > 0 then result.sessions * 100 / maxSessions else 0
26992714 let fedAvg = if result.federationLatencyCount > 0 then result.federationLatencyTotal / result.federationLatencyCount else 0
2700− yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, draining: result.draining, federationEnabled: federationEnabled, recentGossip: result.recentGossip, totalMessageCount: result.totalMessageCount, recipientCounts: result.recipientCounts, federationInFlight: result.federationInFlight, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, drainCompletedAt: result.drainCompletedAt, ok: True {}})
2715+ yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pruneCount: result.pruneCount, pruneRunning: result.pruneRunning, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, draining: result.draining, federationEnabled: federationEnabled, recentGossip: result.recentGossip, totalMessageCount: result.totalMessageCount, recipientCounts: result.recipientCounts, federationInFlight: result.federationInFlight, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, drainCompletedAt: result.drainCompletedAt, ok: True {}})
27012716 else yield (respond zeroStats)
27022717
27032718handleStatsPeers = \authorization respond -> do