05bf0ef · skrepka · git.iko.soy

git.iko.soy/skrepka

Implement 9 server improvements: forward TTL propagation, batch federation cancel, delivery receipts, health monitoring, rotate secret enforcement, security warnings, slow request logging, snapshot contention reduction

05bf0efiko6 months ago

server.knot+234−36

compare with master

unifiedsplit

@@ -165,7 +165,7 @@ federationLatencyWindowMs : Int<Ms>

165federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval

166

167-- 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

169

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.

@@ -376,7 +376,7 @@ type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHe

376-- Presence: tracks that a pubkey is online at a given server, with expiry

377type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}

378-- 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}

380-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin

381type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}

382-- 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

420type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, expiresInMs: Int}

421type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}

422

423+-- Batch cancel types

424+type BatchCancelEntry = {messageId: MessageIdHex, to: PubkeyHex}

425+type BatchCancelResult = {messageId: MessageIdHex, cancelled: Bool}

426+

423-- Sender expiration notification: tracks when a locally-originated message expired unread.

427-- Sender expiration notification: tracks when a locally-originated message expired unread.

424-- Pruned after messageMaxAge. Included in the sender's next poll response.

428-- Pruned after messageMaxAge. Included in the sender's next poll response.

425type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}

429type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}

@@ -840,6 +844,7 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe

840*cachedHealthFederation : Bool

844*cachedHealthFederation : Bool

841*cachedHealthSnapshot : Bool

845*cachedHealthSnapshot : Bool

842*cachedHealthWatchdog : Bool

846*cachedHealthWatchdog : Bool

847+*cachedHealthNonce : Bool

843

848

844-- ============================================

849-- ============================================

845-- Routes

850-- Routes

@@ -906,6 +911,11 @@ route ClientApi where

906 POST {messageId: MessageIdHex, to: PubkeyHex} /cancel headers {authorization: Text, clientIp: Text}

911 POST {messageId: MessageIdHex, to: PubkeyHex} /cancel headers {authorization: Text, clientIp: Text}

907 -> {ok: Bool, authorized: Bool, cancelled: Bool, error: Text}

912 -> {ok: Bool, authorized: Bool, cancelled: Bool, error: Text}

908 = CancelMessage

913 = 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

909 -- Look up which server a pubkey is present on (for federation routing)

919 -- Look up which server a pubkey is present on (for federation routing)

910 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}

920 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}

911 -> {server: AnyServerName, authorized: Bool, error: Text}

921 -> {server: AnyServerName, authorized: Bool, error: Text}

@@ -930,7 +940,7 @@ route FederationApi where

930 -- privacy model (the server never learns who sent a message). /messages/cancel for

940 -- privacy model (the server never learns who sent a message). /messages/cancel for

931 -- federated messages requires the client to include sender identity in the encrypted

941 -- federated messages requires the client to include sender identity in the encrypted

932 -- payload; the server cannot verify sender identity for federation-forwarded messages.

942 -- 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}

934 -> {ok: Bool, error: Text}

944 -> {ok: Bool, error: Text}

935 = ForwardMessage

945 = ForwardMessage

936 -- Acknowledge delivery of messages back to the origin server

946 -- Acknowledge delivery of messages back to the origin server

@@ -943,6 +953,10 @@ route FederationApi where

943 POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /cancel headers {authorization: Text}

953 POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /cancel headers {authorization: Text}

944 -> {ok: Bool, error: Text, cancelled: Bool}

954 -> {ok: Bool, error: Text, cancelled: Bool}

945 = FederationCancel

955 = 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

946

960

947-- Federation callback verification: the receiving server calls back to the sender

961-- Federation callback verification: the receiving server calls back to the sender

948-- to confirm a request token is genuine. No shared secret needed.

962-- to confirm a request token is genuine. No shared secret needed.

@@ -1053,7 +1067,7 @@ route MaintenanceApi where

1053 = RotateSecret

1067 = RotateSecret

1054

1068

1055route HealthApi where

1069route 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

1057

1071

1058-- Returns counts of in-memory collections (requires maintenance secret)

1072-- Returns counts of in-memory collections (requires maintenance secret)

1059-- TODO: If Knot adds raw text response support, add a /metrics endpoint

