53a4c04 · skrepka · git.iko.soy

git.iko.soy/skrepka

Add batch send endpoint, snapshot versioning, structured logging, clock jump detection, and optimize server.knot security, performance, and reliability

commit53a4c046317e6aa90cecc0ef0fd0e43c65b81b71 authoriko <mail@iko.soy> dateSat 25 Apr 2026, 10:28 (6 months ago) parentca4b8d3

1 file changed+333−129

compare with master

unifiedsplit

server.knot+333−129

server.knot+333−129

@@ -136,6 +136,9 @@ reconciliationFrequency = 10

136136federationLatencyWindowMs : Int<Ms>

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

138138

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

140+snapshotVersion = 1

141+

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

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

141144gossipEventRateWindowMs : Int<Ms>

@@ -161,6 +164,10 @@ presenceRefreshInterval = 300000 -- 5 minutes

161164presenceRefreshThreshold : Int<Ms>

162165presenceRefreshThreshold = presenceRefreshInterval * 2 -- refresh entries expiring within this window

163166federationEnabled = True {}

167+-- WARNING: DNS rebinding risk — federationUrl resolves DNS at fetch time, not at

168+-- ServerName validation time. A hostname passing isBadServerName could rebind to a

169+-- private IP between validation and fetch. Mitigate at the network layer (e.g.,

170+-- firewall rules blocking outbound connections to RFC1918 ranges from the server process).

164171federationUrl = \server -> federationScheme ++ server

165172-- Callback verification: outbound federation requests include a random single-use token;

166173-- the receiver calls back to /federation/verify to confirm the sender actually issued it.

@@ -303,6 +310,10 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest

303310type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}

304311type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}

305312

313+-- Batch send types

314+type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId}

315+type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}

316+

306317-- ============================================

307318-- Persisted State

308319-- ============================================

@@ -499,8 +510,8 @@ type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}

499510-- scope binds the token to a specific request type (gossip/forward/ack) to prevent cross-use

500511*pendingVerifications : [{token: Text, createdAt: Timestamp, scope: Text}]

501512

502−-- Forward nonce dedup: (messageId, toKey, seenAt) for replay protection beyond message lifetime

503−*forwardNonces : [{messageId: MessageIdHex, toKey: PubkeyHex, seenAt: Timestamp}]

513+-- Forward nonce dedup: (messageId, toKey, originServer, seenAt) for replay protection beyond message lifetime

514+*forwardNonces : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}]

504515

505516-- Federation verify per-IP rate limiting (tighter than general IP rate)

506517*verifyIpRates : [IpRateEntry]

@@ -514,6 +525,10 @@ type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}

514525-- Global federation connection counter (tracks total concurrent outbound connections)

515526*federationActiveConnections : Int

516527

528+-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)

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

530+*presenceVersion : Timestamp

531+

517532-- ============================================

518533-- Routes

519534-- ============================================

@@ -557,6 +572,10 @@ route ClientApi where

557572 POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text, clientIp: Text}

558573 -> {ok: Bool, authorized: Bool, ackedCount: Int}

559574 = AckMessages

575+ -- Send multiple messages in one request (auth checked once; each message rate-limited independently)

576+ POST {messages: [BatchMessage]} /batch headers {authorization: Text, clientIp: Text}

577+ -> {results: [BatchResult], authorized: Bool}

578+ = SendBatch

560579 -- Look up which server a pubkey is present on (for federation routing)

561580 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}

562581 -> {server: AnyServerName, authorized: Bool}

@@ -598,11 +617,11 @@ route MaintenanceApi where

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

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

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

601− -> {ok: Bool, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry]}

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

602621 = Snapshot

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

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

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

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

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

607626 = Restore

608627

@@ -612,7 +631,7 @@ route HealthApi where

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

613632route StatsApi where

614633 GET /stats headers {authorization: Text}

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

634+ -> {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}

616635 = Stats

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

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

@@ -627,6 +646,11 @@ route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | Maintena

627646-- Helpers

628647-- ============================================

629648

649+-- Structured log helpers (severity-prefixed for operator filtering)

650+logInfo = \msg -> println ("[INFO] " ++ msg)

651+logWarn = \msg -> println ("[WARN] " ++ msg)

652+logError = \msg -> println ("[ERROR] " ++ msg)

653+

630654any : [a] -> (a -> Bool) -> Bool

631655any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs

632656none : [a] -> (a -> Bool) -> Bool

@@ -916,7 +940,7 @@ verifyFederationCallback = \fromServer authorization scope ->

916940 result <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})

917941 case result of

918942 Err e -> do

919− println ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)

943+ logWarn ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)

920944 yield (False {})

921945 Ok r -> yield r.ok

922946

@@ -1020,6 +1044,20 @@ hasSessionForKey = \pubkey b ->

10201044 if any bucket (\s -> s.pubkey == pubkey) then yield (True {})

10211045 else hasSessionForKey pubkey (b + 1)

10221046

1047+-- Count sessions for a specific pubkey across all buckets (avoids materializing full session list).

1048+-- Used in verify fast path to skip readAllSessions.

1049+countSessionsForKey : PubkeyHex -> IO Int

1050+countSessionsForKey = \pubkey -> countSessionsInBucket pubkey 0

1051+

1052+countSessionsInBucket : PubkeyHex -> Int -> IO Int

1053+countSessionsInBucket = \pubkey b ->

1054+ if b > 15 then yield 0

1055+ else do

1056+ bucket <- readSessionBucket b

1057+ let n = count (do s <- bucket; where s.pubkey == pubkey; yield s)

1058+ rest <- countSessionsInBucket pubkey (b + 1)

1059+ yield (n + rest)

1060+

10231061-- Apply a filter/transform to every session bucket

10241062transformAllSessionBuckets : ([Session] -> [Session]) -> IO ()

10251063transformAllSessionBuckets = \f -> do

@@ -1124,6 +1162,17 @@ countAllPresence = do

11241162 p15 <- *presenceBucket15

11251163 yield (count p0 + count p1 + count p2 + count p3 + count p4 + count p5 + count p6 + count p7 + count p8 + count p9 + count p10 + count p11 + count p12 + count p13 + count p14 + count p15)

11261164

1165+-- Read only the presence buckets at the given indices (avoids reading all 16)

1166+readSelectedPresenceBuckets : [Int] -> IO [PresenceEntry]

1167+readSelectedPresenceBuckets = \buckets ->

1168+ case headOf buckets of

1169+ Nothing {} -> yield []

1170+ Just {value: b} -> do

1171+ let rest = do x <- buckets; where x != b; yield x

1172+ entries <- readPresenceBucket b

1173+ remaining <- readSelectedPresenceBuckets rest

1174+ yield (union entries remaining)

1175+

11271176-- Single-pass partitioning: each entry's bucket is computed once (vs 16x in the list-comprehension approach)

11281177writeAllPresence : [PresenceEntry] -> IO ()

11291178writeAllPresence = \entries -> do

@@ -1374,6 +1423,9 @@ hasHungFederationRequest = \server t inFlight ->

13741423

13751424-- Wrapper around fetch for federation calls: tracks in-flight state, global connection count,

13761425-- logs slow responses, and records latency metrics.

1426+-- NOTE: If fetch panics (runtime error), in-flight and connection counters leak until

1427+-- pruneExpired reclaims them (stale threshold: 1.5x federationFetchTimeoutMs). Knot

1428+-- lacks try-finally, so this safety net in pruneExpired is the cleanup mechanism.

