e5cfebc · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Remove message TTL feature and simplify message expiration to always use 7-day default

e5cfebciko6 months ago

server.knot+34−69

compare with master

unifiedsplit

@@ -39,10 +39,6 @@ maxPresenceTtl : Int<S>

39maxPresenceTtl = 3600 -- max presence TTL from gossip

40maxPubkeyLen = 256

41maxBlobLen = 65536

42−-- Minimum message TTL (sender-specified). Prevents very short TTLs that would expire

43−-- before the recipient polls, wasting mailbox slots and federation bandwidth.

44−minMessageTtl : Int<Ms>

45−minMessageTtl = 60000 -- 1 minute

46maxSignatureLen = 256

42maxSignatureLen = 256

47maxIpLen = 45 -- max length of an IP address string (IPv6 = 45 chars); used in matchIp instead of maxServerNameLen

43maxIpLen = 45 -- max length of an IP address string (IPv6 = 45 chars); used in matchIp instead of maxServerNameLen

48maxMessageIds = 100

44maxMessageIds = 100

@@ -339,7 +335,6 @@ type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp, ip:

339-- Authenticated session; token is a random hex string

335-- Authenticated session; token is a random hex string

340type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp, ip: Text, createdAt: Timestamp}

336type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp, ip: Text, createdAt: Timestamp}

341-- Stored message; originServer is "" for locally-originated messages, set to sender server for federated forwards.

337-- Stored message; originServer is "" for locally-originated messages, set to sender server for federated forwards.

342−-- expiresAt: computed from sender-specified TTL or default unackedMessageMaxAge.

343-- Messages with expiresAt <= now are pruned and excluded from polls.

338-- Messages with expiresAt <= now are pruned and excluded from polls.

344-- senderKey: pubkey of the authenticated sender (locally-originated only; "" for federation forwards).

339-- senderKey: pubkey of the authenticated sender (locally-originated only; "" for federation forwards).

345-- Used by /messages/cancel to verify the caller is the actual sender.

340-- Used by /messages/cancel to verify the caller is the actual sender.

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

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

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

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

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

349-- Queued federation forward: message waiting to be delivered to a remote server.

344-- Queued federation forward: message waiting to be delivered to a remote server.

350−-- messageTtl: remaining TTL in ms at time of enqueue (0 = default). Passed to the

351−-- receiving server so it can compute its own expiresAt.

352−type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int, messageTtl: Int}

345+type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int}

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

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

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

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

355-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)

348-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)

@@ -387,7 +380,7 @@ type UniqueKeyIpEntry = {ip: Text, keys: [PubkeyHex], windowStart: Timestamp}

387type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}

380type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}

388

381

389-- Batch send types

382-- Batch send types

390−type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, messageTtl: Int}

383+type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId}

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

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

392

385

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

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

@@ -800,10 +793,7 @@ route ClientApi where

800 = Poll

793 = Poll

801 /messages

794 /messages

802 -- Send an E2E encrypted message to a recipient pubkey.

795 -- Send an E2E encrypted message to a recipient pubkey.

803− -- messageTtl: optional TTL in milliseconds (0 = default unackedMessageMaxAge).

804− -- Clamped to [minMessageTtl, unackedMessageMaxAge]. Short TTLs for ephemeral data

805− -- (typing indicators, read receipts) free mailbox slots faster than the default 7 days.

806− POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, messageTtl: Int} / headers {authorization: Text, clientIp: Text}

796+ POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text, clientIp: Text}

807 -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int, error: Text}

797 -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int, error: Text}

808 = SendMessage

798 = SendMessage

809 -- Check delivery status of a sent message (caller must know recipient to prove ownership)

799 -- Check delivery status of a sent message (caller must know recipient to prove ownership)

@@ -853,14 +843,12 @@ route FederationApi where

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

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

854 = RecvGossip

844 = RecvGossip

855 -- Receive a forwarded message from a peer server on behalf of a sender.

845 -- Receive a forwarded message from a peer server on behalf of a sender.

856− -- messageTtl: remaining TTL in ms (0 = default). Allows short-lived messages to

857− -- expire on the destination server without waiting for the full unackedMessageMaxAge.

858 -- NOTE: senderKey is intentionally NOT included in the forward protocol. The sender's

846 -- NOTE: senderKey is intentionally NOT included in the forward protocol. The sender's

859 -- identity is inside the E2E encrypted blob — exposing it in plaintext would break the

847 -- identity is inside the E2E encrypted blob — exposing it in plaintext would break the

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

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

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

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

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

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

863− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, messageTtl: Int} /forward headers {authorization: Text}