1073-- TODO: If Knot adds raw text response support, add a /metrics endpoint

@@ -3655,7 +3669,7 @@ incrementForwardRetries = \messageId toServer toKey -> atomic do

3655 tfc <- *totalForwardCount

3669 tfc <- *totalForwardCount

3656 set *totalForwardCount = max 0 (tfc - 1)

3670 set *totalForwardCount = max 0 (tfc - 1)

3657 logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")

3671 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))

3659

3673

3660retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}

3674retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}

3661retryForwardsForServer = \entries remaining ->

3675retryForwardsForServer = \entries remaining ->

@@ -3672,7 +3686,7 @@ retryForwardsForServer = \entries remaining ->

3672 toKey: f.toKey, encryptedBlob: f.encryptedBlob,

3686 toKey: f.toKey, encryptedBlob: f.encryptedBlob,

3673 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,

3687 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,

3674 fromServer: serverHostN, requestTimestamp: t,

3688 fromServer: serverHostN, requestTimestamp: t,

3675− authorization: auth, correlationId: corrId})

3689+ authorization: auth, correlationId: corrId, expiresAt: f.expiresAt})

3676 case result of

3690 case result of

3677 Err e -> do

3691 Err e -> do

3678 logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)

3692 logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)

@@ -4342,7 +4356,7 @@ forwardCandidates = \targets messages forwards t ->

4342 let origin = if m.originServer == "" then serverHostN else m.originServer

4356 let origin = if m.originServer == "" then serverHostN else m.originServer

4343 yield {messageId: m.messageId, toServer: target.server, createdAt: t,

4357 yield {messageId: m.messageId, toServer: target.server, createdAt: t,

4344 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,

4358 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,

4345− originServer: origin, retries: 0})) [] targets

4359+ originServer: origin, retries: 0, expiresAt: m.expiresAt})) [] targets

4346

4360

4347

4361

4348-- Challenge-IP binding rules:

4362-- Challenge-IP binding rules:

@@ -4745,7 +4759,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn

4745 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do

4759 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do

4746 let fbi = pubkeyBucket to

4760 let fbi = pubkeyBucket to

4747 fwdBkt <- readFwdBucket fbi

4761 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}

4749 totalFwdCount <- *totalForwardCount

4763 totalFwdCount <- *totalForwardCount

4750 let globalCap = maxForwardsEntries - totalFwdCount

4764 let globalCap = maxForwardsEntries - totalFwdCount

4751 -- Per-key cap: prevents one key from consuming disproportionate forward slots

4765 -- Per-key cap: prevents one key from consuming disproportionate forward slots

@@ -4766,10 +4780,12 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn

4766forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->

4780forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->

4767 when (count remoteServers > 0 && federationEnabled) (

4781 when (count remoteServers > 0 && federationEnabled) (

4768 fork (sendToEach remoteServers (\server -> do

4782 fork (sendToEach remoteServers (\server -> do

4769− stillPending <- atomic do

4783+ fwdInfo <- atomic do

4770 fwdBkt <- readFwdBucket (pubkeyBucket to)

4784 fwdBkt <- readFwdBucket (pubkeyBucket to)

4771− yield (any fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == server))

4785+ case headOf (do f <- fwdBkt; where f.messageId == effectiveMsgId && f.toServer == server; yield f) of

4772− when stillPending (do

4786+ Nothing {} -> yield {pending: False {}, fwdExpiresAt: 0}

4787+ Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt}

4788+ when fwdInfo.pending (do

4773 fwdT <- now

4789 fwdT <- now

4774 fwdAuth <- generateFederationAuth "forward" server

4790 fwdAuth <- generateFederationAuth "forward" server

4775 fwdCorrId <- genTokenN 1

4791 fwdCorrId <- genTokenN 1

@@ -4777,7 +4793,8 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->

4777 toKey: to, encryptedBlob: encryptedBlob,

4793 toKey: to, encryptedBlob: encryptedBlob,

4778 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,

4794 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,

4779 fromServer: serverHostN, requestTimestamp: fwdT,

4795 fromServer: serverHostN, requestTimestamp: fwdT,

4780− authorization: fwdAuth, correlationId: fwdCorrId})

4796+ authorization: fwdAuth, correlationId: fwdCorrId,

4797+ expiresAt: fwdInfo.fwdExpiresAt})

4781 case fwdResult of

4798 case fwdResult of

4782 Err e -> do

4799 Err e -> do

4783 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)

4800 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)

