@@ -449,9 +449,9 @@ countWhere = \xs pred -> count (filter xs pred)
449449any : [a] -> (a -> Bool) -> Bool
450450any = \xs pred -> countWhere xs pred > 0
451451none : [a] -> (a -> Bool) -> Bool
452−none = \xs pred -> not (any xs pred)
452+none = \xs pred -> countWhere xs pred == 0
453453all : [a] -> (a -> Bool) -> Bool
454−all = \xs pred -> none xs (\x -> not (pred x))
454+all = \xs pred -> countWhere xs pred == count xs
455455elem = \xs x -> any xs (\y -> y == x)
456456findFirst = \xs pred -> headOf (filter xs pred)
457457when = \cond action -> if cond then action else yield {}
@@ -524,6 +524,14 @@ emptyPollResp = \cursor authorized rateLimited draining gen retryAfterMs cursorW
524524-- Shared error shape for /messages send.
525525sendErrResp = \status err -> {status: status, messageId: "", retryAfterMs: 0, error: err, loadPct: 0}
526526
527+-- True when a send produced (or matched) a deliverable message.
528+isSendSuccess = \s -> case s of
529+ MsDelivered {} -> True {}
530+ MsFederated {} -> True {}
531+ MsQueued {} -> True {}
532+ MsDuplicate {} -> True {}
533+ _ -> False {}
534+
527535-- Sequence an IO action over each unique item (dedup by getKey).
528536forEach = \items getKey action -> case headOf items of
529537 Nothing {} -> yield {}
@@ -547,6 +555,9 @@ genTokenN = \n -> if n <= 0 then yield "" else do
547555
548556genToken = genTokenN 48
549557
558+-- Use the supplied message id when non-empty, otherwise generate a fresh one.
559+resolveMessageId = \mid -> if mid != "" then yield mid else genToken
560+
550561tooLong : Text -> Int -> Bool
551562tooLong = \s maxLen -> count (chars s) > maxLen
552563
@@ -756,10 +767,10 @@ recordFailure = \server -> do
756767 atomic do
757768 failures <- *forwardFailures
758769 let same = \f -> f.server == server
759− set *forwardFailures = union (removeWhere failures same)
760− [case findFirst failures same of
761− Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
762− Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}]
770+ let entry = case findFirst failures same of
771+ Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
772+ Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}
773+ set *forwardFailures = union (removeWhere failures same) [entry]
763774
764775clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))
765776
@@ -805,8 +816,9 @@ knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.serve
805816-- Last-write-wins merge by (pubkey, server), keeping the entry with the larger expiresAt.
806817mergePresence = \existing incoming ->
807818 fold (\acc e ->
808− if any acc (\x -> x.pubkey == e.pubkey && x.server == e.server && x.expiresAt >= e.expiresAt) then acc
809− else union (removeWhere acc (\x -> x.pubkey == e.pubkey && x.server == e.server)) [e]) existing incoming
819+ let same = \x -> x.pubkey == e.pubkey && x.server == e.server in
820+ if any acc (\x -> same x && x.expiresAt >= e.expiresAt) then acc
821+ else union (removeWhere acc same) [e]) existing incoming
810822
811823-- ============================================
812824-- Pruning
@@ -829,7 +841,8 @@ pruneExpired = \t -> atomic do
829841 -- presence: remember locally-expired keys for offline gossip
830842 pres <- *presence
831843 pruneRef presence (\p -> p.expiresAt > t)
832− yield (filter pres (\p -> p.expiresAt <= t && isLocalServer p.server && none pres (\k -> isLocalServer k.server && k.pubkey == p.pubkey && k.expiresAt > t)))
844+ let stillLive = \pk -> any pres (\k -> isLocalServer k.server && k.pubkey == pk && k.expiresAt > t)
845+ yield (filter pres (\p -> p.expiresAt <= t && isLocalServer p.server && not (stillLive p.pubkey)))
833846
834847-- ============================================
835848-- Federation gossip
@@ -901,12 +914,11 @@ retryPendingForwards = do
901914 Err _ -> do recordFailure server; yield False
902915 Ok r -> do when r.ok (clearFailure server); yield r.ok
903916 let isMatch = \x -> x.messageId == f.messageId && x.toServer == server
917+ let isDone = \x -> isMatch x && (success || x.retries + 1 >= maxForwardRetries)
904918 atomic do
905919 cur <- *forwards
906− set *forwards = do
907− x <- cur
908− where not (isMatch x && (success || x.retries + 1 >= maxForwardRetries))
909− yield (if isMatch x
920+ set *forwards = pluck (removeWhere cur isDone) (\x ->
921+ if isMatch x
910922 then {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}
911923 else x)))
912924
@@ -929,8 +941,10 @@ sendAcksAndClean = \pendingAcks ->
929941forwardQueuedToServer = \pubkey server t -> atomic do
930942 msgs <- *messages
931943 fwds <- *forwards
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)
944+ let candidates = filter msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.messageId == m.messageId && f.toServer == server))
945+ let newFwds = pluck (takeRelation maxForwardsPerServer candidates) (\m ->
946+ {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})
947+ set *forwards = capList (union fwds newFwds) maxForwardsEntries (\f -> f.createdAt)
934948
935949-- ============================================
936950-- Background loops
@@ -995,12 +1009,13 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
9951009 result <- atomic do
9961010 ch <- *challenges
9971011 dp <- *drainPhase
1012+ let same = \c -> c.pubkey == pubkey && c.challenge == challenge
9981013 if dp >= 1
999− || none ch (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && matchIp clientIp c.ip)
1014+ || none ch (\c -> same c && c.expiresAt > t && matchIp clientIp c.ip)
10001015 || not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))
10011016 then yield {valid: False {}, peers: []}
10021017 else do
1003− set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)
1018+ set *challenges = removeWhere ch same
10041019 ses <- *sessions
10051020 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)
10061021 pres <- *presence
@@ -1126,7 +1141,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
11261141 t <- now
11271142 dp <- *drainPhase
11281143 g <- checkGlobalRate t
1129− effMsgId <- if messageId != "" then yield messageId else genToken
1144+ effMsgId <- resolveMessageId messageId
11301145 let err = \s e -> yield (respond (sendErrResp s e))
11311146 if dp >= 2 then err (MsDraining {}) "draining"
11321147 else if g then err (MsRateLimited {}) "rate_limited"
@@ -1139,9 +1154,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
11391154 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effMsgId expiresInMs (clamp 0 maxMessagePriority priority) t)
11401155 when federationEnabled
11411156 (fork (forkForwardsTo effMsgId to encryptedBlob clampedTs result.remoteServers))
1142− let ok = case result.status of
1143− MsDelivered {} -> True {}; MsFederated {} -> True {}
1144− MsQueued {} -> True {}; MsDuplicate {} -> True {}; _ -> False {}
1157+ let ok = isSendSuccess result.status
11451158 yield (respond {status: result.status, messageId: if ok then effMsgId else "", retryAfterMs: 0, error: if ok then "" else "rejected", loadPct: 0})
11461159
11471160sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msgPriority t -> do
@@ -1325,7 +1338,7 @@ handleSendBatch = \messages authorization clientIp respond -> do
13251338 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
13261339 gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do
13271340 results <- mapEach (takeRelation maxBatchSize messages) id (\m -> do
1328− effMsgId <- if m.messageId != "" then yield m.messageId else genToken
1341+ effMsgId <- resolveMessageId m.messageId
13291342 let clampedTs = clampTimestamp m.timestamp t
13301343 r <- if t - m.timestamp > messageMaxAge
13311344 then yield {status: MsRejected {}, remoteServers: []}
@@ -1367,8 +1380,10 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
13671380 let expiresAt = min e.timestamp t + clampTtl e.ttl * 1000
13681381 where expiresAt > t
13691382 yield {pubkey: e.pubkey, server: e.server, expiresAt: expiresAt, timestamp: e.timestamp, eventType: e.eventType}
1370− 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})
1371− let offline = mapWhere validEvents (\v -> case v.eventType of GossipOffline {} -> True {}; _ -> False {}) (\v -> {pubkey: v.pubkey, server: v.server, timestamp: v.timestamp})
1383+ let isOnline = \v -> case v.eventType of GossipOnline {} -> True {}; _ -> False {}
1384+ let isOffline = \v -> case v.eventType of GossipOffline {} -> True {}; _ -> False {}
1385+ let online = mapWhere validEvents isOnline (\v -> {pubkey: v.pubkey, server: v.server, expiresAt: v.expiresAt, timestamp: v.timestamp})
1386+ let offline = mapWhere validEvents isOffline (\v -> {pubkey: v.pubkey, server: v.server, timestamp: v.timestamp})
13721387 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)
13731388 yield (mapWhere online (\e -> none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)) (\e -> {pubkey: e.pubkey, server: e.server}))
13741389 fork (forEach newlyOnline (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t))
@@ -1605,9 +1620,10 @@ handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth autho
16051620 (do
16061621 revoked <- atomic do
16071622 ses <- *sessions
1608− set *sessions = removeWhere ses (\s -> s.pubkey == pubkey)
1623+ let same = \s -> s.pubkey == pubkey
1624+ set *sessions = removeWhere ses same
16091625 pruneRef presence (\p -> p.pubkey != pubkey || not (isLocalServer p.server))
1610− yield (countWhere ses (\s -> s.pubkey == pubkey))
1626+ yield (countWhere ses same)
16111627 yield (respond {ok: True {}, revokedCount: revoked}))
16121628
16131629handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth authorization
@@ -1616,11 +1632,12 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
16161632 t <- now
16171633 purged <- atomic do
16181634 msgs <- *messages
1619− let mine = filter msgs (\m -> m.toKey == pubkey)
1620− set *messages = removeWhere msgs (\m -> m.toKey == pubkey)
1635+ let mine = \m -> m.toKey == pubkey
1636+ let purgedMsgs = filter msgs mine
1637+ set *messages = removeWhere msgs mine
16211638 pruneRef forwards (\f -> f.toKey != pubkey)
1622− appendCapped ackedIds (mapWhere mine (\m -> m.originServer != "") (\m -> mkAckedEntry m t)) maxAckedIds (\a -> a.ackedAt)
1623− yield (count mine)
1639+ appendCapped ackedIds (mapWhere purgedMsgs (\m -> m.originServer != "") (\m -> mkAckedEntry m t)) maxAckedIds (\a -> a.ackedAt)
1640+ yield (count purgedMsgs)
16241641 yield (respond {ok: True {}, purgedCount: purged}))
16251642
16261643handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
@@ -1628,8 +1645,9 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
16281645 (do
16291646 keys <- if ip == "" then yield [] else atomic do
16301647 ses <- *sessions
1631− set *sessions = removeWhere ses (\s -> s.ip == ip)
1632− yield (mapWhere ses (\s -> s.ip == ip) (\s -> s.pubkey))
1648+ let same = \s -> s.ip == ip
1649+ set *sessions = removeWhere ses same
1650+ yield (mapWhere ses same (\s -> s.pubkey))
16331651 yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count (unique keys)}))
16341652
16351653handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization