@@ -513,14 +513,14 @@ forEach = \items getKey action -> case headOf items of
513513 Nothing {} -> yield {}
514514 Just {value: x} -> do
515515 action x
516− forEach (do y <- items; where getKey y != getKey x; yield y) getKey action
516+ forEach (filter items (\y -> getKey y != getKey x)) getKey action
517517
518518-- Like forEach, but collects each action's result into a list.
519519mapEach = \items getKey action -> case headOf items of
520520 Nothing {} -> yield []
521521 Just {value: x} -> do
522522 r <- action x
523− rest <- mapEach (do y <- items; where getKey y != getKey x; yield y) getKey action
523+ rest <- mapEach (filter items (\y -> getKey y != getKey x)) getKey action
524524 yield (union [r] rest)
525525
526526-- Hex tokens
@@ -612,22 +612,16 @@ withMaintAuth = \authorization onUnauth onAuth -> do
612612 authorized <- verifyMaintenanceAuth authorization
613613 if authorized then onAuth else onUnauth
614614
615−-- Common gate for client handlers: drain check, global rate, session auth.
615+-- Common gate for handlers: optional drain check, global rate, session auth.
616616-- mkErr is called with one of: "draining", "rate_limited", "unauthorized".
617−gateAuth = \auth ip t respond mkErr onAuth -> do
617+gateAuth = \checkDrain auth ip t respond mkErr onAuth -> do
618618 dp <- *drainPhase
619− if dp >= 2 then yield (respond (mkErr "draining"))
619+ if checkDrain && dp >= 2 then yield (respond (mkErr "draining"))
620620 else do
621621 g <- checkGlobalRate t
622622 if g then yield (respond (mkErr "rate_limited"))
623623 else withSessionAuth auth t ip (yield (respond (mkErr "unauthorized"))) onAuth
624624
625−-- Variant without the drain check (used by lookup/status handlers).
626−gateAuthNoDrain = \auth ip t respond mkErr onAuth -> do
627− g <- checkGlobalRate t
628− if g then yield (respond (mkErr "rate_limited"))
629− else withSessionAuth auth t ip (yield (respond (mkErr "unauthorized"))) onAuth
630−
631625authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
632626authenticateSession = \sessions token t ->
633627 case findFirst sessions (\s -> s.expiresAt > t && s.token == token) of
@@ -764,6 +758,14 @@ checkPreAuth = \clientIp t -> do
764758 else if i then "ip_rate_limited"
765759 else "")
766760
761+-- Pre-auth check + auth-global rate. Returns "" when ok, else an error code.
762+authPreCheck = \clientIp t -> do
763+ p <- checkPreAuth clientIp t
764+ if p != "" then yield p
765+ else do
766+ a <- checkAuthGlobalRate t
767+ yield (if a then "auth_rate_limited" else "")
768+
767769-- ============================================
768770-- Federation failure tracking (simple, no circuit breaker)
769771-- ============================================
@@ -856,8 +858,9 @@ pruneExpired = \t -> atomic do
856858 -- presence: remember locally-expired keys for offline gossip
857859 pres <- *presence
858860 let keptPres = filter pres (\p -> p.expiresAt > t)
861+ let stillLocal = filter keptPres (\k -> isLocalServer k.server)
859862 set *presence = keptPres
860− yield (filter pres (\p -> p.expiresAt <= t && isLocalServer p.server && none keptPres (\k -> k.pubkey == p.pubkey && isLocalServer k.server)))
863+ yield (filter pres (\p -> p.expiresAt <= t && isLocalServer p.server && none stillLocal (\k -> k.pubkey == p.pubkey)))
861864
862865-- ============================================
863866-- Federation gossip
@@ -996,14 +999,11 @@ refreshLocalPresence = \t -> do
996999
9971000handleChallenge = \pubkey clientIp respond -> do
9981001 t <- now
999− if requireClientIp && clientIp == ""
1000− then yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: "client_ip_required"})
1001− else do
1002− preAuth <- checkPreAuth clientIp t
1003− if preAuth != "" then yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: preAuth})
1002+ let mkResp = \err -> respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: err}
1003+ if requireClientIp && clientIp == "" then yield (mkResp "client_ip_required")
10041004 else do
1005− authLimited <- checkAuthGlobalRate t
1006− if authLimited then yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: "auth_rate_limited"})
1005+ preErr <- authPreCheck clientIp t
1006+ if preErr != "" then yield (mkResp preErr)
10071007 else do
10081008 challenge <- genToken
10091009 result <- atomic do
@@ -1025,44 +1025,41 @@ handleChallenge = \pubkey clientIp respond -> do
10251025
10261026handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond -> do
10271027 t <- now
1028− preAuth <- checkPreAuth clientIp t
1029− if preAuth != "" then yield (respond {token: "", expiresAt: 0, error: preAuth})
1028+ preErr <- authPreCheck clientIp t
1029+ if preErr != "" then yield (respond {token: "", expiresAt: 0, error: preErr})
10301030 else do
1031− authLimited <- checkAuthGlobalRate t
1032− if authLimited then yield (respond {token: "", expiresAt: 0, error: "auth_rate_limited"})
1033− else do
1034− let expiry = t + sessionExpiry
1035− token <- genToken
1036− result <- atomic do
1037− ch <- *challenges
1038− let hasMatch = any ch (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && matchIp clientIp c.ip)
1039− if not hasMatch then yield {valid: False {}, peers: []}
1040− else
1041− if not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))
1042− then yield {valid: False {}, peers: []}
1031+ let expiry = t + sessionExpiry
1032+ token <- genToken
1033+ result <- atomic do
1034+ ch <- *challenges
1035+ let hasMatch = any ch (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && matchIp clientIp c.ip)
1036+ if not hasMatch then yield {valid: False {}, peers: []}
1037+ else
1038+ if not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))
1039+ then yield {valid: False {}, peers: []}
1040+ else do
1041+ set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)
1042+ dp <- *drainPhase
1043+ if dp >= 1 then yield {valid: False {}, peers: []}
10431044 else do
1044− set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)
1045− dp <- *drainPhase
1046− if dp >= 1 then yield {valid: False {}, peers: []}
1047− else do
1048− ses <- *sessions
1049− let kept = if revokeOthers then removeWhere ses (\s -> s.pubkey == pubkey) else ses
1050− let forKey = filter kept (\s -> s.pubkey == pubkey)
1051− let afterPerKey = if count forKey >= maxSessionsPerKey
1052− then union (removeWhere kept (\s -> s.pubkey == pubkey)) (evictOldest forKey (\s -> s.expiresAt))
1053− else kept
1054− let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
1055− set *sessions = capList (union afterPerKey [newSession]) maxSessions (\s -> s.expiresAt)
1056− pres <- *presence
1057− let withoutLocal = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
1058− let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1059− set *presence = newPres
1060− yield {valid: True {}, peers: knownPeerServersFromPresence newPres}
1061− if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})
1062− else do
1063− when federationEnabled
1064− (fork (sendBatchedGossip result.peers [mkGossip (GossipOnline {}) pubkey t onlineGossipTtl] gossipOriginHops))
1065− yield (respond {token: token, expiresAt: expiry, error: ""})
1045+ ses <- *sessions
1046+ let kept = if revokeOthers then removeWhere ses (\s -> s.pubkey == pubkey) else ses
1047+ let forKey = filter kept (\s -> s.pubkey == pubkey)
1048+ let afterPerKey = if count forKey >= maxSessionsPerKey
1049+ then union (removeWhere kept (\s -> s.pubkey == pubkey)) (evictOldest forKey (\s -> s.expiresAt))
1050+ else kept
1051+ let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
1052+ set *sessions = capList (union afterPerKey [newSession]) maxSessions (\s -> s.expiresAt)
1053+ pres <- *presence
1054+ let withoutLocal = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
1055+ let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1056+ set *presence = newPres
1057+ yield {valid: True {}, peers: knownPeerServersFromPresence newPres}
1058+ if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})
1059+ else do
1060+ when federationEnabled
1061+ (fork (sendBatchedGossip result.peers [mkGossip (GossipOnline {}) pubkey t onlineGossipTtl] gossipOriginHops))
1062+ yield (respond {token: token, expiresAt: expiry, error: ""})
10661063
10671064handleRevoke = \targetToken authorization clientIp respond -> do
10681065 t <- now
@@ -1306,7 +1303,7 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
13061303handleLookupPresence = \pubkey authorization clientIp respond -> do
13071304 t <- now
13081305 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}
1309− gateAuthNoDrain authorization clientIp t respond mkErr (\senderKey -> do
1306+ gateAuth (False {}) authorization clientIp t respond mkErr (\senderKey -> do
13101307 lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t
13111308 if lim then yield (respond {server: "", authorized: True {}, error: "rate_limited"})
13121309 else do
@@ -1318,7 +1315,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
13181315handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
13191316 t <- now
13201317 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}
1321− gateAuthNoDrain authorization clientIp t respond mkErr (\senderKey -> do
1318+ gateAuth (False {}) authorization clientIp t respond mkErr (\senderKey -> do
13221319 lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t
13231320 if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
13241321 else do
@@ -1333,7 +1330,7 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
13331330handleMessageStatus = \to messageId authorization clientIp respond -> do
13341331 t <- now
13351332 let mkErr = \_ -> {status: "", authorized: False {}}
1336− gateAuthNoDrain authorization clientIp t respond mkErr (\_ -> do
1333+ gateAuth (False {}) authorization clientIp t respond mkErr (\_ -> do
13371334 msgs <- *messages
13381335 fwds <- *forwards
13391336 aids <- *ackedIds
@@ -1352,7 +1349,7 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
13521349handleAckMessages = \messageIds authorization clientIp respond -> do
13531350 t <- now
13541351 let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}
1355− gateAuth authorization clientIp t respond mkErr (\pubkey -> do
1352+ gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do
13561353 lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t
13571354 if lim then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})
13581355 else do
@@ -1371,7 +1368,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
13711368handleListMessages = \offset authorization clientIp respond -> do
13721369 t <- now
13731370 let mkErr = \e -> {messages: [], authorized: False {}, count: 0, error: if e == "rate_limited" then "rate_limited" else ""}
1374− gateAuth authorization clientIp t respond mkErr (\pubkey -> do
1371+ gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do
13751372 lim <- checkPubkeyRateGate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs t
13761373 if lim then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})
13771374 else do
@@ -1386,7 +1383,7 @@ handleListMessages = \offset authorization clientIp respond -> do
13861383handleDeleteMessages = \messageIds authorization clientIp respond -> do
13871384 t <- now
13881385 let mkErr = \e -> {ok: False {}, authorized: False {}, deletedCount: 0, error: if e == "draining" then "" else e}
1389− gateAuth authorization clientIp t respond mkErr (\pubkey -> do
1386+ gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do
13901387 lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t
13911388 if lim then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})
13921389 else do
@@ -1406,7 +1403,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
14061403handleCancelMessage = \messageId to authorization clientIp respond -> do
14071404 t <- now
14081405 let mkErr = \e -> {ok: False {}, authorized: False {}, cancelled: False {}, error: if e == "draining" then "" else e}
1409− gateAuth authorization clientIp t respond mkErr (\senderKey -> do
1406+ gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do
14101407 lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t
14111408 if lim then yield (respond {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited"})
14121409 else do
@@ -1446,7 +1443,7 @@ sendCancelToEach = \messageId to servers ->
14461443handleCancelBatch = \cancels authorization clientIp respond -> do
14471444 t <- now
14481445 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "draining" then "" else e}
1449− gateAuth authorization clientIp t respond mkErr (\senderKey -> do
1446+ gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do
14501447 lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t
14511448 if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
14521449 else do
@@ -1470,7 +1467,7 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
14701467handleSendBatch = \messages authorization clientIp respond -> do
14711468 t <- now
14721469 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
1473− gateAuth authorization clientIp t respond mkErr (\senderKey -> do
1470+ gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do
14741471 results <- mapEach (takeRelation maxBatchSize messages) id (\m -> do
14751472 effMsgId <- if m.messageId != "" then yield m.messageId else genToken
14761473 let clampedTs = clampTimestamp m.timestamp t
@@ -1528,12 +1525,9 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
15281525 let newlyOnline = mapWhere online (\e -> not (isLocalServer e.server) && none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)) (\e -> {pubkey: e.pubkey, server: e.server})
15291526 yield {newlyOnline: newlyOnline}
15301527 when (count valid.newlyOnline > 0)
1531− (fork (forwardForNewlyOnline valid.newlyOnline t))
1528+ (fork (forEach valid.newlyOnline (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t)))
15321529 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
15331530
1534−forwardForNewlyOnline = \targets t ->
1535− forEach targets (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t)
1536−
15371531handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond -> do
15381532 t <- now
15391533 withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do