@@ -4796,10 +4813,12 @@ sendBatchForwardsSequential = \msgs server ->

4796 Nothing {} -> yield {}

4813 Nothing {} -> yield {}

4797 Just {value: r} -> do

4814 Just {value: r} -> do

4798 let rest = do m <- msgs; where m.messageId != r.messageId; yield m

4815 let rest = do m <- msgs; where m.messageId != r.messageId; yield m

4799− stillPending <- atomic do

4816+ batchFwdInfo <- atomic do

4800 fwdBkt <- readFwdBucket (pubkeyBucket r.toKey)

4817 fwdBkt <- readFwdBucket (pubkeyBucket r.toKey)

4801− yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))

4818+ case headOf (do f <- fwdBkt; where f.messageId == r.messageId && f.toServer == server; yield f) of

4802− when stillPending (do

4819+ Nothing {} -> yield {pending: False {}, fwdExpiresAt: 0}

4820+ Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt}

4821+ when batchFwdInfo.pending (do

4803 fwdT <- now

4822 fwdT <- now

4804 fwdAuth <- generateFederationAuth "forward" server

4823 fwdAuth <- generateFederationAuth "forward" server

4805 fwdCorrId <- genTokenN 1

4824 fwdCorrId <- genTokenN 1

@@ -4807,7 +4826,8 @@ sendBatchForwardsSequential = \msgs server ->

4807 toKey: r.toKey, encryptedBlob: r.blob,

4826 toKey: r.toKey, encryptedBlob: r.blob,

4808 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,

4827 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,

4809 fromServer: serverHostN, requestTimestamp: fwdT,

4828 fromServer: serverHostN, requestTimestamp: fwdT,

4810− authorization: fwdAuth, correlationId: fwdCorrId})

4829+ authorization: fwdAuth, correlationId: fwdCorrId,

4830+ expiresAt: batchFwdInfo.fwdExpiresAt})

4811 case fwdResult of

4831 case fwdResult of

4812 Err e -> do

4832 Err e -> do

4813 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)

4833 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)

@@ -5211,7 +5231,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation

5211-- (fromServer != originServer) where an intermediate server re-forwards because

5231-- (fromServer != originServer) where an intermediate server re-forwards because

5212-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored

5232-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored

5213-- twice) and forwardCandidates (won't re-forward to the same server).

5233-- 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

5215 t <- now

5235 t <- now

5216 let corrId = sanitizeCorrelationId correlationId

5236 let corrId = sanitizeCorrelationId correlationId

5217 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward " ++ messageId ++ " from " ++ fromServer))

5237 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward " ++ messageId ++ " from " ++ fromServer))

@@ -5295,7 +5315,9 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

5295 seqCounter <- *seqCounter

5315 seqCounter <- *seqCounter

5296 let seqTs = nextSeqTs t seqCounter

5316 let seqTs = nextSeqTs t seqCounter

5297 set *seqCounter = seqTs

5317 set *seqCounter = seqTs

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.

5298− let fwdExpiresAt = seqTs + unackedMessageMaxAge

5320+ let fwdExpiresAt = if expiresAt > 0 && expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge

5299 writeMessageBucket bkt (union bucket

5321 writeMessageBucket bkt (union bucket

5300 [{messageId: messageId, toKey: toKey,

5322 [{messageId: messageId, toKey: toKey,

5301 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: ""}])

5323 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: ""}])

@@ -5326,6 +5348,26 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

5326 ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}

5348 ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}

5327 _ -> respond {ok: True {}, error: ""}))

5349 _ -> respond {ok: True {}, error: ""}))

5328

5350

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+

5329-- POST /federation/ack: Peer server confirms delivery of messages we originated.

5371-- POST /federation/ack: Peer server confirms delivery of messages we originated.

5330-- Removes corresponding forward entries. Also cleans up the original message if the

5372-- Removes corresponding forward entries. Also cleans up the original message if the

5331-- recipient has no local presence and no other pending forwards for that message.

