@@ -149,9 +149,6 @@ backgroundPruneInterval = 30000
149-- Long-poll hold timeout before returning empty to the client (25 s).
149-- Long-poll hold timeout before returning empty to the client (25 s).
150pollTimeoutMs : Int<Ms>
150pollTimeoutMs : Int<Ms>
151pollTimeoutMs = 25000
151pollTimeoutMs = 25000
152−-- Cadence at which this server refreshes its own presence to peers (5 min).
153−presenceRefreshInterval : Int<Ms>
154−presenceRefreshInterval = 300000
156-- Master switch for federation; when False the server runs in single-node mode.
153-- Master switch for federation; when False the server runs in single-node mode.
157federationEnabled : Bool
154federationEnabled : Bool
@@ -289,8 +286,6 @@ type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
289*seqCounter : Timestamp
286*seqCounter : Timestamp
290-- Bumped by send/heartbeat to wake all blocked /poll holders.
287-- Bumped by send/heartbeat to wake all blocked /poll holders.
291*pollWakeup : Timestamp
288*pollWakeup : Timestamp
292−-- Wall time of the last self-presence refresh broadcast to peers.
293−*lastPresenceRefresh : Timestamp
295-- ============================================
290-- ============================================
@@ -769,40 +764,14 @@ forwardQueuedToServer = \pubkey server -> atomic (do
769-- Background loops
764-- Background loops
770-- ============================================
765-- ============================================
771-- Composite tick run from `backgroundPrune`: expire stale rows, then
766-- Composite tick run from `backgroundPrune`: expire stale rows, then
772−-- opportunistically retry pending forwards and refresh local presence so peers
767+-- opportunistically retry pending forwards.
773−-- don't see us go offline mid-session.
775-- One composite background tick: prune expired state, gossip offline for
769-- One composite background tick: prune expired state, gossip offline for
776−-- locally-expired keys, retry pending forwards, and (occasionally) refresh
770+-- locally-expired keys, and retry pending forwards.
777−-- local presence so peers don't see us drop off.
778pruneAndRetry = \t -> do
771pruneAndRetry = \t -> do
779 expiredLocal <- pruneExpired t
772 expiredLocal <- pruneExpired t
780 fork (broadcastOfflineGossip expiredLocal)
773 fork (broadcastOfflineGossip expiredLocal)
781− when federationEnabled (do
782− fork retryPendingForwards
774+ when federationEnabled (fork retryPendingForwards)
783− lastRefresh <- *lastPresenceRefresh
784− when (t - lastRefresh >= presenceRefreshInterval) (do
785− *lastPresenceRefresh = t
786− refreshLocalPresence t))
788−-- Extend the expiry of local presence rows whose backing session is still live
789−-- but the presence entry is past half its window. Re-broadcasts an `online`
790−-- event for the refreshed entries so peers don't time them out.
791−refreshLocalPresence : Int<Ms> -> IO {rw *presence, r *sessions} {}
792−refreshLocalPresence = \t -> do
793− refreshState <- atomic (do
794− sessions <- *sessions
795− presence <- *presence
796− let sessionKeys = map (\s -> s.pubkey) sessions
797− let needsRefresh = \p -> isLocalServer p.server && elem p.pubkey sessionKeys && p.expiresAt - t < sessionExpiry / 2
798− let kept = filter presence (\p -> not needsRefresh p)
799− let refreshed = map (\p -> {p.pubkey, p.server, expiresAt: t + sessionExpiry}) (filter presence needsRefresh)
800− let updated = union kept refreshed
802− yield {refreshed, peers: knownPeerServersFromPresence updated})
803− when (count refreshState.refreshed > 0) (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> GossipOnline {
807-- ============================================
776-- ============================================
808-- Handler: challenge / verify
777-- Handler: challenge / verify
@@ -902,7 +871,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
903-- POST /poll: top-level dispatch. Validates the global rate gate, then hands
872-- POST /poll: top-level dispatch. Validates the global rate gate, then hands
904-- off to `pollOnce` for the actual long-poll body.
873-- off to `pollOnce` for the actual long-poll body.
905−handlePoll : Int -> Text -> Text -> IO {r *pollWakeup, r *sessions, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
874+handlePoll : Int -> Text -> Text -> IO {r *pollWakeup, r *sessions, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, rw *presence, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
906handlePoll = \cursor authorization clientIp -> do
875handlePoll = \cursor authorization clientIp -> do
908 g <- checkGlobalRate t
877 g <- checkGlobalRate t
@@ -911,18 +880,32 @@ handlePoll = \cursor authorization clientIp -> do
911 else pollOnce authorization cursor clientIp t
880 else pollOnce authorization cursor clientIp t
913-- Body of a single poll. Treats the cursor advance as an implicit ack
882-- Body of a single poll. Treats the cursor advance as an implicit ack
883+-- (`ackPolledMessages`), refreshes the caller's local presence row so peers
914−-- (`ackPolledMessages`), then runs `doPoll` to collect events (or block
884+-- don't time us out mid-session, then runs `doPoll` to collect events (or
915−-- until wakeup) and assembles the response.
885+-- block until wakeup) and assembles the response.
916−pollOnce : Text -> Int -> Text -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *forwards, rw *messages, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
886+pollOnce : Text -> Int -> Text -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *forwards, rw *messages, rw *presence, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
917pollOnce = \authorization cursor clientIp t -> do
887pollOnce = \authorization cursor clientIp t -> do
918 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
888 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
919 let clampedCursor = clamp 0 t cursor
889 let clampedCursor = clamp 0 t cursor
920 when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)
890 when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)
921 startWakeup <- *pollWakeup
891 startWakeup <- *pollWakeup
893+ let mine = \p -> p.pubkey == pubkey && isLocalServer p.server
894+ let stale = not (any pres (\p -> mine p && p.expiresAt - t >= sessionExpiry / 2))
895+ refreshPeers <- if stale
897+ let kept = removeWhere pres mine
898+ let newPres = union kept [{pubkey, server: serverHost, expiresAt: t + sessionExpiry}]
900+ yield (Just {value: knownPeerServersFromPresence newPres})
901+ else yield (Nothing {})
922− yield (Just {value: {pubkey, clampedCursor, startWakeup}}))
902+ yield (Just {value: {pubkey, clampedCursor, startWakeup, refreshPeers}}))
923 case setup of
903 case setup of
924 Nothing {} -> yield (Err {error: {status: 401, message: "unauthorized"}})
904 Nothing {} -> yield (Err {error: {status: 401, message: "unauthorized"}})
925 Just {value: s} -> do
905 Just {value: s} -> do
906+ case s.refreshPeers of
907+ Just {value: peers} -> when federationEnabled (fork (sendBatchedGossip peers [GossipOnline {pubkey: s.pubkey}]))
908+ Nothing {} -> yield {}
926 pollResult <- doPoll s t
909 pollResult <- doPoll s t
927 yield (Ok {value: {pollResult.events, pollResult.cursor}})
910 yield (Ok {value: {pollResult.events, pollResult.cursor}})
@@ -1213,7 +1196,7 @@ api = serve Api where
1213-- Endless tick: prune immediately on first call, then every
1196-- Endless tick: prune immediately on first call, then every
1214-- `backgroundPruneInterval`. The startup prune clears stale rows persisted
1197-- `backgroundPruneInterval`. The startup prune clears stale rows persisted
1215-- by the previous process before the next interval elapses.
1198-- by the previous process before the next interval elapses.
1216−backgroundPrune : IO {rw *challenges, rw *forwardFailures, r *forwards, rw *lastPresenceRefresh, rw *messages, rw *presence, rw *sessions, clock} a
1199+backgroundPrune : IO {rw *challenges, rw *forwardFailures, r *forwards, rw *messages, rw *presence, rw *sessions, clock} a
1217backgroundPrune = do
1200backgroundPrune = do
1218 t <- now
1201 t <- now
1219 pruneAndRetry t
1202 pruneAndRetry t