git.iko.soy/skrepka
on master help sign in
Fix three bugs: duplicate delivery on failed acks, long-poll deadline hang, and thundering herd
5a23463iko
server.knot+98−52
compare with master
unifiedsplit
@@ -336,16 +336,24 @@ clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresen
336evictOldestAckedIds = \items n ->
336evictOldestAckedIds = \items n ->
337 if n <= 0 then items
337 if n <= 0 then items
338 else
338 else
339− let oldest = fold (\acc a -> if acc.ackedAt == -1 || a.ackedAt < acc.ackedAt then a else acc) {messageId: "", toKey: "", ackedAt: -1, originServer: ""} items
339+ let toEvict = (fold (\acc a ->
340− in if oldest.ackedAt == -1 then items
340+ if acc.size < n
341− else evictOldestAckedIds (fold (\acc a ->
341+ then {size: acc.size + 1, heap: union acc.heap [a], heapMax: if a.ackedAt > acc.heapMax then a.ackedAt else acc.heapMax}
342− if not acc.removed && a == oldest
342+ else if a.ackedAt < acc.heapMax
343+ then
344+ let removed = fold (\r x ->
345+ if not r.done && x.ackedAt == acc.heapMax
343− then {removed: True {}, items: acc.items}
346+ then {done: True {}, items: r.items}
344− else {removed: acc.removed, items: union acc.items [a]}) {removed: False {}, items: []} items).items (n - 1)
347+ else {done: r.done, items: union r.items [x]}) {done: False {}, items: []} acc.heap
348+ let newHeap = union removed.items [a]
349+ let newMax = fold (\best x -> if x.ackedAt > best then x.ackedAt else best) -1 newHeap
350+ {size: acc.size, heap: newHeap, heapMax: newMax}
351+ else acc) {size: 0, heap: [], heapMax: -1} items).heap
352+ in diff items toEvict
345
353
346safeCompareLoop = \s1 s2 result ->
354safeCompareLoop = \s1 s2 result ->
347− if s2 == ""
355+ if s1 == "" && s2 == ""
348− then result + (if s1 == "" then 0 else 1)
356+ then result
349 else safeCompareLoop (drop 1 s1) (drop 1 s2) (result + (if take 1 s1 == take 1 s2 then 0 else 1))
357 else safeCompareLoop (drop 1 s1) (drop 1 s2) (result + (if take 1 s1 == take 1 s2 then 0 else 1))
350
358
351safeCompare = \a b ->
359safeCompare = \a b ->
@@ -466,6 +474,71 @@ retryPendingForwards = \pending ->
466 println ("prune: retry forward to " ++ f.toServer ++ " failed: " ++ show err))
474 println ("prune: retry forward to " ++ f.toServer ++ " failed: " ++ show err))
467 retryPendingForwards rest
475 retryPendingForwards rest
468
476
477+sendAcksAndClean = \pendingAcks servers ->
478+ if count servers == 0
479+ then yield {}
480+ else do
481+ let server = fold (\acc s -> if acc == "" then s else acc) "" servers
482+ let rest = do s <- servers; where not (s == server); yield s
483+ if server == ""
484+ then yield {}
485+ else do
486+ let serverAcks = do a <- pendingAcks; where a.originServer == server; yield a
487+ catch (do
488+ fetch (federationScheme ++ server) (FederationAck {
489+ authorization: "Bearer " ++ federationSecret,
490+ messageIds: (do a <- serverAcks; yield a.messageId),
491+ originServer: server, fromServer: serverHostN})
492+ atomic do
493+ ackedIds <- *ackedIds
494+ set *ackedIds = diff ackedIds serverAcks) (\err ->
495+ println ("federation: error sending ack to " ++ server ++ ": " ++ show err))
496+ sendAcksAndClean pendingAcks rest
497+
498+pollOnce = \token cursor ->
499+ atomic do
500+ t <- now
501+ sessions <- *sessions
502+ messages <- *messages
503+ let session = lookupSession sessions token t
504+ let clampedCursor = if cursor < 0 then 0 else if cursor > t then t else cursor
505+ let matchingMsgs = case session of
506+ Nothing {} -> []
507+ Just {value: s} -> do
508+ m <- messages
509+ where m.toKey == s.pubkey
510+ where m.receivedAt >= t - messageMaxAge
511+ where m.receivedAt > clampedCursor
512+ yield m
513+ let limitedMsgs = fold (\acc m ->
514+ if count acc >= maxPollEvents then acc
515+ else union acc [m]) [] matchingMsgs
516+ let authorized = case session of
517+ Nothing {} -> False {}
518+ _ -> True {}
519+ let evts = do
520+ m <- limitedMsgs
521+ yield {id: m.messageId, eventType: "message",
522+ payload: toJson {toKey: m.toKey, fromKey: m.fromKey,
523+ encryptedBlob: m.encryptedBlob,
524+ timestamp: m.timestamp}}
525+ let lastReceivedAt = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) clampedCursor limitedMsgs
526+ let newCursor = case session of
527+ Nothing {} -> -1
528+ _ -> lastReceivedAt
529+ yield {events: evts, cursor: newCursor, authorized: authorized}
530+
531+pollLoop = \token cursor deadline -> do
532+ result <- pollOnce token cursor
533+ t <- now
534+ if result.authorized && count result.events == 0 && t < deadline
535+ then do
536+ let remaining = deadline - t
537+ let sleepMs = if remaining > 500 then 500 else remaining
538+ sleep sleepMs
539+ pollLoop token cursor deadline
540+ else yield result
541+
469-- ============================================
542-- ============================================
470-- Request Handler
543-- Request Handler
471-- ============================================
544-- ============================================
@@ -602,47 +675,12 @@ serve = \req -> case req of
602 let clampedTimeout = if timeout < 0 then 0 else if timeout > 60000 then 60000 else timeout
675 let clampedTimeout = if timeout < 0 then 0 else if timeout > 60000 then 60000 else timeout
603 let deadline = t + clampedTimeout
676 let deadline = t + clampedTimeout
604 let token = extractToken authorization
677 let token = extractToken authorization
605− result <- atomic do
678+ result <- pollLoop token cursor deadline
606− tNow <- now
607− pr <- maybePrune tNow
608− sessions <- *sessions
609− messages <- *messages
610− seqCounter <- *seqCounter
611− let session = lookupSession sessions token tNow
612− let clampedCursor = if cursor < 0 then 0 else if cursor > seqCounter then seqCounter else cursor
613− let matchingMsgs = case session of
614− Nothing {} -> []
615− Just {value: s} -> do
616− m <- messages
617− where m.toKey == s.pubkey
618− where m.receivedAt >= tNow - messageMaxAge
619− where m.receivedAt > clampedCursor
620− yield m
621− let limitedMsgs = fold (\acc m ->
622− if count acc >= maxPollEvents then acc
623− else union acc [m]) [] matchingMsgs
624− let authorized = case session of
625− Nothing {} -> False {}
626− _ -> True {}
627− if authorized && count limitedMsgs == 0 && tNow < deadline
628− then retry
629− else do
630− let evts = do
631− m <- limitedMsgs
632− yield {id: m.messageId, eventType: "message",
633− payload: toJson {toKey: m.toKey, fromKey: m.fromKey,
634− encryptedBlob: m.encryptedBlob,
635− timestamp: m.timestamp}}
636− let lastReceivedAt = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) clampedCursor limitedMsgs
637− let newCursor = case session of
638− Nothing {} -> -1
639− _ -> if count limitedMsgs >= maxPollEvents
640− then lastReceivedAt
641− else if seqCounter > clampedCursor then seqCounter else clampedCursor
642− yield {events: evts, cursor: newCursor, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
643 respond {events: result.events, cursor: result.cursor, authorized: result.authorized}
679 respond {events: result.events, cursor: result.cursor, authorized: result.authorized}
644 tAfterPoll <- now
680 tAfterPoll <- now
681+ pruneResult <- atomic do
682+ maybePrune tAfterPoll
645− broadcastOfflineGossip tAfterPoll result.peerServers result.expiredLocal
683+ broadcastOfflineGossip tAfterPoll pruneResult.peerServers pruneResult.expiredLocal
646
684
647 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
685 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
648 let validInput = not (to == "") && not (encryptedBlob == "") && not (tooLong to maxPubkeyLen) && not (tooLong encryptedBlob maxBlobLen) && isValidHex to && isValidHex encryptedBlob
686 let validInput = not (to == "") && not (encryptedBlob == "") && not (tooLong to maxPubkeyLen) && not (tooLong encryptedBlob maxBlobLen) && isValidHex to && isValidHex encryptedBlob
@@ -1158,11 +1196,7 @@ serve = \req -> case req of
1158 let ackOriginServers = fold (\acc a ->
1196 let ackOriginServers = fold (\acc a ->
1159 if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1197 if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1160 else union acc [a.originServer]) [] result.pendingAcks
1198 else union acc [a.originServer]) [] result.pendingAcks
1161− sendToEach ackOriginServers (\server ->
1199+ sendAcksAndClean result.pendingAcks ackOriginServers
1162− fetch (federationScheme ++ server) (FederationAck {
1163− authorization: "Bearer " ++ federationSecret,
1164− messageIds: (do a <- result.pendingAcks; where a.originServer == server; yield a.messageId),
1165− originServer: server, fromServer: serverHostN}))
1166 else yield {}
1200 else yield {}
1167 else respond {ok: False {}}
1201 else respond {ok: False {}}
1168
1202
@@ -1170,6 +1204,18 @@ serve = \req -> case req of
1170-- Entry Point
1204-- Entry Point
1171-- ============================================
1205-- ============================================
1172
1206
1207+backgroundPrune = do
1208+ sleep 30000
1209+ t <- now
1210+ catch (do
1211+ pruneResult <- atomic do
1212+ set *lastPrunedAt = t
1213+ pruneExpired t
1214+ broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal) (\err ->
1215+ println ("background prune: " ++ show err))
1216+ backgroundPrune
1217+
1173main = do
1218main = do
1174 println "Skrepka relay server starting on port 8080..."
1219 println "Skrepka relay server starting on port 8080..."
1220+ spawn backgroundPrune
1175 listen 8080 serve
1221 listen 8080 serve