5373-- recipient has no local presence and no other pending forwards for that message.

@@ -5356,6 +5398,11 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre

5356 -- Prevents a malicious peer from resetting its backoff by padding ack requests

5398 -- Prevents a malicious peer from resetting its backoff by padding ack requests

5357 -- with known messageIds that no longer have forward entries.

5399 -- with known messageIds that no longer have forward entries.

5358 when (ackedCount > 0 && unmatched == 0) (clearFailure fromServer)

5400 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)

5359 -- Deferred origin message cleanup: only queue cleanup for messages that actually had

5406 -- Deferred origin message cleanup: only queue cleanup for messages that actually had

5360 -- matching forward entries. Messages acked without a forward entry are NOT cleaned up,

5407 -- matching forward entries. Messages acked without a forward entry are NOT cleaned up,

5361 -- preventing premature deletion when a peer acks messages it never actually received.

5408 -- preventing premature deletion when a peer acks messages it never actually received.

@@ -5421,6 +5468,124 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio

5421 (logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")"))

5468 (logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")"))

5422 yield (respond {ok: True {}, error: "", cancelled: result.cancelled}))

5469 yield (respond {ok: True {}, error: "", cancelled: result.cancelled}))

5423

5470

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+

5424-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).

5589-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).

5425-- Call before stopping the server so peers learn keys went offline immediately.

5590-- Call before stopping the server so peers learn keys went offline immediately.

5426-- Gossip is sent via fork; allow a few seconds after response before process exit.

5591-- Gossip is sent via fork; allow a few seconds after response before process exit.

@@ -5633,14 +5798,17 @@ computeAndCacheHealth = \t -> do

5633 || staleThresholdCycles == 0 || pss <= staleThresholdCycles

5798 || staleThresholdCycles == 0 || pss <= staleThresholdCycles

5634 || (tmc == 0 && presenceCount == 0 && forwardCount == 0)

5799 || (tmc == 0 && presenceCount == 0 && forwardCount == 0)

5635 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)

5800 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)

5801+ nonceCount <- countAllNonces 0

5802+ let nonceOk = nonceCount * 10 <= maxForwardNonces * 9

5636 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),

5803 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),

5637 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,

5804 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,

5638 capacityOk: capacityOk,

5805 capacityOk: capacityOk,

5639 federationOk: fedOk,

5806 federationOk: fedOk,

5640 snapshotOk: snapOk,

5807 snapshotOk: snapOk,

5641 watchdogOk: wdOk,

5808 watchdogOk: wdOk,

5809+ nonceOk: nonceOk,

5642 isDraining: dp > 0}

5810 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

5644 atomic do

5812 atomic do

5645 set *cachedHealthOk = allOk

5813 set *cachedHealthOk = allOk

5646 set *cachedHealthPrune = result.pruneOk

5814 set *cachedHealthPrune = result.pruneOk

@@ -5649,6 +5817,7 @@ computeAndCacheHealth = \t -> do

5649 set *cachedHealthFederation = result.federationOk

5817 set *cachedHealthFederation = result.federationOk

5650 set *cachedHealthSnapshot = result.snapshotOk

5818 set *cachedHealthSnapshot = result.snapshotOk

5651 set *cachedHealthWatchdog = result.watchdogOk

5819 set *cachedHealthWatchdog = result.watchdogOk

5820+ set *cachedHealthNonce = result.nonceOk

5652 yield result

5821 yield result

5653

5822

5654handleHealth = \authorization respond -> do

5823handleHealth = \authorization respond -> do

@@ -5667,13 +5836,14 @@ handleHealth = \authorization respond -> do

5667 cF <- *cachedHealthFederation

5836 cF <- *cachedHealthFederation

5668 cS <- *cachedHealthSnapshot

5837 cS <- *cachedHealthSnapshot

5669 cW <- *cachedHealthWatchdog

5838 cW <- *cachedHealthWatchdog

5839+ cN <- *cachedHealthNonce

5670 dp <- *drainPhase

5840 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}

5672 yield (respond cached)

5842 yield (respond cached)

5673 else do

5843 else do

5674 isAuthed <- verifyMaintenanceAuth authorization

5844 isAuthed <- verifyMaintenanceAuth authorization

