56d04b5 · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Delete-on-poll: eliminate /messages/ack endpoint, delete messages atomically at poll time

56d04b5iko6 months ago

client/ApiClient.fs−6

compare with master

unifiedsplit

@@ -108,9 +108,3 @@ module ApiClient =

108 let body = JsonSerializer.Serialize({| cursor = cursor |})

109 postJson<PollResponse> pollClient $"{serverUrl}/poll" body (Some token)

110

111− let ackMessages (serverUrl: string) (token: string) (messageIds: string list) =

112− let body = JsonSerializer.Serialize({| messageIds = messageIds |})

113− async {

114− use! _ = sendRequest client $"{serverUrl}/messages/ack" body (Some token)

115− ()

116− }

client/App.fs−8

@@ -602,8 +602,6 @@ module App =

602 dispatch TokenExpired

603 else

604

605− let ackIds = response.Events |> Array.map _.Id |> Array.toList

606−

607 let results =

605 let results =

608 response.Events

606 response.Events

609 |> Array.choose (fun evt ->

607 |> Array.choose (fun evt ->

@@ -625,12 +623,6 @@ module App =

625 for e in errors do

623 for e in errors do

626 log $"decrypt error: {e}"

624 log $"decrypt error: {e}"

627

625

628− if not ackIds.IsEmpty then

629− try

630− do! ackMessages session.Url session.Token ackIds

631− with ex ->

632− log $"ack error: {ex.Message}"

633−

634 if not events.IsEmpty then

626 if not events.IsEmpty then

635 let acksBySender =

627 let acksBySender =

636 events

628 events

server.knot+27−50

@@ -114,8 +114,7 @@ type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: T

114type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}

115data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}

116data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName], forwardServers: [ServerName]} | SendDuplicate {}

117−data AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}

118−data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp}

117+data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp, pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}

119data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

118data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

120

119

121-- ============================================

120-- ============================================

@@ -166,9 +165,6 @@ route ClientApi where

166 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text}

165 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text}

167 -> {status: MessageStatus, messageId: OptionalMessageId}

166 -> {status: MessageStatus, messageId: OptionalMessageId}

168 = SendMessage

167 = SendMessage

169− POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text}

170− -> {ok: Bool}

171− = AckMessages

172 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text}

168 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text}

173 -> {server: AnyServerName, authorized: Bool}

169 -> {server: AnyServerName, authorized: Bool}

174 = LookupPresence

170 = LookupPresence

@@ -793,19 +789,30 @@ readMessageSignal = \pubkey ->

793 else if b == 2 then do s <- *messageSignal2; yield s

789 else if b == 2 then do s <- *messageSignal2; yield s

794 else do s <- *messageSignal3; yield s

790 else do s <- *messageSignal3; yield s

795

791

796−pollEvents = \allMessages pubkey clampedCursor t ->

792+pollEventsAndDelete = \pubkey clampedCursor t -> do

793+ messages <- *messages

797− let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages allMessages pubkey clampedCursor t)) in

794+ let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages messages pubkey clampedCursor t))

798 if count limitedMsgs > 0

795 if count limitedMsgs > 0

799− then

796+ then do

797+ set *messages = diff messages limitedMsgs

798+ let polledMsgIds = do m <- limitedMsgs; yield m.messageId

799+ forwards <- *forwards

800+ set *forwards = do f <- forwards; where none polledMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f

801+ let ackedFederated = federatedAckEntries limitedMsgs t

802+ ackedIds <- *ackedIds

803+ let combinedAckedIds = union ackedIds ackedFederated

804+ let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in

805+ if excessCount <= 0 then [] else takeRelation excessCount (sortBy (\a -> a.ackedAt) combinedAckedIds)

806+ set *ackedIds = diff combinedAckedIds evictedEntries

800 let evts = do

807 let evts = do

801 m <- limitedMsgs

808 m <- limitedMsgs

802 yield {id: m.messageId, eventType: "message",

809 yield {id: m.messageId, eventType: "message",

803 payload: {toKey: m.toKey,

810 payload: {toKey: m.toKey,

804 encryptedBlob: m.encryptedBlob,

811 encryptedBlob: m.encryptedBlob,

805 timestamp: m.timestamp}}

812 timestamp: m.timestamp}}

806− in let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs

813+ let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs

807− in PollOk {events: evts, cursor: lastReceivedAt}

814+ yield (PollOk {events: evts, cursor: lastReceivedAt, pendingAcks: ackedFederated, evictedEntries: evictedEntries})

808− else PollOk {events: [], cursor: clampedCursor}

815+ else yield (PollOk {events: [], cursor: clampedCursor, pendingAcks: [], evictedEntries: []})

809

816

810pollOnce = \token cursor -> do

817pollOnce = \token cursor -> do

811 t <- now

818 t <- now

@@ -822,7 +829,7 @@ pollOnce = \token cursor -> do

822 initialMessages <- *messages

829 initialMessages <- *messages

