@@ -177,17 +177,17 @@ federationLatencyWindowMs : Int<Ms>
177federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
177federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
179-- Snapshot format version (increment on schema changes to prevent corrupt restores)
179-- Snapshot format version (increment on schema changes to prevent corrupt restores)
180−snapshotVersion = 11 -- bumped for Message.priority and ForwardEntry.priority fields
180+snapshotVersion = 12 -- bumped for ForwardEntry.senderKey and federation forward priority
182-- Server build version: included in /health and /stats responses for deployment verification.
182-- Server build version: included in /health and /stats responses for deployment verification.
183-- Update this string when deploying a new build so operators can confirm which version is running.
183-- Update this string when deploying a new build so operators can confirm which version is running.
184−serverBuildVersion = "2026-04-26a"
184+serverBuildVersion = "2026-04-26b"
186-- Federation protocol version: included in gossip/forward/ack requests.
186-- Federation protocol version: included in gossip/forward/ack requests.
187-- Peers log warnings when they receive requests from older versions.
187-- Peers log warnings when they receive requests from older versions.
188-- Backward compatible: older peers ignore unknown fields (Knot's JSON parser
188-- Backward compatible: older peers ignore unknown fields (Knot's JSON parser
189-- drops unknown keys). 0 = pre-versioning peer (field missing or unset).
189-- drops unknown keys). 0 = pre-versioning peer (field missing or unset).
190−federationProtocolVersion = 11
190+federationProtocolVersion = 12
192-- Snapshot checksum: deterministic hash of entry counts for integrity verification.
192-- Snapshot checksum: deterministic hash of entry counts for integrity verification.
193-- Uses multiplicative hash to distinguish permutations of the same counts
193-- Uses multiplicative hash to distinguish permutations of the same counts
@@ -419,7 +419,10 @@ type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHe
419-- Presence: tracks that a pubkey is online at a given server, with expiry
419-- Presence: tracks that a pubkey is online at a given server, with expiry
420type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
420type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
421-- Queued federation forward: message waiting to be delivered to a remote server.
421-- Queued federation forward: message waiting to be delivered to a remote server.
422+-- senderKey: pubkey of the authenticated sender (locally-originated only; "" for federation relays).
423+-- Used by fedDeliveryReceipts to generate delivery receipts even if the original message was pruned.
424+-- Pre-v12 snapshots without senderKey default to "" (Knot ignores missing Text fields).
422−type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int, expiresAt: Timestamp, priority: Int}
425+type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int, expiresAt: Timestamp, priority: Int, senderKey: Text}
423-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
426-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
424type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}
427type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}
425-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
428-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
@@ -1003,7 +1006,7 @@ route FederationApi where
1003 -- privacy model (the server never learns who sent a message). /messages/cancel for
1006 -- privacy model (the server never learns who sent a message). /messages/cancel for
1004 -- federated messages requires the client to include sender identity in the encrypted
1007 -- federated messages requires the client to include sender identity in the encrypted
1005 -- payload; the server cannot verify sender identity for federation-forwarded messages.
1008 -- payload; the server cannot verify sender identity for federation-forwarded messages.
1006− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, expiresAt: Timestamp, protocolVersion: Int} /forward headers {authorization: Text}
1009+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, expiresAt: Timestamp, priority: Int, protocolVersion: Int} /forward headers {authorization: Text}
1007 -> {ok: Bool, error: Text}
1010 -> {ok: Bool, error: Text}
1008 = ForwardMessage
1011 = ForwardMessage
1009 -- Acknowledge delivery of messages back to the origin server
1012 -- Acknowledge delivery of messages back to the origin server
@@ -1023,7 +1026,7 @@ route FederationApi where
1023 -- Batch forward: forward multiple messages in one request (reduces HTTP round-trips).
1026 -- Batch forward: forward multiple messages in one request (reduces HTTP round-trips).
1024 -- Each message is processed independently; partial success is possible.
1027 -- Each message is processed independently; partial success is possible.
1025 -- Results are returned per-message so the sender knows which forwards succeeded.
1028 -- Results are returned per-message so the sender knows which forwards succeeded.
1026− POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, expiresAt: Timestamp}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /forward-batch headers {authorization: Text}
1029+ POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, expiresAt: Timestamp, priority: Int}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /forward-batch headers {authorization: Text}
1027 -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}]}
1030 -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}]}
1028 = FederationBatchForward
1031 = FederationBatchForward
@@ -1531,6 +1534,8 @@ isBadServerName = \sn ->
1531 || containsSuffix ".ec2.internal" sn -- AWS EC2 internal DNS
1534 || containsSuffix ".ec2.internal" sn -- AWS EC2 internal DNS
1532 || containsSuffix ".amazonaws.com" sn -- AWS service endpoints (S3, SQS, etc.)
1535 || containsSuffix ".amazonaws.com" sn -- AWS service endpoints (S3, SQS, etc.)
1533 || matchesHostname sn "metadata.google.internal" -- GCP metadata server
1536 || matchesHostname sn "metadata.google.internal" -- GCP metadata server
1537+ || containsSuffix ".internal.cloudapp.net" sn -- Azure VM internal DNS
1538+ || matchesHostname sn "metadata.azure.com" -- Azure Instance Metadata Service (IMDS v2)
1534 -- NOTE: Unicode/IDN defense is provided by containsBadUrlChar above, which rejects
1539 -- NOTE: Unicode/IDN defense is provided by containsBadUrlChar above, which rejects
1535 -- any character not in [0-9a-z-.:] via isAllowedServerChar. This blocks homoglyph
1540 -- any character not in [0-9a-z-.:] via isAllowedServerChar. This blocks homoglyph
1536 -- attacks (e.g., Greek omicron in "lοcalhost") assuming Knot's Text type preserves
1541 -- attacks (e.g., Greek omicron in "lοcalhost") assuming Knot's Text type preserves
@@ -3880,7 +3885,7 @@ incrementForwardRetries = \messageId toServer toKey -> atomic do
3880 tfc <- *totalForwardCount
3885 tfc <- *totalForwardCount
3881 set *totalForwardCount = max 0 (tfc - 1)
3886 set *totalForwardCount = max 0 (tfc - 1)
3882 logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
3887 logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
3883− else writeFwdBucket b (do f <- fwdBkt; yield (if f.messageId == messageId && f.toServer == toServer then {messageId: f.messageId, toServer: f.toServer, createdAt: f.createdAt, toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, retries: f.retries + 1, expiresAt: f.expiresAt} else f))
3888+ else writeFwdBucket b (do f <- fwdBkt; yield (if f.messageId == messageId && f.toServer == toServer then {messageId: f.messageId, toServer: f.toServer, createdAt: f.createdAt, toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, retries: f.retries + 1, expiresAt: f.expiresAt, priority: f.priority, senderKey: f.senderKey} else f))
3885-- Retry forwards for a server using batch forward (1 HTTP round-trip per batch of maxBatchSize).
3890-- Retry forwards for a server using batch forward (1 HTTP round-trip per batch of maxBatchSize).
3886-- Falls back to single forward if the peer returns an HTTP-level error (old peer without batch support).
3891-- Falls back to single forward if the peer returns an HTTP-level error (old peer without batch support).
@@ -3893,7 +3898,7 @@ retryForwardsForServer = \entries remaining ->
3893 t <- now
3898 t <- now
3894 auth <- generateFederationAuth "forward" server
3899 auth <- generateFederationAuth "forward" server
3895 corrId <- genTokenN 1
3900 corrId <- genTokenN 1
3896− let batchPayload = do f <- batch; yield {toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId, expiresAt: f.expiresAt}
3901+ let batchPayload = do f <- batch; yield {toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId, expiresAt: f.expiresAt, priority: f.priority}
3897 result <- fetchFederation server (FederationBatchForward {
3902 result <- fetchFederation server (FederationBatchForward {
3898 forwards: batchPayload, fromServer: serverHostN,
3903 forwards: batchPayload, fromServer: serverHostN,
3899 requestTimestamp: t, authorization: auth, correlationId: corrId, protocolVersion: federationProtocolVersion})
3904 requestTimestamp: t, authorization: auth, correlationId: corrId, protocolVersion: federationProtocolVersion})
@@ -4632,7 +4637,7 @@ forwardCandidates = \targets messages forwards t ->
4632 let origin = if m.originServer == "" then serverHostN else m.originServer
4637 let origin = if m.originServer == "" then serverHostN else m.originServer
4633 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
4638 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
4634 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,
4639 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,
4635− originServer: origin, retries: 0, expiresAt: m.expiresAt, priority: m.priority})) [] targets
4640+ originServer: origin, retries: 0, expiresAt: m.expiresAt, priority: m.priority, senderKey: m.senderKey})) [] targets
4638-- Challenge-IP binding rules:
4643-- Challenge-IP binding rules:
@@ -5041,7 +5046,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
5041 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
5046 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
5042 let fbi = pubkeyBucket to
5047 let fbi = pubkeyBucket to
5043 fwdBkt <- readFwdBucket fbi
5048 fwdBkt <- readFwdBucket fbi
5044− let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: msgExpiresAt, priority: msgPriority}
5049+ let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: msgExpiresAt, priority: msgPriority, senderKey: senderKey}
5045 totalFwdCount <- *totalForwardCount
5050 totalFwdCount <- *totalForwardCount
5046 let globalCap = maxForwardsEntries - totalFwdCount
5051 let globalCap = maxForwardsEntries - totalFwdCount
5047 -- Per-key cap: prevents one key from consuming disproportionate forward slots
5052 -- Per-key cap: prevents one key from consuming disproportionate forward slots
@@ -5065,8 +5070,8 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
5065 fwdInfo <- atomic do
5070 fwdInfo <- atomic do
5066 fwdBkt <- readFwdBucket (pubkeyBucket to)
5071 fwdBkt <- readFwdBucket (pubkeyBucket to)
5067 case headOf (do f <- fwdBkt; where f.messageId == effectiveMsgId && f.toServer == server; yield f) of
5072 case headOf (do f <- fwdBkt; where f.messageId == effectiveMsgId && f.toServer == server; yield f) of
5068− Nothing {} -> yield {pending: False {}, fwdExpiresAt: 0}
5073+ Nothing {} -> yield {pending: False {}, fwdExpiresAt: 0, fwdPriority: 0}
5069− Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt}
5074+ Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt, fwdPriority: fe.priority}
5070 when fwdInfo.pending (do
5075 when fwdInfo.pending (do
5071 fwdT <- now
5076 fwdT <- now
5072 fwdAuth <- generateFederationAuth "forward" server
5077 fwdAuth <- generateFederationAuth "forward" server
@@ -5076,7 +5081,7 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
5076 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
5081 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
5077 fromServer: serverHostN, requestTimestamp: fwdT,
5082 fromServer: serverHostN, requestTimestamp: fwdT,
5078 authorization: fwdAuth, correlationId: fwdCorrId,
5083 authorization: fwdAuth, correlationId: fwdCorrId,
5079− expiresAt: fwdInfo.fwdExpiresAt, protocolVersion: federationProtocolVersion})
5084+ expiresAt: fwdInfo.fwdExpiresAt, priority: fwdInfo.fwdPriority, protocolVersion: federationProtocolVersion})
5080 case fwdResult of
5085 case fwdResult of
5081 Err e -> do
5086 Err e -> do
5082 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
5087 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
@@ -5098,8 +5103,8 @@ sendBatchForwardsSequential = \msgs server ->
5098 batchFwdInfo <- atomic do
5103 batchFwdInfo <- atomic do
5099 fwdBkt <- readFwdBucket (pubkeyBucket r.toKey)
5104 fwdBkt <- readFwdBucket (pubkeyBucket r.toKey)
5100 case headOf (do f <- fwdBkt; where f.messageId == r.messageId && f.toServer == server; yield f) of
5105 case headOf (do f <- fwdBkt; where f.messageId == r.messageId && f.toServer == server; yield f) of
5101− Nothing {} -> yield {pending: False {}, fwdExpiresAt: 0}
5106+ Nothing {} -> yield {pending: False {}, fwdExpiresAt: 0, fwdPriority: 0}
5102− Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt}
5107+ Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt, fwdPriority: fe.priority}
5103 when batchFwdInfo.pending (do
5108 when batchFwdInfo.pending (do
5104 fwdT <- now
5109 fwdT <- now
5105 fwdAuth <- generateFederationAuth "forward" server
5110 fwdAuth <- generateFederationAuth "forward" server
@@ -5109,7 +5114,7 @@ sendBatchForwardsSequential = \msgs server ->
5109 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
5114 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
5110 fromServer: serverHostN, requestTimestamp: fwdT,
5115 fromServer: serverHostN, requestTimestamp: fwdT,
5111 authorization: fwdAuth, correlationId: fwdCorrId,
5116 authorization: fwdAuth, correlationId: fwdCorrId,
5112− expiresAt: batchFwdInfo.fwdExpiresAt, protocolVersion: federationProtocolVersion})
5117+ expiresAt: batchFwdInfo.fwdExpiresAt, priority: batchFwdInfo.fwdPriority, protocolVersion: federationProtocolVersion})
5113 case fwdResult of
5118 case fwdResult of
5114 Err e -> do
5119 Err e -> do
5115 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
5120 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
@@ -5519,7 +5524,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
5519-- (fromServer != originServer) where an intermediate server re-forwards because
5524-- (fromServer != originServer) where an intermediate server re-forwards because
5520-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored
5525-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored
5521-- twice) and forwardCandidates (won't re-forward to the same server).
5526-- twice) and forwardCandidates (won't re-forward to the same server).
5522−handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt authorization respond -> do
5527+handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond -> do
5523 t <- now
5528 t <- now
5524 let corrId = sanitizeCorrelationId correlationId
5529 let corrId = sanitizeCorrelationId correlationId
5525 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward " ++ messageId ++ " from " ++ fromServer))
5530 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward " ++ messageId ++ " from " ++ fromServer))
@@ -5606,9 +5611,10 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
5606 -- Use sender-specified TTL if provided and still in the future; clamp to local max.
5611 -- Use sender-specified TTL if provided and still in the future; clamp to local max.
5607 -- expiresAt == 0 means pre-v10 peer or unset; fall back to server default.
5612 -- expiresAt == 0 means pre-v10 peer or unset; fall back to server default.
5608 let fwdExpiresAt = if expiresAt > 0 && expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
5613 let fwdExpiresAt = if expiresAt > 0 && expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
5614+ let clampedPriority = clamp 0 maxMessagePriority priority
5609 writeMessageBucket bkt (union bucket
5615 writeMessageBucket bkt (union bucket
5610 [{messageId: messageId, toKey: toKey,
5616 [{messageId: messageId, toKey: toKey,
5611− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: 0}])
5617+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: clampedPriority}])
5612 set *totalMessageCount = totalMsgCount + 1
5618 set *totalMessageCount = totalMsgCount + 1
5613 writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
5619 writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
5614 signalMessageForKey toKey seqTs
5620 signalMessageForKey toKey seqTs
@@ -5637,7 +5643,9 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
5637 _ -> respond {ok: True {}, error: ""}))
5643 _ -> respond {ok: True {}, error: ""}))
5639-- Generate sender delivery receipts for messages confirmed delivered via federation ack.
5645-- Generate sender delivery receipts for messages confirmed delivered via federation ack.
5646+-- Uses senderKey from the ForwardEntry directly (stored since v12), so receipts are
5640−-- Reads the original message from the origin server's mailbox to extract senderKey.
5647+-- generated even if the original message was already pruned from the mailbox.
5648+-- Falls back to message lookup for pre-v12 forward entries where senderKey is "".
5641-- Per-bucket: only reads message buckets containing the acked recipients.
5649-- Per-bucket: only reads message buckets containing the acked recipients.
5642fedDeliveryReceipts : [Int] -> [ForwardEntry] -> Timestamp -> IO ()
5650fedDeliveryReceipts : [Int] -> [ForwardEntry] -> Timestamp -> IO ()
5643fedDeliveryReceipts = \bucketIds forwards t -> case headOf bucketIds of
5651fedDeliveryReceipts = \bucketIds forwards t -> case headOf bucketIds of
@@ -5645,15 +5653,22 @@ fedDeliveryReceipts = \bucketIds forwards t -> case headOf bucketIds of
5645 Just {value: b} -> do
5653 Just {value: b} -> do
5646 let rest = do x <- bucketIds; where x != b; yield x
5654 let rest = do x <- bucketIds; where x != b; yield x
5647 let bucketFwds = do f <- forwards; where pubkeyBucket f.toKey == b; yield f
5655 let bucketFwds = do f <- forwards; where pubkeyBucket f.toKey == b; yield f
5648− atomic do
5656+ -- Fast path: forward entries with senderKey (v12+) don't need message lookup
5657+ let fromForwards = do f <- bucketFwds; where f.senderKey != ""; yield {senderKey: f.senderKey, messageId: f.messageId, toKey: f.toKey, deliveredAt: t}
5658+ -- Fallback: pre-v12 entries without senderKey — look up the original message
5659+ let needLookup = do f <- bucketFwds; where f.senderKey == ""; yield f
5660+ msgDeliveries <- if count needLookup > 0
5649− msgs <- readMessageBucket b
5662+ msgs <- readMessageBucket b
5650− let deliveries = do f <- bucketFwds; m <- msgs; where m.messageId == f.messageId && m.toKey == f.toKey && m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
5663+ yield (do f <- needLookup; m <- msgs; where m.messageId == f.messageId && m.toKey == f.toKey && m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t})
5665+ let deliveries = union fromForwards msgDeliveries
5651− when (count deliveries > 0) (do
5666+ when (count deliveries > 0) (atomic do
5652− sd <- *senderDeliveries
5667+ sd <- *senderDeliveries
5653− let combined = union sd deliveries
5668+ let combined = union sd deliveries
5654− set *senderDeliveries = if count combined > maxSenderDeliveries
5669+ set *senderDeliveries = if count combined > maxSenderDeliveries
5655− then evictOneOldest combined (\d -> d.deliveredAt)
5670+ then evictOneOldest combined (\d -> d.deliveredAt)
5656− else combined)
5671+ else combined)
5657 fedDeliveryReceipts rest forwards t
5672 fedDeliveryReceipts rest forwards t
5659-- POST /federation/ack: Peer server confirms delivery of messages we originated.
5674-- POST /federation/ack: Peer server confirms delivery of messages we originated.
@@ -5892,9 +5907,10 @@ forwardBatchInner = \entries fromServer t acc ->
5892 let seqTs = nextSeqTs t seqCounter
5907 let seqTs = nextSeqTs t seqCounter
5893 set *seqCounter = seqTs
5908 set *seqCounter = seqTs
5894 let fwdExpiresAt = if entry.expiresAt > 0 && entry.expiresAt > t then min entry.expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
5909 let fwdExpiresAt = if entry.expiresAt > 0 && entry.expiresAt > t then min entry.expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
5910+ let batchClampedPriority = clamp 0 maxMessagePriority entry.priority
5895 writeMessageBucket bkt (union bucket
5911 writeMessageBucket bkt (union bucket
5896 [{messageId: entry.messageId, toKey: entry.toKey,
5912 [{messageId: entry.messageId, toKey: entry.toKey,
5897− encryptedBlob: entry.encryptedBlob, timestamp: clampedTs, originServer: entry.originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: 0}])
5913+ encryptedBlob: entry.encryptedBlob, timestamp: clampedTs, originServer: entry.originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: batchClampedPriority}])
5898 set *totalMessageCount = totalMsgCount + 1
5914 set *totalMessageCount = totalMsgCount + 1
5899 writeRcBucket bkt (adjustRecipientCount recipientCounts entry.toKey 1)
5915 writeRcBucket bkt (adjustRecipientCount recipientCounts entry.toKey 1)
5900 signalMessageForKey entry.toKey seqTs
5916 signalMessageForKey entry.toKey seqTs
@@ -7462,8 +7478,8 @@ serveInner = \req -> case req of
7462 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
7478 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
7463 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
7479 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
7464 warnOldProtocol fromServer protocolVersion; handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
7480 warnOldProtocol fromServer protocolVersion; handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
7465− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, expiresAt, protocolVersion, authorization, respond} -> do
7481+ ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, expiresAt, priority, protocolVersion, authorization, respond} -> do
7466− warnOldProtocol fromServer protocolVersion; handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt authorization respond
7482+ warnOldProtocol fromServer protocolVersion; handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond
7467 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
7483 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
7468 warnOldProtocol fromServer protocolVersion; handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
7484 warnOldProtocol fromServer protocolVersion; handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
7469 FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
7485 FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
@@ -7655,8 +7671,13 @@ backgroundPrune = do
7655 -- to clients with pre-jump cursors. Without a generation bump, these messages are
7671 -- to clients with pre-jump cursors. Without a generation bump, these messages are
7656 -- silently skipped until real time catches up to the inflated timestamps.
7672 -- silently skipped until real time catches up to the inflated timestamps.
7657 when (lp > 0 && t - lp > backgroundPruneInterval * 10) (do
7673 when (lp > 0 && t - lp > backgroundPruneInterval * 10) (do
7658− logWarn ("clock forward jump detected — " ++ show (t - lp) ++ "ms since last prune (expected ~" ++ show backgroundPruneInterval ++ "ms), bumping server generation to reset client cursors")
7674+ logWarn ("clock forward jump detected — " ++ show (t - lp) ++ "ms since last prune (expected ~" ++ show backgroundPruneInterval ++ "ms), bumping server generation and resetting seqCounter")
7659− set *serverGeneration = t)
7675+ set *serverGeneration = t
7676+ -- Also reset seqCounter: messages stored during the forward jump window have
7677+ -- inflated receivedAt timestamps from nextSeqTs using the jumped seqCounter.
7678+ -- Without this reset, new messages would continue getting inflated timestamps
7679+ -- until seqCounter is organically surpassed by real time.
7680+ set *seqCounter = t)
7660 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead
7681 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead
7661 sa <- *startedAt
7682 sa <- *startedAt
7662 lastHb <- *lastHeartbeatAt
7683 lastHb <- *lastHeartbeatAt
@@ -7772,25 +7793,33 @@ backgroundSnapshot = do
7772 lastFp <- *lastSnapshotFingerprint
7793 lastFp <- *lastSnapshotFingerprint
7773 let stateChanged = lastFp == 0 || fingerprint != lastFp
7794 let stateChanged = lastFp == 0 || fingerprint != lastFp
7774 when (hasState && stateChanged && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do
7795 when (hasState && stateChanged && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do
7775− -- Split into per-collection reads to reduce contention (auto-export is best-effort;
7796+ -- Single atomic read for consistent [SNAPSHOT_DATA] output (used for restore).
7776− -- slight inconsistency between collections is acceptable vs. one large atomic block
7797+ -- Previous per-collection reads caused inconsistencies (e.g., ackedIds referencing
7777− -- holding all 80+ bucket locks and competing with hot-path operations).
7798+ -- messageIds from a different snapshot window). The contention cost is acceptable
7799+ -- since auto-export runs at most once per hour.
7800+ exportSnapshot <- atomic do
7778− p <- atomic do pp <- readAllPresence; yield pp
7801+ p <- readAllPresence
7779− a <- atomic do aa <- readAllAcked; yield aa
7802+ a <- readAllAcked
7780− ao <- atomic do aao <- *ackOverflow; yield aao
7803+ ao <- *ackOverflow
7781− d <- atomic do dd <- readAllDelivered; yield dd
7804+ d <- readAllDelivered
7782− ff <- atomic do fff <- *forwardFailures; yield fff
7805+ ff <- *forwardFailures
7783− f <- atomic do ff2 <- readAllForwards; yield ff2
7806+ f <- readAllForwards
7784− fn <- atomic do fnn <- readAllNonces; yield fnn
7807+ fn <- readAllNonces
7785− extraState <- atomic do
7786 bp <- *blockedPeers
7808 bp <- *blockedPeers
7787 ap <- *allowedPeers
7809 ap <- *allowedPeers
7788 pac <- *pendingAckCleanups
7810 pac <- *pendingAckCleanups
7789 se <- *senderExpirations
7811 se <- *senderExpirations
7790 sdl <- *senderDeliveries
7812 sdl <- *senderDeliveries
7791− yield {bp: bp, ap: ap, pac: pac, se: se, sdl: sdl}
7813+ yield {p: p, a: a, ao: ao, d: d, ff: ff, f: f, fn: fn, bp: bp, ap: ap, pac: pac, se: se, sdl: sdl}
7814+ let p = exportSnapshot.p
7815+ let a = exportSnapshot.a
7816+ let ao = exportSnapshot.ao
7817+ let d = exportSnapshot.d
7818+ let ff = exportSnapshot.ff
7819+ let f = exportSnapshot.f
7820+ let fn = exportSnapshot.fn
7792 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
7821 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
7793− let exportData = {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: extraState.bp, allowedPeers: extraState.ap, pendingAckCleanups: extraState.pac, senderExpirations: extraState.se, senderDeliveries: extraState.sdl}
7822+ let exportData = {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: exportSnapshot.bp, allowedPeers: exportSnapshot.ap, pendingAckCleanups: exportSnapshot.pac, senderExpirations: exportSnapshot.se, senderDeliveries: exportSnapshot.sdl}
7794 autoRtSec <- *runtimeMaintenanceSecret
7823 autoRtSec <- *runtimeMaintenanceSecret
7795 let autoEffSec = if autoRtSec != "" then autoRtSec else maintenanceSecret
7824 let autoEffSec = if autoRtSec != "" then autoRtSec else maintenanceSecret
7796 let autoTag = snapshotIntegrityTag exportCs autoEffSec
7825 let autoTag = snapshotIntegrityTag exportCs autoEffSec
@@ -7853,6 +7882,24 @@ backgroundWatchdog = do
7853 logStaleInFlight hungSlots t)
7882 logStaleInFlight hungSlots t)
7854 backgroundWatchdog
7883 backgroundWatchdog
7885+-- Meta-watchdog: monitors the watchdog itself. The backgroundWatchdog can restart prune
7886+-- and heartbeat loops, but if the watchdog itself panics (e.g., a bug in hung-slot detection),
7887+-- no loop monitors it. This ultra-simple loop just checks lastWatchdogAt freshness and
7888+-- restarts backgroundWatchdog if stale. Runs infrequently (10x watchdog interval) to minimize cost.
7889+-- Cannot restart itself — a panic here surfaces as healthWatchdogMaxAge failure in /health.
7890+backgroundMetaWatchdog = do
7891+ sleep (backgroundPruneInterval * 50)
7894+ let warmup = t - sa < backgroundPruneInterval * 60
7895+ if warmup then backgroundMetaWatchdog
7897+ lastWd <- *lastWatchdogAt
7898+ when (lastWd > 0 && t - lastWd > healthWatchdogMaxAge * 3) (do
7899+ logError "meta-watchdog: watchdog loop appears dead, restarting"
7900+ fork backgroundWatchdog)
7901+ backgroundMetaWatchdog
7856-- Startup connectivity probe: send an empty gossip to each known peer and log results.
7903-- Startup connectivity probe: send an empty gossip to each known peer and log results.
7857-- Surfaces DNS, firewall, and TLS issues before the first real gossip cycle.
7904-- Surfaces DNS, firewall, and TLS issues before the first real gossip cycle.
7858-- Probes up to 10 peers sequentially (not forked) to avoid overwhelming the network.
7905-- Probes up to 10 peers sequentially (not forked) to avoid overwhelming the network.
@@ -7977,6 +8024,7 @@ main = do
7977 fork backgroundPrune
8024 fork backgroundPrune
7978 fork pollHeartbeat
8025 fork pollHeartbeat
7979 fork backgroundWatchdog
8026 fork backgroundWatchdog
8027+ fork backgroundMetaWatchdog
7980 when (autoSnapshotInterval > 0) (fork backgroundSnapshot)
8028 when (autoSnapshotInterval > 0) (fork backgroundSnapshot)
7981 -- Startup peer connectivity check: probe known peers after a short delay
8029 -- Startup peer connectivity check: probe known peers after a short delay
7982 -- to surface federation misconfigurations immediately (DNS, firewall, TLS).
8030 -- to surface federation misconfigurations immediately (DNS, firewall, TLS).