@@ -464,13 +464,26 @@ maxBy = \items getKey -> fold (\acc x -> case acc of
464evictOldest = \items getKey ->
464evictOldest = \items getKey ->
465 case minBy items getKey of
465 case minBy items getKey of
466 Nothing {} -> items
466 Nothing {} -> items
467− Just {value: oldest} ->
468− let k = getKey oldest in
469− do x <- items; where getKey x != k; yield x
467+ Just {value: oldest} -> do x <- items; where getKey x != getKey oldest; yield x
471capList = \items maxN getKey ->
469capList = \items maxN getKey ->
472 if count items > maxN then evictOldest items getKey else items
470 if count items > maxN then evictOldest items getKey else items
472+-- Sequence an IO action over each unique item (dedup by getKey).
473+forEach = \items getKey action -> case headOf items of
474+ Nothing {} -> yield {}
475+ Just {value: x} -> do
477+ forEach (do y <- items; where getKey y != getKey x; yield y) getKey action
479+-- Like forEach, but collects each action's result into a list.
480+mapEach = \items getKey action -> case headOf items of
481+ Nothing {} -> yield []
482+ Just {value: x} -> do
484+ rest <- mapEach (do y <- items; where getKey y != getKey x; yield y) getKey action
485+ yield (union [r] rest)
474-- Hex tokens
487-- Hex tokens
475hexDigit : Int -> Text
488hexDigit : Int -> Text
476hexDigit = \n -> take 1 (drop (clamp 0 15 n) "0123456789abcdef")
489hexDigit = \n -> take 1 (drop (clamp 0 15 n) "0123456789abcdef")
@@ -498,11 +511,10 @@ tooLong = \s maxLen -> drop maxLen s != ""
499isValidHex : Text -> Bool
512isValidHex : Text -> Bool
500isValidHex = \s ->
513isValidHex = \s ->
501− if s == "" then False {}
502− else if drop 1 s == "" then False {}
514+ s != "" && drop 1 s != ""
515+ && contains (take 1 s) "0123456789abcdef"
503− else if contains (take 1 s) "0123456789abcdef" && contains (take 1 (drop 1 s)) "0123456789abcdef"
516+ && contains (take 1 (drop 1 s)) "0123456789abcdef"
504− then drop 2 s == "" || isValidHex (drop 2 s)
517+ && (drop 2 s == "" || isValidHex (drop 2 s))
507normalizeHex : Text -> Text
519normalizeHex : Text -> Text
508normalizeHex = \s -> bytesToHex (bytesFromHex s)
520normalizeHex = \s -> bytesToHex (bytesFromHex s)
@@ -852,24 +864,18 @@ pruneExpired = \t -> do
852-- Federation gossip
864-- Federation gossip
853-- ============================================
865-- ============================================
855−sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}
867+sendBatchedGossip = \peers events hops -> forEach peers (\s -> s) (\server -> do
856−sendBatchedGossip = \peers events hops ->
857− case headOf peers of
858− Nothing {} -> yield {}
859− Just {value: server} -> do
860− let rest = do s <- peers; where s != server; yield s
861− t <- now
868+ t <- now
862− failures <- *forwardFailures
869+ failures <- *forwardFailures
863− when (shouldRetryServer server t failures) (do
870+ when (shouldRetryServer server t failures) (do
864− auth <- generateFederationAuth "gossip" t
871+ auth <- generateFederationAuth "gossip" t
865− result <- fetch (federationUrl server) (RecvGossip {
872+ result <- fetch (federationUrl server) (RecvGossip {
866− authorization: auth,
873+ authorization: auth,
867− events: events, hopsRemaining: hops, fromServer: serverHostN,
874+ events: events, hopsRemaining: hops, fromServer: serverHostN,
868− requestTimestamp: t, correlationId: "", protocolVersion: federationProtocolVersion})
875+ requestTimestamp: t, correlationId: "", protocolVersion: federationProtocolVersion})
869− case result of
876+ case result of
870− Err _ -> recordFailure server
877+ Err _ -> recordFailure server
871− Ok _ -> clearFailure server)
878+ Ok _ -> clearFailure server))
872− sendBatchedGossip rest events hops
874broadcastOfflineGossip = \t expiredLocal -> do
880broadcastOfflineGossip = \t expiredLocal -> do
875 presence <- *presence
881 presence <- *presence
@@ -882,37 +888,21 @@ broadcastOfflineGossip = \t expiredLocal -> do
882-- Forward retries and acks
888-- Forward retries and acks
883-- ============================================
889-- ============================================
885−retryPendingForwards : IO {network} {}
886retryPendingForwards = do
891retryPendingForwards = do
888 fwds <- *forwards
893 fwds <- *forwards
889 failures <- *forwardFailures
894 failures <- *forwardFailures
890 blocked <- *blockedPeers
895 blocked <- *blockedPeers
892− let servers = collectPeers fwds (\f -> f.toServer)
893− let eligibleServers = do s <- servers; where shouldRetryServer s t failures && not (isBlockedPeer s blocked); yield s
896+ let servers = do s <- collectPeers fwds (\f -> f.toServer); where shouldRetryServer s t failures && not (isBlockedPeer s blocked); yield s
894− retryServerForwards eligibleServers fwds
897+ forEach servers (\s -> s) (\server -> do
896−retryServerForwards = \servers fwds ->
897− case headOf servers of
898− Nothing {} -> yield {}
899− Just {value: server} -> do
900− let rest = do s <- servers; where s != server; yield s
901− let serverFwds = takeRelation maxBatchSize (do f <- fwds; where f.toServer == server; yield f)
898+ let serverFwds = takeRelation maxBatchSize (do f <- fwds; where f.toServer == server; yield f)
902− sendForwardsToServer serverFwds server
899+ forEach serverFwds (\f -> f.messageId) (\f -> do
903− retryServerForwards rest fwds
905−sendForwardsToServer = \fwds server ->
907− Nothing {} -> yield {}
908− Just {value: f} -> do
909− let rest = do x <- fwds; where x.messageId != f.messageId; yield x
910− t <- now
900+ t2 <- now
911− auth <- generateFederationAuth ("forward:" ++ f.messageId) t
901+ auth <- generateFederationAuth ("forward:" ++ f.messageId) t2
912 result <- fetch (federationUrl server) (ForwardMessage {
902 result <- fetch (federationUrl server) (ForwardMessage {
913 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
903 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
914 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
904 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
915− fromServer: serverHostN, requestTimestamp: t,
905+ fromServer: serverHostN, requestTimestamp: t2,
916 authorization: auth, correlationId: "",
906 authorization: auth, correlationId: "",
917 expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})
907 expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})
918 let success = case result of Ok r -> r.ok; _ -> False {}
908 let success = case result of Ok r -> r.ok; _ -> False {}
@@ -923,37 +913,24 @@ sendForwardsToServer = \fwds server ->
924 where x.messageId != f.messageId || x.toServer != server || (not success && x.retries + 1 < maxForwardRetries)
914 where x.messageId != f.messageId || x.toServer != server || (not success && x.retries + 1 < maxForwardRetries)
925 yield (if success || x.messageId != f.messageId || x.toServer != server then x
915 yield (if success || x.messageId != f.messageId || x.toServer != server then x
926− else {messageId: x.messageId, toServer: x.toServer, createdAt: x.createdAt, toKey: x.toKey, encryptedBlob: x.encryptedBlob, timestamp: x.timestamp, originServer: x.originServer, retries: x.retries + 1, expiresAt: x.expiresAt, priority: x.priority, senderKey: x.senderKey})
916+ else {messageId: x.messageId, toServer: x.toServer, createdAt: x.createdAt, toKey: x.toKey, encryptedBlob: x.encryptedBlob, timestamp: x.timestamp, originServer: x.originServer, retries: x.retries + 1, expiresAt: x.expiresAt, priority: x.priority, senderKey: x.senderKey})))
927− sendForwardsToServer rest server
929−sendAcksAndClean : [AckedEntry] -> IO {network} {}
930sendAcksAndClean = \pendingAcks ->
918sendAcksAndClean = \pendingAcks ->
931− if count pendingAcks == 0 then yield {}
933− let servers = collectPeers pendingAcks (\a -> a.originServer)
919+ forEach (collectPeers pendingAcks (\a -> a.originServer)) (\s -> s) (\server -> do
934− sendAcksToEach servers pendingAcks
936−sendAcksToEach = \servers acks ->
937− case headOf servers of
938− Nothing {} -> yield {}
939− Just {value: server} -> do
940− let rest = do s <- servers; where s != server; yield s
941− let serverAcks = do a <- acks; where a.originServer == server; yield a
920+ let mids = do a <- pendingAcks; where a.originServer == server; yield a.messageId
942− let mids = do a <- serverAcks; yield a.messageId
943− t <- now
921+ t <- now
944− auth <- generateFederationAuth "ack" t
922+ auth <- generateFederationAuth "ack" t
945− result <- fetch (federationUrl server) (FederationAck {
923+ result <- fetch (federationUrl server) (FederationAck {
946− authorization: auth,
924+ authorization: auth,
947− messageIds: mids, originServer: server, fromServer: serverHostN,
925+ messageIds: mids, originServer: server, fromServer: serverHostN,
948− requestTimestamp: t, correlationId: "", protocolVersion: federationProtocolVersion})
926+ requestTimestamp: t, correlationId: "", protocolVersion: federationProtocolVersion})
949− case result of
927+ case result of
950− Err _ -> recordFailure server
928+ Err _ -> recordFailure server
951− Ok _ -> do
929+ Ok _ -> do
952− clearFailure server
930+ clearFailure server
953− atomic do
931+ atomic do
954− cur <- *ackedIds
932+ cur <- *ackedIds
955− set *ackedIds = do a <- cur; where not (a.originServer == server && any mids (\m -> m == a.messageId)); yield a
933+ set *ackedIds = do a <- cur; where not (a.originServer == server && any mids (\m -> m == a.messageId)); yield a)
956− sendAcksToEach rest acks
958-- Forward queued messages to a server when a remote key comes online.
935-- Forward queued messages to a server when a remote key comes online.
959forwardQueuedToServer = \pubkey server t -> atomic do
936forwardQueuedToServer = \pubkey server t -> atomic do
@@ -1322,36 +1299,32 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
1322 yield {status: status, remoteServers: remoteServers}
1299 yield {status: status, remoteServers: remoteServers}
1324forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
1301forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
1325− case headOf servers of
1302+ forEach servers (\s -> s) (\server -> do
1326− Nothing {} -> yield {}
1327− Just {value: server} -> do
1328− let rest = do s <- servers; where s != server; yield s
1329− info <- atomic do
1303+ info <- atomic do
1330− fwds <- *forwards
1304+ fwds <- *forwards
1331− case headOf (do f <- fwds; where f.messageId == effMsgId && f.toServer == server; yield f) of
1305+ case headOf (do f <- fwds; where f.messageId == effMsgId && f.toServer == server; yield f) of
1332− Nothing {} -> yield Nothing {}
1306+ Nothing {} -> yield Nothing {}
1333− Just {value: f} -> yield (Just {value: {expiresAt: f.expiresAt, priority: f.priority}})
1307+ Just {value: f} -> yield (Just {value: {expiresAt: f.expiresAt, priority: f.priority}})
1334− case info of
1308+ case info of
1335− Nothing {} -> yield {}
1309+ Nothing {} -> yield {}
1336− Just {value: i} -> do
1310+ Just {value: i} -> do
1337− t <- now
1311+ t <- now
1338− auth <- generateFederationAuth ("forward:" ++ effMsgId) t
1312+ auth <- generateFederationAuth ("forward:" ++ effMsgId) t
1339− result <- fetch (federationUrl server) (ForwardMessage {
1313+ result <- fetch (federationUrl server) (ForwardMessage {
1340− toKey: to, encryptedBlob: encryptedBlob,
1314+ toKey: to, encryptedBlob: encryptedBlob,
1341− timestamp: clampedTs, originServer: serverHostN, messageId: effMsgId,
1315+ timestamp: clampedTs, originServer: serverHostN, messageId: effMsgId,
1342− fromServer: serverHostN, requestTimestamp: t,
1316+ fromServer: serverHostN, requestTimestamp: t,
1343− authorization: auth, correlationId: "",
1317+ authorization: auth, correlationId: "",
1344− expiresAt: i.expiresAt, priority: i.priority, protocolVersion: federationProtocolVersion})
1318+ expiresAt: i.expiresAt, priority: i.priority, protocolVersion: federationProtocolVersion})
1345− case result of
1319+ case result of
1346− Err _ -> recordFailure server
1320+ Err _ -> recordFailure server
1347− Ok r ->
1321+ Ok r ->
1348− if r.ok then do
1322+ if r.ok then do
1349− clearFailure server
1323+ clearFailure server
1350− atomic do
1324+ atomic do
1351− fwds <- *forwards
1325+ fwds <- *forwards
1352− set *forwards = do f <- fwds; where not (f.messageId == effMsgId && f.toServer == server); yield f
1326+ set *forwards = do f <- fwds; where not (f.messageId == effMsgId && f.toServer == server); yield f
1353− else yield {}
1327+ else yield {})
1354− forkForwardsTo effMsgId to encryptedBlob clampedTs rest
1356-- ============================================
1329-- ============================================
1357-- Handler: lookup
1330-- Handler: lookup
@@ -1535,24 +1508,17 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
1535 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
1508 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
1537sendCancelToEach = \messageId to servers ->
1510sendCancelToEach = \messageId to servers ->
1538− case headOf servers of
1511+ forEach servers (\s -> s) (\server -> do
1539− Nothing {} -> yield {}
1540− Just {value: server} -> do
1541− let rest = do s <- servers; where s != server; yield s
1542− t <- now
1512+ t <- now
1543− auth <- generateFederationAuth ("cancel:" ++ messageId) t
1513+ auth <- generateFederationAuth ("cancel:" ++ messageId) t
1544− result <- fetch (federationUrl server) (FederationCancel {
1514+ result <- fetch (federationUrl server) (FederationCancel {
1545− messageId: messageId, toKey: to, fromServer: serverHostN,
1515+ messageId: messageId, toKey: to, fromServer: serverHostN,
1546− requestTimestamp: t, correlationId: "", authorization: auth,
1516+ requestTimestamp: t, correlationId: "", authorization: auth,
1547− protocolVersion: federationProtocolVersion})
1517+ protocolVersion: federationProtocolVersion})
1548− case result of
1518+ let failed = case result of Err _ -> True {}; Ok r -> not r.ok
1550− pfc <- *pendingFederationCancels
1551− set *pendingFederationCancels = capList (union pfc [{messageId: messageId, toKey: to, toServer: server, createdAt: t, retries: 0}]) 10000 (\c -> c.createdAt)
1552− Ok r -> when (not r.ok) (atomic do
1519+ when failed (atomic do
1553− pfc <- *pendingFederationCancels
1520+ pfc <- *pendingFederationCancels
1554− set *pendingFederationCancels = capList (union pfc [{messageId: messageId, toKey: to, toServer: server, createdAt: t, retries: 0}]) 10000 (\c -> c.createdAt))
1521+ set *pendingFederationCancels = capList (union pfc [{messageId: messageId, toKey: to, toServer: server, createdAt: t, retries: 0}]) 10000 (\c -> c.createdAt)))
1555− sendCancelToEach messageId to rest
1557handleCancelBatch = \cancels authorization clientIp respond -> do
1523handleCancelBatch = \cancels authorization clientIp respond -> do
1558 t <- now
1524 t <- now
@@ -1567,32 +1533,25 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
1567 lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t
1533 lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t
1568 if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
1534 if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
1570− let capped = takeRelation maxBatchSize cancels
1536+ results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do
1571− results <- cancelEach senderKey capped t []
1537+ r <- atomic do
1539+ let target = headOf (do m <- msgs; where m.messageId == e.messageId && m.toKey == e.to; yield m)
1541+ Nothing {} -> yield False
1542+ Just {value: msg} ->
1543+ if msg.originServer != "" || msg.senderKey == "" || msg.senderKey != senderKey
1546+ set *messages = do m <- msgs; where not (m.messageId == e.messageId && m.toKey == e.to); yield m
1548+ set *forwards = do f <- fwds; where not (f.messageId == e.messageId && f.toKey == e.to); yield f
1549+ cn <- *cancelNotifications
1550+ set *cancelNotifications = capList (union cn [{toKey: e.to, messageId: e.messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)
1552+ yield {messageId: e.messageId, cancelled: r})
1572 yield (respond {results: results, authorized: True {}, error: ""}))
1553 yield (respond {results: results, authorized: True {}, error: ""}))
1574−cancelEach = \senderKey entries t acc ->
1575− case headOf entries of
1576− Nothing {} -> yield acc
1577− Just {value: e} -> do
1578− let rest = do x <- entries; where x.messageId != e.messageId; yield x
1581− let target = headOf (do m <- msgs; where m.messageId == e.messageId && m.toKey == e.to; yield m)
1583− Nothing {} -> yield False
1584− Just {value: msg} ->
1585− if msg.originServer != "" || msg.senderKey == "" || msg.senderKey != senderKey
1588− set *messages = do m <- msgs; where not (m.messageId == e.messageId && m.toKey == e.to); yield m
1590− set *forwards = do f <- fwds; where not (f.messageId == e.messageId && f.toKey == e.to); yield f
1591− cn <- *cancelNotifications
1592− set *cancelNotifications = capList (union cn [{toKey: e.to, messageId: e.messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)
1594− cancelEach senderKey rest t (union acc [{messageId: e.messageId, cancelled: r}])
1596handleSendBatch = \messages authorization clientIp respond -> do
1555handleSendBatch = \messages authorization clientIp respond -> do
1597 t <- now
1556 t <- now
1598 dp <- *drainPhase
1557 dp <- *drainPhase
@@ -1603,24 +1562,17 @@ handleSendBatch = \messages authorization clientIp respond -> do
1603 else withSessionAuth authorization t clientIp
1562 else withSessionAuth authorization t clientIp
1604 (yield (respond {results: [], authorized: False {}, retryAfterMs: 0}))
1563 (yield (respond {results: [], authorized: False {}, retryAfterMs: 0}))
1605 (\senderKey -> do
1564 (\senderKey -> do
1606− let capped = takeRelation maxBatchSize messages
1565+ results <- mapEach (takeRelation maxBatchSize messages) (\m -> m) (\m -> do
1607− results <- sendBatchEach senderKey capped t []
1608− yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}, retryAfterMs: 0}))
1610−sendBatchEach = \senderKey msgs t acc ->
1611− case headOf msgs of
1612− Nothing {} -> yield acc
1613− Just {value: m} -> do
1614− let rest = do x <- msgs; where x != m; yield x
1615− effMsgId <- if m.messageId != "" then yield m.messageId else genToken
1566+ effMsgId <- if m.messageId != "" then yield m.messageId else genToken
1616− let clampedTs = clampTimestamp m.timestamp t
1567+ let clampedTs = clampTimestamp m.timestamp t
1617− let clampedPri = clamp 0 maxMessagePriority m.priority
1568+ let clampedPri = clamp 0 maxMessagePriority m.priority
1618− r <- if t - m.timestamp > messageMaxAge
1569+ r <- if t - m.timestamp > messageMaxAge
1619− then yield {status: MsRejected {}, remoteServers: []}
1570+ then yield {status: MsRejected {}, remoteServers: []}
1620− else atomic do sendMessageCore senderKey m.to m.encryptedBlob clampedTs effMsgId m.expiresInMs clampedPri t
1571+ else atomic do sendMessageCore senderKey m.to m.encryptedBlob clampedTs effMsgId m.expiresInMs clampedPri t
1621− when (federationEnabled && count r.remoteServers > 0)
1572+ when (federationEnabled && count r.remoteServers > 0)
1622− (fork (forkForwardsTo effMsgId m.to m.encryptedBlob clampedTs r.remoteServers))
1573+ (fork (forkForwardsTo effMsgId m.to m.encryptedBlob clampedTs r.remoteServers))
1623− sendBatchEach senderKey rest t (union acc [{messageId: effMsgId, status: r.status}])
1574+ yield {messageId: effMsgId, status: r.status})
1575+ yield (respond {results: results, authorized: True {}, retryAfterMs: 0}))
1625-- ============================================
1577-- ============================================
1626-- Handler: federation receive
1578-- Handler: federation receive
@@ -1678,12 +1630,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
1678 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
1630 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
1680forwardForNewlyOnline = \targets t ->
1632forwardForNewlyOnline = \targets t ->
1681− case headOf targets of
1682− Nothing {} -> yield {}
1683− Just {value: tgt} -> do
1684− let rest = do x <- targets; where x.pubkey != tgt.pubkey || x.server != tgt.server; yield x
1685− forwardQueuedToServer tgt.pubkey tgt.server t
1633+ forEach targets (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t)
1686− forwardForNewlyOnline rest t
1688handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond -> do
1635handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond -> do
1689 t <- now
1636 t <- now
@@ -1774,15 +1721,7 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
1774 t <- now
1721 t <- now
1775 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
1722 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
1776 withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (do
1723 withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (do
1777− let capped = takeRelation maxBatchSize cancels
1724+ results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do
1778− results <- federationCancelEach capped fromServer t []
1779− yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
1781−federationCancelEach = \entries fromServer t acc ->
1782− case headOf entries of
1783− Nothing {} -> yield acc
1784− Just {value: e} -> do
1785− let rest = do x <- entries; where x.messageId != e.messageId; yield x
1786 cancelled <- atomic do
1725 cancelled <- atomic do
1787 msgs <- *messages
1726 msgs <- *messages
1788 let target = headOf (do m <- msgs; where m.messageId == e.messageId && m.toKey == e.toKey && m.originServer == fromServer; yield m)
1727 let target = headOf (do m <- msgs; where m.messageId == e.messageId && m.toKey == e.toKey && m.originServer == fromServer; yield m)
@@ -1793,21 +1732,14 @@ federationCancelEach = \entries fromServer t acc ->
1793 fwds <- *forwards
1732 fwds <- *forwards
1794 set *forwards = do f <- fwds; where not (f.messageId == e.messageId && f.toKey == e.toKey); yield f
1733 set *forwards = do f <- fwds; where not (f.messageId == e.messageId && f.toKey == e.toKey); yield f
1795 yield True
1734 yield True
1796− federationCancelEach rest fromServer t (union acc [{messageId: e.messageId, cancelled: cancelled}])
1735+ yield {messageId: e.messageId, cancelled: cancelled})
1736+ yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
1798handleFederationBatchForward = \forwards fromServer requestTimestamp correlationId authorization respond -> do
1738handleFederationBatchForward = \forwards fromServer requestTimestamp correlationId authorization respond -> do
1799 t <- now
1739 t <- now
1800 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
1740 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
1801 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (do
1741 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (do
1802− let capped = takeRelation maxBatchSize forwards
1742+ results <- mapEach (takeRelation maxBatchSize forwards) (\e -> e.messageId) (\e -> do
1803− results <- federationForwardEach capped fromServer t []
1804− yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
1806−federationForwardEach = \entries fromServer t acc ->
1807− case headOf entries of
1808− Nothing {} -> yield acc
1809− Just {value: e} -> do
1810− let rest = do x <- entries; where x.messageId != e.messageId; yield x
1811 ok <- if isLocalServer e.originServer || t - e.timestamp > messageMaxAge then yield False
1743 ok <- if isLocalServer e.originServer || t - e.timestamp > messageMaxAge then yield False
1812 else atomic do
1744 else atomic do
1813 pres <- *presence
1745 pres <- *presence
@@ -1835,7 +1767,8 @@ federationForwardEach = \entries fromServer t acc ->
1835 set *messages = union msgs [{messageId: e.messageId, toKey: e.toKey, encryptedBlob: e.encryptedBlob, timestamp: clampedTs, originServer: e.originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: clampedPri}]
1767 set *messages = union msgs [{messageId: e.messageId, toKey: e.toKey, encryptedBlob: e.encryptedBlob, timestamp: clampedTs, originServer: e.originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: clampedPri}]
1836 set *pollWakeup = seqTs
1768 set *pollWakeup = seqTs
1837 yield True
1769 yield True
1838− federationForwardEach rest fromServer t (union acc [{messageId: e.messageId, ok: ok}])
1770+ yield {messageId: e.messageId, ok: ok})
1771+ yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
1840handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp correlationId authorization respond -> do
1773handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp correlationId authorization respond -> do
1841 t <- now
1774 t <- now
@@ -1857,10 +1790,9 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
1857 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
1790 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
1859-- Pair up parallel lists by index. Builds index arrays and joins on i (O(n*m) but trivial code).
1792-- Pair up parallel lists by index. Builds index arrays and joins on i (O(n*m) but trivial code).
1793+withIndex = \xs -> (fold (\acc x -> {i: acc.i + 1, out: union acc.out [{i: acc.i, v: x}]}) {i: 0, out: []} xs).out
1860zipPairs = \mids keys ->
1794zipPairs = \mids keys ->
1861− let mIdx = (fold (\acc m -> {i: acc.i + 1, out: union acc.out [{i: acc.i, v: m}]}) {i: 0, out: []} mids).out in
1862− let kIdx = (fold (\acc k -> {i: acc.i + 1, out: union acc.out [{i: acc.i, v: k}]}) {i: 0, out: []} keys).out in
1863− do mp <- mIdx; kp <- kIdx; where mp.i == kp.i; yield {messageId: mp.v, toKey: kp.v}
1795+ do mp <- withIndex mids; kp <- withIndex keys; where mp.i == kp.i; yield {messageId: mp.v, toKey: kp.v}
1865-- ============================================
1797-- ============================================
1866-- Maintenance handlers
1798-- Maintenance handlers