@@ -205,7 +205,8 @@ unackedMessageMaxAge : Int<Ms>
205205unackedMessageMaxAge = 604800000 -- 7 days
206206-- Grace period before federation ack can delete the original message (allows local client to poll)
207207ackGracePeriod : Int<Ms>
208−ackGracePeriod = 30000 -- 30 seconds
208+ackGracePeriod = 120000 -- 2 minutes; allows clients on slow connections or backgrounded apps
209+-- to poll before federation ack reclaims the origin message
209210-- Deferred ack cleanup capacity
210211maxPendingAckCleanups = 10000
211212-- Per-key deferred ack cleanup cap (prevents one noisy key from dominating the queue)
@@ -633,6 +634,10 @@ route ClientApi where
633634 GET /messages/list headers {authorization: Text, clientIp: Text}
634635 -> {messages: [{messageId: MessageIdHex, timestamp: Timestamp, receivedAt: Timestamp}], authorized: Bool, count: Int}
635636 = ListMessages
637+ -- Delete pending messages by ID (recipient-only; messages don't store sender pubkey)
638+ POST {messageIds: [MessageIdHex]} /delete headers {authorization: Text, clientIp: Text}
639+ -> {ok: Bool, authorized: Bool, deletedCount: Int}
640+ = DeleteMessages
636641 -- Look up which server a pubkey is present on (for federation routing)
637642 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
638643 -> {server: AnyServerName, authorized: Bool}
@@ -713,13 +718,11 @@ route HealthApi where
713718-- TODO: If Knot adds raw text response support, add a /metrics endpoint
714719-- returning Prometheus exposition format for direct integration with standard
715720-- monitoring stacks. Currently /stats returns JSON which requires a custom scraper.
716−-- TODO: Consider grouping the flat 60+ field response into nested objects
717−-- (e.g., {capacity: {messages, sessions, ...}, federation: {avgLatency, ...}})
718−-- for better monitoring dashboard ergonomics. This is a wire format change
719−-- that would require coordinating with any existing scrapers.
721+-- Stats response grouped into nested objects for monitoring dashboard ergonomics.
722+-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
720723route StatsApi where
721724 GET /stats headers {authorization: Text}
722− -> {sessions: Int, sessionsMax: Int, messages: Int, messagesMax: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pairRates: Int, federationPairRates: Int, pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, federationEnabled: Bool, recentGossip: Int, totalMessageCount: Int, recipientCounts: Int, federationInFlight: Int, hungFederationSlots: Int, messagesPct: Int, presencePct: Int, forwardsPct: Int, sessionsPct: Int, federationAvgLatencyMs: Int, federationMaxLatencyMs: Int, federationCallCount: Int, drainCompletedAt: Timestamp, oldestForwardAge: Int, oldestAckAge: Int, forwardNonces: Int, forwardNoncesMax: Int, pendingAckCleanups: Int, blockedPeers: Int, gossipEventRates: Int, verifyIpRates: Int, uniqueKeyIpRates: Int, ok: Bool}
725+ -> {ok: Bool, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, activePolls: Int, activePollsMax: Int, recipientCounts: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool}}
723726 = Stats
724727 GET /stats/peers headers {authorization: Text}
725728 -> {peers: [PeerStatus], ok: Bool}
@@ -784,8 +787,12 @@ genTokenN = \n -> if n <= 0 then yield "" else do
784787 rest <- genTokenN (n - 1)
785788 yield (intTo3Hex chunk ++ rest)
786789
787−-- Default token: 8 chunks = 48 hex chars = 24 bytes of randomness
788−genToken = genTokenN 8
790+-- Token generation parameters (linked to safeCompareLen for consistency)
791+tokenChunks = 8
792+tokenHexLen = 6 * tokenChunks -- 48; each chunk produces 6 hex chars (3 bytes)
793+
794+-- Default token: tokenChunks chunks = tokenHexLen hex chars = 24 bytes of randomness
795+genToken = genTokenN tokenChunks
789796
790797tooLong : Text -> Int -> Bool
791798tooLong = \s maxLen -> drop maxLen s != ""
@@ -1027,6 +1034,8 @@ generateFederationAuth = \scope -> do
10271034 yield ("Bearer " ++ token)
10281035
10291036-- Verify an inbound federation token by callback to the sender's /federation/verify endpoint.
1037+-- Uses fetchFederation for in-flight tracking, connection counting, and slow-response logging,
1038+-- preventing a malicious peer from slow-loris attacks on the verify endpoint.
10301039-- Single-use: the sender consumes the token on verification.
10311040-- scope must match the scope the token was generated with; empty scope accepts any (backward compat).
10321041-- No retry: a failed verify rejects the request. The peer's federation rate limit + circuit breaker
@@ -1037,7 +1046,7 @@ verifyFederationCallback = \fromServer authorization scope ->
10371046 case extractToken authorization of
10381047 Nothing {} -> yield (False {})
10391048 Just {value: token} -> do
1040− result <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
1049+ result <- fetchFederation fromServer (VerifyToken {token: token, scope: scope})
10411050 case result of
10421051 Err e -> do
10431052 logWarn ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
@@ -1506,6 +1515,10 @@ pruneMessageBucket = \b messageCutoff -> atomic do
15061515-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
15071516-- Each bucket's atomic block only contends on *totalMessageCount (a single Int);
15081517-- rcBuckets and message buckets are per-bucket and independent.
1518+-- STM contention on the shared counter is acceptable because: (1) per-bucket prune
1519+-- is fast (list filter + counter decrement), (2) prune runs every ~30s, not on the
1520+-- hot path, (3) Knot's fork is fire-and-forget so we can't collect per-bucket
1521+-- deltas and apply them in a single final atomic block.
15091522pruneMessageBucketsConcurrent = \b messageCutoff ->
15101523 if b > 15 then yield {}
15111524 else if b < 15 then do
@@ -1704,10 +1717,11 @@ dedupMessageIds = \messageIds ->
17041717 else union acc [mid]) [] capped
17051718
17061719
1707−-- Constant-time comparison iteration count. Must be >= the longest secret
1708−-- compared (tokens = 48 hex chars, maintenance secret = variable). Using 64
1709−-- provides headroom without the waste of maxTokenLen (256 = 5x actual length).
1710−safeCompareLen = 64
1720+-- Constant-time comparison iteration count. Derived from tokenHexLen (48) + 16
1721+-- headroom. Must be >= the longest secret compared (tokens = tokenHexLen hex
1722+-- chars, maintenance secret = variable). If tokenChunks changes, this adjusts
1723+-- automatically.
1724+safeCompareLen = tokenHexLen + 16
17111725
17121726-- Constant-time string comparison to prevent timing side-channels.
17131727-- Always iterates safeCompareLen characters regardless of input length;
@@ -3272,6 +3286,36 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
32723286 else
32733287 logWarn ("federation: forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))
32743288
3289+-- Send all batch forwards for one server sequentially (no per-message fork overhead).
3290+-- Used by handleSendBatch; reduces fork count from O(messages * servers) to O(servers).
3291+sendBatchForwardsSequential = \msgs server ->
3292+ case headOf msgs of
3293+ Nothing {} -> yield {}
3294+ Just {value: r} -> do
3295+ let rest = do m <- msgs; where m.messageId != r.messageId; yield m
3296+ stillPending <- atomic do
3297+ forwards <- *forwards
3298+ yield (any forwards (\f -> f.messageId == r.messageId && f.toServer == server))
3299+ when stillPending (do
3300+ fwdT <- now
3301+ fwdAuth <- generateFederationAuth "forward"
3302+ fwdResult <- fetchFederation server (ForwardMessage {
3303+ toKey: r.toKey, encryptedBlob: r.blob,
3304+ timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
3305+ fromServer: serverHostN, requestTimestamp: fwdT,
3306+ authorization: fwdAuth})
3307+ case fwdResult of
3308+ Err e -> do
3309+ logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
3310+ recordFailure server
3311+ Ok response ->
3312+ if response.ok then do
3313+ removeForward r.messageId server
3314+ clearFailure server
3315+ else
3316+ logWarn ("federation: forward to " ++ server ++ " rejected for " ++ r.messageId ++ ": " ++ response.error))
3317+ sendBatchForwardsSequential rest server
3318+
32753319-- POST /messages: Send an encrypted message to a recipient.
32763320-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
32773321-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
@@ -3535,6 +3579,10 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
35353579 let validated = validGossipEvents events t
35363580 let onlineEntries = validated.online
35373581 let trustedOffline = trustedOfflineEvents validated.offline fromServer
3582+ -- Early return: skip the expensive presence merge when all events are invalid/expired
3583+ if count onlineEntries == 0 && count trustedOffline == 0
3584+ then yield (respond {ok: True {}, error: ""})
3585+ else do
35383586 -- Warn if a peer sends both online and offline for the same key in one batch (suspicious)
35393587 let conflictingKeys = do on <- onlineEntries; off <- trustedOffline; where on.pubkey == off.pubkey && on.server == off.server; yield on.pubkey
35403588 when (count conflictingKeys > 0)
@@ -3622,46 +3670,46 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
36223670 if t - timestamp > messageMaxAge
36233671 then yield (respond {ok: False {}, error: "stale_timestamp"})
36243672 else do
3625− -- Verify recipient has active presence on this server. Peers should only forward
3626− -- to us if gossip indicated the key is here. Accepting messages for completely
3627− -- unknown keys wastes mailbox slots that will never be drained by polling.
3628− hasPresence <- atomic do
3629− bkt <- readPresenceBucket (pubkeyBucket toKey)
3630− yield (any bkt (\p -> p.pubkey == toKey && p.expiresAt > t))
3631− if not hasPresence then do
3632− logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
3633− yield (respond {ok: False {}, error: "no_presence"})
3634− else do
3673+ -- Presence check + message storage in a single atomic block to prevent TOCTOU
3674+ -- race where a key goes offline between presence check and store.
36353675 let clampedTs = clampTimestamp timestamp t
36363676 result <- atomic do
3677+ -- Verify recipient has active presence on this server. Peers should only forward
3678+ -- to us if gossip indicated the key is here. Accepting messages for completely
3679+ -- unknown keys wastes mailbox slots that will never be drained by polling.
36373680 let msgBucket = pubkeyBucket toKey
3681+ bkt <- readPresenceBucket msgBucket
3682+ let hasPresence = any bkt (\p -> p.pubkey == toKey && p.expiresAt > t)
3683+ if not hasPresence
3684+ then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: True {}}
3685+ else do
36383686 bucket <- readMessageBucket msgBucket
36393687 ackedIds <- *ackedIds
36403688 deliveredIds <- *deliveredIds
36413689 forwardNonces <- *forwardNonces
36423690 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)
36433691 if isDup
3644− then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}}
3692+ then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
36453693 else do
36463694 totalMsgCount <- *totalMessageCount
36473695 if totalMsgCount >= maxMessages
3648− then yield {forward: ForwardRejected {}, globalCapReached: True {}, recvRateLimited: False {}}
3696+ then yield {forward: ForwardRejected {}, globalCapReached: True {}, recvRateLimited: False {}, noPresence: False {}}
36493697 else do
36503698 recipientCounts <- readRcBucket msgBucket
36513699 let recipientCount = getRecipientCount toKey recipientCounts
36523700 if recipientCount >= maxMessagesPerRecipient
3653− then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
3701+ then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
36543702 else do
36553703 recvRates <- *recvRates
36563704 let recvResult = checkPubkeyRate recvRates toKey maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
36573705 if recvResult.limited
3658− then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: True {}}
3706+ then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: True {}, noPresence: False {}}
36593707 else do
36603708 set *recvRates = recvResult.entries
36613709 federationPairRates <- *federationPairRates
36623710 let fedPairResult = checkServerRecipientRate federationPairRates fromServer toKey maxFederationPairRatePerPair federationPairRateWindowMs maxFederationPairRateEntries t
36633711 if fedPairResult.limited
3664− then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
3712+ then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
36653713 else do
36663714 set *federationPairRates = fedPairResult.entries
36673715 seqCounter <- *seqCounter
@@ -3684,7 +3732,11 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
36843732 then let evictCount = max 1 (count updatedNonces / 10) in
36853733 (evictSmallest updatedNonces evictCount (\n -> n.seenAt)).kept
36863734 else updatedNonces
3687− yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
3735+ yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
3736+ if result.noPresence then do
3737+ logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
3738+ yield (respond {ok: False {}, error: "no_presence"})
3739+ else do
36883740 case result.forward of
36893741 ForwardRejected {} ->
36903742 if result.globalCapReached
@@ -3877,7 +3929,7 @@ handleHealth = \authorization respond -> do
38773929 snapshotHealthy: True {},
38783930 watchdogHealthy: True {}})
38793931
3880−zeroStats = {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pairRates: 0, federationPairRates: 0, pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, hungFederationSlots: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, drainCompletedAt: 0, oldestForwardAge: 0, oldestAckAge: 0, forwardNonces: 0, forwardNoncesMax: 0, pendingAckCleanups: 0, blockedPeers: 0, gossipEventRates: 0, verifyIpRates: 0, uniqueKeyIpRates: 0, ok: False {}}
3932+zeroStats = {ok: False {}, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0}, rates: {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}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}}}
38813933
38823934handleStats = \authorization respond -> do
38833935 t <- now
@@ -3932,16 +3984,24 @@ handleStats = \authorization respond -> do
39323984 flm <- *federationLatencyMax
39333985 dca <- *drainCompletedAt
39343986 yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, dca: dca}
3935− let result = {sessions: core.sc, messages: core.tmc, presence: core.pc, forwards: count core.f, challenges: count core.c, ackedIds: count core.a, ackOverflow: count core.ao, deliveredIds: count core.d, forwardFailures: count core.ff, activePolls: count core.ap, sendRates: count counters.sr, challengeRates: count counters.cr, federationRates: count counters.fr, globalRateCount: counters.grc, ipRates: count counters.ir, lookupRates: count counters.lr, recvRates: count counters.rr, pairRates: count counters.prr, federationPairRates: count counters.fpr, pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr, recentGossip: count core.rg, totalMessageCount: core.tmc, recipientCounts: count core.rc, federationInFlight: count core.fi, federationLatencyTotal: counters.flt, federationLatencyCount: counters.flc, federationLatencyMax: counters.flm, drainCompletedAt: counters.dca}
3936− let msgPct = if maxMessages > 0 then result.messages * 100 / maxMessages else 0
3937− let presPct = if maxPresenceEntries > 0 then result.presence * 100 / maxPresenceEntries else 0
3938− let fwdPct = if maxForwardsEntries > 0 then result.forwards * 100 / maxForwardsEntries else 0
3939− let sessPct = if maxSessions > 0 then result.sessions * 100 / maxSessions else 0
3940− let fedAvg = if result.federationLatencyCount > 0 then result.federationLatencyTotal / result.federationLatencyCount else 0
3987+ -- Compute derived metrics
3988+ let sessions = core.sc
3989+ let messages = core.tmc
3990+ let presence = core.pc
3991+ let fwds = count core.f
3992+ let msgPct = if maxMessages > 0 then messages * 100 / maxMessages else 0
3993+ let presPct = if maxPresenceEntries > 0 then presence * 100 / maxPresenceEntries else 0
3994+ let fwdPct = if maxForwardsEntries > 0 then fwds * 100 / maxForwardsEntries else 0
3995+ let sessPct = if maxSessions > 0 then sessions * 100 / maxSessions else 0
3996+ let fedAvg = if counters.flc > 0 then counters.flt / counters.flc else 0
39413997 let hungFedSlots = count (do r <- core.fi; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs; yield r)
39423998 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
39433999 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt
3944− yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pairRates: result.pairRates, federationPairRates: result.federationPairRates, pruneCount: result.pruneCount, pruneRunning: result.pruneRunning, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, draining: result.draining, federationEnabled: federationEnabled, recentGossip: result.recentGossip, totalMessageCount: result.totalMessageCount, recipientCounts: result.recipientCounts, federationInFlight: result.federationInFlight, hungFederationSlots: hungFedSlots, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, drainCompletedAt: result.drainCompletedAt, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, forwardNonces: count core.fn, forwardNoncesMax: maxForwardNonces, pendingAckCleanups: count core.pac, blockedPeers: count core.bp, gossipEventRates: count core.ger, verifyIpRates: count core.vir, uniqueKeyIpRates: count core.ukir, ok: True {}})
4000+ yield (respond {ok: True {},
4001+ capacity: {sessions: sessions, sessionsMax: maxSessions, sessionsPct: sessPct, messages: messages, messagesMax: maxMessages, messagesPct: msgPct, totalMessageCount: messages, presence: presence, presenceMax: maxPresenceEntries, presencePct: presPct, forwards: fwds, forwardsMax: maxForwardsEntries, forwardsPct: fwdPct, challenges: count core.c, challengesMax: maxChallenges, ackedIds: count core.a, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: count core.d, deliveredIdsMax: maxDeliveredIds, forwardNonces: count core.fn, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
4002+ federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca},
4003+ rates: {send: count counters.sr, challenge: count counters.cr, federation: count counters.fr, federationPair: count counters.fpr, global: counters.grc, ip: count counters.ir, lookup: count counters.lr, recv: count counters.rr, pair: count counters.prr, gossipEvent: count core.ger, verifyIp: count core.vir, uniqueKeyIp: count core.ukir, pendingAckCleanups: count core.pac},
4004+ timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr}})
39454005 else yield (respond zeroStats)
39464006
39474007handleStatsPeers = \authorization respond -> do
@@ -4153,6 +4213,46 @@ handleListMessages = \authorization clientIp respond -> do
41534213 yield {msgs: do m <- capped; yield {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt}, total: count myMsgs}
41544214 yield (respond {messages: result.msgs, authorized: True {}, count: result.total}))
41554215
4216+-- POST /messages/delete: Delete pending messages by ID (recipient-only).
4217+-- The recipient knows their pubkey (toKey) and can be authenticated via session.
4218+-- Deletes matching messages, adjusts counters, and cancels pending forwards.
4219+-- NOTE: Deleted messages won't trigger federation acks to origin servers,
4220+-- potentially leaving stale forward entries on origin servers until prune.
4221+handleDeleteMessages = \messageIds authorization clientIp respond -> do
4222+ t <- now
4223+ globalRateLimited <- checkGlobalRate t
4224+ if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0})
4225+ else
4226+ withSessionAuth authorization t clientIp
4227+ (yield (respond {ok: False {}, authorized: False {}, deletedCount: 0}))
4228+ (\pubkey -> do
4229+ deleteRateLimited <- atomic do
4230+ lookupRates <- *lookupRates
4231+ let rateResult = checkPubkeyRate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
4232+ set *lookupRates = rateResult.entries
4233+ yield rateResult.limited
4234+ if deleteRateLimited then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0})
4235+ else do
4236+ let dedupedIds = dedupMessageIds messageIds
4237+ deletedCount <- atomic do
4238+ let b = pubkeyBucket pubkey
4239+ bucket <- readMessageBucket b
4240+ let toDelete = do m <- bucket; where m.toKey == pubkey && any dedupedIds (\mid -> mid == m.messageId); yield m
4241+ let removed = count toDelete
4242+ if removed > 0 then do
4243+ writeMessageBucket b (diff bucket toDelete)
4244+ tc <- *totalMessageCount
4245+ set *totalMessageCount = max 0 (tc - removed)
4246+ rc <- readRcBucket b
4247+ writeRcBucket b (adjustRecipientCount rc pubkey (0 - removed))
4248+ -- Cancel pending forwards for deleted messages
4249+ let deletedMsgIds = do m <- toDelete; yield m.messageId
4250+ forwards <- *forwards
4251+ set *forwards = do f <- forwards; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
4252+ yield removed
4253+ else yield 0
4254+ yield (respond {ok: True {}, authorized: True {}, deletedCount: deletedCount}))
4255+
41564256-- POST /messages/batch: Send multiple messages in one request.
41574257-- Auth checked once; each message processed independently with its own rate limiting.
41584258-- Partial success is possible — check per-message status in results.
@@ -4166,18 +4266,21 @@ handleSendBatch = \messages authorization clientIp respond -> do
41664266 (\senderKey -> do
41674267 let capped = takeRelation maxMessageIds messages
41684268 results <- sendBatchInner senderKey capped t []
4169− -- Fork federation forwarding for all successful sends
4269+ -- Fork federation forwarding: one forked task per target server (not per message).
4270+ -- Within each task, forwards are sent sequentially to avoid per-message fork overhead.
41704271 let forwardTargets = do r <- results; where count r.remoteServers > 0; yield r
4272+ let allServers = collectPeerServers forwardTargets (\r -> case headOf r.remoteServers of Nothing {} -> ""; Just {value: s} -> s)
41714273 fork (when (federationEnabled && count forwardTargets > 0)
4172− (sendToEach (collectPeerServers forwardTargets (\r -> case headOf r.remoteServers of Nothing {} -> ""; Just {value: s} -> s)) (\server -> do
4274+ (sendToEach allServers (\server -> do
41734275 let serverMsgs = do r <- forwardTargets; where any r.remoteServers (\s -> s == server); yield r
4174− forEachBatch serverMsgs maxMessageIds (\batch -> do
4175− forEachBatch batch 1 (\single -> case headOf single of
4176− Nothing {} -> yield {}
4177− Just {value: r} -> forkSendForwards r.messageId r.toKey r.blob r.ts (do s <- r.remoteServers; where s == server; yield s))))))
4276+ sendBatchForwardsSequential serverMsgs server)))
41784277 let batchRetryMs = fold (\acc r -> max acc r.retryAfterMs) 0 results
41794278 yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}, retryAfterMs: batchRetryMs}))
41804279
4280+-- TODO: If Knot adds a map/dictionary type, group batch messages by pubkeyBucket
4281+-- and process each group in a single atomic block per bucket. Currently each message
4282+-- gets its own atomic block via sendMessageCore, re-reading the same bucket when
4283+-- multiple messages target the same recipient. Grouping without maps is O(n^2).
41814284sendBatchInner = \senderKey messages t acc ->
41824285 let split = splitAtN messages 1 in
41834286 case headOf split.batch of
@@ -4322,6 +4425,7 @@ requestPath = \req -> case req of
43224425 AckMessages {} -> "POST /messages/ack"
43234426 SendBatch {} -> "POST /messages/batch"
43244427 ListMessages {} -> "GET /messages/list"
4428+ DeleteMessages {} -> "POST /messages/delete"
43254429 LookupPresence {} -> "GET /lookup"
43264430 RecvGossip {} -> "POST /federation/gossip"
43274431 ForwardMessage {} -> "POST /federation/forward"
@@ -4343,6 +4447,40 @@ requestPath = \req -> case req of
43434447 StatsPeers {} -> "GET /stats/peers"
43444448 StatsKey {} -> "GET /stats/key"
43454449
4450+-- Extract client IP for access logging (empty for federation/maintenance routes)
4451+requestClientIp = \req -> case req of
4452+ ReqChallenge {clientIp} -> clientIp
4453+ VerifyAuth {clientIp} -> clientIp
4454+ RevokeSession {clientIp} -> clientIp
4455+ Logout {clientIp} -> clientIp
4456+ Poll {clientIp} -> clientIp
4457+ SendMessage {clientIp} -> clientIp
4458+ MessageStatus {clientIp} -> clientIp
4459+ AckMessages {clientIp} -> clientIp
4460+ SendBatch {clientIp} -> clientIp
4461+ ListMessages {clientIp} -> clientIp
4462+ DeleteMessages {clientIp} -> clientIp
4463+ LookupPresence {clientIp} -> clientIp
4464+ VerifyToken {clientIp} -> clientIp
4465+ RecvGossip {} -> ""
4466+ ForwardMessage {} -> ""
4467+ FederationAck {} -> ""
4468+ Prune {} -> ""
4469+ Drain {} -> ""
4470+ Undrain {} -> ""
4471+ Snapshot {} -> ""
4472+ Restore {} -> ""
4473+ BlockPeer {} -> ""
4474+ UnblockPeer {} -> ""
4475+ ListBlocked {} -> ""
4476+ PingPeer {} -> ""
4477+ AdminListSessions {} -> ""
4478+ AdminRevokeSessions {} -> ""
4479+ Health {} -> ""
4480+ Stats {} -> ""
4481+ StatsPeers {} -> ""
4482+ StatsKey {} -> ""
4483+
43464484-- Slow request threshold for access logging (ms)
43474485slowRequestThreshold : Int<Ms>
43484486slowRequestThreshold = 1000
@@ -4358,6 +4496,7 @@ serveInner = \req -> case req of
43584496 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
43594497 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
43604498 ListMessages {authorization, clientIp, respond} -> handleListMessages authorization clientIp respond
4499+ DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
43614500 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
43624501 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
43634502 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond
@@ -4379,18 +4518,22 @@ serveInner = \req -> case req of
43794518 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
43804519 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
43814520
4382−-- Access-logged request handler: wraps serveInner with timing.
4383−-- Logs all non-poll requests. Polls are only logged when slow (> slowRequestThreshold)
4384−-- to avoid overwhelming logs with high-frequency long-poll traffic.
4521+-- Access-logged request handler: wraps serveInner with timing and correlation ID.
4522+-- Logs all non-poll requests with a 6-hex-char request ID for log correlation.
4523+-- Polls are only logged when slow (> slowRequestThreshold) to avoid overwhelming
4524+-- logs with high-frequency long-poll traffic.
43854525serve = \req -> do
43864526 startTime <- now
4527+ reqId <- genTokenN 1
43874528 serveInner req
43884529 endTime <- now
43894530 let latency = endTime - startTime
43904531 let path = requestPath req
4532+ let ip = requestClientIp req
4533+ let ipSuffix = if ip != "" then " " ++ ip else ""
43914534 if path != ""
4392− then logInfo ("[ACCESS] " ++ path ++ " " ++ show latency ++ "ms")
4393− else when (latency > slowRequestThreshold) (logInfo ("[SLOW] poll " ++ show latency ++ "ms"))
4535+ then logInfo ("[ACCESS] " ++ reqId ++ " " ++ path ++ " " ++ show latency ++ "ms" ++ ipSuffix)
4536+ else when (latency > slowRequestThreshold) (logInfo ("[SLOW] " ++ reqId ++ " poll " ++ show latency ++ "ms" ++ ipSuffix))
43944537
43954538-- ============================================
43964539-- Entry Point
@@ -4560,10 +4703,9 @@ backgroundWatchdog = do
45604703-- Startup connectivity probe: send an empty gossip to each known peer and log results.
45614704-- Surfaces DNS, firewall, and TLS issues before the first real gossip cycle.
45624705-- Probes up to 10 peers sequentially (not forked) to avoid overwhelming the network.
4563−startupProbePeers = \probed -> do
4706+startupProbePeers = do
45644707 knownPeers <- *knownPeerServers
4565− let remaining = do p <- knownPeers; yield p
4566− startupProbeNext remaining 0
4708+ startupProbeNext knownPeers 0
45674709
45684710startupProbeNext = \peers probed ->
45694711 if probed >= 10 then logInfo ("startup: probed " ++ show probed ++ " peers")
@@ -4670,5 +4812,5 @@ main = do
46704812 when federationEnabled (fork (do
46714813 sleep 5000 -- wait for listener to be ready
46724814 logInfo "startup: probing federation peers..."
4673− startupProbePeers 0))
4815+ startupProbePeers))
46744816 listen 8080 serve