@@ -165,7 +165,7 @@ federationLatencyWindowMs : Int<Ms>
165federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
165federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
167-- Snapshot format version (increment on schema changes to prevent corrupt restores)
167-- Snapshot format version (increment on schema changes to prevent corrupt restores)
168−snapshotVersion = 8 -- bumped for senderExpirations field in snapshot
168+snapshotVersion = 9 -- bumped for senderDeliveries field in snapshot
170-- Server build version: included in /health and /stats responses for deployment verification.
170-- Server build version: included in /health and /stats responses for deployment verification.
171-- Update this string when deploying a new build so operators can confirm which version is running.
171-- Update this string when deploying a new build so operators can confirm which version is running.
@@ -259,6 +259,10 @@ maxVerifyRateEntries = 1000
259-- Unacked message aging (messages pruned after this regardless of ack status)
259-- Unacked message aging (messages pruned after this regardless of ack status)
260unackedMessageMaxAge : Int<Ms>
260unackedMessageMaxAge : Int<Ms>
261unackedMessageMaxAge = 604800000 -- 7 days
261unackedMessageMaxAge = 604800000 -- 7 days
262+-- Minimum sender-specified message TTL (prevents abuse with ultra-short TTLs that
263+-- expire before the recipient can poll). 0 in the request = use unackedMessageMaxAge.
264+minMessageTtl : Int<Ms>
265+minMessageTtl = 60000 -- 1 minute
262-- Grace period before federation ack can delete the original message (allows local client to poll)
266-- Grace period before federation ack can delete the original message (allows local client to poll)
263ackGracePeriod : Int<Ms>
267ackGracePeriod : Int<Ms>
264ackGracePeriod = 120000 -- 2 minutes; allows clients on slow connections or backgrounded apps
268ackGracePeriod = 120000 -- 2 minutes; allows clients on slow connections or backgrounded apps
@@ -272,6 +276,10 @@ maxPendingAckCleanupsPerKey = 100
272-- can learn their message was never read (returned in poll response).
276-- can learn their message was never read (returned in poll response).
273maxSenderExpirations = 10000
277maxSenderExpirations = 10000
274maxSenderExpirationsPerKey = 100
278maxSenderExpirationsPerKey = 100
279+-- Delivery receipts: track when a locally-originated message is polled/acked by the recipient
280+-- so senders can learn their message was delivered (returned in poll response).
281+maxSenderDeliveries = 10000
282+maxSenderDeliveriesPerKey = 100
275-- Max pubkeys in a batch lookup request
283-- Max pubkeys in a batch lookup request
276maxBatchLookupKeys = 50
284maxBatchLookupKeys = 50
277-- Verify callback caching: cache successful verifications per-server to reduce
285-- Verify callback caching: cache successful verifications per-server to reduce
@@ -403,16 +411,24 @@ type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
403type UniqueKeyIpEntry = {ip: Text, keys: [PubkeyHex], windowStart: Timestamp}
411type UniqueKeyIpEntry = {ip: Text, keys: [PubkeyHex], windowStart: Timestamp}
405-- Forward nonce for extended replay protection (persisted in snapshots)
413-- Forward nonce for extended replay protection (persisted in snapshots)
414+-- originServer was removed in v9: nonce checks ignore originServer (a messageId+toKey
415+-- pair is globally unique), so storing it wasted memory. Pre-v9 snapshots with the
416+-- extra field are accepted during restore (Knot's JSON parser ignores unknown fields).
406−type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}
417+type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, seenAt: Timestamp}
408-- Batch send types
419-- Batch send types
409−type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId}
420+type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, expiresInMs: Int}
410type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
421type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
412-- Sender expiration notification: tracks when a locally-originated message expired unread.
423-- Sender expiration notification: tracks when a locally-originated message expired unread.
413-- Pruned after messageMaxAge. Included in the sender's next poll response.
424-- Pruned after messageMaxAge. Included in the sender's next poll response.
414type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}
425type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}
427+-- Sender delivery receipt: tracks when a locally-originated message was polled/acked by
428+-- the recipient. Included in the sender's next poll response for delivery confirmation.
429+-- Only for locally-originated messages (senderKey != ""); federated messages have no senderKey.
430+type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
416-- ============================================
432-- ============================================
417-- Persisted State
433-- Persisted State
418-- ============================================
434-- ============================================
@@ -768,6 +784,10 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
768-- Populated during prune for locally-originated messages (senderKey != "").
784-- Populated during prune for locally-originated messages (senderKey != "").
769*senderExpirations : [SenderExpiration]
785*senderExpirations : [SenderExpiration]
787+-- Sender delivery receipts: messages polled/acked by the recipient, queued for the sender's next poll.
788+-- Populated during ackPolledMessages and handleAckMessages for locally-originated messages.
789+*senderDeliveries : [SenderDelivery]
771-- Runtime maintenance secret: when non-empty, used instead of compile-time maintenanceSecret.
791-- Runtime maintenance secret: when non-empty, used instead of compile-time maintenanceSecret.
772-- Set via /maintenance/rotate-secret to enable zero-downtime secret rotation.
792-- Set via /maintenance/rotate-secret to enable zero-downtime secret rotation.
773-- Empty = use compile-time maintenanceSecret (default).
793-- Empty = use compile-time maintenanceSecret (default).
@@ -849,11 +869,13 @@ route AuthApi where
849route ClientApi where
869route ClientApi where
850 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
870 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
851 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
871 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
852− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}]}
872+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}]}
854 /messages
874 /messages
855 -- Send an E2E encrypted message to a recipient pubkey.
875 -- Send an E2E encrypted message to a recipient pubkey.
876+ -- expiresInMs: optional sender-specified TTL in milliseconds (0 = server default).
877+ -- Clamped to [minMessageTtl, unackedMessageMaxAge]. Allows ephemeral messages.
856− POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text, clientIp: Text}
878+ POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, expiresInMs: Int} / headers {authorization: Text, clientIp: Text}
857 -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int, error: Text}
879 -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int, error: Text}
858 = SendMessage
880 = SendMessage
859 -- Check delivery status of a sent message (caller must know recipient to prove ownership)
881 -- Check delivery status of a sent message (caller must know recipient to prove ownership)
@@ -948,12 +970,12 @@ route MaintenanceApi where
948 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
970 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
949 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
971 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
950 GET /maintenance/snapshot headers {authorization: Text}
972 GET /maintenance/snapshot headers {authorization: Text}
951− -> {ok: Bool, version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}], senderExpirations: [SenderExpiration]}
973+ -> {ok: Bool, version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}], senderExpirations: [SenderExpiration], senderDeliveries: [SenderDelivery]}
952 = Snapshot
974 = Snapshot
953 -- Restore: import a previously-exported snapshot. Server must be in drain mode
975 -- Restore: import a previously-exported snapshot. Server must be in drain mode
954 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
976 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
955 -- checksum: integrity check (0 = skip verification, for pre-v6 snapshots)
977 -- checksum: integrity check (0 = skip verification, for pre-v6 snapshots)
956− POST {version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}], senderExpirations: [SenderExpiration]} /maintenance/restore headers {authorization: Text}
978+ POST {version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}], senderExpirations: [SenderExpiration], senderDeliveries: [SenderDelivery]} /maintenance/restore headers {authorization: Text}
957 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
979 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
958 = Restore
980 = Restore
959 -- Block a federation peer (rejects all inbound/outbound federation with this server)
981 -- Block a federation peer (rejects all inbound/outbound federation with this server)
@@ -1041,7 +1063,7 @@ route HealthApi where
1041-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1063-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1042route StatsApi where
1064route StatsApi where
1043 GET /stats headers {authorization: Text}
1065 GET /stats headers {authorization: Text}
1044− -> {ok: Bool, version: Text, 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, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int}, 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, senderExpirations: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, prunesSinceSnapshot: Int, totalEntries: Int}}
1066+ -> {ok: Bool, version: Text, 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, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int}, 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, senderExpirations: Int, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, prunesSinceSnapshot: Int, totalEntries: Int}}
1046 GET /stats/peers headers {authorization: Text}
1068 GET /stats/peers headers {authorization: Text}
1047 -> {peers: [PeerStatus], ok: Bool}
1069 -> {peers: [PeerStatus], ok: Bool}
@@ -1078,7 +1100,19 @@ logError = \msg -> println ("[ERROR] " ++ msg)
1078-- Use for key operational events that monitoring pipelines need to parse
1100-- Use for key operational events that monitoring pipelines need to parse
1079-- (rate limit rejections, federation failures, capacity warnings).
1101-- (rate limit rejections, federation failures, capacity warnings).
1080-- Plain-text logInfo/logWarn/logError remain for human-readable context.
1102-- Plain-text logInfo/logWarn/logError remain for human-readable context.
1103+-- Values are escaped (quotes and backslashes) to produce valid JSON even
1104+-- when msg contains user-controlled data (e.g., error strings, server names).
1081−logEvent = \level event msg -> println ("{\"level\":\"" ++ level ++ "\",\"event\":\"" ++ event ++ "\",\"msg\":\"" ++ msg ++ "\"}")
1105+logEvent = \level event msg -> println ("{\"level\":\"" ++ escapeJsonString level ++ "\",\"event\":\"" ++ escapeJsonString event ++ "\",\"msg\":\"" ++ escapeJsonString msg ++ "\"}")
1107+-- Escape backslashes and double quotes in a string for safe JSON embedding.
1108+-- Processes one character at a time via tail recursion (TCO-safe in Knot).
1109+escapeJsonString : Text -> Text
1110+escapeJsonString = \s ->
1112+ else let c = take 1 s in
1113+ if c == "\\" then "\\\\" ++ escapeJsonString (drop 1 s)
1114+ else if c == "\"" then "\\\"" ++ escapeJsonString (drop 1 s)
1115+ else c ++ escapeJsonString (drop 1 s)
1083any : [a] -> (a -> Bool) -> Bool
1117any : [a] -> (a -> Bool) -> Bool
1084any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
1118any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
@@ -1135,19 +1169,29 @@ genToken = genTokenN tokenChunks
1135tooLong : Text -> Int -> Bool
1169tooLong : Text -> Int -> Bool
1136tooLong = \s maxLen -> drop maxLen s != ""
1170tooLong = \s maxLen -> drop maxLen s != ""
1138−-- Sanitize correlationId: truncate to maxCorrelationIdLen and strip control characters
1172+-- Sanitize correlationId: truncate to maxCorrelationIdLen and strip non-printable characters
1139−-- (newlines, carriage returns, tabs) to prevent log injection attacks.
1140−-- A malicious peer could embed "\n[SNAPSHOT_DATA] ..." to spoof snapshot lines
1173+-- to prevent log injection attacks. A malicious peer could embed "\n[SNAPSHOT_DATA] ..."
1141−-- parsed by sidecar processes, or inject fake [ERROR] entries.
1174+-- to spoof snapshot lines parsed by sidecar processes, inject fake [ERROR] entries, or
1175+-- use ANSI escape codes to manipulate terminal output. Only printable ASCII (0x20-0x7E)
1176+-- is kept; all other bytes (null, control chars, escape sequences, DEL, high bytes) are
1177+-- stripped. This is stricter than the original \n/\r/\t filter.
1142sanitizeCorrelationId : Text -> Text
1178sanitizeCorrelationId : Text -> Text
1143−sanitizeCorrelationId = \cid -> stripControlChars (take maxCorrelationIdLen cid)
1179+sanitizeCorrelationId = \cid -> stripNonPrintable (take maxCorrelationIdLen cid)
1145−stripControlChars : Text -> Text
1181+-- Printable ASCII characters (0x20 through 0x7E): space, symbols, digits, letters, punctuation.
1146−stripControlChars = \s ->
1182+-- Used as an allowlist by stripNonPrintable. Excludes DEL (0x7F) and all control chars (0x00-0x1F).
1183+printableAscii : Text
1184+printableAscii = " !\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~"
1186+-- Strip all non-printable characters from a string (allowlist approach).
1187+-- Keeps only characters in the printable ASCII range (0x20-0x7E).
1188+-- Tail-recursive: Knot compiles to a loop via TCO.
1189+stripNonPrintable : Text -> Text
1190+stripNonPrintable = \s ->
1147 if s == "" then ""
1191 if s == "" then ""
1148 else let c = take 1 s in
1192 else let c = take 1 s in
1149− if c == "\n" || c == "\r" || c == "\t" then stripControlChars (drop 1 s)
1193+ if contains c printableAscii then c ++ stripNonPrintable (drop 1 s)
1150− else c ++ stripControlChars (drop 1 s)
1194+ else stripNonPrintable (drop 1 s)
1152isHexChar : Text -> Bool
1196isHexChar : Text -> Bool
1153isHexChar = \c -> contains c "0123456789abcdef"
1197isHexChar = \c -> contains c "0123456789abcdef"
@@ -2701,6 +2745,13 @@ clampTtl = clamp 1 maxPresenceTtl
2701clampTimestamp : Timestamp -> Timestamp -> Timestamp
2745clampTimestamp : Timestamp -> Timestamp -> Timestamp
2702clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
2746clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
2748+-- Compute effective message TTL: 0 = server default, otherwise clamp to [minMessageTtl, unackedMessageMaxAge].
2749+-- Returns the TTL in milliseconds to add to receivedAt for expiresAt.
2750+effectiveMessageTtl : Int<Ms> -> Int<Ms>
2751+effectiveMessageTtl = \expiresInMs ->
2752+ if expiresInMs <= 0 then unackedMessageMaxAge
2753+ else clamp minMessageTtl unackedMessageMaxAge expiresInMs
2704-- Ensure strictly monotonic timestamps for message ordering
2755-- Ensure strictly monotonic timestamps for message ordering
2705nextSeqTs : Timestamp -> Timestamp -> Timestamp
2756nextSeqTs : Timestamp -> Timestamp -> Timestamp
2706nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
2757nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
@@ -3167,6 +3218,9 @@ pruneExpired = \t -> do
3167 -- still receive expiration notifications on their next poll.
3218 -- still receive expiration notifications on their next poll.
3168 senderExp <- *senderExpirations
3219 senderExp <- *senderExpirations
3169 set *senderExpirations = do e <- senderExp; where t - e.expiredAt < unackedMessageMaxAge; yield e
3220 set *senderExpirations = do e <- senderExp; where t - e.expiredAt < unackedMessageMaxAge; yield e
3221+ -- Prune old sender delivery receipts (same retention as expirations)
3222+ senderDel <- *senderDeliveries
3223+ set *senderDeliveries = do d <- senderDel; where t - d.deliveredAt < unackedMessageMaxAge; yield d
3170 -- Reconcile global federation connection counter with actual in-flight state
3224 -- Reconcile global federation connection counter with actual in-flight state
3171 -- (self-heals drift from leaked slots where a forked task died without decrementing)
3225 -- (self-heals drift from leaked slots where a forked task died without decrementing)
3172 inFlightRec <- *federationInFlight
3226 inFlightRec <- *federationInFlight
@@ -4116,6 +4170,15 @@ ackPolledMessages = \pubkey cursor t -> do
4116 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
4170 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
4117 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
4171 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
4118 writeAckedBucket b eviction.kept
4172 writeAckedBucket b eviction.kept
4173+ -- Queue delivery receipts for locally-originated messages (senderKey != "")
4174+ -- so the sender learns their message was delivered on their next poll.
4175+ let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
4176+ when (count localDelivered > 0) (do
4177+ sd <- *senderDeliveries
4178+ let combined = union sd localDelivered
4179+ set *senderDeliveries = if count combined > maxSenderDeliveries
4180+ then evictOneOldest combined (\d -> d.deliveredAt)
4119 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
4182 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
4120 else yield {pendingAcks: [], evictedEntries: []}
4183 else yield {pendingAcks: [], evictedEntries: []}
@@ -4556,10 +4619,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
4556 let cursorWasReset = generationMismatch && cursor > 0
4619 let cursorWasReset = generationMismatch && cursor > 0
4557 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
4620 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
4558 dp <- *drainPhase
4621 dp <- *drainPhase
4559− if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, messageExpirations: []})
4622+ if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, messageExpirations: [], messageDeliveries: []})
4561 globalRateLimited <- checkGlobalRate t
4624 globalRateLimited <- checkGlobalRate t
4562− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, messageExpirations: []})
4625+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, messageExpirations: [], messageDeliveries: []})
4564 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
4627 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
4565 startPresenceVersion <- *presenceVersion
4628 startPresenceVersion <- *presenceVersion
@@ -4598,14 +4661,24 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
4598 set *senderExpirations = remaining
4661 set *senderExpirations = remaining
4599 let capped = takeRelation maxSenderExpirationsPerKey forKey
4662 let capped = takeRelation maxSenderExpirationsPerKey forKey
4600 yield capped
4663 yield capped
4664+ -- Deliver message delivery receipts: atomically read and drain delivery notifications
4665+ -- for this sender's pubkey so each receipt is delivered exactly once.
4666+ deliveries <- if not result.authorized || result.pubkey == "" then yield []
4668+ sd <- *senderDeliveries
4669+ let forKey = do d <- sd; where d.senderKey == result.pubkey; yield {messageId: d.messageId, toKey: d.toKey, deliveredAt: d.deliveredAt}
4670+ let remaining = do d <- sd; where d.senderKey != result.pubkey; yield d
4671+ set *senderDeliveries = remaining
4672+ let capped = takeRelation maxSenderDeliveriesPerKey forKey
4601 -- Poll slot exhaustion: signal clients to back off for one poll cycle
4674 -- Poll slot exhaustion: signal clients to back off for one poll cycle
4602 let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
4675 let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
4603− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, messageExpirations: expirations})
4676+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, messageExpirations: expirations, messageDeliveries: deliveries})
4605-- Core send logic: dedup, rate limiting, store message, create forwards.
4678-- Core send logic: dedup, rate limiting, store message, create forwards.
4606-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
4679-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
4607-- Caller handles auth, drain checks, and messageId generation.
4680-- Caller handles auth, drain checks, and messageId generation.
4608−sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
4681+sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs t ->
4609 if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4682 if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4611 let msgBucket = pubkeyBucket to
4684 let msgBucket = pubkeyBucket to
@@ -4648,7 +4721,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
4648 seqCounter <- *seqCounter
4721 seqCounter <- *seqCounter
4649 let seqTs = nextSeqTs t seqCounter
4722 let seqTs = nextSeqTs t seqCounter
4650 set *seqCounter = seqTs
4723 set *seqCounter = seqTs
4651− let msgExpiresAt = seqTs + unackedMessageMaxAge
4724+ let msgExpiresAt = seqTs + effectiveMessageTtl expiresInMs
4652 writeMessageBucket msgBucket (union bucket
4725 writeMessageBucket msgBucket (union bucket
4653 [{messageId: effectiveMsgId, toKey: to,
4726 [{messageId: effectiveMsgId, toKey: to,
4654 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt, senderKey: senderKey}])
4727 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt, senderKey: senderKey}])
@@ -4662,7 +4735,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
4662 -- Sharded: writes only to the bucket for this recipient's pubkey.
4735 -- Sharded: writes only to the bucket for this recipient's pubkey.
4663 let nb = pubkeyBucket to
4736 let nb = pubkeyBucket to
4664 fwdNonces <- readNonceBucket nb
4737 fwdNonces <- readNonceBucket nb
4665− let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, originServer: "", seenAt: t}]
4738+ let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, seenAt: t}]
4666 writeNonceBucket nb (if count updatedNonces > maxNoncesPerBucket
4739 writeNonceBucket nb (if count updatedNonces > maxNoncesPerBucket
4667 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
4740 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
4668 else updatedNonces)
4741 else updatedNonces)
@@ -4750,7 +4823,7 @@ sendBatchForwardsSequential = \msgs server ->
4750-- POST /messages: Send an encrypted message to a recipient.
4823-- POST /messages: Send an encrypted message to a recipient.
4751-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
4824-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
4752-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
4825-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
4753−handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp respond -> do
4826+handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs authorization clientIp respond -> do
4754 t <- now
4827 t <- now
4755 globalRateLimited <- checkGlobalRate t
4828 globalRateLimited <- checkGlobalRate t
4756 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0, error: "rate_limited"})
4829 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0, error: "rate_limited"})
@@ -4769,7 +4842,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
4769 if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
4842 if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
4770 else withSessionAuth authorization t clientIp
4843 else withSessionAuth authorization t clientIp
4771 (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
4844 (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
4772− (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId t)
4845+ (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs t)
4773 case result.status of
4846 case result.status of
4774 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
4847 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
4775 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))
4848 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))
@@ -5005,9 +5078,14 @@ handleVerifyToken = \token scope clientIp respond -> do
5005 -- The prefix filter is not constant-time, but single-use tokens with 24 bytes of randomness
5078 -- The prefix filter is not constant-time, but single-use tokens with 24 bytes of randomness
5006 -- make first-char leakage irrelevant (16^47 remaining possibilities in 30s window).
5079 -- make first-char leakage irrelevant (16^47 remaining possibilities in 30s window).
5007 let firstChar = take 1 token
5080 let firstChar = take 1 token
5008− let candidates = do p <- pending; where p.scope == scope && t - p.createdAt < verificationTokenExpiry && take 1 p.token == firstChar; yield p
5081+ let isCandidate = \p -> p.scope == scope && t - p.createdAt < verificationTokenExpiry && take 1 p.token == firstChar
5082+ let candidates = do p <- pending; where isCandidate p; yield p
5009 let match = any candidates (\p -> safeCompare p.token token)
5083 let match = any candidates (\p -> safeCompare p.token token)
5084+ -- Optimized removal: run safeCompare only on candidates (O(n/16)) instead of
5085+ -- all pending (O(n)). Non-candidates are kept without constant-time comparison.
5010− when match (set *pendingVerifications = do p <- pending; where not (safeCompare p.token token); yield p)
5086+ when match (set *pendingVerifications = union
5087+ (do p <- pending; where not (isCandidate p); yield p)
5088+ (do p <- candidates; where not (safeCompare p.token token); yield p))
5011 yield match
5089 yield match
5012 -- Omit scope from log to prevent scope enumeration by attackers probing the endpoint
5090 -- Omit scope from log to prevent scope enumeration by attackers probing the endpoint
5013 when (not found) (do
5091 when (not found) (do
@@ -5227,7 +5305,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
5227 -- Record forward nonce for replay protection beyond message/ack dedup lifetime.
5305 -- Record forward nonce for replay protection beyond message/ack dedup lifetime.
5228 -- Sharded: writes only to the bucket for this recipient's pubkey.
5306 -- Sharded: writes only to the bucket for this recipient's pubkey.
5229 fwdNonces <- readNonceBucket bkt
5307 fwdNonces <- readNonceBucket bkt
5230− let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
5308+ let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, seenAt: t}]
5231 writeNonceBucket bkt (if count updatedNonces > maxNoncesPerBucket
5309 writeNonceBucket bkt (if count updatedNonces > maxNoncesPerBucket
5232 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5310 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5233 else updatedNonces)
5311 else updatedNonces)
@@ -5401,8 +5479,9 @@ handleDrain = \phase authorization respond -> do
5401 ap <- *allowedPeers
5479 ap <- *allowedPeers
5402 pac <- *pendingAckCleanups
5480 pac <- *pendingAckCleanups
5403 se <- *senderExpirations
5481 se <- *senderExpirations
5482+ sdl <- *senderDeliveries
5404 let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
5483 let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
5405− yield {version: snapshotVersion, checksum: drainCs, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se}
5484+ yield {version: snapshotVersion, checksum: drainCs, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se, senderDeliveries: sdl}
5406 -- Sentinel-wrapped: sidecar processes should verify [SNAPSHOT_END] follows to detect truncation
5485 -- Sentinel-wrapped: sidecar processes should verify [SNAPSHOT_END] follows to detect truncation
5407 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
5486 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
5408 println "[SNAPSHOT_END]"
5487 println "[SNAPSHOT_END]"
@@ -5614,7 +5693,7 @@ handleHealth = \authorization respond -> do
5614 watchdogHealthy: True {},
5693 watchdogHealthy: True {},
5615 draining: result.isDraining})
5694 draining: result.isDraining})
5617−zeroStats = {ok: False {}, version: serverBuildVersion, 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, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 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, senderExpirations: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, prunesSinceSnapshot: 0, totalEntries: 0}}
5696+zeroStats = {ok: False {}, version: serverBuildVersion, 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, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 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, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, prunesSinceSnapshot: 0, totalEntries: 0}}
5619handleStats = \authorization respond -> do
5698handleStats = \authorization respond -> do
5620 t <- now
5699 t <- now
@@ -5651,7 +5730,8 @@ handleStats = \authorization respond -> do
5651 vir <- *verifyIpRates
5730 vir <- *verifyIpRates
5652 ukir <- *uniqueKeyIpRates
5731 ukir <- *uniqueKeyIpRates
5653 se <- *senderExpirations
5732 se <- *senderExpirations
5733+ sdl <- *senderDeliveries
5654− yield {sc: sc, pc: pc, c: c, aCount: aCount, ao: ao, dCount: dCount, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg, fnCount: fnCount, pac: pac, bp: bp, ger: ger, vir: vir, ukir: ukir, se: se}
5734+ yield {sc: sc, pc: pc, c: c, aCount: aCount, ao: ao, dCount: dCount, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg, fnCount: fnCount, pac: pac, bp: bp, ger: ger, vir: vir, ukir: ukir, se: se, sdl: sdl}
5655 counters <- atomic do
5735 counters <- atomic do
5656 sr <- *sendRates
5736 sr <- *sendRates
5657 cr <- *challengeRates
5737 cr <- *challengeRates
@@ -5703,7 +5783,7 @@ handleStats = \authorization respond -> do
5703 yield (respond {ok: True {}, version: serverBuildVersion,
5783 yield (respond {ok: True {}, version: serverBuildVersion,
5704 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: core.aCount, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: core.dCount, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
5784 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: core.aCount, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: core.dCount, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
5705 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, slowVerifies: counters.svc, verifyFailures: counters.vfc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca, forwardServers: fwdServerCount, maxForwardQueueDepth: maxFwdDepth},
5785 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, slowVerifies: counters.svc, verifyFailures: counters.vfc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca, forwardServers: fwdServerCount, maxForwardQueueDepth: maxFwdDepth},
5706− 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, senderExpirations: count core.se, rejectedGlobal: counters.rlg, rejectedIp: counters.rli, rejectedAuth: counters.rla, rejectedPerKey: counters.rlk, rejectedFederation: counters.rlf},
5786+ 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, senderExpirations: count core.se, senderDeliveries: count core.sdl, rejectedGlobal: counters.rlg, rejectedIp: counters.rli, rejectedAuth: counters.rla, rejectedPerKey: counters.rlk, rejectedFederation: counters.rlf},
5707 timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0, drainPhase: counters.dr, prunesSinceSnapshot: counters.pss, totalEntries: sessions + messages + presence + fwds + count core.c + core.aCount + count core.ao + core.dCount + core.fnCount + count core.ap + count core.fi}})
5787 timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0, drainPhase: counters.dr, prunesSinceSnapshot: counters.pss, totalEntries: sessions + messages + presence + fwds + count core.c + core.aCount + count core.ao + core.dCount + core.fnCount + count core.ap + count core.fi}})
5708 else yield (respond zeroStats)
5788 else yield (respond zeroStats)
@@ -5789,7 +5869,7 @@ handleReady = \respond -> do
5789 let isReady = lastPrune > 0 && lastPrune >= sa
5869 let isReady = lastPrune > 0 && lastPrune >= sa
5790 yield (respond {ready: isReady})
5870 yield (respond {ready: isReady})
5792−emptySnapshot = {ok: False {}, version: 0, checksum: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: [], allowedPeers: [], pendingAckCleanups: [], senderExpirations: []}
5872+emptySnapshot = {ok: False {}, version: 0, checksum: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: [], allowedPeers: [], pendingAckCleanups: [], senderExpirations: [], senderDeliveries: []}
5794-- GET /maintenance/snapshot: Export federation metadata for persistence.
5874-- GET /maintenance/snapshot: Export federation metadata for persistence.
5795-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
5875-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
@@ -5812,27 +5892,28 @@ handleSnapshot = \authorization respond -> do
5812 ap <- *allowedPeers
5892 ap <- *allowedPeers
5813 pac <- *pendingAckCleanups
5893 pac <- *pendingAckCleanups
5814 se <- *senderExpirations
5894 se <- *senderExpirations
5895+ sdl <- *senderDeliveries
5815− yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se}
5896+ yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se, senderDeliveries: sdl}
5816 atomic do
5897 atomic do
5817 set *lastSnapshotExportAt = t
5898 set *lastSnapshotExportAt = t
5818 set *prunesSinceSnapshot = 0
5899 set *prunesSinceSnapshot = 0
5819 let cs = snapshotChecksum (count result.presence) (count result.ackedIds) (count result.deliveredIds) (count result.forwards) (count result.forwardNonces)
5900 let cs = snapshotChecksum (count result.presence) (count result.ackedIds) (count result.deliveredIds) (count result.forwards) (count result.forwardNonces)
5820− logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards, " ++ show (count result.forwardNonces) ++ " forwardNonces, " ++ show (count result.blockedPeers) ++ " blockedPeers, " ++ show (count result.allowedPeers) ++ " allowedPeers, " ++ show (count result.pendingAckCleanups) ++ " pendingAckCleanups, " ++ show (count result.senderExpirations) ++ " senderExpirations, checksum=" ++ show cs)
5901+ logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards, " ++ show (count result.forwardNonces) ++ " forwardNonces, " ++ show (count result.blockedPeers) ++ " blockedPeers, " ++ show (count result.allowedPeers) ++ " allowedPeers, " ++ show (count result.pendingAckCleanups) ++ " pendingAckCleanups, " ++ show (count result.senderExpirations) ++ " senderExpirations, " ++ show (count result.senderDeliveries) ++ " senderDeliveries, checksum=" ++ show cs)
5821− yield (respond {ok: True {}, version: snapshotVersion, checksum: cs, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces, blockedPeers: result.blockedPeers, allowedPeers: result.allowedPeers, pendingAckCleanups: result.pendingAckCleanups, senderExpirations: result.senderExpirations})
5902+ yield (respond {ok: True {}, version: snapshotVersion, checksum: cs, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces, blockedPeers: result.blockedPeers, allowedPeers: result.allowedPeers, pendingAckCleanups: result.pendingAckCleanups, senderExpirations: result.senderExpirations, senderDeliveries: result.senderDeliveries})
5823-- POST /maintenance/restore: Import a previously-exported snapshot.
5904-- POST /maintenance/restore: Import a previously-exported snapshot.
5824-- Requires drain mode to prevent conflicts with live state. Merges imported data
5905-- Requires drain mode to prevent conflicts with live state. Merges imported data
5825-- with existing state (dedup entries are unioned, presence is merged by freshness).
5906-- with existing state (dedup entries are unioned, presence is merged by freshness).
5826-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
5907-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
5827-- trigger federation forward creation from messages + restored presence.
5908-- trigger federation forward creation from messages + restored presence.
5828−handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData blockedPeersData allowedPeersData pendingAckCleanupsData senderExpirationsData authorization respond -> do
5909+handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData blockedPeersData allowedPeersData pendingAckCleanupsData senderExpirationsData senderDeliveriesData authorization respond -> do
5829 authorized <- verifyMaintenanceAuth authorization
5910 authorized <- verifyMaintenanceAuth authorization
5830 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
5911 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
5832 -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers,
5913 -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers,
5833 -- version 3 = pre-ForwardEntry changes, version 4 = pre-allowedPeers, version 5 = pre-checksum
5914 -- version 3 = pre-ForwardEntry changes, version 4 = pre-allowedPeers, version 5 = pre-checksum
5834 -- (accept all for backwards compatibility; missing fields default to empty list or 0)
5915 -- (accept all for backwards compatibility; missing fields default to empty list or 0)
5835− if version != 0 && version != 1 && version != 2 && version != 3 && version != 4 && version != 5 && version != 6 && version != 7 && version != snapshotVersion then do
5916+ if version != 0 && version != 1 && version != 2 && version != 3 && version != 4 && version != 5 && version != 6 && version != 7 && version != 8 && version != snapshotVersion then do
5836 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
5917 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
5837 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
5918 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
@@ -5897,10 +5978,10 @@ handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData de
5897 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
5978 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
5898 writeAllForwards cappedForwards
5979 writeAllForwards cappedForwards
5899 set *totalForwardCount = count cappedForwards
5980 set *totalForwardCount = count cappedForwards
5900− -- Restore forward nonces (dedup by messageId + toKey + originServer, keep fresher on conflict)
5981+ -- Restore forward nonces (dedup by messageId + toKey, keep fresher on conflict)
5901 existingNonces <- readAllNonces
5982 existingNonces <- readAllNonces
5902 let validNonces = do n <- forwardNoncesData; where n.seenAt <= t && t - n.seenAt < forwardNonceRetention; yield n
5983 let validNonces = do n <- forwardNoncesData; where n.seenAt <= t && t - n.seenAt < forwardNonceRetention; yield n
5903− let mergedNonces = mergeByKey (\a b -> a.messageId == b.messageId && a.toKey == b.toKey && a.originServer == b.originServer) (\x -> x.seenAt) existingNonces validNonces
5984+ let mergedNonces = mergeByKey (\a b -> a.messageId == b.messageId && a.toKey == b.toKey) (\x -> x.seenAt) existingNonces validNonces
5904 let cappedNonces = takeRelation maxForwardNonces mergedNonces
5985 let cappedNonces = takeRelation maxForwardNonces mergedNonces
5905 writeAllNonces cappedNonces
5986 writeAllNonces cappedNonces
5906 -- Restore blocked peers (union with existing, cap at maxBlockedPeers)
5987 -- Restore blocked peers (union with existing, cap at maxBlockedPeers)
@@ -5923,6 +6004,11 @@ handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData de
5923 existingSenderExp <- *senderExpirations
6004 existingSenderExp <- *senderExpirations
5924 let mergedSenderExp = union existingSenderExp senderExpirationsData
6005 let mergedSenderExp = union existingSenderExp senderExpirationsData
5925 set *senderExpirations = takeRelation maxSenderExpirations mergedSenderExp
6006 set *senderExpirations = takeRelation maxSenderExpirations mergedSenderExp
6007+ -- Restore sender delivery receipts (union with existing, cap at maxSenderDeliveries)
6008+ -- pre-v9 snapshots have empty senderDeliveries (field defaults to [])
6009+ existingSenderDel <- *senderDeliveries
6010+ let mergedSenderDel = union existingSenderDel senderDeliveriesData
6011+ set *senderDeliveries = takeRelation maxSenderDeliveries mergedSenderDel
5926 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
6012 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
5927 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
6013 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
5928 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
6014 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
@@ -5971,6 +6057,14 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
5971 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
6057 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
5972 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
6058 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
5973 writeAckedBucket b eviction.kept
6059 writeAckedBucket b eviction.kept
6060+ -- Queue delivery receipts for locally-originated messages
6061+ let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
6062+ when (count localDelivered > 0) (do
6063+ sd <- *senderDeliveries
6064+ let combined = union sd localDelivered
6065+ set *senderDeliveries = if count combined > maxSenderDeliveries
6066+ then evictOneOldest combined (\d -> d.deliveredAt)
5974 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
6068 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
5975 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}
6069 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}
5976 fork (when federationEnabled (do
6070 fork (when federationEnabled (do
@@ -6187,34 +6281,73 @@ handleSendBatch = \messages authorization clientIp respond -> do
6187 let batchRetryMs = fold (\acc r -> max acc r.retryAfterMs) 0 results
6281 let batchRetryMs = fold (\acc r -> max acc r.retryAfterMs) 0 results
6188 yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}, retryAfterMs: batchRetryMs}))
6282 yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}, retryAfterMs: batchRetryMs}))
6190−-- TODO: If Knot adds a map/dictionary type, group batch messages by pubkeyBucket
6284+-- Batch messages are sorted by pubkeyBucket(to) by the caller, so consecutive messages
6191−-- and process each group in a single atomic block per bucket. Currently each message
6285+-- sharing a bucket are grouped into a single atomic block. This reduces bucket reads
6192−-- gets its own atomic block via sendMessageCore, re-reading the same bucket when
6286+-- from O(batchSize) to O(uniqueBuckets) and avoids per-message transaction overhead.
6193−-- multiple messages target the same recipient. Grouping without maps is O(n^2).
6287+-- Messages requiring genToken (empty messageId) are pre-processed outside the atomic block.
6194sendBatchInner = \senderKey messages t acc ->
6288sendBatchInner = \senderKey messages t acc ->
6195− let split = splitAtN messages 1 in
6289+ case headOf messages of
6290+ Nothing {} -> yield acc
6291+ Just {value: first} -> do
6292+ let bucket = pubkeyBucket first.to
6293+ -- Collect consecutive messages targeting the same bucket
6294+ let sameR = fold (\a m -> if pubkeyBucket m.to == bucket then {same: union a.same [m], rest: a.rest} else {same: a.same, rest: union a.rest [m]}) {same: [], rest: []} messages
6295+ -- Pre-generate message IDs and validate timestamps (IO, outside atomic block)
6296+ prepared <- prepareBatchMessages sameR.same t []
6297+ -- Process the entire bucket group in a single atomic block
6298+ results <- atomic do
6300+ sendBatchBucketGroup senderKey prepared t dp []
6301+ -- Log warnings for rate-limited/rejected messages outside the atomic block
6302+ logBatchResults results
6303+ sendBatchInner senderKey sameR.rest t (union acc results)
6305+-- Pre-generate message IDs and validate timestamps for a group of messages.
6306+-- IO actions (genToken, logWarn) can't run inside atomic, so this runs outside.
6307+prepareBatchMessages = \messages t acc ->
6196− case headOf split.batch of
6308+ case headOf messages of
6197 Nothing {} -> yield acc
6309 Nothing {} -> yield acc
6198 Just {value: msg} -> do
6310 Just {value: msg} -> do
6311+ let rest = do m <- messages; where m != msg; yield m
6199 effectiveMsgId <- if msg.messageId != "" then yield msg.messageId else genToken
6312 effectiveMsgId <- if msg.messageId != "" then yield msg.messageId else genToken
6200 when (msg.timestamp > t + clockSkewTolerance)
6313 when (msg.timestamp > t + clockSkewTolerance)
6201 (logWarn ("sendBatch: future timestamp detected (+" ++ show (msg.timestamp - t) ++ "ms ahead) for " ++ effectiveMsgId))
6314 (logWarn ("sendBatch: future timestamp detected (+" ++ show (msg.timestamp - t) ++ "ms ahead) for " ++ effectiveMsgId))
6202 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
6315 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
6203 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
6316 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
6204− result <- if t - msg.timestamp > messageMaxAge
6317+ let entry = if t - msg.timestamp > messageMaxAge
6318+ then {msg: msg, effectiveMsgId: effectiveMsgId, stale: True {}}
6319+ else {msg: msg, effectiveMsgId: effectiveMsgId, stale: False {}}
6320+ prepareBatchMessages rest t (union acc [entry])
6322+-- Process all prepared messages for one bucket in a single atomic block.
6323+-- Each call to sendMessageCore sees the previous call's writes (updated counters,
6324+-- new messages), avoiding redundant bucket reads.
6325+sendBatchBucketGroup = \senderKey prepared t dp acc ->
6326+ case headOf prepared of
6327+ Nothing {} -> yield acc
6328+ Just {value: entry} -> do
6329+ let rest = do p <- prepared; where p != entry; yield p
6330+ result <- if entry.stale
6205− then yield {messageId: effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, retryAfterMs: 0}
6331+ then yield {messageId: entry.effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, retryAfterMs: 0}
6207− let clampedTs = clampTimestamp msg.timestamp t
6333+ let clampedTs = clampTimestamp entry.msg.timestamp t
6208− sendResult <- atomic do
6210− if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
6334+ if dp >= 2 then yield {messageId: entry.effectiveMsgId, status: MsDraining {}, remoteServers: [], toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: 0}
6211− else sendMessageCore senderKey msg.to msg.encryptedBlob clampedTs effectiveMsgId t
6336+ sendResult <- sendMessageCore senderKey entry.msg.to entry.msg.encryptedBlob clampedTs entry.effectiveMsgId entry.msg.expiresInMs t
6337+ yield {messageId: entry.effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryAfterMs}
6338+ sendBatchBucketGroup senderKey rest t dp (union acc [result])
6340+-- Log warnings for batch results outside the atomic block
6341+logBatchResults = \results ->
6342+ case headOf results of
6343+ Nothing {} -> yield {}
6344+ Just {value: r} -> do
6345+ let rest = do x <- results; where x != r; yield x
6212− case sendResult.status of
6346+ case r.status of
6213− MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
6347+ MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ r.messageId)
6214− MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")
6348+ MsRejected {} -> logWarn ("sendBatch: rejected for " ++ r.messageId ++ " (capacity or mailbox full)")
6215− _ -> yield {}
6349+ _ -> yield {}
6216− yield {messageId: effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryAfterMs}
6350+ logBatchResults rest
6217− sendBatchInner senderKey split.rest t (union acc [result])
6219-- POST /maintenance/block: Add a server to the federation blocklist.
6352-- POST /maintenance/block: Add a server to the federation blocklist.
6220-- Blocked peers are rejected from all inbound and outbound federation operations.
6353-- Blocked peers are rejected from all inbound and outbound federation operations.
@@ -6689,7 +6822,7 @@ serveInner = \req -> case req of
6689 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
6822 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
6690 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
6823 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
6691 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond
6824 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond
6692− SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond
6825+ SendMessage {to, encryptedBlob, timestamp, messageId, expiresInMs, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId expiresInMs authorization clientIp respond
6693 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
6826 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
6694 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
6827 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
6695 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
6828 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
@@ -6708,7 +6841,7 @@ serveInner = \req -> case req of
6708 Shutdown {authorization, respond} -> handleShutdown authorization respond
6841 Shutdown {authorization, respond} -> handleShutdown authorization respond
6709 Undrain {authorization, respond} -> handleUndrain authorization respond
6842 Undrain {authorization, respond} -> handleUndrain authorization respond
6710 Snapshot {authorization, respond} -> handleSnapshot authorization respond
6843 Snapshot {authorization, respond} -> handleSnapshot authorization respond
6711− Restore {version, checksum, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, blockedPeers, allowedPeers, pendingAckCleanups, senderExpirations, authorization, respond} -> handleRestore version checksum presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces blockedPeers allowedPeers pendingAckCleanups senderExpirations authorization respond
6844+ Restore {version, checksum, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, blockedPeers, allowedPeers, pendingAckCleanups, senderExpirations, senderDeliveries, authorization, respond} -> handleRestore version checksum presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces blockedPeers allowedPeers pendingAckCleanups senderExpirations senderDeliveries authorization respond
6712 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
6845 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
6713 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
6846 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
6714 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
6847 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
@@ -6924,8 +7057,9 @@ backgroundSnapshot = do
6924 ap <- *allowedPeers
7057 ap <- *allowedPeers
6925 pac <- *pendingAckCleanups
7058 pac <- *pendingAckCleanups
6926 se <- *senderExpirations
7059 se <- *senderExpirations
7060+ sdl <- *senderDeliveries
6927 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
7061 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
6928− yield {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se}
7062+ yield {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se, senderDeliveries: sdl}
6929 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
7063 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
6930 println "[SNAPSHOT_END]"
7064 println "[SNAPSHOT_END]"
6931 atomic do
7065 atomic do
@@ -7031,6 +7165,7 @@ main = do
7031 && retryBackoffBase <= retryBackoffMax
7165 && retryBackoffBase <= retryBackoffMax
7032 && backgroundPruneInterval < healthPruneMaxAge
7166 && backgroundPruneInterval < healthPruneMaxAge
7033 && ackGracePeriod < unackedMessageMaxAge
7167 && ackGracePeriod < unackedMessageMaxAge
7168+ && minMessageTtl > 0 && minMessageTtl <= unackedMessageMaxAge
7034 && clockSkewTolerance < challengeExpiry
7169 && clockSkewTolerance < challengeExpiry
7035 && maxMessagesPerRecipient <= maxMessages
7170 && maxMessagesPerRecipient <= maxMessages
7036 && maxForwardsPerKey <= maxForwardsEntries
7171 && maxForwardsPerKey <= maxForwardsEntries