00797dc · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Consolidate server.knot: refactor send logic, add federation/watchdog monitoring, improve code quality

00797dciko6 months ago

server.knot+219−316

compare with master

unifiedsplit

@@ -139,7 +139,7 @@ federationLatencyWindowMs : Int<Ms>

139federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval

140

141-- Snapshot format version (increment on schema changes to prevent corrupt restores)

142−snapshotVersion = 2

142+snapshotVersion = 3

143

144-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)

145maxGossipEventsPerServer = 500 -- max total gossip events per server per window

@@ -211,6 +211,8 @@ healthPruneMaxAge : Int<Ms>

211healthPruneMaxAge = backgroundPruneInterval * 3

212healthHeartbeatMaxAge : Int<Ms>

213healthHeartbeatMaxAge = pollTimeoutMs * 3

214+healthWatchdogMaxAge : Int<Ms>

215+healthWatchdogMaxAge = backgroundPruneInterval * 15 -- watchdog runs every ~5 prune intervals

214

216

215-- ============================================

217-- ============================================

216-- Types

218-- Types

@@ -257,9 +259,6 @@ impl ToJSON MessageStatus where

257 MsDuplicate {} -> "duplicate"

259 MsDuplicate {} -> "duplicate"

258 MsDraining {} -> "draining")

260 MsDraining {} -> "draining")

259

261

260−-- Internal delivery status (same as MessageStatus but without error variants)

261−data DeliveryStatus = DsDelivered {} | DsFederated {} | DsQueued {}

262−

263-- Validated hex-encoded types with maximum length constraints

262-- Validated hex-encoded types with maximum length constraints

264isValidBoundedHex : Int -> Text -> Bool

263isValidBoundedHex : Int -> Text -> Bool

265isValidBoundedHex = \maxLen s -> not (tooLong s maxLen) && isValidHex s && s == normalizeHex s

264isValidBoundedHex = \maxLen s -> not (tooLong s maxLen) && isValidHex s && s == normalizeHex s

@@ -306,7 +305,6 @@ type FederationPairRateEntry = {server: Text, recipient: PubkeyHex, windowStart:

306-- Tracks active long-poll connections to enforce per-key concurrency limits

305-- Tracks active long-poll connections to enforce per-key concurrency limits

307type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}

306type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}

308data AuthResult = Authorized {pubkey: PubkeyHex, ip: Text} | Unauthorized {}

307data AuthResult = Authorized {pubkey: PubkeyHex, ip: Text} | Unauthorized {}

309−data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName]} | SendDuplicate {}

310data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

308data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

311

309

312type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int}

310type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int}

@@ -554,6 +552,9 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}

554-- Polls compare against this to skip expensive presence lookups when nothing changed.

552-- Polls compare against this to skip expensive presence lookups when nothing changed.

555*presenceVersion : Timestamp

553*presenceVersion : Timestamp

556

554

555+-- Last time the background watchdog loop ran (for health check liveness detection)

556+*lastWatchdogAt : Timestamp

557+

557-- ============================================

558-- ============================================

558-- Routes

559-- Routes

559-- ============================================

560-- ============================================

@@ -646,11 +647,11 @@ route MaintenanceApi where

646 -- Messages are excluded (ephemeral by design; clients handle re-delivery).

647 -- Messages are excluded (ephemeral by design; clients handle re-delivery).

647 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.

648 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.

648 GET /maintenance/snapshot headers {authorization: Text}

649 GET /maintenance/snapshot headers {authorization: Text}

649− -> {ok: Bool, version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce]}

650+ -> {ok: Bool, version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text]}

650 = Snapshot

651 = Snapshot

651 -- Restore: import a previously-exported snapshot. Server must be in drain mode

652 -- Restore: import a previously-exported snapshot. Server must be in drain mode

652 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.

653 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.

653− POST {version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce]} /maintenance/restore headers {authorization: Text}

654+ POST {version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text]} /maintenance/restore headers {authorization: Text}

654 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}

655 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}

655 = Restore

656 = Restore

656 -- Block a federation peer (rejects all inbound/outbound federation with this server)

657 -- Block a federation peer (rejects all inbound/outbound federation with this server)

@@ -679,7 +680,7 @@ route MaintenanceApi where

679 = AdminRevokeSessions

680 = AdminRevokeSessions

680

681

681route HealthApi where

682route HealthApi where

682− GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool} = Health

683+ GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool} = Health

683

684

684-- Returns counts of in-memory collections (requires maintenance secret)

685-- Returns counts of in-memory collections (requires maintenance secret)

685-- TODO: If Knot adds raw text response support, add a /metrics endpoint

686-- TODO: If Knot adds raw text response support, add a /metrics endpoint

@@ -687,7 +688,7 @@ route HealthApi where

687-- monitoring stacks. Currently /stats returns JSON which requires a custom scraper.

688-- monitoring stacks. Currently /stats returns JSON which requires a custom scraper.

688route StatsApi where

689route StatsApi where

689 GET /stats headers {authorization: Text}

690 GET /stats headers {authorization: Text}

690− -> {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, pairRates: Int, federationPairRates: 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, hungFederationSlots: Int, messagesPct: Int, presencePct: Int, forwardsPct: Int, sessionsPct: Int, federationAvgLatencyMs: Int, federationMaxLatencyMs: Int, federationCallCount: Int, drainCompletedAt: Timestamp, oldestForwardAge: Int, oldestAckAge: Int, ok: Bool}

691+ -> {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, pairRates: Int, federationPairRates: 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, hungFederationSlots: Int, messagesPct: Int, presencePct: Int, forwardsPct: Int, sessionsPct: Int, federationAvgLatencyMs: Int, federationMaxLatencyMs: Int, federationCallCount: Int, drainCompletedAt: Timestamp, oldestForwardAge: Int, oldestAckAge: Int, forwardNonces: Int, forwardNoncesMax: Int, pendingAckCleanups: Int, blockedPeers: Int, gossipEventRates: Int, verifyIpRates: Int, ok: Bool}

691 = Stats

692 = Stats

692 GET /stats/peers headers {authorization: Text}

693 GET /stats/peers headers {authorization: Text}

693 -> {peers: [PeerStatus], ok: Bool}

694 -> {peers: [PeerStatus], ok: Bool}

@@ -1146,6 +1147,46 @@ transformAllSessionBuckets = \f -> do

1146 b14 <- *sessionBucket14; set *sessionBucket14 = f b14

1147 b14 <- *sessionBucket14; set *sessionBucket14 = f b14

1147 b15 <- *sessionBucket15; set *sessionBucket15 = f b15

1148 b15 <- *sessionBucket15; set *sessionBucket15 = f b15

1148

1149

1150+-- Single-pass partition of sessions by token bucket and write to all 16 buckets.

1151+-- Mirrors writeAllPresence for the session sharding scheme.

1152+writeAllSessions : [Session] -> IO ()

1153+writeAllSessions = \sessions -> do

