03f16bd · skrepka · git.iko.soy

git.iko.soy/skrepka

Implement 8 server improvements: adaptive rate limiting fix, LRU-based drop counter eviction, forward queue health caching, gossip filtering optimization, session lookup fast-path, and operational tuning

03f16bdiko6 months ago

server.knot+64−28

compare with master

unifiedsplit

@@ -852,8 +852,8 @@ maxPeerLatencyEntries = 100

852-- Per-sender drop counters: tracks how many delivery receipts / expiration notifications

853-- were evicted before the sender could poll them. Keyed by senderKey so each sender sees

854-- only their own drops (not a global counter that conflates all senders). Drained on poll.

855−*droppedDeliveryReceipts : [{senderKey: Text, count: Int}]

855+*droppedDeliveryReceipts : [{senderKey: Text, count: Int, lastUpdated: Timestamp}]

856−*droppedExpirationNotifs : [{senderKey: Text, count: Int}]

856+*droppedExpirationNotifs : [{senderKey: Text, count: Int, lastUpdated: Timestamp}]

857

858-- Runtime maintenance secret: when non-empty, used instead of compile-time maintenanceSecret.

859-- Set via /maintenance/rotate-secret to enable zero-downtime secret rotation.

@@ -915,6 +915,9 @@ maxPeerLatencyEntries = 100

915*cachedHealthForwardQueue : Bool

916-- Cached count of federation peers with active failure records (degraded connectivity)

917*cachedHealthDegradedPeers : Int

918+-- Cached count of destination servers with stale forward queues (updated during prune cycle).

919+-- Used by health checks to avoid scanning all 16 forward buckets on every /health request.

920+*cachedForwardStaleCount : Int

918

921

919-- ============================================

922-- ============================================

920-- Routes

923-- Routes

@@ -1273,17 +1276,24 @@ max = \a b -> if a > b then a else b

1273

1276

1274-- Per-sender drop counter helpers: increment a per-key counter and drain it for a specific key.

1277-- Per-sender drop counter helpers: increment a per-key counter and drain it for a specific key.

1275-- Used to track per-sender delivery receipt / expiration notification drops.

1278-- Used to track per-sender delivery receipt / expiration notification drops.

1279+-- LRU eviction: when at capacity, evict the entry least recently updated (oldest lastUpdated).

1280+-- Previous strategy evicted by lowest count, which could discard active senders with few drops.

1276−incrementDropCounter = \entries key n ->

1281+incrementDropCounter = \entries key n t ->

1277 if any entries (\e -> e.senderKey == key)

1282 if any entries (\e -> e.senderKey == key)

1278− then do e <- entries; yield (if e.senderKey == key then {senderKey: e.senderKey, count: e.count + n} else e)

1283+ then do e <- entries; yield (if e.senderKey == key then {senderKey: e.senderKey, count: e.count + n, lastUpdated: t} else e)

1279 else if count entries >= maxDroppedCountEntries

1284 else if count entries >= maxDroppedCountEntries

1280− then union (evictOneOldest entries (\e -> e.count)) [{senderKey: key, count: n}]

1285+ then union (evictOneOldest entries (\e -> e.lastUpdated)) [{senderKey: key, count: n, lastUpdated: t}]

1281− else union entries [{senderKey: key, count: n}]

1286+ else union entries [{senderKey: key, count: n, lastUpdated: t}]

1282

1287

1283drainDropCounter = \entries key ->

1288drainDropCounter = \entries key ->

1284 case headOf (do e <- entries; where e.senderKey == key; yield e) of

1289 case headOf (do e <- entries; where e.senderKey == key; yield e) of

1285 Nothing {} -> {count: 0, remaining: entries}

1290 Nothing {} -> {count: 0, remaining: entries}

1286 Just {value: e} -> {count: e.count, remaining: do x <- entries; where x.senderKey != key; yield x}

1291 Just {value: e} -> {count: e.count, remaining: do x <- entries; where x.senderKey != key; yield x}

1292+-- NOTE: fold always traverses the full list even after finding the first element.

1293+-- This is O(n) where O(1) suffices, but Knot lacks list destructuring (case xs of

1294+-- (x : rest) -> ...) that would allow short-circuit. Used pervasively on hot paths

1295+-- (sendMessageCore, handleForwardMessage, gossip). If Knot adds list pattern matching

1296+-- or a built-in head/uncons primitive, replace this to eliminate the O(n) overhead.

