602ee35 · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Remove MessageStatus enum; signal send outcome via error field

602ee35iko5 months ago

client/ApiClient.fs+10−8

compare with master

unifiedsplit

@@ -51,9 +51,11 @@ module ApiClient =

51 let! text = response.Content.ReadAsStringAsync() |> awaitTask

52 let doc = JsonDocument.Parse(text)

53 match doc.RootElement.TryGetProperty("error") with

54− | true, err ->

54+ | true, err when not (String.IsNullOrEmpty(err.GetString())) ->

55+ let code = err.GetString()

55 doc.Dispose()

56 doc.Dispose()

57+ if code = "unauthorized" then return raise Unauthorized

56− return raise (ApiError $"{url}: {err.GetString()}")

58+ else return raise (ApiError $"{url}: {code}")

57 | _ -> return doc

59 | _ -> return doc

58 }

60 }

@@ -79,12 +81,12 @@ module ApiClient =

79 let sendMessage (serverUrl: string) (token: string) (toHex: string) (blobHex: string) =

81 let sendMessage (serverUrl: string) (token: string) (toHex: string) (blobHex: string) =

80 let body = JsonSerializer.Serialize({| ``to`` = toHex; encryptedBlob = blobHex |})

82 let body = JsonSerializer.Serialize({| ``to`` = toHex; encryptedBlob = blobHex |})

81 async {

83 async {

84+ try

82− use! doc = sendRequest client $"{serverUrl}/messages" body (Some token)

85+ use! _ = sendRequest client $"{serverUrl}/messages" body (Some token)

83− match doc.RootElement.GetProperty("status").GetString() with

86+ return ()

84− | "delivered" | "federated" | "queued" | "duplicate" -> ()

87+ with

88+ | ApiError msg when msg.EndsWith(": self_send") ->

85− | "rejected" -> raise (ServerRejected "Message rejected by server")

89+ return raise (ServerRejected "Message rejected by server")

86− | "unauthorized" -> raise Unauthorized

87− | s -> raise (ApiError $"Unexpected status: {s}")

88 }

90 }

90 let poll (serverUrl: string) (token: string) (cursor: int64) =

92 let poll (serverUrl: string) (token: string) (cursor: int64) =

server.knot+32−68

@@ -248,29 +248,11 @@ impl FromJSON GossipEventType where

248-- One presence change exchanged between federated servers.

249type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex}

250

251−-- Outcome classification returned to the sender of /messages and per-entry in

251+-- /messages and /messages/batch responses signal failure via the `error`

252−-- /messages/batch. The first four are success-ish (see `isSendSuccess`); the

252+-- field: empty = success (delivered, federated, queued, or silent duplicate),

253−-- rest are refusals.

253+-- non-empty = "draining" | "rate_limited" | "unauthorized" | "self_send".

254−data MessageStatus

254+-- The sender's UI treats every success the same and learns "they got it"

255− = MsDelivered {}

255+-- only via the recipient's E2E ack.

256− | MsFederated {}

257− | MsQueued {}

258− | MsUnauthorized {}

259− | MsRejected {}

260− | MsRateLimited {}

261− | MsDuplicate {}

262− | MsDraining {}

263−

264−impl ToJSON MessageStatus where

265− toJson ms = toJson (case ms of

266− MsDelivered {} -> "delivered"

267− MsFederated {} -> "federated"

268− MsQueued {} -> "queued"

269− MsUnauthorized {} -> "unauthorized"

270− MsRejected {} -> "rejected"

271− MsRateLimited {} -> "rate_limited"

272− MsDuplicate {} -> "duplicate"

273− MsDraining {} -> "draining")

274

256

275-- True when `s` is non-empty lowercase hex of even length, with `maxLen` ceiling.

257-- True when `s` is non-empty lowercase hex of even length, with `maxLen` ceiling.

276isValidBoundedHex : Int -> Text -> Bool

258isValidBoundedHex : Int -> Text -> Bool

@@ -346,7 +328,7 @@ type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}

346-- One entry of a /messages/batch request body.

328-- One entry of a /messages/batch request body.

347type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex}

329type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex}

348-- Per-entry result returned by /messages/batch.

330-- Per-entry result returned by /messages/batch.

349−type BatchResult = {messageId: MessageIdHex, status: MessageStatus}

331+type BatchResult = {messageId: MessageIdHex, error: Text}

350

332

351-- ============================================

333-- ============================================

352-- Persisted State (all flat lists, no sharding)

334-- Persisted State (all flat lists, no sharding)

@@ -422,7 +404,7 @@ route AuthApi where

422

404

423route ClientApi where

405route ClientApi where

424 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

406 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

425− POST {to: PubkeyHex, encryptedBlob: BlobHex} /messages headers {authorization: Text, clientIp: Text} -> {status: MessageStatus, messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int} = SendMessage

407+ POST {to: PubkeyHex, encryptedBlob: BlobHex} /messages headers {authorization: Text, clientIp: Text} -> {messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int} = SendMessage

426 POST {to: PubkeyHex, messageId: MessageIdHex} /messages/status headers {authorization: Text, clientIp: Text} -> {status: Text, authorized: Bool} = MessageStatus

408 POST {to: PubkeyHex, messageId: MessageIdHex} /messages/status headers {authorization: Text, clientIp: Text} -> {status: Text, authorized: Bool} = MessageStatus

427 POST {messageIds: [MessageIdHex]} /messages/ack headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, ackedCount: Int, error: Text} = AckMessages

409 POST {messageIds: [MessageIdHex]} /messages/ack headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, ackedCount: Int, error: Text} = AckMessages

428 POST {messages: [BatchMessage]} /messages/batch headers {authorization: Text, clientIp: Text} -> {results: [BatchResult], authorized: Bool, retryAfterMs: Int} = SendBatch

410 POST {messages: [BatchMessage]} /messages/batch headers {authorization: Text, clientIp: Text} -> {results: [BatchResult], authorized: Bool, retryAfterMs: Int} = SendBatch

@@ -590,17 +572,8 @@ emptyPollResp = \cursor authorized rateLimited draining gen retryAfterMs cursorW

590}

572}

591

573

592-- Shared error shape for /messages send.

574-- Shared error shape for /messages send.

593−sendErrResp : MessageStatus -> Text -> MessageIdHex -> {status: MessageStatus, messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int}

575+sendErrResp : Text -> MessageIdHex -> {messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int}

594−sendErrResp = \status err messageId -> {status, messageId, retryAfterMs: 0, error: err, loadPct: 0}

576+sendErrResp = \err messageId -> {messageId, retryAfterMs: 0, error: err, loadPct: 0}

595−

596−-- True when a send produced (or matched) a deliverable message.

597−isSendSuccess : MessageStatus -> Bool

598−isSendSuccess = \s -> case s of

599− MsDelivered {} -> True {}

600− MsFederated {} -> True {}

601− MsQueued {} -> True {}

602− MsDuplicate {} -> True {}

603− _ -> False {}

604

577

605-- Sequence an IO action over each item, in order.

578-- Sequence an IO action over each item, in order.

606forEach : [a] -> (a -> IO {} {}) -> IO {} {}

579forEach : [a] -> (a -> IO {} {}) -> IO {} {}

@@ -1412,11 +1385,11 @@ ackPolledMessages = \pubkey cursor -> do

1412-- ============================================

1385-- ============================================

1413-- Handler: send

1386-- Handler: send

1414-- ============================================

1387-- ============================================

1415−-- Accept a message for a recipient pubkey and decide its placement:

1388+-- Accept a message for a recipient pubkey. Placement (live local session,

1416−-- * `MsDelivered` -- recipient has a live local session

1389+-- federated forward, or held-for-presence) is internal; the response only

1417−-- * `MsFederated` -- recipient lives on a remote peer; queued as a forward

1390+-- says whether the message was accepted (`error: ""`) or refused (`error:

1418−-- * `MsQueued` -- no presence yet; held until presence appears

1391+-- "draining" | "rate_limited" | "unauthorized" | "self_send"`). Duplicates

1419−-- * `MsRejected` / `MsRateLimited` / `MsDuplicate` / `MsDraining` -- refused

1392+-- are silently treated as success so retries are idempotent.

1420-- Send is gated by drain phase, global rate, and the per-sender / per-pair /

1393-- Send is gated by drain phase, global rate, and the per-sender / per-pair /

1421-- per-recipient rate limits charged together inside `sendMessageCore`.

1394-- per-recipient rate limits charged together inside `sendMessageCore`.

1422-- Federated forwards are kicked off in a background fork after the atomic.

1395-- Federated forwards are kicked off in a background fork after the atomic.

@@ -1424,27 +1397,25 @@ ackPolledMessages = \pubkey cursor -> do

1424-- POST /messages: send one ciphertext to `to`. Rejects on draining/rate

1397-- POST /messages: send one ciphertext to `to`. Rejects on draining/rate

1425-- limit; otherwise calls `sendMessageCore` inside an atomic block and forks

1398-- limit; otherwise calls `sendMessageCore` inside an atomic block and forks

1426-- the federation forward (if any) once the local commit lands.

1399-- the federation forward (if any) once the local commit lands.

1427−handleSendMessage : PubkeyHex -> BlobHex -> 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

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

1428handleSendMessage = \to encryptedBlob authorization clientIp respond -> do

1401handleSendMessage = \to encryptedBlob authorization clientIp respond -> do

1429 t <- now

1402 t <- now

1430 dp <- *drainPhase

1403 dp <- *drainPhase

1431 g <- checkGlobalRate t

1404 g <- checkGlobalRate t

1432 let effMsgId = messageIdOf encryptedBlob

1405 let effMsgId = messageIdOf encryptedBlob

1433− let err = \s e -> yield (respond (sendErrResp s e effMsgId))

1406+ let err = \e -> yield (respond (sendErrResp e effMsgId))

1434 if dp >= 2

1407 if dp >= 2

1435− then err (MsDraining {}) "draining"

1408+ then err "draining"

1436 else if g

1409 else if g

1437− then err (MsRateLimited {}) "rate_limited"

1410+ then err "rate_limited"

1438 else do

1411 else do

1439 result <- atomic (do

1412 result <- atomic (do

1440− withSessionAuth authorization t clientIp (yield {status: MsUnauthorized {}, remoteServers: []}) (\senderKey -> sendMessageCore senderKey to encryptedBlob t))

1413+ withSessionAuth authorization t clientIp (yield {error: "unauthorized", remoteServers: []}) (\senderKey -> sendMessageCore senderKey to encryptedBlob t))

1441 when federationEnabled (fork (forkForwardsTo to encryptedBlob result.remoteServers))

1414 when federationEnabled (fork (forkForwardsTo to encryptedBlob result.remoteServers))

1442− let ok = isSendSuccess result.status

1443 yield (respond {

1415 yield (respond {

1444− result.status,

1445 messageId: effMsgId,

1416 messageId: effMsgId,

1446 retryAfterMs: 0,

1417 retryAfterMs: 0,

1447− error: if ok then "" else "rejected",

1418+ error: result.error,

1448 loadPct: 0

1419 loadPct: 0

1449 })

1420 })

1450

1421

@@ -1452,9 +1423,10 @@ handleSendMessage = \to encryptedBlob authorization clientIp respond -> do

1452-- counters together (so any limit denies the message), enforces caps and

1423-- counters together (so any limit denies the message), enforces caps and

1453-- self-send/duplicate rejections, allocates a monotonic receivedAt via

1424-- self-send/duplicate rejections, allocates a monotonic receivedAt via

1454-- `seqCounter`, appends the message, and pre-stages forwards for any remote

1425-- `seqCounter`, appends the message, and pre-stages forwards for any remote

1455−-- presence rows that already exist for `to`. Returns the chosen status and

1426+-- presence rows that already exist for `to`. Returns the error code (empty

1427+-- on success or silent duplicate) and the list of remote servers the caller

1456−-- the list of remote servers the caller should kick forwards to.

1428+-- should kick forwards to.

1457−sendMessageCore : PubkeyHex -> PubkeyHex -> BlobHex -> 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}

1429+sendMessageCore : PubkeyHex -> PubkeyHex -> BlobHex -> Int<Ms> -> IO {r *deliveredIds, rw *forwards, rw *messages, rw *pairRates, rw *pollWakeup, r *presence, rw *recvRates, rw *sendRates, rw *seqCounter} {remoteServers: [Text], error: Text}

