@@ -47,6 +47,8 @@ gossipOriginHops = 2
4747
4848backgroundPruneInterval : Int<Ms>
4949backgroundPruneInterval = 30000 -- 30 seconds
50+pollTimeoutMs : Int<Ms>
51+pollTimeoutMs = 25000 -- long-poll heartbeat interval
5052federationEnabled = federationSecret != ""
5153federationAuth = "Bearer " ++ federationSecret
5254federationUrl = \server -> federationScheme ++ server
@@ -137,6 +139,8 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
137139
138140*lastPrunedAt : Timestamp
139141
142+*pollWakeup : Timestamp
143+
140144-- ============================================
141145-- Routes
142146-- ============================================
@@ -816,7 +820,7 @@ pruneAndRetry = \t -> do
816820 retryPendingForwards result.pendingForwards
817821 sendAcksAndClean result.pendingAcks)
818822
819−pollOnce = \token cursor -> do
823+pollOnce = \token cursor startWakeup -> do
820824 t <- now
821825 sessions <- *sessions
822826 let auth = authenticateSession sessions token t
@@ -825,23 +829,27 @@ pollOnce = \token cursor -> do
825829 Authorized {pubkey} -> do
826830 seqCounter <- *seqCounter
827831 let clampedCursor = clamp 0 (max seqCounter t) cursor
828− allMessages <- *messages
829− let matchingMsgs = do
830− m <- allMessages
831− where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
832− yield m
833− let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
834− if count limitedMsgs == 0
835− then yield (PollOk {events: [], cursor: clampedCursor})
836− else
837− let evts = do
838− m <- limitedMsgs
839− yield {id: m.messageId, eventType: "message",
840− payload: {toKey: m.toKey, fromKey: m.fromKey,
841− encryptedBlob: m.encryptedBlob,
842− timestamp: m.timestamp}}
843− in let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
844− in yield (PollOk {events: evts, cursor: lastReceivedAt})
832+ atomic do
833+ wakeup <- *pollWakeup
834+ allMessages <- *messages
835+ let matchingMsgs = do
836+ m <- allMessages
837+ where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
838+ yield m
839+ let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
840+ if count limitedMsgs > 0
841+ then
842+ let evts = do
843+ m <- limitedMsgs
844+ yield {id: m.messageId, eventType: "message",
845+ payload: {toKey: m.toKey, fromKey: m.fromKey,
846+ encryptedBlob: m.encryptedBlob,
847+ timestamp: m.timestamp}}
848+ in let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
849+ in yield (PollOk {events: evts, cursor: lastReceivedAt})
850+ else if wakeup > startWakeup
851+ then yield (PollOk {events: [], cursor: clampedCursor})
852+ else retry
845853
846854-- ============================================
847855-- Request Handler
@@ -910,9 +918,10 @@ serve = \req -> case req of
910918 yield (respond {token: token, expiresAt: expiry})
911919
912920 Poll {cursor, authorization, respond} -> do
921+ startWakeup <- *pollWakeup
913922 result <- case extractToken authorization of
914923 Nothing {} -> yield (PollUnauthorized {})
915− Just {value: token} -> pollOnce token cursor
924+ Just {value: token} -> pollOnce token cursor startWakeup
916925 tAfterPoll <- now
917926 maybePruneAndBroadcast tAfterPoll
918927 yield (respond (case result of
@@ -1124,7 +1133,15 @@ backgroundPrune = do
11241133 pruneAndRetry t
11251134 backgroundPrune
11261135
1136+pollHeartbeat = do
1137+ sleep pollTimeoutMs
1138+ t <- now
1139+ atomic do
1140+ set *pollWakeup = t
1141+ pollHeartbeat
1142+
11271143main = do
11281144 println "Skrepka relay server starting on port 8080..."
11291145 fork backgroundPrune
1146+ fork pollHeartbeat
11301147 listen 8080 serve