e51c63d · skrepka · git.iko.soy

git.iko.soy/skrepka

Implement 8 server improvements: structured logging, two-pass ack cleanup, federation retry jitter, nonce eviction logging, code simplifications, missing enum case, and adaptive poll timeout

e51c63diko6 months ago

server.knot+75−63

compare with master

unifiedsplit

@@ -1212,6 +1212,14 @@ logError = \msg -> println ("[ERROR] " ++ msg)

1212-- when msg contains user-controlled data (e.g., error strings, server names).

1213logEvent = \level event msg -> println ("{\"level\":\"" ++ escapeJsonString level ++ "\",\"event\":\"" ++ escapeJsonString event ++ "\",\"msg\":\"" ++ escapeJsonString msg ++ "\"}")

1214

1215+-- Structured key=value logging: formats message as "k1=v1 k2=v2 ..." for consistent

1216+-- machine-parseable log entries. Wraps logEvent with key-value pair formatting.

1217+-- Use for events where individual fields need to be extracted by log pipelines.

1218+logKV = \level event kvs ->

1219+ logEvent level event (joinKVPairs kvs)

1220+

1221+joinKVPairs = \kvs -> fold (\acc kv -> if acc == "" then kv else acc ++ " " ++ kv) "" kvs

1222+

1215-- Log a warning when a federation peer sends a request with an older protocol version.

1223-- Log a warning when a federation peer sends a request with an older protocol version.

1216-- protocolVersion == 0 means pre-versioning peer (field missing). Only logs once per

1224-- protocolVersion == 0 means pre-versioning peer (field missing). Only logs once per

1217-- request — callers invoke this before delegating to the handler.

1225-- request — callers invoke this before delegating to the handler.

@@ -4025,6 +4033,9 @@ ackCleanupForwards = \entries t ->

4025 let bucketIds = uniqueBucketsFor entries (\e -> e.toKey) in

4033 let bucketIds = uniqueBucketsFor entries (\e -> e.toKey) in

4026 ackCleanupByBucket bucketIds entries t

4034 ackCleanupByBucket bucketIds entries t

4027

4035

4036+-- Two-pass ack cleanup: first filter entries to those eligible for cleanup,

4037+-- then single-pass filter messages. Avoids O(entries * messages) nested scan

4038+-- from the fold-with-shrinking-accumulator pattern.

4028ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of

4039ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of

4029 Nothing {} -> yield {}

4040 Nothing {} -> yield {}

4030 Just {value: b} -> do

4041 Just {value: b} -> do

@@ -4037,17 +4048,20 @@ ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of

4037 msgs <- readMessageBucket b

4048 msgs <- readMessageBucket b

4038 tc <- *totalMessageCount

4049 tc <- *totalMessageCount

4039 rc <- readRcBucket b

4050 rc <- readRcBucket b

4040− let cleanup = fold (\acc rf ->

4051+ -- Phase 1: filter to entries eligible for cleanup (no local presence, no pending forwards)

4052+ let eligible = do rf <- bucketEntries

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

4053+ where not (any bucketPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t))

4042− let hasOtherFwds = any fwdBkt (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey) in

4054+ where not (any fwdBkt (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey))

4043− if hasLocal || hasOtherFwds then acc

4055+ yield rf

4056+ -- Phase 2: single pass over messages — remove those matching any eligible entry past grace period

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

4057+ let shouldRemove = \m -> any eligible (\rf -> m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod)

4045− Nothing {} -> acc

4058+ let cleanMsgs = do m <- msgs; where not (shouldRemove m); yield m

4046− Just {} -> {msgs: (do m <- acc.msgs; where not (m.messageId == rf.messageId && m.toKey == rf.toKey); yield m), removed: acc.removed + 1, rc: adjustRecipientCount acc.rc rf.toKey (0 - 1)}

4059+ let removedMsgs = do m <- msgs; where shouldRemove m; yield m

4047− ) {msgs: msgs, removed: 0, rc: rc} bucketEntries

4060+ let removedCount = count removedMsgs