13771429fetchFederation = \server request -> do

13781430 -- Acquire global federation connection slot

13791431 atomic do

@@ -1390,7 +1442,7 @@ fetchFederation = \server request -> do

13901442 endTime <- now

13911443 let latency = endTime - startTime

13921444 when (latency > federationFetchTimeoutMs)

1393− (println ("federation: slow response from " ++ server ++ " (" ++ show latency ++ "ms)"))

1445+ (logWarn ("federation: slow response from " ++ server ++ " (" ++ show latency ++ "ms)"))

13941446 atomic do

13951447 total <- *federationLatencyTotal

13961448 cnt <- *federationLatencyCount

@@ -1659,6 +1711,7 @@ pruneExpired = \t -> do

16591711 tpc <- *totalPresenceCount

16601712 set *totalPresenceCount = max 0 (tpc - presenceResults.totalRemoved)

16611713 set *knownPeerServers = presenceResults.keptPeerServers

1714+ when (presenceResults.totalRemoved > 0) (set *presenceVersion = t)

16621715 let presenceResult = {expiredLocal: presenceResults.expiredLocal, peerServers: presenceResults.allPeerServers}

16631716 -- Prune auth state

16641717 atomic do

@@ -1741,7 +1794,7 @@ pruneExpired = \t -> do

17411794 let staleThreshold = federationFetchTimeoutMs * 3 / 2

17421795 let staleInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= staleThreshold; yield r

17431796 when (count staleInFlight > 0)

1744− (println ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))

1797+ (logWarn ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))

17451798 set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < staleThreshold); yield r

17461799 -- Prune stale federation replay dedup entries

17471800 recentFedReqs <- *recentFederationRequests

@@ -1785,7 +1838,7 @@ sendBatchedGossipToServer = \server allEvents hops ->

17851838 events: batch, hopsRemaining: hops, fromServer: serverHostN,

17861839 requestTimestamp: t})

17871840 case result of

1788− Err e -> println ("federation: gossip to " ++ server ++ " failed: " ++ show e)

1841+ Err e -> logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)

17891842 Ok _ -> yield {})

17901843

17911844sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}

@@ -1796,11 +1849,11 @@ broadcastOfflineGossip = \t peerServers expiredLocal ->

17961849 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do

17971850 failures <- *forwardFailures

17981851 if isFederationCircuitOpen failures t

1799− then println ("federation: circuit breaker open, skipping offline gossip broadcast")

1852+ then logWarn ("federation: circuit breaker open, skipping offline gossip broadcast")

18001853 else do

18011854 let cappedExpired = takeRelation maxPresenceEntries expiredLocal

18021855 when (count expiredLocal > maxPresenceEntries)

1803− (println ("gossip: capped offline broadcast to " ++ show maxPresenceEntries ++ " entries (had " ++ show (count expiredLocal) ++ ")"))

1856+ (logWarn ("gossip: capped offline broadcast to " ++ show maxPresenceEntries ++ " entries (had " ++ show (count expiredLocal) ++ ")"))

18041857 let gossipEvents = mergeByKey (\a b -> a.pubkey == b.pubkey) (\x -> x.timestamp) [] (offlineGossipEvents cappedExpired)

18051858 sendBatchedGossip peerServers gossipEvents gossipOriginHops)

18061859

@@ -1893,7 +1946,7 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp

18931946 -- requestTimestamp == 0 means the peer doesn't support this field yet (accept).

18941947 else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance)

18951948 then do

1896− println ("federation: clock skew " ++ show (t - requestTimestamp) ++ "ms from " ++ fromServer ++ ", rejecting (tolerance: " ++ show clockSkewTolerance ++ "ms)")

1949+ logWarn ("federation: clock skew " ++ show (t - requestTimestamp) ++ "ms from " ++ fromServer ++ ", rejecting (tolerance: " ++ show clockSkewTolerance ++ "ms)")

18971950 yield (respond {ok: False {}, error: "clock_skew"})

18981951 else if not precondition

18991952 then yield (respond {ok: False {}, error: "invalid_request"})

@@ -1912,19 +1965,19 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp

19121965 else yield (False {})

19131966 if replayDetected

19141967 then do

1915− println ("federation: replay detected from " ++ fromServer ++ " (timestamp: " ++ show requestTimestamp ++ ")")

1968+ logWarn ("federation: replay detected from " ++ fromServer ++ " (timestamp: " ++ show requestTimestamp ++ ")")

19161969 yield (respond {ok: False {}, error: "replay_detected"})

19171970 else do

19181971 globalLimited <- checkGlobalRate t

19191972 if globalLimited

19201973 then do

1921− println ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)

1974+ logWarn ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)

19221975 yield (respond {ok: False {}, error: "rate_limited"})

19231976 else do

19241977 rateLimited <- checkFederationRate fromServer t

19251978 if rateLimited

19261979 then do

1927− println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)

1980+ logWarn ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)

19281981 yield (respond {ok: False {}, error: "rate_limited"})

19291982 else do

19301983 clearFailure fromServer

@@ -1960,13 +2013,14 @@ ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of

19602013 let remainingBuckets = do x <- bucketIds; where x != b; yield x

19612014 let bucketEntries = do e <- entries; where pubkeyBucket e.toKey == b; yield e

19622015 atomic do

1963− allPresence <- readAllPresence

2016+ -- All entries in this batch share pubkeyBucket == b, so only that presence bucket is needed

2017+ bucketPresence <- readPresenceBucket b

19642018 forwards <- *forwards

19652019 msgs <- readMessageBucket b

19662020 tc <- *totalMessageCount

19672021 rc <- *recipientMsgCounts

19682022 let cleanup = fold (\acc rf ->

1969− let hasLocal = any allPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in

2023+ let hasLocal = any bucketPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in

19702024 let hasOtherFwds = any forwards (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey) in

19712025 if hasLocal || hasOtherFwds then acc

19722026 else case headOf (do m <- acc.msgs; where m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod; yield m) of

@@ -2029,7 +2083,7 @@ incrementForwardRetries = \messageId toServer -> atomic do

20292083 if shouldDrop

20302084 then do

20312085 set *forwards = do f <- forwards; where not (f.messageId == messageId && f.toServer == toServer); yield f

2032− println ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")

2086+ logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")

20332087 else set *forwards = do f <- forwards; yield (if f.messageId == messageId && f.toServer == toServer then {messageId: f.messageId, toServer: f.toServer, createdAt: f.createdAt, toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, retries: f.retries + 1} else f)

20342088

20352089retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}

@@ -2049,7 +2103,7 @@ retryForwardsForServer = \entries remaining ->

20492103 authorization: auth})

20502104 case result of

20512105 Err e -> do

2052− println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)

2106+ logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)

20532107 recordFailure f.toServer

20542108 incrementForwardRetries f.messageId f.toServer

20552109 -- Network error: stop retrying this server

@@ -2058,7 +2112,7 @@ retryForwardsForServer = \entries remaining ->

20582112 removeForward f.messageId f.toServer

20592113 clearFailure f.toServer

20602114 else do

2061− println ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)

2115+ logWarn ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)

20622116 incrementForwardRetries f.messageId f.toServer

20632117 retryForwardsForServer rest (remaining - 1)

20642118

@@ -2070,7 +2124,7 @@ retryPendingForwards = \pending -> do

20702124 t <- now

20712125 failures <- *forwardFailures

20722126 if isFederationCircuitOpen failures t

2073− then println ("federation: circuit breaker open (" ++ show (count failures) ++ " peers failing), skipping forward retries")

