@@ -196,11 +196,12 @@ impl FromJSON GossipEventType where
196-- One presence change exchanged between federated servers.
196-- One presence change exchanged between federated servers.
197type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex}
197type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex}
199−-- /messages signals failure via the per-entry `error` field on each result:
199+-- /messages is all-or-nothing per call: either every message in the batch is
200−-- empty = success (delivered, federated, or queued),
200+-- enqueued (delivered, federated, or buffered for federation), or none are
201−-- non-empty = "rate_limited" | "self_send". Top-level rejections (auth,
201+-- and the call returns a non-2xx HTTP status. Per-message rejection codes
202−-- global rate limit) come back as a non-2xx HTTP status with an empty
202+-- ("self_send", "rate_limited") surface as the HTTP error body. Rate
203−-- response body.
203+-- counters are still charged for every attempted message even on rejection,
204+-- so a sender can't probe limits cheaply.
204-- The sender's UI treats every success the same and learns "they got it"
205-- The sender's UI treats every success the same and learns "they got it"
205-- only via the recipient's E2E ack.
206-- only via the recipient's E2E ack.
@@ -326,7 +327,7 @@ route AuthApi where
327route ClientApi where
328route ClientApi where
328 POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, generation: Int} = Poll
329 POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, generation: Int} = Poll
329− POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, clientIp: Text} -> {results: [{error: Text}]} = SendMessages
330+ POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, clientIp: Text} -> {} = SendMessages
331route FederationApi where
332route FederationApi where
332 POST {events: [GossipEvent], fromServer: ServerName} /federation/gossip -> {} = RecvGossip
333 POST {events: [GossipEvent], fromServer: ServerName} /federation/gossip -> {} = RecvGossip
@@ -994,79 +995,93 @@ ackPolledMessages = \pubkey cursor -> do
994-- ============================================
995-- ============================================
995-- Handler: send
996-- Handler: send
996-- ============================================
997-- ============================================
997−-- Accept a message for a recipient pubkey. Placement (live local session,
998+-- Accept a batch of messages for recipient pubkeys. Placement (live local
998−-- federated forward, or held-for-presence) is internal; the response only
999+-- session, federated forward, or held-for-presence) is internal; the call
999−-- says whether the message was accepted (`error: ""`) or refused (`error:
1000+-- either succeeds (every message enqueued) or fails with a non-2xx HTTP
1000−-- "rate_limited" | "unauthorized" | "self_send"`). Duplicates are silently
1001+-- status carrying the rejection code ("rate_limited" | "self_send" | ...).
1001−-- treated as success so retries are idempotent.
1002+-- Duplicates are silently treated as success so retries are idempotent.
1002-- Send is gated by global rate and the per-sender / per-pair / per-recipient
1003-- Send is gated by global rate and the per-sender / per-pair / per-recipient
1003−-- rate limits charged together inside `sendMessageCore`.
1004+-- rate limits charged together inside `checkSendRate`.
1004−-- Federated forwards are kicked off in a background fork after the atomic.
1005+-- Federated forwards are kicked off in background forks after the atomic.
1006−-- POST /messages: send a list of ciphertexts. Each entry runs through
1007+-- POST /messages: send a list of ciphertexts atomically. Every message is
1007−-- `sendMessageCore` separately so per-entry rate limits/errors don't abort
1008+-- validated (self-send + send/pair/recv rate windows) inside a single atomic
1008−-- the rest; federated forwards are forked per entry. Top-level rejections
1009+-- block; if any one would be rejected, none are enqueued and the call returns
1009−-- (auth, global rate) come back as a non-2xx HTTP status via `gateAuth`.
1010+-- a non-2xx HTTP status. Rate counters are still charged for every attempted
1011+-- message even when the batch is rejected, so a sender can't probe limits
1012+-- cheaply (e.g. by sending to themselves). Federation forwards are kicked
1013+-- per message after commit.
1010−handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} (Result HttpError {results: [{error: Text}]})
1014+handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} (Result HttpError {})
1011handleSendMessages = \messages authorization clientIp -> do
1015handleSendMessages = \messages authorization clientIp -> do
1012 t <- now
1016 t <- now
1013 gateAuth authorization clientIp t (\senderKey -> do
1017 gateAuth authorization clientIp t (\senderKey -> do
1014− results <- mapEach (takeRelation maxBatchSize messages) (\m -> do
1018+ let limited = takeRelation maxBatchSize messages
1015− r <- atomic (do sendMessageCore senderKey m.to m.encryptedBlob t)
1019+ decision <- atomic (do
1020+ checked <- mapEach limited (\m -> do
1021+ c <- checkSendRate senderKey m.to t
1022+ yield {error: c.error, remoteServers: c.remoteServers, to: m.to, encryptedBlob: m.encryptedBlob})
1023+ case findFirst checked (\c -> c.error != "") of
1024+ Just {value: bad} -> yield (Err {error: bad.error})
1026+ forEach checked (\c -> commitSendMessage c.to c.encryptedBlob c.remoteServers t)
1027+ yield (Ok {value: checked}))
1029+ Err {error: code} -> yield (Err {error: {status: if code == "self_send" then 400 else 429, message: code}})
1030+ Ok {value: checked} -> do
1016− when federationEnabled (fork (forkForwardsTo m.to m.encryptedBlob r.remoteServers))
1031+ when federationEnabled (forEach checked (\c -> fork (forkForwardsTo c.to c.encryptedBlob c.remoteServers)))
1018− yield (Ok {value: {results}}))
1032+ yield (Ok {value: {}}))
1020−-- Atomic core of /messages. Charges send/pair/recv rate
1034+-- Charge send/pair/recv rate counters for one (sender, recipient) pair and
1021−-- counters together (so any limit denies the message), enforces caps and
1035+-- decide whether the message would be accepted. Mutates only the rate
1022−-- self-send rejections, allocates a monotonic receivedAt via `seqCounter`,
1036+-- counters, never the mailbox; the caller commits via `commitSendMessage`
1023−-- appends the message, and pre-stages forwards for any remote presence rows
1037+-- only after every message in the batch has passed.
1024−-- that already exist for `to`. Returns the error code (empty on success) and
1025−-- the list of remote servers the caller should kick forwards to.
1026-- All three rate counters are charged unconditionally (even on a failure)
1038-- All three rate counters are charged unconditionally (even on a failure)
1027-- so that a sender can't probe limits cheaply by sending to themselves.
1039-- so that a sender can't probe limits cheaply by sending to themselves.
1028-- Order matters: send-rate is the sender's identity, pair-rate covers
1040-- Order matters: send-rate is the sender's identity, pair-rate covers
1029-- conversation spam, and recv-rate is per-recipient.
1041-- conversation spam, and recv-rate is per-recipient.
1030−sendMessageCore : PubkeyHex -> PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *forwards, rw *messages, rw *pairRates, rw *pollWakeup, r *presence, rw *recvRates, rw *sendRates, rw *seqCounter} {remoteServers: [Text], error: Text}
1042+checkSendRate : PubkeyHex -> PubkeyHex -> Int<Ms> -> IO {rw *pairRates, r *presence, rw *recvRates, rw *sendRates} {remoteServers: [Text], error: Text}
1031−sendMessageCore = \senderKey to encryptedBlob t -> do
1043+checkSendRate = \senderKey to t -> do
1033 sRates <- *sendRates
1044 sRates <- *sendRates
1034 let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t
1045 let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t
1035 *sendRates = sR.entries
1046 *sendRates = sR.entries
1036− let srLim = sR.limited
1037 pRates <- *pairRates
1047 pRates <- *pairRates
1038 let pR = checkRate pRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t
1048 let pR = checkRate pRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t
1039 *pairRates = pR.entries
1049 *pairRates = pR.entries
1040− let prLim = pR.limited
1041 rRates <- *recvRates
1050 rRates <- *recvRates
1042 let rR = checkRate rRates to maxRecvRatePerKey recvRateWindowMs t
1051 let rR = checkRate rRates to maxRecvRatePerKey recvRateWindowMs t
1043 *recvRates = rR.entries
1052 *recvRates = rR.entries
1044− let rrLim = rR.limited
1045− let reject = \e -> yield {error: e, remoteServers: []}
1046 if senderKey == to
1053 if senderKey == to
1047− then reject "self_send"
1054+ then yield {error: "self_send", remoteServers: []}
1048− else if srLim || prLim || rrLim
1055+ else if sR.limited || pR.limited || rR.limited
1049− then reject "rate_limited"
1056+ then yield {error: "rate_limited", remoteServers: []}
1051 pres <- *presence
1058 pres <- *presence
1053− let seqTs = nextSeqTs t seq
1054− *seqCounter = seqTs
1055− *messages = union msgs [{toKey: to, encryptedBlob, receivedAt: seqTs}]
1056− *pollWakeup = seqTs
1057 let toPresence = presenceForKey pres to t
1059 let toPresence = presenceForKey pres to t
1058 let remoteServers = collectPeers toPresence (\p -> p.server)
1060 let remoteServers = collectPeers toPresence (\p -> p.server)
1059− when federationEnabled (do
1061− let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s)) (\s -> {
1067− *forwards = union fwds newFwds)
1068 yield {error: "", remoteServers}
1061 yield {error: "", remoteServers}
1063+-- Allocate a monotonic receivedAt via `seqCounter`, append the message to
1064+-- the mailbox, and pre-stage forwards for any remote presence rows that
1065+-- already exist for `to`. Caller must have already validated the message
1066+-- via `checkSendRate`.
1067+commitSendMessage : PubkeyHex -> BlobHex -> [Text] -> Int<Ms> -> IO {rw *forwards, rw *messages, rw *pollWakeup, rw *seqCounter} {}
1068+commitSendMessage = \to encryptedBlob remoteServers t -> do
1071+ let seqTs = nextSeqTs t seq
1072+ *seqCounter = seqTs
1073+ *messages = union msgs [{toKey: to, encryptedBlob, receivedAt: seqTs}]
1074+ *pollWakeup = seqTs
1075+ when federationEnabled (do
1077+ let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s)) (\s -> {
1083+ *forwards = union fwds newFwds)
1070-- Fire one /federation/forward call per pre-staged server. On success the
1085-- Fire one /federation/forward call per pre-staged server. On success the
1071-- corresponding row in `forwards` is pruned, and if no other forwards remain
1086-- corresponding row in `forwards` is pruned, and if no other forwards remain
1072-- for this blob AND the recipient has no local presence, the backing
1087-- for this blob AND the recipient has no local presence, the backing