@@ -1,6 +1,7 @@
11-- Skrepka Relay Server (maximally simplified)
22-- Single-server implementation of the Skrepka protocol (v0.1).
3−-- Every collection is a flat list; rate limiters are plain count-per-window.
3+-- Every collection is a flat list; rate limiting is delegated to the
4+-- compiler's per-endpoint `rateLimit` clauses on each route.
45-- Token comparison is plain `==` (timing-attack vulnerable), retry backoff is linear,
56-- and many helpers are inlined. Performance suffers; clarity is the only goal.
67
@@ -8,7 +9,7 @@
89-- Configuration
910-- ============================================
1011-- Top-level constants: bind address, federation identity, capacity ceilings,
11−-- per-key/window rate limits, and expiry/backoff/TTL durations.
12+-- per-endpoint rate-limit budgets, and expiry/backoff/TTL durations.
1213-- Values written here are defaults; every scalar (Int/Float/Text/Bool/Maybe ...)
1314-- top-level constant can be overridden at startup via `--name=value` on the
1415-- command line (see Knot's `knot_override_lookup`). Run with `--help` for the
@@ -81,36 +82,20 @@ maxSessionsPerKey = 10
8182-- Validity window for an auth challenge before it must be re-issued (1 min).
8283challengeExpiry : Int<Ms>
8384challengeExpiry = 60000
84−-- Challenge requests permitted per key inside its rate window.
85+-- Challenge requests permitted per pubkey inside its rate window.
86+-- Wired into the /auth/challenge route's `rateLimit` clause.
8587maxChallengeRatePerKey : Int
8688maxChallengeRatePerKey = 5
87−-- Window over which the per-key challenge rate is measured (1 s).
89+-- Window over which the per-pubkey challenge rate is measured (1 s).
8890challengeRateWindowMs : Int<Ms>
8991challengeRateWindowMs = 1000
90−-- Send-message requests permitted per key per window.
92+-- Send-message requests permitted per session token per window.
93+-- Wired into the /messages route's `rateLimit` clause.
9194maxSendRatePerKey : Int
9295maxSendRatePerKey = 20
93−-- Window for the per-key send rate limit (1 s).
96+-- Window for the per-session send rate limit (1 s).
9497sendRateWindowMs : Int<Ms>
9598sendRateWindowMs = 1000
96−-- Requests permitted per source IP per window (covers unauthenticated traffic).
97−maxIpRatePerIp : Int
98−maxIpRatePerIp = 10
99−-- Window for the per-IP rate limit (1 s).
100−ipRateWindowMs : Int<Ms>
101−ipRateWindowMs = 1000
102−-- Global ceiling on total request rate across all clients.
103−maxGlobalRequestRate : Int
104−maxGlobalRequestRate = 1000
105−-- Window for the global request rate limit (1 s).
106−globalRateWindowMs : Int<Ms>
107−globalRateWindowMs = 1000
108−-- Global ceiling on auth-endpoint requests (challenge + verify combined).
109−maxAuthGlobalRate : Int
110−maxAuthGlobalRate = 200
111−-- Window for the global auth rate limit (1 s).
112−authGlobalRateWindowMs : Int<Ms>
113−authGlobalRateWindowMs = 1000
11499
115100-- How long a session token remains valid after issue (1 hour).
116101sessionExpiry : Int<Ms>
@@ -175,10 +160,10 @@ impl FromJSON GossipEventType where
175160
176161-- /messages is all-or-nothing per call: either every message in the batch is
177162-- enqueued (delivered, federated, or buffered for federation), or none are
178−-- and the call returns a non-2xx HTTP status. Per-message rejection codes
179−-- ("self_send", "rate_limited") surface as the HTTP error body. Rate
180−-- counters are still charged for every attempted message even on rejection,
181−-- so a sender can't probe limits cheaply.
163+-- and the call returns a non-2xx HTTP status. The per-message rejection code
164+-- "self_send" surfaces in the HTTP error body. Per-session rate limiting is
165+-- enforced by the route's `rateLimit` clause, which short-circuits with 429
166+-- before the handler runs.
182167-- The sender's UI treats every success the same and learns "they got it"
183168-- only via the recipient's E2E ack.
184169
@@ -224,11 +209,6 @@ type PresenceEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Timestam
224209type ForwardEntry = {toServer: ServerName, toKey: PubkeyHex, encryptedBlob: BlobHex, retries: Int}
225210-- Running failure counter for a peer; keyed by `server`. Drives `retryBackoff`.
226211type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFailedAt: Timestamp}
227−-- One fixed-window rate counter. `key` identifies the subject (pubkey, ip,
228−-- server, or composite "sender|recipient" string). `expiresAt` is the wall
229−-- time at which the current window ends; once `t >= expiresAt` the entry is
230−-- recycled (counter resets to 1).
231−type RateEntry = {key: Text, expiresAt: Timestamp, count: Int}
232212
233213-- ============================================
234214-- Persisted State (all flat lists, no sharding)
@@ -236,7 +216,9 @@ type RateEntry = {key: Text, expiresAt: Timestamp, count: Int}
236216-- Mutable refs (`*x`) holding the entire server state in memory. Every list
237217-- scan is O(n); scaling is the first thing the production version replaces.
238218-- Grouped here: auth (challenges/sessions), mailbox (messages/forwards),
239−-- presence, rate-limit windows, and bookkeeping counters.
219+-- presence, and bookkeeping counters. Per-endpoint rate-limit buckets live
220+-- in the runtime's hidden `_knot_rate_limits` SQLite table and don't appear
221+-- here.
240222-- The Knot runtime persists every `*ref` declared here through restarts (see
241223-- the project memory note about SQLite-backed refs); nothing in this file is
242224-- truly volatile despite being typed as plain in-memory lists.
@@ -254,20 +236,6 @@ type RateEntry = {key: Text, expiresAt: Timestamp, count: Int}
254236-- Per-peer failure counters; drives `retryBackoff`.
255237*forwardFailures : [ForwardFailure]
256238
257−-- Per-key fixed-window counters for /messages send-side rate limiting.
258−*sendRates : [RateEntry]
259−-- Per-key /auth/challenge counters (separate from `*challenges` capacity gate).
260−*challengeRates : [RateEntry]
261−-- Per-source-IP counters; gate for unauthenticated traffic.
262−*ipRates : [RateEntry]
263−
264−-- Global request-rate gate (single counter shared across all clients).
265−*globalRateCount : Int
266−*globalRateWindowStart : Timestamp
267−-- Global gate charged only by /auth/* endpoints, on top of `*globalRateCount`.
268−*authGlobalRateCount : Int
269−*authGlobalRateWindowStart : Timestamp
270−
271239-- Strictly-monotonic sequence used as `Message.receivedAt`; guarantees poll-cursor ordering.
272240*seqCounter : Timestamp
273241-- Bumped by send/heartbeat to wake all blocked /poll holders.
@@ -284,13 +252,32 @@ type RateEntry = {key: Text, expiresAt: Timestamp, count: Int}
284252-- declared body shape, `Err` produces a non-2xx response whose body is
285253-- `{"error": "<code>"}`.
286254
255+-- Per-pubkey token bucket on /auth/challenge: prevents one key from
256+-- exhausting the challenge pool by repeated nonce requests.
257+-- (`export` is a workaround for a compiler bug where the unused-name pass
258+-- doesn't walk route `rateLimit` clauses.)
259+export challengeRateLimit = {
260+ key: \input ctx -> Just {value: input.pubkey},
261+ limit: {requests: maxChallengeRatePerKey, window: challengeRateWindowMs}
262+}
263+
264+-- Per-session token bucket on /messages, keyed by the Bearer token. A pubkey
265+-- is capped at `maxSessionsPerKey` live sessions, so the effective per-pubkey
266+-- send budget is bounded as well.
267+export sendRateLimit = {
268+ key: \input ctx -> Just {value: input.authorization},
269+ limit: {requests: maxSendRatePerKey, window: sendRateWindowMs}
270+}
271+
287272route AuthApi where
288− POST {pubkey: PubkeyHex} /auth/challenge headers {xForwardedFor: Maybe Text} -> {challenge: Text, expiresAt: Timestamp} = ReqChallenge
273+ POST {pubkey: PubkeyHex} /auth/challenge headers {xForwardedFor: Maybe Text} -> {challenge: Text, expiresAt: Timestamp}
274+ rateLimit challengeRateLimit = ReqChallenge
289275 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /auth/verify headers {xForwardedFor: Maybe Text} -> {token: Text, expiresAt: Timestamp} = VerifyAuth
290276
291277route ClientApi where
292278 POST {cursor: Timestamp} /poll headers {authorization: Text, xForwardedFor: Maybe Text} -> {events: [PollEvent], cursor: Timestamp} = Poll
293− POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, xForwardedFor: Maybe Text} -> {} = SendMessages
279+ POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, xForwardedFor: Maybe Text} -> {}
280+ rateLimit sendRateLimit = SendMessages
294281
295282route FederationApi where
296283 POST {events: [GossipEventType], fromServer: ServerName} /federation/gossip -> {} = RecvGossip
@@ -323,15 +310,6 @@ capList = \items maxN -> drop (max 0 (count items - maxN)) (sortBy (\x -> x.expi
323310capPerKey : [{pubkey: PubkeyHex, expiresAt: Timestamp | r}] -> PubkeyHex -> Int -> [{pubkey: PubkeyHex, expiresAt: Timestamp | r}]
324311capPerKey = \items pubkey maxN -> union (filter (\x -> not (x.pubkey == pubkey)) items) (capList (filter (\x -> x.pubkey == pubkey) items) (maxN - 1))
325312
326−-- Like forEach, but collects each action's result into a list.
327−mapEach : [a] -> (a -> IO {} b) -> IO {} [b]
328−mapEach = \items action -> case items of
329− [] -> yield []
330− Cons x rest -> do
331− r <- action x
332− rs <- mapEach rest action
333− yield (union [r] rs)
334−
335313-- Hex tokens: one random hex char per recursive call. `n` is the desired
336314-- length in characters (so n=48 yields 24 random bytes of entropy).
337315genTokenN : Int -> IO {random} Text
@@ -403,8 +381,9 @@ isBadServerName = \sn -> sn == "" || sn == "localhost" || not all (\c -> contain
403381-- ============================================
404382-- Auth helpers
405383-- ============================================
406−-- Bearer token extraction, session lookup, and the `gateAuth` combinator that
407−-- every client handler funnels through (global rate → session).
384+-- Bearer token extraction and session lookup. Authenticated handlers funnel
385+-- through `withSessionAuth`; per-endpoint rate limiting is enforced by the
386+-- compiler-generated `rateLimit` clauses on each route.
408387-- Federation requests are unauthenticated: anyone can hit /federation/*.
409388-- The `no_presence` check bounds federated forwards to currently-online keys.
410389
@@ -413,8 +392,7 @@ matchIp : Text -> Text -> Bool
413392matchIp = \clientIp sessionIp -> sessionIp == "" || clientIp == sessionIp
414393
415394-- Unwrap an X-Forwarded-For header value to a plain IP string. Empty when the
416−-- header is absent (no reverse proxy in front), which downstream gates treat
417−-- as "skip IP-based rate limiting".
395+-- header is absent (no reverse proxy in front).
418396clientIpOf : Maybe Text -> Text
419397clientIpOf = \h -> case h of
420398 Just {value: ip} -> ip
@@ -427,16 +405,6 @@ extractToken = \h -> if hasPrefix "Bearer " h && drop 7 h != ""
427405then Just {value: drop 7 h}
428406else Nothing {}
429407
430−-- Common gate for handlers: global rate, session auth. Short-circuits with
431−-- an `Err` response on failure; the `onAuth` callback returns its own
432−-- `Result HttpError T` for the success path.
433−gateAuth : Text -> Text -> Timestamp -> (PubkeyHex -> IO {} (Result HttpError t)) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions} (Result HttpError t)
434−gateAuth = \auth ip t onAuth -> do
435− g <- checkGlobalRate t
436− if g
437− then yield (Err {error: {status: 429, message: "rate_limited"}})
438− else withSessionAuth auth t ip (yield (Err {error: {status: 401, message: "unauthorized"}})) onAuth
439−
440408-- Resolve a Bearer token to its session pubkey if the session is unexpired and
441409-- the client IP matches. Runs `onAuth pubkey` on success, `onUnauth` otherwise.
442410withSessionAuth : Text -> Int<Ms> -> Text -> IO e a -> (PubkeyHex -> IO e a) -> IO {r *sessions | e} a
@@ -448,83 +416,6 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
448416 Just {value: s} -> onAuth s.pubkey
449417 _ -> onUnauth
450418
451−-- ============================================
452−-- Rate limiting (plain count-per-window, no sliding interpolation)
453−-- ============================================
454−-- Two flavors of fixed-window counters:
455−-- * Per-key (`checkRate`) — keyed by pubkey, ip, server, or pair string;
456−-- entries live in a `RateEntry` list ref capped at 5000 with LRU eviction.
457−-- * Global (`checkGlobalRate`/`checkAuthGlobalRate`) — single counter+windowStart
458−-- pair, used for the request-wide and auth-wide ceilings.
459−-- `authPreCheck` composes global + ip + auth-global into a one-shot
460−-- gate for unauthenticated handlers (challenge/verify).
461−
462−-- Returns {limited, entries}. Increments count when the key fits within the
463−-- window; otherwise resets to a fresh window expiring at `t + windowMs`. The
464−-- returned `entries` list is what the caller must write back to the
465−-- corresponding ref; it has been capped at 5000 rows (LRU by `expiresAt`) to
466−-- bound memory.
467−checkRate : [RateEntry] -> Text -> Int -> Int<Ms> -> Timestamp -> {limited: Bool, entries: [RateEntry]}
468−checkRate = \entries key maxRate windowMs t -> (\cur -> if cur.count > maxRate then {limited: True {}, entries} else {limited: False {}, entries: capList (union (filter (\x -> not (x.key == key)) entries) [{key, cur.expiresAt, cur.count}]) 5000}) (case findFirst entries (\r -> r.key == key && r.expiresAt > t) of
469− Just {value: r} -> {r.expiresAt, count: r.count + 1}
470− _ -> {expiresAt: t + windowMs, count: 1})
471−
472−-- Single-counter global rate gate. Resets the counter when the window expires;
473−-- returns True when the limit has been hit (caller responds rate_limited).
474−checkGlobalRate : Timestamp -> IO {rw *globalRateCount, rw *globalRateWindowStart} Bool
475−checkGlobalRate = \t -> atomic (do
476− ws <- *globalRateWindowStart
477− c <- *globalRateCount
478− if t - ws >= globalRateWindowMs
479− then do
480− *globalRateWindowStart = t
481− *globalRateCount = 1
482− yield (False {})
483− else if c >= maxGlobalRequestRate
484− then yield (True {})
485− else do
486− *globalRateCount = c + 1
487− yield (False {}))
488−
489−-- Same shape as `checkGlobalRate` but charged only by the auth endpoints
490−-- (challenge + verify). Stops a single attacker from exhausting the global
491−-- counter via auth churn.
492−checkAuthGlobalRate = \t -> atomic (do
493− ws <- *authGlobalRateWindowStart
494− c <- *authGlobalRateCount
495− if t - ws >= authGlobalRateWindowMs
496− then do
497− *authGlobalRateWindowStart = t
498− *authGlobalRateCount = 1
499− yield (False {})
500− else if c >= maxAuthGlobalRate
501− then yield (True {})
502− else do
503− *authGlobalRateCount = c + 1
504− yield (False {}))
505−
506−-- Per-source-IP rate gate. When the IP is unknown (no X-Forwarded-For from a
507−-- reverse proxy), the gate is skipped — direct deployments without a proxy
508−-- still rely on global rate limits.
509−checkIpRateGate : Text -> Int<Ms> -> IO {rw *ipRates} Bool
510−checkIpRateGate = \ip t -> if ip == ""
511−then yield (False {})
512−else atomic (do
513− rates <- *ipRates
514− let r = checkRate rates ip maxIpRatePerIp ipRateWindowMs t
515− *ipRates = r.entries
516− yield r.limited)
517−
518−-- Pre-auth gate: global + ip + auth-global. Returns "" when ok, else an error code.
519−authPreCheck : Text -> Int<Ms> -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates} Text
520−authPreCheck = \clientIp t -> do
521− g <- checkGlobalRate t
522− i <- checkIpRateGate clientIp t
523− a <- checkAuthGlobalRate t
524− yield (if g
525− then "global_rate_limited"
526− else if i then "ip_rate_limited" else if a then "auth_rate_limited" else "")
527−
528419-- ============================================
529420-- Federation failure tracking (simple, no circuit breaker)
530421-- ============================================
@@ -714,78 +605,65 @@ pruneAndRetry = \t -> do
714605-- revoke/logout.
715606
716607-- POST /auth/challenge: issue a fresh signed-nonce challenge for `pubkey`.
717−-- Capped per-key (maxChallengesPerKey) and rate-limited; the IP is recorded
718−-- so /auth/verify can require the same client.
719−handleChallenge : PubkeyHex -> Text -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *challengeRates, rw *challenges, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates, clock, random} (Result HttpError {challenge: Text, expiresAt: Int})
608+-- Capped per-key (maxChallengesPerKey); per-pubkey rate limiting is enforced
609+-- by the route's `rateLimit` clause. The IP is recorded so /auth/verify can
610+-- require the same client.
611+handleChallenge : PubkeyHex -> Text -> IO {rw *challenges, clock, random} (Result HttpError {challenge: Text, expiresAt: Int})
720612handleChallenge = \pubkey clientIp -> do
721613 t <- now
722− preErr <- authPreCheck clientIp t
723− if preErr == ""
724− then do
725− challenge <- genToken
726− result <- atomic (do
727− ch <- *challenges
728− let cappedActive = capPerKey (filter (\c -> c.expiresAt > t) ch) pubkey maxChallengesPerKey
729− if not (count cappedActive >= maxChallenges)
730− then do
731− chRates <- *challengeRates
732− let chR = checkRate chRates pubkey maxChallengeRatePerKey challengeRateWindowMs t
733− *challengeRates = chR.entries
734− if not chR.limited
735− then do
736− *challenges = union cappedActive [
737− {pubkey, challenge, expiresAt: t + challengeExpiry, ip: clientIp}
738− ]
739− yield ""
740− else yield "rate_limited"
741− else yield "capacity")
742− if result == ""
743− then yield (Ok {value: {challenge, expiresAt: t + challengeExpiry}})
744− else yield (Err {error: {status: if result == "capacity" then 503 else 429, message: result}})
745− else yield (Err {error: {status: 429, message: preErr}})
614+ challenge <- genToken
615+ full <- atomic (do
616+ ch <- *challenges
617+ let cappedActive = capPerKey (filter (\c -> c.expiresAt > t) ch) pubkey maxChallengesPerKey
618+ if count cappedActive >= maxChallenges
619+ then yield (True {})
620+ else do
621+ *challenges = union cappedActive [
622+ {pubkey, challenge, expiresAt: t + challengeExpiry, ip: clientIp}
623+ ]
624+ yield (False {}))
625+ if full
626+ then yield (Err {error: {status: 503, message: "capacity"}})
627+ else yield (Ok {value: {challenge, expiresAt: t + challengeExpiry}})
746628
747629-- POST /auth/verify: validate `signature` over `challenge` for `pubkey`.
748630-- On success, mints a session token, records local presence (`pubkey` is now
749631-- live on this server), and gossips an `online` event to known peers. If
750632-- `revokeOthers` is set, all other sessions for the same pubkey are dropped.
751−handleVerifyAuth : PubkeyHex -> Text -> Text -> Bool -> Text -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *challenges, rw *forwardFailures, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates, rw *presence, rw *sessions, network, clock, random} (Result HttpError {expiresAt: Int, token: Text})
633+handleVerifyAuth : PubkeyHex -> Text -> Text -> Bool -> Text -> IO {rw *challenges, rw *forwardFailures, rw *presence, rw *sessions, network, clock, random} (Result HttpError {expiresAt: Int, token: Text})
752634handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
753635 t <- now
754− preErr <- authPreCheck clientIp t
755− if preErr != ""
756− then yield (Err {error: {status: 429, message: preErr}})
757− else do
758− let expiry = t + sessionExpiry
759− token <- genToken
760− result <- atomic (do
761− ch <- *challenges
762− let same = \c -> c.pubkey == pubkey && c.challenge == challenge
763− let sigValid = case bytesFromHex pubkey of
764− Just {value: pk} -> case bytesFromHex signature of
765− Just {value: sigBytes} -> verify pk (textToBytes challenge) sigBytes
766− _ -> False {}
636+ let expiry = t + sessionExpiry
637+ token <- genToken
638+ result <- atomic (do
639+ ch <- *challenges
640+ let same = \c -> c.pubkey == pubkey && c.challenge == challenge
641+ let sigValid = case bytesFromHex pubkey of
642+ Just {value: pk} -> case bytesFromHex signature of
643+ Just {value: sigBytes} -> verify pk (textToBytes challenge) sigBytes
767644 _ -> False {}
768− let valid = any (\c -> same c && c.expiresAt > t && matchIp clientIp c.ip) ch && sigValid
769− if not valid
770− then yield {valid: False {}, peers: []}
771− else do
772− *challenges = filter (\c -> not same c) ch
773− ses <- *sessions
774− let pruned = if revokeOthers
775− then filter (\s -> not (s.pubkey == pubkey)) ses
776− else capPerKey ses pubkey maxSessionsPerKey
777− let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp}
778− *sessions = capList (union pruned [newSession]) maxSessions
779− pres <- *presence
780− let prunedPres = filter (\p -> not (p.pubkey == pubkey && isLocalServer p.server)) pres
781− let newPres = union prunedPres [{pubkey, server: serverHost, expiresAt: expiry}]
782− *presence = newPres
783− yield {valid: True {}, peers: knownPeerServersFromPresence newPres})
784− if not result.valid
785− then yield (Err {error: {status: 401, message: "invalid"}})
645+ _ -> False {}
646+ let valid = any (\c -> same c && c.expiresAt > t && matchIp clientIp c.ip) ch && sigValid
647+ if not valid
648+ then yield {valid: False {}, peers: []}
786649 else do
787− when federationEnabled (fork (sendBatchedGossip result.peers [GossipOnline {pubkey}]))
788− yield (Ok {value: {token, expiresAt: expiry}})
650+ *challenges = filter (\c -> not same c) ch
651+ ses <- *sessions
652+ let pruned = if revokeOthers
653+ then filter (\s -> not (s.pubkey == pubkey)) ses
654+ else capPerKey ses pubkey maxSessionsPerKey
655+ let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp}
656+ *sessions = capList (union pruned [newSession]) maxSessions
657+ pres <- *presence
658+ let prunedPres = filter (\p -> not (p.pubkey == pubkey && isLocalServer p.server)) pres
659+ let newPres = union prunedPres [{pubkey, server: serverHost, expiresAt: expiry}]
660+ *presence = newPres
661+ yield {valid: True {}, peers: knownPeerServersFromPresence newPres})
662+ if not result.valid
663+ then yield (Err {error: {status: 401, message: "invalid"}})
664+ else do
665+ when federationEnabled (fork (sendBatchedGossip result.peers [GossipOnline {pubkey}]))
666+ yield (Ok {value: {token, expiresAt: expiry}})
789667
790668-- ============================================
791669-- Handler: poll
@@ -795,15 +673,12 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
795673-- * Wait on `pollWakeup` (bumped by send/heartbeat) and return new events
796674-- plus piggybacked sender-side expirations/deliveries.
797675
798−-- POST /poll: top-level dispatch. Validates the global rate gate, then hands
799−-- off to `pollOnce` for the actual long-poll body.
800−handlePoll : Int -> Text -> Text -> IO {r *pollWakeup, r *sessions, rw *forwardFailures, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, rw *presence, network, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
676+-- POST /poll: top-level dispatch. Hands off to `pollOnce` for the actual
677+-- long-poll body.
678+handlePoll : Int -> Text -> Text -> IO {r *pollWakeup, r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, network, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
801679handlePoll = \cursor authorization clientIp -> do
802680 t <- now
803− g <- checkGlobalRate t
804− if g
805− then yield (Err {error: {status: 429, message: "rate_limited"}})
806− else pollOnce authorization cursor clientIp t
681+ pollOnce authorization cursor clientIp t
807682
808683-- Body of a single poll. Treats the cursor advance as an implicit ack
809684-- (`ackPolledMessages`), refreshes the caller's local presence row so peers
@@ -882,66 +757,39 @@ ackPolledMessages = \pubkey cursor -> do
882757-- Accept a batch of messages for recipient pubkeys. Placement (live local
883758-- session, federated forward, or held-for-presence) is internal; the call
884759-- either succeeds (every message enqueued) or fails with a non-2xx HTTP
885−-- status carrying the rejection code ("rate_limited" | "self_send" | ...).
760+-- status carrying the rejection code ("self_send" | ...).
886761-- Duplicates are silently treated as success so retries are idempotent.
887−-- Send is gated by global rate and the per-sender rate limit charged inside
888−-- `checkSendRate`.
762+-- Per-session send rate is enforced by the route's `rateLimit` clause.
889763-- Federated forwards are kicked off in background forks after the atomic.
890764
891−-- POST /messages: send a list of ciphertexts atomically. Every message is
892−-- validated (self-send + per-sender rate window) inside a single atomic
893−-- block; if any one would be rejected, none are enqueued and the call returns
894−-- a non-2xx HTTP status. The send-rate counter is still charged for every
895−-- attempted message even when the batch is rejected, so a sender can't probe
896−-- limits cheaply (e.g. by sending to themselves). Federation forwards are
897−-- kicked per message after commit.
898−handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} (Result HttpError {})
765+-- POST /messages: send a list of ciphertexts atomically. Self-send is
766+-- rejected up-front; the remaining batch is committed inside one atomic
767+-- block (all-or-nothing) which reads presence to pre-stage federation
768+-- forwards. Federation forwards are kicked per message after commit.
769+handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> IO {r *presence, r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *pollWakeup, rw *seqCounter, network, clock} (Result HttpError {})
899770handleSendMessages = \messages authorization clientIp -> do
900771 t <- now
901− gateAuth authorization clientIp t (\senderKey -> do
772+ withSessionAuth authorization t clientIp (yield (Err {error: {status: 401, message: "unauthorized"}})) (\senderKey -> do
902773 let limited = take maxBatchSize messages
903− decision <- atomic (do
904− checked <- mapEach limited (\m -> do
905− c <- checkSendRate senderKey m.to t
906− yield {c.error, c.remoteServers, m.to, m.encryptedBlob})
907− case findFirst checked (\c -> c.error != "") of
908− Just {value: bad} -> yield (Err {bad.error})
909− Nothing {} -> do
910− forEach checked (\c -> commitSendMessage c.to c.encryptedBlob c.remoteServers t)
911− yield (Ok {value: checked}))
912− case decision of
913− Err {error: code} -> yield (Err {
914− error: {status: if code == "self_send" then 400 else 429, message: code}
915− })
916− Ok {value: checked} -> do
917− when federationEnabled (forEach checked (\c -> fork (forkForwardsTo c.to c.encryptedBlob c.remoteServers)))
918− yield (Ok {value: {}}))
919−
920−-- Charge the send rate counter for the sender and decide whether the message
921−-- would be accepted. Mutates only the rate counter, never the mailbox; the
922−-- caller commits via `commitSendMessage` only after every message in the
923−-- batch has passed. The counter is charged unconditionally (even on a
924−-- failure) so that a sender can't probe limits cheaply by sending to
925−-- themselves.
926−checkSendRate : PubkeyHex -> PubkeyHex -> Int<Ms> -> IO {r *presence, rw *sendRates} {remoteServers: [Text], error: Text}
927−checkSendRate = \senderKey to t -> do
928− sRates <- *sendRates
929− let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t
930− *sendRates = sR.entries
931− if senderKey == to
932− then yield {error: "self_send", remoteServers: []}
933− else if sR.limited
934− then yield {error: "rate_limited", remoteServers: []}
935− else do
936− pres <- *presence
937− let toPresence = presenceForKey pres to t
938− let remoteServers = collectPeers toPresence (\p -> p.server)
939− yield {error: "", remoteServers}
774+ if any (\m -> m.to == senderKey) limited
775+ then yield (Err {error: {status: 400, message: "self_send"}})
776+ else do
777+ committed <- atomic (do
778+ pres <- *presence
779+ let staged = map (\m -> {
780+ m.to,
781+ m.encryptedBlob,
782+ remoteServers: collectPeers (presenceForKey pres m.to t) (\p -> p.server)
783+ }) limited
784+ forEach staged (\c -> commitSendMessage c.to c.encryptedBlob c.remoteServers t)
785+ yield staged)
786+ when federationEnabled (forEach committed (\c -> fork (forkForwardsTo c.to c.encryptedBlob c.remoteServers)))
787+ yield (Ok {value: {}}))
940788
941789-- Allocate a monotonic receivedAt via `seqCounter`, append the message to
942790-- the mailbox, and pre-stage forwards for any remote presence rows that
943791-- already exist for `to`. Caller must have already validated the message
944−-- via `checkSendRate`.
792+-- via `resolveSendTarget`.
945793commitSendMessage : PubkeyHex -> BlobHex -> [Text] -> Int<Ms> -> IO {rw *forwards, rw *messages, rw *pollWakeup, rw *seqCounter} {}
946794commitSendMessage = \to encryptedBlob remoteServers t -> do
947795 msgs <- *messages