5675 result <- computeAndCacheHealth t

5845 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

5677 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,

5847 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,

5678 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.

5848 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.

5679 -- Draining status is always visible (load balancers need it for routing decisions).

5849 -- Draining status is always visible (load balancers need it for routing decisions).

@@ -5684,6 +5854,7 @@ handleHealth = \authorization respond -> do

5684 federationHealthy: result.federationOk,

5854 federationHealthy: result.federationOk,

5685 snapshotHealthy: result.snapshotOk,

5855 snapshotHealthy: result.snapshotOk,

5686 watchdogHealthy: result.watchdogOk,

5856 watchdogHealthy: result.watchdogOk,

5857+ nonceHealthy: result.nonceOk,

5687 draining: result.isDraining})

5858 draining: result.isDraining})

5688 else yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: True {},

5859 else yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: True {},

5689 heartbeatHealthy: True {},

5860 heartbeatHealthy: True {},

@@ -5691,6 +5862,7 @@ handleHealth = \authorization respond -> do

5691 federationHealthy: True {},

5862 federationHealthy: True {},

5692 snapshotHealthy: True {},

5863 snapshotHealthy: True {},

5693 watchdogHealthy: True {},

5864 watchdogHealthy: True {},

5865+ nonceHealthy: True {},

5694 draining: result.isDraining})

5866 draining: result.isDraining})

5695

5867

5696zeroStats = {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}}

5868zeroStats = {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

5913 -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers,

6085 -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers,

5914 -- version 3 = pre-ForwardEntry changes, version 4 = pre-allowedPeers, version 5 = pre-checksum

6086 -- version 3 = pre-ForwardEntry changes, version 4 = pre-allowedPeers, version 5 = pre-checksum

5915 -- (accept all for backwards compatibility; missing fields default to empty list or 0)

6087 -- (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

5917 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")

6089 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")

5918 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})

6090 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})

5919 else do

6091 else do

@@ -6151,6 +6323,14 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do

6151 -- Create ack entries for federated messages so origin servers learn about the delete

6323 -- Create ack entries for federated messages so origin servers learn about the delete

6152 let federatedDeleted = do m <- toDelete; where m.originServer != ""; yield m

6324 let federatedDeleted = do m <- toDelete; where m.originServer != ""; yield m

6153 let ackedFederated = federatedAckEntries federatedDeleted t

6325 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)

6154 if count ackedFederated > 0 then do

6334 if count ackedFederated > 0 then do

6155 ackedBkt <- readAckedBucket b

6335 ackedBkt <- readAckedBucket b

6156 let combinedAcked = union ackedBkt ackedFederated

6336 let combinedAcked = union ackedBkt ackedFederated

@@ -6659,6 +6839,9 @@ handleFlushNonces = \authorization respond -> do

6659handleRotateSecret = \newSecret authorization respond -> do

6839handleRotateSecret = \newSecret authorization respond -> do

6660 authorized <- verifyMaintenanceAuth authorization

6840 authorized <- verifyMaintenanceAuth authorization

6661 if not authorized then yield (respond {ok: False {}, error: "unauthorized"})

6841 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"})

6662 else if newSecret != "" && not (tooLong newSecret 31) then do

6845 else if newSecret != "" && not (tooLong newSecret 31) then do

6663 logWarn "rotate-secret: new secret is shorter than 32 characters — 32+ recommended for production"

6846 logWarn "rotate-secret: new secret is shorter than 32 characters — 32+ recommended for production"

6664 atomic do set *runtimeMaintenanceSecret = newSecret

6847 atomic do set *runtimeMaintenanceSecret = newSecret

@@ -6724,12 +6907,14 @@ requestPath = \req -> case req of

6724 ListMessages {} -> "POST /messages/list"

6907 ListMessages {} -> "POST /messages/list"

6725 DeleteMessages {} -> "POST /messages/delete"

6908 DeleteMessages {} -> "POST /messages/delete"

6726 CancelMessage {} -> "POST /messages/cancel"

6909 CancelMessage {} -> "POST /messages/cancel"

6910+ CancelBatch {} -> "POST /messages/cancel-batch"

6727 LookupPresence {} -> "GET /lookup"

6911 LookupPresence {} -> "GET /lookup"