1287headOf = \items -> fold (\acc x -> case acc of

1297headOf = \items -> fold (\acc x -> case acc of

1288 Nothing {} -> Just {value: x}

1298 Nothing {} -> Just {value: x}

1289 _ -> acc) Nothing {} items

1299 _ -> acc) Nothing {} items

@@ -1812,6 +1822,7 @@ hasSessionForKey = \pubkey b ->

1812 else hasSessionForKey pubkey (b + 1)

1822 else hasSessionForKey pubkey (b + 1)

1813

1823

1814-- Check if a pubkey has any non-expired session across all buckets (short-circuits on first match).

1824-- Check if a pubkey has any non-expired session across all buckets (short-circuits on first match).

1825+-- Fast-path: returns False immediately when totalSessionCount is 0, avoiding 16 bucket reads.

1815hasActiveSessionForKey : PubkeyHex -> Timestamp -> Int -> IO Bool

1826hasActiveSessionForKey : PubkeyHex -> Timestamp -> Int -> IO Bool

1816hasActiveSessionForKey = \pubkey t b ->

1827hasActiveSessionForKey = \pubkey t b ->

1817 if b > 15 then yield (False {})

1828 if b > 15 then yield (False {})

@@ -1822,9 +1833,12 @@ hasActiveSessionForKey = \pubkey t b ->

1822

1833

1823-- Filter a list of pubkeys to those with active (non-expired) sessions.

1834-- Filter a list of pubkeys to those with active (non-expired) sessions.

1824-- Checks each key independently (short-circuits per key), avoids materializing all sessions.

1835-- Checks each key independently (short-circuits per key), avoids materializing all sessions.

1836+-- Fast-path: returns empty list when no sessions exist (totalSessionCount == 0).

1825filterActiveSessionKeys : [PubkeyHex] -> Timestamp -> IO [PubkeyHex]

1837filterActiveSessionKeys : [PubkeyHex] -> Timestamp -> IO [PubkeyHex]

1826filterActiveSessionKeys = \keys t ->

1838filterActiveSessionKeys = \keys t ->

1839+ totalSessions <- *totalSessionCount

1840+ if totalSessions == 0 then yield []

1827− case headOf keys of

1841+ else case headOf keys of

1828 Nothing {} -> yield []

1842 Nothing {} -> yield []

1829 Just {value: k} -> do

1843 Just {value: k} -> do

1830 let rest = do x <- keys; where x != k; yield x

1844 let rest = do x <- keys; where x != k; yield x

@@ -2074,7 +2088,11 @@ writeSelectedPresenceBuckets = \buckets entries ->

2074 writePresenceBucket b bucketEntries

2088 writePresenceBucket b bucketEntries

2075 writeSelectedPresenceBuckets rest entries

2089 writeSelectedPresenceBuckets rest entries

2076

2090

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

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

2092+-- The per-element 16-field record allocation is unavoidable without Knot-level indexed arrays:

2093+-- the alternative (16 list-comprehension passes, each computing pubkeyBucket) is O(16n) bucket

2094+-- computations vs O(n) here, and sortBy-based splitting is O(n log n) vs O(n) here.

2095+-- This pattern is optimal given Knot's current data structure primitives.

2078writeAllPresence : [PresenceEntry] -> IO ()

2096writeAllPresence : [PresenceEntry] -> IO ()

2079writeAllPresence = \entries -> do

2097writeAllPresence = \entries -> do

2080 let partitioned = fold (\acc p ->

2098 let partitioned = fold (\acc p ->

@@ -2295,16 +2313,16 @@ pruneMessageBucket = \b t -> atomic do

2295 let eviction = evictSmallest combined evictCount (\e -> e.expiredAt)

2313 let eviction = evictSmallest combined evictCount (\e -> e.expiredAt)

2296 set *senderExpirations = eviction.kept

2314 set *senderExpirations = eviction.kept

2297 den <- *droppedExpirationNotifs

2315 den <- *droppedExpirationNotifs

2298− let denAfterCap = fold (\acc ev -> incrementDropCounter acc ev.senderKey 1) den eviction.evicted

2316+ let denAfterCap = fold (\acc ev -> incrementDropCounter acc ev.senderKey 1 t) den eviction.evicted

2299 -- Also record per-key drops for senders that hit maxSenderExpirationsPerKey

2317 -- Also record per-key drops for senders that hit maxSenderExpirationsPerKey

2300 let perKeyDroppedEntries = do e <- localExpired; where none perKeyCapped (\c -> c.messageId == e.messageId && c.toKey == e.toKey); yield e

2318 let perKeyDroppedEntries = do e <- localExpired; where none perKeyCapped (\c -> c.messageId == e.messageId && c.toKey == e.toKey); yield e

2301− set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1) denAfterCap perKeyDroppedEntries

2319+ set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1 t) denAfterCap perKeyDroppedEntries