851+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /forward headers {authorization: Text}

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

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

865 = ForwardMessage

853 = ForwardMessage

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

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

@@ -2499,13 +2487,6 @@ fetchFederation = \server request -> do

2499clampTtl : Int<S> -> Int<S>

2487clampTtl : Int<S> -> Int<S>

2500clampTtl = clamp 1 maxPresenceTtl

2488clampTtl = clamp 1 maxPresenceTtl

2501

2489

2502−-- Compute message expiresAt from a sender-specified TTL (in ms).

2503−-- 0 = default (unackedMessageMaxAge). Non-zero values are clamped to [minMessageTtl, unackedMessageMaxAge].

2504−computeMessageExpiresAt : Timestamp -> Int -> Timestamp

2505−computeMessageExpiresAt = \receivedAt messageTtl ->

2506− if messageTtl <= 0 then receivedAt + unackedMessageMaxAge

2507− else receivedAt + clamp minMessageTtl unackedMessageMaxAge messageTtl

2508−

2509clampTimestamp : Timestamp -> Timestamp -> Timestamp

2490clampTimestamp : Timestamp -> Timestamp -> Timestamp

2510clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp

2491clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp

2511

2492

@@ -3370,7 +3351,7 @@ retryForwardsForServer = \entries remaining ->

3370 toKey: f.toKey, encryptedBlob: f.encryptedBlob,

3351 toKey: f.toKey, encryptedBlob: f.encryptedBlob,

3371 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,

3352 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,

3372 fromServer: serverHostN, requestTimestamp: t,

3353 fromServer: serverHostN, requestTimestamp: t,

3373− authorization: auth, correlationId: corrId, messageTtl: f.messageTtl})

3354+ authorization: auth, correlationId: corrId})

3374 case result of

3355 case result of

3375 Err e -> do

3356 Err e -> do

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

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

@@ -3961,10 +3942,9 @@ forwardCandidates = \targets messages forwards t ->

3961 where m.toKey == target.pubkey

3942 where m.toKey == target.pubkey

3962 where none serverFwdIds (\fid -> fid == m.messageId)

3943 where none serverFwdIds (\fid -> fid == m.messageId)

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

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

3964− let remainingTtl = if m.expiresAt > 0 && m.expiresAt > t then max 0 (m.expiresAt - t) else 0

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

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

3966 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,

3946 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,

3967− originServer: origin, retries: 0, messageTtl: remainingTtl})) [] targets

3947+ originServer: origin, retries: 0})) [] targets

3968

3948

3969

3949

3970challengeMatch = \challenges pubkey challenge ip t ->

3950challengeMatch = \challenges pubkey challenge ip t ->

@@ -4278,9 +4258,8 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do

4278-- Core send logic: dedup, rate limiting, store message, create forwards.

4258-- Core send logic: dedup, rate limiting, store message, create forwards.

4279-- Called inside an atomic block by both handleSendMessage and sendBatchInner.

4259-- Called inside an atomic block by both handleSendMessage and sendBatchInner.

4280-- Caller handles auth, drain checks, and messageId generation.

4260-- Caller handles auth, drain checks, and messageId generation.

4281−-- messageTtl: sender-specified TTL in ms (0 = default unackedMessageMaxAge).

4282−sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTtl t ->

4261+sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->

4283− if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}

4262+ if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}

4284 else do

4263 else do

4285 let msgBucket = pubkeyBucket to

4264 let msgBucket = pubkeyBucket to

4286 bucket <- readMessageBucket msgBucket

4265 bucket <- readMessageBucket msgBucket

@@ -4291,37 +4270,37 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTt

4291 -- expire but before the nonce expires.

4270 -- expire but before the nonce expires.

4292 nonceBkt <- readNonceBucket msgBucket

4271 nonceBkt <- readNonceBucket msgBucket

4293 let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt || any nonceBkt (\n -> n.messageId == effectiveMsgId && n.toKey == to)

4272 let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt || any nonceBkt (\n -> n.messageId == effectiveMsgId && n.toKey == to)

4294− if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}

4273+ if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}

4295 else do

4274 else do

4296 sendRates <- *sendRates

4275 sendRates <- *sendRates

4297 sendRate <- effectiveSendRate

4276 sendRate <- effectiveSendRate

4298 let rateResult = checkPubkeyRate sendRates senderKey sendRate sendRateWindowMs maxSendRateEntries t

4277 let rateResult = checkPubkeyRate sendRates senderKey sendRate sendRateWindowMs maxSendRateEntries t

4299− if rateResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: rateResult.retryAfterMs, msgExpiresAt: 0}