4061+ let newRc = fold (\accRc m -> adjustRecipientCount accRc m.toKey (0 - 1)) rc removedMsgs

4048− writeMessageBucket b cleanup.msgs

4062+ writeMessageBucket b cleanMsgs

4049− set *totalMessageCount = max 0 (tc - cleanup.removed)

4063+ set *totalMessageCount = max 0 (tc - removedCount)

4050− writeRcBucket b cleanup.rc

4064+ writeRcBucket b newRc

4051 ackCleanupByBucket remainingBuckets entries t

4065 ackCleanupByBucket remainingBuckets entries t

4052

4066

4053pow2 : Int -> Int

4067pow2 : Int -> Int

@@ -4088,12 +4102,18 @@ isFederationCircuitOpen = \failures t ->

4088

4102

4089recordFailure = \server -> do

4103recordFailure = \server -> do

4090 t <- now

4104 t <- now

4105+ -- Random jitter (0-25% of base backoff) prevents synchronized retries when

4106+ -- multiple operations to the same peer fail in quick succession. Shifts failedAt

4107+ -- forward, delaying when shouldRetryServer allows the next attempt. Combined with

4108+ -- the deterministic per-server spread in retryBackoffFor, this breaks both

4109+ -- cross-server and same-server retry correlation (thundering herd).

4110+ jitter <- randomInt (retryBackoffBase / 4)

4091 atomic do

4111 atomic do

4092 failures <- *forwardFailures

4112 failures <- *forwardFailures

4093 let existing = headOf (do f <- failures; where f.server == server; yield f)

4113 let existing = headOf (do f <- failures; where f.server == server; yield f)

4094 let updated = case existing of

4114 let updated = case existing of

4095− Nothing {} -> union failures [{server: server, failedAt: t, failures: 1, firstFailedAt: t}]

4115+ Nothing {} -> union failures [{server: server, failedAt: t + jitter, failures: 1, firstFailedAt: t}]

4096− Just {value: f} -> union (do x <- failures; where x.server != server; yield x) [{server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}]

4116+ Just {value: f} -> union (do x <- failures; where x.server != server; yield x) [{server: server, failedAt: t + jitter, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}]

4097 set *forwardFailures = if count updated > maxForwardFailures then evictOneOldest updated (\x -> x.failedAt) else updated

4117 set *forwardFailures = if count updated > maxForwardFailures then evictOneOldest updated (\x -> x.failedAt) else updated

4098

4118

4099clearFailure = \server -> do

4119clearFailure = \server -> do

@@ -4113,7 +4133,7 @@ incrementForwardRetries = \messageId toServer toKey -> atomic do

4113 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toServer == toServer); yield f)

4133 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toServer == toServer); yield f)

4114 tfc <- *totalForwardCount

4134 tfc <- *totalForwardCount

4115 set *totalForwardCount = max 0 (tfc - 1)

4135 set *totalForwardCount = max 0 (tfc - 1)

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

4136+ logKV "warn" "federation_forward_dropped" ["messageId=" ++ messageId, "server=" ++ toServer, "retries=" ++ show maxForwardRetries]

4117 else writeFwdBucket b (do f <- fwdBkt; 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, expiresAt: f.expiresAt, priority: f.priority, senderKey: f.senderKey} else f))

4137 else writeFwdBucket b (do f <- fwdBkt; 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, expiresAt: f.expiresAt, priority: f.priority, senderKey: f.senderKey} else f))

4118

4138

4119-- Retry forwards for a server using batch forward (1 HTTP round-trip per batch of maxBatchSize).

4139-- Retry forwards for a server using batch forward (1 HTTP round-trip per batch of maxBatchSize).

@@ -4141,7 +4161,7 @@ retryForwardsForServer = \entries remaining ->

4141 requestTimestamp: t, authorization: auth, correlationId: corrId, protocolVersion: federationProtocolVersion})

4161 requestTimestamp: t, authorization: auth, correlationId: corrId, protocolVersion: federationProtocolVersion})

4142 case result of

4162 case result of

