@@ -497,6 +497,24 @@ evictOldest = \items getKey -> drop 1 (sortBy getKey items)
497capList = \items maxN getKey ->
497capList = \items maxN getKey ->
498 if count items > maxN then evictOldest items getKey else items
498 if count items > maxN then evictOldest items getKey else items
500+-- Integer percentage with zero-denominator guard.
501+pct = \cur maxN -> if maxN > 0 then cur * 100 / maxN else 0
503+-- Read entries matching `pred` (transformed by `extract`) and prune them in one step.
504+collectAndPrune = \ref pred extract -> do
506+ let mine = mapWhere items pred extract
507+ pruneRef ref (\x -> not (pred x))
510+-- Common ack/delete tail: pull matching messages, drop them and their forwards.
511+removeForKey = \pred -> do
513+ let toRemove = filter msgs pred
514+ set *messages = removeWhere msgs pred
515+ pruneRef forwards (\f -> not (pred f))
500-- If `pred` matches at least maxN items, drop the oldest matching one.
518-- If `pred` matches at least maxN items, drop the oldest matching one.
501capPerKey = \items pred maxN getKey ->
519capPerKey = \items pred maxN getKey ->
502 let mine = filter items pred in
520 let mine = filter items pred in
@@ -526,12 +544,12 @@ mapEach = \items getKey action -> case headOf items of
526 rest <- mapEach (filter items (\y -> getKey y != getKey x)) getKey action
544 rest <- mapEach (filter items (\y -> getKey y != getKey x)) getKey action
527 yield (union [r] rest)
545 yield (union [r] rest)
529−-- Hex tokens
547+-- Hex tokens (modulo lives inside hexDigit, so callers just pass raw bits).
530hexDigit : Int -> Text
548hexDigit : Int -> Text
531−hexDigit = \n -> take 1 (drop (clamp 0 15 n) "0123456789abcdef")
549+hexDigit = \n -> take 1 (drop (n - n / 16 * 16) "0123456789abcdef")
533byteToHex : Int -> Text
551byteToHex : Int -> Text
534−byteToHex = \n -> let m = n - n / 256 * 256 in hexDigit (m / 16) ++ hexDigit (m - m / 16 * 16)
552+byteToHex = \n -> hexDigit (n / 16) ++ hexDigit n
536intTo3Hex : Int -> Text
554intTo3Hex : Int -> Text
537intTo3Hex = \n -> byteToHex (n / 65536) ++ byteToHex (n / 256) ++ byteToHex n
555intTo3Hex = \n -> byteToHex (n / 65536) ++ byteToHex (n / 256) ++ byteToHex n
@@ -1135,15 +1153,9 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -
1135 yield (pluck watchKeys (\k -> {pubkey: k, online: count (presenceForKey pres k t2) > 0}))
1153 yield (pluck watchKeys (\k -> {pubkey: k, online: count (presenceForKey pres k t2) > 0}))
1136 else yield []
1154 else yield []
1137 drains <- atomic do
1155 drains <- atomic do
1138− se <- *senderExpirations
1139− let myExp = mapWhere se (\x -> x.senderKey == s.pubkey) (\x -> {messageId: x.messageId, toKey: x.toKey, expiredAt: x.expiredAt})
1156+ myExp <- collectAndPrune senderExpirations (\x -> x.senderKey == s.pubkey) (\x -> {messageId: x.messageId, toKey: x.toKey, expiredAt: x.expiredAt})
1140− pruneRef senderExpirations (\x -> x.senderKey != s.pubkey)
1141− sd <- *senderDeliveries
1142− let myDel = mapWhere sd (\x -> x.senderKey == s.pubkey) (\x -> {messageId: x.messageId, toKey: x.toKey, deliveredAt: x.deliveredAt})
1157+ myDel <- collectAndPrune senderDeliveries (\x -> x.senderKey == s.pubkey) (\x -> {messageId: x.messageId, toKey: x.toKey, deliveredAt: x.deliveredAt})
1143− pruneRef senderDeliveries (\x -> x.senderKey != s.pubkey)
1144− cn <- *cancelNotifications
1145− let myCancels = mapWhere cn (\x -> x.toKey == s.pubkey) (\x -> {messageId: x.messageId, cancelledAt: x.cancelledAt})
1158+ myCancels <- collectAndPrune cancelNotifications (\x -> x.toKey == s.pubkey) (\x -> {messageId: x.messageId, cancelledAt: x.cancelledAt})
1146− pruneRef cancelNotifications (\x -> x.toKey != s.pubkey)
1147 yield {expirations: myExp, deliveries: myDel, cancellations: myCancels}
1159 yield {expirations: myExp, deliveries: myDel, cancellations: myCancels}
1148 let isHeartbeat = pollResult.authorized && count pollResult.events == 0
1160 let isHeartbeat = pollResult.authorized && count pollResult.events == 0
1149 yield (respond {events: pollResult.events, cursor: pollResult.cursor, authorized: pollResult.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: pollResult.hasMore, cursorStale: False {}, messageExpirations: drains.expirations, messageDeliveries: drains.deliveries, droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: drains.cancellations})
1161 yield (respond {events: pollResult.events, cursor: pollResult.cursor, authorized: pollResult.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: pollResult.hasMore, cursorStale: False {}, messageExpirations: drains.expirations, messageDeliveries: drains.deliveries, droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: drains.cancellations})
@@ -1312,10 +1324,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
1312 let dedupedIds = dedupMessageIds messageIds
1324 let dedupedIds = dedupMessageIds messageIds
1313 let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId
1325 let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId
1314 result <- atomic do
1326 result <- atomic do
1315− msgs <- *messages
1327+ toAck <- removeForKey matches
1316− let toAck = filter msgs matches
1317− set *messages = removeWhere msgs matches
1318− pruneRef forwards (\f -> not (matches f))
1319 ackedFed <- recordAcksAndDeliveries toAck t
1328 ackedFed <- recordAcksAndDeliveries toAck t
1320 yield {ackedCount: count toAck, pendingAcks: ackedFed}
1329 yield {ackedCount: count toAck, pendingAcks: ackedFed}
1321 fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
1330 fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
@@ -1340,10 +1349,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
1340 let dedupedIds = dedupMessageIds messageIds
1349 let dedupedIds = dedupMessageIds messageIds
1341 let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId
1350 let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId
1342 result <- atomic do
1351 result <- atomic do
1343− msgs <- *messages
1352+ toDel <- removeForKey matches
1344− let toDel = filter msgs matches
1345− set *messages = removeWhere msgs matches
1346− pruneRef forwards (\f -> not (matches f))
1347 let ackedFed = mapWhere toDel (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
1353 let ackedFed = mapWhere toDel (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
1348 appendCapped ackedIds ackedFed maxAckedIds (\a -> a.ackedAt)
1354 appendCapped ackedIds ackedFed maxAckedIds (\a -> a.ackedAt)
1349 yield {deletedCount: count toDel, pendingAcks: ackedFed}
1355 yield {deletedCount: count toDel, pendingAcks: ackedFed}
@@ -1812,7 +1818,7 @@ handleStats = \authorization respond -> withMaintAuth authorization
1812 let fCount = count fwds
1818 let fCount = count fwds
1813 let zeroFed = {enabled: federationEnabled, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: count bp, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: [], pendingCancels: count pfc, gossipSuccess: 0, gossipFail: 0, latencyBuckets: {lt100: 0, lt500: 0, lt2000: 0, gte2000: 0}}
1819 let zeroFed = {enabled: federationEnabled, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: count bp, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: [], pendingCancels: count pfc, gossipSuccess: 0, gossipFail: 0, latencyBuckets: {lt100: 0, lt500: 0, lt2000: 0, gte2000: 0}}
1814 let zeroRates = {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0, clientCancels: 0, federationCancels: 0}
1820 let zeroRates = {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0, clientCancels: 0, federationCancels: 0}
1815− let cap = {sessions: sCount, sessionsMax: maxSessions, sessionsPct: if maxSessions > 0 then sCount * 100 / maxSessions else 0, messages: mCount, messagesMax: maxMessages, messagesPct: if maxMessages > 0 then mCount * 100 / maxMessages else 0, totalMessageCount: mCount, presence: pCount, presenceMax: maxPresenceEntries, presencePct: if maxPresenceEntries > 0 then pCount * 100 / maxPresenceEntries else 0, forwards: fCount, forwardsMax: maxForwardsEntries, forwardsPct: if maxForwardsEntries > 0 then fCount * 100 / maxForwardsEntries else 0, challenges: count chs, challengesMax: maxChallenges, ackedIds: count aids, ackedIdsMax: maxAckedIds, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: count dids, deliveredIdsMax: maxDeliveredIds, forwardNonces: 0, forwardNoncesMax: 0, nonceEvictions: 0, activePolls: count aps, activePollsMax: maxActivePolls, recipientCounts: 0, messageBucketMax: 0, messageBucketMin: 0}
1821+ let cap = {sessions: sCount, sessionsMax: maxSessions, sessionsPct: pct sCount maxSessions, messages: mCount, messagesMax: maxMessages, messagesPct: pct mCount maxMessages, totalMessageCount: mCount, presence: pCount, presenceMax: maxPresenceEntries, presencePct: pct pCount maxPresenceEntries, forwards: fCount, forwardsMax: maxForwardsEntries, forwardsPct: pct fCount maxForwardsEntries, challenges: count chs, challengesMax: maxChallenges, ackedIds: count aids, ackedIdsMax: maxAckedIds, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: count dids, deliveredIdsMax: maxDeliveredIds, forwardNonces: 0, forwardNoncesMax: 0, nonceEvictions: 0, activePolls: count aps, activePollsMax: maxActivePolls, recipientCounts: 0, messageBucketMax: 0, messageBucketMin: 0}
1816 let timing = {pruneCount: pc, pruneRunning: 0, heartbeatCount: hc, uptime: t - sa, lastPruneAt: lp, lastHeartbeatAt: lh, draining: dp > 0, drainPhase: dp, totalEntries: sCount + mCount + pCount + fCount, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, pollLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0, sendLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, authAvgMs: 0, authMaxMs: 0, authCount: 0}
1822 let timing = {pruneCount: pc, pruneRunning: 0, heartbeatCount: hc, uptime: t - sa, lastPruneAt: lp, lastHeartbeatAt: lh, draining: dp > 0, drainPhase: dp, totalEntries: sCount + mCount + pCount + fCount, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, pollLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0, sendLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, authAvgMs: 0, authMaxMs: 0, authCount: 0}
1817 yield (respond {ok: True {}, version: serverBuildVersion, capacity: cap, federation: zeroFed, rates: zeroRates, timing: timing}))
1823 yield (respond {ok: True {}, version: serverBuildVersion, capacity: cap, federation: zeroFed, rates: zeroRates, timing: timing}))
@@ -1864,8 +1870,7 @@ handleStatsForwards = \authorization respond -> withMaintAuth authorization
1864 let servers = collectPeers fwds (\f -> f.toServer)
1870 let servers = collectPeers fwds (\f -> f.toServer)
1865 let result = pluck servers (\s ->
1871 let result = pluck servers (\s ->
1866 let svFwds = filter fwds (\f -> f.toServer == s) in
1872 let svFwds = filter fwds (\f -> f.toServer == s) in
1867− let oldest = case minBy svFwds (\f -> f.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt in
1868− {server: s, count: count svFwds, oldestAge: oldest, totalRetries: fold (\acc f -> acc + f.retries) 0 svFwds})
1873+ {server: s, count: count svFwds, oldestAge: t - fold (\acc f -> min acc f.createdAt) t svFwds, totalRetries: fold (\acc f -> acc + f.retries) 0 svFwds})
1869 yield (respond {servers: result, ok: True {}, total: count fwds}))
1874 yield (respond {servers: result, ok: True {}, total: count fwds}))
1871handleStatsRates = \authorization respond -> do
1876handleStatsRates = \authorization respond -> do