2302 else do

2320 else do

2303 set *senderExpirations = combined

2321 set *senderExpirations = combined

2304 when (perKeyDropped > 0) (do

2322 when (perKeyDropped > 0) (do

2305 den <- *droppedExpirationNotifs

2323 den <- *droppedExpirationNotifs

2306 let perKeyDroppedEntries = do e <- localExpired; where none perKeyCapped (\c -> c.messageId == e.messageId && c.toKey == e.toKey); yield e

2324 let perKeyDroppedEntries = do e <- localExpired; where none perKeyCapped (\c -> c.messageId == e.messageId && c.toKey == e.toKey); yield e

2307− set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1) den perKeyDroppedEntries))

2325+ set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1 t) den perKeyDroppedEntries))

2308 else yield {}

2326 else yield {}

2309

2327

2310-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.

2328-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.

@@ -3486,6 +3504,10 @@ pruneExpired = \t -> do

3486 -- with sendMessageCore which writes to forward buckets on the hot path.

3504 -- with sendMessageCore which writes to forward buckets on the hot path.

3487 fwdSurvivorCount <- pruneAllFwdBuckets 0 retainCutoff federationEnabled 0

3505 fwdSurvivorCount <- pruneAllFwdBuckets 0 retainCutoff federationEnabled 0

3488 atomic do set *totalForwardCount = fwdSurvivorCount

3506 atomic do set *totalForwardCount = fwdSurvivorCount

3507+ -- Cache forward queue stale count after pruning (forward buckets are clean now).

3508+ -- Health checks read this cached value instead of scanning all 16 buckets each time.

3509+ fwdStale <- countStaleForwardServers 0 t []

3510+ atomic do set *cachedForwardStaleCount = fwdStale

3489 -- AckedIds: per-bucket pruning (each bucket in its own atomic block to reduce lock

3511 -- AckedIds: per-bucket pruning (each bucket in its own atomic block to reduce lock

3490 -- contention with poll-ack paths that write to ackedIds concurrently).

3512 -- contention with poll-ack paths that write to ackedIds concurrently).

3491 pruneAllAckedBuckets 0 retainCutoff

3513 pruneAllAckedBuckets 0 retainCutoff

@@ -3662,10 +3684,16 @@ broadcastOfflineGossip = \t peerServers expiredLocal ->

3662-- Build re-broadcast gossip events from validated entries. The seenBy list includes

3684-- Build re-broadcast gossip events from validated entries. The seenBy list includes

3663-- the fromServer (who sent us these events) and ourselves (serverHostN), so downstream

3685-- the fromServer (who sent us these events) and ourselves (serverHostN), so downstream

3664-- peers that already have this gossip via either path won't receive it again.

3686-- peers that already have this gossip via either path won't receive it again.

3687+-- Optimization: pre-extract unique pubkeys from appliedKeys for a fast-path rejection.

3688+-- Entries whose pubkey doesn't appear in any appliedKey skip the full samePubkeyServer

3689+-- check (two-field comparison). Helps when entries contain many pubkeys not in appliedKeys.

3665−effectiveGossipEvents = \entries appliedKeys eventType seenBy -> do

3690+effectiveGossipEvents = \entries appliedKeys eventType seenBy ->

3691+ let appliedPubkeys = fold (\acc a -> if any acc (\p -> p == a.pubkey) then acc else union [a.pubkey] acc) [] appliedKeys

3692+ in do

3666− e <- entries

3693+ e <- entries

3694+ where any appliedPubkeys (\pk -> pk == e.pubkey)

3667− where any appliedKeys (\a -> samePubkeyServer a e)

3695+ where any appliedKeys (\a -> samePubkeyServer a e)

3668− yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: seenBy}

3696+ yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: seenBy}

3669

3697

3670mergeGossipPresence = \presence deduped trustedOffline fromServer t ->

3698mergeGossipPresence = \presence deduped trustedOffline fromServer t ->