4143 Err e -> do

4163 Err e -> do

4144− logWarn ("federation: batch forward to " ++ server ++ " failed: " ++ show e)

4164+ logKV "warn" "federation_batch_forward_failed" ["server=" ++ server, "batchSize=" ++ show (count batch), "error=" ++ show e]

4145 recordFailure server

4165 recordFailure server

4146 -- HTTP-level failure (connectivity, timeout): don't increment per-message retries.

4166 -- HTTP-level failure (connectivity, timeout): don't increment per-message retries.

4147 -- The messages weren't rejected — the server was unreachable. Server backoff

4167 -- The messages weren't rejected — the server was unreachable. Server backoff

@@ -4154,7 +4174,7 @@ retryForwardsForServer = \entries remaining ->

4154 -- Process per-message results: remove succeeded, increment retries for failed

4174 -- Process per-message results: remove succeeded, increment retries for failed

4155 retryForwardsBatchResults batch response.results server

4175 retryForwardsBatchResults batch response.results server

4156 else do

4176 else do

4157− logWarn ("federation: batch forward to " ++ server ++ " rejected: " ++ response.error ++ ", falling back to single-message forwards")

4177+ logKV "warn" "federation_batch_forward_rejected" ["server=" ++ server, "error=" ++ response.error]

4158 -- Application-level batch rejection (old peer, invalid request): fall back to

4178 -- Application-level batch rejection (old peer, invalid request): fall back to

4159 -- sequential single-message forwards instead of burning retries. The peer likely

4179 -- sequential single-message forwards instead of burning retries. The peer likely

4160 -- doesn't support /federation/forward-batch, so individual forwards may succeed.

4180 -- doesn't support /federation/forward-batch, so individual forwards may succeed.

@@ -4188,7 +4208,7 @@ retryForwardsSingle = \batch server ->

4188 expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})

4208 expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})

4189 case result of

4209 case result of

4190 Err e -> do

4210 Err e -> do

4191− logWarn ("federation: single forward to " ++ server ++ " failed for " ++ f.messageId ++ ": " ++ show e)

4211+ logKV "warn" "federation_single_forward_failed" ["server=" ++ server, "messageId=" ++ f.messageId, "error=" ++ show e]

4192 recordFailure server

4212 recordFailure server

4193 Ok response ->

4213 Ok response ->

4194 if response.ok then do

4214 if response.ok then do

@@ -5499,9 +5519,15 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn

5499 let nb = pubkeyBucket to

5519 let nb = pubkeyBucket to

5500 fwdNonces <- readNonceBucket nb

5520 fwdNonces <- readNonceBucket nb

5501 let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, seenAt: t}]

5521 let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, seenAt: t}]

5502− writeNonceBucket nb (if count updatedNonces > maxNoncesPerBucket

5522+ let evictedNonces = if count updatedNonces > maxNoncesPerBucket

5503− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t

5523+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t

5504− else updatedNonces)

5524+ else updatedNonces

5525+ -- Log LRU nonce eviction: evicted nonces create a bounded replay window

5526+ -- where messages with those nonces could be re-accepted until natural expiry.

5527+ let lruEvicted = count updatedNonces - count evictedNonces

5528+ when (lruEvicted > 0)

5529+ (logKV "warn" "nonce_lru_eviction" ["bucket=" ++ show nb, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket])

5530+ writeNonceBucket nb evictedNonces

5505 let toPresence = presenceForKey recipientPresence to t

5531 let toPresence = presenceForKey recipientPresence to t

5506 let localOnline = hasLocalPresence toPresence

5532 let localOnline = hasLocalPresence toPresence

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

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

@@ -5547,14 +5573,14 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->

5547 expiresAt: fwdInfo.fwdExpiresAt, priority: fwdInfo.fwdPriority, protocolVersion: federationProtocolVersion})

5573 expiresAt: fwdInfo.fwdExpiresAt, priority: fwdInfo.fwdPriority, protocolVersion: federationProtocolVersion})

5548 case fwdResult of

