@@ -634,16 +634,13 @@ generateFederationAuth = \scope requestTimestamp ->
634634 if serverFederationPrivateKeyHex != ""
635635 then do
636636 nonce <- genTokenN 48
637− let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN ++ "|" ++ nonce)
638− let sig = sign (bytesFromHex serverFederationPrivateKeyHex) message
639− yield ("Signed " ++ bytesToHex sig ++ ":" ++ nonce)
637+ yield ("Signed " ++ bytesToHex (sign (bytesFromHex serverFederationPrivateKeyHex) (textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN ++ "|" ++ nonce))) ++ ":" ++ nonce)
640638 else do
641639 token <- genToken
642640 t <- now
643641 atomic do
644642 pending <- *pendingVerifications
645− let updated = union pending [{token: token, createdAt: t, scope: scope}]
646− set *pendingVerifications = capList updated maxPendingVerifications (\p -> p.createdAt)
643+ set *pendingVerifications = capList (union pending [{token: token, createdAt: t, scope: scope}]) maxPendingVerifications (\p -> p.createdAt)
647644 yield ("Bearer " ++ token)
648645
649646-- Verify a federation request: signed (Ed25519 over scope|ts|fromServer|nonce) or callback Bearer token.
@@ -661,9 +658,7 @@ verifyFederationAuth = \authorization fromServer scope requestTimestamp -> do
661658 case result of
662659 Ok r ->
663660 if not r.ok || r.pubkey == "" then yield False
664− else
665− let msg = scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer ++ (if parts.nonce == "" then "" else "|" ++ parts.nonce) in
666− yield (verify (bytesFromHex r.pubkey) (textToBytes msg) (bytesFromHex parts.sig))
661+ else yield (verify (bytesFromHex r.pubkey) (textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer ++ (if parts.nonce == "" then "" else "|" ++ parts.nonce))) (bytesFromHex parts.sig))
667662 _ -> yield False
668663 else case extractToken authorization of
669664 Just {value: token} -> do
@@ -760,10 +755,10 @@ recordFailure = \server -> do
760755 t <- now
761756 atomic do
762757 failures <- *forwardFailures
763− let new = case findFirst failures (\f -> f.server == server) of
764− Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
765− Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}
766− set *forwardFailures = union (removeWhere failures (\f -> f.server == server)) [new]
758+ set *forwardFailures = union (removeWhere failures (\f -> f.server == server))
759+ [case findFirst failures (\f -> f.server == server) of
760+ Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
761+ Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}]
767762
768763clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))
769764
@@ -852,9 +847,8 @@ mkAckedEntry = \m t -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, orig
852847recordAcksAndDeliveries = \toAck t -> do
853848 let ackedFed = mapWhere toAck (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
854849 appendCapped ackedIds ackedFed maxAckedIds (\a -> a.ackedAt)
855− let localDel = mapWhere toAck (\m -> m.senderKey != "" && m.originServer == "")
856− (\m -> {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t})
857− appendCapped senderDeliveries localDel 10000 (\d -> d.deliveredAt)
850+ appendCapped senderDeliveries (mapWhere toAck (\m -> m.senderKey != "" && m.originServer == "")
851+ (\m -> {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t})) 10000 (\d -> d.deliveredAt)
858852 yield ackedFed
859853
860854-- Common cancel block: drop matching messages + forwards, record cancel notification.
@@ -894,10 +888,8 @@ retryPendingForwards = do
894888 fwds <- *forwards
895889 failures <- *forwardFailures
896890 blocked <- *blockedPeers
897− let servers = filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures && not (elem blocked s))
898− forEach servers id (\server -> do
899− let serverFwds = takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))
900− forEach serverFwds (\f -> f.messageId) (\f -> do
891+ forEach (filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures && not (elem blocked s))) id (\server ->
892+ forEach (takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))) (\f -> f.messageId) (\f -> do
901893 t2 <- now
902894 auth <- generateFederationAuth ("forward:" ++ f.messageId) t2
903895 result <- fetch (federationUrl server) (ForwardMessage {
@@ -937,9 +929,8 @@ sendAcksAndClean = \pendingAcks ->
937929forwardQueuedToServer = \pubkey server t -> atomic do
938930 msgs <- *messages
939931 fwds <- *forwards
940− let candidates = mapWhere msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.messageId == m.messageId && f.toServer == server)) (\m ->
941− {messageId: m.messageId, toServer: server, createdAt: t, toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, originServer: (if m.originServer == "" then serverHostN else m.originServer), retries: 0, expiresAt: m.expiresAt, priority: m.priority, senderKey: m.senderKey})
942− set *forwards = capList (union fwds (takeRelation maxForwardsPerServer candidates)) maxForwardsEntries (\f -> f.createdAt)
932+ set *forwards = capList (union fwds (takeRelation maxForwardsPerServer (mapWhere msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.messageId == m.messageId && f.toServer == server)) (\m ->
933+ {messageId: m.messageId, toServer: server, createdAt: t, toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, originServer: (if m.originServer == "" then serverHostN else m.originServer), retries: 0, expiresAt: m.expiresAt, priority: m.priority, senderKey: m.senderKey})))) maxForwardsEntries (\f -> f.createdAt)
943934
944935-- ============================================
945936-- Background loops
@@ -964,16 +955,14 @@ refreshLocalPresence = \t -> do
964955 refreshState <- atomic do
965956 sessions <- *sessions
966957 presence <- *presence
967− let activeKeys = unique (pluck sessions (\s -> s.pubkey))
968958 let updated = pluck presence (\p ->
969− if isLocalServer p.server && elem activeKeys p.pubkey && p.expiresAt - t < sessionExpiry / 2
959+ if isLocalServer p.server && elem (unique (pluck sessions (\s -> s.pubkey))) p.pubkey && p.expiresAt - t < sessionExpiry / 2
970960 then {pubkey: p.pubkey, server: p.server, expiresAt: t + sessionExpiry, timestamp: t}
971961 else p)
972962 set *presence = updated
973963 yield {refreshed: filter updated (\p -> isLocalServer p.server && p.expiresAt == t + sessionExpiry), peers: knownPeerServersFromPresence updated}
974− when (count refreshState.refreshed > 0 && count refreshState.peers > 0) (do
975− let events = pluck refreshState.refreshed (\p -> mkGossip (GossipOnline {}) p.pubkey t (clampTtl (sessionExpiry / 1000)))
976− fork (sendBatchedGossip refreshState.peers events gossipOriginHops))
964+ when (count refreshState.refreshed > 0 && count refreshState.peers > 0)
965+ (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> mkGossip (GossipOnline {}) p.pubkey t (clampTtl (sessionExpiry / 1000)))) gossipOriginHops))
977966
978967-- ============================================
979968-- Handler: challenge / verify / revoke / logout
@@ -1017,8 +1006,8 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
10171006 ses <- *sessions
10181007 set *sessions = capList (union (if revokeOthers then removeWhere ses (\s -> s.pubkey == pubkey) else capPerKey ses (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt)) [{token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}]) maxSessions (\s -> s.expiresAt)
10191008 pres <- *presence
1020− let newPres = capList (union (removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)) [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1021− set *presence = newPres
1009+ set *presence = capList (union (removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)) [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1010+ newPres <- *presence
10221011 yield {valid: True {}, peers: knownPeerServersFromPresence newPres}
10231012 if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})
10241013 else do
@@ -1172,12 +1161,11 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
11721161 msgs <- *messages
11731162 aids <- *ackedIds
11741163 dids <- *deliveredIds
1175− let isDup = hasMessage effMsgId to msgs || hasMessage effMsgId to aids || hasMessage effMsgId to dids
11761164 srLim <- chargeRate sendRates senderKey maxSendRatePerKey sendRateWindowMs t
11771165 prLim <- chargeRate pairRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t
11781166 rrLim <- chargeRate recvRates to maxRecvRatePerKey recvRateWindowMs t
11791167 if senderKey == to then yield {status: MsRejected {}, remoteServers: []}
1180− else if isDup then yield {status: MsDuplicate {}, remoteServers: []}
1168+ else if hasMessage effMsgId to msgs || hasMessage effMsgId to aids || hasMessage effMsgId to dids then yield {status: MsDuplicate {}, remoteServers: []}
11811169 else if srLim || prLim || rrLim then yield {status: MsRateLimited {}, remoteServers: []}
11821170 else if count msgs >= maxMessages then yield {status: MsRejected {}, remoteServers: []}
11831171 else if countWhere msgs (\m -> m.toKey == to) >= maxMessagesPerRecipient then yield {status: MsQueued {}, remoteServers: []}
@@ -1190,7 +1178,6 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
11901178 set *messages = union msgs [{messageId: effMsgId, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: expiresAt, senderKey: senderKey, priority: msgPriority}]
11911179 set *pollWakeup = seqTs
11921180 let toPresence = presenceForKey pres to t
1193− let localOnline = any toPresence (\p -> isLocalServer p.server)
11941181 let remoteServers = collectPeers toPresence (\p -> p.server)
11951182 forwardsEnqueued <- if federationEnabled && count remoteServers > 0
11961183 then do
@@ -1199,10 +1186,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
11991186 appendCapped forwards newFwds maxForwardsEntries (\f -> f.createdAt)
12001187 yield (count newFwds > 0)
12011188 else yield False
1202− let status = if localOnline then MsDelivered {}
1203− else if forwardsEnqueued then MsFederated {}
1204− else MsQueued {}
1205− yield {status: status, remoteServers: remoteServers}
1189+ yield {status: if any toPresence (\p -> isLocalServer p.server) then MsDelivered {} else if forwardsEnqueued then MsFederated {} else MsQueued {}, remoteServers: remoteServers}
12061190
12071191forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
12081192 forEach servers id (\server -> do
@@ -1396,8 +1380,7 @@ handleFederationPubkey = \respond -> do
13961380
13971381handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlationId authorization respond -> do
13981382 t <- now
1399− let cappedHops = clamp 0 maxGossipHops hopsRemaining
1400− withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
1383+ withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (clamp 0 maxGossipHops hopsRemaining >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
14011384 valid <- atomic do
14021385 pres <- *presence
14031386 let validEvents = do
@@ -1406,14 +1389,10 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
14061389 let expiresAt = min e.timestamp t + clampTtl e.ttl * 1000
14071390 where expiresAt > t
14081391 yield {pubkey: e.pubkey, server: e.server, expiresAt: expiresAt, timestamp: e.timestamp, eventType: e.eventType}
1409− let isOnline = \v -> case v.eventType of GossipOnline {} -> True {}; _ -> False {}
1410− let isOffline = \v -> case v.eventType of GossipOffline {} -> True {}; _ -> False {}
1411− let online = mapWhere validEvents isOnline (\v -> {pubkey: v.pubkey, server: v.server, expiresAt: v.expiresAt, timestamp: v.timestamp})
1412− let offline = mapWhere validEvents isOffline (\v -> {pubkey: v.pubkey, server: v.server, timestamp: v.timestamp})
1413− let withoutOffline = filter pres (\p -> none offline (\o -> o.pubkey == p.pubkey && o.server == p.server && o.timestamp > p.timestamp))
1414− set *presence = capList (mergePresence withoutOffline online) maxPresenceEntries (\p -> p.expiresAt)
1415− let newlyOnline = mapWhere online (\e -> not (isLocalServer e.server) && none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)) (\e -> {pubkey: e.pubkey, server: e.server})
1416− yield {newlyOnline: newlyOnline}
1392+ let online = mapWhere validEvents (\v -> case v.eventType of GossipOnline {} -> True {}; _ -> False {}) (\v -> {pubkey: v.pubkey, server: v.server, expiresAt: v.expiresAt, timestamp: v.timestamp})
1393+ let offline = mapWhere validEvents (\v -> case v.eventType of GossipOffline {} -> True {}; _ -> False {}) (\v -> {pubkey: v.pubkey, server: v.server, timestamp: v.timestamp})
1394+ set *presence = capList (mergePresence (filter pres (\p -> none offline (\o -> o.pubkey == p.pubkey && o.server == p.server && o.timestamp > p.timestamp))) online) maxPresenceEntries (\p -> p.expiresAt)
1395+ yield {newlyOnline: mapWhere online (\e -> not (isLocalServer e.server) && none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)) (\e -> {pubkey: e.pubkey, server: e.server})}
14171396 when (count valid.newlyOnline > 0)
14181397 (fork (forEach valid.newlyOnline (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t)))
14191398 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
@@ -1444,10 +1423,7 @@ storeFederated = \toKey encryptedBlob timestamp originServer messageId expiresAt
14441423 seq <- *seqCounter
14451424 let seqTs = nextSeqTs t seq
14461425 set *seqCounter = seqTs
1447− let clampedTs = clampTimestamp timestamp t
1448− let fwdExpiresAt = if expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
1449− let clampedPri = clamp 0 maxMessagePriority priority
1450− set *messages = union msgs [{messageId: messageId, toKey: toKey, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: clampedPri}]
1426+ set *messages = union msgs [{messageId: messageId, toKey: toKey, encryptedBlob: encryptedBlob, timestamp: clampTimestamp timestamp t, originServer: originServer, receivedAt: seqTs, expiresAt: if expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge, senderKey: "", priority: clamp 0 maxMessagePriority priority}]
14511427 set *pollWakeup = seqTs
14521428 yield ""
14531429
@@ -1570,8 +1546,7 @@ handleDrain = \phase authorization respond -> withMaintAuth authorization
15701546 let localEntries = filter pres (\p -> isLocalServer p.server && p.expiresAt > t)
15711547 set *presence = filter pres (\p -> not (isLocalServer p.server) || p.expiresAt <= t)
15721548 set *pollWakeup = t
1573− let peers = knownPeerServersFromPresence pres
1574− yield {drainedCount: count localEntries, peers: peers, localKeys: localEntries}
1549+ yield {drainedCount: count localEntries, peers: knownPeerServersFromPresence pres, localKeys: localEntries}
15751550 when (effectivePhase == 2 && count drainResult.localKeys > 0 && count drainResult.peers > 0 && federationEnabled)
15761551 (fork (sendBatchedGossip drainResult.peers (pluck drainResult.localKeys (\p -> mkGossip (GossipOffline {}) p.pubkey t offlineGossipMinTtl)) gossipOriginHops))
15771552 yield (respond {ok: True {}, drained: drainResult.drainedCount, phase: effectivePhase}))
@@ -1769,9 +1744,8 @@ handleStatsPeers = \authorization respond -> withMaintAuth authorization
17691744 t <- now
17701745 failures <- *forwardFailures
17711746 fwds <- *forwards
1772− let peers = pluck failures (\f ->
1773− {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoff f.failures, retriable: shouldRetryServer f.server t failures, pendingForwards: countWhere fwds (\x -> x.toServer == f.server), avgLatencyMs: 0, maxLatencyMs: 0, latencySamples: 0, gossipEventsPerMin: 0})
1774− yield (respond {peers: peers, ok: True {}}))
1747+ yield (respond {peers: pluck failures (\f ->
1748+ {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoff f.failures, retriable: shouldRetryServer f.server t failures, pendingForwards: countWhere fwds (\x -> x.toServer == f.server), avgLatencyMs: 0, maxLatencyMs: 0, latencySamples: 0, gossipEventsPerMin: 0}), ok: True {}}))
17751749
17761750handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization
17771751 (yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, forwardDetails: [], activePolls: 0, ok: False {}}))