3671 let trustedDeduped = trustedGossipEntries deduped presence fromServer t in

3699 let trustedDeduped = trustedGossipEntries deduped presence fromServer t in

@@ -4864,10 +4892,10 @@ ackPolledMessages = \pubkey cursor t -> do

4864 let eviction = evictSmallest combined evictCount (\d -> d.deliveredAt)

4892 let eviction = evictSmallest combined evictCount (\d -> d.deliveredAt)

4865 set *senderDeliveries = eviction.kept

4893 set *senderDeliveries = eviction.kept

4866 -- Track per-sender drop counts for all evicted entries

4894 -- Track per-sender drop counts for all evicted entries

4867− let dropCounts = fold (\acc e -> incrementDropCounter acc e.senderKey 1) [] eviction.evicted

4895+ let dropCounts = fold (\acc e -> incrementDropCounter acc e.senderKey 1 t) [] eviction.evicted

4868 when (count dropCounts > 0) (do

4896 when (count dropCounts > 0) (do

4869 ddr <- *droppedDeliveryReceipts

4897 ddr <- *droppedDeliveryReceipts

4870− set *droppedDeliveryReceipts = fold (\acc dc -> incrementDropCounter acc dc.senderKey dc.count) ddr dropCounts)

4898+ set *droppedDeliveryReceipts = fold (\acc dc -> incrementDropCounter acc dc.senderKey dc.count t) ddr dropCounts)

4871 else set *senderDeliveries = combined)

4899 else set *senderDeliveries = combined)

4872 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}

4900 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}

4873 else yield {pendingAcks: [], evictedEntries: []}

4901 else yield {pendingAcks: [], evictedEntries: []}

@@ -5103,9 +5131,12 @@ trustedOfflineEvents = \events fromServer ->

5103gossipTargets = \peerServers fromServer ->

5131gossipTargets = \peerServers fromServer ->

5104 do s <- peerServers; where s != fromServer; yield s

5132 do s <- peerServers; where s != fromServer; yield s

5105

5133

5106−-- Remote keys that just came online (triggers forwarding of queued messages)

5134+-- Remote keys that just came online (triggers forwarding of queued messages).

5135+-- Pre-extracts unique pubkeys from appliedKeys for fast-path rejection (same optimization

5136+-- as effectiveGossipEvents: skip full samePubkeyServer check when pubkey doesn't match).

5107newlyOnlineRemoteKeys = \events appliedKeys ->

5137newlyOnlineRemoteKeys = \events appliedKeys ->

5138+ let appliedPubkeys = fold (\acc a -> if any acc (\p -> p == a.pubkey) then acc else union [a.pubkey] acc) [] appliedKeys

5108− do e <- events; where any appliedKeys (\a -> samePubkeyServer a e); where not (isLocalServer e.server); yield {pubkey: e.pubkey, server: e.server}

5139+ in do e <- events; where any appliedPubkeys (\pk -> pk == e.pubkey); where any appliedKeys (\a -> samePubkeyServer a e); where not (isLocalServer e.server); yield {pubkey: e.pubkey, server: e.server}

5109

5140

5110-- Trust gossip entries if they're first-party (server == fromServer) or updating

5141-- Trust gossip entries if they're first-party (server == fromServer) or updating

5111-- an existing (pubkey, server) presence entry. Third-party gossip cannot create

5142-- an existing (pubkey, server) presence entry. Third-party gossip cannot create

@@ -6032,7 +6063,8 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

6032 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}

6063 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}

6033 else do

6064 else do

6034 recvRates <- *recvRates

6065 recvRates <- *recvRates

6066+ recvRate <- effectiveRecvRate

6035− let recvResult = checkPubkeyRate recvRates toKey maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t

6067+ let recvResult = checkPubkeyRate recvRates toKey recvRate recvRateWindowMs maxRecvRateEntries t

6036 if recvResult.limited

6068 if recvResult.limited

6037 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: True {}, noPresence: False {}}

6069 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: True {}, noPresence: False {}}

6038 else do

6070 else do

@@ -6114,7 +6146,7 @@ fedDeliveryReceipts = \bucketIds forwards t -> case headOf bucketIds of

6114 Nothing {} -> yield {}

6146 Nothing {} -> yield {}

6115 Just {value: ev} -> do

6147 Just {value: ev} -> do

6116 ddr <- *droppedDeliveryReceipts