5574 case fwdResult of

5549 Err e -> do

5575 Err e -> do

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

5576+ logKV "warn" "federation_forward_failed" ["server=" ++ server, "messageId=" ++ effectiveMsgId, "error=" ++ show e]

5551 recordFailure server

5577 recordFailure server

5552 Ok response ->

5578 Ok response ->

5553 if response.ok then do

5579 if response.ok then do

5554 removeForward effectiveMsgId server to

5580 removeForward effectiveMsgId server to

5555 clearFailure server

5581 clearFailure server

5556 else

5582 else

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

5583+ logKV "warn" "federation_forward_rejected" ["server=" ++ server, "messageId=" ++ effectiveMsgId, "error=" ++ response.error]))))

5558

5584

5559-- Send all batch forwards for one server sequentially (no per-message fork overhead).

5585-- Send all batch forwards for one server sequentially (no per-message fork overhead).

5560-- Used by handleSendBatch; reduces fork count from O(messages * servers) to O(servers).

5586-- Used by handleSendBatch; reduces fork count from O(messages * servers) to O(servers).

@@ -5581,14 +5607,14 @@ sendBatchForwardsSequential = \msgs server ->

5581 expiresAt: r.fwdExpiresAt, priority: r.fwdPriority, protocolVersion: federationProtocolVersion})

5607 expiresAt: r.fwdExpiresAt, priority: r.fwdPriority, protocolVersion: federationProtocolVersion})

5582 case fwdResult of

5608 case fwdResult of

5583 Err e -> do

5609 Err e -> do

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

5610+ logKV "warn" "federation_forward_failed" ["server=" ++ server, "messageId=" ++ r.messageId, "error=" ++ show e]

5585 recordFailure server

5611 recordFailure server

5586 Ok response ->

5612 Ok response ->

5587 if response.ok then do

5613 if response.ok then do

5588 removeForward r.messageId server r.toKey

5614 removeForward r.messageId server r.toKey

5589 clearFailure server

5615 clearFailure server

5590 else

5616 else

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

5617+ logKV "warn" "federation_forward_rejected" ["server=" ++ server, "messageId=" ++ r.messageId, "error=" ++ response.error])

5592 sendBatchForwardsSequential rest server

5618 sendBatchForwardsSequential rest server

5593

5619

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

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

@@ -6092,9 +6118,13 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

6092 -- Sharded: writes only to the bucket for this recipient's pubkey.

6118 -- Sharded: writes only to the bucket for this recipient's pubkey.

6093 fwdNonces <- readNonceBucket bkt

6119 fwdNonces <- readNonceBucket bkt

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

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

6095− writeNonceBucket bkt (if count updatedNonces > maxNoncesPerBucket

6121+ let evictedNonces = if count updatedNonces > maxNoncesPerBucket

6096− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t

6122+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t

6097− else updatedNonces)

6123+ else updatedNonces

6124+ let lruEvicted = count updatedNonces - count evictedNonces

6125+ when (lruEvicted > 0)

6126+ (logKV "warn" "nonce_lru_eviction" ["bucket=" ++ show bkt, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket])

6127+ writeNonceBucket bkt evictedNonces

6098 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}

6128 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}

6099 if result.noPresence then do

6129 if result.noPresence then do

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

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

@@ -7922,9 +7952,12 @@ requestPath = \req -> case req of

7922 StatsKey {} -> "GET /stats/key"

7952 StatsKey {} -> "GET /stats/key"

7923 TopRecipients {} -> "POST /stats/top-recipients"

7953 TopRecipients {} -> "POST /stats/top-recipients"

7924 StatsForwards {} -> "GET /stats/forwards"

7954 StatsForwards {} -> "GET /stats/forwards"

7955+ StatsRates {} -> "GET /stats/rates"

7925 Ready {} -> "GET /ready"

7956 Ready {} -> "GET /ready"

7926

7957

7927−-- Extract client IP for access logging (empty for federation/maintenance routes)

7958+-- Extract client IP for access logging (empty for federation/maintenance/stats routes).