2127+ then logWarn ("federation: circuit breaker open (" ++ show (count failures) ++ " peers failing), skipping forward retries")

20742128 else do

20752129 inFlight <- *federationInFlight

20762130 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s

@@ -2096,7 +2150,7 @@ forwardQueuedMessages = \targets t -> do

20962150 let servers = collectPeerServers newForwards (\x -> x.toServer)

20972151 failures <- *forwardFailures

20982152 if isFederationCircuitOpen failures t

2099− then println ("federation: circuit breaker open, skipping queued message forwarding")

2153+ then logWarn ("federation: circuit breaker open, skipping queued message forwarding")

21002154 else do

21012155 inFlight <- *federationInFlight

21022156 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s

@@ -2118,7 +2172,7 @@ sendBatchedAcksToServer = \serverAcks server ->

21182172 result <- fetchFederationAck server (do a <- batch; yield a.messageId)

21192173 case result of

21202174 Err e -> do

2121− println ("federation: ack to " ++ server ++ " failed: " ++ show e)

2175+ logWarn ("federation: ack to " ++ server ++ " failed: " ++ show e)

21222176 recordFailure server

21232177 -- Increment retry counters; drop entries exceeding maxAckRetries to deliveredIds

21242178 dropped <- atomic do

@@ -2136,7 +2190,7 @@ sendBatchedAcksToServer = \serverAcks server ->

21362190 set *deliveredIds = if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined)

21372191 yield dropCount

21382192 when (dropped > 0)

2139− (println ("federation: dropped " ++ show dropped ++ " ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))

2193+ (logWarn ("federation: dropped " ++ show dropped ++ " ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))

21402194 Ok _ -> do

21412195 clearFailure server

21422196 atomic do

@@ -2151,7 +2205,7 @@ sendAcksAndClean = \pendingAcks -> do

21512205 t <- now

21522206 failures <- *forwardFailures

21532207 if isFederationCircuitOpen failures t

2154− then println ("federation: circuit breaker open, skipping ack retries")

2208+ then logWarn ("federation: circuit breaker open, skipping ack retries")

21552209 else do

21562210 inFlight <- *federationInFlight

21572211 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s

@@ -2169,7 +2223,7 @@ retryEvictedAcks = \evictedEntries pendingAcks ->

21692223 fetchResult <- fetchFederationAck server (do a <- acksForServer evictedToRetry server; yield a.messageId)

21702224 case fetchResult of

21712225 Err e -> do

2172− println ("federation: evicted ack to " ++ server ++ " failed: " ++ show e)

2226+ logWarn ("federation: evicted ack to " ++ server ++ " failed: " ++ show e)

21732227 overflow <- atomic do

21742228 ackedIds <- *ackedIds

21752229 let reinsert = do a <- acksForServer evictedToRetry server; yield {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1}

@@ -2198,10 +2252,10 @@ retryEvictedAcks = \evictedEntries pendingAcks ->

21982252 set *deliveredIds = if deliveredExcess <= 0 then combinedDelivered

21992253 else (evictSmallest combinedDelivered (max 0 deliveredExcess) (\d -> d.deliveredAt)).kept

22002254 when (deliveredExcess > 0)

2201− (println ("ackMessages: deliveredIds at capacity (" ++ show maxDeliveredIds ++ "), " ++ show deliveredExcess ++ " oldest dedup entries permanently removed for " ++ server))

2255+ (logWarn ("ackMessages: deliveredIds at capacity (" ++ show maxDeliveredIds ++ "), " ++ show deliveredExcess ++ " oldest dedup entries permanently removed for " ++ server))

22022256 yield overflowExcess

22032257 when (dropped > 0)

2204− (println ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries moved to delivered set for " ++ server)))

2258+ (logWarn ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries moved to delivered set for " ++ server)))

22052259 Ok _ -> yield {})

22062260

22072261-- Retry acks sitting in the overflow buffer (last chance before they become deliveredIds)

@@ -2210,7 +2264,7 @@ retryOverflowAcks = \overflow -> do

22102264 t <- now

22112265 failures <- *forwardFailures

22122266 if isFederationCircuitOpen failures t

2213− then println ("federation: circuit breaker open, skipping overflow ack retries")

2267+ then logWarn ("federation: circuit breaker open, skipping overflow ack retries")

22142268 else do

22152269 inFlight <- *federationInFlight

22162270 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s

@@ -2219,7 +2273,7 @@ retryOverflowAcks = \overflow -> do

22192273 result <- fetchFederationAck server (do a <- batch; yield a.messageId)

22202274 case result of

22212275 Err e -> do

2222− println ("federation: overflow ack to " ++ server ++ " failed: " ++ show e)

2276+ logWarn ("federation: overflow ack to " ++ server ++ " failed: " ++ show e)

22232277 recordFailure server

22242278 -- Increment retry counters in overflow; drop entries exceeding maxAckRetries to deliveredIds

22252279 dropped <- atomic do

@@ -2237,7 +2291,7 @@ retryOverflowAcks = \overflow -> do

22372291 set *deliveredIds = if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined)

22382292 yield dropCount

22392293 when (dropped > 0)

2240− (println ("federation: dropped " ++ show dropped ++ " overflow ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))

2294+ (logWarn ("federation: dropped " ++ show dropped ++ " overflow ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))

22412295 Ok _ -> do

22422296 clearFailure server

22432297 atomic do

@@ -2266,7 +2320,7 @@ waitForInFlightDrain = \deadline -> do

22662320 if activeCount <= 0 then yield {}

22672321 else retry

22682322 timedOut <- *drainTimedOut

2269− when timedOut (println "drain: timed out waiting for in-flight federation tasks")

2323+ when timedOut (logWarn "drain: timed out waiting for in-flight federation tasks")

22702324

22712325-- Full maintenance cycle: prune expired state, retry pending forwards and acks,

22722326-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys

@@ -2288,7 +2342,7 @@ pruneAndRetry = \t -> do

22882342 set *totalSessionCount = actual

22892343 yield (actual - stored)

22902344 when (presDrift != 0 || sessDrift != 0)

2291− (println ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))

2345+ (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))

22922346 -- Message reconciliation is expensive (reads all 16 buckets + rebuilds recipient counts), keep periodic

22932347 pc <- *pruneCount

22942348 when (pc > 0 && pc - (pc / reconciliationFrequency) * reconciliationFrequency == 0) (do

@@ -2300,7 +2354,7 @@ pruneAndRetry = \t -> do

23002354 set *recipientMsgCounts = countByRecipient allMsgs

23012355 yield (actual - stored)

23022356 when (msgDrift != 0)

2303− (println ("reconciliation: corrected message drift: " ++ show msgDrift)))

2357+ (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift)))

23042358 pendingForwards <- atomic do f <- *forwards; yield f

23052359 ackState <- atomic do

23062360 ackedIds2 <- *ackedIds

@@ -2311,19 +2365,22 @@ pruneAndRetry = \t -> do

23112365 refreshResult <- atomic do

23122366 lastRefresh <- *lastPresenceRefresh

23132367 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval

2314− allPresence <- readAllPresence

2315− allSessions <- readAllSessions

2316− -- Extend presence for keys with active sessions nearing expiry (sliding window)

2317− let localNearExpiry = if shouldRefresh then (do p <- allPresence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []

2318− let toExtend = do p <- localNearExpiry; where any allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p

2319− let notExtended = do p <- localNearExpiry; where none allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p

2320− when (count toExtend > 0) (do

2321− let updatedPresence = do pp <- allPresence; yield (if any toExtend (\e -> e.pubkey == pp.pubkey && isLocalServer pp.server) then {pubkey: pp.pubkey, server: pp.server, expiresAt: t + sessionExpiry, timestamp: t} else pp)

2322− writeAllPresence updatedPresence)

2323− knownPeers <- *knownPeerServers

2324− let refreshPeers = if shouldRefresh then knownPeers else []

2325− when shouldRefresh (set *lastPresenceRefresh = t)

2326− yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}

2368+ -- Early return: skip expensive readAllPresence + readAllSessions on ~9/10 prune cycles

2369+ if not shouldRefresh then yield {extendedActive: [], nearExpiryActive: [], refreshPeers: [], shouldRefresh: False {}}

2370+ else do

2371+ allPresence <- readAllPresence

2372+ allSessions <- readAllSessions

2373+ -- Extend presence for keys with active sessions nearing expiry (sliding window)

2374+ let localNearExpiry = do p <- allPresence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p

2375+ let toExtend = do p <- localNearExpiry; where any allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p

2376+ let notExtended = do p <- localNearExpiry; where none allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p

2377+ when (count toExtend > 0) (do

2378+ let updatedPresence = do pp <- allPresence; yield (if any toExtend (\e -> e.pubkey == pp.pubkey && isLocalServer pp.server) then {pubkey: pp.pubkey, server: pp.server, expiresAt: t + sessionExpiry, timestamp: t} else pp)

2379+ writeAllPresence updatedPresence

2380+ set *presenceVersion = t)

2381+ knownPeers <- *knownPeerServers

2382+ set *lastPresenceRefresh = t

2383+ yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: knownPeers, shouldRefresh: True {}}

23272384 -- Reset federation latency metrics periodically for windowed averages (prevents

23282385 -- lifetime-cumulative average from masking current conditions)

23292386 atomic do

@@ -2649,7 +2706,7 @@ handleChallenge = \pubkey clientIp respond -> do

26492706 preAuth <- checkPreAuth clientIp t

26502707 if preAuth != ""

26512708 then do

2652− when (preAuth == "ip_rate_limited") (println ("challenge: IP rate limited for " ++ clientIp))

2709+ when (preAuth == "ip_rate_limited") (logWarn ("challenge: IP rate limited for " ++ clientIp))

26532710 yield (respond {challenge: "", ok: False {}, expiresAt: 0})

26542711 else do

26552712 authLimited <- checkAuthGlobalRate t

@@ -2681,9 +2738,9 @@ handleChallenge = \pubkey clientIp respond -> do

26812738 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]

26822739 yield {accepted: True {}, rateLimited: False {}}

26832740 when result.rateLimited

2684− (println ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))

2741+ (logWarn ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))