1458sendMessageCore = \senderKey to encryptedBlob t -> do

1430sendMessageCore = \senderKey to encryptedBlob t -> do

1459 msgs <- *messages

1431 msgs <- *messages

1460 dids <- *deliveredIds

1432 dids <- *deliveredIds

@@ -1470,13 +1442,13 @@ sendMessageCore = \senderKey to encryptedBlob t -> do

1470 let rR = checkRate rRates to maxRecvRatePerKey recvRateWindowMs t

1442 let rR = checkRate rRates to maxRecvRatePerKey recvRateWindowMs t

1471 *recvRates = rR.entries

1443 *recvRates = rR.entries

1472 let rrLim = rR.limited

1444 let rrLim = rR.limited

1473− let reject = \s -> yield {status: s, remoteServers: []}

1445+ let reject = \e -> yield {error: e, remoteServers: []}

1474 if senderKey == to

1446 if senderKey == to

1475− then reject (MsRejected {})

1447+ then reject "self_send"

1476 else if isDuplicateMessage encryptedBlob to msgs dids

1448 else if isDuplicateMessage encryptedBlob to msgs dids

1477− then reject (MsDuplicate {})

1449+ then reject ""

1478 else if srLim || prLim || rrLim

1450 else if srLim || prLim || rrLim

1479− then reject (MsRateLimited {})

1451+ then reject "rate_limited"

1480 else do

1452 else do

1481 pres <- *presence

1453 pres <- *presence

1482 seq <- *seqCounter

1454 seq <- *seqCounter

@@ -1486,8 +1458,7 @@ sendMessageCore = \senderKey to encryptedBlob t -> do

1486 *pollWakeup = seqTs

1458 *pollWakeup = seqTs

1487 let toPresence = presenceForKey pres to t

1459 let toPresence = presenceForKey pres to t

1488 let remoteServers = collectPeers toPresence (\p -> p.server)

1460 let remoteServers = collectPeers toPresence (\p -> p.server)

1489− forwardsEnqueued <- if federationEnabled

1461+ when federationEnabled (do

1490− then do

1491 fwds <- *forwards

1462 fwds <- *forwards

1492 let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s)) (\s -> {

1463 let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s)) (\s -> {

1493 toServer: s,

1464 toServer: s,

@@ -1496,15 +1467,8 @@ sendMessageCore = \senderKey to encryptedBlob t -> do

1496 encryptedBlob,

1467 encryptedBlob,

1497 retries: 0

1468 retries: 0

1498 })

1469 })

1499− *forwards = union fwds newFwds

1470+ *forwards = union fwds newFwds)

1500− yield (count newFwds > 0)

1501− else yield (False {})

1502− yield {

1503− status: if any toPresence (\p -> isLocalServer p.server)

1504− then MsDelivered {}

1505− else if forwardsEnqueued then MsFederated {} else MsQueued {},

1506− remoteServers

1471+ yield {error: "", remoteServers}

1507− }

1508

1472

1509-- Fire one /federation/forward call per pre-staged server. On success the

1473-- Fire one /federation/forward call per pre-staged server. On success the

1510-- corresponding row in `forwards` is pruned, and if no other forwards remain

1474-- corresponding row in `forwards` is pruned, and if no other forwards remain

@@ -1692,7 +1656,7 @@ handleSendBatch = \messages authorization clientIp respond -> do

1692 let effMsgId = messageIdOf m.encryptedBlob

1656 let effMsgId = messageIdOf m.encryptedBlob

1693 r <- atomic (do sendMessageCore senderKey m.to m.encryptedBlob t)

1657 r <- atomic (do sendMessageCore senderKey m.to m.encryptedBlob t)

1694 when federationEnabled (fork (forkForwardsTo m.to m.encryptedBlob r.remoteServers))

1658 when federationEnabled (fork (forkForwardsTo m.to m.encryptedBlob r.remoteServers))

1695− yield {messageId: effMsgId, r.status})

1659+ yield {messageId: effMsgId, r.error})

1696 yield (respond {results, authorized: True {}, retryAfterMs: 0}))

1660 yield (respond {results, authorized: True {}, retryAfterMs: 0}))

1697

1661

1698-- ============================================

1662-- ============================================