1154+ let partitioned = fold (\acc s ->

1155+ let b = tokenBucket s.token in

1156+ {b0: if b == 0 then union [s] acc.b0 else acc.b0,

1157+ b1: if b == 1 then union [s] acc.b1 else acc.b1,

1158+ b2: if b == 2 then union [s] acc.b2 else acc.b2,

1159+ b3: if b == 3 then union [s] acc.b3 else acc.b3,

1160+ b4: if b == 4 then union [s] acc.b4 else acc.b4,

1161+ b5: if b == 5 then union [s] acc.b5 else acc.b5,

1162+ b6: if b == 6 then union [s] acc.b6 else acc.b6,

1163+ b7: if b == 7 then union [s] acc.b7 else acc.b7,

1164+ b8: if b == 8 then union [s] acc.b8 else acc.b8,

1165+ b9: if b == 9 then union [s] acc.b9 else acc.b9,

1166+ b10: if b == 10 then union [s] acc.b10 else acc.b10,

1167+ b11: if b == 11 then union [s] acc.b11 else acc.b11,

1168+ b12: if b == 12 then union [s] acc.b12 else acc.b12,

1169+ b13: if b == 13 then union [s] acc.b13 else acc.b13,

1170+ b14: if b == 14 then union [s] acc.b14 else acc.b14,

1171+ b15: if b == 15 then union [s] acc.b15 else acc.b15})

1172+ {b0: [], b1: [], b2: [], b3: [], b4: [], b5: [], b6: [], b7: [], b8: [], b9: [], b10: [], b11: [], b12: [], b13: [], b14: [], b15: []} sessions

1173+ writeSessionBucket 0 partitioned.b0

1174+ writeSessionBucket 1 partitioned.b1

1175+ writeSessionBucket 2 partitioned.b2

1176+ writeSessionBucket 3 partitioned.b3

1177+ writeSessionBucket 4 partitioned.b4

1178+ writeSessionBucket 5 partitioned.b5

1179+ writeSessionBucket 6 partitioned.b6

1180+ writeSessionBucket 7 partitioned.b7

1181+ writeSessionBucket 8 partitioned.b8

1182+ writeSessionBucket 9 partitioned.b9

1183+ writeSessionBucket 10 partitioned.b10

1184+ writeSessionBucket 11 partitioned.b11

1185+ writeSessionBucket 12 partitioned.b12

1186+ writeSessionBucket 13 partitioned.b13

1187+ writeSessionBucket 14 partitioned.b14

1188+ writeSessionBucket 15 partitioned.b15

1189+

1149-- ============================================

1190-- ============================================

1150-- Presence Bucket Helpers

1191-- Presence Bucket Helpers

1151-- ============================================

1192-- ============================================

@@ -1555,15 +1596,6 @@ clampTtl = clamp 1 maxPresenceTtl

1555clampTimestamp : Timestamp -> Timestamp -> Timestamp

1596clampTimestamp : Timestamp -> Timestamp -> Timestamp

1556clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp

1597clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp

1557

1598

1558−deliveryToMessageStatus : DeliveryStatus -> MessageStatus

1559−deliveryToMessageStatus = \ds -> case ds of

1560− DsDelivered {} -> MsDelivered {}

1561− DsFederated {} -> MsFederated {}

1562− DsQueued {} -> MsQueued {}

1563−

1564−isQueued : DeliveryStatus -> Bool

1565−isQueued = \ds -> case ds of DsQueued {} -> True {}; _ -> False {}

1566−

1567-- Ensure strictly monotonic timestamps for message ordering

1599-- Ensure strictly monotonic timestamps for message ordering

1568nextSeqTs : Timestamp -> Timestamp -> Timestamp

1600nextSeqTs : Timestamp -> Timestamp -> Timestamp

1569nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1

1601nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1

@@ -2048,7 +2080,7 @@ checkAuthGlobalRate = \t -> atomic do

2048 yield (False {})

2080 yield (False {})

2049

2081

2050-- IP-based rate limiting; skips if clientIp header is empty (no reverse proxy)

2082-- IP-based rate limiting; skips if clientIp header is empty (no reverse proxy)

2051−checkIpRate = \ip t ->

2083+checkClientIpRate = \ip t ->

2052 if ip == "" then yield (False {})

2084 if ip == "" then yield (False {})

2053 else atomic do

2085 else atomic do

2054 ipRates <- *ipRates

2086 ipRates <- *ipRates

@@ -2064,7 +2096,7 @@ checkPreAuth = \clientIp t -> do

2064 globalLimited <- checkGlobalRate t

2096 globalLimited <- checkGlobalRate t

2065 if globalLimited then yield "global_rate_limited"

2097 if globalLimited then yield "global_rate_limited"

2066 else do

2098 else do

2067− ipLimited <- checkIpRate clientIp t

2099+ ipLimited <- checkClientIpRate clientIp t

2068 if ipLimited then yield "ip_rate_limited"

2100 if ipLimited then yield "ip_rate_limited"

2069 else yield ""

2101 else yield ""

2070

2102

@@ -2985,42 +3017,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->

2985 let capped = capPerKeyAndGlobal afterRevoke (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions

3017 let capped = capPerKeyAndGlobal afterRevoke (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions

2986 let sessionsWithNew = union capped [newSession]

3018 let sessionsWithNew = union capped [newSession]

2987 set *totalSessionCount = count sessionsWithNew

3019 set *totalSessionCount = count sessionsWithNew

2988− -- Single-pass partition by token bucket (O(n) instead of 16x O(n) scan)

3020+ writeAllSessions sessionsWithNew

2989− let partitioned = fold (\acc s ->

2990− let b = tokenBucket s.token in

2991− {b0: if b == 0 then union [s] acc.b0 else acc.b0,

2992− b1: if b == 1 then union [s] acc.b1 else acc.b1,

2993− b2: if b == 2 then union [s] acc.b2 else acc.b2,

2994− b3: if b == 3 then union [s] acc.b3 else acc.b3,

2995− b4: if b == 4 then union [s] acc.b4 else acc.b4,

2996− b5: if b == 5 then union [s] acc.b5 else acc.b5,

2997− b6: if b == 6 then union [s] acc.b6 else acc.b6,

2998− b7: if b == 7 then union [s] acc.b7 else acc.b7,

2999− b8: if b == 8 then union [s] acc.b8 else acc.b8,

3000− b9: if b == 9 then union [s] acc.b9 else acc.b9,

3001− b10: if b == 10 then union [s] acc.b10 else acc.b10,

3002− b11: if b == 11 then union [s] acc.b11 else acc.b11,

3003− b12: if b == 12 then union [s] acc.b12 else acc.b12,

3004− b13: if b == 13 then union [s] acc.b13 else acc.b13,

3005− b14: if b == 14 then union [s] acc.b14 else acc.b14,

3006− b15: if b == 15 then union [s] acc.b15 else acc.b15})

3007− {b0: [], b1: [], b2: [], b3: [], b4: [], b5: [], b6: [], b7: [], b8: [], b9: [], b10: [], b11: [], b12: [], b13: [], b14: [], b15: []} sessionsWithNew

3008− writeSessionBucket 0 partitioned.b0

3009− writeSessionBucket 1 partitioned.b1

3010− writeSessionBucket 2 partitioned.b2

3011− writeSessionBucket 3 partitioned.b3

3012− writeSessionBucket 4 partitioned.b4

3013− writeSessionBucket 5 partitioned.b5

3014− writeSessionBucket 6 partitioned.b6

3015− writeSessionBucket 7 partitioned.b7

3016− writeSessionBucket 8 partitioned.b8

3017− writeSessionBucket 9 partitioned.b9

3018− writeSessionBucket 10 partitioned.b10

3019− writeSessionBucket 11 partitioned.b11

3020− writeSessionBucket 12 partitioned.b12

3021− writeSessionBucket 13 partitioned.b13

3022− writeSessionBucket 14 partitioned.b14

3023− writeSessionBucket 15 partitioned.b15

3024 yield (Just {value: {peerServers: peerServers, revokedCount: revokedCount}})

3021 yield (Just {value: {peerServers: peerServers, revokedCount: revokedCount}})

3025 case authResult of

3022 case authResult of

3026 Nothing {} -> yield (respond {token: "", expiresAt: 0})

3023 Nothing {} -> yield (respond {token: "", expiresAt: 0})

@@ -3046,17 +3043,22 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do

3046 globalRateLimited <- checkGlobalRate t

3043 globalRateLimited <- checkGlobalRate t

3047 if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs})

3044 if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs})

3048 else do

3045 else do

3046+ -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups

3047+ startPresenceVersion <- *presenceVersion

3049 result <- case extractToken authorization of

3048 result <- case extractToken authorization of

3050 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}

3049 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}

3051 Just {value: token} -> pollOnce token effectiveCursor clientIp

3050 Just {value: token} -> pollOnce token effectiveCursor clientIp

3052− -- Check presence for watched keys (piggybacks on poll cycle; max latency = pollTimeoutMs)

3051+ -- Check presence for watched keys (piggybacks on poll cycle; max latency = pollTimeoutMs).

3052+ -- Skip the expensive bucket reads if presenceVersion hasn't changed since poll start.

3053 presenceUpdates <- if result.authorized && count cappedWatchKeys > 0 then do

3054+ currentPresenceVersion <- *presenceVersion

3055+ if currentPresenceVersion == startPresenceVersion then yield []

3056+ else do

3054− t2 <- now

3057+ t2 <- now

3055− atomic do

3058+ atomic do

3056− -- Read only the presence buckets needed for watched keys (not all 16)

3057− let neededBuckets = uniqueBucketsFor cappedWatchKeys (\k -> k)

3059+ let neededBuckets = uniqueBucketsFor cappedWatchKeys (\k -> k)

3058− relevantPresence <- readSelectedPresenceBuckets neededBuckets

3060+ relevantPresence <- readSelectedPresenceBuckets neededBuckets

3059− yield (do key <- cappedWatchKeys; yield {pubkey: key, online: count (presenceForKey relevantPresence key t2) > 0})

3061+ yield (do key <- cappedWatchKeys; yield {pubkey: key, online: count (presenceForKey relevantPresence key t2) > 0})

3060 else yield []

3062 else yield []

3061 fork (when federationEnabled (do

3063 fork (when federationEnabled (do

3062 sendAcksAndClean result.pendingAcks

3064 sendAcksAndClean result.pendingAcks

@@ -3064,6 +3066,90 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do

3064 let isHeartbeat = result.authorized && count result.events == 0

3066 let isHeartbeat = result.authorized && count result.events == 0

3065 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0})

3067 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0})

3066

3068

3069+-- Core send logic: dedup, rate limiting, store message, create forwards.

3070+-- Called inside an atomic block by both handleSendMessage and sendBatchInner.

3071+-- Caller handles auth, drain checks, and messageId generation.

3072+sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->

3073+ if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}

3074+ else do

3075+ let msgBucket = pubkeyBucket to

3076+ bucket <- readMessageBucket msgBucket

3077+ ackedIds <- *ackedIds

3078+ deliveredIds <- *deliveredIds

3079+ let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedIds || hasMessage effectiveMsgId to deliveredIds

3080+ if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}

3081+ else do

3082+ sendRates <- *sendRates

3083+ let rateResult = checkPubkeyRate sendRates senderKey maxSendRatePerKey sendRateWindowMs maxSendRateEntries t

3084+ if rateResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: rateResult.retryAfterMs}

3085+ else do

3086+ set *sendRates = rateResult.entries

3087+ pairRates <- *pairRates

3088+ let pairResult = checkPairRate pairRates senderKey to maxPairRatePerPair pairRateWindowMs maxPairRateEntries t

3089+ set *pairRates = pairResult.entries

3090+ if pairResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: pairResult.retryAfterMs}

3091+ else do

3092+ recvRates <- *recvRates

3093+ let recvResult = checkPubkeyRate recvRates to maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t

3094+ if recvResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: recvResult.retryAfterMs}

3095+ else do

3096+ set *recvRates = recvResult.entries

3097+ totalMsgCount <- *totalMessageCount

3098+ if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}

3099+ else do

3100+ recipientCounts <- *recipientMsgCounts

3101+ let recipientCount = getRecipientCount to recipientCounts

3102+ if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}

3103+ else do

3104+ recipientPresence <- readPresenceBucket (pubkeyBucket to)

3105+ seqCounter <- *seqCounter

3106+ let seqTs = nextSeqTs t seqCounter

3107+ set *seqCounter = seqTs

3108+ writeMessageBucket msgBucket (union bucket

3109+ [{messageId: effectiveMsgId, toKey: to,

3110+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])

3111+ set *totalMessageCount = totalMsgCount + 1

3112+ set *recipientMsgCounts = adjustRecipientCount recipientCounts to 1

3113+ signalMessageForKey to seqTs

3114+ let toPresence = presenceForKey recipientPresence to t

3115+ let localOnline = hasLocalPresence toPresence

3116+ let remoteServers = collectPeerServers toPresence (\p -> p.server)

3117+ when (federationEnabled && count remoteServers > 0) (do

3118+ forwards <- *forwards

3119+ 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, originServer: serverHostN, retries: 0}

3120+ let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds

3121+ set *forwards = union forwards cappedFwds)

3122+ let status = if localOnline then MsDelivered {}

3123+ else if federationEnabled && count remoteServers > 0 then MsFederated {}

3124+ else MsQueued {}

3125+ yield {status: status, remoteServers: remoteServers, retryAfterMs: 0}

3126+

3127+-- Fork federation forwards for messages that need to be delivered to remote servers

3128+forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->