823 let initial = pollableMessages initialMessages pubkey clampedCursor t

830 let initial = pollableMessages initialMessages pubkey clampedCursor t

824 if count initial > 0

831 if count initial > 0

825− then yield (pollEvents initialMessages pubkey clampedCursor t)

832+ then atomic do pollEventsAndDelete pubkey clampedCursor t

826 else do

833 else do

827 -- Phase 2: wait for signal or heartbeat (no *messages dependency)

834 -- Phase 2: wait for signal or heartbeat (no *messages dependency)

828 atomic do

835 atomic do

@@ -831,9 +838,8 @@ pollOnce = \token cursor -> do

831 if wakeup > startWakeup || signal > startSignal

838 if wakeup > startWakeup || signal > startSignal

832 then yield {}

839 then yield {}

833 else retry

840 else retry

834− -- Phase 3: read messages after waking

841+ -- Phase 3: read messages after waking, delete atomically

835− newMessages <- *messages

836− yield (pollEvents newMessages pubkey clampedCursor t)

842+ atomic do pollEventsAndDelete pubkey clampedCursor t

837

843

838-- ============================================

844-- ============================================

839-- Derived Relations

845-- Derived Relations

@@ -860,9 +866,6 @@ bestServerForKey = \entries ->

860pollableMessages = \messages pubkey cursor t ->

866pollableMessages = \messages pubkey cursor t ->

861 do m <- messages; where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > cursor; yield m

867 do m <- messages; where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > cursor; yield m

862

868

863−matchingAckMessages = \messages pubkey messageIds ->

864− do m <- messages; where m.toKey == pubkey && any messageIds (\mid -> m.messageId == mid); yield m

865−

866federatedAckEntries = \ackedMessages t ->

869federatedAckEntries = \ackedMessages t ->

867 do m <- ackedMessages; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}

870 do m <- ackedMessages; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}

868

871

@@ -998,6 +1001,12 @@ serve = \req -> case req of

998 Just {value: token} -> pollOnce token cursor

1001 Just {value: token} -> pollOnce token cursor

999 tAfterPoll <- now

1002 tAfterPoll <- now

1000 maybePruneAndBroadcast tAfterPoll

1003 maybePruneAndBroadcast tAfterPoll

1004+ case result of

1005+ PollOk {pendingAcks, evictedEntries} ->

1006+ fork (when federationEnabled (do

1007+ sendAcksAndClean pendingAcks

1008+ retryEvictedAcks evictedEntries pendingAcks))

1009+ PollUnauthorized {} -> yield {}

1001 yield (respond (case result of

1010 yield (respond (case result of

1002 PollUnauthorized {} -> {events: [], cursor: 0, authorized: False {}}

1011 PollUnauthorized {} -> {events: [], cursor: 0, authorized: False {}}

1003 PollOk {events: evts, cursor: c} -> {events: evts, cursor: c, authorized: True {}}))

1012 PollOk {events: evts, cursor: c} -> {events: evts, cursor: c, authorized: True {}}))

@@ -1062,38 +1071,6 @@ serve = \req -> case req of

1062 SendDuplicate {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId}

1071 SendDuplicate {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId}

1063 SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})

1072 SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})

1064

1073

1065− AckMessages {messageIds, authorization, respond} -> do

1066− t <- now

1067− let dedupedIds = dedupMessageIds messageIds

1068− result <- atomic do

1069− pr <- maybePrune t

1070− withSessionAuth authorization t

1071− (yield {pr: pr, ack: AckUnauthorized {}})

1072− (\pubkey -> do

1073− messages <- *messages

1074− let ackedMessages = matchingAckMessages messages pubkey dedupedIds

1075− let remainingMessages = diff messages ackedMessages

1076− set *messages = remainingMessages

1077− forwards <- *forwards

1078− let ackedMsgIds = do m <- ackedMessages; yield m.messageId

1079− set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f

1080− let ackedFederated = federatedAckEntries ackedMessages t

1081− ackedIds <- *ackedIds

1082− let combinedAckedIds = union ackedIds ackedFederated

1083− let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in

1084− if excessCount <= 0 then [] else takeRelation excessCount (sortBy (\a -> a.ackedAt) combinedAckedIds)

1085− let remainingAckedIds = diff combinedAckedIds evictedEntries

1086− set *ackedIds = remainingAckedIds

1087− yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}})

1088− fork (do

1089− case result.ack of

1090− AckDone {pendingAcks, evictedEntries} ->

1091− when federationEnabled (do

1092− sendAcksAndClean pendingAcks

1093− retryEvictedAcks evictedEntries pendingAcks)

1094− AckUnauthorized {} -> yield {}

1095− broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)

1096− yield (respond {ok: case result.ack of AckDone _ -> True {}; _ -> False {}})

1097

1074

1098 LookupPresence {pubkey, authorization, respond} -> do

1075 LookupPresence {pubkey, authorization, respond} -> do

1099 t <- now

1076 t <- now