@@ -464,13 +464,26 @@ maxBy = \items getKey -> fold (\acc x -> case acc of
464464evictOldest = \items getKey ->
465465 case minBy items getKey of
466466 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
470468
471469capList = \items maxN getKey ->
472470 if count items > maxN then evictOldest items getKey else items
473471
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
476+ action x
477+ forEach (do y <- items; where getKey y != getKey x; yield y) getKey action
478+
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
483+ r <- action x
484+ rest <- mapEach (do y <- items; where getKey y != getKey x; yield y) getKey action
485+ yield (union [r] rest)
486+
474487-- Hex tokens
475488hexDigit : Int -> Text
476489hexDigit = \n -> take 1 (drop (clamp 0 15 n) "0123456789abcdef")
@@ -498,11 +511,10 @@ tooLong = \s maxLen -> drop maxLen s != ""
498511
499512isValidHex : Text -> Bool
500513isValidHex = \s ->
501− if s == "" then False {}
502− else if drop 1 s == "" then False {}
503− else if contains (take 1 s) "0123456789abcdef" && contains (take 1 (drop 1 s)) "0123456789abcdef"
504− then drop 2 s == "" || isValidHex (drop 2 s)
505− else False {}
514+ s != "" && drop 1 s != ""
515+ && contains (take 1 s) "0123456789abcdef"
516+ && contains (take 1 (drop 1 s)) "0123456789abcdef"
517+ && (drop 2 s == "" || isValidHex (drop 2 s))
506518
507519normalizeHex : Text -> Text
508520normalizeHex = \s -> bytesToHex (bytesFromHex s)
@@ -852,24 +864,18 @@ pruneExpired = \t -> do
852864-- Federation gossip
853865-- ============================================
854866
855−sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}
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
862− failures <- *forwardFailures
863− when (shouldRetryServer server t failures) (do
864− auth <- generateFederationAuth "gossip" t
865− result <- fetch (federationUrl server) (RecvGossip {
866− authorization: auth,
867− events: events, hopsRemaining: hops, fromServer: serverHostN,
868− requestTimestamp: t, correlationId: "", protocolVersion: federationProtocolVersion})
869− case result of
870− Err _ -> recordFailure server
871− Ok _ -> clearFailure server)
872− sendBatchedGossip rest events hops
867+sendBatchedGossip = \peers events hops -> forEach peers (\s -> s) (\server -> do
868+ t <- now
869+ failures <- *forwardFailures
870+ when (shouldRetryServer server t failures) (do
871+ auth <- generateFederationAuth "gossip" t
872+ result <- fetch (federationUrl server) (RecvGossip {
873+ authorization: auth,
874+ events: events, hopsRemaining: hops, fromServer: serverHostN,
875+ requestTimestamp: t, correlationId: "", protocolVersion: federationProtocolVersion})
876+ case result of
877+ Err _ -> recordFailure server
878+ Ok _ -> clearFailure server))
873879
874880broadcastOfflineGossip = \t expiredLocal -> do
875881 presence <- *presence
@@ -882,37 +888,21 @@ broadcastOfflineGossip = \t expiredLocal -> do
882888-- Forward retries and acks
883889-- ============================================
884890
885−retryPendingForwards : IO {network} {}
886891retryPendingForwards = do
887892 t <- now
888893 fwds <- *forwards
889894 failures <- *forwardFailures
890895 blocked <- *blockedPeers
891− -- Group by server
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
894− retryServerForwards eligibleServers fwds
895−
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)
902− sendForwardsToServer serverFwds server
903− retryServerForwards rest fwds
904−
905−sendForwardsToServer = \fwds server ->
906− case headOf fwds of
907− Nothing {} -> yield {}
908− Just {value: f} -> do
909− let rest = do x <- fwds; where x.messageId != f.messageId; yield x
910− t <- now
911− auth <- generateFederationAuth ("forward:" ++ f.messageId) t
896+ let servers = do s <- collectPeers fwds (\f -> f.toServer); where shouldRetryServer s t failures && not (isBlockedPeer s blocked); yield s
897+ forEach servers (\s -> s) (\server -> do
898+ let serverFwds = takeRelation maxBatchSize (do f <- fwds; where f.toServer == server; yield f)
899+ forEach serverFwds (\f -> f.messageId) (\f -> do
900+ t2 <- now
901+ auth <- generateFederationAuth ("forward:" ++ f.messageId) t2
912902 result <- fetch (federationUrl server) (ForwardMessage {
913903 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
914904 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
915− fromServer: serverHostN, requestTimestamp: t,
905+ fromServer: serverHostN, requestTimestamp: t2,
916906 authorization: auth, correlationId: "",
917907 expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})
918908 let success = case result of Ok r -> r.ok; _ -> False {}
@@ -923,37 +913,24 @@ sendForwardsToServer = \fwds server ->
923913 x <- cur
924914 where x.messageId != f.messageId || x.toServer != server || (not success && x.retries + 1 < maxForwardRetries)
925915 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})
927− sendForwardsToServer rest server
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})))
928917
929−sendAcksAndClean : [AckedEntry] -> IO {network} {}
930918sendAcksAndClean = \pendingAcks ->
931− if count pendingAcks == 0 then yield {}
932− else do
933− let servers = collectPeers pendingAcks (\a -> a.originServer)
934− sendAcksToEach servers pendingAcks
935−
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
942− let mids = do a <- serverAcks; yield a.messageId
943− t <- now
944− auth <- generateFederationAuth "ack" t
945− result <- fetch (federationUrl server) (FederationAck {
946− authorization: auth,
947− messageIds: mids, originServer: server, fromServer: serverHostN,
948− requestTimestamp: t, correlationId: "", protocolVersion: federationProtocolVersion})
949− case result of
950− Err _ -> recordFailure server
951− Ok _ -> do
952− clearFailure server
953− atomic do
954− cur <- *ackedIds
955− set *ackedIds = do a <- cur; where not (a.originServer == server && any mids (\m -> m == a.messageId)); yield a
956− sendAcksToEach rest acks
919+ forEach (collectPeers pendingAcks (\a -> a.originServer)) (\s -> s) (\server -> do
920+ let mids = do a <- pendingAcks; where a.originServer == server; yield a.messageId
921+ t <- now
922+ auth <- generateFederationAuth "ack" t
923+ result <- fetch (federationUrl server) (FederationAck {
924+ authorization: auth,
925+ messageIds: mids, originServer: server, fromServer: serverHostN,
926+ requestTimestamp: t, correlationId: "", protocolVersion: federationProtocolVersion})
927+ case result of
928+ Err _ -> recordFailure server
929+ Ok _ -> do
930+ clearFailure server
931+ atomic do
932+ cur <- *ackedIds
933+ set *ackedIds = do a <- cur; where not (a.originServer == server && any mids (\m -> m == a.messageId)); yield a)
957934
958935-- Forward queued messages to a server when a remote key comes online.
959936forwardQueuedToServer = \pubkey server t -> atomic do
@@ -1322,36 +1299,32 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
13221299 yield {status: status, remoteServers: remoteServers}
13231300
13241301forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
1325− case headOf servers of
1326− Nothing {} -> yield {}
1327− Just {value: server} -> do
1328− let rest = do s <- servers; where s != server; yield s
1329− info <- atomic do
1330− fwds <- *forwards
1331− case headOf (do f <- fwds; where f.messageId == effMsgId && f.toServer == server; yield f) of
1332− Nothing {} -> yield Nothing {}
1333− Just {value: f} -> yield (Just {value: {expiresAt: f.expiresAt, priority: f.priority}})
1334− case info of
1335− Nothing {} -> yield {}
1336− Just {value: i} -> do
1337− t <- now
1338− auth <- generateFederationAuth ("forward:" ++ effMsgId) t
1339− result <- fetch (federationUrl server) (ForwardMessage {
1340− toKey: to, encryptedBlob: encryptedBlob,
1341− timestamp: clampedTs, originServer: serverHostN, messageId: effMsgId,
1342− fromServer: serverHostN, requestTimestamp: t,
1343− authorization: auth, correlationId: "",
1344− expiresAt: i.expiresAt, priority: i.priority, protocolVersion: federationProtocolVersion})
1345− case result of
1346− Err _ -> recordFailure server
1347− Ok r ->
1348− if r.ok then do
1349− clearFailure server
1350− atomic do
1351− fwds <- *forwards
1352− set *forwards = do f <- fwds; where not (f.messageId == effMsgId && f.toServer == server); yield f
1353− else yield {}
1354− forkForwardsTo effMsgId to encryptedBlob clampedTs rest
1302+ forEach servers (\s -> s) (\server -> do
1303+ info <- atomic do
1304+ fwds <- *forwards
1305+ case headOf (do f <- fwds; where f.messageId == effMsgId && f.toServer == server; yield f) of
1306+ Nothing {} -> yield Nothing {}
1307+ Just {value: f} -> yield (Just {value: {expiresAt: f.expiresAt, priority: f.priority}})
1308+ case info of
1309+ Nothing {} -> yield {}
1310+ Just {value: i} -> do
1311+ t <- now
1312+ auth <- generateFederationAuth ("forward:" ++ effMsgId) t
1313+ result <- fetch (federationUrl server) (ForwardMessage {
1314+ toKey: to, encryptedBlob: encryptedBlob,
1315+ timestamp: clampedTs, originServer: serverHostN, messageId: effMsgId,
1316+ fromServer: serverHostN, requestTimestamp: t,
1317+ authorization: auth, correlationId: "",
1318+ expiresAt: i.expiresAt, priority: i.priority, protocolVersion: federationProtocolVersion})
1319+ case result of
1320+ Err _ -> recordFailure server
1321+ Ok r ->
1322+ if r.ok then do
1323+ clearFailure server
1324+ atomic do
1325+ fwds <- *forwards
1326+ set *forwards = do f <- fwds; where not (f.messageId == effMsgId && f.toServer == server); yield f
1327+ else yield {})
13551328
13561329-- ============================================
13571330-- Handler: lookup
@@ -1535,24 +1508,17 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
15351508 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
15361509
15371510sendCancelToEach = \messageId to servers ->
1538− case headOf servers of
1539− Nothing {} -> yield {}
1540− Just {value: server} -> do
1541− let rest = do s <- servers; where s != server; yield s
1542− t <- now
1543− auth <- generateFederationAuth ("cancel:" ++ messageId) t
1544− result <- fetch (federationUrl server) (FederationCancel {
1545− messageId: messageId, toKey: to, fromServer: serverHostN,
1546− requestTimestamp: t, correlationId: "", authorization: auth,
1547− protocolVersion: federationProtocolVersion})
1548− case result of
1549− Err _ -> atomic do
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
1553− pfc <- *pendingFederationCancels
1554− set *pendingFederationCancels = capList (union pfc [{messageId: messageId, toKey: to, toServer: server, createdAt: t, retries: 0}]) 10000 (\c -> c.createdAt))
1555− sendCancelToEach messageId to rest
1511+ forEach servers (\s -> s) (\server -> do
1512+ t <- now
1513+ auth <- generateFederationAuth ("cancel:" ++ messageId) t
1514+ result <- fetch (federationUrl server) (FederationCancel {
1515+ messageId: messageId, toKey: to, fromServer: serverHostN,
1516+ requestTimestamp: t, correlationId: "", authorization: auth,
1517+ protocolVersion: federationProtocolVersion})
1518+ let failed = case result of Err _ -> True {}; Ok r -> not r.ok
1519+ when failed (atomic do
1520+ pfc <- *pendingFederationCancels
1521+ set *pendingFederationCancels = capList (union pfc [{messageId: messageId, toKey: to, toServer: server, createdAt: t, retries: 0}]) 10000 (\c -> c.createdAt)))
15561522
15571523handleCancelBatch = \cancels authorization clientIp respond -> do
15581524 t <- now
@@ -1567,32 +1533,25 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
15671533 lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t
15681534 if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
15691535 else do
1570− let capped = takeRelation maxBatchSize cancels
1571− results <- cancelEach senderKey capped t []
1536+ results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do
1537+ r <- atomic do
1538+ msgs <- *messages
1539+ let target = headOf (do m <- msgs; where m.messageId == e.messageId && m.toKey == e.to; yield m)
1540+ case target of
1541+ Nothing {} -> yield False
1542+ Just {value: msg} ->
1543+ if msg.originServer != "" || msg.senderKey == "" || msg.senderKey != senderKey
1544+ then yield False
1545+ else do
1546+ set *messages = do m <- msgs; where not (m.messageId == e.messageId && m.toKey == e.to); yield m
1547+ fwds <- *forwards
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)
1551+ yield True
1552+ yield {messageId: e.messageId, cancelled: r})
15721553 yield (respond {results: results, authorized: True {}, error: ""}))
15731554
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
1579− r <- atomic do
1580− msgs <- *messages
1581− let target = headOf (do m <- msgs; where m.messageId == e.messageId && m.toKey == e.to; yield m)
1582− case target of
1583− Nothing {} -> yield False
1584− Just {value: msg} ->
1585− if msg.originServer != "" || msg.senderKey == "" || msg.senderKey != senderKey
1586− then yield False
1587− else do
1588− set *messages = do m <- msgs; where not (m.messageId == e.messageId && m.toKey == e.to); yield m
1589− fwds <- *forwards
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)
1593− yield True
1594− cancelEach senderKey rest t (union acc [{messageId: e.messageId, cancelled: r}])
1595−
15961555handleSendBatch = \messages authorization clientIp respond -> do
15971556 t <- now
15981557 dp <- *drainPhase
@@ -1603,24 +1562,17 @@ handleSendBatch = \messages authorization clientIp respond -> do
16031562 else withSessionAuth authorization t clientIp
16041563 (yield (respond {results: [], authorized: False {}, retryAfterMs: 0}))
16051564 (\senderKey -> do
1606− let capped = takeRelation maxBatchSize messages
1607− results <- sendBatchEach senderKey capped t []
1608− yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}, retryAfterMs: 0}))
1609−
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
1616− let clampedTs = clampTimestamp m.timestamp t
1617− let clampedPri = clamp 0 maxMessagePriority m.priority
1618− r <- if t - m.timestamp > messageMaxAge
1619− then yield {status: MsRejected {}, remoteServers: []}
1620− else atomic do sendMessageCore senderKey m.to m.encryptedBlob clampedTs effMsgId m.expiresInMs clampedPri t
1621− when (federationEnabled && count r.remoteServers > 0)
1622− (fork (forkForwardsTo effMsgId m.to m.encryptedBlob clampedTs r.remoteServers))
1623− sendBatchEach senderKey rest t (union acc [{messageId: effMsgId, status: r.status}])
1565+ results <- mapEach (takeRelation maxBatchSize messages) (\m -> m) (\m -> do
1566+ effMsgId <- if m.messageId != "" then yield m.messageId else genToken
1567+ let clampedTs = clampTimestamp m.timestamp t
1568+ let clampedPri = clamp 0 maxMessagePriority m.priority
1569+ r <- if t - m.timestamp > messageMaxAge
1570+ then yield {status: MsRejected {}, remoteServers: []}
1571+ else atomic do sendMessageCore senderKey m.to m.encryptedBlob clampedTs effMsgId m.expiresInMs clampedPri t
1572+ when (federationEnabled && count r.remoteServers > 0)
1573+ (fork (forkForwardsTo effMsgId m.to m.encryptedBlob clampedTs r.remoteServers))
1574+ yield {messageId: effMsgId, status: r.status})
1575+ yield (respond {results: results, authorized: True {}, retryAfterMs: 0}))
16241576
16251577-- ============================================
16261578-- Handler: federation receive
@@ -1678,12 +1630,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
16781630 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
16791631
16801632forwardForNewlyOnline = \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
1686− forwardForNewlyOnline rest t
1633+ forEach targets (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t)
16871634
16881635handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond -> do
16891636 t <- now
@@ -1774,15 +1721,7 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
17741721 t <- now
17751722 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
17761723 withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (do
1777− let capped = takeRelation maxBatchSize cancels
1778− results <- federationCancelEach capped fromServer t []
1779− yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
1780−
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
1724+ results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do
17861725 cancelled <- atomic do
17871726 msgs <- *messages
17881727 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 ->
17931732 fwds <- *forwards
17941733 set *forwards = do f <- fwds; where not (f.messageId == e.messageId && f.toKey == e.toKey); yield f
17951734 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}))
17971737
17981738handleFederationBatchForward = \forwards fromServer requestTimestamp correlationId authorization respond -> do
17991739 t <- now
18001740 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
18011741 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (do
1802− let capped = takeRelation maxBatchSize forwards
1803− results <- federationForwardEach capped fromServer t []
1804− yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
1805−
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
1742+ results <- mapEach (takeRelation maxBatchSize forwards) (\e -> e.messageId) (\e -> do
18111743 ok <- if isLocalServer e.originServer || t - e.timestamp > messageMaxAge then yield False
18121744 else atomic do
18131745 pres <- *presence
@@ -1835,7 +1767,8 @@ federationForwardEach = \entries fromServer t acc ->
18351767 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}]
18361768 set *pollWakeup = seqTs
18371769 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}))
18391772
18401773handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp correlationId authorization respond -> do
18411774 t <- now
@@ -1857,10 +1790,9 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
18571790 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
18581791
18591792-- 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
18601794zipPairs = \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}
18641796
18651797-- ============================================
18661798-- Maintenance handlers