3129+ when (count remoteServers > 0 && federationEnabled) (

3130+ fork (sendToEach remoteServers (\server -> do

3131+ stillPending <- atomic do

3132+ forwards <- *forwards

3133+ yield (any forwards (\f -> f.messageId == effectiveMsgId && f.toServer == server))

3134+ when stillPending (do

3135+ fwdT <- now

3136+ fwdAuth <- generateFederationAuth "forward"

3137+ fwdResult <- fetchFederation server (ForwardMessage {

3138+ toKey: to, encryptedBlob: encryptedBlob,

3139+ timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,

3140+ fromServer: serverHostN, requestTimestamp: fwdT,

3141+ authorization: fwdAuth})

3142+ case fwdResult of

3143+ Err e -> do

3144+ logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)

3145+ recordFailure server

3146+ Ok response ->

3147+ if response.ok then do

3148+ removeForward effectiveMsgId server

3149+ clearFailure server

3150+ else

3151+ logWarn ("federation: forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))

3152+

3067-- POST /messages: Send an encrypted message to a recipient.

3153-- POST /messages: Send an encrypted message to a recipient.

3068-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.

3154-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.

3069-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.

3155-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.

@@ -3083,113 +3169,21 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp

3083 let clampedTs = clampTimestamp timestamp t

3169 let clampedTs = clampTimestamp timestamp t

3084 result <- atomic do

3170 result <- atomic do

3085 draining <- *draining

3171 draining <- *draining

3086− if draining then yield {send: SendFailed {status: MsDraining {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}

3172+ if draining then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}

3087 else withSessionAuth authorization t clientIp

3173 else withSessionAuth authorization t clientIp

3088− (yield {send: SendFailed {status: MsUnauthorized {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0})

3174+ (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})

3089− (\senderKey -> do

3175+ (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId t)

3090− -- Reject self-sends (sender == recipient wastes a mailbox slot)

3176+ case result.status of

3091− if senderKey == to then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}

3092− else do

3093− let msgBucket = pubkeyBucket to

3094− bucket <- readMessageBucket msgBucket

3095− ackedIds <- *ackedIds

3096− deliveredIds <- *deliveredIds

3097− -- Dedup check (only when messageId is non-empty; scans only the recipient's bucket)

3098− let isDup = checkMessageDup bucket messageId to || (messageId != "" && (hasMessage messageId to ackedIds || hasMessage messageId to deliveredIds))

3099− if isDup

3100− then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}

3101− else do

3102− sendRates <- *sendRates

3103− let rateResult = checkPubkeyRate sendRates senderKey maxSendRatePerKey sendRateWindowMs maxSendRateEntries t

3104− if rateResult.limited

3105− then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}, globalCapReached: False {}, retryAfterMs: rateResult.retryAfterMs}

3106− else do

3107− set *sendRates = rateResult.entries

3108− pairRates <- *pairRates

3109− let pairResult = checkPairRate pairRates senderKey to maxPairRatePerPair pairRateWindowMs maxPairRateEntries t

3110− set *pairRates = pairResult.entries

3111− if pairResult.limited

3112− then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: pairResult.retryAfterMs}

3113− else do

3114− recvRates <- *recvRates

3115− let recvResult = checkPubkeyRate recvRates to maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t

3116− if recvResult.limited

3117− then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: recvResult.retryAfterMs}

3118− else do

3119− set *recvRates = recvResult.entries

3120− totalMsgCount <- *totalMessageCount

3121− if totalMsgCount >= maxMessages

3122− then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: True {}, retryAfterMs: 0}

3123− else do

3124− -- Recipient count from maintained counts (O(unique_recipients) instead of O(all_messages))

3125− recipientCounts <- *recipientMsgCounts

3126− let recipientCount = getRecipientCount to recipientCounts

3127− if recipientCount >= maxMessagesPerRecipient

3128− then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}

3129− else do

3130− -- Hot path: read only recipient's presence bucket

3131− recipientPresence <- readPresenceBucket (pubkeyBucket to)

3132− seqCounter <- *seqCounter

3133− let seqTs = nextSeqTs t seqCounter

3134− set *seqCounter = seqTs

3135− writeMessageBucket msgBucket (union bucket

3136− [{messageId: effectiveMsgId, toKey: to,

3137− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])

3138− set *totalMessageCount = totalMsgCount + 1

3139− set *recipientMsgCounts = adjustRecipientCount recipientCounts to 1

3140− signalMessageForKey to seqTs

3141− let toPresence = presenceForKey recipientPresence to t

3142− let localOnline = hasLocalPresence toPresence

3143− let remoteServers = collectPeerServers toPresence (\p -> p.server)

3144− -- Note: DsFederated means "queued for federation forwarding", not

3145− -- "confirmed delivered to remote server". The actual HTTP forward

3146− -- happens asynchronously in a fork below; retries are handled by

3147− -- the background prune cycle if the initial attempt fails.

3148− let delivery = if localOnline then DsDelivered {}

3149− else if federationEnabled && count remoteServers > 0 then DsFederated {}

3150− else DsQueued {}

3151− -- Create forward entries atomically with message storage (prevents race with ack)

3152− when (federationEnabled && count remoteServers > 0) (do

3153− forwards <- *forwards

3154− 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, originServer: serverHostN, retries: 0}

3155− let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds

3156− set *forwards = union forwards cappedFwds)

3157− yield {send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0})

3158− when result.senderRateLimited

3159− (logWarn ("sendMessage: rate limited (" ++ show maxSendRatePerKey ++ "/s per key)"))

3177+ MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)

3160− when result.globalCapReached

3178+ MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))

3161− (logWarn ("sendMessage: global message cap reached (" ++ show maxMessages ++ "), rejecting"))

3162− case result.send of

3163− SendDone {delivery, remoteServers} ->

