@@ -336,16 +336,24 @@ clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresen
336336evictOldestAckedIds = \items n ->
337337 if n <= 0 then items
338338 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
340− in if oldest.ackedAt == -1 then items
341− else evictOldestAckedIds (fold (\acc a ->
342− if not acc.removed && a == oldest
343− then {removed: True {}, items: acc.items}
344− else {removed: acc.removed, items: union acc.items [a]}) {removed: False {}, items: []} items).items (n - 1)
339+ let toEvict = (fold (\acc a ->
340+ if acc.size < n
341+ then {size: acc.size + 1, heap: union acc.heap [a], heapMax: if a.ackedAt > acc.heapMax then a.ackedAt else acc.heapMax}
342+ else if a.ackedAt < acc.heapMax
343+ then
344+ let removed = fold (\r x ->
345+ if not r.done && x.ackedAt == acc.heapMax
346+ then {done: True {}, items: r.items}
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
345353
346354safeCompareLoop = \s1 s2 result ->
347− if s2 == ""
348− then result + (if s1 == "" then 0 else 1)
355+ if s1 == "" && s2 == ""
356+ then result
349357 else safeCompareLoop (drop 1 s1) (drop 1 s2) (result + (if take 1 s1 == take 1 s2 then 0 else 1))
350358
351359safeCompare = \a b ->
@@ -466,6 +474,71 @@ retryPendingForwards = \pending ->
466474 println ("prune: retry forward to " ++ f.toServer ++ " failed: " ++ show err))
467475 retryPendingForwards rest
468476
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+
469542-- ============================================
470543-- Request Handler
471544-- ============================================
@@ -602,47 +675,12 @@ serve = \req -> case req of
602675 let clampedTimeout = if timeout < 0 then 0 else if timeout > 60000 then 60000 else timeout
603676 let deadline = t + clampedTimeout
604677 let token = extractToken authorization
605− result <- atomic do
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}
678+ result <- pollLoop token cursor deadline
643679 respond {events: result.events, cursor: result.cursor, authorized: result.authorized}
644680 tAfterPoll <- now
645− broadcastOfflineGossip tAfterPoll result.peerServers result.expiredLocal
681+ pruneResult <- atomic do
682+ maybePrune tAfterPoll
683+ broadcastOfflineGossip tAfterPoll pruneResult.peerServers pruneResult.expiredLocal
646684
647685 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
648686 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
11581196 let ackOriginServers = fold (\acc a ->
11591197 if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
11601198 else union acc [a.originServer]) [] result.pendingAcks
1161− sendToEach ackOriginServers (\server ->
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}))
1199+ sendAcksAndClean result.pendingAcks ackOriginServers
11661200 else yield {}
11671201 else respond {ok: False {}}
11681202
@@ -1170,6 +1204,18 @@ serve = \req -> case req of
11701204-- Entry Point
11711205-- ============================================
11721206
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+
11731218main = do
11741219 println "Skrepka relay server starting on port 8080..."
1220+ spawn backgroundPrune
11751221 listen 8080 serve