26852742 when (not result.accepted && not result.rateLimited)

2686− (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))

2743+ (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))

26872744 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})

26882745

26892746-- POST /auth/verify: Verify Ed25519 signature over challenge string.

@@ -2694,7 +2751,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->

26942751 preAuth <- checkPreAuth clientIp t

26952752 if preAuth != ""

26962753 then do

2697− when (preAuth == "ip_rate_limited") (println ("verify: IP rate limited for " ++ clientIp))

2754+ when (preAuth == "ip_rate_limited") (logWarn ("verify: IP rate limited for " ++ clientIp))

26982755 yield (respond {token: "", expiresAt: 0})

26992756 else do

27002757 authRateLimited <- atomic do

@@ -2706,7 +2763,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->

27062763 set *challengeRates = rateResult.entries

27072764 yield rateResult.limited

27082765 if authRateLimited then do

2709− println ("verify: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey)

2766+ logWarn ("verify: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey)

27102767 yield (respond {token: "", expiresAt: 0})

27112768 else do

27122769 authGlobalLimited <- checkAuthGlobalRate t

@@ -2742,12 +2799,11 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->

27422799 isDraining <- *draining

27432800 if isDraining then yield (Nothing {})

27442801 else do

2745− -- Read all session buckets for per-key enforcement (revokeOthers, capPerKey)

2746− allSessions <- readAllSessions

27472802 tpc <- *totalPresenceCount

27482803 knownPeers <- *knownPeerServers

2749− let sessionsForKey = count (do s <- allSessions; where s.pubkey == pubkey; yield s)

2750− let totalSessions = count allSessions

2804+ -- Count per-key sessions without materializing all sessions (O(sessions) but no union)

2805+ sessionsForKey <- countSessionsForKey pubkey

2806+ tsc <- *totalSessionCount

27512807 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp}

27522808 -- Update presence: replace local entry for this pubkey, evict oldest if over capacity

27532809 let pb = pubkeyBucket pubkey

@@ -2759,18 +2815,19 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->

27592815 writePresenceBucket pb (union bucketEvicted [newPresenceEntry])

27602816 let presenceDelta = count (union bucketEvicted [newPresenceEntry]) - count presenceBkt

27612817 set *totalPresenceCount = tpc + presenceDelta

2818+ set *presenceVersion = t

27622819 let peerServers = if federationEnabled then knownPeers else []

27632820 -- Fast path: no revocation, under per-key and global limits — only write one bucket

2764− if not revokeOthers && sessionsForKey < maxSessionsPerKey && totalSessions < maxSessions

2821+ if not revokeOthers && sessionsForKey < maxSessionsPerKey && tsc < maxSessions

27652822 then do

27662823 let nb = tokenBucket token

27672824 currentBucket <- readSessionBucket nb

27682825 writeSessionBucket nb (union currentBucket [newSession])

2769− sc <- *totalSessionCount

2770− set *totalSessionCount = sc + 1

2826+ set *totalSessionCount = tsc + 1

27712827 yield (Just {value: {peerServers: peerServers, revokedCount: 0}})

27722828 else do

2773− -- Slow path: revocation or capacity pressure — redistribute across all buckets

2829+ -- Slow path: revocation or capacity pressure — need all sessions

2830+ allSessions <- readAllSessions

27742831 let revokedCount = if revokeOthers then sessionsForKey else 0

27752832 let afterRevoke = if revokeOthers then do s <- allSessions; where s.pubkey != pubkey; yield s else allSessions

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

@@ -2798,7 +2855,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->

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

27992856 Just {value: result} -> do

28002857 when (revokeOthers && result.revokedCount > 0)

2801− (println ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))

2858+ (logInfo ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))

28022859 fork (when federationEnabled

28032860 (sendBatchedGossip result.peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops))

28042861 yield (respond {token: token, expiresAt: expiry})

@@ -2825,8 +2882,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do

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

28262883 t2 <- now

28272884 atomic do

2828− allPresence <- readAllPresence

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

2885+ -- Read only the presence buckets needed for watched keys (not all 16)

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

2887+ relevantPresence <- readSelectedPresenceBuckets neededBuckets

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

28302889 else yield []