3164− when (count remoteServers > 0 && federationEnabled) (

3165− fork (sendToEach remoteServers (\server -> do

3166− -- Check if forward still exists (recipient may have acked during fork delay)

3167− stillPending <- atomic do

3168− forwards <- *forwards

3169− yield (any forwards (\f -> f.messageId == effectiveMsgId && f.toServer == server))

3170− when stillPending (do

3171− fwdT <- now

3172− fwdAuth <- generateFederationAuth "forward"

3173− fwdResult <- fetchFederation server (ForwardMessage {

3174− toKey: to, encryptedBlob: encryptedBlob,

3175− timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,

3176− fromServer: serverHostN, requestTimestamp: fwdT,

3177− authorization: fwdAuth})

3178− case fwdResult of

3179− Err e -> do

3180− logWarn ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)

3181− recordFailure server

3182− Ok response ->

3183− if response.ok then do

3184− removeForward effectiveMsgId server

3185− clearFailure server

3186− else

3187− logWarn ("federation: send forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))

3188 _ -> yield {}

3179 _ -> yield {}

3180+ forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.remoteServers

3189− yield (case result.send of

3181+ yield (case result.status of

3182+ MsDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0}

3190− SendFailed {status} -> respond {status: status, messageId: "", retryAfterMs: result.retryAfterMs}

3183+ MsDelivered {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId, retryAfterMs: 0}

3191− SendDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0}

3184+ MsFederated {} -> respond {status: MsFederated {}, messageId: effectiveMsgId, retryAfterMs: 0}

3192− SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId, retryAfterMs: 0})

3185+ MsQueued {} -> respond {status: MsQueued {}, messageId: effectiveMsgId, retryAfterMs: 0}

3186+ _ -> respond {status: result.status, messageId: "", retryAfterMs: result.retryAfterMs})

3193

3187

3194-- GET /lookup/:pubkey: Look up the best known server for a pubkey.

3188-- GET /lookup/:pubkey: Look up the best known server for a pubkey.

3195-- Returns local server if key is online locally, otherwise the remote with latest expiry.

3189-- Returns local server if key is online locally, otherwise the remote with latest expiry.

@@ -3363,7 +3357,7 @@ handleVerifyToken = \token scope clientIp respond -> do

3363 globalRateLimited <- checkGlobalRate t

3357 globalRateLimited <- checkGlobalRate t

3364 if globalRateLimited then yield (respond {ok: False {}})

3358 if globalRateLimited then yield (respond {ok: False {}})

3365 else do

3359 else do

3366− ipLimited <- checkIpRate clientIp t

3360+ ipLimited <- checkClientIpRate clientIp t

3367 if ipLimited then yield (respond {ok: False {}})

3361 if ipLimited then yield (respond {ok: False {}})

3368 else do

3362 else do

3369 -- Dedicated verify rate limit (tighter than general IP rate to limit token probing)

3363 -- Dedicated verify rate limit (tighter than general IP rate to limit token probing)

@@ -3694,6 +3688,7 @@ handleHealth = \authorization respond -> do

3694 result <- atomic do

3688 result <- atomic do

3695 lastPrune <- *lastPrunedAt

3689 lastPrune <- *lastPrunedAt

3696 lastHb <- *lastHeartbeatAt

3690 lastHb <- *lastHeartbeatAt

3691+ lastWd <- *lastWatchdogAt

3697 sa <- *startedAt

3692 sa <- *startedAt

3698 presenceCount <- *totalPresenceCount

3693 presenceCount <- *totalPresenceCount

3699 forwards <- *forwards

3694 forwards <- *forwards

@@ -3721,12 +3716,15 @@ handleHealth = \authorization respond -> do

3721 let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup

3716 let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup

3722 || (lastSnap > 0 && t - lastSnap <= snapshotStalenessWarning)

3717 || (lastSnap > 0 && t - lastSnap <= snapshotStalenessWarning)

3723 || (tmc == 0 && presenceCount == 0 && count forwards == 0)

3718 || (tmc == 0 && presenceCount == 0 && count forwards == 0)

3719+ -- Watchdog: healthy if recently bumped or in warmup (watchdog starts after 10x prune interval)

3720+ let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)

3724 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),

3721 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),

3725 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,

3722 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,

3726 capacityOk: capacityOk,

3723 capacityOk: capacityOk,

3727 federationOk: fedOk,

3724 federationOk: fedOk,

3728− snapshotOk: snapOk}

3725+ snapshotOk: snapOk,

3726+ watchdogOk: wdOk}

3729− let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk

3727+ let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk

3730 -- Unauthenticated: always return ok:true to prevent information leakage.

3728 -- Unauthenticated: always return ok:true to prevent information leakage.

3731 -- Attackers could correlate ok:false transitions with targeted load to infer subsystem pressure.

3729 -- Attackers could correlate ok:false transitions with targeted load to infer subsystem pressure.

3732 if isAuthed

3730 if isAuthed

@@ -3734,14 +3732,16 @@ handleHealth = \authorization respond -> do

3734 heartbeatHealthy: result.heartbeatOk,

3732 heartbeatHealthy: result.heartbeatOk,

3735 capacityHealthy: result.capacityOk,

3733 capacityHealthy: result.capacityOk,

3736 federationHealthy: result.federationOk,

3734 federationHealthy: result.federationOk,

3737− snapshotHealthy: result.snapshotOk})

3735+ snapshotHealthy: result.snapshotOk,

3736+ watchdogHealthy: result.watchdogOk})

3738 else yield (respond {ok: True {}, pruneHealthy: True {},

3737 else yield (respond {ok: True {}, pruneHealthy: True {},

3739 heartbeatHealthy: True {},

3738 heartbeatHealthy: True {},

3740 capacityHealthy: True {},

3739 capacityHealthy: True {},

3741 federationHealthy: True {},

3740 federationHealthy: True {},

3742− snapshotHealthy: True {}})

3741+ snapshotHealthy: True {},

3742+ watchdogHealthy: True {}})

3743

3744−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, pairRates: 0, federationPairRates: 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, hungFederationSlots: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, drainCompletedAt: 0, oldestForwardAge: 0, oldestAckAge: 0, ok: False {}}

3744+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, pairRates: 0, federationPairRates: 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, hungFederationSlots: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, drainCompletedAt: 0, oldestForwardAge: 0, oldestAckAge: 0, forwardNonces: 0, forwardNoncesMax: 0, pendingAckCleanups: 0, blockedPeers: 0, gossipEventRates: 0, verifyIpRates: 0, ok: False {}}

3745

3746handleStats = \authorization respond -> do

3747 t <- now

@@ -3767,7 +3767,12 @@ handleStats = \authorization respond -> do

3767 rc <- *recipientMsgCounts

3768 fi <- *federationInFlight

3769 rg <- *recentGossip

3770+ fn <- *forwardNonces

3771+ pac <- *pendingAckCleanups

3772+ bp <- *blockedPeers

3773+ ger <- *gossipEventRates

3774+ vir <- *verifyIpRates

3770− yield {sc: sc, pc: pc, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg}

3775+ yield {sc: sc, pc: pc, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg, fn: fn, pac: pac, bp: bp, ger: ger, vir: vir}

3771 counters <- atomic do

3776 counters <- atomic do

3772 sr <- *sendRates

3777 sr <- *sendRates

3773 cr <- *challengeRates

3778 cr <- *challengeRates

@@ -3799,7 +3804,7 @@ handleStats = \authorization respond -> do

3799 let hungFedSlots = count (do r <- core.fi; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs; yield r)

3804 let hungFedSlots = count (do r <- core.fi; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs; yield r)

3800 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt

3805 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt

3801 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt

3806 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt

3802− 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, pairRates: result.pairRates, federationPairRates: result.federationPairRates, 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, hungFederationSlots: hungFedSlots, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, drainCompletedAt: result.drainCompletedAt, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, ok: True {}})

3807+ 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, pairRates: result.pairRates, federationPairRates: result.federationPairRates, 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, hungFederationSlots: hungFedSlots, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, drainCompletedAt: result.drainCompletedAt, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, forwardNonces: count core.fn, forwardNoncesMax: maxForwardNonces, pendingAckCleanups: count core.pac, blockedPeers: count core.bp, gossipEventRates: count core.ger, verifyIpRates: count core.vir, ok: True {}})

3803 else yield (respond zeroStats)

3808 else yield (respond zeroStats)

3804

3809

3805handleStatsPeers = \authorization respond -> do

3810handleStatsPeers = \authorization respond -> do

@@ -3839,7 +3844,7 @@ handleStatsKey = \pubkey authorization respond -> do

3839 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})

3844 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})

3840 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})

3845 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})

3841

3846

3842−emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: []}

3847+emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: []}

3843

3848

3844-- GET /maintenance/snapshot: Export federation metadata for persistence.

3849-- GET /maintenance/snapshot: Export federation metadata for persistence.

