@@ -1,15 +1,24 @@
11-- Skrepka Relay Server
22-- Single-server implementation of the Skrepka protocol (v0.1)
33-- Binary data (keys, signatures, blobs) uses hex encoding (bytesToHex/bytesFromHex)
4+--
5+-- Architecture:
6+-- Clients authenticate via Ed25519 challenge-response, then poll for messages via long-polling.
7+-- Messages are E2E encrypted blobs routed by recipient pubkey.
8+-- Federation: servers gossip presence info and forward messages for keys homed elsewhere.
9+-- Presence tracks which server a pubkey was last seen on (online/offline via gossip).
10+-- All state is in-memory; background tasks handle pruning and retry of failed federation ops.
411
512-- ============================================
613-- Configuration
714-- ============================================
815
16+-- Server identity and federation credentials
917serverHost = "localhost:8080"
1018federationSecret = "" -- must be set to enable federation
1119maintenanceSecret = "" -- set to enable /maintenance/prune independently of federation
1220federationScheme = "https://" -- scheme for federation HTTP calls
21+-- Message and data size limits
1322messageMaxAge : Int<Ms>
1423messageMaxAge = 2592000000 -- 30 days
1524maxPresenceTtl : Int<S>
@@ -22,6 +31,7 @@ maxGossipEvents = 100
2231maxPollEvents = 50
2332maxPresenceEntries = 10000
2433maxMessagesPerRecipient = 500
34+-- Global capacity limits for in-memory collections
2535clockSkewTolerance : Int<Ms>
2636clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers
2737maxForwardsEntries = 10000
@@ -31,40 +41,44 @@ maxTokenLen = 256
3141maxChallenges = 10000
3242maxSessions = 10000
3343maxAckedIds = 10000
34−maxAckOverflow = 200
44+maxAckOverflow = 200 -- overflow buffer for acks that couldn't be sent yet
3545maxDeliveredIds = 10000
46+-- Per-key limits and rate limiting
3647maxChallengesPerKey = 5
3748maxSessionsPerKey = 10
3849challengeExpiry : Int<Ms>
3950challengeExpiry = 60000 -- 1 minute
40−maxChallengeRatePerKey = 5
51+maxChallengeRatePerKey = 5 -- max challenge requests per key per window
4152maxChallengeRateEntries = 1000
4253challengeRateWindowMs : Int<Ms>
4354challengeRateWindowMs = 1000
44−maxSendRatePerKey = 20
55+maxSendRatePerKey = 20 -- max messages sent per key per window
4556sendRateWindowMs : Int<Ms>
4657sendRateWindowMs = 1000
4758maxSendRateEntries = 1000
59+-- Session and gossip timing
4860sessionExpiry : Int<Ms>
4961sessionExpiry = 3600000 -- 1 hour
5062onlineGossipTtl : Int<S>
51−onlineGossipTtl = 3600 -- 1 hour
63+onlineGossipTtl = 3600 -- 1 hour; TTL announced when a key authenticates
5264offlineGossipMinTtl : Int<S>
5365offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events
5466maxForwardRetriesPerServer = 100
55−maxGossipHops = 5
56−gossipOriginHops = 2
67+maxGossipHops = 5 -- max hops a gossip event can propagate across servers
68+gossipOriginHops = 2 -- hops set when this server originates a gossip event
69+-- Exponential backoff for failed federation forwards
5770retryBackoffBase : Int<Ms>
5871retryBackoffBase = 30000 -- 30 seconds
5972retryBackoffMax : Int<Ms>
6073retryBackoffMax = 480000 -- 8 minutes
61−maxForwardFailures = 1000
74+maxForwardFailures = 1000 -- max tracked server failure records
6275
76+-- Background task intervals
6377backgroundPruneInterval : Int<Ms>
64−backgroundPruneInterval = 30000 -- 30 seconds
78+backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
6579pollTimeoutMs : Int<Ms>
66−pollTimeoutMs = 25000 -- long-poll heartbeat interval
67−maxPollsPerKey = 3
80+pollTimeoutMs = 25000 -- long-poll heartbeat interval (how long a poll blocks before returning empty)
81+maxPollsPerKey = 3 -- max concurrent long-poll connections per pubkey
6882maxActivePolls = 5000
6983presenceRefreshInterval : Int<Ms>
7084presenceRefreshInterval = 300000 -- 5 minutes
@@ -78,8 +92,10 @@ federationUrl = \server -> federationScheme ++ server
7892-- Types
7993-- ============================================
8094
95+-- Event delivered to a polling client: an encrypted message blob
8196type PollEvent = {id: MessageIdHex, eventType: Text, payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex}}
8297
98+-- Gossip events propagate presence changes between federated servers
8399data GossipEventType = GossipOnline {} | GossipOffline {} | GossipUnknown {}
84100
85101impl ToJSON GossipEventType where
@@ -95,10 +111,15 @@ impl FromJSON GossipEventType where
95111 else if s == "offline" then GossipOffline {}
96112 else GossipUnknown {}
97113
114+-- Wire format for gossip events sent between servers
98115type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex, server: ServerName, timestamp: Timestamp, ttl: Int<S>}
99116
117+-- A gossip event that passed validation (timestamp not in future, TTL > 0, not expired)
100118type ValidGossipEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Timestamp, timestamp: Timestamp, ttl: Int<S>}
101119
120+-- Status returned to the sender after SendMessage
121+-- Delivered: recipient is online locally; Federated: forwarded to remote server;
122+-- Queued: stored but recipient location unknown; Unauthorized/Rejected: not accepted
102123data MessageStatus = MsDelivered {} | MsFederated {} | MsQueued {} | MsUnauthorized {} | MsRejected {}
103124
104125impl ToJSON MessageStatus where
@@ -109,8 +130,10 @@ impl ToJSON MessageStatus where
109130 MsUnauthorized {} -> "unauthorized"
110131 MsRejected {} -> "rejected")
111132
133+-- Internal delivery status (same as MessageStatus but without error variants)
112134data DeliveryStatus = DsDelivered {} | DsFederated {} | DsQueued {}
113135
136+-- Validated hex-encoded types with maximum length constraints
114137isValidBoundedHex : Int -> Text -> Bool
115138isValidBoundedHex = \maxLen s -> isValidHex s && not (tooLong s maxLen) && s == normalizeHex s
116139type PubkeyHex = Text where isValidBoundedHex maxPubkeyLen
@@ -120,6 +143,8 @@ isValidMessageId : Text -> Bool
120143isValidMessageId = isValidBoundedHex maxMessageIdLen
121144type MessageIdHex = Text where isValidMessageId
122145type OptionalMessageId = Text where \s -> s == "" || isValidMessageId s
146+-- ServerName: validated server hostname (rejects private IPs, localhost, etc.)
147+-- AnyServerName: normalized server name without the security checks (used in presence lookups)
123148isValidServerName : Text -> Bool
124149isValidServerName = \s -> not (tooLong s maxServerNameLen) && s == normalizeServerName s && not (isBadServerName s)
125150type ServerName = Text where isValidServerName
@@ -127,16 +152,26 @@ type AnyServerName = Text where \s -> not (tooLong s maxServerNameLen) && s == n
127152
128153type Timestamp = Int<Ms>
129154
155+-- Pending challenge for Ed25519 challenge-response auth
130156type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp}
157+-- Authenticated session; token is a random hex string
131158type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp}
159+-- Stored message; originServer is "" for locally-originated messages, set to sender server for federated forwards
132160type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp}
161+-- Presence: tracks that a pubkey is online at a given server, with expiry
133162type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
163+-- Queued federation forward: message waiting to be delivered to a remote server
134164type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
165+-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
135166type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
167+-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
136168type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
169+-- Tracks consecutive failures per federation peer for exponential backoff
137170type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}
171+-- Sliding-window rate limit state per pubkey
138172type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
139173type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
174+-- Tracks active long-poll connections to enforce per-key concurrency limits
140175type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
141176data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
142177data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName]} | SendDuplicate {}
@@ -145,6 +180,8 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
145180-- ============================================
146181-- Persisted State
147182-- ============================================
183+-- All state is in-memory (no disk persistence). Prefixed with * (mutable refs).
184+-- Atomic blocks provide transactional consistency for reads + writes.
148185
149186*challenges : [Challenge]
150187
@@ -154,10 +191,12 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
154191
155192*presence : [PresenceEntry]
156193
194+-- Monotonic counter ensuring unique receivedAt ordering even within the same millisecond
157195*seqCounter : Timestamp
158196
159197*forwards : [ForwardEntry]
160198
199+-- Messages acked by local recipients, pending federation ack to origin server
161200*ackedIds : [AckedEntry]
162201
163202*lastPrunedAt : Timestamp
@@ -170,8 +209,12 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
170209
171210*challengeRates : [ChallengeRateEntry]
172211
212+-- Heartbeat signal: incremented periodically to wake all blocking long-polls
173213*pollWakeup : Timestamp
174214
215+-- Per-bucket message signals (16 buckets, keyed by first hex digit of pubkey).
216+-- Updated when a message is stored for a recipient; polls for that pubkey's
217+-- bucket watch for changes to avoid waking unrelated polls.
175218*messageSignal0 : Timestamp
176219*messageSignal1 : Timestamp
177220*messageSignal2 : Timestamp
@@ -191,14 +234,20 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
191234
192235*activePolls : [ActivePoll]
193236
237+-- Overflow buffer: acks evicted from ackedIds before they could be sent to origin
194238*ackOverflow : [AckedEntry]
195239
240+-- Final dedup set: messages confirmed delivered (ack sent or overflow drained)
196241*deliveredIds : [DeliveredEntry]
197242
198243-- ============================================
199244-- Routes
200245-- ============================================
201246
247+-- Authentication: Ed25519 challenge-response flow
248+-- 1. Client POSTs pubkey to /auth/challenge, gets a random challenge string
249+-- 2. Client signs the challenge with their private key and POSTs to /auth/verify
250+-- 3. Server verifies signature, returns a session token
202251route AuthApi where
203252 /auth
204253 POST {pubkey: PubkeyHex} /challenge
@@ -208,30 +257,39 @@ route AuthApi where
208257 -> {token: Text, expiresAt: Timestamp}
209258 = VerifyAuth
210259
260+-- Client-facing API (requires session token in Authorization header)
211261route ClientApi where
262+ -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
212263 POST {cursor: Timestamp} /poll headers {authorization: Text}
213264 -> {events: [PollEvent], cursor: Timestamp, authorized: Bool}
214265 = Poll
215266 /messages
267+ -- Send an E2E encrypted message to a recipient pubkey
216268 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text}
217269 -> {status: MessageStatus, messageId: OptionalMessageId}
218270 = SendMessage
271+ -- Look up which server a pubkey is present on (for federation routing)
219272 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text}
220273 -> {server: AnyServerName, authorized: Bool}
221274 = LookupPresence
222275
276+-- Server-to-server federation API (requires federation secret)
223277route FederationApi where
224278 /federation
279+ -- Receive presence gossip events from a peer server
225280 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName} /gossip headers {authorization: Text}
226281 -> {ok: Bool}
227282 = RecvGossip
283+ -- Receive a forwarded message from a peer server on behalf of a sender
228284 POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName} /forward headers {authorization: Text}
229285 -> {ok: Bool}
230286 = ForwardMessage
287+ -- Acknowledge delivery of messages back to the origin server
231288 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName} /ack headers {authorization: Text}
232289 -> {ok: Bool}
233290 = FederationAck
234291
292+-- Triggers pruning of expired state and retry of pending federation ops
235293route MaintenanceApi where
236294 POST {} /maintenance/prune headers {authorization: Text}
237295 -> {ok: Bool}
@@ -240,6 +298,7 @@ route MaintenanceApi where
240298route HealthApi where
241299 GET /health -> {ok: Bool} = Health
242300
301+-- Returns counts of in-memory collections (requires federation or maintenance secret)
243302route StatsApi where
244303 GET /stats headers {authorization: Text}
245304 -> {sessions: Int, messages: Int, presence: Int, forwards: Int, challenges: Int, ackedIds: Int, ok: Bool}
@@ -274,12 +333,14 @@ minBy = \items getKey -> fold (\acc x -> case acc of
274333 Nothing {} -> Just {value: x}
275334 Just {value: v} -> if getKey x < getKey v then Just {value: x} else acc) Nothing {} items
276335
336+-- Hex encoding utilities for token generation
277337hexDigit : Int -> Text
278338hexDigit = \n -> take 1 (drop (clamp 0 15 n) "0123456789abcdef")
279339
280340byteToHex : Int -> Text
281341byteToHex = \n -> hexDigit (n / 16) ++ hexDigit (n - (n / 16) * 16)
282342
343+-- Encode a 24-bit integer as 6 hex characters (3 bytes)
283344intTo3Hex : Int -> Text
284345intTo3Hex = \n ->
285346 let b0 = n / 65536 in
@@ -288,11 +349,13 @@ intTo3Hex = \n ->
288349 let b2 = r0 - b1 * 256 in
289350 byteToHex b0 ++ byteToHex b1 ++ byteToHex b2
290351
352+-- Generate a random hex token (n chunks of 3 bytes each = 6n hex chars)
291353genTokenN = \n -> if n <= 0 then yield "" else do
292354 chunk <- randomInt 16777216
293355 rest <- genTokenN (n - 1)
294356 yield (intTo3Hex chunk ++ rest)
295357
358+-- Default token: 8 chunks = 48 hex chars = 24 bytes of randomness
296359genToken = genTokenN 8
297360
298361tooLong : Text -> Int -> Bool
@@ -323,6 +386,7 @@ isAllowedServerChar = \c -> contains c "0123456789abcdefghijklmnopqrstuvwxyz-.:"
323386containsBadUrlChar : Text -> Bool
324387containsBadUrlChar = \s -> any (chars s) (\c -> not (isAllowedServerChar c))
325388
389+-- Carrier-grade NAT range: 100.64.0.0/10 (RFC 6598)
326390isCgnatRange : Text -> Bool
327391isCgnatRange = \s ->
328392 take 4 s == "100."
@@ -431,6 +495,9 @@ hasDotBeforePort = \s ->
431495 else if take 1 s == "." then True {}
432496 else hasDotBeforePort (drop 1 s)
433497
498+-- DNS rebinding / SSRF protection: reject server names that resolve to private/reserved addresses.
499+-- Blocks localhost, private RFC1918 ranges, link-local, loopback, CGNAT, documentation ranges,
500+-- IPv6 shorthand, octal/hex IP tricks, bare IPs without dots, and .arpa domains.
434501isBadServerName : Text -> Bool
435502isBadServerName = \sn ->
436503 sn == ""
@@ -462,18 +529,21 @@ isBadServerName = \sn ->
462529 || not (hasDotBeforePort sn)
463530 || containsSuffix ".arpa" sn
464531
532+-- Look up a non-expired session matching the given token (constant-time comparison)
465533authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
466534authenticateSession = \sessions token t ->
467535 case headOf (do s <- sessions; where s.expiresAt > t && safeCompare s.token token; yield s) of
468536 Nothing {} -> Unauthorized {}
469537 Just {value: s} -> Authorized {pubkey: s.pubkey}
470538
539+-- Extract bearer token from "Authorization: Bearer <token>" header
471540extractToken = \authHeader ->
472541 let token = drop 7 authHeader in
473542 if toLower (take 7 authHeader) == "bearer " && not (tooLong token maxTokenLen)
474543 then Just {value: token}
475544 else Nothing {}
476545
546+-- Constant-time secret comparison (prevents timing attacks on federation/maintenance secrets)
477547verifySecret : Text -> Text -> Bool
478548verifySecret = \token secret ->
479549 secret != "" && safeCompare token secret
@@ -505,9 +575,11 @@ deliveryToMessageStatus = \ds -> case ds of
505575isQueued : DeliveryStatus -> Bool
506576isQueued = \ds -> case ds of DsQueued {} -> True {}; _ -> False {}
507577
578+-- Ensure strictly monotonic timestamps for message ordering
508579nextSeqTs : Timestamp -> Timestamp -> Timestamp
509580nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
510581
582+-- Deduplicate message IDs, preserving order, capped at maxMessageIds
511583dedupMessageIds : [MessageIdHex] -> [MessageIdHex]
512584dedupMessageIds = \messageIds ->
513585 fold (\acc mid ->
@@ -515,10 +587,14 @@ dedupMessageIds = \messageIds ->
515587 else if any acc (\a -> a == mid) then acc
516588 else union acc [mid]) [] messageIds
517589
590+-- Remove the N oldest acked entries (by ackedAt timestamp)
518591evictOldestAckedIds = \items n ->
519592 if n <= 0 then items
520593 else diff items (takeRelation n (sortBy (\a -> a.ackedAt) items))
521594
595+-- Constant-time string comparison to prevent timing side-channels.
596+-- Always iterates maxTokenLen characters regardless of input length;
597+-- rejects empty strings to avoid vacuous equality.
522598safeCompareLoop : Text -> Text -> Int -> Int -> Int
523599safeCompareLoop = \s1 s2 n result ->
524600 if n <= 0 && s1 == "" && s2 == ""
@@ -529,6 +605,7 @@ safeCompare : Text -> Text -> Bool
529605safeCompare = \a b ->
530606 safeCompareLoop a b maxTokenLen 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0
531607
608+-- Remove the single item with the smallest key value (LRU-style eviction)
532609evictOneOldest = \items getKey ->
533610 let result = fold (\acc x ->
534611 case acc.candidate of
@@ -542,6 +619,8 @@ evictOneOldest = \items getKey ->
542619 Nothing {} -> items
543620 Just {} -> result.items
544621
622+-- Enforce per-key capacity: if a key has >= maxPerKey items, evict the oldest for that key.
623+-- Also prunes expired items.
545624capPerKey = \items matchKey maxPerKey getExpiry t ->
546625 let active = do x <- items; where getExpiry x > t; yield x in
547626 let forKey = do x <- active; where matchKey x; yield x in
@@ -550,11 +629,13 @@ capPerKey = \items matchKey maxPerKey getExpiry t ->
550629 union others (evictOneOldest forKey getExpiry)
551630 else active
552631
632+-- Enforce global capacity: if at maxCount, evict the single oldest item
553633capGlobal = \items maxCount getKey ->
554634 if count items >= maxCount
555635 then evictOneOldest items getKey
556636 else items
557637
638+-- Two-tier capacity enforcement: per-key first, then global
558639capPerKeyAndGlobal = \items matchKey maxPerKey getExpiry t maxGlobal ->
559640 capGlobal (capPerKey items matchKey maxPerKey getExpiry t) maxGlobal getExpiry
560641
@@ -566,6 +647,7 @@ removeOneForKey = \items pubkey ->
566647 ) {removed: False {}, items: []} items
567648 in result.items
568649
650+-- Extract unique remote (non-local) server names from a list of items
569651collectPeerServers : [a] -> (a -> Text) -> [Text]
570652collectPeerServers = \items getServer ->
571653 fold (\acc item ->
@@ -577,6 +659,7 @@ collectPeerServers = \items getServer ->
577659samePubkeyServer = \a b -> a.pubkey == b.pubkey && a.server == b.server
578660pubkeyServer = \e -> {pubkey: e.pubkey, server: e.server}
579661
662+-- Upsert items by key, keeping the fresher entry on conflict (last-write-wins by freshness)
580663mergeByKey = \sameKey getFreshness initial items ->
581664 fold (\acc e ->
582665 if any acc (\a -> sameKey a e && getFreshness a >= getFreshness e)
@@ -585,11 +668,14 @@ mergeByKey = \sameKey getFreshness initial items ->
585668 (do a <- acc; where not (sameKey a e); yield a)
586669 [e]) initial items
587670
671+-- Merge presence entries by (pubkey, server), keeping the one with the latest expiresAt
588672mergeByPubkeyServer = mergeByKey samePubkeyServer (\x -> x.expiresAt)
589673
674+-- Keep entries from setA that are not superseded by a newer-timestamped entry in setB
590675resolveByTimestamp = \setA setB ->
591676 do e <- setA; where none setB (\o -> samePubkeyServer o e && o.timestamp > e.timestamp); yield e
592677
678+-- Remove presence entries that have a corresponding offline event with a newer timestamp
593679removeOffline = \presenceSet trustedOffline t ->
594680 diff presenceSet (do
595681 p <- presenceSet
@@ -597,6 +683,8 @@ removeOffline = \presenceSet trustedOffline t ->
597683 where samePubkeyServer p o && min o.timestamp t > p.timestamp
598684 yield p)
599685
686+-- Cap presence table after gossip merge. If over capacity, only allow updates to
687+-- existing entries and reclaim slots from expired ones for new entries.
600688capGossipPresence = \args ->
601689 if count args.afterOffline <= maxPresenceEntries
602690 then {finalPresence: args.afterOffline, appliedOnline: args.appliedOnline, withinCapacity: True {}}
@@ -620,6 +708,8 @@ countMessagesFor = \messages toKey t ->
620708hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool
621709hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)
622710
711+-- Validate incoming gossip events: reject local-server events, future timestamps,
712+-- and already-expired entries. Clamp TTLs and split into online/offline buckets.
623713validGossipEvents : [GossipEvent] -> Timestamp -> {online: [ValidGossipEntry], offline: [ValidGossipEntry]}
624714validGossipEvents = \events t ->
625715 let validate = \e ->
@@ -636,6 +726,7 @@ validGossipEvents = \events t ->
636726 GossipOffline {} -> {online: acc.online, offline: union acc.offline (validate e)}
637727 _ -> acc) {online: [], offline: []} events
638728
729+-- Execute an action for each server in parallel (fork per server)
639730sendToEach : [ServerName] -> (ServerName -> (a {})) -> a {}
640731sendToEach = \servers action ->
641732 case headOf servers of
@@ -645,6 +736,8 @@ sendToEach = \servers action ->
645736 fork (action server)
646737 sendToEach rest action
647738
739+-- Remove all expired entries from every collection. Returns expired local presence
740+-- entries (for broadcasting offline gossip) and the set of peer servers known before pruning.
648741pruneExpired = \t -> do
649742 challenges <- *challenges
650743 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
@@ -675,6 +768,7 @@ pruneExpired = \t -> do
675768 set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > messageCutoff; yield d
676769 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
677770
771+-- Opportunistic pruning: runs at most once per second, triggered by request handlers
678772maybePrune = \t -> do
679773 lastPruned <- *lastPrunedAt
680774 if t - lastPruned < 1000
@@ -759,6 +853,7 @@ removeForward = \mid srv -> atomic do
759853 forwards <- *forwards
760854 set *forwards = forwardsExcluding forwards srv [mid]
761855
856+-- Exponential backoff: 30s -> 60s -> 120s -> 240s -> 480s (capped)
762857retryBackoff : Int -> Int<Ms>
763858retryBackoff = \failures ->
764859 if failures <= 1 then retryBackoffBase
@@ -808,6 +903,7 @@ retryForwardsForServer = \entries remaining ->
808903 clearFailure f.toServer
809904 retryForwardsForServer rest (remaining - 1)
810905
906+-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
811907retryPendingForwards : [ForwardEntry] -> IO {network} {}
812908retryPendingForwards = \pending -> do
813909 let servers = collectPeerServers pending (\x -> x.toServer)
@@ -818,6 +914,7 @@ retryPendingForwards = \pending -> do
818914 retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)
819915 yield {}
820916
917+-- When a remote key comes online, create forward entries for queued messages and send them
821918forwardQueuedMessages = \targets t -> do
822919 newForwards <- atomic do
823920 messages <- *messages
@@ -851,12 +948,16 @@ sendBatchedAcksToServer = \serverAcks server ->
851948 let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch
852949 set *ackedIds = updatedAckedIds)
853950
951+-- Send delivery acknowledgments back to origin servers and remove acked entries on success
854952sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
855953sendAcksAndClean = \pendingAcks ->
856954 let servers = collectPeerServers pendingAcks (\a -> a.originServer) in
857955 sendToEach servers (\server ->
858956 sendBatchedAcksToServer (acksForServer pendingAcks server) server)
859957
958+-- Retry sending acks that were evicted from ackedIds due to capacity limits.
959+-- On failure, re-insert into ackedIds or overflow buffer; overflow entries eventually
960+-- move to deliveredIds (terminal dedup state) if the buffer is also full.
860961retryEvictedAcks = \evictedEntries pendingAcks ->
861962 let evictedToRetry = retriableEvictedAcks evictedEntries pendingAcks
862963 in do
@@ -897,6 +998,7 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
897998 (println ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries moved to delivered set for " ++ server)))
898999 Ok _ -> yield {})
8991000
1001+-- Retry acks sitting in the overflow buffer (last chance before they become deliveredIds)
9001002retryOverflowAcks = \overflow ->
9011003 let servers = collectPeerServers overflow (\a -> a.originServer) in
9021004 sendToEach servers (\server ->
@@ -908,6 +1010,9 @@ retryOverflowAcks = \overflow ->
9081010 ackOverflow <- *ackOverflow
9091011 set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))
9101012
1013+-- Full maintenance cycle: prune expired state, retry pending forwards and acks,
1014+-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
1015+-- nearing expiry. Called by both the background task and the /maintenance/prune endpoint.
9111016pruneAndRetry = \t -> do
9121017 result <- atomic do
9131018 set *lastPrunedAt = t
@@ -941,9 +1046,13 @@ hexCharIndex = \c ->
9411046 else if c == "8" then 8 else if c == "9" then 9 else if c == "a" then 10 else if c == "b" then 11
9421047 else if c == "c" then 12 else if c == "d" then 13 else if c == "e" then 14 else 15
9431048
1049+-- Map pubkey to one of 16 signal buckets based on first hex digit.
1050+-- This partitions the signal space so storing a message for one pubkey
1051+-- only wakes polls in the same bucket, not all active polls.
9441052pubkeyBucket : PubkeyHex -> Int
9451053pubkeyBucket = \pubkey -> hexCharIndex (take 1 pubkey)
9461054
1055+-- Update the signal timestamp for a pubkey's bucket, waking any blocked polls in that bucket
9471056signalMessageForKey = \pubkey ts ->
9481057 let b = pubkeyBucket pubkey in
9491058 if b == 0 then set *messageSignal0 = ts
@@ -982,6 +1091,9 @@ readMessageSignal = \pubkey ->
9821091 else if b == 14 then do s <- *messageSignal14; yield s
9831092 else do s <- *messageSignal15; yield s
9841093
1094+-- Acknowledge messages up to the given cursor: remove from messages, cancel pending forwards,
1095+-- and create ack entries for federated messages (to notify origin servers).
1096+-- Returns pending acks and any evicted entries that exceeded ackedIds capacity.
9851097ackPolledMessages = \pubkey cursor t -> do
9861098 messages <- *messages
9871099 let toAck = do m <- messages; where m.toKey == pubkey && m.receivedAt <= cursor; yield m
@@ -1012,6 +1124,12 @@ pollNewEvents = \pubkey clampedCursor t -> do
10121124 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
10131125 yield {events: evts, cursor: lastReceivedAt}
10141126
1127+-- Execute a single long-poll cycle:
1128+-- 1. Authenticate and ack messages up to cursor (atomic)
1129+-- 2. Acquire a poll slot (limit concurrent polls per key)
1130+-- 3. Check for existing messages; if none, block on signal/heartbeat (retry-based wait)
1131+-- 4. Re-authenticate after waking (session may have expired during wait)
1132+-- 5. Return new events and release the poll slot
10151133pollOnce = \token cursor -> do
10161134 t <- now
10171135 setup <- atomic do
@@ -1079,41 +1197,51 @@ pollOnce = \token cursor -> do
10791197-- ============================================
10801198-- Derived Relations
10811199-- ============================================
1200+-- Pure query functions over in-memory state. No side effects.
10821201
1202+-- All non-expired presence entries for a given pubkey
10831203presenceForKey = \presence pubkey t ->
10841204 do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield p
10851205
1206+-- Pubkeys that are online on this server
10861207localOnlineKeys = \presence t ->
10871208 do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p.pubkey
10881209
1210+-- Remove the local presence entry for a pubkey (used before re-inserting on auth)
10891211presenceWithoutLocalKey = \presence pubkey ->
10901212 do p <- presence; where p.pubkey != pubkey || not (isLocalServer p.server); yield p
10911213
1214+-- Local presence entries that expired during this prune cycle (for offline gossip)
10921215expiredLocalPresence = \before after t ->
10931216 do p <- before; where p.expiresAt <= t && isLocalServer p.server && none after (\q -> q.pubkey == p.pubkey && isLocalServer q.server); yield p
10941217
1218+-- Resolve the best server for a pubkey: prefer local, otherwise pick the remote with latest expiry
10951219bestServerForKey = \entries ->
10961220 if any entries (\p -> isLocalServer p.server) then serverHostN
10971221 else case maxBy (do p <- entries; where not (isLocalServer p.server); yield p) (\p -> p.expiresAt) of
10981222 Just {value: p} -> p.server
10991223 Nothing {} -> ""
11001224
1225+-- Messages for a pubkey newer than cursor and within messageMaxAge
11011226pollableMessages = \messages pubkey cursor t ->
11021227 do m <- messages; where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > cursor; yield m
11031228
1229+-- Build ack entries for messages that came from a remote origin server (need federation ack)
11041230federatedAckEntries = \ackedMessages t ->
11051231 do m <- ackedMessages; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
11061232
1233+-- Remove forwards for specific message IDs destined to a given server
11071234forwardsExcluding = \forwards server messageIds ->
11081235 do f <- forwards; where not (f.toServer == server && any messageIds (\mid -> f.messageId == mid)); yield f
11091236
1237+-- Find locally-originated messages that haven't been forwarded yet to each target's server
11101238forwardCandidates = \targets messages forwards t ->
11111239 fold (\acc target ->
11121240 let serverFwdIds = do f <- forwards; where f.toServer == target.server; yield f.messageId in
11131241 union acc (do
11141242 m <- messages
11151243 where m.toKey == target.pubkey
1116− where m.originServer == ""
1244+ where m.originServer == "" -- only forward locally-originated messages
11171245 where none serverFwdIds (\fid -> fid == m.messageId)
11181246 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
11191247 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp})) [] targets
@@ -1137,32 +1265,41 @@ acksForServer = \acks server -> do a <- acks; where a.originServer == server; yi
11371265
11381266forwardsForServer = \forwards server -> do f <- forwards; where f.toServer == server; yield f
11391267
1268+-- Only trust offline events that claim to be from the sending server itself
11401269trustedOfflineEvents = \events fromServer ->
11411270 do e <- events; where e.server == fromServer; yield e
11421271
1272+-- Peer servers to re-gossip to, excluding the server that sent us the gossip
11431273gossipTargets = \peerServers fromServer ->
11441274 do s <- peerServers; where s != fromServer; yield s
11451275
1276+-- Remote keys that just came online (triggers forwarding of queued messages)
11461277newlyOnlineRemoteKeys = \events appliedKeys ->
11471278 do e <- events; where any appliedKeys (\a -> samePubkeyServer a e); where not (isLocalServer e.server); yield {pubkey: e.pubkey, server: e.server}
11481279
1280+-- Trust gossip entries if they're from the sender or about a server we already know about
11491281trustedGossipEntries = \deduped presence fromServer t ->
11501282 do e <- deduped; where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t); yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp}
11511283
1284+-- Keys whose presence was actually updated (new or extended expiry)
11521285appliedOnlineKeys = \entries presence ->
11531286 do e <- entries; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield (pubkeyServer e)
11541287
1288+-- Gossip entries that update existing presence (vs. creating new entries)
11551289existingPresenceUpdates = \entries presence ->
11561290 do e <- entries; where any presence (\p -> samePubkeyServer p e); yield e
11571291
1292+-- Gossip entries for keys not yet in the presence table
11581293newPresenceEntries = \entries presence ->
11591294 do e <- entries; where none presence (\p -> samePubkeyServer p e); yield e
11601295
1296+-- Generate offline gossip events for locally-expired presence entries
11611297offlineGossipEvents = \expiredLocal -> do
11621298 p <- expiredLocal
11631299 let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
11641300 yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.expiresAt, ttl: ttl}
11651301
1302+-- Filter targets to only those not locally online (avoid forwarding to keys that can poll directly)
11661303remoteOnlyTargets = \targets localOnline ->
11671304 do target <- targets; where none localOnline (\pk -> pk == target.pubkey); yield target
11681305
@@ -1172,6 +1309,8 @@ remoteOnlyTargets = \targets localOnline ->
11721309
11731310serve = \req -> case req of
11741311
1312+ -- POST /auth/challenge: Generate a random challenge for Ed25519 authentication.
1313+ -- Rate-limited per key; globally capped. Challenge expires in 1 minute.
11751314 ReqChallenge {pubkey, respond} -> do
11761315 challenge <- genToken
11771316 t <- now
@@ -1208,6 +1347,8 @@ serve = \req -> case req of
12081347 forkBroadcastOffline t result.pr
12091348 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
12101349
1350+ -- POST /auth/verify: Verify Ed25519 signature over challenge string.
1351+ -- On success: creates a session, registers local presence, and gossips online status.
12111352 VerifyAuth {pubkey, challenge, signature, respond} -> do
12121353 t <- now
12131354 let expiry = t + sessionExpiry
@@ -1249,6 +1390,9 @@ serve = \req -> case req of
12491390 forkBroadcastOffline t validation.pr
12501391 yield (respond {token: "", expiresAt: 0})
12511392
1393+ -- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor
1394+ -- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.
1395+ -- Acks for federated messages are sent to origin servers in the background.
12521396 Poll {cursor, authorization, respond} -> do
12531397 result <- case extractToken authorization of
12541398 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
@@ -1260,6 +1404,9 @@ serve = \req -> case req of
12601404 retryEvictedAcks result.evictedEntries result.pendingAcks))
12611405 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
12621406
1407+ -- POST /messages: Send an encrypted message to a recipient.
1408+ -- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
1409+ -- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
12631410 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, respond} -> do
12641411 t <- now
12651412 effectiveMsgId <- if messageId != "" then yield messageId else genToken
@@ -1329,6 +1476,8 @@ serve = \req -> case req of
13291476 SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})
13301477
13311478
1479+ -- GET /lookup/:pubkey: Look up the best known server for a pubkey.
1480+ -- Returns local server if key is online locally, otherwise the remote with latest expiry.
13321481 LookupPresence {pubkey, authorization, respond} -> do
13331482 t <- now
13341483 result <- atomic do
@@ -1342,6 +1491,10 @@ serve = \req -> case req of
13421491 forkBroadcastOffline t result.pr
13431492 yield (respond {server: result.server, authorized: result.authorized})
13441493
1494+ -- POST /federation/gossip: Receive presence gossip from a peer server.
1495+ -- Validates events, merges into presence table (last-write-wins), and re-gossips
1496+ -- to other known peers with decremented hop count. Triggers message forwarding
1497+ -- when a remote key comes online.
13451498 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
13461499 t <- now
13471500 let cappedHops = clamp 0 maxGossipHops hopsRemaining
@@ -1370,6 +1523,9 @@ serve = \req -> case req of
13701523 (forwardQueuedMessages newlyOnlineRemote t))
13711524 yield (respond {ok: True {}}))
13721525
1526+ -- POST /federation/forward: Receive a message forwarded from a peer server.
1527+ -- Only accepts messages where fromServer == originServer (no cascade forwarding).
1528+ -- Deduplicates and stores; rejects if recipient mailbox is full.
13731529 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, authorization, respond} -> do
13741530 t <- now
13751531 withFederationAuth authorization t respond (not (isLocalServer originServer) && fromServer == originServer) (do
@@ -1401,6 +1557,9 @@ serve = \req -> case req of
14011557 ForwardRejected {} -> False {}
14021558 _ -> True {}}))
14031559
1560+ -- POST /federation/ack: Peer server confirms delivery of messages we originated.
1561+ -- Removes corresponding forward entries. Also cleans up the original message if the
1562+ -- recipient has no local presence and no other pending forwards for that message.
14041563 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
14051564 t <- now
14061565 withFederationAuth authorization t respond (isLocalServer originServer && not (isLocalServer fromServer)) (do
@@ -1423,6 +1582,8 @@ serve = \req -> case req of
14231582 forkBroadcastOffline t pruneResult
14241583 yield (respond {ok: True {}}))
14251584
1585+ -- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).
1586+ -- Accepts either federation or maintenance secret.
14261587 Prune {authorization, respond} -> do
14271588 t <- now
14281589 let authorized = verifyMaintenanceAuth authorization
@@ -1450,6 +1611,8 @@ serve = \req -> case req of
14501611-- Entry Point
14511612-- ============================================
14521613
1614+-- Background loop: periodically prunes expired state and retries pending federation ops.
1615+-- Jitter prevents thundering herd when multiple servers restart simultaneously.
14531616backgroundPrune = do
14541617 jitter <- randomInt 5000
14551618 sleep (backgroundPruneInterval + jitter)
@@ -1457,6 +1620,8 @@ backgroundPrune = do
14571620 fork (pruneAndRetry t)
14581621 backgroundPrune
14591622
1623+-- Background loop: periodically bumps pollWakeup to unblock long-polls that have been
1624+-- waiting too long (heartbeat). Ensures polls return within pollTimeoutMs even if no messages arrive.
14601625pollHeartbeat = do
14611626 jitter <- randomInt 2000
14621627 sleep (pollTimeoutMs + jitter)