@@ -71,10 +71,9 @@ type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp
7171type AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
7272type PruneResult = {expiredLocal: [PresenceEntry], peerServers: [ServerName]}
7373type PollResult = PollUnauthorized {} | PollAuthorized {events: [PollEvent], cursor: Timestamp}
74−type VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [ServerName]}
75−type SendResult = SendUnauthorized {} | SendRejected {} | SendDone {status: MessageStatus, remoteServers: [ServerName], fromKey: PubkeyHex}
74+type DeliveryStatus = Text where \s -> s == "delivered" || s == "federated" || s == "queued"
75+type SendResult = SendUnauthorized {} | SendRejected {} | SendDone {status: DeliveryStatus, remoteServers: [ServerName], fromKey: PubkeyHex}
7676type AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
77−type LookupResult = LookupUnauthorized {} | LookupDone {foundServer: OptionalServerName}
7877
7978
8079-- ============================================
@@ -107,12 +106,12 @@ route AuthApi where
107106 -> {challenge: Text}
108107 = ReqChallenge
109108 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex} /verify
110− -> {token: Text, expiresAt: Int}
109+ -> {token: Text, expiresAt: Timestamp}
111110 = VerifyAuth
112111
113112route ClientApi where
114− POST {cursor: Int, timeout: Int} /poll headers {authorization: Text}
115− -> {events: [PollEvent], cursor: Int, authorized: Bool}
113+ POST {cursor: Timestamp, timeout: Int} /poll headers {authorization: Text}
114+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool}
116115 = Poll
117116 /messages
118117 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp} / headers {authorization: Text}
@@ -652,6 +651,33 @@ sendAcksAndClean = \pendingAcks servers ->
652651 sendToEach servers (\server ->
653652 sendBatchedAcksToServer (do a <- pendingAcks; where a.originServer == server; yield a) server)
654653
654+retryEvictedAcks = \evictedEntries pendingAcks ->
655+ let evictedToRetry = do
656+ a <- evictedEntries
657+ where none pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)
658+ yield a
659+ in when (count evictedToRetry > 0) (do
660+ let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
661+ sendToEach evictedOriginServers (\server -> do
662+ fetchResult <- fetch (federationScheme ++ server) (FederationAck {
663+ authorization: "Bearer " ++ federationSecret,
664+ messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
665+ originServer: server, fromServer: serverHostN})
666+ case fetchResult of
667+ Err _ -> do
668+ dropped <- atomic do
669+ ackedIds <- *ackedIds
670+ let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
671+ let combined = union ackedIds reinsert
672+ let excess = count combined - maxAckedIds
673+ let afterEviction = if excess <= 0 then combined
674+ else evictOldestAckedIds combined excess
675+ set *ackedIds = afterEviction
676+ yield excess
677+ when (dropped > 0)
678+ (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
679+ _ -> yield {}))
680+
655681pruneAndRetry = \t -> do
656682 result <- atomic do
657683 set *lastPrunedAt = t
@@ -743,7 +769,7 @@ serve = \req -> case req of
743769 let hasMatch = any challenges (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t)
744770 let sigValid = if hasMatch then verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature) else False {}
745771 if not sigValid
746− then yield {pr: pr, verify: VerifyFailed {}}
772+ then yield {pr: pr, verified: False {}, peerServers: []}
747773 else do
748774 set *challenges = do
749775 c <- challenges
@@ -762,20 +788,17 @@ serve = \req -> case req of
762788 let peerServers = if federationEnabled
763789 then collectPeerServers updatedPresence (\p -> p.server)
764790 else []
765− yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}
766− let expiredLocal = case result.verify of
767− VerifyOk _ -> do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p
768− _ -> result.pr.expiredLocal
791+ yield {pr: pr, verified: True {}, peerServers: peerServers}
792+ let expiredLocal = if result.verified
793+ then do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p
794+ else result.pr.expiredLocal
769795 fork (do
770− case result.verify of
771− VerifyOk {peerServers} ->
772− when federationEnabled
773− (sendBatchedGossip peerServers [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] 2)
774− _ -> yield {}
796+ when (result.verified && federationEnabled)
797+ (sendBatchedGossip result.peerServers [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] 2)
775798 broadcastOfflineGossip t result.pr.peerServers expiredLocal)
776− yield (case result.verify of
777− VerifyFailed _ -> respond {token: "", expiresAt: 0}
778− VerifyOk _ -> respond {token: token, expiresAt: expiry})
799+ yield (if result.verified
800+ then respond {token: token, expiresAt: expiry}
801+ else respond {token: "", expiresAt: 0})
779802
780803 Poll {cursor, timeout, authorization, respond} -> do
781804 t <- now
@@ -881,31 +904,7 @@ serve = \req -> case req of
881904 let originServers = collectPeerServers pendingAcks (\a -> a.originServer)
882905 when (count originServers > 0)
883906 (sendAcksAndClean pendingAcks originServers)
884− let evictedToRetry = do
885− a <- evictedEntries
886− where none pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)
887− yield a
888− when (count evictedToRetry > 0) (do
889− let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
890− sendToEach evictedOriginServers (\server -> do
891− fetchResult <- fetch (federationScheme ++ server) (FederationAck {
892− authorization: "Bearer " ++ federationSecret,
893− messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
894− originServer: server, fromServer: serverHostN})
895− case fetchResult of
896− Err _ -> do
897− dropped <- atomic do
898− ackedIds <- *ackedIds
899− let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
900− let combined = union ackedIds reinsert
901− let excess = count combined - maxAckedIds
902− let afterEviction = if excess <= 0 then combined
903− else evictOldestAckedIds combined excess
904− set *ackedIds = afterEviction
905− yield excess
906− when (dropped > 0)
907− (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
908− _ -> yield {})))
907+ retryEvictedAcks evictedEntries pendingAcks)
909908 _ -> yield {}
910909 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
911910 yield (case result.ack of
@@ -921,7 +920,7 @@ serve = \req -> case req of
921920 let auth = authenticateRequest sessions authorization t
922921 case auth of
923922 Unauthorized {} ->
924− yield {pr: pr, lookup: LookupUnauthorized {}}
923+ yield {pr: pr, authorized: False {}, foundServer: ""}
925924 Authorized _ -> do
926925 presence <- *presence
927926 let valid = do
@@ -933,11 +932,11 @@ serve = \req -> case req of
933932 else if p.server == serverHostN then {server: serverHostN, best: p.expiresAt}
934933 else if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt}
935934 else acc) {server: "", best: 0} valid).server
936− yield {pr: pr, lookup: LookupDone {foundServer: foundServer}}
935+ yield {pr: pr, authorized: True {}, foundServer: foundServer}
937936 forkBroadcastOffline t result.pr
938− yield (case result.lookup of
939− LookupUnauthorized _ -> respond {online: False {}, server: "", authorized: False {}}
940− LookupDone {foundServer} -> respond {online: not (foundServer == ""), server: foundServer, authorized: True {}})
937+ yield (if result.authorized
938+ then respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: True {}}
939+ else respond {online: False {}, server: "", authorized: False {}})
941940
942941 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
943942 t <- now