3845-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,

3850-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,

@@ -3858,23 +3863,24 @@ handleSnapshot = \authorization respond -> do

3858 ff <- *forwardFailures

3863 ff <- *forwardFailures

3859 f <- *forwards

3864 f <- *forwards

3860 fn <- *forwardNonces

3865 fn <- *forwardNonces

3866+ bp <- *blockedPeers

3861− yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn}

3867+ yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}

3862 atomic do set *lastSnapshotExportAt = t

3868 atomic do set *lastSnapshotExportAt = t

3863− logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards, " ++ show (count result.forwardNonces) ++ " forwardNonces")

3869+ logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards, " ++ show (count result.forwardNonces) ++ " forwardNonces, " ++ show (count result.blockedPeers) ++ " blockedPeers")

3864− yield (respond {ok: True {}, version: snapshotVersion, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces})

3870+ yield (respond {ok: True {}, version: snapshotVersion, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces, blockedPeers: result.blockedPeers})

3865

3871

3866-- POST /maintenance/restore: Import a previously-exported snapshot.

3872-- POST /maintenance/restore: Import a previously-exported snapshot.

3867-- Requires drain mode to prevent conflicts with live state. Merges imported data

3873-- Requires drain mode to prevent conflicts with live state. Merges imported data

3868-- with existing state (dedup entries are unioned, presence is merged by freshness).

3874-- with existing state (dedup entries are unioned, presence is merged by freshness).

3869-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to

3875-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to

3870-- trigger federation forward creation from messages + restored presence.

3876-- trigger federation forward creation from messages + restored presence.

3871−handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData authorization respond -> do

3877+handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData blockedPeersData authorization respond -> do

3872 let authorized = verifyMaintenanceAuth authorization

3878 let authorized = verifyMaintenanceAuth authorization

3873 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})

3879 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})

3874 else do

3880 else do

3875− -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces (accept both for backwards compatibility;

3881+ -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers

3876− -- missing forwardNonces field will default to empty list)

3882+ -- (accept all for backwards compatibility; missing fields default to empty list)

3877− if version != 0 && version != 1 && version != snapshotVersion then do

3883+ if version != 0 && version != 1 && version != 2 && version != snapshotVersion then do

3878 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")

3884 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")

3879 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})

3885 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})

3880 else do

3886 else do

@@ -3928,6 +3934,11 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD

3928 let mergedNonces = mergeByKey (\a b -> a.messageId == b.messageId && a.toKey == b.toKey && a.originServer == b.originServer) (\x -> x.seenAt) forwardNonces validNonces

3934 let mergedNonces = mergeByKey (\a b -> a.messageId == b.messageId && a.toKey == b.toKey && a.originServer == b.originServer) (\x -> x.seenAt) forwardNonces validNonces

3929 let cappedNonces = takeRelation maxForwardNonces mergedNonces

3935 let cappedNonces = takeRelation maxForwardNonces mergedNonces

3930 set *forwardNonces = cappedNonces

3936 set *forwardNonces = cappedNonces

3937+ -- Restore blocked peers (union with existing, cap at maxBlockedPeers)

3938+ blocked <- *blockedPeers

3939+ let mergedBlocked = fold (\acc b -> if any acc (\x -> x == b) then acc else union acc [b]) blocked blockedPeersData

3940+ let cappedBlocked = takeRelation maxBlockedPeers mergedBlocked

3941+ set *blockedPeers = cappedBlocked

3931 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}

3942 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}

3932 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")

3943 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")

3933 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})

3944 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})

@@ -4019,35 +4030,14 @@ handleSendBatch = \messages authorization clientIp respond -> do

4019 let capped = takeRelation maxMessageIds messages

4030 let capped = takeRelation maxMessageIds messages

4020 results <- sendBatchInner senderKey capped t []

4031 results <- sendBatchInner senderKey capped t []

4021 -- Fork federation forwarding for all successful sends

4032 -- Fork federation forwarding for all successful sends

4022− let forwardTargets = do r <- results; where r.needsForward; yield r

4033+ let forwardTargets = do r <- results; where count r.remoteServers > 0; yield r

4023 fork (when (federationEnabled && count forwardTargets > 0)

4034 fork (when (federationEnabled && count forwardTargets > 0)

4024− (sendToEach (collectPeerServers forwardTargets (\r -> r.forwardServer)) (\server -> do

4035+ (sendToEach (collectPeerServers forwardTargets (\r -> case headOf r.remoteServers of Nothing {} -> ""; Just {value: s} -> s)) (\server -> do

4025− let serverMsgs = do r <- forwardTargets; where r.forwardServer == server; yield r

4036+ let serverMsgs = do r <- forwardTargets; where any r.remoteServers (\s -> s == server); yield r

4026 forEachBatch serverMsgs maxMessageIds (\batch -> do

4037 forEachBatch serverMsgs maxMessageIds (\batch -> do

4027− fwdT <- now

4028 forEachBatch batch 1 (\single -> case headOf single of

4038 forEachBatch batch 1 (\single -> case headOf single of

4029 Nothing {} -> yield {}

4039 Nothing {} -> yield {}

4030− Just {value: r} -> do

4040+ Just {value: r} -> forkSendForwards r.messageId r.toKey r.blob r.ts (do s <- r.remoteServers; where s == server; yield s))))))

4031− stillPending <- atomic do

4032− forwards <- *forwards

4033− yield (any forwards (\f -> f.messageId == r.messageId && f.toServer == server))

4034− when stillPending (do

4035− fwdAuth <- generateFederationAuth "forward"

4036− fwdResult <- fetchFederation server (ForwardMessage {

4037− toKey: r.toKey, encryptedBlob: r.blob,

4038− timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,

4039− fromServer: serverHostN, requestTimestamp: fwdT,

4040− authorization: fwdAuth})

4041− case fwdResult of

4042− Err e -> do

4043− logWarn ("federation: batch forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)

4044− recordFailure server

4045− Ok response ->

4046− if response.ok then do

4047− removeForward r.messageId server

4048− clearFailure server

4049− else logWarn ("federation: batch forward to " ++ server ++ " rejected for " ++ r.messageId ++ ": " ++ response.error)))))))

4050− -- Aggregate retryAfterMs: use the maximum across all rate-limited messages

4051 let batchRetryMs = fold (\acc r -> max acc r.retryAfterMs) 0 results

4041 let batchRetryMs = fold (\acc r -> max acc r.retryAfterMs) 0 results

4052 yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}, retryAfterMs: batchRetryMs}))

4042 yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}, retryAfterMs: batchRetryMs}))

4053

4043

@@ -4057,80 +4047,23 @@ sendBatchInner = \senderKey messages t acc ->

4057 Nothing {} -> yield acc

4047 Nothing {} -> yield acc

4058 Just {value: msg} -> do

4048 Just {value: msg} -> do

4059 effectiveMsgId <- if msg.messageId != "" then yield msg.messageId else genToken

4049 effectiveMsgId <- if msg.messageId != "" then yield msg.messageId else genToken