4278+ if rateResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: rateResult.retryAfterMs}

4300 else do

4279 else do

4301 set *sendRates = rateResult.entries

4280 set *sendRates = rateResult.entries

4302 pairRates <- *pairRates

4281 pairRates <- *pairRates

4303 let pairResult = checkPairRate pairRates senderKey to maxPairRatePerPair pairRateWindowMs maxPairRateEntries t

4282 let pairResult = checkPairRate pairRates senderKey to maxPairRatePerPair pairRateWindowMs maxPairRateEntries t

4304 set *pairRates = pairResult.entries

4283 set *pairRates = pairResult.entries

4305− if pairResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: pairResult.retryAfterMs, msgExpiresAt: 0}

4284+ if pairResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: pairResult.retryAfterMs}

4306 else do

4285 else do

4307 recvRates <- *recvRates

4286 recvRates <- *recvRates

4308 recvRate <- effectiveRecvRate

4287 recvRate <- effectiveRecvRate

4309 let recvResult = checkPubkeyRate recvRates to recvRate recvRateWindowMs maxRecvRateEntries t

4288 let recvResult = checkPubkeyRate recvRates to recvRate recvRateWindowMs maxRecvRateEntries t

4310− if recvResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: recvResult.retryAfterMs, msgExpiresAt: 0}

4289+ if recvResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: recvResult.retryAfterMs}

4311 else do

4290 else do

4312 set *recvRates = recvResult.entries

4291 set *recvRates = recvResult.entries

4313 totalMsgCount <- *totalMessageCount

4292 totalMsgCount <- *totalMessageCount

4314− if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}

4293+ if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}

4315 else do

4294 else do

4316 recipientCounts <- readRcBucket msgBucket

4295 recipientCounts <- readRcBucket msgBucket

4317 let recipientCount = getRecipientCount to recipientCounts

4296 let recipientCount = getRecipientCount to recipientCounts

4318− if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}

4297+ if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}

4319 else do

4298 else do

4320 recipientPresence <- readPresenceBucket (pubkeyBucket to)

4299 recipientPresence <- readPresenceBucket (pubkeyBucket to)

4321 seqCounter <- *seqCounter

4300 seqCounter <- *seqCounter

4322 let seqTs = nextSeqTs t seqCounter

4301 let seqTs = nextSeqTs t seqCounter

4323 set *seqCounter = seqTs

4302 set *seqCounter = seqTs

4324− let msgExpiresAt = computeMessageExpiresAt seqTs messageTtl

4303+ let msgExpiresAt = seqTs + unackedMessageMaxAge

4325 writeMessageBucket msgBucket (union bucket

4304 writeMessageBucket msgBucket (union bucket

4326 [{messageId: effectiveMsgId, toKey: to,

4305 [{messageId: effectiveMsgId, toKey: to,

4327 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt, senderKey: senderKey}])

4306 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt, senderKey: senderKey}])

@@ -4345,8 +4324,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTt

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

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

4346 let fbi = pubkeyBucket to

4325 let fbi = pubkeyBucket to

4347 fwdBkt <- readFwdBucket fbi

4326 fwdBkt <- readFwdBucket fbi

4348− let remainingTtl = if messageTtl > 0 then max 0 (msgExpiresAt - t) else 0

4349− 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, messageTtl: remainingTtl}

4327+ 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}

4350 totalFwdCount <- countAllForwards 0

4328 totalFwdCount <- countAllForwards 0

4351 let globalCap = maxForwardsEntries - totalFwdCount

4329 let globalCap = maxForwardsEntries - totalFwdCount

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

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

@@ -4360,12 +4338,10 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTt

4360 let status = if localOnline then MsDelivered {}

4338 let status = if localOnline then MsDelivered {}

4361 else if federationEnabled && forwardsEnqueued then MsFederated {}

4339 else if federationEnabled && forwardsEnqueued then MsFederated {}

4362 else MsQueued {}

4340 else MsQueued {}

4363− yield {status: status, remoteServers: remoteServers, retryAfterMs: 0, msgExpiresAt: msgExpiresAt}

4341+ yield {status: status, remoteServers: remoteServers, retryAfterMs: 0}

4364

4342

4365-- Fork federation forwards for messages that need to be delivered to remote servers.

4343-- Fork federation forwards for messages that need to be delivered to remote servers.

4366−-- msgExpiresAt: the stored message's expiration timestamp (used to compute remaining TTL

4367−-- at forward time, ensuring the remote server gets the correct remaining lifetime).

4368−forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs msgExpiresAt remoteServers ->