28312890 fork (when federationEnabled (do

28322891 sendAcksAndClean result.pendingAcks

@@ -2847,9 +2906,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp

28472906 then yield (respond {status: MsRejected {}, messageId: ""})

28482907 else do

28492908 when (timestamp > t + clockSkewTolerance)

2850− (println ("sendMessage: future timestamp detected (+" ++ show (timestamp - t) ++ "ms ahead), clamping to now"))

2909+ (logWarn ("sendMessage: future timestamp detected (+" ++ show (timestamp - t) ++ "ms ahead), clamping to now"))

28512910 when (timestamp < t - messageMaxAge / 2)

2852− (println ("sendMessage: stale timestamp (" ++ show ((t - timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))

2911+ (logWarn ("sendMessage: stale timestamp (" ++ show ((t - timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))

28532912 let clampedTs = clampTimestamp timestamp t

28542913 result <- atomic do

28552914 draining <- *draining

@@ -2935,9 +2994,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp

29352994 set *forwards = union forwards cappedFwds)

29362995 yield {send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}, globalCapReached: False {}})

29372996 when result.senderRateLimited

2938− (println ("sendMessage: rate limited (" ++ show maxSendRatePerKey ++ "/s per key)"))

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

29392998 when result.globalCapReached

2940− (println ("sendMessage: global message cap reached (" ++ show maxMessages ++ "), rejecting"))

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

29413000 case result.send of

29423001 SendDone {delivery, remoteServers} ->

29433002 when (count remoteServers > 0 && federationEnabled) (

@@ -2956,14 +3015,14 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp

29563015 authorization: fwdAuth})

29573016 case fwdResult of

29583017 Err e -> do

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

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

29603019 recordFailure server

29613020 Ok response ->

29623021 if response.ok then do

29633022 removeForward effectiveMsgId server

29643023 clearFailure server

29653024 else

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

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

29673026 _ -> yield {}

29683027 yield (case result.send of

29693028 SendFailed {status} -> respond {status: status, messageId: ""}

@@ -3077,7 +3136,7 @@ handleRevoke = \targetToken authorization clientIp respond -> do

30773136 set *totalSessionCount = max 0 (sc - 1)

30783137 yield (True {})

30793138 else yield (False {})

3080− when removed (println ("revoke: session revoked for " ++ callerPubkey))

3139+ when removed (logInfo ("revoke: session revoked for " ++ callerPubkey))

30813140 yield (respond {ok: removed, authorized: True {}}))

30823141

30833142-- POST /auth/logout: Voluntary disconnect. Removes the caller's session and, if it was

@@ -3121,6 +3180,7 @@ handleLogout = \authorization clientIp respond -> do

31213180 let removedCount = count presenceBkt - count (do p <- presenceBkt; where p.pubkey != pubkey || not (isLocalServer p.server); yield p)

31223181 tpc <- *totalPresenceCount

31233182 set *totalPresenceCount = max 0 (tpc - removedCount)

3183+ set *presenceVersion = t

31243184 peerServers <- *knownPeerServers

31253185 yield (Just {value: {pubkey: pubkey, gossipOffline: True {}, peerServers: peerServers}})

31263186 else yield (Just {value: {pubkey: pubkey, gossipOffline: False {}, peerServers: []}})

@@ -3128,7 +3188,7 @@ handleLogout = \authorization clientIp respond -> do

31283188 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})

31293189 Just {value: r} -> do

31303190 when r.gossipOffline (do

3131− println ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")

3191+ logInfo ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")

31323192 when (federationEnabled && count r.peerServers > 0)

31333193 (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))

31343194 yield (respond {ok: True {}, authorized: True {}})

@@ -3155,7 +3215,7 @@ handleVerifyToken = \token scope clientIp respond -> do

31553215 set *verifyIpRates = result.entries

31563216 yield result.limited

31573217 if verifyLimited then do

3158− println ("federation/verify: rate limited for IP " ++ clientIp)

3218+ logWarn ("federation/verify: rate limited for IP " ++ clientIp)

31593219 yield (respond {ok: False {}})

31603220 else do

31613221 found <- atomic do

@@ -3164,7 +3224,7 @@ handleVerifyToken = \token scope clientIp respond -> do

31643224 let match = any pending (\p -> safeCompare p.token token && p.scope == scope && t - p.createdAt < verificationTokenExpiry)

31653225 when match (set *pendingVerifications = do p <- pending; where not (safeCompare p.token token); yield p)

31663226 yield match

3167− when (not found) (println ("federation/verify: failed attempt from " ++ clientIp ++ " (scope: " ++ scope ++ ")"))

3227+ when (not found) (logWarn ("federation/verify: failed attempt from " ++ clientIp ++ " (scope: " ++ scope ++ ")"))

31683228 yield (respond {ok: found})

31693229

31703230-- POST /federation/gossip: Receive presence gossip from a peer server.

@@ -3193,7 +3253,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati

31933253 set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs; yield r) else newEntries

31943254 yield (False {})

31953255 if eventRateLimited then do

3196− println ("gossip: event volume rate limited for " ++ fromServer ++ " (" ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms)")

3256+ logWarn ("gossip: event volume rate limited for " ++ fromServer ++ " (" ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms)")

31973257 yield (respond {ok: False {}, error: "event_rate_limited"})

31983258 else do

31993259 let validated = validGossipEvents events t

@@ -3202,7 +3262,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati

32023262 -- Warn if a peer sends both online and offline for the same key in one batch (suspicious)

32033263 let conflictingKeys = do on <- onlineEntries; off <- trustedOffline; where on.pubkey == off.pubkey && on.server == off.server; yield on.pubkey

32043264 when (count conflictingKeys > 0)

3205− (println ("gossip: warning: " ++ fromServer ++ " sent conflicting online+offline for same key(s)"))

3265+ (logWarn ("gossip: " ++ fromServer ++ " sent conflicting online+offline for same key(s)"))

32063266 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline

32073267 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries

32083268 let deduped = mergeByPubkeyServer [] filteredOnline

@@ -3218,10 +3278,11 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati

32183278 writeAllPresence gm.finalPresence

32193279 set *totalPresenceCount = count gm.finalPresence

32203280 set *knownPeerServers = gm.gossipPeerServers

3281+ set *presenceVersion = t

32213282 yield {gm: gm, preFiltered: count deduped - count capacityDeduped}

32223283 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))

32233284 when (not gossipResult.gm.withinCapacity || gossipResult.preFiltered > 0)

3224− (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer ++ " (pre-filtered: " ++ show gossipResult.preFiltered ++ ")"))

3285+ (logWarn ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer ++ " (pre-filtered: " ++ show gossipResult.preFiltered ++ ")"))

32253286 -- Suppress re-broadcast of events we've already gossiped recently (prevents amplification)

32263287 freshEvents <- atomic do

32273288 recentGossip <- *recentGossip

@@ -3258,7 +3319,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

32583319 t <- now

32593320 withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do

32603321 when (fromServer != originServer)

3261− (println ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))

3322+ (logInfo ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))

32623323 -- Relay origin verification: for relay forwards, verify originServer is a known peer

32633324 -- to prevent ack routing (and metadata leakage) to arbitrary servers.

32643325 originOk <- if fromServer == originServer then yield (True {})

@@ -3267,7 +3328,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

32673328 yield (any knownPeers (\p -> p == originServer))

32683329 if not originOk

32693330 then do

3270− println ("forward: rejected relay with unknown origin " ++ originServer ++ " from " ++ fromServer)

3331+ logWarn ("forward: rejected relay with unknown origin " ++ originServer ++ " from " ++ fromServer)

32713332 yield (respond {ok: False {}, error: "unknown_origin"})

32723333 else do

32733334 -- Replay protection: reject forwards with timestamps older than messageMaxAge.

@@ -3283,7 +3344,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

32833344 bkt <- readPresenceBucket (pubkeyBucket toKey)

32843345 yield (any bkt (\p -> p.pubkey == toKey && p.expiresAt > t))