4060− -- Log timestamp anomalies (matching handleSendMessage warnings)

4061 when (msg.timestamp > t + clockSkewTolerance)

4050 when (msg.timestamp > t + clockSkewTolerance)

4062 (logWarn ("sendBatch: future timestamp detected (+" ++ show (msg.timestamp - t) ++ "ms ahead) for " ++ effectiveMsgId))

4051 (logWarn ("sendBatch: future timestamp detected (+" ++ show (msg.timestamp - t) ++ "ms ahead) for " ++ effectiveMsgId))

4063 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)

4052 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)

4064 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))

4053 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))

4065 result <- if t - msg.timestamp > messageMaxAge

4054 result <- if t - msg.timestamp > messageMaxAge

4066− then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0, retryAfterMs: 0}

4055+ then yield {messageId: "", status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, retryAfterMs: 0}

4067− else if senderKey == msg.to

4068− then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0, retryAfterMs: 0}

4069 else do

4056 else do

4070 let clampedTs = clampTimestamp msg.timestamp t

4057 let clampedTs = clampTimestamp msg.timestamp t

4071 sendResult <- atomic do

4058 sendResult <- atomic do

4072 draining <- *draining

4059 draining <- *draining

4073− if draining then yield {status: MsDraining {}, fwdServer: "", needsFwd: False {}, retryMs: 0}

4060+ if draining then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}

4074− else do

4061+ else sendMessageCore senderKey msg.to msg.encryptedBlob clampedTs effectiveMsgId t

4075− let msgBucket = pubkeyBucket msg.to

4076− bucket <- readMessageBucket msgBucket

4077− ackedIds <- *ackedIds

4078− deliveredIds <- *deliveredIds

4079− let isDup = checkMessageDup bucket effectiveMsgId msg.to || (effectiveMsgId != "" && (hasMessage effectiveMsgId msg.to ackedIds || hasMessage effectiveMsgId msg.to deliveredIds))

4080− if isDup then yield {status: MsDuplicate {}, fwdServer: "", needsFwd: False {}, retryMs: 0}

4081− else do

4082− sendRates <- *sendRates

4083− let rateResult = checkPubkeyRate sendRates senderKey maxSendRatePerKey sendRateWindowMs maxSendRateEntries t

4084− if rateResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: rateResult.retryAfterMs}

4085− else do

4086− set *sendRates = rateResult.entries

4087− recvRates <- *recvRates

4088− let recvResult = checkPubkeyRate recvRates msg.to maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t

4089− if recvResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: recvResult.retryAfterMs}

4090− else do

4091− set *recvRates = recvResult.entries

4092− -- Per-sender-per-recipient pair rate limiting (prevents one sender filling a recipient's mailbox)

4093− pairRates <- *pairRates

4094− let pairResult = checkPairRate pairRates senderKey msg.to maxPairRatePerPair pairRateWindowMs maxPairRateEntries t

4095− if pairResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: pairResult.retryAfterMs}

4096− else do

4097− set *pairRates = pairResult.entries

4098− totalMsgCount <- *totalMessageCount

4099− if totalMsgCount >= maxMessages then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}, retryMs: 0}

4100− else do

4101− recipientCounts <- *recipientMsgCounts

4102− let recipientCount = getRecipientCount msg.to recipientCounts

4103− if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}, retryMs: 0}

4104− else do

4105− recipientPresence <- readPresenceBucket (pubkeyBucket msg.to)

4106− seqCounter <- *seqCounter

4107− let seqTs = nextSeqTs t seqCounter

4108− set *seqCounter = seqTs

4109− writeMessageBucket msgBucket (union bucket

4110− [{messageId: effectiveMsgId, toKey: msg.to,

4111− encryptedBlob: msg.encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])

4112− set *totalMessageCount = totalMsgCount + 1

4113− set *recipientMsgCounts = adjustRecipientCount recipientCounts msg.to 1

4114− signalMessageForKey msg.to seqTs

4115− let toPresence = presenceForKey recipientPresence msg.to t

4116− let localOnline = hasLocalPresence toPresence

4117− let remoteServers = collectPeerServers toPresence (\p -> p.server)

4118− let fwdServer = case headOf remoteServers of Nothing {} -> ""; Just {value: s} -> s

4119− when (federationEnabled && count remoteServers > 0) (do

4120− forwards <- *forwards

4121− let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: msg.to, encryptedBlob: msg.encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}

4122− let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds

4123− set *forwards = union forwards cappedFwds)

4124− let delivery = if localOnline then MsDelivered {}

4125− else if federationEnabled && count remoteServers > 0 then MsFederated {}

4126− else MsQueued {}

4127− yield {status: delivery, fwdServer: fwdServer, needsFwd: federationEnabled && count remoteServers > 0, retryMs: 0}

4128− -- Log rate limiting and capacity warnings (matching handleSendMessage)

4129 case sendResult.status of

4062 case sendResult.status of

4130 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)

4063 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)

4131 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")

4064 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")

4132 _ -> yield {}

4065 _ -> yield {}

4133− yield {messageId: effectiveMsgId, status: sendResult.status, needsForward: sendResult.needsFwd, forwardServer: sendResult.fwdServer, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryMs}

4066+ yield {messageId: effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryAfterMs}

4134 sendBatchInner senderKey split.rest t (union acc [result])

4067 sendBatchInner senderKey split.rest t (union acc [result])

4135

4068

4136-- POST /maintenance/block: Add a server to the federation blocklist.

4069-- POST /maintenance/block: Add a server to the federation blocklist.

@@ -4143,8 +4076,11 @@ handleBlockPeer = \server authorization respond -> do

4143 blocked <- *blockedPeers

4076 blocked <- *blockedPeers

4144 if isBlockedPeer server blocked then yield {alreadyBlocked: True {}, total: count blocked}

4077 if isBlockedPeer server blocked then yield {alreadyBlocked: True {}, total: count blocked}

4145 else do

4078 else do

4079+ -- FIFO eviction: remove the oldest entry (head of list) since union appends

4146 let updated = if count blocked >= maxBlockedPeers

4080 let updated = if count blocked >= maxBlockedPeers

4147− then union (evictOneOldest blocked (\_ -> 0)) [server]

4081+ then case headOf blocked of

4082+ Nothing {} -> [server]

4083+ Just {value: oldest} -> union (do b <- blocked; where b != oldest; yield b) [server]

4148 else union blocked [server]

4084 else union blocked [server]

4149 set *blockedPeers = updated

4085 set *blockedPeers = updated

4150 yield {alreadyBlocked: False {}, total: count updated}

4086 yield {alreadyBlocked: False {}, total: count updated}

@@ -4173,25 +4109,26 @@ handleListBlocked = \authorization respond -> do

4173 yield (respond {peers: blocked, ok: True {}})

4109 yield (respond {peers: blocked, ok: True {}})

4174

4110

4175-- POST /maintenance/ping: Test federation connectivity to a specific peer.

4111-- POST /maintenance/ping: Test federation connectivity to a specific peer.

4176−-- Does a verify callback round-trip to measure latency and confirm reachability.

