@@ -257,7 +257,6 @@ type GossipEvent = {
257257 eventType: GossipEventType,
258258 pubkey: PubkeyHex,
259259 server: ServerName,
260− timestamp: Timestamp,
261260 seenBy: [Text]
262261}
263262
@@ -338,8 +337,7 @@ type Message = {toKey: PubkeyHex, encryptedBlob: BlobHex, receivedAt: Timestamp}
338337type PresenceEntry = {
339338 pubkey: PubkeyHex,
340339 server: AnyServerName,
341− expiresAt: Timestamp,
342− timestamp: Timestamp
340+ expiresAt: Timestamp
343341}
344342-- A message queued for cross-server delivery to `toServer`. `retries` is
345343-- incremented by `retryPendingForwards` until `maxForwardRetries` is hit.
@@ -986,9 +984,9 @@ clearFailure = \server -> atomic (do
986984-- ============================================
987985-- Common utility queries
988986-- ============================================
989−-- Pure helpers shared by send/poll/federation paths: TTL clamping, monotonic
990−-- sequence generation, presence lookup, duplicate detection across
991−-- (messages, deliveredIds), and last-write-wins presence merge.
987+-- Pure helpers shared by send/poll/federation paths: monotonic sequence
988+-- generation, presence lookup, and duplicate detection across
989+-- (messages, deliveredIds).
992990
993991-- Strictly-monotonic next sequence timestamp: max of wall clock and prior+1.
994992-- Guarantees /poll cursors order events even when many arrive in the same ms.
@@ -1027,10 +1025,6 @@ collectPeers = \items getServer -> unique (filter (pluck items getServer) isRemo
10271025knownPeerServersFromPresence : [PresenceEntry] -> [Text]
10281026knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.server)
10291027
1030−-- Last-write-wins merge by (pubkey, server), keeping the entry with the larger expiresAt.
1031−mergePresence : [PresenceEntry] -> [PresenceEntry] -> [PresenceEntry]
1032−mergePresence = \existing incoming -> fold (\acc e -> (\same -> if any acc (\x -> same x && x.expiresAt >= e.expiresAt) then acc else union (removeWhere acc same) [e]) (\x -> x.pubkey == e.pubkey && x.server == e.server)) existing incoming
1033−
10341028-- ============================================
10351029-- Pruning
10361030-- ============================================
@@ -1041,7 +1035,7 @@ mergePresence = \existing incoming -> fold (\acc e -> (\same -> if any acc (\x -
10411035-- One sweep over every state ref: drops expired rows in a single atomic block.
10421036-- Returns the locally-expired presence rows whose pubkey has no other live
10431037-- local session, so the caller can broadcast offline gossip for them.
1044−pruneExpired : Int<Ms> -> IO {rw *activePolls, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, rw *sessions} [{expiresAt: Timestamp, pubkey: PubkeyHex, server: AnyServerName, timestamp: Timestamp}]
1038+pruneExpired : Int<Ms> -> IO {rw *activePolls, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, rw *sessions} [{expiresAt: Timestamp, pubkey: PubkeyHex, server: AnyServerName}]
10451039pruneExpired = \t -> atomic do
10461040 ch <- *challenges
10471041 *challenges = filter ch (\x -> x.expiresAt > t)
@@ -1072,12 +1066,11 @@ pruneExpired = \t -> atomic do
10721066-- both subsystems back off together.
10731067
10741068-- Build a gossip event with this server's identity.
1075−mkGossip : GossipEventType -> PubkeyHex -> Timestamp -> GossipEvent
1076−mkGossip = \etype pk t -> {
1069+mkGossip : GossipEventType -> PubkeyHex -> GossipEvent
1070+mkGossip = \etype pk -> {
10771071 eventType: etype,
10781072 pubkey: pk,
10791073 server: serverHostN,
1080− timestamp: t,
10811074 seenBy: [serverHostN]
10821075}
10831076
@@ -1103,10 +1096,10 @@ sendBatchedGossip = \peers events -> forEach peers (\server -> do
11031096
11041097-- Tell known peers that the listed local pubkeys just went offline. No-op when
11051098-- federation is disabled or the list is empty.
1106−broadcastOfflineGossip : Int<Ms> -> [{pubkey: PubkeyHex | a}] -> IO {r *presence} {}
1107−broadcastOfflineGossip = \t expiredLocal -> do
1099+broadcastOfflineGossip : [{pubkey: PubkeyHex | a}] -> IO {r *presence} {}
1100+broadcastOfflineGossip = \expiredLocal -> do
11081101 presence <- *presence
1109− when (federationEnabled && count expiredLocal > 0) (sendBatchedGossip (knownPeerServersFromPresence presence) (pluck expiredLocal (\p -> mkGossip (GossipOffline {}) p.pubkey t)))
1102+ when (federationEnabled && count expiredLocal > 0) (sendBatchedGossip (knownPeerServersFromPresence presence) (pluck expiredLocal (\p -> mkGossip (GossipOffline {}) p.pubkey)))
11101103
11111104-- ============================================
11121105-- Forward retries
@@ -1182,7 +1175,7 @@ forwardQueuedToServer = \pubkey server t -> atomic (do
11821175pruneAndRetry = \t -> do
11831176 *lastPrunedAt = t
11841177 expiredLocal <- pruneExpired t
1185− fork (broadcastOfflineGossip t expiredLocal)
1178+ fork (broadcastOfflineGossip expiredLocal)
11861179 when federationEnabled (do
11871180 fork retryPendingForwards
11881181 lastRefresh <- *lastPresenceRefresh
@@ -1203,13 +1196,12 @@ refreshLocalPresence = \t -> do
12031196 let refreshed = mapWhere presence needsRefresh (\p -> {
12041197 p.pubkey,
12051198 p.server,
1206− expiresAt: t + sessionExpiry,
1207− timestamp: t
1199+ expiresAt: t + sessionExpiry
12081200 })
12091201 let updated = union (removeWhere presence needsRefresh) refreshed
12101202 replace *presence = updated
12111203 yield {refreshed, peers: knownPeerServersFromPresence updated})
1212− when (count refreshState.refreshed > 0) (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> mkGossip (GossipOnline {}) p.pubkey t))))
1204+ when (count refreshState.refreshed > 0) (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> mkGossip (GossipOnline {}) p.pubkey))))
12131205
12141206-- ============================================
12151207-- Handler: challenge / verify / revoke / logout
@@ -1299,7 +1291,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
12991291 pres <- *presence
13001292 let prunedPres = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
13011293 let newPres = union prunedPres [
1302− {pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}
1294+ {pubkey, server: serverHostN, expiresAt: expiry}
13031295 ]
13041296 replace *presence = newPres
13051297 yield {valid: True {}, peers: knownPeerServersFromPresence newPres})
@@ -1307,7 +1299,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
13071299 then yield (respond {token: "", expiresAt: 0, error: "invalid"})
13081300 else do
13091301 when federationEnabled (fork (sendBatchedGossip result.peers [
1310− mkGossip (GossipOnline {}) pubkey t
1302+ mkGossip (GossipOnline {}) pubkey
13111303 ]))
13121304 yield (respond {token, expiresAt: expiry, error: ""})
13131305
@@ -1348,7 +1340,7 @@ handleLogout = \authorization clientIp respond -> do
13481340 yield (knownPeerServersFromPresence pres)
13491341 yield {ok: True {}, pubkey, peers})
13501342 when (removed.ok && federationEnabled) (fork (sendBatchedGossip removed.peers [
1351− mkGossip (GossipOffline {}) removed.pubkey t
1343+ mkGossip (GossipOffline {}) removed.pubkey
13521344 ]))
13531345 yield (respond {removed.ok, authorized: removed.ok})
13541346
@@ -1754,45 +1746,33 @@ handleFederationPubkey = \respond -> do
17541746 hex <- *federationPublicKeyHex
17551747 yield (respond {pubkey: hex, ok: hex != ""})
17561748
1757−-- POST /federation/gossip: ingest peer presence events. Filters out events
1758−-- whose origin is local (loops) or whose timestamp is in the future beyond
1759−-- skew tolerance, applies last-write-wins merge per (pubkey, server), and
1760−-- — for keys that became newly online — flushes any locally-buffered messages
1761−-- for them via `forwardQueuedToServer` (the catch-up path).
1762−handleRecvGossip : [{eventType: <GossipOffline {} | GossipOnline {} | a>, pubkey: PubkeyHex, server: Text, timestamp: Int<Ms> | b}] -> Int -> Text -> Int -> Text -> ({error: Text, ok: Bool, retryAfterMs: Int} -> c) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, rw *presence, clock, network} c
1749+-- POST /federation/gossip: ingest peer presence events. Drops loop-back
1750+-- events (origin == this server), then applies each event in arrival order:
1751+-- `GossipOnline` upserts the (pubkey, server) row with a fresh receiver-side
1752+-- TTL, `GossipOffline` deletes it. For keys that transition offline → online,
1753+-- flushes any locally-buffered messages via `forwardQueuedToServer`.
1754+handleRecvGossip : [{eventType: <GossipOffline {} | GossipOnline {} | a>, pubkey: PubkeyHex, server: Text | b}] -> Int -> Text -> Int -> Text -> ({error: Text, ok: Bool, retryAfterMs: Int} -> c) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, rw *presence, clock, network} c
17631755handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorization respond -> do
17641756 t <- now
17651757 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (clamp 0 maxGossipHops hopsRemaining >= 1 && isRemote fromServer && count events <= maxGossipEvents) (do
17661758 newlyOnline <- atomic (do
17671759 pres <- *presence
1768− -- Clamp e.timestamp to receive time so a peer can't push it into the
1769− -- future to game the offline-vs-online LWW comparison below. The skew
1770− -- check is a redundant sanity filter against broken senders.
1771− let validEvents = do
1772− e <- events
1773− where isRemote e.server && e.timestamp <= t + clockSkewTolerance
1774− let effTs = min e.timestamp t
1775− yield {e.pubkey, e.server, effTs, e.eventType}
1776− let isOnline = \v -> case v.eventType of
1777− GossipOnline {} -> True {}
1778− _ -> False {}
1779− let isOffline = \v -> case v.eventType of
1780− GossipOffline {} -> True {}
1781− _ -> False {}
1782− -- Receiver anchors expiresAt to its own clock + onlineGossipTtl; senders
1783− -- don't carry a TTL on the wire.
1784− let online = mapWhere validEvents isOnline (\v -> {
1785− v.pubkey,
1786− v.server,
1787− expiresAt: t + onlineGossipTtl,
1788− timestamp: v.effTs
1789− })
1790− let offline = mapWhere validEvents isOffline (\v -> {v.pubkey, v.server, timestamp: v.effTs})
1791− *presence = mergePresence (filter pres (\p -> none offline (\o -> o.pubkey == p.pubkey && o.server == p.server && o.timestamp > p.timestamp))) online
1792− yield (mapWhere online (\e -> none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)) (\e -> {
1793− e.pubkey,
1794− e.server
1795− })))
1760+ let final = fold (\state e ->
1761+ if not isRemote e.server
1762+ then state
1763+ else (\matches -> case e.eventType of
1764+ GossipOnline {} -> {
1765+ pres: union (removeWhere state.pres matches) [{e.pubkey, e.server, expiresAt: t + onlineGossipTtl}],
1766+ newlyOnline: if any state.pres (\p -> matches p && p.expiresAt > t) then state.newlyOnline else union state.newlyOnline [{e.pubkey, e.server}]
1767+ }
1768+ GossipOffline {} -> {
1769+ pres: removeWhere state.pres matches,
1770+ state.newlyOnline
1771+ }
1772+ _ -> state) (\p -> p.pubkey == e.pubkey && p.server == e.server)
1773+ ) {pres, newlyOnline: []} events
1774+ *presence = final.pres
1775+ yield final.newlyOnline)
17961776 fork (forEach newlyOnline (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t))
17971777 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
17981778
@@ -1893,7 +1873,7 @@ handleDrain = \phase authorization respond -> withMaintAuth authorization (yield
18931873 *presence = removeWhere pres isLocalLive
18941874 *pollWakeup = t
18951875 yield {peers: knownPeerServersFromPresence pres, localKeys: filter pres isLocalLive})
1896− when (effectivePhase == 2 && count drainResult.localKeys > 0 && federationEnabled) (fork (sendBatchedGossip drainResult.peers (pluck drainResult.localKeys (\p -> mkGossip (GossipOffline {}) p.pubkey t))))
1876+ when (effectivePhase == 2 && count drainResult.localKeys > 0 && federationEnabled) (fork (sendBatchedGossip drainResult.peers (pluck drainResult.localKeys (\p -> mkGossip (GossipOffline {}) p.pubkey))))
18971877 yield (respond {ok: True {}, drained: count drainResult.localKeys, phase: effectivePhase}))
18981878
18991879-- POST /maintenance/undrain: clear drain state and resume normal operation.