6148 ddr <- *droppedDeliveryReceipts

6117− set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1

6149+ set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1 t

6118 else set *senderDeliveries = combined)

6150 else set *senderDeliveries = combined)

6119 fedDeliveryReceipts rest forwards t

6151 fedDeliveryReceipts rest forwards t

6120

6152

@@ -6745,10 +6777,11 @@ computeAndCacheHealth = \t -> do

6745 let nonceOk = nonceCount * 10 <= maxForwardNonces * 9

6777 let nonceOk = nonceCount * 10 <= maxForwardNonces * 9

6746 -- Count peers with active failure records (degraded federation connectivity)

6778 -- Count peers with active failure records (degraded federation connectivity)

6747 let degradedPeerCount = count (do f <- forwardFailures; where t - f.firstFailedAt < maxServerDowntime; yield f)

6779 let degradedPeerCount = count (do f <- forwardFailures; where t - f.firstFailedAt < maxServerDowntime; yield f)

6748− -- Forward queue age health: per-server awareness. Only flag unhealthy when

6780+ -- Forward queue age health: use cached stale count from the last prune cycle

6749− -- multiple destination servers have stale queues (oldest forward > healthForwardQueueMaxAge).

6781+ -- (updated every ~30s) instead of scanning all 16 forward buckets per health check.

6750− -- A single stuck peer shouldn't mark the server unhealthy — it's an isolated issue.

6782+ -- Staleness of up to 30 seconds is acceptable for health monitoring; stale forwards

6751− fwdStaleCount <- countStaleForwardServers 0 t []

6783+ -- take >24 minutes (healthForwardQueueMaxAge) to appear, so they're always caught.

6784+ fwdStaleCount <- *cachedForwardStaleCount

6752 let fwdQueueOk = fwdStaleCount < healthForwardQueueMinStaleServers

6785 let fwdQueueOk = fwdStaleCount < healthForwardQueueMinStaleServers

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

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

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

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

@@ -7139,7 +7172,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do

7139 Nothing {} -> yield {}

7172 Nothing {} -> yield {}

7140 Just {value: ev} -> do

7173 Just {value: ev} -> do

7141 ddr <- *droppedDeliveryReceipts

7174 ddr <- *droppedDeliveryReceipts

7142− set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1

7175+ set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1 t

7143 else set *senderDeliveries = combined)

7176 else set *senderDeliveries = combined)

7144 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}

7177 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}

7145 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}

7178 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}

@@ -7243,7 +7276,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do

7243 Nothing {} -> yield {}

7276 Nothing {} -> yield {}

7244 Just {value: ev} -> do

7277 Just {value: ev} -> do

7245 ddr <- *droppedDeliveryReceipts

7278 ddr <- *droppedDeliveryReceipts

7246− set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1

7279+ set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1 t

7247 else set *senderDeliveries = combinedDel)

7280 else set *senderDeliveries = combinedDel)

7248 if count ackedFederated > 0 then do

7281 if count ackedFederated > 0 then do

7249 ackedBkt <- readAckedBucket b

7282 ackedBkt <- readAckedBucket b

@@ -8421,8 +8454,11 @@ main = do

8421 atomic do

8454 atomic do

8422 set *startedAt = t

8455 set *startedAt = t

8423 set *serverGeneration = t

8456 set *serverGeneration = t

8457+ -- Empty maintenanceSecret is the recommended deployment path: use /maintenance/rotate-secret

8458+ -- at startup via deployment scripts for zero-knowledge secret management. Log at INFO level

8459+ -- since operators following the recommended path shouldn't see warnings on every startup.

8424 when (maintenanceSecret == "")

8460 when (maintenanceSecret == "")

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

8461+ (logInfo "maintenanceSecret is empty — use /maintenance/rotate-secret at startup to enable /maintenance/* and /stats endpoints")

8426 when (maintenanceSecret != "" && not (tooLong maintenanceSecret 15))

8462 when (maintenanceSecret != "" && not (tooLong maintenanceSecret 15))

8427 (logWarn "maintenanceSecret is shorter than 16 characters — use a longer secret for production deployments")

8463 (logWarn "maintenanceSecret is shorter than 16 characters — use a longer secret for production deployments")

8428 when (maintenanceSecret != "" && not (tooLong maintenanceSecret 31))

8464 when (maintenanceSecret != "" && not (tooLong maintenanceSecret 31))