6728 BatchLookupPresence {} -> "POST /lookup/batch"

6912 BatchLookupPresence {} -> "POST /lookup/batch"

6729 RecvGossip {} -> "POST /federation/gossip"

6913 RecvGossip {} -> "POST /federation/gossip"

6730 ForwardMessage {} -> "POST /federation/forward"

6914 ForwardMessage {} -> "POST /federation/forward"

6731 FederationAck {} -> "POST /federation/ack"

6915 FederationAck {} -> "POST /federation/ack"

6732 FederationCancel {} -> "POST /federation/cancel"

6916 FederationCancel {} -> "POST /federation/cancel"

6917+ FederationBatchCancel {} -> "POST /federation/cancel-batch"

6733 VerifyToken {} -> "POST /federation/verify"

6918 VerifyToken {} -> "POST /federation/verify"

6734 Prune {} -> "POST /maintenance/prune"

6919 Prune {} -> "POST /maintenance/prune"

6735 Drain {} -> "POST /maintenance/drain"

6920 Drain {} -> "POST /maintenance/drain"

@@ -6775,6 +6960,7 @@ requestClientIp = \req -> case req of

6775 ListMessages {clientIp} -> clientIp

6960 ListMessages {clientIp} -> clientIp

6776 DeleteMessages {clientIp} -> clientIp

6961 DeleteMessages {clientIp} -> clientIp

6777 CancelMessage {clientIp} -> clientIp

6962 CancelMessage {clientIp} -> clientIp

6963+ CancelBatch {clientIp} -> clientIp

6778 LookupPresence {clientIp} -> clientIp

6964 LookupPresence {clientIp} -> clientIp

6779 BatchLookupPresence {clientIp} -> clientIp

6965 BatchLookupPresence {clientIp} -> clientIp

6780 VerifyToken {clientIp} -> clientIp

6966 VerifyToken {clientIp} -> clientIp

@@ -6782,6 +6968,7 @@ requestClientIp = \req -> case req of

6782 ForwardMessage {} -> ""

6968 ForwardMessage {} -> ""

6783 FederationAck {} -> ""

6969 FederationAck {} -> ""

6784 FederationCancel {} -> ""

6970 FederationCancel {} -> ""

6971+ FederationBatchCancel {} -> ""

6785 Prune {} -> ""

6972 Prune {} -> ""

6786 Drain {} -> ""

6973 Drain {} -> ""

6787 Shutdown {} -> ""

6974 Shutdown {} -> ""

@@ -6829,12 +7016,14 @@ serveInner = \req -> case req of

6829 ListMessages {offset, authorization, clientIp, respond} -> handleListMessages offset authorization clientIp respond

7016 ListMessages {offset, authorization, clientIp, respond} -> handleListMessages offset authorization clientIp respond

6830 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond

7017 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond

6831 CancelMessage {messageId, to, authorization, clientIp, respond} -> handleCancelMessage messageId to authorization clientIp respond

7018 CancelMessage {messageId, to, authorization, clientIp, respond} -> handleCancelMessage messageId to authorization clientIp respond

7019+ CancelBatch {cancels, authorization, clientIp, respond} -> handleCancelBatch cancels authorization clientIp respond

6832 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond

7020 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond

6833 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond

7021 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond

6834 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond

7022 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

6836 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond

7024 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond

6837 FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationCancel messageId toKey fromServer requestTimestamp correlationId authorization respond

7025 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

6838 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond

7027 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond

6839 Prune {authorization, respond} -> handlePrune authorization respond

7028 Prune {authorization, respond} -> handlePrune authorization respond

6840 Drain {phase, authorization, respond} -> handleDrain phase authorization respond

7029 Drain {phase, authorization, respond} -> handleDrain phase authorization respond

@@ -6880,7 +7069,10 @@ serve = \req -> do

6880 let ip = requestClientIp req

7069 let ip = requestClientIp req

6881 let ipSuffix = if ip != "" then " " ++ ip else ""

7070 let ipSuffix = if ip != "" then " " ++ ip else ""

6882 if path != ""

7071 if path != ""

7072+ then do

6883− then logInfo ("[ACCESS] " ++ reqId ++ " " ++ path ++ " " ++ show latency ++ "ms" ++ ipSuffix)

