@@ -180,21 +180,22 @@ requireClientIp = True {}
180-- polling pubkey. The server cannot read the blob.
180-- polling pubkey. The server cannot read the blob.
181type PollEvent = {encryptedBlob: BlobHex}
181type PollEvent = {encryptedBlob: BlobHex}
183−-- Kind of presence event flowing through federation gossip. `GossipUnknown` is
183+-- One presence change exchanged between federated servers. Per-variant
184+-- payloads keep room for future event kinds that aren't tied to a pubkey.
184−-- the parser's fallback for forward-compatibility with future event types.
185+-- `GossipUnknown` is the parser's fallback for forward-compatibility.
185−data GossipEventType = GossipOnline {} | GossipOffline {} | GossipUnknown {}
186+data GossipEventType
187+ = GossipOnline {pubkey: PubkeyHex}
188+ | GossipOffline {pubkey: PubkeyHex}
187impl ToJSON GossipEventType where
191impl ToJSON GossipEventType where
188− toJson et = toJson (case et of
192+ toJson et = case et of
189− GossipOnline {} -> "online"
193+ GossipOnline {pubkey} -> toJson {eventType: "online", pubkey}
190− GossipOffline {} -> "offline"
194+ GossipOffline {pubkey} -> toJson {eventType: "offline", pubkey}
191− GossipUnknown {} -> "unknown")
195+ GossipUnknown {} -> toJson {eventType: "unknown"}
193impl FromJSON GossipEventType where
197impl FromJSON GossipEventType where
194− parseJson t = (\s -> if s == "online" then GossipOnline {} else if s == "offline" then GossipOffline {} else GossipUnknown {}) (parseJson t : Text)
198+ parseJson t = (\raw -> if raw.eventType == "online" then GossipOnline {raw.pubkey} else if raw.eventType == "offline" then GossipOffline {raw.pubkey} else GossipUnknown {}) (parseJson t : {eventType: Text, pubkey: PubkeyHex})
196−-- One presence change exchanged between federated servers.
197−type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex}
199-- /messages is all-or-nothing per call: either every message in the batch is
200-- /messages is all-or-nothing per call: either every message in the batch is
200-- enqueued (delivered, federated, or buffered for federation), or none are
201-- enqueued (delivered, federated, or buffered for federation), or none are
@@ -322,7 +323,7 @@ route ClientApi where
322 POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, clientIp: Text} -> {} = SendMessages
323 POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, clientIp: Text} -> {} = SendMessages
324route FederationApi where
325route FederationApi where
325− POST {events: [GossipEvent], fromServer: ServerName} /federation/gossip -> {} = RecvGossip
326+ POST {events: [GossipEventType], fromServer: ServerName} /federation/gossip -> {} = RecvGossip
326 POST {toKey: PubkeyHex, encryptedBlob: BlobHex, fromServer: ServerName} /federation/forward -> {} = ForwardMessage
327 POST {toKey: PubkeyHex, encryptedBlob: BlobHex, fromServer: ServerName} /federation/forward -> {} = ForwardMessage
327 POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex}], fromServer: ServerName} /federation/forward-batch -> {results: [{ok: Bool}]} = FederationBatchForward
328 POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex}], fromServer: ServerName} /federation/forward-batch -> {results: [{ok: Bool}]} = FederationBatchForward
@@ -692,13 +693,9 @@ pruneExpired = \t -> atomic do
692-- a peer record into `forwardFailures` (shared with forward retries) so both
693-- a peer record into `forwardFailures` (shared with forward retries) so both
693-- subsystems back off together.
694-- subsystems back off together.
695−-- Build a gossip event with the originating pubkey's status.
696−mkGossip : GossipEventType -> PubkeyHex -> GossipEvent
697−mkGossip = \etype pk -> {eventType: etype, pubkey: pk}
699-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.
696-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.
700-- Records or clears federation-failure counters per call.
697-- Records or clears federation-failure counters per call.
701−sendBatchedGossip : [Text] -> [GossipEvent] -> IO {} {}
698+sendBatchedGossip : [Text] -> [GossipEventType] -> IO {} {}
702sendBatchedGossip = \peers events -> forEach peers (\server -> do
699sendBatchedGossip = \peers events -> forEach peers (\server -> do
704 failures <- *forwardFailures
701 failures <- *forwardFailures
@@ -713,7 +710,9 @@ sendBatchedGossip = \peers events -> forEach peers (\server -> do
713broadcastOfflineGossip : [{pubkey: PubkeyHex | a}] -> IO {r *presence} {}
710broadcastOfflineGossip : [{pubkey: PubkeyHex | a}] -> IO {r *presence} {}
714broadcastOfflineGossip = \expiredLocal -> do
711broadcastOfflineGossip = \expiredLocal -> do
715 presence <- *presence
712 presence <- *presence
716− when (federationEnabled && count expiredLocal > 0) (sendBatchedGossip (knownPeerServersFromPresence presence) (pluck expiredLocal (\p -> mkGossip (GossipOffline {}) p.pubkey)))
713+ when (federationEnabled && count expiredLocal > 0) (sendBatchedGossip (knownPeerServersFromPresence presence) (pluck expiredLocal (\p -> GossipOffline {
718-- ============================================
717-- ============================================
719-- Forward retries
718-- Forward retries
@@ -802,7 +801,9 @@ refreshLocalPresence = \t -> do
802 let updated = union kept refreshed
801 let updated = union kept refreshed
803 *presence = updated
802 *presence = updated
804 yield {refreshed, peers: knownPeerServersFromPresence updated})
803 yield {refreshed, peers: knownPeerServersFromPresence updated})
805− when (count refreshState.refreshed > 0) (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> mkGossip (GossipOnline {}) p.pubkey))))
804+ when (count refreshState.refreshed > 0) (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> GossipOnline {
807-- ============================================
808-- ============================================
808-- Handler: challenge / verify
809-- Handler: challenge / verify
@@ -893,9 +894,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
893 if not result.valid
894 if not result.valid
894 then yield (Err {error: {status: 401, message: "invalid"}})
895 then yield (Err {error: {status: 401, message: "invalid"}})
896− when federationEnabled (fork (sendBatchedGossip result.peers [
897+ when federationEnabled (fork (sendBatchedGossip result.peers [GossipOnline {pubkey}]))
897− mkGossip (GossipOnline {}) pubkey
899 yield (Ok {value: {token, expiresAt: expiry}})
898 yield (Ok {value: {token, expiresAt: expiry}})
901-- ============================================
900-- ============================================
@@ -930,9 +929,7 @@ pollOnce = \authorization cursor clientIp t -> do
930 Nothing {} -> yield (Err {error: {status: 401, message: "unauthorized"}})
929 Nothing {} -> yield (Err {error: {status: 401, message: "unauthorized"}})
931 Just {value: s} -> do
930 Just {value: s} -> do
932 pollResult <- doPoll s t
931 pollResult <- doPoll s t
934− value: {pollResult.events, pollResult.cursor}
932+ yield (Ok {value: {pollResult.events, pollResult.cursor}})
937-- One pass: collect events at `t`. If empty, retry-wait on `pollWakeup`
934-- One pass: collect events at `t`. If empty, retry-wait on `pollWakeup`
938-- (bumped by send/heartbeat) and re-collect once at the new wall time.
935-- (bumped by send/heartbeat) and re-collect once at the new wall time.
@@ -1000,14 +997,16 @@ handleSendMessages = \messages authorization clientIp -> do
1000 decision <- atomic (do
997 decision <- atomic (do
1001 checked <- mapEach limited (\m -> do
998 checked <- mapEach limited (\m -> do
1002 c <- checkSendRate senderKey m.to t
999 c <- checkSendRate senderKey m.to t
1003− yield {error: c.error, remoteServers: c.remoteServers, to: m.to, encryptedBlob: m.encryptedBlob})
1000+ yield {c.error, c.remoteServers, m.to, m.encryptedBlob})
1004 case findFirst checked (\c -> c.error != "") of
1001 case findFirst checked (\c -> c.error != "") of
1005− Just {value: bad} -> yield (Err {error: bad.error})
1002+ Just {value: bad} -> yield (Err {bad.error})
1006 Nothing {} -> do
1003 Nothing {} -> do
1007 forEach checked (\c -> commitSendMessage c.to c.encryptedBlob c.remoteServers t)
1004 forEach checked (\c -> commitSendMessage c.to c.encryptedBlob c.remoteServers t)
1008 yield (Ok {value: checked}))
1005 yield (Ok {value: checked}))
1009 case decision of
1006 case decision of
1007+ Err {error: code} -> yield (Err {
1010− Err {error: code} -> yield (Err {error: {status: if code == "self_send" then 400 else 429, message: code}})
1008+ error: {status: if code == "self_send" then 400 else 429, message: code}
1011 Ok {value: checked} -> do
1010 Ok {value: checked} -> do
1012 when federationEnabled (forEach checked (\c -> fork (forkForwardsTo c.to c.encryptedBlob c.remoteServers)))
1011 when federationEnabled (forEach checked (\c -> fork (forkForwardsTo c.to c.encryptedBlob c.remoteServers)))
1013 yield (Ok {value: {}}))
1012 yield (Ok {value: {}}))
@@ -1110,7 +1109,7 @@ forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
1110-- pubkeys with a `GossipOnline` event (so Online wins on collision). For keys
1109-- pubkeys with a `GossipOnline` event (so Online wins on collision). For keys
1111-- that transition offline → online, flushes any locally-buffered messages via
1110-- that transition offline → online, flushes any locally-buffered messages via
1112-- `forwardQueuedToServer`.
1111-- `forwardQueuedToServer`.
1113−handleRecvGossip : [{eventType: <GossipOffline {} | GossipOnline {} | a>, pubkey: PubkeyHex | b}] -> Text -> IO {rw *presence, clock} (Result HttpError {})
1112+handleRecvGossip : [GossipEventType] -> Text -> IO {rw *presence, clock} (Result HttpError {})
1114handleRecvGossip = \events fromServer -> do
1113handleRecvGossip = \events fromServer -> do
1115 t <- now
1114 t <- now
1116 if not isRemote fromServer || isBadServerName fromServer || count events > maxGossipEvents
1115 if not isRemote fromServer || isBadServerName fromServer || count events > maxGossipEvents
@@ -1118,10 +1117,18 @@ handleRecvGossip = \events fromServer -> do
1119 newlyOnline <- atomic (do
1118 newlyOnline <- atomic (do
1120 pres <- *presence
1119 pres <- *presence
1121− let touchedPubkeys = pluck events (\e -> e.pubkey)
1120+ let isOnline = \e -> case e of
1121+ GossipOnline {pubkey: _} -> True {}
1123+ let isOffline = \e -> case e of
1124+ GossipOffline {pubkey: _} -> True {}
1126+ let pubkeyOf = \e -> case e of
1127+ GossipOnline {pubkey} -> pubkey
1128+ GossipOffline {pubkey} -> pubkey
1122− let onlinePubkeys = pluck (filter events (\e -> case e.eventType of
1130+ let onlinePubkeys = pluck (filter events isOnline) pubkeyOf
1123− GossipOnline {} -> True {}
1131+ let touchedPubkeys = pluck (filter events (\e -> isOnline e || isOffline e)) pubkeyOf
1124− _ -> False {})) (\e -> e.pubkey)
1125 let kept = filter pres (\p -> p.server != fromServer || not elem p.pubkey touchedPubkeys)
1132 let kept = filter pres (\p -> p.server != fromServer || not elem p.pubkey touchedPubkeys)
1126 let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys
1133 let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys
1127 *presence = union kept added
1134 *presence = union kept added