@@ -141,7 +141,10 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
141141
142142*pollWakeup : Timestamp
143143
144−*messageSignal : Timestamp
144+*messageSignal0 : Timestamp
145+*messageSignal1 : Timestamp
146+*messageSignal2 : Timestamp
147+*messageSignal3 : Timestamp
145148
146149-- ============================================
147150-- Routes
@@ -150,7 +153,7 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
150153route AuthApi where
151154 /auth
152155 POST {pubkey: PubkeyHex} /challenge
153− -> {challenge: Text}
156+ -> {challenge: Text, ok: Bool}
154157 = ReqChallenge
155158 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex} /verify
156159 -> {token: Text, expiresAt: Timestamp}
@@ -188,7 +191,10 @@ route MaintenanceApi where
188191 -> {ok: Bool}
189192 = Prune
190193
191−route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi
194+route HealthApi where
195+ GET /health -> {ok: Bool} = Health
196+
197+route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi | HealthApi
192198
193199-- ============================================
194200-- Helpers
@@ -761,6 +767,28 @@ pruneAndRetry = \t -> do
761767 retryPendingForwards result.pendingForwards
762768 sendAcksAndClean result.pendingAcks)
763769
770+pubkeyBucket : PubkeyHex -> Int
771+pubkeyBucket = \pubkey ->
772+ let c = take 1 pubkey in
773+ if contains c "0123" then 0
774+ else if contains c "4567" then 1
775+ else if contains c "89ab" then 2
776+ else 3
777+
778+signalMessageForKey = \pubkey ts ->
779+ let b = pubkeyBucket pubkey in
780+ if b == 0 then set *messageSignal0 = ts
781+ else if b == 1 then set *messageSignal1 = ts
782+ else if b == 2 then set *messageSignal2 = ts
783+ else set *messageSignal3 = ts
784+
785+readMessageSignal = \pubkey ->
786+ let b = pubkeyBucket pubkey in
787+ if b == 0 then do s <- *messageSignal0; yield s
788+ else if b == 1 then do s <- *messageSignal1; yield s
789+ else if b == 2 then do s <- *messageSignal2; yield s
790+ else do s <- *messageSignal3; yield s
791+
764792pollOnce = \token cursor -> do
765793 t <- now
766794 sessions <- *sessions
@@ -773,7 +801,7 @@ pollOnce = \token cursor -> do
773801 startWakeup <- *pollWakeup
774802 atomic do
775803 wakeup <- *pollWakeup
776− messageSignal <- *messageSignal
804+ messageSignal <- readMessageSignal pubkey
777805 allMessages <- *messages
778806 let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages allMessages pubkey clampedCursor t))
779807 if count limitedMsgs > 0
@@ -907,46 +935,45 @@ serve = \req -> case req of
907935 when (not result.accepted)
908936 (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
909937 forkBroadcastOffline t result.pr
910− yield (respond {challenge: if result.accepted then challenge else ""})
938+ yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted})
911939
912940 VerifyAuth {pubkey, challenge, signature, respond} -> do
913941 t <- now
914− token <- genToken
915942 let expiry = t + sessionExpiry
916− result <- atomic do
943+ validation <- atomic do
917944 pr <- maybePrune t
918945 challenges <- *challenges
919− sessions <- *sessions
920− presence <- *presence
921946 let hasMatch = challengeMatch challenges pubkey challenge t
922947 let sigValid = hasMatch && verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
923948 if not sigValid
924− then yield {pr: pr, verify: VerifyFailed {}}
949+ then yield {pr: pr, valid: False {}}
925950 else do
926− let remainingChallenges = challengesExcluding challenges pubkey challenge
927− set *challenges = remainingChallenges
951+ set *challenges = challengesExcluding challenges pubkey challenge
952+ yield {pr: pr, valid: True {}}
953+ if validation.valid
954+ then do
955+ token <- genToken
956+ peerServers <- atomic do
957+ sessions <- *sessions
958+ presence <- *presence
928959 let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
929− let updatedSessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
930− set *sessions = updatedSessions
960+ set *sessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
931961 let presenceWithout = presenceWithoutLocalKey presence pubkey
932962 let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
933963 let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
934964 set *presence = updatedPresence
935− let peerServers = if federationEnabled
965+ yield (if federationEnabled
936966 then collectPeerServers updatedPresence (\p -> p.server)
937− else []
938− yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}
939− case result.verify of
940− VerifyFailed {} -> do
941− forkBroadcastOffline t result.pr
942− yield (respond {token: "", expiresAt: 0})
943− VerifyOk {peerServers} -> do
944− let expiredLocal = do p <- result.pr.expiredLocal; where p.pubkey != pubkey; yield p
967+ else [])
968+ let expiredLocal = do p <- validation.pr.expiredLocal; where p.pubkey != pubkey; yield p
945969 fork (do
946970 when federationEnabled
947971 (sendBatchedGossip peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops)
948− broadcastOfflineGossip t result.pr.peerServers expiredLocal)
972+ broadcastOfflineGossip t validation.pr.peerServers expiredLocal)
949973 yield (respond {token: token, expiresAt: expiry})
974+ else do
975+ forkBroadcastOffline t validation.pr
976+ yield (respond {token: "", expiresAt: 0})
950977
951978 Poll {cursor, authorization, respond} -> do
952979 result <- case extractToken authorization of
@@ -981,7 +1008,7 @@ serve = \req -> case req of
9811008 set *messages = union messages
9821009 [{messageId: effectiveMsgId, toKey: to,
9831010 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
984− set *messageSignal = seqTs
1011+ signalMessageForKey to seqTs
9851012 let toPresence = presenceForKey presence to t
9861013 let localOnline = hasLocalPresence toPresence
9871014 let remoteServers = collectPeerServers toPresence (\p -> p.server)
@@ -1109,8 +1136,11 @@ serve = \req -> case req of
11091136 [{messageId: messageId, toKey: toKey,
11101137 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
11111138 set *messages = updatedMessages
1112− set *messageSignal = seqTs
1139+ signalMessageForKey toKey seqTs
11131140 yield {pr: pr, forward: ForwardStored {}}
1141+ case result.forward of
1142+ ForwardRejected {} -> println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
1143+ _ -> yield {}
11141144 forkBroadcastOffline t result.pr
11151145 yield (respond {ok: case result.forward of
11161146 ForwardRejected {} -> False {}
@@ -1134,18 +1164,22 @@ serve = \req -> case req of
11341164 when authorized (fork (pruneAndRetry t))
11351165 yield (respond {ok: authorized})
11361166
1167+ Health {respond} -> yield (respond {ok: True {}})
1168+
11371169-- ============================================
11381170-- Entry Point
11391171-- ============================================
11401172
11411173backgroundPrune = do
1142− sleep backgroundPruneInterval
1174+ jitter <- randomInt 5000
1175+ sleep (backgroundPruneInterval + jitter)
11431176 t <- now
11441177 pruneAndRetry t
11451178 backgroundPrune
11461179
11471180pollHeartbeat = do
1148− sleep pollTimeoutMs
1181+ jitter <- randomInt 2000
1182+ sleep (pollTimeoutMs + jitter)
11491183 t <- now
11501184 atomic do
11511185 set *pollWakeup = t