7073+ logInfo ("[ACCESS] " ++ reqId ++ " " ++ path ++ " " ++ show latency ++ "ms" ++ ipSuffix)

7074+ when (latency > slowRequestThreshold)

7075+ (logEvent "warn" "slow_request" (path ++ " " ++ show latency ++ "ms"))

6884 else when (latency > slowRequestThreshold) (do

7076 else when (latency > slowRequestThreshold) (do

6885 logInfo ("[SLOW] " ++ reqId ++ " poll " ++ show latency ++ "ms" ++ ipSuffix)

7077 logInfo ("[SLOW] " ++ reqId ++ " poll " ++ show latency ++ "ms" ++ ipSuffix)

6886 logEvent "warn" "slow_request" ("poll " ++ show latency ++ "ms"))

7078 logEvent "warn" "slow_request" ("poll " ++ show latency ++ "ms"))

@@ -7045,23 +7237,28 @@ backgroundSnapshot = do

7045 lastFp <- *lastSnapshotFingerprint

7237 lastFp <- *lastSnapshotFingerprint

7046 let stateChanged = lastFp == 0 || fingerprint != lastFp

7238 let stateChanged = lastFp == 0 || fingerprint != lastFp

7047 when (hasState && stateChanged && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do

7239 when (hasState && stateChanged && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do

7048− exportData <- atomic do

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).

7049− p <- readAllPresence

7243+ p <- atomic do pp <- readAllPresence; yield pp

7050− a <- readAllAcked

7244+ a <- atomic do aa <- readAllAcked; yield aa

7051− ao <- *ackOverflow

7245+ ao <- atomic do aao <- *ackOverflow; yield aao

7052− d <- readAllDelivered

7246+ d <- atomic do dd <- readAllDelivered; yield dd

7053− ff <- *forwardFailures

7247+ ff <- atomic do fff <- *forwardFailures; yield fff

7054− f <- readAllForwards

7248+ f <- atomic do ff2 <- readAllForwards; yield ff2

7055− fn <- readAllNonces

7249+ fn <- atomic do fnn <- readAllNonces; yield fnn

7250+ extraState <- atomic do

7056 bp <- *blockedPeers

7251 bp <- *blockedPeers

7057 ap <- *allowedPeers

7252 ap <- *allowedPeers

7058 pac <- *pendingAckCleanups

7253 pac <- *pendingAckCleanups

7059 se <- *senderExpirations

7254 se <- *senderExpirations

7060 sdl <- *senderDeliveries

7255 sdl <- *senderDeliveries

7061− let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)

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)

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}

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}

7063 println ("[SNAPSHOT_DATA] " ++ toJson exportData)

7259 println ("[SNAPSHOT_DATA] " ++ toJson exportData)

7064 println "[SNAPSHOT_END]"

7260 println "[SNAPSHOT_END]"

7261+ logEvent "warn" "snapshot_unencrypted" "snapshot exported to stdout — ensure log pipeline is secured"

7065 atomic do

7262 atomic do

7066 set *lastAutoSnapshotExportAt = t

7263 set *lastAutoSnapshotExportAt = t

7067 set *lastSnapshotFingerprint = fingerprint

7264 set *lastSnapshotFingerprint = fingerprint

@@ -7228,8 +7425,9 @@ main = do

7228 logInfo (" federation allowlist: " ++ show maxAllowedPeers ++ " max (empty = open federation)")

7425 logInfo (" federation allowlist: " ++ show maxAllowedPeers ++ " max (empty = open federation)")

7229 when (autoSnapshotInterval > 0)

7426 when (autoSnapshotInterval > 0)

7230 (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))

7427 (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))

7231− when (autoSnapshotExportInterval > 0)

7428+ when (autoSnapshotExportInterval > 0) (do

7232− (logInfo (" auto-snapshot export: " ++ show autoSnapshotExportInterval ++ "ms (full [SNAPSHOT_DATA] to stdout)"))

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")

7233 fork backgroundPrune

7431 fork backgroundPrune

7234 fork pollHeartbeat

7432 fork pollHeartbeat

7235 fork backgroundWatchdog

7433 fork backgroundWatchdog