4344+forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->

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

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

4370 fork (sendToEach remoteServers (\server -> do

4346 fork (sendToEach remoteServers (\server -> do

4371 stillPending <- atomic do

4347 stillPending <- atomic do

@@ -4375,15 +4351,11 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs msgExpiresAt remot

4375 fwdT <- now

4351 fwdT <- now

4376 fwdAuth <- generateFederationAuth "forward" server

4352 fwdAuth <- generateFederationAuth "forward" server

4377 fwdCorrId <- genTokenN 1

4353 fwdCorrId <- genTokenN 1

4378− -- Compute remaining TTL from the stored message's expiration (not the original TTL).

4379− -- This ensures the remote server gets the correct remaining lifetime, even if

4380− -- the fork is delayed by network issues or server load.

4381− let fwdTtl = if msgExpiresAt > 0 then max 0 (msgExpiresAt - fwdT) else 0

4382 fwdResult <- fetchFederation server (ForwardMessage {

4354 fwdResult <- fetchFederation server (ForwardMessage {

4383 toKey: to, encryptedBlob: encryptedBlob,

4355 toKey: to, encryptedBlob: encryptedBlob,

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

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

4385 fromServer: serverHostN, requestTimestamp: fwdT,

4357 fromServer: serverHostN, requestTimestamp: fwdT,

4386− authorization: fwdAuth, correlationId: fwdCorrId, messageTtl: fwdTtl})

4358+ authorization: fwdAuth, correlationId: fwdCorrId})

4387 case fwdResult of

4359 case fwdResult of

4388 Err e -> do

4360 Err e -> do

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

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

@@ -4409,13 +4381,11 @@ sendBatchForwardsSequential = \msgs server ->

4409 fwdT <- now

4381 fwdT <- now

4410 fwdAuth <- generateFederationAuth "forward" server

4382 fwdAuth <- generateFederationAuth "forward" server

4411 fwdCorrId <- genTokenN 1

4383 fwdCorrId <- genTokenN 1

4412− -- Use stored msgExpiresAt to compute remaining TTL (not original messageTtl)

4413− let fwdTtl = if r.msgExpiresAt > 0 then max 0 (r.msgExpiresAt - fwdT) else 0

4414 fwdResult <- fetchFederation server (ForwardMessage {

4384 fwdResult <- fetchFederation server (ForwardMessage {

4415 toKey: r.toKey, encryptedBlob: r.blob,

4385 toKey: r.toKey, encryptedBlob: r.blob,

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

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

4417 fromServer: serverHostN, requestTimestamp: fwdT,

4387 fromServer: serverHostN, requestTimestamp: fwdT,

4418− authorization: fwdAuth, correlationId: fwdCorrId, messageTtl: fwdTtl})

4388+ authorization: fwdAuth, correlationId: fwdCorrId})

4419 case fwdResult of

4389 case fwdResult of

4420 Err e -> do

4390 Err e -> do

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

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

@@ -4431,7 +4401,7 @@ sendBatchForwardsSequential = \msgs server ->

4431-- POST /messages: Send an encrypted message to a recipient.

4401-- POST /messages: Send an encrypted message to a recipient.

4432-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.

4402-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.

4433-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.

4403-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.

4434−handleSendMessage = \to encryptedBlob timestamp messageId messageTtl authorization clientIp respond -> do

4404+handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp respond -> do

4435 t <- now

4405 t <- now

4436 globalRateLimited <- checkGlobalRate t

4406 globalRateLimited <- checkGlobalRate t

4437 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0, error: "rate_limited"})

4407 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0, error: "rate_limited"})

@@ -4447,15 +4417,15 @@ handleSendMessage = \to encryptedBlob timestamp messageId messageTtl authorizati

4447 let clampedTs = clampTimestamp timestamp t

4417 let clampedTs = clampTimestamp timestamp t

4448 result <- atomic do

4418 result <- atomic do

4449 dp <- *drainPhase

4419 dp <- *drainPhase

4450− if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}

4420+ if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}

4451 else withSessionAuth authorization t clientIp

4421 else withSessionAuth authorization t clientIp

4452− (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0})

4422+ (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})

4453− (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId messageTtl t)

4423+ (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId t)

4454 case result.status of

4424 case result.status of

4455 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)

4425 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)

4456 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))

4426 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))

4457 _ -> yield {}

4427 _ -> yield {}

4458− forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.msgExpiresAt result.remoteServers

4428+ forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.remoteServers

4459 yield (case result.status of

4429 yield (case result.status of

4460 MsDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0, error: ""}

4430 MsDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0, error: ""}

