@@ -248,9 +248,11 @@ impl FromJSON GossipEventType where
248248-- One presence change exchanged between federated servers.
249249type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex}
250250
251−-- /messages and /messages/batch responses signal failure via the `error`
252−-- field: empty = success (delivered, federated, queued, or silent duplicate),
253−-- non-empty = "draining" | "rate_limited" | "unauthorized" | "self_send".
251+-- /messages signals failure via the per-entry `error` field on each result:
252+-- empty = success (delivered, federated, queued, or silent duplicate),
253+-- non-empty = "rate_limited" | "self_send". A non-empty top-level `error`
254+-- ("draining" | "rate_limited" | "unauthorized") means the request was
255+-- rejected before any entry was processed and `results` is empty.
254256-- The sender's UI treats every success the same and learns "they got it"
255257-- only via the recipient's E2E ack.
256258
@@ -325,11 +327,6 @@ type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
325327-- A live long-poll slot held by a pubkey; used to cap concurrency per key.
326328type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
327329
328−-- One entry of a /messages/batch request body.
329−type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex}
330−-- Per-entry result returned by /messages/batch.
331−type BatchResult = {messageId: MessageIdHex, error: Text}
332−
333330-- ============================================
334331-- Persisted State (all flat lists, no sharding)
335332-- ============================================
@@ -404,10 +401,9 @@ route AuthApi where
404401
405402route ClientApi where
406403 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
407− POST {to: PubkeyHex, encryptedBlob: BlobHex} /messages headers {authorization: Text, clientIp: Text} -> {messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int} = SendMessage
404+ POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, clientIp: Text} -> {results: [{messageId: MessageIdHex, error: Text}], error: Text, retryAfterMs: Int} = SendMessages
408405 POST {to: PubkeyHex, messageId: MessageIdHex} /messages/status headers {authorization: Text, clientIp: Text} -> {status: Text, authorized: Bool} = MessageStatus
409406 POST {messageIds: [MessageIdHex]} /messages/ack headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, ackedCount: Int, error: Text} = AckMessages
410− POST {messages: [BatchMessage]} /messages/batch headers {authorization: Text, clientIp: Text} -> {results: [BatchResult], authorized: Bool, retryAfterMs: Int} = SendBatch
411407 POST {offset: Int} /messages/list headers {authorization: Text, clientIp: Text} -> {messages: [{messageId: MessageIdHex, receivedAt: Timestamp}], authorized: Bool, count: Int, error: Text} = ListMessages
412408 POST {messageIds: [MessageIdHex]} /messages/delete headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, deletedCount: Int, error: Text} = DeleteMessages
413409 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text} -> {server: AnyServerName, authorized: Bool, error: Text} = LookupPresence
@@ -571,10 +567,6 @@ emptyPollResp = \cursor authorized rateLimited draining gen retryAfterMs cursorW
571567 loadPct: 0
572568}
573569
574−-- Shared error shape for /messages send.
575−sendErrResp : Text -> MessageIdHex -> {messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int}
576−sendErrResp = \err messageId -> {messageId, retryAfterMs: 0, error: err, loadPct: 0}
577−
578570-- Sequence an IO action over each item, in order.
579571forEach : [a] -> (a -> IO {} {}) -> IO {} {}
580572forEach = \items action -> case single (takeRelation 1 items) of
@@ -1394,32 +1386,24 @@ ackPolledMessages = \pubkey cursor -> do
13941386-- per-recipient rate limits charged together inside `sendMessageCore`.
13951387-- Federated forwards are kicked off in a background fork after the atomic.
13961388
1397−-- POST /messages: send one ciphertext to `to`. Rejects on draining/rate
1398−-- limit; otherwise calls `sendMessageCore` inside an atomic block and forks
1399−-- the federation forward (if any) once the local commit lands.
1400−handleSendMessage : PubkeyHex -> BlobHex -> Text -> Text -> ({error: Text, loadPct: Int, messageId: MessageIdHex, retryAfterMs: Int} -> 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
1401−handleSendMessage = \to encryptedBlob authorization clientIp respond -> do
1389+-- POST /messages: send a list of ciphertexts. Each entry runs through
1390+-- `sendMessageCore` separately so per-entry rate limits/errors don't abort
1391+-- the rest; federated forwards are forked per entry. Top-level `error`
1392+-- carries global gate failures ("draining" | "rate_limited" | "unauthorized")
1393+-- and short-circuits with empty `results`.
1394+handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> ({error: Text, results: [{error: Text, messageId: MessageIdHex}], retryAfterMs: Int} -> b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} b
1395+handleSendMessages = \messages authorization clientIp respond -> do
14021396 t <- now
1403− dp <- *drainPhase
1404− g <- checkGlobalRate t
1405− let effMsgId = messageIdOf encryptedBlob
1406− let err = \e -> yield (respond (sendErrResp e effMsgId))
1407− if dp >= 2
1408− then err "draining"
1409− else if g
1410− then err "rate_limited"
1411− else do
1412− result <- atomic (do
1413− withSessionAuth authorization t clientIp (yield {error: "unauthorized", remoteServers: []}) (\senderKey -> sendMessageCore senderKey to encryptedBlob t))
1414− when federationEnabled (fork (forkForwardsTo to encryptedBlob result.remoteServers))
1415− yield (respond {
1416− messageId: effMsgId,
1417− retryAfterMs: 0,
1418− error: result.error,
1419− loadPct: 0
1420− })
1397+ let mkErr = \e -> {results: [], error: e, retryAfterMs: 0}
1398+ gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do
1399+ results <- mapEach (takeRelation maxBatchSize messages) (\m -> do
1400+ let effMsgId = messageIdOf m.encryptedBlob
1401+ r <- atomic (do sendMessageCore senderKey m.to m.encryptedBlob t)
1402+ when federationEnabled (fork (forkForwardsTo m.to m.encryptedBlob r.remoteServers))
1403+ yield {messageId: effMsgId, r.error})
1404+ yield (respond {results, error: "", retryAfterMs: 0}))
14211405
1422−-- Atomic core of /messages and /messages/batch. Charges send/pair/recv rate
1406+-- Atomic core of /messages. Charges send/pair/recv rate
14231407-- counters together (so any limit denies the message), enforces caps and
14241408-- self-send/duplicate rejections, allocates a monotonic receivedAt via
14251409-- `seqCounter`, appends the message, and pre-stages forwards for any remote
@@ -1644,21 +1628,6 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
16441628 yield (count toDel))
16451629 yield (respond {ok: True {}, authorized: True {}, deletedCount, error: ""}))
16461630
1647−-- POST /messages/batch: multi-recipient send in one call. Each entry runs
1648−-- through `sendMessageCore` separately so per-entry rate limits/errors don't
1649−-- abort the rest; federated forwards are forked per entry.
1650−handleSendBatch : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> ({authorized: Bool, results: [BatchResult], retryAfterMs: Int} -> b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} b
1651−handleSendBatch = \messages authorization clientIp respond -> do
1652− t <- now
1653− let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
1654− gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do
1655− results <- mapEach (takeRelation maxBatchSize messages) (\m -> do
1656− let effMsgId = messageIdOf m.encryptedBlob
1657− r <- atomic (do sendMessageCore senderKey m.to m.encryptedBlob t)
1658− when federationEnabled (fork (forkForwardsTo m.to m.encryptedBlob r.remoteServers))
1659− yield {messageId: effMsgId, r.error})
1660− yield (respond {results, authorized: True {}, retryAfterMs: 0}))
1661−
16621631-- ============================================
16631632-- Handler: federation receive
16641633-- ============================================
@@ -2453,10 +2422,9 @@ serve = \req -> case req of
24532422 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
24542423 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
24552424 Poll {cursor, generation, authorization, clientIp, respond} -> handlePoll cursor generation authorization clientIp respond
2456− SendMessage {to, encryptedBlob, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob authorization clientIp respond
2425+ SendMessages {messages, authorization, clientIp, respond} -> handleSendMessages messages authorization clientIp respond
24572426 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
24582427 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
2459− SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
24602428 ListMessages {offset, authorization, clientIp, respond} -> handleListMessages offset authorization clientIp respond
24612429 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
24622430 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond