@@ -626,7 +626,7 @@ retryPendingForwards = \pending -> do
626626 retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)
627627 yield {}
628628
629−sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, | a}] -> Text -> IO {network} {}
629+sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
630630sendBatchedAcksToServer = \serverAcks server ->
631631 forEachBatch serverAcks maxMessageIds (\batch -> do
632632 result <- fetch (federationScheme ++ server) (FederationAck {
@@ -643,7 +643,7 @@ sendBatchedAcksToServer = \serverAcks server ->
643643 yield a
644644 Err _ -> yield {})
645645
646−sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, | a}] -> [Text] -> IO {network} {}
646+sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> [ServerName] -> IO {network} {}
647647sendAcksAndClean = \pendingAcks servers ->
648648 sendToEach servers (\server ->
649649 sendBatchedAcksToServer (do a <- pendingAcks; where a.originServer == server; yield a) server)
@@ -668,15 +668,15 @@ pollOnce = \token cursor -> do
668668 t <- now
669669 atomic do
670670 sessions <- *sessions
671− seqCounter <- *seqCounter
672671 let auth = authenticateSession sessions token t
673− let cursorCeil = if seqCounter > t then seqCounter else t
674− let clampedCursor = if cursor < 0 then 0 else if cursor > cursorCeil then cursorCeil else cursor
675672 case auth of
676673 Unauthorized {} ->
677674 yield {events: [], cursor: -1, authorized: False {}}
678675 Authorized {pubkey} -> do
676+ seqCounter <- *seqCounter
679677 messages <- *messages
678+ let cursorCeil = if seqCounter > t then seqCounter else t
679+ let clampedCursor = if cursor < 0 then 0 else if cursor > cursorCeil then cursorCeil else cursor
680680 let matchingMsgs = do
681681 m <- messages
682682 where m.toKey == pubkey
@@ -808,7 +808,7 @@ serve = \req -> case req of
808808 let auth = authenticateSession sessions (extractToken authorization) t
809809 case auth of
810810 Unauthorized {} ->
811− yield {status: "unauthorized", shouldInsert: False {}, remoteServers: [], fromKey: "", expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: False {}}
811+ yield {status: "unauthorized", shouldInsert: False {}, remoteServers: [], fromKey: "", expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
812812 Authorized {pubkey: fromKey} -> do
813813 messages <- *messages
814814 presence <- *presence
@@ -825,17 +825,15 @@ serve = \req -> case req of
825825 let localOnline = any toPresence (\p -> p.server == serverHostN)
826826 let remoteServers = do p <- toPresence; where not (p.server == serverHostN); yield p.server
827827 forwards <- *forwards
828− let forwardsAtCapacity = count forwards >= maxForwardsEntries
828+ let forwardServers = takeN remoteServers (maxForwardsEntries - count forwards)
829829 let status = if not shouldInsert then "rejected"
830830 else if localOnline then "delivered"
831− else if count remoteServers > 0 && federationEnabled && not forwardsAtCapacity then "federated"
831+ else if federationEnabled && count forwardServers > 0 then "federated"
832832 else "queued"
833− let forwardSlots = maxForwardsEntries - count forwards
834− let forwardServers = takeN remoteServers forwardSlots
835− set *forwards = if shouldInsert && not forwardsAtCapacity && count forwardServers > 0 && federationEnabled
833+ set *forwards = if shouldInsert && federationEnabled && count forwardServers > 0
836834 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
837835 else forwards
838− yield {status: status, shouldInsert: shouldInsert, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}
836+ yield {status: status, shouldInsert: shouldInsert, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
839837 fork (do
840838 when (count result.remoteServers > 0 && result.shouldInsert && federationEnabled)
841839 (sendToEach result.remoteServers (\server -> do
@@ -846,7 +844,7 @@ serve = \req -> case req of
846844 case fwdResult of
847845 Ok _ -> removeForward msgId server
848846 Err _ -> yield {}))
849− when (result.forwardsAtCapacity && count result.remoteServers > 0 && result.shouldInsert)
847+ when (result.status == "queued" && count result.remoteServers > 0 && result.shouldInsert && federationEnabled)
850848 (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking"))
851849 broadcastOfflineGossip t result.peerServers result.expiredLocal)
852850 yield (respond {status: result.status, messageId: if result.shouldInsert then msgId else ""})
@@ -953,9 +951,14 @@ serve = \req -> case req of
953951 e <- onlineEntries
954952 where none offlineEntries (\o -> o.server == fromServer && samePubkeyServer o e && o.timestamp > e.timestamp)
955953 yield e
956− let affectedKeys = do
957− e <- allValidEntries
958− yield {pubkey: e.pubkey, server: e.server}
954+ let filteredOffline = do
955+ e <- offlineEntries
956+ where none onlineEntries (\o -> samePubkeyServer o e && o.timestamp > e.timestamp)
957+ yield e
958+ let trustedOffline = do
959+ e <- offlineEntries
960+ where e.server == fromServer
961+ yield e
959962 let deduped = mergeByPubkeyServer [] filteredOnline
960963 gossipResult <- atomic do
961964 pr <- maybePrune t
@@ -964,19 +967,15 @@ serve = \req -> case req of
964967 e <- deduped
965968 where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t)
966969 yield e
967− let trustedOffline = do
968− e <- offlineEntries
969− where e.server == fromServer
970− yield e
971970 let appliedOnline = do
972971 e <- trustedDeduped
973972 where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
974973 yield {pubkey: e.pubkey, server: e.server}
975974 let merged = mergeByPubkeyServer presence trustedDeduped
976975 let offlineOnlyKeys = do
977− k <- affectedKeys
978− where none trustedDeduped (\d -> samePubkeyServer d k)
979− yield k
976+ e <- allValidEntries
977+ where none trustedDeduped (\d -> samePubkeyServer d e)
978+ yield e
980979 let appliedOffline = do
981980 k <- offlineOnlyKeys
982981 where any presence (\p -> samePubkeyServer p k && any trustedOffline (\o -> samePubkeyServer o k && (if o.timestamp > t then t else o.timestamp) > p.timestamp))
@@ -1000,10 +999,6 @@ serve = \req -> case req of
1000999 set *presence = finalPresence
10011000 let gossipPeerServers = collectPeerServers finalPresence (\p -> p.server)
10021001 yield {appliedKeys: union capResult.appliedOnline appliedOffline, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1003− let filteredOffline = do
1004− e <- offlineEntries
1005− where none onlineEntries (\o -> samePubkeyServer o e && o.timestamp > e.timestamp)
1006− yield e
10071002 let effectiveEvents = do
10081003 e <- union filteredOnline filteredOffline
10091004 where any gossipResult.appliedKeys (\a -> samePubkeyServer a e)