@@ -815,34 +815,35 @@ serve = \req -> case req of
815 yield SendUnauthorized {pr: pr}
815 yield SendUnauthorized {pr: pr}
816 Authorized {pubkey: fromKey} -> do
816 Authorized {pubkey: fromKey} -> do
817 messages <- *messages
817 messages <- *messages
818− presence <- *presence
819− seqCounter <- *seqCounter
820− let seqTs = nextSeqTs t seqCounter
821− let shouldInsert = countMessagesFor messages to t < maxMessagesPerRecipient
818+ if countMessagesFor messages to t >= maxMessagesPerRecipient
822− set *seqCounter = if shouldInsert then seqTs else seqCounter
819+ then yield SendRejected {pr: pr}
823− set *messages = if shouldInsert
820+ else do
821+ presence <- *presence
822+ seqCounter <- *seqCounter
823+ let seqTs = nextSeqTs t seqCounter
824+ set *seqCounter = seqTs
824− then union messages
825+ set *messages = union messages
825− [{messageId: msgId, toKey: to, fromKey: fromKey,
826+ [{messageId: msgId, toKey: to, fromKey: fromKey,
826− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
827+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
828− let toPresence = do p <- presence; where p.pubkey == to; where p.expiresAt > t; yield p
828+ let toPresence = do p <- presence; where p.pubkey == to; where p.expiresAt > t; yield p
829− let localOnline = any toPresence (\p -> p.server == serverHostN)
829+ let localOnline = any toPresence (\p -> p.server == serverHostN)
830− let remoteServers = do p <- toPresence; where not (p.server == serverHostN); yield p.server
830+ let remoteServers = do p <- toPresence; where not (p.server == serverHostN); yield p.server
831− forwards <- *forwards
831+ forwards <- *forwards
832− let forwardServers = takeN remoteServers (maxForwardsEntries - count forwards)
832+ let forwardServers = takeN remoteServers (maxForwardsEntries - count forwards)
833− let status = if not shouldInsert then "rejected"
834− else if localOnline then "delivered"
833+ let status = if localOnline then "delivered"
835− else if federationEnabled && count forwardServers > 0 then "federated"
834+ else if federationEnabled && count forwardServers > 0 then "federated"
836− else "queued"
835+ else "queued"
837− set *forwards = if shouldInsert && federationEnabled && count forwardServers > 0
836+ set *forwards = if federationEnabled && count forwardServers > 0
838− then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
837+ then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
839− else forwards
838+ else forwards
840− yield SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey, pr: pr}
839+ yield SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey, pr: pr}
841 fork (case result of
840 fork (case result of
842 SendUnauthorized {pr} ->
841 SendUnauthorized {pr} ->
843 broadcastOfflineGossip t pr.peerServers pr.expiredLocal
842 broadcastOfflineGossip t pr.peerServers pr.expiredLocal
843+ SendRejected {pr} ->
844+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal
844 SendDone {status, remoteServers, fromKey, pr} -> do
845 SendDone {status, remoteServers, fromKey, pr} -> do
845− when (count remoteServers > 0 && not (status == "rejected") && federationEnabled)
846+ when (count remoteServers > 0 && federationEnabled)
846 (sendToEach remoteServers (\server -> do
847 (sendToEach remoteServers (\server -> do
847 fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
848 fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
848 toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
849 toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
@@ -856,7 +857,8 @@ serve = \req -> case req of
856 broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
857 broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
857 yield (case result of
858 yield (case result of
858 SendUnauthorized _ -> respond {status: "unauthorized", messageId: ""}
859 SendUnauthorized _ -> respond {status: "unauthorized", messageId: ""}
860+ SendRejected _ -> respond {status: "rejected", messageId: ""}
859− SendDone {status, remoteServers, fromKey, pr} -> respond {status: status, messageId: if status == "rejected" then "" else msgId})
861+ SendDone {status} -> respond {status: status, messageId: msgId})
861 AckMessages {messageIds, authorization, respond} -> do
863 AckMessages {messageIds, authorization, respond} -> do
@@ -956,82 +958,86 @@ serve = \req -> case req of
956 let authorized = verifyFederationAuth authorization
958 let authorized = verifyFederationAuth authorization
957 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining
959 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining
958 let fromServer = normalizeServerName fromServer
960 let fromServer = normalizeServerName fromServer
959− let validRequest = authorized && cappedHops >= 1 && not (fromServer == serverHostN) && count events <= maxGossipEvents
961+ if not (authorized && cappedHops >= 1 && not (fromServer == serverHostN) && count events <= maxGossipEvents)
963+ pruneResult <- atomic do maybePrune t
964+ forkBroadcastOffline t pruneResult
965+ yield (respond {ok: False {}})
960− let allValidEntries = if validRequest then validGossipEvents events t else []
967+ let allValidEntries = validGossipEvents events t
961− let onlineEntries = do e <- allValidEntries; where e.eventType == "online"; yield e
968+ let onlineEntries = do e <- allValidEntries; where e.eventType == "online"; yield e
962− let offlineEntries = do e <- allValidEntries; where e.eventType == "offline"; yield e
969+ let offlineEntries = do e <- allValidEntries; where e.eventType == "offline"; yield e
963− let filteredOnline = do
970+ let filteredOnline = do
964− e <- onlineEntries
971+ e <- onlineEntries
965− where none offlineEntries (\o -> o.server == fromServer && samePubkeyServer o e && o.timestamp > e.timestamp)
972+ where none offlineEntries (\o -> o.server == fromServer && samePubkeyServer o e && o.timestamp > e.timestamp)
967− let filteredOffline = do
974+ let filteredOffline = do
968− e <- offlineEntries
975+ e <- offlineEntries
969− where none onlineEntries (\o -> samePubkeyServer o e && o.timestamp > e.timestamp)
976+ where none onlineEntries (\o -> samePubkeyServer o e && o.timestamp > e.timestamp)
971− let trustedOffline = do
978+ let trustedOffline = do
972− e <- offlineEntries
979+ e <- offlineEntries
973− where e.server == fromServer
980+ where e.server == fromServer
975− let deduped = mergeByPubkeyServer [] filteredOnline
982+ let deduped = mergeByPubkeyServer [] filteredOnline
976− gossipResult <- atomic do
983+ gossipResult <- atomic do
977− pr <- maybePrune t
984+ pr <- maybePrune t
978− presence <- *presence
985+ presence <- *presence
979− let trustedDeduped = do
986+ let trustedDeduped = do
980− e <- deduped
987+ e <- deduped
981− where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t)
988+ where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t)
983− let appliedOnline = do
990+ let appliedOnline = do
984− e <- trustedDeduped
991+ e <- trustedDeduped
985− where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
992+ where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
986− yield {pubkey: e.pubkey, server: e.server}
993+ yield {pubkey: e.pubkey, server: e.server}
987− let merged = mergeByPubkeyServer presence trustedDeduped
994+ let merged = mergeByPubkeyServer presence trustedDeduped
988− let offlineOnlyKeys = do
995+ let offlineOnlyKeys = do
989− e <- allValidEntries
996+ e <- allValidEntries
990− where none trustedDeduped (\d -> samePubkeyServer d e)
997+ where none trustedDeduped (\d -> samePubkeyServer d e)
992− let appliedOffline = do
999+ let appliedOffline = do
993− k <- offlineOnlyKeys
1000+ k <- offlineOnlyKeys
994− where any presence (\p -> samePubkeyServer p k && any trustedOffline (\o -> samePubkeyServer o k && (if o.timestamp > t then t else o.timestamp) > p.timestamp))
1001+ where any presence (\p -> samePubkeyServer p k && any trustedOffline (\o -> samePubkeyServer o k && (if o.timestamp > t then t else o.timestamp) > p.timestamp))
995− yield {pubkey: k.pubkey, server: k.server}
1002+ yield {pubkey: k.pubkey, server: k.server}
996− let afterOffline = removeOffline merged offlineOnlyKeys trustedOffline t
1003+ let afterOffline = removeOffline merged offlineOnlyKeys trustedOffline t
997− let withinCapacity = count afterOffline <= maxPresenceEntries
1004+ let withinCapacity = count afterOffline <= maxPresenceEntries
998− let capResult = if withinCapacity
1005+ let capResult = if withinCapacity
999− then {finalPresence: afterOffline, appliedOnline: appliedOnline}
1006+ then {finalPresence: afterOffline, appliedOnline: appliedOnline}
1001− let updatesOnly = do e <- trustedDeduped; where any presence (\p -> samePubkeyServer p e); yield e in
1008+ let updatesOnly = do e <- trustedDeduped; where any presence (\p -> samePubkeyServer p e); yield e in
1002− let updatesOnlyMerged = mergeByPubkeyServer presence updatesOnly in
1009+ let updatesOnlyMerged = mergeByPubkeyServer presence updatesOnly in
1003− let updatesOnlyAfterOffline = removeOffline updatesOnlyMerged offlineOnlyKeys trustedOffline t in
1010+ let updatesOnlyAfterOffline = removeOffline updatesOnlyMerged offlineOnlyKeys trustedOffline t in
1004− let reclaimSlots = maxPresenceEntries - count updatesOnlyAfterOffline in
1011+ let reclaimSlots = maxPresenceEntries - count updatesOnlyAfterOffline in
1005− let newEntries = do e <- trustedDeduped; where none presence (\p -> samePubkeyServer p e); where none updatesOnlyAfterOffline (\p -> samePubkeyServer p e); yield e in
1012+ let newEntries = do e <- trustedDeduped; where none presence (\p -> samePubkeyServer p e); where none updatesOnlyAfterOffline (\p -> samePubkeyServer p e); yield e in
1006− let reclaimedEntries = takeN newEntries reclaimSlots in
1013+ let reclaimedEntries = takeN newEntries reclaimSlots in
1007− {finalPresence: union updatesOnlyAfterOffline reclaimedEntries,
1014+ {finalPresence: union updatesOnlyAfterOffline reclaimedEntries,
1008− appliedOnline: union
1015+ appliedOnline: union
1009− (do e <- updatesOnly; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield {pubkey: e.pubkey, server: e.server})
1016+ (do e <- updatesOnly; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield {pubkey: e.pubkey, server: e.server})
1010− (do e <- reclaimedEntries; yield {pubkey: e.pubkey, server: e.server})}
1017+ (do e <- reclaimedEntries; yield {pubkey: e.pubkey, server: e.server})}
1011− let finalPresence = if validRequest then capResult.finalPresence else presence
1012− set *presence = finalPresence
1018+ set *presence = capResult.finalPresence
1013− let gossipPeerServers = collectPeerServers finalPresence (\p -> p.server)
1019+ let gossipPeerServers = collectPeerServers capResult.finalPresence (\p -> p.server)
1014− yield {appliedKeys: union capResult.appliedOnline appliedOffline, pr: pr, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1020+ yield {appliedKeys: union capResult.appliedOnline appliedOffline, pr: pr, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1015− let effectiveEvents = do
1021+ let effectiveEvents = do
1016− e <- union filteredOnline filteredOffline
1022+ e <- union filteredOnline filteredOffline
1017− where any gossipResult.appliedKeys (\a -> samePubkeyServer a e)
1023+ where any gossipResult.appliedKeys (\a -> samePubkeyServer a e)
1018− yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
1024+ yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
1019− when (not gossipResult.withinCapacity)
1025+ when (not gossipResult.withinCapacity)
1020− (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
1026+ (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
1021− fork (do
1027+ fork (do
1022− when (validRequest && cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (do
1028+ when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (do
1023− let targets = do
1029+ let targets = do
1024− s <- gossipResult.gossipPeerServers
1030+ s <- gossipResult.gossipPeerServers
1025− where not (s == fromServer)
1031+ where not (s == fromServer)
1026− yield s
1032+ yield s
1027− sendToEach targets (\server -> do
1033+ sendToEach targets (\server -> do
1028− _ <- fetch (federationScheme ++ server) (RecvGossip {
1034+ _ <- fetch (federationScheme ++ server) (RecvGossip {
1029− authorization: "Bearer " ++ federationSecret,
1035+ authorization: "Bearer " ++ federationSecret,
1030− events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
1036+ events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
1031− yield {}))
1037+ yield {}))
1032− let pr = gossipResult.pr
1038+ let pr = gossipResult.pr
1033− broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
1039+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
1034− yield (respond {ok: validRequest})
1040+ yield (respond {ok: True {}})
1036 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
1042 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
1037 let toKey = normalizeHex toKey
1043 let toKey = normalizeHex toKey
@@ -1045,23 +1051,24 @@ serve = \req -> case req of
1045 let clampedTs = clampTimestamp timestamp t
1051 let clampedTs = clampTimestamp timestamp t
1046 result <- atomic do
1052 result <- atomic do
1047 pr <- maybePrune t
1053 pr <- maybePrune t
1048− messages <- *messages
1049− seqCounter <- *seqCounter
1050− let seqTs = nextSeqTs t seqCounter
1051− ackedIds <- *ackedIds
1052− let valid = authorized && validOrigin
1054+ if not (authorized && validOrigin)
1053− let alreadyExists = valid
1055+ then yield {pr: pr, ok: False {}}
1057+ messages <- *messages
1058+ ackedIds <- *ackedIds
1054− && (any messages (\m -> m.messageId == messageId && m.toKey == toKey)
1059+ if any messages (\m -> m.messageId == messageId && m.toKey == toKey)
1055− || any ackedIds (\a -> a.messageId == messageId && a.toKey == toKey))
1060+ || any ackedIds (\a -> a.messageId == messageId && a.toKey == toKey)
1056− let shouldInsert = valid && not alreadyExists
1061+ then yield {pr: pr, ok: True {}}
1057− && countMessagesFor messages toKey t < maxMessagesPerRecipient
1062+ else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1058− set *seqCounter = if shouldInsert then seqTs else seqCounter
1063+ then yield {pr: pr, ok: False {}}
1059− set *messages = if shouldInsert
1064+ else do
1065+ seqCounter <- *seqCounter
1066+ let seqTs = nextSeqTs t seqCounter
1067+ set *seqCounter = seqTs
1060− then union messages
1068+ set *messages = union messages
1061− [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1069+ [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1062− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1070+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1064− yield {pr: pr, ok: shouldInsert || alreadyExists}
1071+ yield {pr: pr, ok: True {}}
1065 forkBroadcastOffline t result.pr
1072 forkBroadcastOffline t result.pr
1066 yield (respond {ok: result.ok})
1073 yield (respond {ok: result.ok})