32853346 if not hasPresence then do

3286− println ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")

3347+ logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")

32873348 yield (respond {ok: False {}, error: "no_presence"})

32883349 else do

32893350 let clampedTs = clampTimestamp timestamp t

@@ -3293,7 +3354,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

32933354 ackedIds <- *ackedIds

32943355 deliveredIds <- *deliveredIds

32953356 forwardNonces <- *forwardNonces

3296− let isDup = checkMessageDup bucket messageId toKey || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds || any forwardNonces (\n -> n.messageId == messageId && n.toKey == toKey)

3357+ let isDup = checkMessageDup bucket messageId toKey || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds || any forwardNonces (\n -> n.messageId == messageId && n.toKey == toKey && n.originServer == originServer)

32973358 if isDup

32983359 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}}

32993360 else do

@@ -3335,16 +3396,16 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

33353396 signalMessageForKey toKey seqTs

33363397 -- Record forward nonce for replay protection beyond message/ack dedup lifetime

33373398 fwdNonces <- *forwardNonces

3338− let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, seenAt: t}]

3399+ let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]

33393400 set *forwardNonces = if count updatedNonces > maxForwardNonces then evictOneOldest updatedNonces (\n -> n.seenAt) else updatedNonces

33403401 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}

33413402 case result.forward of

33423403 ForwardRejected {} ->

33433404 if result.globalCapReached

3344− then println ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)

3405+ then logWarn ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)

33453406 else if result.recvRateLimited

3346− then println ("forward: recv rate limited (" ++ show maxRecvRatePerKey ++ "/s per key), rejecting for " ++ toKey)

3347− else println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")

3407+ then logWarn ("forward: recv rate limited (" ++ show maxRecvRatePerKey ++ "/s per key), rejecting for " ++ toKey)

3408+ else logWarn ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")

33483409 _ -> yield {}

33493410 yield (case result.forward of

33503411 ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}

@@ -3390,6 +3451,7 @@ handleDrain = \authorization respond -> do

33903451 let remainingPresence = do p <- allPresence; where not (isLocalServer p.server) || p.expiresAt <= t; yield p

33913452 writeAllPresence remainingPresence

33923453 set *totalPresenceCount = count remainingPresence

3454+ set *presenceVersion = t

33933455 set *pollWakeup = t

33943456 yield {localEntries: localEntries, peerServers: collectPeerServers allPresence (\p -> p.server)}

33953457 let drainedCount = count drainResult.localEntries

@@ -3407,10 +3469,10 @@ handleDrain = \authorization respond -> do

34073469 fork (do

34083470 -- Best-effort flush of pending forwards and acks before stopping

34093471 when (federationEnabled && pendingForwardCount > 0) (do

3410− println ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards")

3472+ logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards")

34113473 retryPendingForwards flushState.forwards)

34123474 when (federationEnabled && pendingAckCount > 0) (do

3413− println ("drain: flushing " ++ show pendingAckCount ++ " pending acks")

3475+ logInfo ("drain: flushing " ++ show pendingAckCount ++ " pending acks")

34143476 sendAcksAndClean flushState.pendingAcks)

34153477 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do

34163478 let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}

@@ -3419,11 +3481,11 @@ handleDrain = \authorization respond -> do

34193481 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))

34203482 t2 <- now

34213483 atomic do set *drainCompletedAt = t2)

3422− println ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")

3484+ logInfo ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")

34233485 when (pendingForwardCount > 0)

3424− (println ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))

3486+ (logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))

34253487 when (pendingAckCount > 0)

3426− (println ("drain: flushing " ++ show pendingAckCount ++ " pending acks"))

3488+ (logInfo ("drain: flushing " ++ show pendingAckCount ++ " pending acks"))

34273489 yield (respond {ok: True {}, drained: drainedCount})

34283490

34293491-- POST /maintenance/undrain: Resume accepting requests after a drain.

@@ -3437,8 +3499,8 @@ handleUndrain = \authorization respond -> do

34373499 set *draining = False {}

34383500 set *drainCompletedAt = 0

34393501 yield d

3440− when wasDraining (println "undrain: server resumed, accepting new requests")

3441− when (not wasDraining) (println "undrain: server was not draining")

3502+ when wasDraining (logInfo "undrain: server resumed, accepting new requests")

3503+ when (not wasDraining) (logInfo "undrain: server was not draining")

34423504 yield (respond {ok: True {}})

34433505

34443506-- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).

@@ -3487,7 +3549,7 @@ handleHealth = \respond -> do

34873549 capacityHealthy: result.capacityOk,

34883550 federationHealthy: result.federationOk})

34893551

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

3552+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 {}}

34913553

34923554handleStats = \authorization respond -> do

34933555 t <- now

@@ -3585,7 +3647,7 @@ handleStatsKey = \pubkey authorization respond -> do

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

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

35873649

3588−emptySnapshot = {ok: False {}, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: []}

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

35893651

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

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

@@ -3605,18 +3667,23 @@ handleSnapshot = \authorization respond -> do

36053667 f <- *forwards

36063668 yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f}

36073669 atomic do set *lastSnapshotExportAt = t

3608− println ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards")

3609− yield (respond {ok: True {}, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards})

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

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

36103672

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

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

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

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

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

3616−handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData authorization respond -> do

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

36173679 let authorized = verifyMaintenanceAuth authorization

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

36193681 else do

3682+ -- Version 0 = pre-versioning snapshot (accept for backwards compatibility)

3683+ if version != 0 && version != snapshotVersion then do

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

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

3686+ else do

36203687 t <- now

36213688 isDraining <- *draining

36223689 if not isDraining then yield (respond {ok: False {}, error: "server_must_be_draining", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})

@@ -3629,7 +3696,7 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw

36293696 let validAcked = do a <- ackedIdsData; where not (isLocalServer a.originServer); yield a

36303697 let validOverflow = do a <- ackOverflowData; where not (isLocalServer a.originServer); yield a

36313698 let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow)

3632− when (rejectedCount > 0) (println ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))

3699+ when (rejectedCount > 0) (logWarn ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))

36333700 result <- atomic do

36343701 -- Merge presence (existing entries win on conflict via freshness)

36353702 allPresence <- readAllPresence

@@ -3638,6 +3705,7 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw

36383705 writeAllPresence cappedPresence

36393706 set *totalPresenceCount = count cappedPresence

36403707 set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)

3708+ set *presenceVersion = t

36413709 -- Union dedup state, respecting capacity limits (using validated data)

36423710 ackedIds <- *ackedIds

36433711 let mergedAcked = union ackedIds validAcked

@@ -3661,7 +3729,7 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw

36613729 let cappedForwards = takeRelation maxForwardsEntries mergedForwards

36623730 set *forwards = cappedForwards

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

3664− println ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")

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

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

36663734

36673735-- POST /messages/ack: Explicitly acknowledge received messages by ID.

@@ -3677,6 +3745,16 @@ handleAckMessages = \messageIds authorization clientIp respond -> do

36773745 withSessionAuth authorization t clientIp

36783746 (yield (respond {ok: False {}, authorized: False {}, ackedCount: 0}))

36793747 (\pubkey -> do

3748+ ackRateLimited <- atomic do

3749+ lookupRates <- *lookupRates

3750+ let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == pubkey)

3751+ (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})

3752+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})

3753+ maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t

3754+ set *lookupRates = rateResult.entries

3755+ yield rateResult.limited

3756+ if ackRateLimited then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0})

3757+ else do

