@@ -399,9 +399,9 @@ clientIpOf = \h -> case h of
399-- Pull the token out of an `Authorization: Bearer <token>` header. Returns
399-- Pull the token out of an `Authorization: Bearer <token>` header. Returns
400-- Nothing if the prefix is missing or the token is empty.
400-- Nothing if the prefix is missing or the token is empty.
401extractToken : Text -> Maybe Text
401extractToken : Text -> Maybe Text
402−extractToken = \h -> if hasPrefix "Bearer " h && drop 7 h != ""
402+extractToken = \h -> (\token -> if hasPrefix "Bearer " h && token != ""
403−then Just {value: drop 7 h}
403+then Just {value: token}
404−else Nothing {}
404+else Nothing {}) (drop 7 h)
406-- Resolve a Bearer token to its session pubkey if the session is unexpired and
406-- Resolve a Bearer token to its session pubkey if the session is unexpired and
407-- the client IP matches. Runs `onAuth pubkey` on success, `onUnauth` otherwise.
407-- the client IP matches. Runs `onAuth pubkey` on success, `onUnauth` otherwise.
@@ -469,6 +469,10 @@ nextSeqTs = \t seqCounter -> max t (seqCounter + 1)
469presenceForKey : [PresenceEntry] -> PubkeyHex -> Timestamp -> [PresenceEntry]
469presenceForKey : [PresenceEntry] -> PubkeyHex -> Timestamp -> [PresenceEntry]
470presenceForKey = \presence pubkey t -> filter (\p -> p.pubkey == pubkey && p.expiresAt > t) presence
470presenceForKey = \presence pubkey t -> filter (\p -> p.pubkey == pubkey && p.expiresAt > t) presence
472+-- True when `pk` has a live local presence row (i.e. is currently online here).
473+hasLocalPresence : [PresenceEntry] -> PubkeyHex -> Timestamp -> Bool
474+hasLocalPresence = \pres pk t -> any (\p -> p.pubkey == pk && isLocalServer p.server && p.expiresAt > t) pres
472-- Project items through `getServer`, drop our own hostname, and dedupe.
476-- Project items through `getServer`, drop our own hostname, and dedupe.
473-- Used to derive peer recipient lists from forwards/presence/etc.
477-- Used to derive peer recipient lists from forwards/presence/etc.
474collectPeers : [a] -> (a -> Text) -> [Text]
478collectPeers : [a] -> (a -> Text) -> [Text]
@@ -502,8 +506,7 @@ pruneExpired = \t -> atomic do
502 -- presence: remember locally-expired keys for offline gossip
506 -- presence: remember locally-expired keys for offline gossip
503 pres <- *presence
507 pres <- *presence
504 *presence = filter (\p -> p.expiresAt > t) pres
508 *presence = filter (\p -> p.expiresAt > t) pres
505− let stillLive = \pk -> any (\k -> isLocalServer k.server && k.pubkey == pk && k.expiresAt > t) pres
506− yield (filter (\p -> p.expiresAt <= t && isLocalServer p.server && not (stillLive p.pubkey)) pres)
509+ yield (filter (\p -> p.expiresAt <= t && isLocalServer p.server && not (hasLocalPresence pres p.pubkey t)) pres)
508-- ============================================
511-- ============================================
509-- Federation gossip
512-- Federation gossip
@@ -576,19 +579,6 @@ forwardQueuedToServer = \pubkey server -> atomic (do
576 let newFwds = map (\m -> {toServer: server, m.toKey, m.encryptedBlob, retries: 0}) (take maxForwardsPerServer candidates)
579 let newFwds = map (\m -> {toServer: server, m.toKey, m.encryptedBlob, retries: 0}) (take maxForwardsPerServer candidates)
577 *forwards = union fwds newFwds)
580 *forwards = union fwds newFwds)
579−-- ============================================
581−-- ============================================
582−-- Composite tick run from `backgroundPrune`: expire stale rows, then
583−-- opportunistically retry pending forwards.
585−-- One composite background tick: prune expired state, gossip offline for
586−-- locally-expired keys, and retry pending forwards.
587−pruneAndRetry = \t -> do
588− expiredLocal <- pruneExpired t
589− fork (broadcastOfflineGossip expiredLocal)
590− when federationEnabled (fork retryPendingForwards)
592-- ============================================
582-- ============================================
593-- Handler: challenge / verify
583-- Handler: challenge / verify
594-- ============================================
584-- ============================================
@@ -610,19 +600,16 @@ handleChallenge : PubkeyHex -> Text -> IO {rw *challenges, clock, random} (Resul
610handleChallenge = \pubkey clientIp -> do
600handleChallenge = \pubkey clientIp -> do
612 challenge <- genToken
602 challenge <- genToken
613− full <- atomic (do
603+ atomic (do
614 ch <- *challenges
604 ch <- *challenges
615 let cappedActive = capPerKey (filter (\c -> c.expiresAt > t) ch) pubkey maxChallengesPerKey
605 let cappedActive = capPerKey (filter (\c -> c.expiresAt > t) ch) pubkey maxChallengesPerKey
616 if count cappedActive >= maxChallenges
606 if count cappedActive >= maxChallenges
617− then yield (True {})
607+ then yield (Err {error: {status: 503, message: "capacity"}})
619 *challenges = union cappedActive [
609 *challenges = union cappedActive [
620 {pubkey, challenge, expiresAt: t + challengeExpiry, ip: clientIp}
610 {pubkey, challenge, expiresAt: t + challengeExpiry, ip: clientIp}
624− then yield (Err {error: {status: 503, message: "capacity"}})
625− else yield (Ok {value: {challenge, expiresAt: t + challengeExpiry}})
612+ yield (Ok {value: {challenge, expiresAt: t + challengeExpiry}}))
627-- POST /auth/verify: validate `signature` over `challenge` for `pubkey`.
614-- POST /auth/verify: validate `signature` over `challenge` for `pubkey`.
628-- On success, mints a session token, records local presence (`pubkey` is now
615-- On success, mints a session token, records local presence (`pubkey` is now
@@ -671,19 +658,13 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
671-- * Wait on `pollWakeup` (bumped by send/heartbeat) and return new events
658-- * Wait on `pollWakeup` (bumped by send/heartbeat) and return new events
672-- plus piggybacked sender-side expirations/deliveries.
659-- plus piggybacked sender-side expirations/deliveries.
674−-- POST /poll: top-level dispatch. Hands off to `pollOnce` for the actual
676−handlePoll : Int -> Text -> Text -> IO {r *pollWakeup, r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, network, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
677−handlePoll = \cursor authorization clientIp -> do
679− pollOnce authorization cursor clientIp t
681−-- Body of a single poll. Treats the cursor advance as an implicit ack
661+-- POST /poll: long-poll body. Treats the cursor advance as an implicit ack
682-- (`ackPolledMessages`), refreshes the caller's local presence row so peers
662-- (`ackPolledMessages`), refreshes the caller's local presence row so peers
683-- don't time us out mid-session, then runs `doPoll` to collect events (or
663-- don't time us out mid-session, then runs `doPoll` to collect events (or
684-- block until wakeup) and assembles the response.
664-- block until wakeup) and assembles the response.
685−pollOnce : Text -> Int -> Text -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, network, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
665+handlePoll : Int -> Text -> Text -> IO {r *pollWakeup, r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, network, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
686−pollOnce = \authorization cursor clientIp t -> do
666+handlePoll = \cursor authorization clientIp -> do
687 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
668 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
688 let clampedCursor = clamp 0 t cursor
669 let clampedCursor = clamp 0 t cursor
689 when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)
670 when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)
@@ -825,7 +806,7 @@ forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
825 let remaining = filter (\x -> not same x) fwds2
806 let remaining = filter (\x -> not same x) fwds2
826 *forwards = remaining
807 *forwards = remaining
827 pres <- *presence
808 pres <- *presence
828− let stillLocal = any (\p -> p.pubkey == to && isLocalServer p.server && p.expiresAt > t) pres
809+ let stillLocal = hasLocalPresence pres to t
829 let othersPending = any (\x -> x.encryptedBlob == encryptedBlob) remaining
810 let othersPending = any (\x -> x.encryptedBlob == encryptedBlob) remaining
830 when (not stillLocal && not othersPending) (do
811 when (not stillLocal && not othersPending) (do
831 msgs <- *messages
812 msgs <- *messages
@@ -883,7 +864,7 @@ storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWa
883storeFederated = \toKey encryptedBlob t -> do
864storeFederated = \toKey encryptedBlob t -> do
884 pres <- *presence
865 pres <- *presence
885 msgs <- *messages
866 msgs <- *messages
886− if not any (\p -> p.pubkey == toKey && isLocalServer p.server && p.expiresAt > t) pres
867+ if not (hasLocalPresence pres toKey t)
887 then yield "no_presence"
868 then yield "no_presence"
889 seq <- *seqCounter
870 seq <- *seqCounter
@@ -925,18 +906,20 @@ api = serve Api where
925-- ============================================
906-- ============================================
926-- Background loops
907-- Background loops
927-- ============================================
908-- ============================================
928−-- `backgroundPrune` runs `pruneAndRetry` every `backgroundPruneInterval` ms.
909+-- `backgroundPrune` runs the prune/retry sweep every `backgroundPruneInterval` ms.
929-- `pollHeartbeat` bumps `pollWakeup` every `pollTimeoutMs` so long-poll
910-- `pollHeartbeat` bumps `pollWakeup` every `pollTimeoutMs` so long-poll
930-- holders unblock and return an empty heartbeat response, keeping connections
911-- holders unblock and return an empty heartbeat response, keeping connections
933−-- Endless tick: prune immediately on first call, then every
914+-- Endless tick: prune expired state, gossip offline for locally-expired keys,
934−-- `backgroundPruneInterval`. The startup prune clears stale rows persisted
915+-- and retry pending forwards. First sweep happens immediately so stale rows
935−-- by the previous process before the next interval elapses.
916+-- persisted by the previous process clear before the next interval elapses.
936backgroundPrune : IO {rw *challenges, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, rw *sessions, network, clock} a
917backgroundPrune : IO {rw *challenges, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, rw *sessions, network, clock} a
937backgroundPrune = do
918backgroundPrune = do
939− pruneAndRetry t
920+ expiredLocal <- pruneExpired t
921+ fork (broadcastOfflineGossip expiredLocal)
922+ when federationEnabled (fork retryPendingForwards)
940 sleep backgroundPruneInterval
923 sleep backgroundPruneInterval
941 backgroundPrune
924 backgroundPrune