7959+-- Client-facing routes extract clientIp; all others use catch-all to avoid enumerating

7960+-- every federation/maintenance/stats route (reduces maintenance burden on new routes).

7928requestClientIp = \req -> case req of

7961requestClientIp = \req -> case req of

7929 ReqChallenge {clientIp} -> clientIp

7962 ReqChallenge {clientIp} -> clientIp

7930 VerifyAuth {clientIp} -> clientIp

7963 VerifyAuth {clientIp} -> clientIp

@@ -7942,40 +7975,7 @@ requestClientIp = \req -> case req of

7942 LookupPresence {clientIp} -> clientIp

7975 LookupPresence {clientIp} -> clientIp

7943 BatchLookupPresence {clientIp} -> clientIp

7976 BatchLookupPresence {clientIp} -> clientIp

7944 VerifyToken {clientIp} -> clientIp

7977 VerifyToken {clientIp} -> clientIp

7945− RecvGossip {} -> ""

7978+ _ -> ""

7946− ForwardMessage {} -> ""

7947− FederationAck {} -> ""

7948− FederationCancel {} -> ""

7949− FederationBatchCancel {} -> ""

7950− FederationBatchForward {} -> ""

7951− Prune {} -> ""

7952− Drain {} -> ""

7953− Shutdown {} -> ""

7954− Undrain {} -> ""

7955− BlockPeer {} -> ""

7956− UnblockPeer {} -> ""

7957− ListBlocked {} -> ""

7958− PingPeer {} -> ""

7959− AdminListSessions {} -> ""

7960− AdminRevokeSessions {} -> ""

7961− AdminListAllSessions {} -> ""

7962− AdminPurgeMessages {} -> ""

7963− AdminRevokeIp {} -> ""

7964− GetRateLimits {} -> ""

7965− SetRateLimits {} -> ""

7966− FlushNonces {} -> ""

7967− AllowPeer {} -> ""

7968− UnallowPeer {} -> ""

7969− ListAllowed {} -> ""

7970− Reconcile {} -> ""

7971− RotateSecret {} -> ""

7972− Health {} -> ""

7973− Stats {} -> ""

7974− StatsPeers {} -> ""

7975− StatsKey {} -> ""

7976− TopRecipients {} -> ""

7977− StatsForwards {} -> ""

7978− Ready {} -> ""

7979

7980-- Slow request threshold for access logging (ms)

7981slowRequestThreshold : Int<Ms>

@@ -8295,9 +8295,21 @@ backgroundPrune = do

8295

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

8297-- waiting too long (heartbeat). Ensures polls return within pollTimeoutMs even if no messages arrive.

8298+-- Adaptive: under high load (>80% capacity), the heartbeat fires more frequently

8299+-- (down to 5s at peak) to release poll slots faster and reduce connection hold time.

8300+-- Normal load uses the full pollTimeoutMs (~25s).

8298pollHeartbeat = do

8301pollHeartbeat = do

8302+ -- Adaptive poll timeout: scale heartbeat interval based on server load.

8303+ -- adaptiveRateScale returns 10-100 (percentage). At 100% (normal): full pollTimeoutMs.

8304+ -- At 10% (peak load): max(5000, pollTimeoutMs * 10 / 100) = 5000ms.

8305+ scale <- atomic do

8306+ tmc <- *totalMessageCount

8307+ sc <- *totalSessionCount

8308+ pc <- *totalPresenceCount

8309+ yield (adaptiveRateScale tmc sc pc)

8310+ let effectiveTimeout = max 5000 (pollTimeoutMs * scale / 100)

8299 jitter <- randomInt 2000

8311 jitter <- randomInt 2000

8300− sleep (pollTimeoutMs + jitter)

8312+ sleep (effectiveTimeout + jitter)

8301 dp <- *drainPhase

8313 dp <- *drainPhase

8302 shutdownReq <- *shutdownRequested

8314 shutdownReq <- *shutdownRequested

8303 dca <- *drainCompletedAt

8315 dca <- *drainCompletedAt