@@ -29,9 +29,8 @@ federationScheme = "https://" -- scheme for federation HTTP calls
2929-- Optional pre-shared key for federation auth. When non-empty and a peer's server name
3030-- matches, the key is used instead of callback verification — eliminates the verify
3131-- round-trip for trusted peers. Format: "server1:key1,server2:key2".
32−-- TODO: Implement pre-shared key parsing and lookup. Currently callback verification
33−-- is always used. Pre-shared keys would halve federation latency for configured peers
34−-- and remove the /federation/verify endpoint as an attack surface for those peers.
32+-- Keys must be at least 16 hex characters. Peers with pre-shared keys skip the
33+-- /federation/verify callback entirely, halving federation latency.
3534federationPresharedKeys = ""
3635-- NOTE: listenHost is not configurable — Knot's listen primitive only accepts a port.
3736-- To restrict the bind address (e.g., 127.0.0.1 behind a reverse proxy), use
@@ -882,6 +881,12 @@ route FederationApi where
882881 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /ack headers {authorization: Text}
883882 -> {ok: Bool, error: Text, ackedCount: Int}
884883 = FederationAck
884+ -- Cancel (retract) a previously forwarded message. The origin server sends this
885+ -- when the sender cancels a message via /messages/cancel. The receiving server
886+ -- deletes the message if it hasn't been polled by the recipient yet.
887+ POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /cancel headers {authorization: Text}
888+ -> {ok: Bool, error: Text, cancelled: Bool}
889+ = FederationCancel
885890
886891-- Federation callback verification: the receiving server calls back to the sender
887892-- to confirm a request token is genuine. No shared secret needed.
@@ -1324,41 +1329,107 @@ isBlockedPeer = \server blocked -> any blocked (\b -> b == server)
13241329isAllowedByAllowlist : Text -> [Text] -> Bool
13251330isAllowedByAllowlist = \server allowed -> count allowed == 0 || any allowed (\a -> a == server)
13261331
1332+-- ============================================
1333+-- Pre-shared Federation Key Helpers
1334+-- ============================================
1335+-- Parse "server1:key1,server2:key2" into a lookup-ready list.
1336+-- Keys provide HMAC-style auth: the outbound side sends "PresharedKey <key>" in the
1337+-- Authorization header; the inbound side checks safeCompare against the stored key.
1338+-- No callback round-trip needed for configured peers.
1339+
1340+-- Parse one "server:key" entry from the head of a comma-separated string.
1341+-- Returns {server, key, rest} where rest is everything after the comma (or "").
1342+parsePskEntry : Text -> {server: Text, key: Text, rest: Text}
1343+parsePskEntry = \s -> parsePskServer s ""
1344+
1345+parsePskServer : Text -> Text -> {server: Text, key: Text, rest: Text}
1346+parsePskServer = \s acc ->
1347+ if s == "" then {server: acc, key: "", rest: ""}
1348+ else if take 1 s == ":" then parsePskKey (drop 1 s) acc ""
1349+ else if take 1 s == "," then {server: acc, key: "", rest: drop 1 s}
1350+ else parsePskServer (drop 1 s) (acc ++ take 1 s)
1351+
1352+parsePskKey : Text -> Text -> Text -> {server: Text, key: Text, rest: Text}
1353+parsePskKey = \s server acc ->
1354+ if s == "" then {server: server, key: acc, rest: ""}
1355+ else if take 1 s == "," then {server: server, key: acc, rest: drop 1 s}
1356+ else parsePskKey (drop 1 s) server (acc ++ take 1 s)
1357+
1358+-- Parse the full config string into a list of {server, key} pairs.
1359+-- Filters out entries with empty server/key or keys shorter than 16 chars.
1360+parsePresharedKeys : Text -> [{server: Text, key: Text}]
1361+parsePresharedKeys = \s ->
1362+ if s == "" then []
1363+ else
1364+ let entry = parsePskEntry s in
1365+ let rest = parsePresharedKeys entry.rest in
1366+ if entry.server == "" || entry.key == "" || not (tooLong entry.key 15)
1367+ then rest
1368+ else union [{server: normalizeServerName entry.server, key: entry.key}] rest
1369+
1370+-- Pre-parsed at startup (evaluated once since federationPresharedKeys is a constant)
1371+presharedKeyList = parsePresharedKeys federationPresharedKeys
1372+
1373+-- Look up a pre-shared key for a server. Returns "" if no key configured.
1374+lookupPresharedKey : Text -> Text
1375+lookupPresharedKey = \server ->
1376+ fold (\acc entry -> if acc != "" then acc else if entry.server == server then entry.key else "") "" presharedKeyList
1377+
1378+-- Check if an authorization header contains a valid pre-shared key for the given server.
1379+-- Uses constant-time comparison to prevent timing side-channels.
1380+verifyPresharedKey : Text -> Text -> Bool
1381+verifyPresharedKey = \server authorization ->
1382+ case extractToken authorization of
1383+ Nothing {} -> False {}
1384+ Just {value: token} ->
1385+ let psk = lookupPresharedKey server in
1386+ psk != "" && safeCompare token psk
1387+
13271388-- Generate a single-use token for outbound federation auth.
1328−-- The receiving server calls back to /federation/verify to confirm we issued it.
1389+-- If a pre-shared key is configured for the target server, uses it directly
1390+-- (no token generation, no pending verification entry, no callback needed).
1391+-- Otherwise falls back to the callback verification flow.
13291392-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
13301393-- issued for one operation cannot authorize a different one.
1331−generateFederationAuth : Text -> IO {network} Text
1332−generateFederationAuth = \scope -> do
1333− token <- genToken
1334− t <- now
1335− atomic do
1336− pending <- *pendingVerifications
1337− let updated = union pending [{token: token, createdAt: t, scope: scope}]
1338− set *pendingVerifications = if count updated > maxPendingVerifications
1339− then evictOneOldest updated (\p -> p.createdAt)
1340− else updated
1341− yield ("Bearer " ++ token)
1342−
1343−-- Verify an inbound federation token by callback to the sender's /federation/verify endpoint.
1344−-- Uses fetchFederation for in-flight tracking, connection counting, and slow-response logging,
1345−-- preventing a malicious peer from slow-loris attacks on the verify endpoint.
1346−-- Single-use: the sender consumes the token on verification.
1347−-- scope must match the scope the token was generated with; empty scope accepts any (backward compat).
1394+generateFederationAuth : Text -> Text -> IO {network} Text
1395+generateFederationAuth = \scope server ->
1396+ -- Fast path: use pre-shared key if configured for this peer (no token generation,
1397+ -- no pending verification entry, no callback round-trip on the receiving side)
1398+ let psk = lookupPresharedKey server in
1399+ if psk != "" then yield ("Bearer " ++ psk)
1400+ else do
1401+ token <- genToken
1402+ t <- now
1403+ atomic do
1404+ pending <- *pendingVerifications
1405+ let updated = union pending [{token: token, createdAt: t, scope: scope}]
1406+ set *pendingVerifications = if count updated > maxPendingVerifications
1407+ then evictOneOldest updated (\p -> p.createdAt)
1408+ else updated
1409+ yield ("Bearer " ++ token)
1410+
1411+-- Verify an inbound federation token. First checks if a pre-shared key is configured
1412+-- for the sender — if so, verifies via constant-time comparison (no network round-trip).
1413+-- Otherwise falls back to callback verification via the sender's /federation/verify endpoint.
1414+-- Pre-shared key auth halves federation latency and removes the verify endpoint as an
1415+-- attack surface for configured peers.
13481416-- No retry: a failed verify rejects the request. The peer's federation rate limit + circuit breaker
13491417-- handle unreliable peers; retrying here doubles verify traffic and blocks with a 1s sleep per op,
13501418-- which a malicious peer can exploit for amplification.
13511419verifyFederationCallback : Text -> Text -> Text -> IO {network} Bool
13521420verifyFederationCallback = \fromServer authorization scope ->
1353− case extractToken authorization of
1354− Nothing {} -> yield (False {})
1355− Just {value: token} -> do
1356− result <- fetchFederation fromServer (VerifyToken {token: token, scope: scope})
1357− case result of
1358− Err e -> do
1359− logWarn ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
1360− yield (False {})
1361− Ok r -> yield r.ok
1421+ -- Fast path: pre-shared key verification (no network round-trip)
1422+ if verifyPresharedKey fromServer authorization then yield (True {})
1423+ else
1424+ case extractToken authorization of
1425+ Nothing {} -> yield (False {})
1426+ Just {value: token} -> do
1427+ result <- fetchFederation fromServer (VerifyToken {token: token, scope: scope})
1428+ case result of
1429+ Err e -> do
1430+ logWarn ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
1431+ yield (False {})
1432+ Ok r -> yield r.ok
13621433
13631434-- ============================================
13641435-- Session Bucket Helpers
@@ -2726,11 +2797,9 @@ removeOneForKey = \items pubkey ->
27262797
27272798-- Extract unique remote (non-local) server names from a list of items.
27282799-- O(n*k) where k = unique servers; acceptable since k is typically small (<20).
2729−-- NOTE: `union acc [s]` copies acc (O(|acc|) per step, O(n*k) total). Using
2730−-- `union [s] acc` would be O(1) per step but reverses insertion order. Since this
2731−-- is a set (order doesn't matter), prepend is preferred for large collections.
2732−-- TODO: Switch to `union [s] acc` across set-accumulating folds (collectPeerServers,
2733−-- uniqueBucketsFor, dedupMessageIds) if Knot confirms union copies left operand.
2800+-- NOTE: `union [s] acc` prepends (O(1) per step) vs `union acc [s]` which copies
2801+-- acc (O(|acc|) per step). Since this is a set (order doesn't matter), prepend is
2802+-- preferred for large collections. All set-accumulating folds use this pattern.
27342803collectPeerServers : [a] -> (a -> Text) -> [Text]
27352804collectPeerServers = \items getServer ->
27362805 fold (\acc item ->
@@ -2861,7 +2930,7 @@ pruneExpired = \t -> do
28612930 presenceResults <- prunePresenceBuckets 0 t
28622931 -- Dedup peer servers once after collecting from all 16 buckets (O(n) total
28632932 -- vs O(n*k) if deduped per-bucket during aggregation).
2864− let dedupServers = \servers -> fold (\acc s -> if any acc (\x -> x == s) then acc else union acc [s]) [] servers
2933+ let dedupServers = \servers -> fold (\acc s -> if any acc (\x -> x == s) then acc else union [s] acc) [] servers
28652934 let keptPeers = dedupServers presenceResults.keptPeerServers
28662935 let allPeers = dedupServers presenceResults.allPeerServers
28672936 atomic do
@@ -3017,7 +3086,7 @@ sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {
30173086sendBatchedGossipToServer = \server allEvents hops ->
30183087 forEachBatch allEvents maxGossipEvents (\batch -> do
30193088 t <- now
3020− auth <- generateFederationAuth "gossip"
3089+ auth <- generateFederationAuth "gossip" server
30213090 corrId <- genTokenN 1
30223091 result <- fetchFederation server (RecvGossip {
30233092 authorization: auth,
@@ -3378,7 +3447,7 @@ retryForwardsForServer = \entries remaining ->
33783447 Just {value: f} -> do
33793448 let rest = do x <- entries; where x.messageId != f.messageId; yield x
33803449 t <- now
3381− auth <- generateFederationAuth "forward"
3450+ auth <- generateFederationAuth "forward" f.toServer
33823451 corrId <- genTokenN 1
33833452 result <- fetchFederation f.toServer (ForwardMessage {
33843453 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
@@ -3456,7 +3525,7 @@ forwardQueuedMessages = \targets t -> do
34563525
34573526fetchFederationAck = \origin messageIds -> do
34583527 t <- now
3459− auth <- generateFederationAuth "ack"
3528+ auth <- generateFederationAuth "ack" origin
34603529 corrId <- genTokenN 1
34613530 fetchFederation origin (FederationAck {
34623531 authorization: auth,
@@ -3684,7 +3753,7 @@ pruneAndRetry = \t -> do
36843753 yield {extendedActive: [], nearExpiryActive: [], refreshPeers: knownPeers, shouldRefresh: True {}}
36853754 else do
36863755 -- Check per-key session existence (avoids materializing all 16 session buckets)
3687− let nearExpiryPubkeys = fold (\acc p -> if any acc (\a -> a == p.pubkey) then acc else union acc [p.pubkey]) [] localNearExpiry
3756+ let nearExpiryPubkeys = fold (\acc p -> if any acc (\a -> a == p.pubkey) then acc else union [p.pubkey] acc) [] localNearExpiry
36883757 activeKeys <- filterActiveSessionKeys nearExpiryPubkeys t
36893758 let toExtend = do p <- localNearExpiry; where any activeKeys (\k -> k == p.pubkey); yield p
36903759 let notExtended = do p <- localNearExpiry; where none activeKeys (\k -> k == p.pubkey); yield p
@@ -3730,6 +3799,10 @@ pruneAndRetry = \t -> do
37303799 let refreshEvents = do p <- result.extendedActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000)}
37313800 when (count result.refreshPeers > 0 && count refreshEvents > 0)
37323801 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
3802+ -- Update cached health state so /health returns fresh data even when rate-limited.
3803+ -- This runs every prune cycle (~30s), keeping the cache within one interval of reality.
3804+ computeAndCacheHealth t
3805+ yield {}
37333806
37343807-- Map a single hex character to its integer value (0-15).
37353808-- Handles both lowercase and uppercase hex to prevent hot-bucket skew from
@@ -3857,7 +3930,7 @@ pollOnce = \token cursor clientIp -> do
38573930 startSignal <- readMessageSignal pubkey
38583931 yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, startSignal: startSignal, pendingAcks: ackResult.pendingAcks, evictedEntries: ackResult.evictedEntries}})
38593932 case setup of
3860− Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
3933+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: [], pubkey: "", pollSlotFull: False {}}
38613934 Just {value: s} -> do
38623935 -- Acquire poll slot (limit concurrent polls per key)
38633936 acquired <- atomic do
@@ -3871,7 +3944,8 @@ pollOnce = \token cursor clientIp -> do
38713944 else union activePolls [{pubkey: s.pubkey, startedAt: t}]
38723945 yield (True {})
38733946 if not acquired
3874− then yield {authorized: True {}, events: [], cursor: cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
3947+ -- Poll slot exhausted: tell client to back off for one poll cycle instead of busy-retrying
3948+ then yield {authorized: True {}, events: [], cursor: cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries, pubkey: s.pubkey, pollSlotFull: True {}}
38753949 else do
38763950 pollResult <- do
38773951 -- Phase 1: check for existing messages (atomic read)
@@ -3916,7 +3990,7 @@ pollOnce = \token cursor clientIp -> do
39163990 atomic do
39173991 activePolls <- *activePolls
39183992 set *activePolls = removeOneForKey activePolls s.pubkey
3919− yield {authorized: pollResult.authorized, events: pollResult.events, cursor: pollResult.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
3993+ yield {authorized: pollResult.authorized, events: pollResult.events, cursor: pollResult.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries, pubkey: s.pubkey, pollSlotFull: False {}}
39203994
39213995-- ============================================
39223996-- Derived Relations
@@ -4249,7 +4323,7 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
42494323 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
42504324 startPresenceVersion <- *presenceVersion
42514325 result <- case extractToken authorization of
4252− Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
4326+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: [], pubkey: "", pollSlotFull: False {}}
42534327 Just {value: token} -> pollOnce token effectiveCursor clientIp
42544328 -- Check presence for watched keys (piggybacks on poll cycle; max latency = pollTimeoutMs).
42554329 -- Skip the expensive bucket reads if presenceVersion hasn't changed since poll start.
@@ -4269,23 +4343,20 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
42694343 let isHeartbeat = result.authorized && count result.events == 0
42704344 -- Deliver message expiration notifications: atomically read and drain expirations
42714345 -- for this sender's pubkey so each notification is delivered exactly once.
4272− expirations <- if not result.authorized then yield []
4273− else case extractToken authorization of
4274− Nothing {} -> yield []
4275− Just {value: token} -> do
4276− let bucket = tokenBucket token
4277− sessions <- readSessionBucket bucket
4278− t3 <- now
4279− case authenticateSession sessions token t3 of
4280− Unauthorized {} -> yield []
4281− Authorized {pubkey} -> atomic do
4282− se <- *senderExpirations
4283− let forKey = do e <- se; where e.senderKey == pubkey; yield {messageId: e.messageId, toKey: e.toKey, expiredAt: e.expiredAt}
4284− let remaining = do e <- se; where e.senderKey != pubkey; yield e
4285− set *senderExpirations = remaining
4286− let capped = takeRelation maxSenderExpirationsPerKey forKey
4287− yield capped
4288− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0, cursorReset: cursorWasReset, messageExpirations: expirations})
4346+ -- Uses result.pubkey from pollOnce (already authenticated) instead of re-authenticating,
4347+ -- avoiding a redundant safeCompare scan and preventing a race where a concurrent poll
4348+ -- could double-drain or lose expirations between the two separate atomic blocks.
4349+ expirations <- if not result.authorized || result.pubkey == "" then yield []
4350+ else atomic do
4351+ se <- *senderExpirations
4352+ let forKey = do e <- se; where e.senderKey == result.pubkey; yield {messageId: e.messageId, toKey: e.toKey, expiredAt: e.expiredAt}
4353+ let remaining = do e <- se; where e.senderKey != result.pubkey; yield e
4354+ set *senderExpirations = remaining
4355+ let capped = takeRelation maxSenderExpirationsPerKey forKey
4356+ yield capped
4357+ -- Poll slot exhaustion: signal clients to back off for one poll cycle
4358+ let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
4359+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, messageExpirations: expirations})
42894360
42904361-- Core send logic: dedup, rate limiting, store message, create forwards.
42914362-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
@@ -4385,7 +4456,7 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs msgExpiresAt remot
43854456 yield (any fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == server))
43864457 when stillPending (do
43874458 fwdT <- now
4388− fwdAuth <- generateFederationAuth "forward"
4459+ fwdAuth <- generateFederationAuth "forward" server
43894460 fwdCorrId <- genTokenN 1
43904461 -- Compute remaining TTL from the stored message's expiration (not the original TTL).
43914462 -- This ensures the remote server gets the correct remaining lifetime, even if
@@ -4419,7 +4490,7 @@ sendBatchForwardsSequential = \msgs server ->
44194490 yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))
44204491 when stillPending (do
44214492 fwdT <- now
4422− fwdAuth <- generateFederationAuth "forward"
4493+ fwdAuth <- generateFederationAuth "forward" server
44234494 fwdCorrId <- genTokenN 1
44244495 -- Use stored msgExpiresAt to compute remaining TTL (not original messageTtl)
44254496 let fwdTtl = if r.msgExpiresAt > 0 then max 0 (r.msgExpiresAt - fwdT) else 0
@@ -4630,6 +4701,9 @@ handleRevoke = \targetToken authorization clientIp respond -> do
46304701-- This gives faster offline detection than waiting for session expiry (1 hour).
46314702handleLogout = \authorization clientIp respond -> do
46324703 t <- now
4704+ dp <- *drainPhase
4705+ if dp >= 2 then yield (respond {ok: False {}, authorized: False {}})
4706+ else do
46334707 globalRateLimited <- checkGlobalRate t
46344708 if globalRateLimited then yield (respond {ok: False {}, authorized: False {}})
46354709 else do
@@ -4774,7 +4848,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
47744848 -- Merge new peer servers with existing known peers (selected buckets may not contain all peers)
47754849 knownPeers <- *knownPeerServers
47764850 let newPeers = gm.gossipPeerServers
4777− set *knownPeerServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union acc [s]) knownPeers newPeers
4851+ set *knownPeerServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union [s] acc) knownPeers newPeers
47784852 set *presenceVersion = t
47794853 yield {gm: gm, preFiltered: count safeDeduped - count capacityDeduped, localClashCount: count localClashes}
47804854 when (gossipResult.localClashCount > 0)
@@ -4788,7 +4862,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
47884862 let active = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
47894863 -- Pre-filter active entries to only those matching (pubkey, server) pairs in this batch.
47904864 -- Reduces the inner dedup scan from O(n*m) to O(n*k) where k << m (k = matching pairs).
4791− let batchPairs = fold (\acc e -> if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server) then acc else union acc [{pubkey: e.pubkey, server: e.server}]) [] effectiveEvents
4865+ let batchPairs = fold (\acc e -> if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server) then acc else union [{pubkey: e.pubkey, server: e.server}] acc) [] effectiveEvents
47924866 let relevantActive = do r <- active; where any batchPairs (\p -> p.pubkey == r.pubkey && p.server == r.server); yield r
47934867 let fresh = do e <- effectiveEvents; where none relevantActive (\r -> r.pubkey == e.pubkey && r.server == e.server && r.timestamp == e.timestamp); yield e
47944868 let newRecent = do e <- fresh; yield {pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, seenAt: t}
@@ -4979,6 +5053,37 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
49795053 set *pendingAckCleanups = if count perKeyCapped > maxPendingAckCleanups then takeRelation maxPendingAckCleanups perKeyCapped else perKeyCapped)
49805054 yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
49815055
5056+-- POST /federation/cancel: Cancel a previously forwarded message.
5057+-- Called by the origin server when the sender retracts a message via /messages/cancel.
5058+-- Deletes the message from the recipient's mailbox if it hasn't been polled yet.
5059+-- If the message was already polled, returns cancelled=False (no-op).
5060+handleFederationCancel = \messageId toKey fromServer requestTimestamp correlationId authorization respond -> do
5061+ t <- now
5062+ let corrId = sanitizeCorrelationId correlationId
5063+ when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel " ++ messageId ++ " from " ++ fromServer))
5064+ let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}}
5065+ withFederationAuth authorization t cancelRespond fromServer "cancel" requestTimestamp (not (isLocalServer fromServer)) (False {}) (do
5066+ result <- atomic do
5067+ let b = pubkeyBucket toKey
5068+ bucket <- readMessageBucket b
5069+ -- Only cancel messages that originated from the requesting server
5070+ let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == toKey && m.originServer == fromServer; yield m)
5071+ case toCancel of
5072+ Nothing {} -> yield {cancelled: False {}}
5073+ Just {} -> do
5074+ writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == toKey); yield m)
5075+ tc <- *totalMessageCount
5076+ set *totalMessageCount = max 0 (tc - 1)
5077+ rc <- readRcBucket b
5078+ writeRcBucket b (adjustRecipientCount rc toKey (0 - 1))
5079+ -- Also cancel pending forwards for this message
5080+ fwdBkt <- readFwdBucket b
5081+ writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == toKey); yield f)
5082+ yield {cancelled: True {}}
5083+ when result.cancelled
5084+ (logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")"))
5085+ yield (respond {ok: True {}, error: "", cancelled: result.cancelled}))
5086+
49825087-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
49835088-- Call before stopping the server so peers learn keys went offline immediately.
49845089-- Gossip is sent via fork; allow a few seconds after response before process exit.
@@ -5120,26 +5225,10 @@ handlePrune = \authorization respond -> do
51205225 when authorized (fork (pruneAndRetry t))
51215226 yield (respond {ok: authorized})
51225227
5123−handleHealth = \authorization respond -> do
5124− t <- now
5125− -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
5126− -- but have their own dedicated rate limiter to prevent DDoS on /health from saturating reads.
5127− healthLimited <- checkHealthRate t
5128− if healthLimited then do
5129− -- Return cached last-known health (prevents DDoS on /health from masking real failures)
5130− cached <- atomic do
5131− cOk <- *cachedHealthOk
5132− cP <- *cachedHealthPrune
5133− cH <- *cachedHealthHeartbeat
5134− cC <- *cachedHealthCapacity
5135− cF <- *cachedHealthFederation
5136− cS <- *cachedHealthSnapshot
5137− cW <- *cachedHealthWatchdog
5138− dp <- *drainPhase
5139− yield {ok: cOk && dp == 0, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, draining: dp > 0}
5140− yield (respond cached)
5141− else do
5142− let isAuthed = verifyMaintenanceAuth authorization
5228+-- Compute health state and update the cached health values.
5229+-- Called from both handleHealth (request-driven) and pruneAndRetry (background-driven)
5230+-- so the cache stays fresh even when /health is rate-limited under sustained monitoring load.
5231+computeAndCacheHealth = \t -> do
51435232 result <- atomic do
51445233 lastPrune <- *lastPrunedAt
51455234 lastHb <- *lastHeartbeatAt
@@ -5153,30 +5242,22 @@ handleHealth = \authorization respond -> do
51535242 lastFedSuccess <- *lastFederationSuccessAt
51545243 forwardFailures <- *forwardFailures
51555244 pruneRun <- *pruneRunning
5156− -- lastSnapshotExportAt no longer used in health check (replaced by prunesSinceSnapshot)
51575245 dp <- *drainPhase
51585246 let warmup = t - sa < healthPruneMaxAge
51595247 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
5160− -- Capacity pressure: flag when any capped collection exceeds 90% full
51615248 let capacityOk = tmc * 10 <= maxMessages * 9
51625249 && presenceCount * 10 <= maxPresenceEntries * 9
51635250 && forwardCount * 10 <= maxForwardsEntries * 9
51645251 && sessionCount * 10 <= maxSessions * 9
51655252 && count activePolls * 10 <= maxActivePolls * 9
5166− -- Federation: healthy if disabled, in warmup, no peer issues, or recent success
51675253 let fedOk = not federationEnabled || warmup
51685254 || count forwardFailures == 0
51695255 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
5170− -- Snapshot staleness: unhealthy if maintenance secret is configured but no snapshot
5171− -- has been exported within the staleness window (federation state will be lost on restart)
5172− -- Snapshot staleness: uses prune-cycle counter (clock-jump resilient).
5173− -- Stale when prunesSinceSnapshot exceeds threshold computed from staleness/prune interval.
51745256 pss <- *prunesSinceSnapshot
51755257 let staleThresholdCycles = if backgroundPruneInterval > 0 then snapshotStalenessWarning / backgroundPruneInterval else 0
51765258 let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup
51775259 || staleThresholdCycles == 0 || pss <= staleThresholdCycles
51785260 || (tmc == 0 && presenceCount == 0 && forwardCount == 0)
5179− -- Watchdog: healthy if recently bumped or in warmup (watchdog starts after 10x prune interval)
51805261 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
51815262 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
51825263 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
@@ -5185,9 +5266,7 @@ handleHealth = \authorization respond -> do
51855266 snapshotOk: snapOk,
51865267 watchdogOk: wdOk,
51875268 isDraining: dp > 0}
5188− -- Draining servers report ok=False so load balancers stop routing traffic to them
51895269 let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
5190− -- Cache health result for rate-limited responses
51915270 atomic do
51925271 set *cachedHealthOk = allOk
51935272 set *cachedHealthPrune = result.pruneOk
@@ -5196,6 +5275,30 @@ handleHealth = \authorization respond -> do
51965275 set *cachedHealthFederation = result.federationOk
51975276 set *cachedHealthSnapshot = result.snapshotOk
51985277 set *cachedHealthWatchdog = result.watchdogOk
5278+ yield result
5279+
5280+handleHealth = \authorization respond -> do
5281+ t <- now
5282+ -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
5283+ -- but have their own dedicated rate limiter to prevent DDoS on /health from saturating reads.
5284+ healthLimited <- checkHealthRate t
5285+ if healthLimited then do
5286+ -- Return cached last-known health (prevents DDoS on /health from masking real failures).
5287+ -- Cache is kept fresh by computeAndCacheHealth called from both here and background prune.
5288+ cached <- atomic do
5289+ cOk <- *cachedHealthOk
5290+ cP <- *cachedHealthPrune
5291+ cH <- *cachedHealthHeartbeat
5292+ cC <- *cachedHealthCapacity
5293+ cF <- *cachedHealthFederation
5294+ cS <- *cachedHealthSnapshot
5295+ cW <- *cachedHealthWatchdog
5296+ dp <- *drainPhase
5297+ yield {ok: cOk && dp == 0, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, draining: dp > 0}
5298+ yield (respond cached)
5299+ else do
5300+ let isAuthed = verifyMaintenanceAuth authorization
5301+ result <- computeAndCacheHealth t
51995302 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
52005303 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
52015304 -- Draining status is always visible (load balancers need it for routing decisions).
@@ -5491,12 +5594,12 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
54915594 writeAllNonces cappedNonces
54925595 -- Restore blocked peers (union with existing, cap at maxBlockedPeers)
54935596 blocked <- *blockedPeers
5494− let mergedBlocked = fold (\acc b -> if any acc (\x -> x == b) then acc else union acc [b]) blocked blockedPeersData
5597+ let mergedBlocked = fold (\acc b -> if any acc (\x -> x == b) then acc else union [b] acc) blocked blockedPeersData
54955598 let cappedBlocked = takeRelation maxBlockedPeers mergedBlocked
54965599 set *blockedPeers = cappedBlocked
54975600 -- Restore allowed peers (union with existing, cap at maxAllowedPeers)
54985601 allowed <- *allowedPeers
5499− let mergedAllowed = fold (\acc a -> if any acc (\x -> x == a) then acc else union acc [a]) allowed allowedPeersData
5602+ let mergedAllowed = fold (\acc a -> if any acc (\x -> x == a) then acc else union [a] acc) allowed allowedPeersData
55005603 let cappedAllowed = takeRelation maxAllowedPeers mergedAllowed
55015604 set *allowedPeers = cappedAllowed
55025605 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
@@ -5672,28 +5775,41 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
56725775 bucket <- readMessageBucket b
56735776 let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == to; yield m)
56745777 case toCancel of
5675− Nothing {} -> yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5778+ Nothing {} -> yield {cancelled: False {}, cancelServers: []}
56765779 Just {value: msg} ->
56775780 -- Only allow cancelling locally-originated messages (originServer == "")
56785781 -- where the caller is the actual sender. Federation forwards have no
56795782 -- senderKey and should not be cancellable via this endpoint.
5680− if msg.originServer != "" then yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5681− else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5783+ if msg.originServer != "" then yield {cancelled: False {}, cancelServers: []}
5784+ else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, cancelServers: []}
56825785 else do
56835786 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)
56845787 tc <- *totalMessageCount
56855788 set *totalMessageCount = max 0 (tc - 1)
56865789 rc <- readRcBucket b
56875790 writeRcBucket b (adjustRecipientCount rc to (0 - 1))
5688− -- Cancel pending forwards
5791+ -- Cancel pending forwards and collect servers that already received the message
56895792 fwdBkt <- readFwdBucket b
5793+ let cancelledFwds = do f <- fwdBkt; where f.messageId == messageId && f.toKey == to; yield f
56905794 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == to); yield f)
5691− -- Only locally-originated messages reach this point (originServer == ""),
5692− -- so no federation acks are needed.
5693− yield {cancelled: True {}, pendingAcks: [], evictedEntries: []}
5694− fork (when (federationEnabled && count cancelResult.pendingAcks > 0) (do
5695− sendAcksAndClean cancelResult.pendingAcks
5696− retryEvictedAcks cancelResult.evictedEntries cancelResult.pendingAcks))
5795+ -- Collect servers from recipient's presence that may have received the forward
5796+ -- (forwards that completed are no longer in fwdBucket, so check presence)
5797+ presenceBkt <- readPresenceBucket b
5798+ let remotePresence = collectPeerServers (presenceForKey presenceBkt to t) (\p -> p.server)
5799+ yield {cancelled: True {}, cancelServers: remotePresence}
5800+ -- Send federation cancel to remote servers that may have the message
5801+ fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)
5802+ (sendToEach cancelResult.cancelServers (\server -> do
5803+ cancelT <- now
5804+ cancelAuth <- generateFederationAuth "cancel" server
5805+ cancelCorrId <- genTokenN 1
5806+ cancelFedResult <- fetchFederation server (FederationCancel {
5807+ messageId: messageId, toKey: to, fromServer: serverHostN,
5808+ requestTimestamp: cancelT, correlationId: cancelCorrId,
5809+ authorization: cancelAuth})
5810+ case cancelFedResult of
5811+ Err e -> logWarn ("federation: cancel to " ++ server ++ " failed for " ++ messageId ++ ": " ++ show e)
5812+ Ok r -> when (not r.ok) (logWarn ("federation: cancel to " ++ server ++ " rejected for " ++ messageId ++ ": " ++ r.error)))))
56975813 yield (respond {ok: True {}, authorized: True {}, cancelled: cancelResult.cancelled, error: ""}))
56985814
56995815-- POST /messages/batch: Send multiple messages in one request.
@@ -5858,7 +5974,7 @@ handlePingPeer = \server authorization respond -> do
58585974 if not authorized then yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"})
58595975 else do
58605976 startTime <- now
5861− auth <- generateFederationAuth "gossip"
5977+ auth <- generateFederationAuth "gossip" server
58625978 pingCorrId <- genTokenN 1
58635979 result <- fetchFederation server (RecvGossip {
58645980 authorization: auth,
@@ -6028,6 +6144,7 @@ requestPath = \req -> case req of
60286144 RecvGossip {} -> "POST /federation/gossip"
60296145 ForwardMessage {} -> "POST /federation/forward"
60306146 FederationAck {} -> "POST /federation/ack"
6147+ FederationCancel {} -> "POST /federation/cancel"
60316148 VerifyToken {} -> "POST /federation/verify"
60326149 Prune {} -> "POST /maintenance/prune"
60336150 Drain {} -> "POST /maintenance/drain"
@@ -6075,6 +6192,7 @@ requestClientIp = \req -> case req of
60756192 RecvGossip {} -> ""
60766193 ForwardMessage {} -> ""
60776194 FederationAck {} -> ""
6195+ FederationCancel {} -> ""
60786196 Prune {} -> ""
60796197 Drain {} -> ""
60806198 Shutdown {} -> ""
@@ -6123,6 +6241,7 @@ serveInner = \req -> case req of
61236241 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
61246242 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, messageTtl, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId messageTtl authorization respond
61256243 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
6244+ FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationCancel messageId toKey fromServer requestTimestamp correlationId authorization respond
61266245 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
61276246 Prune {authorization, respond} -> handlePrune authorization respond
61286247 Drain {phase, authorization, respond} -> handleDrain phase authorization respond
@@ -6474,6 +6593,8 @@ main = do
64746593 logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
64756594 when federationEnabled
64766595 (logWarn "SECURITY: DNS rebinding risk — federation URLs resolve DNS at fetch time, not at validation time. A peer hostname passing isBadServerName could rebind to a private IP between validation and fetch. Mitigate by blocking outbound connections to RFC1918/link-local/CGNAT ranges at the firewall level.")
6596+ when (count presharedKeyList > 0)
6597+ (logInfo ("pre-shared federation keys configured for " ++ show (count presharedKeyList) ++ " peer(s) — callback verification bypassed for those peers"))
64776598 logInfo ("Skrepka relay server starting on port " ++ show listenPort ++ "...")
64786599 logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")
64796600 logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")