4461 MsDelivered {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId, retryAfterMs: 0, error: ""}

4431 MsDelivered {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId, retryAfterMs: 0, error: ""}

@@ -4806,7 +4776,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation

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

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

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

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

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

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

4809−handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId messageTtl authorization respond -> do

4779+handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId authorization respond -> do

4810 t <- now

4780 t <- now

4811 let corrId = sanitizeCorrelationId correlationId

4781 let corrId = sanitizeCorrelationId correlationId

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

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

@@ -4890,12 +4860,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

4890 seqCounter <- *seqCounter

4860 seqCounter <- *seqCounter

4891 let seqTs = nextSeqTs t seqCounter

4861 let seqTs = nextSeqTs t seqCounter

4892 set *seqCounter = seqTs

4862 set *seqCounter = seqTs

4893− -- Federation forwards: use raw TTL without minMessageTtl clamp.

4894− -- The TTL represents remaining lifetime from the origin server, which

4895− -- may legitimately be less than minMessageTtl near expiry. Clamping

4896− -- to minMessageTtl would extend the message's life on each hop.

4897− let fwdExpiresAt = if messageTtl <= 0 then seqTs + unackedMessageMaxAge

4863+ let fwdExpiresAt = seqTs + unackedMessageMaxAge

4898− else seqTs + clamp 1 unackedMessageMaxAge messageTtl

4899 writeMessageBucket bkt (union bucket

4864 writeMessageBucket bkt (union bucket

4900 [{messageId: messageId, toKey: toKey,

4865 [{messageId: messageId, toKey: toKey,

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

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

@@ -5443,7 +5408,7 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD

5443 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})

5408 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})

5444 else do

5409 else do

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

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

5446− -- version 3 = pre-messageTtl in ForwardEntry, version 4 = pre-allowedPeers

5411+ -- version 3 = pre-ForwardEntry changes, version 4 = pre-allowedPeers

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

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

5448 if version != 0 && version != 1 && version != 2 && version != 3 && version != 4 && version != snapshotVersion then do

5413 if version != 0 && version != 1 && version != 2 && version != 3 && version != 4 && version != snapshotVersion then do

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

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

@@ -5781,18 +5746,18 @@ sendBatchInner = \senderKey messages t acc ->

5781 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)

5746 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)

5782 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))

5747 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))

5783 result <- if t - msg.timestamp > messageMaxAge

5748 result <- if t - msg.timestamp > messageMaxAge

5784− then yield {messageId: effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, msgExpiresAt: 0, retryAfterMs: 0}

5749+ then yield {messageId: effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, retryAfterMs: 0}

5785 else do

5750 else do

5786 let clampedTs = clampTimestamp msg.timestamp t

5751 let clampedTs = clampTimestamp msg.timestamp t

5787 sendResult <- atomic do

5752 sendResult <- atomic do

5788 dp <- *drainPhase

5753 dp <- *drainPhase

5789− if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}

5754+ if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}

5790− else sendMessageCore senderKey msg.to msg.encryptedBlob clampedTs effectiveMsgId msg.messageTtl t

5755+ else sendMessageCore senderKey msg.to msg.encryptedBlob clampedTs effectiveMsgId t

5791 case sendResult.status of

5756 case sendResult.status of

5792 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)

5757 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)

5793 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")

5758 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")

5794 _ -> yield {}

5759 _ -> yield {}

5795− yield {messageId: effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, msgExpiresAt: sendResult.msgExpiresAt, retryAfterMs: sendResult.retryAfterMs}

5760+ yield {messageId: effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryAfterMs}

5796 sendBatchInner senderKey split.rest t (union acc [result])

5761 sendBatchInner senderKey split.rest t (union acc [result])

5797

5762

5798-- POST /maintenance/block: Add a server to the federation blocklist.

5763-- POST /maintenance/block: Add a server to the federation blocklist.

@@ -6146,7 +6111,7 @@ serveInner = \req -> case req of

6146 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond

6111 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond

6147 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond

6112 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond

6148 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond

6113 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond

6149− SendMessage {to, encryptedBlob, timestamp, messageId, messageTtl, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId messageTtl authorization clientIp respond

6114+ SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond

6150 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond

6115 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond

6151 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond

6116 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond

6152 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond

6117 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond

@@ -6156,7 +6121,7 @@ serveInner = \req -> case req of

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

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

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

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

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

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

6159− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, messageTtl, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId messageTtl authorization respond

6124+ ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId authorization respond

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

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

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

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

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

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