36803758 let dedupedIds = dedupMessageIds messageIds

36813759 result <- atomic do

36823760 let b = pubkeyBucket pubkey

@@ -3705,6 +3783,122 @@ handleAckMessages = \messageIds authorization clientIp respond -> do

37053783 retryEvictedAcks result.evictedEntries result.pendingAcks))

37063784 yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount}))

37073785

3786+-- POST /messages/batch: Send multiple messages in one request.

3787+-- Auth checked once; each message processed independently with its own rate limiting.

3788+-- Partial success is possible — check per-message status in results.

3789+handleSendBatch = \messages authorization clientIp respond -> do

3790+ t <- now

3791+ globalRateLimited <- checkGlobalRate t

3792+ if globalRateLimited then yield (respond {results: [], authorized: False {}})

3793+ else

3794+ withSessionAuth authorization t clientIp

3795+ (yield (respond {results: [], authorized: False {}}))

3796+ (\senderKey -> do

3797+ let capped = takeRelation maxMessageIds messages

3798+ results <- sendBatchInner senderKey capped t []

3799+ -- Fork federation forwarding for all successful sends

3800+ let forwardTargets = do r <- results; where r.needsForward; yield r

3801+ fork (when (federationEnabled && count forwardTargets > 0)

3802+ (sendToEach (collectPeerServers forwardTargets (\r -> r.forwardServer)) (\server -> do

3803+ let serverMsgs = do r <- forwardTargets; where r.forwardServer == server; yield r

3804+ forEachBatch serverMsgs maxMessageIds (\batch -> do

3805+ fwdT <- now

3806+ forEachBatch batch 1 (\single -> case headOf single of

3807+ Nothing {} -> yield {}

3808+ Just {value: r} -> do

3809+ stillPending <- atomic do

3810+ forwards <- *forwards

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

3812+ when stillPending (do

3813+ fwdAuth <- generateFederationAuth "forward"

3814+ fwdResult <- fetchFederation server (ForwardMessage {

3815+ toKey: r.toKey, encryptedBlob: r.blob,

3816+ timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,

3817+ fromServer: serverHostN, requestTimestamp: fwdT,

3818+ authorization: fwdAuth})

3819+ case fwdResult of

3820+ Err e -> do

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

3822+ recordFailure server

3823+ Ok response ->

3824+ if response.ok then do

3825+ removeForward r.messageId server

3826+ clearFailure server

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

3828+ yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}}))

3829+

3830+sendBatchInner = \senderKey messages t acc ->

3831+ let split = splitAtN messages 1 in

3832+ case headOf split.batch of

3833+ Nothing {} -> yield acc

3834+ Just {value: msg} -> do

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

3836+ result <- if t - msg.timestamp > messageMaxAge

3837+ then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0}

3838+ else if senderKey == msg.to

3839+ then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0}

3840+ else do

3841+ let clampedTs = clampTimestamp msg.timestamp t

3842+ sendResult <- atomic do

3843+ draining <- *draining

3844+ if draining then yield {status: MsDraining {}, fwdServer: "", needsFwd: False {}}

3845+ else do

3846+ let msgBucket = pubkeyBucket msg.to

3847+ bucket <- readMessageBucket msgBucket

3848+ ackedIds <- *ackedIds

3849+ deliveredIds <- *deliveredIds

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

3851+ if isDup then yield {status: MsDuplicate {}, fwdServer: "", needsFwd: False {}}

3852+ else do

3853+ sendRates <- *sendRates

3854+ let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)

3855+ (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})

3856+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})

3857+ maxSendRatePerKey sendRateWindowMs maxSendRateEntries t

3858+ if rateResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}}

3859+ else do

3860+ set *sendRates = rateResult.entries

3861+ recvRates <- *recvRates

3862+ let recvResult = checkRateWindow recvRates (\r -> r.pubkey == msg.to)

3863+ (\ts -> {pubkey: msg.to, windowStart: ts, count: 1})

3864+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})

3865+ maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t

3866+ if recvResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}}

3867+ else do

3868+ set *recvRates = recvResult.entries

3869+ totalMsgCount <- *totalMessageCount

3870+ if totalMsgCount >= maxMessages then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}}

3871+ else do

3872+ recipientCounts <- *recipientMsgCounts

3873+ let recipientCount = getRecipientCount msg.to recipientCounts

3874+ if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}}

3875+ else do

3876+ recipientPresence <- readPresenceBucket (pubkeyBucket msg.to)

3877+ seqCounter <- *seqCounter

3878+ let seqTs = nextSeqTs t seqCounter

3879+ set *seqCounter = seqTs

3880+ writeMessageBucket msgBucket (union bucket

3881+ [{messageId: effectiveMsgId, toKey: msg.to,

3882+ encryptedBlob: msg.encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])

3883+ set *totalMessageCount = totalMsgCount + 1

3884+ set *recipientMsgCounts = adjustRecipientCount recipientCounts msg.to 1

3885+ signalMessageForKey msg.to seqTs

3886+ let toPresence = presenceForKey recipientPresence msg.to t

3887+ let localOnline = hasLocalPresence toPresence

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

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

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

3891+ forwards <- *forwards

3892+ 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}

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

3894+ set *forwards = union forwards cappedFwds)

3895+ let delivery = if localOnline then MsDelivered {}

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

3897+ else MsQueued {}

3898+ yield {status: delivery, fwdServer: fwdServer, needsFwd: federationEnabled && count remoteServers > 0}

3899+ yield {messageId: effectiveMsgId, status: sendResult.status, needsForward: sendResult.needsFwd, forwardServer: sendResult.fwdServer, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs}

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

3901+

37083902serve = \req -> case req of

37093903 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond

37103904 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond

@@ -3714,6 +3908,7 @@ serve = \req -> case req of

37143908 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond

37153909 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond

37163910 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond

3911+ SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond

37173912 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond

37183913 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond

37193914 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond

@@ -3723,7 +3918,7 @@ serve = \req -> case req of

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

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

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

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

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

37273922 Health {respond} -> handleHealth respond

37283923 Stats {authorization, respond} -> handleStats authorization respond

37293924 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond

@@ -3744,11 +3939,18 @@ backgroundPrune = do

37443939 backgroundPrune

37453940 else do

37463941 t <- now

3942+ -- Detect backward clock jump: if seqCounter is far ahead of now, reset to prevent

3943+ -- all messages getting unreachable receivedAt values for an extended period

3944+ atomic do

3945+ sc <- *seqCounter

3946+ when (sc > t + clockSkewTolerance) (do

3947+ logWarn ("clock jump detected — seqCounter " ++ show (sc - t) ++ "ms ahead of now, resetting")

3948+ set *seqCounter = t)

37473949 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead

37483950 sa <- *startedAt

37493951 lastHb <- *lastHeartbeatAt

37503952 when (t - sa > healthHeartbeatMaxAge && t - lastHb > healthHeartbeatMaxAge) (do

3751− println "backgroundPrune: heartbeat appears dead, bumping pollWakeup"

3953+ logWarn "backgroundPrune: heartbeat appears dead, bumping pollWakeup"

37523954 atomic do

37533955 set *pollWakeup = t

37543956 set *lastHeartbeatAt = t)

@@ -3765,7 +3967,7 @@ backgroundPrune = do

37653967 set *pruneCount = pc + 1

37663968 yield {canRun: True {}, recovered: running > 0, gen: newGen, stuckSince: running}

