@@ -114,8 +114,7 @@ type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: T
114114type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
115115data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
116116data 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]}
119118data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
120119
121120-- ============================================
@@ -166,9 +165,6 @@ route ClientApi where
166165 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text}
167166 -> {status: MessageStatus, messageId: OptionalMessageId}
168167 = SendMessage
169− POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text}
170− -> {ok: Bool}
171− = AckMessages
172168 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text}
173169 -> {server: AnyServerName, authorized: Bool}
174170 = LookupPresence
@@ -793,19 +789,30 @@ readMessageSignal = \pubkey ->
793789 else if b == 2 then do s <- *messageSignal2; yield s
794790 else do s <- *messageSignal3; yield s
795791
796−pollEvents = \allMessages pubkey clampedCursor t ->
797− let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages allMessages pubkey clampedCursor t)) in
792+pollEventsAndDelete = \pubkey clampedCursor t -> do
793+ messages <- *messages
794+ let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages messages pubkey clampedCursor t))
798795 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
800807 let evts = do
801808 m <- limitedMsgs
802809 yield {id: m.messageId, eventType: "message",
803810 payload: {toKey: m.toKey,
804811 encryptedBlob: m.encryptedBlob,
805812 timestamp: m.timestamp}}
806− in let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
807− in PollOk {events: evts, cursor: lastReceivedAt}
808− else PollOk {events: [], cursor: clampedCursor}
813+ let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
814+ yield (PollOk {events: evts, cursor: lastReceivedAt, pendingAcks: ackedFederated, evictedEntries: evictedEntries})
815+ else yield (PollOk {events: [], cursor: clampedCursor, pendingAcks: [], evictedEntries: []})
809816
810817pollOnce = \token cursor -> do
811818 t <- now
@@ -822,7 +829,7 @@ pollOnce = \token cursor -> do
822829 initialMessages <- *messages
823830 let initial = pollableMessages initialMessages pubkey clampedCursor t
824831 if count initial > 0
825− then yield (pollEvents initialMessages pubkey clampedCursor t)
832+ then atomic do pollEventsAndDelete pubkey clampedCursor t
826833 else do
827834 -- Phase 2: wait for signal or heartbeat (no *messages dependency)
828835 atomic do
@@ -831,9 +838,8 @@ pollOnce = \token cursor -> do
831838 if wakeup > startWakeup || signal > startSignal
832839 then yield {}
833840 else retry
834− -- Phase 3: read messages after waking
835− newMessages <- *messages
836− yield (pollEvents newMessages pubkey clampedCursor t)
841+ -- Phase 3: read messages after waking, delete atomically
842+ atomic do pollEventsAndDelete pubkey clampedCursor t
837843
838844-- ============================================
839845-- Derived Relations
@@ -860,9 +866,6 @@ bestServerForKey = \entries ->
860866pollableMessages = \messages pubkey cursor t ->
861867 do m <- messages; where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > cursor; yield m
862868
863−matchingAckMessages = \messages pubkey messageIds ->
864− do m <- messages; where m.toKey == pubkey && any messageIds (\mid -> m.messageId == mid); yield m
865−
866869federatedAckEntries = \ackedMessages t ->
867870 do m <- ackedMessages; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
868871
@@ -998,6 +1001,12 @@ serve = \req -> case req of
9981001 Just {value: token} -> pollOnce token cursor
9991002 tAfterPoll <- now
10001003 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 {}
10011010 yield (respond (case result of
10021011 PollUnauthorized {} -> {events: [], cursor: 0, authorized: False {}}
10031012 PollOk {events: evts, cursor: c} -> {events: evts, cursor: c, authorized: True {}}))
@@ -1062,38 +1071,6 @@ serve = \req -> case req of
10621071 SendDuplicate {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId}
10631072 SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})
10641073
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 {}})
10971074
10981075 LookupPresence {pubkey, authorization, respond} -> do
10991076 t <- now