@@ -196,11 +196,12 @@ impl FromJSON GossipEventType where
196196-- One presence change exchanged between federated servers.
197197type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex}
198198
199−-- /messages signals failure via the per-entry `error` field on each result:
200−-- empty = success (delivered, federated, or queued),
201−-- non-empty = "rate_limited" | "self_send". Top-level rejections (auth,
202−-- global rate limit) come back as a non-2xx HTTP status with an empty
203−-- response body.
199+-- /messages is all-or-nothing per call: either every message in the batch is
200+-- enqueued (delivered, federated, or buffered for federation), or none are
201+-- and the call returns a non-2xx HTTP status. Per-message rejection codes
202+-- ("self_send", "rate_limited") surface as the HTTP error body. Rate
203+-- counters are still charged for every attempted message even on rejection,
204+-- so a sender can't probe limits cheaply.
204205-- The sender's UI treats every success the same and learns "they got it"
205206-- only via the recipient's E2E ack.
206207
@@ -326,7 +327,7 @@ route AuthApi where
326327
327328route ClientApi where
328329 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
330331
331332route FederationApi where
332333 POST {events: [GossipEvent], fromServer: ServerName} /federation/gossip -> {} = RecvGossip
@@ -994,79 +995,93 @@ ackPolledMessages = \pubkey cursor -> do
994995-- ============================================
995996-- Handler: send
996997-- ============================================
997−-- Accept a message for a recipient pubkey. Placement (live local session,
998−-- federated forward, or held-for-presence) is internal; the response only
999−-- says whether the message was accepted (`error: ""`) or refused (`error:
1000−-- "rate_limited" | "unauthorized" | "self_send"`). Duplicates are silently
1001−-- treated as success so retries are idempotent.
998+-- Accept a batch of messages for recipient pubkeys. Placement (live local
999+-- session, federated forward, or held-for-presence) is internal; the call
1000+-- either succeeds (every message enqueued) or fails with a non-2xx HTTP
1001+-- status carrying the rejection code ("rate_limited" | "self_send" | ...).
1002+-- Duplicates are silently treated as success so retries are idempotent.
10021003-- Send is gated by global rate and the per-sender / per-pair / per-recipient
1003−-- rate limits charged together inside `sendMessageCore`.
1004−-- Federated forwards are kicked off in a background fork after the atomic.
1005−
1006−-- POST /messages: send a list of ciphertexts. Each entry runs through
1007−-- `sendMessageCore` separately so per-entry rate limits/errors don't abort
1008−-- the rest; federated forwards are forked per entry. Top-level rejections
1009−-- (auth, global rate) come back as a non-2xx HTTP status via `gateAuth`.
1010−handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} (Result HttpError {results: [{error: Text}]})
1004+-- rate limits charged together inside `checkSendRate`.
1005+-- Federated forwards are kicked off in background forks after the atomic.
1006+
1007+-- POST /messages: send a list of ciphertexts atomically. Every message is
1008+-- validated (self-send + send/pair/recv rate windows) inside a single atomic
1009+-- block; if any one would be rejected, none are enqueued and the call returns
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.
1014+handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} (Result HttpError {})
10111015handleSendMessages = \messages authorization clientIp -> do
10121016 t <- now
10131017 gateAuth authorization clientIp t (\senderKey -> do
1014− results <- mapEach (takeRelation maxBatchSize messages) (\m -> do
1015− r <- atomic (do sendMessageCore senderKey m.to m.encryptedBlob t)
1016− when federationEnabled (fork (forkForwardsTo m.to m.encryptedBlob r.remoteServers))
1017− yield {r.error})
1018− yield (Ok {value: {results}}))
1019−
1020−-- Atomic core of /messages. Charges send/pair/recv rate
1021−-- counters together (so any limit denies the message), enforces caps and
1022−-- self-send rejections, allocates a monotonic receivedAt via `seqCounter`,
1023−-- appends the message, and pre-stages forwards for any remote presence rows
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.
1018+ let limited = takeRelation maxBatchSize messages
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})
1025+ Nothing {} -> do
1026+ forEach checked (\c -> commitSendMessage c.to c.encryptedBlob c.remoteServers t)
1027+ yield (Ok {value: checked}))
1028+ case decision of
1029+ Err {error: code} -> yield (Err {error: {status: if code == "self_send" then 400 else 429, message: code}})
1030+ Ok {value: checked} -> do
1031+ when federationEnabled (forEach checked (\c -> fork (forkForwardsTo c.to c.encryptedBlob c.remoteServers)))
1032+ yield (Ok {value: {}}))
1033+
1034+-- Charge send/pair/recv rate counters for one (sender, recipient) pair and
1035+-- decide whether the message would be accepted. Mutates only the rate
1036+-- counters, never the mailbox; the caller commits via `commitSendMessage`
1037+-- only after every message in the batch has passed.
10261038-- All three rate counters are charged unconditionally (even on a failure)
10271039-- so that a sender can't probe limits cheaply by sending to themselves.
10281040-- Order matters: send-rate is the sender's identity, pair-rate covers
10291041-- 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}
1031−sendMessageCore = \senderKey to encryptedBlob t -> do
1032− msgs <- *messages
1042+checkSendRate : PubkeyHex -> PubkeyHex -> Int<Ms> -> IO {rw *pairRates, r *presence, rw *recvRates, rw *sendRates} {remoteServers: [Text], error: Text}
1043+checkSendRate = \senderKey to t -> do
10331044 sRates <- *sendRates
10341045 let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t
10351046 *sendRates = sR.entries
1036− let srLim = sR.limited
10371047 pRates <- *pairRates
10381048 let pR = checkRate pRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t
10391049 *pairRates = pR.entries
1040− let prLim = pR.limited
10411050 rRates <- *recvRates
10421051 let rR = checkRate rRates to maxRecvRatePerKey recvRateWindowMs t
10431052 *recvRates = rR.entries
1044− let rrLim = rR.limited
1045− let reject = \e -> yield {error: e, remoteServers: []}
10461053 if senderKey == to
1047− then reject "self_send"
1048− else if srLim || prLim || rrLim
1049− then reject "rate_limited"
1054+ then yield {error: "self_send", remoteServers: []}
1055+ else if sR.limited || pR.limited || rR.limited
1056+ then yield {error: "rate_limited", remoteServers: []}
10501057 else do
10511058 pres <- *presence
1052− seq <- *seqCounter
1053− let seqTs = nextSeqTs t seq
1054− *seqCounter = seqTs
1055− *messages = union msgs [{toKey: to, encryptedBlob, receivedAt: seqTs}]
1056− *pollWakeup = seqTs
10571059 let toPresence = presenceForKey pres to t
10581060 let remoteServers = collectPeers toPresence (\p -> p.server)
1059− when federationEnabled (do
1060− fwds <- *forwards
1061− let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s)) (\s -> {
1062− toServer: s,
1063− toKey: to,
1064− encryptedBlob,
1065− retries: 0
1066− })
1067− *forwards = union fwds newFwds)
10681061 yield {error: "", remoteServers}
10691062
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
1069+ msgs <- *messages
1070+ seq <- *seqCounter
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
1076+ fwds <- *forwards
1077+ let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s)) (\s -> {
1078+ toServer: s,
1079+ toKey: to,
1080+ encryptedBlob,
1081+ retries: 0
1082+ })
1083+ *forwards = union fwds newFwds)
1084+
10701085-- Fire one /federation/forward call per pre-staged server. On success the
10711086-- corresponding row in `forwards` is pruned, and if no other forwards remain
10721087-- for this blob AND the recipient has no local presence, the backing