37673969 when status.recovered (do

3768− println ("backgroundPrune: previous cycle stuck for " ++ show (t - status.stuckSince) ++ "ms, forcing recovery")

3970+ logWarn ("backgroundPrune: previous cycle stuck for " ++ show (t - status.stuckSince) ++ "ms, forcing recovery")

37693971 -- Reset fork counter: the stuck fork leaked its slot, so reclaim it to prevent permanent lockout

37703972 atomic do set *activePruneForks = 0)

37713973 if status.canRun

@@ -3773,7 +3975,7 @@ backgroundPrune = do

37733975 -- Guard against unbounded concurrent prune forks from repeated stuck recovery

37743976 forks <- *activePruneForks

37753977 if forks >= 2

3776− then println "backgroundPrune: skipping (max concurrent prune forks reached)"

3978+ then logWarn "backgroundPrune: skipping (max concurrent prune forks reached)"

37773979 else do

37783980 atomic do

37793981 af <- *activePruneForks

@@ -3787,7 +3989,7 @@ backgroundPrune = do

37873989 when (currentGen == status.gen) (set *pruneRunning = 0)

37883990 af <- *activePruneForks

37893991 set *activePruneForks = max 0 (af - 1))

3790− else println "backgroundPrune: skipping (previous cycle still running)"

3992+ else logWarn "backgroundPrune: skipping (previous cycle still running)"

37913993 backgroundPrune

37923994

37933995-- Background loop: periodically bumps pollWakeup to unblock long-polls that have been

@@ -3827,14 +4029,16 @@ backgroundSnapshot = do

38274029 d <- *deliveredIds

38284030 ff <- *forwardFailures

38294031 yield {messages: tmc, sessions: sc, presence: pc, forwards: count f, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff}

3830− println ("snapshot: messages=" ++ show result.messages ++ " sessions=" ++ show result.sessions ++ " presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " ackOverflow=" ++ show result.ackOverflow ++ " deliveredIds=" ++ show result.deliveredIds ++ " forwardFailures=" ++ show result.forwardFailures)

4032+ logInfo ("snapshot: messages=" ++ show result.messages ++ " sessions=" ++ show result.sessions ++ " presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " ackOverflow=" ++ show result.ackOverflow ++ " deliveredIds=" ++ show result.deliveredIds ++ " forwardFailures=" ++ show result.forwardFailures)

38314033 -- Warn if federation state hasn't been exported recently (will be lost on restart)

38324034 t <- now

38334035 lastExport <- *lastSnapshotExportAt

38344036 when (maintenanceSecret != "" && snapshotStalenessWarning > 0

38354037 && (lastExport == 0 || t - lastExport > snapshotStalenessWarning)

38364038 && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))

3837− (println ("WARNING: no snapshot exported in " ++ show (if lastExport == 0 then 0 else (t - lastExport) / 1000) ++ "s — federation state will be lost on restart (presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " deliveredIds=" ++ show result.deliveredIds ++ ")"))

4039+ (do

4040+ logError ("no snapshot exported in " ++ show (if lastExport == 0 then 0 else (t - lastExport) / 1000) ++ "s — federation state will be lost on restart (presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " deliveredIds=" ++ show result.deliveredIds ++ ")")

4041+ logError "ACTION REQUIRED: run GET /maintenance/snapshot to persist federation state before restart")

38384042 else yield {}

38394043 backgroundSnapshot

38404044

@@ -3874,40 +4078,40 @@ main = do

38744078 && maxGlobalFederationConcurrency >= maxFederationConcurrency

38754079 && maxGossipEventsPerServer > maxGossipEvents

38764080 if not configOk

3877− then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"

4081+ then logError "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"

38784082 else if not configRelationsOk

3879− then println "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, per-key limits <= global limits), server not started"

4083+ then logError "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, per-key limits <= global limits), server not started"

38804084 else do

38814085 t <- now

38824086 atomic do

38834087 set *startedAt = t

38844088 set *serverGeneration = t

38854089 when (maintenanceSecret == "")

3886− (println "WARNING: maintenanceSecret is empty — /maintenance/* and /stats endpoints are disabled")

4090+ (logWarn "maintenanceSecret is empty — /maintenance/* and /stats endpoints are disabled")

38874091 when (matchesHostnameOrSubdomain serverHostN "localhost")

3888− (println ("WARNING: serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))

3889− println "NOTE: IP binding is active — sessions created with a ClientIp header will reject requests without one"

3890− println "Skrepka relay server starting on port 8080..."

3891− println (" federation: " ++ if federationEnabled then "enabled" else "disabled")

3892− println (" prune interval: " ++ show backgroundPruneInterval ++ "ms")

3893− println (" poll timeout: " ++ show pollTimeoutMs ++ "ms")

3894− println (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")

3895− println (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")

3896− println (" max presence: " ++ show maxPresenceEntries)

3897− println (" max forwards: " ++ show maxForwardsEntries)

3898− println (" session expiry: " ++ show sessionExpiry ++ "ms")

3899− println (" message max age: " ++ show messageMaxAge ++ "ms")

3900− println (" global rate limit: " ++ show maxGlobalRequestRate ++ "/" ++ show globalRateWindowMs ++ "ms")

3901− println (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")

3902− println (" pair rate limit: " ++ show maxPairRatePerPair ++ "/" ++ show pairRateWindowMs ++ "ms per sender-recipient")

3903− println (" federation pair rate limit: " ++ show maxFederationPairRatePerPair ++ "/" ++ show federationPairRateWindowMs ++ "ms per server-recipient")

3904− println (" gossip event rate limit: " ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms per server")

3905− println (" global federation concurrency: " ++ show maxGlobalFederationConcurrency)

3906− println (" ack retry limit: " ++ show maxAckRetries ++ " per entry")

4092+ (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))

4093+ logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"

4094+ logInfo "Skrepka relay server starting on port 8080..."

4095+ logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")

4096+ logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")

4097+ logInfo (" poll timeout: " ++ show pollTimeoutMs ++ "ms")

4098+ logInfo (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")

4099+ logInfo (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")

4100+ logInfo (" max presence: " ++ show maxPresenceEntries)

4101+ logInfo (" max forwards: " ++ show maxForwardsEntries)

4102+ logInfo (" session expiry: " ++ show sessionExpiry ++ "ms")

4103+ logInfo (" message max age: " ++ show messageMaxAge ++ "ms")

4104+ logInfo (" global rate limit: " ++ show maxGlobalRequestRate ++ "/" ++ show globalRateWindowMs ++ "ms")

4105+ logInfo (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")

4106+ logInfo (" pair rate limit: " ++ show maxPairRatePerPair ++ "/" ++ show pairRateWindowMs ++ "ms per sender-recipient")

4107+ logInfo (" federation pair rate limit: " ++ show maxFederationPairRatePerPair ++ "/" ++ show federationPairRateWindowMs ++ "ms per server-recipient")

4108+ logInfo (" gossip event rate limit: " ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms per server")

4109+ logInfo (" global federation concurrency: " ++ show maxGlobalFederationConcurrency)

4110+ logInfo (" ack retry limit: " ++ show maxAckRetries ++ " per entry")

39074111 when (federationCircuitBreakerThreshold > 0)

3908− (println (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))

4112+ (logInfo (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))

39094113 when (autoSnapshotInterval > 0)

3910− (println (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))

4114+ (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))

39114115 fork backgroundPrune

39124116 fork pollHeartbeat

39134117 when (autoSnapshotInterval > 0) (fork backgroundSnapshot)