@@ -165,7 +165,7 @@ federationLatencyWindowMs : Int<Ms>
165165federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
166166
167167-- Snapshot format version (increment on schema changes to prevent corrupt restores)
168−snapshotVersion = 9 -- bumped for senderDeliveries field in snapshot
168+snapshotVersion = 10 -- bumped for ForwardEntry.expiresAt field
169169
170170-- Server build version: included in /health and /stats responses for deployment verification.
171171-- Update this string when deploying a new build so operators can confirm which version is running.
@@ -376,7 +376,7 @@ type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHe
376376-- Presence: tracks that a pubkey is online at a given server, with expiry
377377type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
378378-- Queued federation forward: message waiting to be delivered to a remote server.
379−type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int}
379+type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int, expiresAt: Timestamp}
380380-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
381381type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}
382382-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
@@ -420,6 +420,10 @@ type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, seenAt: Timestam
420420type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, expiresInMs: Int}
421421type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
422422
423+-- Batch cancel types
424+type BatchCancelEntry = {messageId: MessageIdHex, to: PubkeyHex}
425+type BatchCancelResult = {messageId: MessageIdHex, cancelled: Bool}
426+
423427-- Sender expiration notification: tracks when a locally-originated message expired unread.
424428-- Pruned after messageMaxAge. Included in the sender's next poll response.
425429type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}
@@ -840,6 +844,7 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
840844*cachedHealthFederation : Bool
841845*cachedHealthSnapshot : Bool
842846*cachedHealthWatchdog : Bool
847+*cachedHealthNonce : Bool
843848
844849-- ============================================
845850-- Routes
@@ -906,6 +911,11 @@ route ClientApi where
906911 POST {messageId: MessageIdHex, to: PubkeyHex} /cancel headers {authorization: Text, clientIp: Text}
907912 -> {ok: Bool, authorized: Bool, cancelled: Bool, error: Text}
908913 = CancelMessage
914+ -- Cancel (retract) multiple sent messages in one request.
915+ -- Each cancel is processed independently; partial success is possible.
916+ POST {cancels: [BatchCancelEntry]} /cancel-batch headers {authorization: Text, clientIp: Text}
917+ -> {results: [BatchCancelResult], authorized: Bool, error: Text}
918+ = CancelBatch
909919 -- Look up which server a pubkey is present on (for federation routing)
910920 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
911921 -> {server: AnyServerName, authorized: Bool, error: Text}
@@ -930,7 +940,7 @@ route FederationApi where
930940 -- privacy model (the server never learns who sent a message). /messages/cancel for
931941 -- federated messages requires the client to include sender identity in the encrypted
932942 -- payload; the server cannot verify sender identity for federation-forwarded messages.
933− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /forward headers {authorization: Text}
943+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, expiresAt: Timestamp} /forward headers {authorization: Text}
934944 -> {ok: Bool, error: Text}
935945 = ForwardMessage
936946 -- Acknowledge delivery of messages back to the origin server
@@ -943,6 +953,10 @@ route FederationApi where
943953 POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /cancel headers {authorization: Text}
944954 -> {ok: Bool, error: Text, cancelled: Bool}
945955 = FederationCancel
956+ -- Batch cancel: cancel multiple previously forwarded messages in one request.
957+ POST {cancels: [{messageId: MessageIdHex, toKey: PubkeyHex}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /cancel-batch headers {authorization: Text}
958+ -> {ok: Bool, error: Text, results: [BatchCancelResult]}
959+ = FederationBatchCancel
946960
947961-- Federation callback verification: the receiving server calls back to the sender
948962-- to confirm a request token is genuine. No shared secret needed.
@@ -1053,7 +1067,7 @@ route MaintenanceApi where
10531067 = RotateSecret
10541068
10551069route HealthApi where
1056− GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool, draining: Bool} = Health
1070+ GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, draining: Bool} = Health
10571071
10581072-- Returns counts of in-memory collections (requires maintenance secret)
10591073-- TODO: If Knot adds raw text response support, add a /metrics endpoint
@@ -3655,7 +3669,7 @@ incrementForwardRetries = \messageId toServer toKey -> atomic do
36553669 tfc <- *totalForwardCount
36563670 set *totalForwardCount = max 0 (tfc - 1)
36573671 logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
3658− 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} else f))
3672+ 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))
36593673
36603674retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
36613675retryForwardsForServer = \entries remaining ->
@@ -3672,7 +3686,7 @@ retryForwardsForServer = \entries remaining ->
36723686 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
36733687 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
36743688 fromServer: serverHostN, requestTimestamp: t,
3675− authorization: auth, correlationId: corrId})
3689+ authorization: auth, correlationId: corrId, expiresAt: f.expiresAt})
36763690 case result of
36773691 Err e -> do
36783692 logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
@@ -4342,7 +4356,7 @@ forwardCandidates = \targets messages forwards t ->
43424356 let origin = if m.originServer == "" then serverHostN else m.originServer
43434357 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
43444358 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,
4345− originServer: origin, retries: 0})) [] targets
4359+ originServer: origin, retries: 0, expiresAt: m.expiresAt})) [] targets
43464360
43474361
43484362-- Challenge-IP binding rules:
@@ -4745,7 +4759,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
47454759 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
47464760 let fbi = pubkeyBucket to
47474761 fwdBkt <- readFwdBucket fbi
4748− 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}
4762+ 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}
47494763 totalFwdCount <- *totalForwardCount
47504764 let globalCap = maxForwardsEntries - totalFwdCount
47514765 -- Per-key cap: prevents one key from consuming disproportionate forward slots
@@ -4766,10 +4780,12 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
47664780forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
47674781 when (count remoteServers > 0 && federationEnabled) (
47684782 fork (sendToEach remoteServers (\server -> do
4769− stillPending <- atomic do
4783+ fwdInfo <- atomic do
47704784 fwdBkt <- readFwdBucket (pubkeyBucket to)
4771− yield (any fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == server))
4772− when stillPending (do
4785+ case headOf (do f <- fwdBkt; where f.messageId == effectiveMsgId && f.toServer == server; yield f) of
4786+ Nothing {} -> yield {pending: False {}, fwdExpiresAt: 0}
4787+ Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt}
4788+ when fwdInfo.pending (do
47734789 fwdT <- now
47744790 fwdAuth <- generateFederationAuth "forward" server
47754791 fwdCorrId <- genTokenN 1
@@ -4777,7 +4793,8 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
47774793 toKey: to, encryptedBlob: encryptedBlob,
47784794 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
47794795 fromServer: serverHostN, requestTimestamp: fwdT,
4780− authorization: fwdAuth, correlationId: fwdCorrId})
4796+ authorization: fwdAuth, correlationId: fwdCorrId,
4797+ expiresAt: fwdInfo.fwdExpiresAt})
47814798 case fwdResult of
47824799 Err e -> do
47834800 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
@@ -4796,10 +4813,12 @@ sendBatchForwardsSequential = \msgs server ->
47964813 Nothing {} -> yield {}
47974814 Just {value: r} -> do
47984815 let rest = do m <- msgs; where m.messageId != r.messageId; yield m
4799− stillPending <- atomic do
4816+ batchFwdInfo <- atomic do
48004817 fwdBkt <- readFwdBucket (pubkeyBucket r.toKey)
4801− yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))
4802− when stillPending (do
4818+ case headOf (do f <- fwdBkt; where f.messageId == r.messageId && f.toServer == server; yield f) of
4819+ Nothing {} -> yield {pending: False {}, fwdExpiresAt: 0}
4820+ Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt}
4821+ when batchFwdInfo.pending (do
48034822 fwdT <- now
48044823 fwdAuth <- generateFederationAuth "forward" server
48054824 fwdCorrId <- genTokenN 1
@@ -4807,7 +4826,8 @@ sendBatchForwardsSequential = \msgs server ->
48074826 toKey: r.toKey, encryptedBlob: r.blob,
48084827 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
48094828 fromServer: serverHostN, requestTimestamp: fwdT,
4810− authorization: fwdAuth, correlationId: fwdCorrId})
4829+ authorization: fwdAuth, correlationId: fwdCorrId,
4830+ expiresAt: batchFwdInfo.fwdExpiresAt})
48114831 case fwdResult of
48124832 Err e -> do
48134833 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
@@ -5211,7 +5231,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
52115231-- (fromServer != originServer) where an intermediate server re-forwards because
52125232-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored
52135233-- twice) and forwardCandidates (won't re-forward to the same server).
5214−handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId authorization respond -> do
5234+handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt authorization respond -> do
52155235 t <- now
52165236 let corrId = sanitizeCorrelationId correlationId
52175237 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward " ++ messageId ++ " from " ++ fromServer))
@@ -5295,7 +5315,9 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
52955315 seqCounter <- *seqCounter
52965316 let seqTs = nextSeqTs t seqCounter
52975317 set *seqCounter = seqTs
5298− let fwdExpiresAt = seqTs + unackedMessageMaxAge
5318+ -- Use sender-specified TTL if provided and still in the future; clamp to local max.
5319+ -- expiresAt == 0 means pre-v10 peer or unset; fall back to server default.
5320+ let fwdExpiresAt = if expiresAt > 0 && expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
52995321 writeMessageBucket bkt (union bucket
53005322 [{messageId: messageId, toKey: toKey,
53015323 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: ""}])
@@ -5326,6 +5348,26 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
53265348 ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}
53275349 _ -> respond {ok: True {}, error: ""}))
53285350
5351+-- Generate sender delivery receipts for messages confirmed delivered via federation ack.
5352+-- Reads the original message from the origin server's mailbox to extract senderKey.
5353+-- Per-bucket: only reads message buckets containing the acked recipients.
5354+fedDeliveryReceipts : [Int] -> [ForwardEntry] -> Timestamp -> IO ()
5355+fedDeliveryReceipts = \bucketIds forwards t -> case headOf bucketIds of
5356+ Nothing {} -> yield {}
5357+ Just {value: b} -> do
5358+ let rest = do x <- bucketIds; where x != b; yield x
5359+ let bucketFwds = do f <- forwards; where pubkeyBucket f.toKey == b; yield f
5360+ atomic do
5361+ msgs <- readMessageBucket b
5362+ 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}
5363+ when (count deliveries > 0) (do
5364+ sd <- *senderDeliveries
5365+ let combined = union sd deliveries
5366+ set *senderDeliveries = if count combined > maxSenderDeliveries
5367+ then evictOneOldest combined (\d -> d.deliveredAt)
5368+ else combined)
5369+ fedDeliveryReceipts rest forwards t
5370+
53295371-- POST /federation/ack: Peer server confirms delivery of messages we originated.
53305372-- Removes corresponding forward entries. Also cleans up the original message if the
53315373-- recipient has no local presence and no other pending forwards for that message.
@@ -5356,6 +5398,11 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
53565398 -- Prevents a malicious peer from resetting its backoff by padding ack requests
53575399 -- with known messageIds that no longer have forward entries.
53585400 when (ackedCount > 0 && unmatched == 0) (clearFailure fromServer)
5401+ -- Generate delivery receipts for locally-originated messages confirmed delivered via federation.
5402+ -- Looks up the original message (still on this server) to get senderKey.
5403+ when (ackedCount > 0) (do
5404+ let receiptBuckets = uniqueBucketsFor removedForwards (\f -> f.toKey)
5405+ fedDeliveryReceipts receiptBuckets removedForwards t)
53595406 -- Deferred origin message cleanup: only queue cleanup for messages that actually had
53605407 -- matching forward entries. Messages acked without a forward entry are NOT cleaned up,
53615408 -- preventing premature deletion when a peer acks messages it never actually received.
@@ -5421,6 +5468,124 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
54215468 (logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")"))
54225469 yield (respond {ok: True {}, error: "", cancelled: result.cancelled}))
54235470
5471+-- POST /federation/cancel-batch: Batch cancel multiple previously forwarded messages.
5472+-- Reuses the cancel logic from handleFederationCancel for each entry.
5473+handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId authorization respond -> do
5474+ t <- now
5475+ let corrId = sanitizeCorrelationId correlationId
5476+ when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel-batch (" ++ show (count cancels) ++ " entries) from " ++ fromServer))
5477+ let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: []}
5478+ withFederationAuth authorization t batchRespond fromServer "cancel" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (False {}) (do
5479+ let capped = takeRelation maxBatchSize cancels
5480+ results <- cancelBatchInner capped fromServer t []
5481+ let cancelledCount = count (do r <- results; where r.cancelled; yield r)
5482+ when (cancelledCount > 0)
5483+ (logInfo ("federation/cancel-batch: cancelled " ++ show cancelledCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer))
5484+ yield (respond {ok: True {}, error: "", results: results}))
5485+
5486+-- Process each cancel entry in a batch, grouping by bucket for efficiency.
5487+cancelBatchInner = \entries fromServer t acc ->
5488+ case headOf entries of
5489+ Nothing {} -> yield acc
5490+ Just {value: entry} -> do
5491+ let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
5492+ result <- atomic do
5493+ let b = pubkeyBucket entry.toKey
5494+ bucket <- readMessageBucket b
5495+ let toCancel = headOf (do m <- bucket; where m.messageId == entry.messageId && m.toKey == entry.toKey && m.originServer == fromServer; yield m)
5496+ case toCancel of
5497+ Nothing {} -> yield {cancelled: False {}}
5498+ Just {} -> do
5499+ writeMessageBucket b (do m <- bucket; where not (m.messageId == entry.messageId && m.toKey == entry.toKey); yield m)
5500+ tc <- *totalMessageCount
5501+ set *totalMessageCount = max 0 (tc - 1)
5502+ rc <- readRcBucket b
5503+ writeRcBucket b (adjustRecipientCount rc entry.toKey (0 - 1))
5504+ fwdBkt <- readFwdBucket b
5505+ let keptFwds = do f <- fwdBkt; where not (f.messageId == entry.messageId && f.toKey == entry.toKey); yield f
5506+ writeFwdBucket b keptFwds
5507+ let fwdRemoved = count fwdBkt - count keptFwds
5508+ when (fwdRemoved > 0) (do
5509+ tfc <- *totalForwardCount
5510+ set *totalForwardCount = max 0 (tfc - fwdRemoved))
5511+ yield {cancelled: True {}}
5512+ cancelBatchInner rest fromServer t (union acc [{messageId: entry.messageId, cancelled: result.cancelled}])
5513+
5514+-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
5515+-- Auth checked once; each cancel processed independently. Partial success possible.
5516+handleCancelBatch = \cancels authorization clientIp respond -> do
5517+ t <- now
5518+ dp <- *drainPhase
5519+ if dp >= 2 then yield (respond {results: [], authorized: False {}, error: ""})
5520+ else do
5521+ globalRateLimited <- checkGlobalRate t
5522+ if globalRateLimited then yield (respond {results: [], authorized: False {}, error: "rate_limited"})
5523+ else
5524+ withSessionAuth authorization t clientIp
5525+ (yield (respond {results: [], authorized: False {}, error: "unauthorized"}))
5526+ (\senderKey -> do
5527+ cancelRateLimited <- atomic do
5528+ cwr <- *clientWriteRates
5529+ cwRate <- effectiveClientWriteRate
5530+ let rateResult = checkPubkeyRate cwr senderKey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t
5531+ set *clientWriteRates = rateResult.entries
5532+ yield rateResult.limited
5533+ if cancelRateLimited then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
5534+ else do
5535+ let capped = takeRelation maxBatchSize cancels
5536+ -- Process each cancel, collecting results and per-server cancel targets
5537+ results <- clientCancelBatchInner senderKey capped t [] []
5538+ -- Send federation batch cancels grouped by server
5539+ fork (when (federationEnabled && count results.cancelServers > 0)
5540+ (sendToEach results.cancelServers (\server -> do
5541+ let serverCancels = do c <- results.fedCancels; where any c.servers (\s -> s == server); yield {messageId: c.messageId, toKey: c.toKey}
5542+ when (count serverCancels > 0) (do
5543+ cancelT <- now
5544+ cancelAuth <- generateFederationAuth "cancel" server
5545+ cancelCorrId <- genTokenN 1
5546+ fetchFederation server (FederationBatchCancel {
5547+ cancels: serverCancels, fromServer: serverHostN,
5548+ requestTimestamp: cancelT, correlationId: cancelCorrId,
5549+ authorization: cancelAuth})
5550+ yield {}))))
5551+ yield (respond {results: results.results, authorized: True {}, error: ""}))
5552+
5553+-- Process each client cancel entry, returning results and federation cancel targets.
5554+clientCancelBatchInner = \senderKey entries t resultAcc fedAcc ->
5555+ case headOf entries of
5556+ Nothing {} -> do
5557+ let allServers = fold (\acc c -> fold (\a s -> if any a (\x -> x == s) then a else union [s] a) acc c.servers) [] fedAcc
5558+ yield {results: resultAcc, fedCancels: fedAcc, cancelServers: allServers}
5559+ Just {value: entry} -> do
5560+ let rest = do e <- entries; where e.messageId != entry.messageId || e.to != entry.to; yield e
5561+ cancelResult <- atomic do
5562+ let b = pubkeyBucket entry.to
5563+ bucket <- readMessageBucket b
5564+ let toCancel = headOf (do m <- bucket; where m.messageId == entry.messageId && m.toKey == entry.to; yield m)
5565+ case toCancel of
5566+ Nothing {} -> yield {cancelled: False {}, cancelServers: []}
5567+ Just {value: msg} ->
5568+ if msg.originServer != "" then yield {cancelled: False {}, cancelServers: []}
5569+ else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, cancelServers: []}
5570+ else do
5571+ writeMessageBucket b (do m <- bucket; where not (m.messageId == entry.messageId && m.toKey == entry.to); yield m)
5572+ tc <- *totalMessageCount
5573+ set *totalMessageCount = max 0 (tc - 1)
5574+ rc <- readRcBucket b
5575+ writeRcBucket b (adjustRecipientCount rc entry.to (0 - 1))
5576+ fwdBkt <- readFwdBucket b
5577+ let cancelledFwds = do f <- fwdBkt; where f.messageId == entry.messageId && f.toKey == entry.to; yield f
5578+ let forwardTargetServers = collectPeerServers cancelledFwds (\f -> f.toServer)
5579+ writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == entry.messageId && f.toKey == entry.to); yield f)
5580+ presenceBkt <- readPresenceBucket b
5581+ let remotePresence = collectPeerServers (presenceForKey presenceBkt entry.to t) (\p -> p.server)
5582+ let allCancelServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union [s] acc) remotePresence forwardTargetServers
5583+ yield {cancelled: True {}, cancelServers: allCancelServers}
5584+ let newFedAcc = if cancelResult.cancelled && count cancelResult.cancelServers > 0
5585+ then union fedAcc [{messageId: entry.messageId, toKey: entry.to, servers: cancelResult.cancelServers}]
5586+ else fedAcc
5587+ clientCancelBatchInner senderKey rest t (union resultAcc [{messageId: entry.messageId, cancelled: cancelResult.cancelled}]) newFedAcc
5588+
54245589-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
54255590-- Call before stopping the server so peers learn keys went offline immediately.
54265591-- Gossip is sent via fork; allow a few seconds after response before process exit.
@@ -5633,14 +5798,17 @@ computeAndCacheHealth = \t -> do
56335798 || staleThresholdCycles == 0 || pss <= staleThresholdCycles
56345799 || (tmc == 0 && presenceCount == 0 && forwardCount == 0)
56355800 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
5801+ nonceCount <- countAllNonces 0
5802+ let nonceOk = nonceCount * 10 <= maxForwardNonces * 9
56365803 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
56375804 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
56385805 capacityOk: capacityOk,
56395806 federationOk: fedOk,
56405807 snapshotOk: snapOk,
56415808 watchdogOk: wdOk,
5809+ nonceOk: nonceOk,
56425810 isDraining: dp > 0}
5643− let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
5811+ let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk && result.nonceOk
56445812 atomic do
56455813 set *cachedHealthOk = allOk
56465814 set *cachedHealthPrune = result.pruneOk
@@ -5649,6 +5817,7 @@ computeAndCacheHealth = \t -> do
56495817 set *cachedHealthFederation = result.federationOk
56505818 set *cachedHealthSnapshot = result.snapshotOk
56515819 set *cachedHealthWatchdog = result.watchdogOk
5820+ set *cachedHealthNonce = result.nonceOk
56525821 yield result
56535822
56545823handleHealth = \authorization respond -> do
@@ -5667,13 +5836,14 @@ handleHealth = \authorization respond -> do
56675836 cF <- *cachedHealthFederation
56685837 cS <- *cachedHealthSnapshot
56695838 cW <- *cachedHealthWatchdog
5839+ cN <- *cachedHealthNonce
56705840 dp <- *drainPhase
5671− yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, draining: dp > 0}
5841+ yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, nonceHealthy: cN, draining: dp > 0}
56725842 yield (respond cached)
56735843 else do
56745844 isAuthed <- verifyMaintenanceAuth authorization
56755845 result <- computeAndCacheHealth t
5676− let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
5846+ let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk && result.nonceOk
56775847 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
56785848 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
56795849 -- Draining status is always visible (load balancers need it for routing decisions).
@@ -5684,6 +5854,7 @@ handleHealth = \authorization respond -> do
56845854 federationHealthy: result.federationOk,
56855855 snapshotHealthy: result.snapshotOk,
56865856 watchdogHealthy: result.watchdogOk,
5857+ nonceHealthy: result.nonceOk,
56875858 draining: result.isDraining})
56885859 else yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: True {},
56895860 heartbeatHealthy: True {},
@@ -5691,6 +5862,7 @@ handleHealth = \authorization respond -> do
56915862 federationHealthy: True {},
56925863 snapshotHealthy: True {},
56935864 watchdogHealthy: True {},
5865+ nonceHealthy: True {},
56945866 draining: result.isDraining})
56955867
56965868zeroStats = {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}}
@@ -5913,7 +6085,7 @@ handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData de
59136085 -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers,
59146086 -- version 3 = pre-ForwardEntry changes, version 4 = pre-allowedPeers, version 5 = pre-checksum
59156087 -- (accept all for backwards compatibility; missing fields default to empty list or 0)
5916− if version != 0 && version != 1 && version != 2 && version != 3 && version != 4 && version != 5 && version != 6 && version != 7 && version != 8 && version != snapshotVersion then do
6088+ if version != 0 && version != 1 && version != 2 && version != 3 && version != 4 && version != 5 && version != 6 && version != 7 && version != 8 && version != 9 && version != snapshotVersion then do
59176089 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
59186090 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
59196091 else do
@@ -6151,6 +6323,14 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
61516323 -- Create ack entries for federated messages so origin servers learn about the delete
61526324 let federatedDeleted = do m <- toDelete; where m.originServer != ""; yield m
61536325 let ackedFederated = federatedAckEntries federatedDeleted t
6326+ -- Queue delivery receipts for locally-originated deleted messages
6327+ let localDeleted = do m <- toDelete; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
6328+ when (count localDeleted > 0) (do
6329+ sd <- *senderDeliveries
6330+ let combinedDel = union sd localDeleted
6331+ set *senderDeliveries = if count combinedDel > maxSenderDeliveries
6332+ then evictOneOldest combinedDel (\d -> d.deliveredAt)
6333+ else combinedDel)
61546334 if count ackedFederated > 0 then do
61556335 ackedBkt <- readAckedBucket b
61566336 let combinedAcked = union ackedBkt ackedFederated
@@ -6659,6 +6839,9 @@ handleFlushNonces = \authorization respond -> do
66596839handleRotateSecret = \newSecret authorization respond -> do
66606840 authorized <- verifyMaintenanceAuth authorization
66616841 if not authorized then yield (respond {ok: False {}, error: "unauthorized"})
6842+ else if newSecret != "" && federationEnabled && not (tooLong newSecret 15) then do
6843+ logError "rotate-secret: rejected — secret must be at least 16 characters when federation is enabled"
6844+ yield (respond {ok: False {}, error: "secret_too_short"})
66626845 else if newSecret != "" && not (tooLong newSecret 31) then do
66636846 logWarn "rotate-secret: new secret is shorter than 32 characters — 32+ recommended for production"
66646847 atomic do set *runtimeMaintenanceSecret = newSecret
@@ -6724,12 +6907,14 @@ requestPath = \req -> case req of
67246907 ListMessages {} -> "POST /messages/list"
67256908 DeleteMessages {} -> "POST /messages/delete"
67266909 CancelMessage {} -> "POST /messages/cancel"
6910+ CancelBatch {} -> "POST /messages/cancel-batch"
67276911 LookupPresence {} -> "GET /lookup"
67286912 BatchLookupPresence {} -> "POST /lookup/batch"
67296913 RecvGossip {} -> "POST /federation/gossip"
67306914 ForwardMessage {} -> "POST /federation/forward"
67316915 FederationAck {} -> "POST /federation/ack"
67326916 FederationCancel {} -> "POST /federation/cancel"
6917+ FederationBatchCancel {} -> "POST /federation/cancel-batch"
67336918 VerifyToken {} -> "POST /federation/verify"
67346919 Prune {} -> "POST /maintenance/prune"
67356920 Drain {} -> "POST /maintenance/drain"
@@ -6775,6 +6960,7 @@ requestClientIp = \req -> case req of
67756960 ListMessages {clientIp} -> clientIp
67766961 DeleteMessages {clientIp} -> clientIp
67776962 CancelMessage {clientIp} -> clientIp
6963+ CancelBatch {clientIp} -> clientIp
67786964 LookupPresence {clientIp} -> clientIp
67796965 BatchLookupPresence {clientIp} -> clientIp
67806966 VerifyToken {clientIp} -> clientIp
@@ -6782,6 +6968,7 @@ requestClientIp = \req -> case req of
67826968 ForwardMessage {} -> ""
67836969 FederationAck {} -> ""
67846970 FederationCancel {} -> ""
6971+ FederationBatchCancel {} -> ""
67856972 Prune {} -> ""
67866973 Drain {} -> ""
67876974 Shutdown {} -> ""
@@ -6829,12 +7016,14 @@ serveInner = \req -> case req of
68297016 ListMessages {offset, authorization, clientIp, respond} -> handleListMessages offset authorization clientIp respond
68307017 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
68317018 CancelMessage {messageId, to, authorization, clientIp, respond} -> handleCancelMessage messageId to authorization clientIp respond
7019+ CancelBatch {cancels, authorization, clientIp, respond} -> handleCancelBatch cancels authorization clientIp respond
68327020 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
68337021 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
68347022 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
6835− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId authorization respond
7023+ ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, expiresAt, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt authorization respond
68367024 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
68377025 FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationCancel messageId toKey fromServer requestTimestamp correlationId authorization respond
7026+ FederationBatchCancel {cancels, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationBatchCancel cancels fromServer requestTimestamp correlationId authorization respond
68387027 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
68397028 Prune {authorization, respond} -> handlePrune authorization respond
68407029 Drain {phase, authorization, respond} -> handleDrain phase authorization respond
@@ -6880,7 +7069,10 @@ serve = \req -> do
68807069 let ip = requestClientIp req
68817070 let ipSuffix = if ip != "" then " " ++ ip else ""
68827071 if path != ""
6883− then logInfo ("[ACCESS] " ++ reqId ++ " " ++ path ++ " " ++ show latency ++ "ms" ++ ipSuffix)
7072+ then do
7073+ logInfo ("[ACCESS] " ++ reqId ++ " " ++ path ++ " " ++ show latency ++ "ms" ++ ipSuffix)
7074+ when (latency > slowRequestThreshold)
7075+ (logEvent "warn" "slow_request" (path ++ " " ++ show latency ++ "ms"))
68847076 else when (latency > slowRequestThreshold) (do
68857077 logInfo ("[SLOW] " ++ reqId ++ " poll " ++ show latency ++ "ms" ++ ipSuffix)
68867078 logEvent "warn" "slow_request" ("poll " ++ show latency ++ "ms"))
@@ -7045,23 +7237,28 @@ backgroundSnapshot = do
70457237 lastFp <- *lastSnapshotFingerprint
70467238 let stateChanged = lastFp == 0 || fingerprint != lastFp
70477239 when (hasState && stateChanged && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do
7048− exportData <- atomic do
7049− p <- readAllPresence
7050− a <- readAllAcked
7051− ao <- *ackOverflow
7052− d <- readAllDelivered
7053− ff <- *forwardFailures
7054− f <- readAllForwards
7055− fn <- readAllNonces
7240+ -- Split into per-collection reads to reduce contention (auto-export is best-effort;
7241+ -- slight inconsistency between collections is acceptable vs. one large atomic block
7242+ -- holding all 80+ bucket locks and competing with hot-path operations).
7243+ p <- atomic do pp <- readAllPresence; yield pp
7244+ a <- atomic do aa <- readAllAcked; yield aa
7245+ ao <- atomic do aao <- *ackOverflow; yield aao
7246+ d <- atomic do dd <- readAllDelivered; yield dd
7247+ ff <- atomic do fff <- *forwardFailures; yield fff
7248+ f <- atomic do ff2 <- readAllForwards; yield ff2
7249+ fn <- atomic do fnn <- readAllNonces; yield fnn
7250+ extraState <- atomic do
70567251 bp <- *blockedPeers
70577252 ap <- *allowedPeers
70587253 pac <- *pendingAckCleanups
70597254 se <- *senderExpirations
70607255 sdl <- *senderDeliveries
7061− let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
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}
7256+ yield {bp: bp, ap: ap, pac: pac, se: se, sdl: sdl}
7257+ let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
7258+ 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}
70637259 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
70647260 println "[SNAPSHOT_END]"
7261+ logEvent "warn" "snapshot_unencrypted" "snapshot exported to stdout — ensure log pipeline is secured"
70657262 atomic do
70667263 set *lastAutoSnapshotExportAt = t
70677264 set *lastSnapshotFingerprint = fingerprint
@@ -7228,8 +7425,9 @@ main = do
72287425 logInfo (" federation allowlist: " ++ show maxAllowedPeers ++ " max (empty = open federation)")
72297426 when (autoSnapshotInterval > 0)
72307427 (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
7231− when (autoSnapshotExportInterval > 0)
7232− (logInfo (" auto-snapshot export: " ++ show autoSnapshotExportInterval ++ "ms (full [SNAPSHOT_DATA] to stdout)"))
7428+ when (autoSnapshotExportInterval > 0) (do
7429+ logInfo (" auto-snapshot export: " ++ show autoSnapshotExportInterval ++ "ms (full [SNAPSHOT_DATA] to stdout)")
7430+ logWarn "SECURITY: auto-snapshot exports unencrypted federation state to stdout — ensure log pipeline is secured and access-restricted")
72337431 fork backgroundPrune
72347432 fork pollHeartbeat
72357433 fork backgroundWatchdog