git.iko.soy/skrepka
on master help sign in
Replace sleep-based polling with STM retry and heartbeat for low-latency message delivery
7523d53iko
client/App.fs−3
compare with master
unifiedsplit
@@ -583,9 +583,6 @@ module App =
583 with ex ->
583 with ex ->
584 log $"delivery ack error: {ex.Message}"
584 log $"delivery ack error: {ex.Message}"
585
585
586− if response.Events.Length = 0 then
587− do! Async.Sleep Constants.pollEmptyDelayMs
588−
589 let chatCount =
586 let chatCount =
590 events
587 events
591 |> List.sumBy (fun e ->
588 |> List.sumBy (fun e ->
client/Constants.fs−1
@@ -13,7 +13,6 @@ module Constants =
13 let schemaVersion = 2
13 let schemaVersion = 2
14 let maxBitmapCacheSize = 128
14 let maxBitmapCacheSize = 128
15 let pollRetryBaseMs = 3000
15 let pollRetryBaseMs = 3000
16− let pollEmptyDelayMs = 5000
17 let outboxRetryDelayMs = 3000
16 let outboxRetryDelayMs = 3000
18 let maxMessages = 200
17 let maxMessages = 200
19 let blobVersion = 1uy
18 let blobVersion = 1uy
server.knot+36−19
@@ -47,6 +47,8 @@ gossipOriginHops = 2
47
47
48backgroundPruneInterval : Int<Ms>
48backgroundPruneInterval : Int<Ms>
49backgroundPruneInterval = 30000 -- 30 seconds
49backgroundPruneInterval = 30000 -- 30 seconds
50+pollTimeoutMs : Int<Ms>
51+pollTimeoutMs = 25000 -- long-poll heartbeat interval
50federationEnabled = federationSecret != ""
52federationEnabled = federationSecret != ""
51federationAuth = "Bearer " ++ federationSecret
53federationAuth = "Bearer " ++ federationSecret
52federationUrl = \server -> federationScheme ++ server
54federationUrl = \server -> federationScheme ++ server
@@ -137,6 +139,8 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
137
139
138*lastPrunedAt : Timestamp
140*lastPrunedAt : Timestamp
139
141
142+*pollWakeup : Timestamp
143+
140-- ============================================
144-- ============================================
141-- Routes
145-- Routes
142-- ============================================
146-- ============================================
@@ -816,7 +820,7 @@ pruneAndRetry = \t -> do
816 retryPendingForwards result.pendingForwards
820 retryPendingForwards result.pendingForwards
817 sendAcksAndClean result.pendingAcks)
821 sendAcksAndClean result.pendingAcks)
818
822
819−pollOnce = \token cursor -> do
823+pollOnce = \token cursor startWakeup -> do
820 t <- now
824 t <- now
821 sessions <- *sessions
825 sessions <- *sessions
822 let auth = authenticateSession sessions token t
826 let auth = authenticateSession sessions token t
@@ -825,23 +829,27 @@ pollOnce = \token cursor -> do
825 Authorized {pubkey} -> do
829 Authorized {pubkey} -> do
826 seqCounter <- *seqCounter
830 seqCounter <- *seqCounter
827 let clampedCursor = clamp 0 (max seqCounter t) cursor
831 let clampedCursor = clamp 0 (max seqCounter t) cursor
832+ atomic do
833+ wakeup <- *pollWakeup
828− allMessages <- *messages
834+ allMessages <- *messages
829− let matchingMsgs = do
835+ let matchingMsgs = do
830− m <- allMessages
836+ m <- allMessages
831− where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
837+ where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
832− yield m
838+ yield m
833− let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
839+ let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
834− if count limitedMsgs == 0
840+ if count limitedMsgs > 0
835− then yield (PollOk {events: [], cursor: clampedCursor})
841+ then
836− else
837− let evts = do
842+ let evts = do
838− m <- limitedMsgs
843+ m <- limitedMsgs
839− yield {id: m.messageId, eventType: "message",
844+ yield {id: m.messageId, eventType: "message",
840− payload: {toKey: m.toKey, fromKey: m.fromKey,
845+ payload: {toKey: m.toKey, fromKey: m.fromKey,
841− encryptedBlob: m.encryptedBlob,
846+ encryptedBlob: m.encryptedBlob,
842− timestamp: m.timestamp}}
847+ timestamp: m.timestamp}}
843− in let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
848+ in let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
844− in yield (PollOk {events: evts, cursor: lastReceivedAt})
849+ in yield (PollOk {events: evts, cursor: lastReceivedAt})
850+ else if wakeup > startWakeup
851+ then yield (PollOk {events: [], cursor: clampedCursor})
852+ else retry
845
853
846-- ============================================
854-- ============================================
847-- Request Handler
855-- Request Handler
@@ -910,9 +918,10 @@ serve = \req -> case req of
910 yield (respond {token: token, expiresAt: expiry})
918 yield (respond {token: token, expiresAt: expiry})
911
919
912 Poll {cursor, authorization, respond} -> do
920 Poll {cursor, authorization, respond} -> do
921+ startWakeup <- *pollWakeup
913 result <- case extractToken authorization of
922 result <- case extractToken authorization of
914 Nothing {} -> yield (PollUnauthorized {})
923 Nothing {} -> yield (PollUnauthorized {})
915− Just {value: token} -> pollOnce token cursor
924+ Just {value: token} -> pollOnce token cursor startWakeup
916 tAfterPoll <- now
925 tAfterPoll <- now
917 maybePruneAndBroadcast tAfterPoll
926 maybePruneAndBroadcast tAfterPoll
918 yield (respond (case result of
927 yield (respond (case result of
@@ -1124,7 +1133,15 @@ backgroundPrune = do
1124 pruneAndRetry t
1133 pruneAndRetry t
1125 backgroundPrune
1134 backgroundPrune
1126
1135
1136+pollHeartbeat = do
1137+ sleep pollTimeoutMs
1138+ t <- now
1139+ atomic do
1140+ set *pollWakeup = t
1141+ pollHeartbeat
1142+
1127main = do
1143main = do
1128 println "Skrepka relay server starting on port 8080..."
1144 println "Skrepka relay server starting on port 8080..."
1129 fork backgroundPrune
1145 fork backgroundPrune
1146+ fork pollHeartbeat
1130 listen 8080 serve
1147 listen 8080 serve