4112+-- Sends an empty gossip request via fetchFederation, triggering a full verify callback

4113+-- round-trip (peer calls back to our /federation/verify to authenticate the request).

4114+-- This exercises the real federation auth path and tracks in-flight/latency metrics.

4177handlePingPeer = \server authorization respond -> do

4115handlePingPeer = \server authorization respond -> do

4178 let authorized = verifyMaintenanceAuth authorization

4116 let authorized = verifyMaintenanceAuth authorization

4179 if not authorized then yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"})

4117 if not authorized then yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"})

4180 else do

4118 else do

4181− token <- genToken

4182− t <- now

4183− atomic do

4184− pending <- *pendingVerifications

4185− set *pendingVerifications = union pending [{token: token, createdAt: t, scope: "ping"}]

4186 startTime <- now

4119 startTime <- now

4187− result <- fetch (federationUrl server) (VerifyToken {token: token, scope: "ping"})

4120+ auth <- generateFederationAuth "gossip"

4121+ result <- fetchFederation server (RecvGossip {

4122+ authorization: auth,

4123+ events: [], hopsRemaining: 1, fromServer: serverHostN,

4124+ requestTimestamp: startTime})

4188 endTime <- now

4125 endTime <- now

4189 let latency = endTime - startTime

4126 let latency = endTime - startTime

4190 case result of

4127 case result of

4191 Err e -> do

4128 Err e -> do

4192 logWarn ("ping: " ++ server ++ " failed: " ++ show e)

4129 logWarn ("ping: " ++ server ++ " failed: " ++ show e)

4193 yield (respond {ok: False {}, latencyMs: latency, error: show e})

4130 yield (respond {ok: False {}, latencyMs: latency, error: show e})

4194− Ok r -> yield (respond {ok: r.ok, latencyMs: latency, error: ""})

4131+ Ok r -> yield (respond {ok: r.ok, latencyMs: latency, error: r.error})

4195

4132

4196-- GET /maintenance/sessions/{pubkey}: List sessions for a specific pubkey.

4133-- GET /maintenance/sessions/{pubkey}: List sessions for a specific pubkey.

4197-- Returns masked tokens (first 8 chars only) for identification without exposing full tokens.

4134-- Returns masked tokens (first 8 chars only) for identification without exposing full tokens.

@@ -4216,42 +4153,7 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do

4216 allSessions <- readAllSessions

4153 allSessions <- readAllSessions

4217 let revokedCount = count (do s <- allSessions; where s.pubkey == pubkey; yield s)

4154 let revokedCount = count (do s <- allSessions; where s.pubkey == pubkey; yield s)

4218 let remaining = do s <- allSessions; where s.pubkey != pubkey; yield s

4155 let remaining = do s <- allSessions; where s.pubkey != pubkey; yield s

4219− -- Single-pass partition by token bucket

4156+ writeAllSessions remaining

4220− let partitioned = fold (\acc s ->

4221− let b = tokenBucket s.token in

4222− {b0: if b == 0 then union [s] acc.b0 else acc.b0,

4223− b1: if b == 1 then union [s] acc.b1 else acc.b1,

4224− b2: if b == 2 then union [s] acc.b2 else acc.b2,

4225− b3: if b == 3 then union [s] acc.b3 else acc.b3,

4226− b4: if b == 4 then union [s] acc.b4 else acc.b4,

4227− b5: if b == 5 then union [s] acc.b5 else acc.b5,

4228− b6: if b == 6 then union [s] acc.b6 else acc.b6,

4229− b7: if b == 7 then union [s] acc.b7 else acc.b7,

4230− b8: if b == 8 then union [s] acc.b8 else acc.b8,

4231− b9: if b == 9 then union [s] acc.b9 else acc.b9,

4232− b10: if b == 10 then union [s] acc.b10 else acc.b10,

4233− b11: if b == 11 then union [s] acc.b11 else acc.b11,

4234− b12: if b == 12 then union [s] acc.b12 else acc.b12,

4235− b13: if b == 13 then union [s] acc.b13 else acc.b13,

4236− b14: if b == 14 then union [s] acc.b14 else acc.b14,

4237− b15: if b == 15 then union [s] acc.b15 else acc.b15})

4238− {b0: [], b1: [], b2: [], b3: [], b4: [], b5: [], b6: [], b7: [], b8: [], b9: [], b10: [], b11: [], b12: [], b13: [], b14: [], b15: []} remaining

4239− writeSessionBucket 0 partitioned.b0

4240− writeSessionBucket 1 partitioned.b1

4241− writeSessionBucket 2 partitioned.b2

4242− writeSessionBucket 3 partitioned.b3

4243− writeSessionBucket 4 partitioned.b4

4244− writeSessionBucket 5 partitioned.b5

4245− writeSessionBucket 6 partitioned.b6

4246− writeSessionBucket 7 partitioned.b7

4247− writeSessionBucket 8 partitioned.b8

4248− writeSessionBucket 9 partitioned.b9

4249− writeSessionBucket 10 partitioned.b10

4250− writeSessionBucket 11 partitioned.b11

4251− writeSessionBucket 12 partitioned.b12

4252− writeSessionBucket 13 partitioned.b13

4253− writeSessionBucket 14 partitioned.b14

4254− writeSessionBucket 15 partitioned.b15

4255 set *totalSessionCount = count remaining

4157 set *totalSessionCount = count remaining

4256 -- Clear local presence

4158 -- Clear local presence

4257 let pb = pubkeyBucket pubkey

4159 let pb = pubkeyBucket pubkey

@@ -4289,7 +4191,7 @@ serve = \req -> case req of

4289 Drain {authorization, respond} -> handleDrain authorization respond

4191 Drain {authorization, respond} -> handleDrain authorization respond

4290 Undrain {authorization, respond} -> handleUndrain authorization respond

4192 Undrain {authorization, respond} -> handleUndrain authorization respond

4291 Snapshot {authorization, respond} -> handleSnapshot authorization respond

4193 Snapshot {authorization, respond} -> handleSnapshot authorization respond

4292− Restore {version, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, authorization, respond} -> handleRestore version presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces authorization respond

4194+ Restore {version, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, blockedPeers, authorization, respond} -> handleRestore version presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces blockedPeers authorization respond

4293 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond

4195 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond

4294 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond

4196 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond

4295 ListBlocked {authorization, respond} -> handleListBlocked authorization respond

4197 ListBlocked {authorization, respond} -> handleListBlocked authorization respond

@@ -4432,6 +4334,7 @@ backgroundSnapshot = do

4432backgroundWatchdog = do

4334backgroundWatchdog = do

4433 sleep (backgroundPruneInterval * 5)

4335 sleep (backgroundPruneInterval * 5)

4434 t <- now

4336 t <- now

4337+ atomic do set *lastWatchdogAt = t

4435 sa <- *startedAt

4338 sa <- *startedAt

4436 let warmup = t - sa < backgroundPruneInterval * 10

4339 let warmup = t - sa < backgroundPruneInterval * 10

4437 if warmup then backgroundWatchdog

4340 if warmup then backgroundWatchdog