@@ -46,7 +46,9 @@ federationScheme = "https://"
46-- The Ed25519 federation keypair lives in `*federationPrivateKeyHex` /
46-- The Ed25519 federation keypair lives in `*federationPrivateKeyHex` /
47-- `*federationPublicKeyHex` refs and is generated on first startup; see `main`.
47-- `*federationPublicKeyHex` refs and is generated on first startup; see `main`.
49+-- Storage retention for messages (and clamp on client-supplied wall-clock
50+-- timestamps + dedup-id retention). Stored messages live this long; sends
49−-- Drop stored messages older than this (30 days).
51+-- whose `timestamp` is older than this are rejected (30 days).
50messageMaxAge : Int<Ms>
52messageMaxAge : Int<Ms>
51messageMaxAge = 2592000000
53messageMaxAge = 2592000000
52-- Upper bound on TTL a client may request for a presence entry (1 hour).
54-- Upper bound on TTL a client may request for a presence entry (1 hour).
@@ -203,12 +205,6 @@ maxPollsPerKey = 3
203presenceRefreshInterval : Int<Ms>
205presenceRefreshInterval : Int<Ms>
204presenceRefreshInterval = 300000
206presenceRefreshInterval = 300000
206−-- Drop unacked messages older than this (7 days).
207−unackedMessageMaxAge : Int<Ms>
208−unackedMessageMaxAge = 604800000
209−-- Floor on the TTL a client may request when sending a message (1 min).
210−minMessageTtl : Int<Ms>
211−minMessageTtl = 60000
212-- Window during which a delivered-but-unacked message is retained for re-delivery (2 min).
208-- Window during which a delivered-but-unacked message is retained for re-delivery (2 min).
213ackGracePeriod : Int<Ms>
209ackGracePeriod : Int<Ms>
214ackGracePeriod = 120000
210ackGracePeriod = 120000
@@ -246,7 +242,7 @@ sToMs = \s -> stripUnit s * 1000
246-- delivery metadata for the recipient; the server cannot read the blob.
242-- delivery metadata for the recipient; the server cannot read the blob.
247type PollEvent = {
243type PollEvent = {
248 id: MessageIdHex,
244 id: MessageIdHex,
249− payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex, expiresAt: Timestamp}
245+ payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex}
252-- Kind of presence event flowing through federation gossip. `GossipUnknown` is
248-- Kind of presence event flowing through federation gossip. `GossipUnknown` is
@@ -340,13 +336,13 @@ type Session = {
341-- Mailbox row: a stored message awaiting recipient pickup or federation forward.
337-- Mailbox row: a stored message awaiting recipient pickup or federation forward.
342-- Identity is `messageIdOf encryptedBlob` (BLAKE3 of the ciphertext); we don't
338-- Identity is `messageIdOf encryptedBlob` (BLAKE3 of the ciphertext); we don't
343−-- persist it because it's deterministically derivable.
339+-- persist it because it's deterministically derivable. Storage TTL is fixed
340+-- at `messageMaxAge`, so expiry is derived from `receivedAt`.
344type Message = {
341type Message = {
345 toKey: PubkeyHex,
342 toKey: PubkeyHex,
346 encryptedBlob: BlobHex,
343 encryptedBlob: BlobHex,
347 timestamp: Timestamp,
344 timestamp: Timestamp,
348− receivedAt: Timestamp,
345+ receivedAt: Timestamp
349− expiresAt: Timestamp
351-- Last-known location of a pubkey, either local (`server == serverHostN`) or
347-- Last-known location of a pubkey, either local (`server == serverHostN`) or
352-- learned from a peer via gossip. Multiple entries per pubkey are allowed.
348-- learned from a peer via gossip. Multiple entries per pubkey are allowed.
@@ -365,8 +361,7 @@ type ForwardEntry = {
365 toKey: PubkeyHex,
361 toKey: PubkeyHex,
366 encryptedBlob: BlobHex,
362 encryptedBlob: BlobHex,
367 timestamp: Timestamp,
363 timestamp: Timestamp,
368− retries: Int,
364+ retries: Int
369− expiresAt: Timestamp
371-- Replay-protection record: a message id we already delivered locally.
366-- Replay-protection record: a message id we already delivered locally.
372type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
367type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
@@ -379,7 +374,7 @@ type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
379type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
374type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
381-- One entry of a /messages/batch request body.
376-- One entry of a /messages/batch request body.
382−type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, expiresInMs: Int}
377+type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
383-- Per-entry result returned by /messages/batch.
378-- Per-entry result returned by /messages/batch.
384type BatchResult = {messageId: MessageIdHex, status: MessageStatus}
379type BatchResult = {messageId: MessageIdHex, status: MessageStatus}
@@ -457,7 +452,7 @@ route AuthApi where
458route ClientApi where
453route ClientApi where
459 POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int} = Poll
454 POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int} = Poll
460− POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, expiresInMs: Int} /messages headers {authorization: Text, clientIp: Text} -> {status: MessageStatus, messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int} = SendMessage
455+ POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp} /messages headers {authorization: Text, clientIp: Text} -> {status: MessageStatus, messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int} = SendMessage
461 POST {to: PubkeyHex, messageId: MessageIdHex} /messages/status headers {authorization: Text, clientIp: Text} -> {status: Text, authorized: Bool} = MessageStatus
456 POST {to: PubkeyHex, messageId: MessageIdHex} /messages/status headers {authorization: Text, clientIp: Text} -> {status: Text, authorized: Bool} = MessageStatus
462 POST {messageIds: [MessageIdHex]} /messages/ack headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, ackedCount: Int, error: Text} = AckMessages
457 POST {messageIds: [MessageIdHex]} /messages/ack headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, ackedCount: Int, error: Text} = AckMessages
463 POST {messages: [BatchMessage]} /messages/batch headers {authorization: Text, clientIp: Text} -> {results: [BatchResult], authorized: Bool, retryAfterMs: Int} = SendBatch
458 POST {messages: [BatchMessage]} /messages/batch headers {authorization: Text, clientIp: Text} -> {results: [BatchResult], authorized: Bool, retryAfterMs: Int} = SendBatch
@@ -468,8 +463,8 @@ route ClientApi where
469route FederationApi where
464route FederationApi where
470 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/gossip headers {authorization: Text} -> {ok: Bool, error: Text, retryAfterMs: Int} = RecvGossip
465 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/gossip headers {authorization: Text} -> {ok: Bool, error: Text, retryAfterMs: Int} = RecvGossip
471− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, expiresAt: Timestamp} /federation/forward headers {authorization: Text} -> {ok: Bool, error: Text, retryAfterMs: Int} = ForwardMessage
466+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/forward headers {authorization: Text} -> {ok: Bool, error: Text, retryAfterMs: Int} = ForwardMessage
472− POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: MessageIdHex, expiresAt: Timestamp}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/forward-batch headers {authorization: Text} -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} = FederationBatchForward
467+ POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: MessageIdHex}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/forward-batch headers {authorization: Text} -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} = FederationBatchForward
474route FederationPubkeyApi where
469route FederationPubkeyApi where
475 GET /federation/pubkey -> {pubkey: Text, ok: Bool} = FederationPubkey
470 GET /federation/pubkey -> {pubkey: Text, ok: Bool} = FederationPubkey
@@ -602,7 +597,7 @@ pct : Int -> Int -> Int
602pct = \cur maxN -> if maxN > 0 then cur * 100 / maxN else 0
597pct = \cur maxN -> if maxN > 0 then cur * 100 / maxN else 0
604-- Common ack/delete tail: pull matching messages, drop them and their forwards.
599-- Common ack/delete tail: pull matching messages, drop them and their forwards.
605−removeForKey : (forall r. {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, expiresAt: Timestamp | r} -> Bool) -> IO {rw *forwards, rw *messages} [Message]
600+removeForKey : (forall r. {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp | r} -> Bool) -> IO {rw *forwards, rw *messages} [Message]
606removeForKey = \pred -> do
601removeForKey = \pred -> do
607 msgs <- *messages
602 msgs <- *messages
608 fwds <- *forwards
603 fwds <- *forwards
@@ -1016,13 +1011,6 @@ clampTtl = clamp 1 maxPresenceTtl
1016clampTimestamp : Timestamp -> Timestamp -> Timestamp
1011clampTimestamp : Timestamp -> Timestamp -> Timestamp
1017clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
1012clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
1019−-- Resolve the effective TTL for a stored message: 0/negative means default,
1020−-- otherwise clamp the requested value into [minMessageTtl, unackedMessageMaxAge].
1021−effectiveMessageTtl : Int<Ms> -> Int<Ms>
1022−effectiveMessageTtl = \e -> if e <= 0
1023−then unackedMessageMaxAge
1024−else clamp minMessageTtl unackedMessageMaxAge e
1026-- Strictly-monotonic next sequence timestamp: max of wall clock and prior+1.
1014-- Strictly-monotonic next sequence timestamp: max of wall clock and prior+1.
1027-- Guarantees /poll cursors order events even when many arrive in the same ms.
1015-- Guarantees /poll cursors order events even when many arrive in the same ms.
1028nextSeqTs : Timestamp -> Timestamp -> Timestamp
1016nextSeqTs : Timestamp -> Timestamp -> Timestamp
@@ -1048,9 +1036,7 @@ maskedServer = \pres pubkey t -> (\here -> if any here (\p -> isLocalServer p.se
1048-- The blob comparison avoids hashing every live row on each check; the
1036-- The blob comparison avoids hashing every live row on each check; the
1049-- tombstone log only stores the id, so we compare it directly.
1037-- tombstone log only stores the id, so we compare it directly.
1050isDuplicateMessage : BlobHex -> PubkeyHex -> [Message] -> [DeliveredEntry] -> Bool
1038isDuplicateMessage : BlobHex -> PubkeyHex -> [Message] -> [DeliveredEntry] -> Bool
1051−isDuplicateMessage = \blob toKey msgs dids ->
1052− any msgs (\m -> m.toKey == toKey && m.encryptedBlob == blob) ||
1053− (\mid -> any dids (\d -> d.toKey == toKey && d.messageId == mid)) (messageIdOf blob)
1039+isDuplicateMessage = \blob toKey msgs dids -> any msgs (\m -> m.toKey == toKey && m.encryptedBlob == blob) || (\mid -> any dids (\d -> d.toKey == toKey && d.messageId == mid)) (messageIdOf blob)
1055-- Project items through `getServer`, drop our own hostname, and dedupe.
1041-- Project items through `getServer`, drop our own hostname, and dedupe.
1056-- Used to derive peer recipient lists from forwards/presence/etc.
1042-- Used to derive peer recipient lists from forwards/presence/etc.
@@ -1083,9 +1069,9 @@ pruneExpired = \t -> atomic do
1083 ses <- *sessions
1069 ses <- *sessions
1084 *sessions = filter ses (\x -> x.expiresAt > t)
1070 *sessions = filter ses (\x -> x.expiresAt > t)
1085 msgs <- *messages
1071 msgs <- *messages
1086− *messages = filter msgs (\m -> m.expiresAt > t)
1072+ *messages = filter msgs (\m -> t - m.receivedAt < messageMaxAge)
1087 fwds <- *forwards
1073 fwds <- *forwards
1088− *forwards = filter fwds (\f -> f.expiresAt > t)
1074+ *forwards = filter fwds (\f -> t - f.createdAt < messageMaxAge)
1089 dids <- *deliveredIds
1075 dids <- *deliveredIds
1090 *deliveredIds = filter dids (\d -> t - d.deliveredAt < messageMaxAge)
1076 *deliveredIds = filter dids (\d -> t - d.deliveredAt < messageMaxAge)
1091 ff <- *forwardFailures
1077 ff <- *forwardFailures
@@ -1174,8 +1160,7 @@ retryPendingForwards = do
1174 fromServer: serverHostN,
1160 fromServer: serverHostN,
1175 requestTimestamp: t2,
1161 requestTimestamp: t2,
1176 authorization: auth,
1162 authorization: auth,
1177− correlationId: "",
1163+ correlationId: ""
1180 success <- case result of
1165 success <- case result of
1181 Err _ -> do
1166 Err _ -> do
@@ -1189,15 +1174,7 @@ retryPendingForwards = do
1189 atomic (do
1174 atomic (do
1190 cur <- *forwards
1175 cur <- *forwards
1191 *forwards = pluck (removeWhere cur isDone) (\x -> if isMatch x
1176 *forwards = pluck (removeWhere cur isDone) (\x -> if isMatch x
1192− then {
1177+ then {x.toServer, x.createdAt, x.toKey, x.encryptedBlob, x.timestamp, retries: x.retries + 1}
1198− retries: x.retries + 1,
1201 else x))))
1178 else x))))
1203-- Forward queued messages to a server when a remote key comes online.
1180-- Forward queued messages to a server when a remote key comes online.
@@ -1212,8 +1189,7 @@ forwardQueuedToServer = \pubkey server t -> atomic (do
1212 m.toKey,
1189 m.toKey,
1213 m.encryptedBlob,
1190 m.encryptedBlob,
1214 m.timestamp,
1191 m.timestamp,
1215− retries: 0,
1192+ retries: 0
1218 *forwards = union fwds newFwds)
1194 *forwards = union fwds newFwds)
@@ -1470,7 +1446,7 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
1471-- One pass: collect events at `t`. If empty, retry-wait on `pollWakeup`
1447-- One pass: collect events at `t`. If empty, retry-wait on `pollWakeup`
1472-- (bumped by send/heartbeat) and re-collect once at the new wall time.
1448-- (bumped by send/heartbeat) and re-collect once at the new wall time.
1473−doPoll : {clampedCursor: Int, pubkey: PubkeyHex, startWakeup: Timestamp} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{id: MessageIdHex, payload: {encryptedBlob: BlobHex, expiresAt: Timestamp, timestamp: Timestamp, toKey: PubkeyHex}}], hasMore: Bool}
1449+doPoll : {clampedCursor: Int, pubkey: PubkeyHex, startWakeup: Timestamp} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{id: MessageIdHex, payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex}}], hasMore: Bool}
1474doPoll = \s t -> do
1450doPoll = \s t -> do
1475 let collectAt = \tnow -> atomic (do
1451 let collectAt = \tnow -> atomic (do
1476 msgs <- *messages
1452 msgs <- *messages
@@ -1486,7 +1462,7 @@ doPoll = \s t -> do
1486-- Pull this pubkey's unexpired messages newer than `cursor`, sort by
1462-- Pull this pubkey's unexpired messages newer than `cursor`, sort by
1487-- receivedAt, take up to `maxPollEvents`, and report `hasMore` if truncated.
1463-- receivedAt, take up to `maxPollEvents`, and report `hasMore` if truncated.
1488collectPollEvents : [Message] -> PubkeyHex -> Timestamp -> Timestamp -> {events: [PollEvent], cursor: Timestamp, hasMore: Bool}
1464collectPollEvents : [Message] -> PubkeyHex -> Timestamp -> Timestamp -> {events: [PollEvent], cursor: Timestamp, hasMore: Bool}
1489−collectPollEvents = \messages pubkey cursor t -> (\myMsgs -> (\limited -> {events: pluck limited (\m -> {id: messageIdOf m.encryptedBlob, payload: {m.toKey, m.encryptedBlob, m.timestamp, m.expiresAt}}), cursor: fold (\acc m -> max acc m.receivedAt) cursor limited, hasMore: count myMsgs > count limited}) (takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs))) (filter messages (\m -> m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor))
1465+collectPollEvents = \messages pubkey cursor t -> (\myMsgs -> (\limited -> {events: pluck limited (\m -> {id: messageIdOf m.encryptedBlob, payload: {m.toKey, m.encryptedBlob, m.timestamp}}), cursor: fold (\acc m -> max acc m.receivedAt) cursor limited, hasMore: count myMsgs > count limited}) (takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs))) (filter messages (\m -> m.toKey == pubkey && t - m.receivedAt < messageMaxAge && m.receivedAt > cursor))
1491-- Treat the supplied cursor as an ack: drop all messages with
1467-- Treat the supplied cursor as an ack: drop all messages with
1492-- receivedAt <= cursor, plus their pending forwards.
1468-- receivedAt <= cursor, plus their pending forwards.
@@ -1515,8 +1491,8 @@ ackPolledMessages = \pubkey cursor -> do
1515-- POST /messages: send one ciphertext to `to`. Rejects on draining/rate
1491-- POST /messages: send one ciphertext to `to`. Rejects on draining/rate
1516-- limit/stale timestamp; otherwise calls `sendMessageCore` inside an atomic
1492-- limit/stale timestamp; otherwise calls `sendMessageCore` inside an atomic
1517-- block and forks the federation forward (if any) once the local commit lands.
1493-- block and forks the federation forward (if any) once the local commit lands.
1518−handleSendMessage : PubkeyHex -> BlobHex -> Int<Ms> -> Int -> Text -> Text -> ({error: Text, loadPct: Int, messageId: MessageIdHex, retryAfterMs: Int, status: MessageStatus} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock, random} a
1494+handleSendMessage : PubkeyHex -> BlobHex -> Int<Ms> -> Text -> Text -> ({error: Text, loadPct: Int, messageId: MessageIdHex, retryAfterMs: Int, status: MessageStatus} -> a) -> IO {r *deliveredIds, r *drainPhase, r *presence, r *sessions, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, rw *pairRates, rw *pollWakeup, rw *recvRates, rw *sendRates, rw *seqCounter, clock} a
1519−handleSendMessage = \to encryptedBlob timestamp expiresInMs authorization clientIp respond -> do
1495+handleSendMessage = \to encryptedBlob timestamp authorization clientIp respond -> do
1520 t <- now
1496 t <- now
1521 dp <- *drainPhase
1497 dp <- *drainPhase
1522 g <- checkGlobalRate t
1498 g <- checkGlobalRate t
@@ -1531,7 +1507,7 @@ handleSendMessage = \to encryptedBlob timestamp expiresInMs authorization client
1532 let clampedTs = clampTimestamp timestamp t
1508 let clampedTs = clampTimestamp timestamp t
1533 result <- atomic (do
1509 result <- atomic (do
1534− withSessionAuth authorization t clientIp (yield {status: MsUnauthorized {}, remoteServers: []}) (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs expiresInMs t))
1510+ withSessionAuth authorization t clientIp (yield {status: MsUnauthorized {}, remoteServers: []}) (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs t))
1535 when federationEnabled (fork (forkForwardsTo to encryptedBlob clampedTs result.remoteServers))
1511 when federationEnabled (fork (forkForwardsTo to encryptedBlob clampedTs result.remoteServers))
1536 let ok = isSendSuccess result.status
1512 let ok = isSendSuccess result.status
1537 yield (respond {
1513 yield (respond {
@@ -1548,8 +1524,8 @@ handleSendMessage = \to encryptedBlob timestamp expiresInMs authorization client
1548-- `seqCounter`, appends the message, and pre-stages forwards for any remote
1524-- `seqCounter`, appends the message, and pre-stages forwards for any remote
1549-- presence rows that already exist for `to`. Returns the chosen status and
1525-- presence rows that already exist for `to`. Returns the chosen status and
1550-- the list of remote servers the caller should kick forwards to.
1526-- the list of remote servers the caller should kick forwards to.
1551−sendMessageCore : PubkeyHex -> PubkeyHex -> BlobHex -> Int<Ms> -> Int -> Int<Ms> -> IO {r *deliveredIds, rw *forwards, rw *messages, rw *pairRates, rw *pollWakeup, r *presence, rw *recvRates, rw *sendRates, rw *seqCounter} {remoteServers: [Text], status: MessageStatus}
1527+sendMessageCore : PubkeyHex -> PubkeyHex -> BlobHex -> Int<Ms> -> Int<Ms> -> IO {r *deliveredIds, rw *forwards, rw *messages, rw *pairRates, rw *pollWakeup, r *presence, rw *recvRates, rw *sendRates, rw *seqCounter} {remoteServers: [Text], status: MessageStatus}
1552−sendMessageCore = \senderKey to encryptedBlob clampedTs expiresInMs t -> do
1528+sendMessageCore = \senderKey to encryptedBlob clampedTs t -> do
1553 msgs <- *messages
1529 msgs <- *messages
1554 dids <- *deliveredIds
1530 dids <- *deliveredIds
1555 sRates <- *sendRates
1531 sRates <- *sendRates
@@ -1576,16 +1552,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs expiresInMs t -> do
1576 seq <- *seqCounter
1552 seq <- *seqCounter
1577 let seqTs = nextSeqTs t seq
1553 let seqTs = nextSeqTs t seq
1578 *seqCounter = seqTs
1554 *seqCounter = seqTs
1579− let expiresAt = seqTs + effectiveMessageTtl expiresInMs
1580− *messages = union msgs [
1584− timestamp: clampedTs,
1555+ *messages = union msgs [{toKey: to, encryptedBlob, timestamp: clampedTs, receivedAt: seqTs}]
1589 *pollWakeup = seqTs
1556 *pollWakeup = seqTs
1590 let toPresence = presenceForKey pres to t
1557 let toPresence = presenceForKey pres to t
1591 let remoteServers = collectPeers toPresence (\p -> p.server)
1558 let remoteServers = collectPeers toPresence (\p -> p.server)
@@ -1598,8 +1565,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs expiresInMs t -> do
1598 toKey: to,
1565 toKey: to,
1599 encryptedBlob,
1566 encryptedBlob,
1600 timestamp: clampedTs,
1567 timestamp: clampedTs,
1601− retries: 0,
1568+ retries: 0
1604 *forwards = union fwds newFwds
1570 *forwards = union fwds newFwds
1605 yield (count newFwds > 0)
1571 yield (count newFwds > 0)
@@ -1635,8 +1601,7 @@ forkForwardsTo = \to encryptedBlob clampedTs servers -> forEach servers (\server
1635 fromServer: serverHostN,
1601 fromServer: serverHostN,
1636 requestTimestamp: t,
1602 requestTimestamp: t,
1637 authorization: auth,
1603 authorization: auth,
1638− correlationId: "",
1604+ correlationId: ""
1641 case result of
1606 case result of
1642 Err _ -> recordFailure server
1607 Err _ -> recordFailure server
@@ -1757,7 +1722,7 @@ handleListMessages = \offset authorization clientIp respond -> do
1757 then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})
1722 then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})
1759 msgs <- *messages
1724 msgs <- *messages
1760− let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))
1725+ let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && t - m.receivedAt < messageMaxAge))
1761 let total = count myMsgs
1726 let total = count myMsgs
1762 let page = pluck (takeRelation maxPollEvents (dropRelation (clamp 0 total offset) myMsgs)) (\m -> {
1727 let page = pluck (takeRelation maxPollEvents (dropRelation (clamp 0 total offset) myMsgs)) (\m -> {
1763 messageId: messageIdOf m.encryptedBlob,
1728 messageId: messageIdOf m.encryptedBlob,
@@ -1791,7 +1756,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
1791-- POST /messages/batch: multi-recipient send in one call. Each entry runs
1756-- POST /messages/batch: multi-recipient send in one call. Each entry runs
1792-- through `sendMessageCore` separately so per-entry rate limits/errors don't
1757-- through `sendMessageCore` separately so per-entry rate limits/errors don't
1793-- abort the rest; federated forwards are forked per entry.
1758-- abort the rest; federated forwards are forked per entry.
1794−handleSendBatch : [{encryptedBlob: BlobHex, expiresInMs: Int, timestamp: Int<Ms>, to: PubkeyHex | a}] -> Text -> Text -> ({authorized: Bool, results: [BatchResult], retryAfterMs: Int} -> b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} b
1759+handleSendBatch : [{encryptedBlob: BlobHex, timestamp: Int<Ms>, to: PubkeyHex | a}] -> Text -> Text -> ({authorized: Bool, results: [BatchResult], retryAfterMs: Int} -> b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} b
1795handleSendBatch = \messages authorization clientIp respond -> do
1760handleSendBatch = \messages authorization clientIp respond -> do
1796 t <- now
1761 t <- now
1797 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
1762 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
@@ -1801,8 +1766,7 @@ handleSendBatch = \messages authorization clientIp respond -> do
1801 let clampedTs = clampTimestamp m.timestamp t
1766 let clampedTs = clampTimestamp m.timestamp t
1802 r <- if t - m.timestamp > messageMaxAge
1767 r <- if t - m.timestamp > messageMaxAge
1803 then yield {status: MsRejected {}, remoteServers: []}
1768 then yield {status: MsRejected {}, remoteServers: []}
1805− sendMessageCore senderKey m.to m.encryptedBlob clampedTs m.expiresInMs t)
1769+ else atomic (do sendMessageCore senderKey m.to m.encryptedBlob clampedTs t)
1806 when federationEnabled (fork (forkForwardsTo m.to m.encryptedBlob clampedTs r.remoteServers))
1770 when federationEnabled (fork (forkForwardsTo m.to m.encryptedBlob clampedTs r.remoteServers))
1807 yield {messageId: effMsgId, r.status})
1771 yield {messageId: effMsgId, r.status})
1808 yield (respond {results, authorized: True {}, retryAfterMs: 0}))
1772 yield (respond {results, authorized: True {}, retryAfterMs: 0}))
@@ -1870,8 +1834,8 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
1870-- on stale gossip about R could land on B with no actively-polling client and
1834-- on stale gossip about R could land on B with no actively-polling client and
1871-- get stuck. If B refuses, A retries via `retryPendingForwards` against the
1835-- get stuck. If B refuses, A retries via `retryPendingForwards` against the
1872-- next presence target.
1836-- next presence target.
1873−storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> Int<Ms> -> Bool -> Int<Ms> -> IO {r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter} Text
1837+storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> Bool -> Int<Ms> -> IO {r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter} Text
1874−storeFederated = \toKey encryptedBlob timestamp expiresAt checkRecvRate t -> do
1838+storeFederated = \toKey encryptedBlob timestamp checkRecvRate t -> do
1875 pres <- *presence
1839 pres <- *presence
1876 msgs <- *messages
1840 msgs <- *messages
1877 dids <- *deliveredIds
1841 dids <- *deliveredIds
@@ -1892,18 +1856,9 @@ storeFederated = \toKey encryptedBlob timestamp expiresAt checkRecvRate t -> do
1893 seq <- *seqCounter
1857 seq <- *seqCounter
1894 let seqTs = nextSeqTs t seq
1858 let seqTs = nextSeqTs t seq
1895− let cappedExpiry = if expiresAt > t
1896− then min expiresAt (seqTs + unackedMessageMaxAge)
1897− else seqTs + unackedMessageMaxAge
1898 *seqCounter = seqTs
1859 *seqCounter = seqTs
1899 *messages = union msgs [
1860 *messages = union msgs [
1903− timestamp: clampTimestamp timestamp t,
1861+ {toKey, encryptedBlob, timestamp: clampTimestamp timestamp t, receivedAt: seqTs}
1905− expiresAt: cappedExpiry
1908 *pollWakeup = seqTs
1863 *pollWakeup = seqTs
1909 yield ""
1864 yield ""
@@ -1915,21 +1870,21 @@ storeFederated = \toKey encryptedBlob timestamp expiresAt checkRecvRate t -> do
1915-- the blob ourselves so a hostile peer cannot bypass dedup by varying the
1870-- the blob ourselves so a hostile peer cannot bypass dedup by varying the
1916-- id. Honest senders derive the same value, so the "forward:" ++ messageId
1871-- id. Honest senders derive the same value, so the "forward:" ++ messageId
1917-- auth scope still verifies.
1872-- auth scope still verifies.
1918−handleForwardMessage : PubkeyHex -> BlobHex -> Int<Ms> -> Text -> Text -> Int -> Int<Ms> -> Text -> ({error: Text, ok: Bool, retryAfterMs: Int} -> a) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter, clock, network} a
1873+handleForwardMessage : PubkeyHex -> BlobHex -> Int<Ms> -> Text -> Text -> Int -> Text -> ({error: Text, ok: Bool, retryAfterMs: Int} -> a) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter, clock, network} a
1919−handleForwardMessage = \toKey encryptedBlob timestamp _ fromServer requestTimestamp expiresAt authorization respond -> do
1874+handleForwardMessage = \toKey encryptedBlob timestamp _ fromServer requestTimestamp authorization respond -> do
1920 let messageId = messageIdOf encryptedBlob
1875 let messageId = messageIdOf encryptedBlob
1921 t <- now
1876 t <- now
1922 withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (isRemote fromServer) (do
1877 withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (isRemote fromServer) (do
1923 if t - timestamp > messageMaxAge
1878 if t - timestamp > messageMaxAge
1924 then yield (respond {ok: False {}, error: "stale_timestamp", retryAfterMs: 0})
1879 then yield (respond {ok: False {}, error: "stale_timestamp", retryAfterMs: 0})
1926− result <- atomic storeFederated toKey encryptedBlob timestamp expiresAt (True {}) t
1881+ result <- atomic storeFederated toKey encryptedBlob timestamp (True {}) t
1927 yield (respond {ok: result == "", error: result, retryAfterMs: 0}))
1882 yield (respond {ok: result == "", error: result, retryAfterMs: 0}))
1929-- POST /federation/forward-batch: bulk variant of `handleForwardMessage`.
1884-- POST /federation/forward-batch: bulk variant of `handleForwardMessage`.
1930-- Skips per-entry recv-rate charging (the per-server federation rate limit
1885-- Skips per-entry recv-rate charging (the per-server federation rate limit
1931-- already gates the call as a whole).
1886-- already gates the call as a whole).
1932−handleFederationBatchForward : [{encryptedBlob: BlobHex, expiresAt: Int<Ms>, messageId: MessageIdHex, timestamp: Int<Ms>, toKey: PubkeyHex | a}] -> Text -> Int -> Text -> ({error: Text, ok: Bool, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} -> b) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, clock, network} b
1887+handleFederationBatchForward : [{encryptedBlob: BlobHex, messageId: MessageIdHex, timestamp: Int<Ms>, toKey: PubkeyHex | a}] -> Text -> Int -> Text -> ({error: Text, ok: Bool, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} -> b) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, clock, network} b
1933handleFederationBatchForward = \forwards fromServer requestTimestamp authorization respond -> do
1888handleFederationBatchForward = \forwards fromServer requestTimestamp authorization respond -> do
1934 t <- now
1889 t <- now
1935 let batchRespond = \r -> respond {r.ok, r.error, results: [], r.retryAfterMs}
1890 let batchRespond = \r -> respond {r.ok, r.error, results: [], r.retryAfterMs}
@@ -1940,7 +1895,7 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp authorizati
1940 ok <- if t - e.timestamp > messageMaxAge
1895 ok <- if t - e.timestamp > messageMaxAge
1941 then yield (False {})
1896 then yield (False {})
1943− result <- atomic storeFederated e.toKey e.encryptedBlob e.timestamp e.expiresAt (False {}) t
1898+ result <- atomic storeFederated e.toKey e.encryptedBlob e.timestamp (False {}) t
1944 yield (result == "")
1899 yield (result == "")
1945 yield {messageId, ok})
1900 yield {messageId, ok})
1946 yield (respond {ok: True {}, error: "", results, retryAfterMs: 0}))
1901 yield (respond {ok: True {}, error: "", results, retryAfterMs: 0}))
@@ -2640,7 +2595,7 @@ serve = \req -> case req of
2640 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
2595 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
2641 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
2596 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
2642 Poll {cursor, generation, authorization, clientIp, respond} -> handlePoll cursor generation authorization clientIp respond
2597 Poll {cursor, generation, authorization, clientIp, respond} -> handlePoll cursor generation authorization clientIp respond
2643− SendMessage {to, encryptedBlob, timestamp, expiresInMs, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp expiresInMs authorization clientIp respond
2598+ SendMessage {to, encryptedBlob, timestamp, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp authorization clientIp respond
2644 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
2599 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
2645 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
2600 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
2646 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
2601 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
@@ -2649,7 +2604,7 @@ serve = \req -> case req of
2649 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
2604 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
2650 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
2605 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
2651 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
2606 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
2652− ForwardMessage {toKey, encryptedBlob, timestamp, messageId, fromServer, requestTimestamp, expiresAt, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp messageId fromServer requestTimestamp expiresAt authorization respond
2607+ ForwardMessage {toKey, encryptedBlob, timestamp, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp messageId fromServer requestTimestamp authorization respond
2653 FederationBatchForward {forwards, fromServer, requestTimestamp, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp authorization respond
2608 FederationBatchForward {forwards, fromServer, requestTimestamp, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp authorization respond
2654 FederationPubkey {respond} -> handleFederationPubkey respond
2609 FederationPubkey {respond} -> handleFederationPubkey respond
2655 Prune {authorization, respond} -> handlePrune authorization respond
2610 Prune {authorization, respond} -> handlePrune authorization respond