@@ -149,9 +149,6 @@ backgroundPruneInterval = 30000
149149-- Long-poll hold timeout before returning empty to the client (25 s).
150150pollTimeoutMs : Int<Ms>
151151pollTimeoutMs = 25000
152−-- Cadence at which this server refreshes its own presence to peers (5 min).
153−presenceRefreshInterval : Int<Ms>
154−presenceRefreshInterval = 300000
155152
156153-- Master switch for federation; when False the server runs in single-node mode.
157154federationEnabled : Bool
@@ -289,8 +286,6 @@ type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
289286*seqCounter : Timestamp
290287-- Bumped by send/heartbeat to wake all blocked /poll holders.
291288*pollWakeup : Timestamp
292−-- Wall time of the last self-presence refresh broadcast to peers.
293−*lastPresenceRefresh : Timestamp
294289
295290-- ============================================
296291-- Routes
@@ -769,40 +764,14 @@ forwardQueuedToServer = \pubkey server -> atomic (do
769764-- Background loops
770765-- ============================================
771766-- Composite tick run from `backgroundPrune`: expire stale rows, then
772−-- opportunistically retry pending forwards and refresh local presence so peers
773−-- don't see us go offline mid-session.
767+-- opportunistically retry pending forwards.
774768
775769-- One composite background tick: prune expired state, gossip offline for
776−-- locally-expired keys, retry pending forwards, and (occasionally) refresh
777−-- local presence so peers don't see us drop off.
770+-- locally-expired keys, and retry pending forwards.
778771pruneAndRetry = \t -> do
779772 expiredLocal <- pruneExpired t
780773 fork (broadcastOfflineGossip expiredLocal)
781− when federationEnabled (do
782− fork retryPendingForwards
783− lastRefresh <- *lastPresenceRefresh
784− when (t - lastRefresh >= presenceRefreshInterval) (do
785− *lastPresenceRefresh = t
786− refreshLocalPresence t))
787−
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
801− *presence = updated
802− yield {refreshed, peers: knownPeerServersFromPresence updated})
803− when (count refreshState.refreshed > 0) (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> GossipOnline {
804− p.pubkey
805− }))))
774+ when federationEnabled (fork retryPendingForwards)
806775
807776-- ============================================
808777-- Handler: challenge / verify
@@ -902,7 +871,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
902871
903872-- POST /poll: top-level dispatch. Validates the global rate gate, then hands
904873-- 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]})
906875handlePoll = \cursor authorization clientIp -> do
907876 t <- now
908877 g <- checkGlobalRate t
@@ -911,18 +880,32 @@ handlePoll = \cursor authorization clientIp -> do
911880 else pollOnce authorization cursor clientIp t
912881
913882-- Body of a single poll. Treats the cursor advance as an implicit ack
914−-- (`ackPolledMessages`), then runs `doPoll` to collect events (or block
915−-- 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]})
883+-- (`ackPolledMessages`), refreshes the caller's local presence row so peers
884+-- don't time us out mid-session, then runs `doPoll` to collect events (or
885+-- block until wakeup) and assembles the response.
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]})
917887pollOnce = \authorization cursor clientIp t -> do
918888 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
919889 let clampedCursor = clamp 0 t cursor
920890 when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)
921891 startWakeup <- *pollWakeup
922− yield (Just {value: {pubkey, clampedCursor, startWakeup}}))
892+ pres <- *presence
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
896+ then do
897+ let kept = removeWhere pres mine
898+ let newPres = union kept [{pubkey, server: serverHost, expiresAt: t + sessionExpiry}]
899+ *presence = newPres
900+ yield (Just {value: knownPeerServersFromPresence newPres})
901+ else yield (Nothing {})
902+ yield (Just {value: {pubkey, clampedCursor, startWakeup, refreshPeers}}))
923903 case setup of
924904 Nothing {} -> yield (Err {error: {status: 401, message: "unauthorized"}})
925905 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 {}
926909 pollResult <- doPoll s t
927910 yield (Ok {value: {pollResult.events, pollResult.cursor}})
928911
@@ -1213,7 +1196,7 @@ api = serve Api where
12131196-- Endless tick: prune immediately on first call, then every
12141197-- `backgroundPruneInterval`. The startup prune clears stale rows persisted
12151198-- 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
12171200backgroundPrune = do
12181201 t <- now
12191202 pruneAndRetry t