@@ -7,13 +7,36 @@
77-- ============================================
88-- Configuration
99-- ============================================
10−-- Compile-time constants: bind address, federation identity, capacity ceilings,
10+-- Top-level constants: bind address, federation identity, capacity ceilings,
1111-- per-key/window rate limits, expiry/backoff/TTL durations, and protocol version.
12−-- All values are plain literals; nothing is read from disk or env at runtime.
13−
14−-- Hostname this server binds to. Use 0.0.0.0 to accept connections on every interface.
12+-- Values written here are defaults; every scalar (Int/Float/Text/Bool/Maybe ...)
13+-- top-level constant can be overridden at startup via `--name=value` on the
14+-- command line (see Knot's `knot_override_lookup`). Run with `--help` for the
15+-- full list.
16+
17+-- This server's canonical federation identity. Used as the `server` field in
18+-- presence/gossip events, in signed federation challenges, and for self-checks
19+-- (`isLocalServer`). Not a bind address — the listener accepts connections on
20+-- every interface regardless of this value. May also differ from the URL
21+-- clients use to reach the server (e.g. behind a reverse proxy or alternate
22+-- DNS name); peers must resolve this name to reach us for federation.
23+--
24+-- Must be unique per deployment. Every operator participating in federation
25+-- has to recompile with their own public hostname here — peers reject events
26+-- whose origin fails `isBadServerName` (which blocks the default "localhost"
27+-- and other private/reserved names), so two default-built servers cannot
28+-- federate with each other. The default is only suitable for a single-node,
29+-- non-federated deployment.
1530serverHost : Text
1631serverHost = "localhost"
32+-- Network interface the HTTP listener binds to. "0.0.0.0" accepts connections
33+-- on every IPv4 interface (the typical default behind a reverse proxy);
34+-- "127.0.0.1" restricts the listener to localhost-only; "::" / "::1" do the
35+-- IPv6 equivalents. Override with `--bindHost=...` at startup. Unrelated to
36+-- `serverHost` above — this is the kernel-level bind address; that is the
37+-- federation identity peers see.
38+bindHost : Text
39+bindHost = "0.0.0.0"
1740-- TCP port for incoming HTTP connections.
1841listenPort : Int
1942listenPort = 8080
@@ -68,9 +91,6 @@ maxChallenges = 10000
6891-- Ceiling on concurrent active sessions.
6992maxSessions : Int
7093maxSessions = 10000
71−-- Ceiling on remembered acked message IDs (used for replay protection).
72−maxAckedIds : Int
73−maxAckedIds = 10000
7494-- Ceiling on remembered delivered message IDs.
7595maxDeliveredIds : Int
7696maxDeliveredIds = 10000
@@ -319,15 +339,14 @@ type Session = {
319339 createdAt: Timestamp
320340}
321341-- Mailbox row: a stored message awaiting recipient pickup or federation forward.
322−-- `originServer` is "" for messages submitted directly by a local sender.
323342-- `senderKey` is "" for messages received via federation (we don't see the
324−-- remote sender's pubkey).
343+-- remote sender's pubkey); non-empty for messages submitted by a local sender.
344+-- That predicate is the local-vs-federation discriminator throughout the code.
325345type Message = {
326346 messageId: MessageIdHex,
327347 toKey: PubkeyHex,
328348 encryptedBlob: BlobHex,
329349 timestamp: Timestamp,
330− originServer: Text,
331350 receivedAt: Timestamp,
332351 expiresAt: Timestamp,
333352 senderKey: Text
@@ -349,20 +368,10 @@ type ForwardEntry = {
349368 toKey: PubkeyHex,
350369 encryptedBlob: BlobHex,
351370 timestamp: Timestamp,
352− originServer: Text,
353371 retries: Int,
354372 expiresAt: Timestamp,
355373 senderKey: Text
356374}
357−-- Record that a federated message has been acked locally; flushed back to
358−-- `originServer` by `sendAcksAndClean`.
359−type AckedEntry = {
360− messageId: MessageIdHex,
361− toKey: PubkeyHex,
362− ackedAt: Timestamp,
363− originServer: ServerName,
364− retries: Int
365−}
366375-- Replay-protection record: a message id we already delivered locally.
367376type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
368377-- Running failure counter for a peer; keyed by `server`. Drives `retryBackoff`.
@@ -390,12 +399,7 @@ type SenderDelivery = {
390399 deliveredAt: Timestamp
391400}
392401-- One entry of a /messages/batch request body.
393−type BatchMessage = {
394− to: PubkeyHex,
395− encryptedBlob: BlobHex,
396− timestamp: Timestamp,
397− expiresInMs: Int
398−}
402+type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, expiresInMs: Int}
399403-- Per-entry result returned by /messages/batch.
400404type BatchResult = {messageId: MessageIdHex, status: MessageStatus}
401405
@@ -412,8 +416,20 @@ type BatchResult = {messageId: MessageIdHex, status: MessageStatus}
412416*messages : [Message]
413417*presence : [PresenceEntry]
414418*forwards : [ForwardEntry]
415−*ackedIds : [AckedEntry]
416419*deliveredIds : [DeliveredEntry]
420+
421+-- Drop the obsolete `originServer` field. Federation now uses presence-based
422+-- routing exclusively; ack-back is gone, so there's no consumer for the value.
423+migrate *messages
424+ from {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp, expiresAt: Timestamp, senderKey: Text}
425+ to {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, receivedAt: Timestamp, expiresAt: Timestamp, senderKey: Text}
426+ using (\old -> {old.messageId, old.toKey, old.encryptedBlob, old.timestamp, old.receivedAt, old.expiresAt, old.senderKey})
427+
428+migrate *forwards
429+ from {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int, expiresAt: Timestamp, senderKey: Text}
430+ to {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, retries: Int, expiresAt: Timestamp, senderKey: Text}
431+ using (\old -> {old.messageId, old.toServer, old.createdAt, old.toKey, old.encryptedBlob, old.timestamp, old.retries, old.expiresAt, old.senderKey})
432+
417433*forwardFailures : [ForwardFailure]
418434*activePolls : [ActivePoll]
419435
@@ -488,10 +504,8 @@ route ClientApi where
488504
489505route FederationApi where
490506 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/gossip headers {authorization: Text} -> {ok: Bool, error: Text, retryAfterMs: Int} = RecvGossip
491− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, expiresAt: Timestamp} /federation/forward headers {authorization: Text} -> {ok: Bool, error: Text, retryAfterMs: Int} = ForwardMessage
492− POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/ack headers {authorization: Text} -> {ok: Bool, error: Text, ackedCount: Int, results: [{messageId: MessageIdHex, found: Bool}], retryAfterMs: Int} = FederationAck
493− POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, expiresAt: Timestamp}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/forward-batch headers {authorization: Text} -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} = FederationBatchForward
494− POST {messageIds: [MessageIdHex], toKeys: [PubkeyHex], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/delivery-receipt headers {authorization: Text} -> {ok: Bool, error: Text, receivedCount: Int, retryAfterMs: Int} = FederationDeliveryReceipt
507+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, expiresAt: Timestamp} /federation/forward headers {authorization: Text} -> {ok: Bool, error: Text, retryAfterMs: Int} = ForwardMessage
508+ POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: MessageIdHex, expiresAt: Timestamp}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/forward-batch headers {authorization: Text} -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} = FederationBatchForward
495509
496510route FederationPubkeyApi where
497511 GET /federation/pubkey -> {pubkey: Text, ok: Bool} = FederationPubkey
@@ -544,7 +558,7 @@ type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
544558type RecipientCount = {pubkey: PubkeyHex, count: Int}
545559
546560route StatsApi where
547− GET /stats headers {authorization: Text} -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, totalMessageCount: Int, presence: Int, forwards: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, nonceEvictions: Int, activePolls: Int, recipientCounts: Int, messageBucketMax: Int, messageBucketMin: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}], gossipSuccess: Int, gossipFail: Int, latencyBuckets: {lt100: Int, lt500: Int, lt2000: Int, gte2000: Int}}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, senderExpirations: Int, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, pollLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int, sendLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, authAvgMs: Int, authMaxMs: Int, authCount: Int}} = Stats
561+ GET /stats headers {authorization: Text} -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, totalMessageCount: Int, presence: Int, forwards: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, nonceEvictions: Int, activePolls: Int, recipientCounts: Int, messageBucketMax: Int, messageBucketMin: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}], gossipSuccess: Int, gossipFail: Int, latencyBuckets: {lt100: Int, lt500: Int, lt2000: Int, gte2000: Int}}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, senderExpirations: Int, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, pollLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int, sendLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, authAvgMs: Int, authMaxMs: Int, authCount: Int}} = Stats
548562 GET /stats/peers headers {authorization: Text} -> {peers: [PeerStatus], ok: Bool} = StatsPeers
549563 GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text} -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, forwardDetails: [{server: Text, retries: Int, createdAt: Timestamp}], activePolls: Int, ok: Bool} = StatsKey
550564 POST {limit: Int} /stats/top-recipients headers {authorization: Text} -> {recipients: [RecipientCount], ok: Bool, total: Int} = TopRecipients
@@ -624,7 +638,7 @@ pct : Int -> Int -> Int
624638pct = \cur maxN -> if maxN > 0 then cur * 100 / maxN else 0
625639
626640-- Common ack/delete tail: pull matching messages, drop them and their forwards.
627−removeForKey : (forall r. {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, expiresAt: Timestamp, senderKey: Text | r} -> Bool) -> IO {rw *forwards, rw *messages} [Message]
641+removeForKey : (forall r. {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, expiresAt: Timestamp, senderKey: Text | r} -> Bool) -> IO {rw *forwards, rw *messages} [Message]
628642removeForKey = \pred -> do
629643 msgs <- *messages
630644 fwds <- *forwards
@@ -1024,7 +1038,7 @@ clearFailure = \server -> atomic (do
10241038-- ============================================
10251039-- Pure helpers shared by send/poll/federation paths: TTL/timestamp clamping,
10261040-- monotonic sequence generation, presence lookup, duplicate detection across
1027−-- (messages, ackedIds, deliveredIds), and last-write-wins presence merge.
1041+-- (messages, deliveredIds), and last-write-wins presence merge.
10281042
10291043-- Clamp a client-supplied TTL to (0, maxPresenceTtl].
10301044clampTtl : Int<Ms> -> Int<Ms>
@@ -1066,8 +1080,8 @@ hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey
10661080
10671081-- Replay-protection check: True if the (messageId, toKey) is already present in
10681082-- the live mailbox, the federated-acked log, or the delivered log.
1069−isDuplicateMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> [{messageId: MessageIdHex, toKey: PubkeyHex | b}] -> [{messageId: MessageIdHex, toKey: PubkeyHex | c}] -> Bool
1070−isDuplicateMessage = \mid toKey msgs aids dids -> hasMessage mid toKey msgs || hasMessage mid toKey aids || hasMessage mid toKey dids
1083+isDuplicateMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> [{messageId: MessageIdHex, toKey: PubkeyHex | b}] -> Bool
1084+isDuplicateMessage = \mid toKey msgs dids -> hasMessage mid toKey msgs || hasMessage mid toKey dids
10711085
10721086-- Project items through `getServer`, drop our own hostname, and dedupe.
10731087-- Used to derive peer recipient lists from forwards/presence/etc.
@@ -1093,7 +1107,7 @@ mergePresence = \existing incoming -> fold (\acc e -> (\same -> if any acc (\x -
10931107-- One sweep over every state ref: drops expired rows in a single atomic block.
10941108-- Returns the locally-expired presence rows whose pubkey has no other live
10951109-- local session, so the caller can broadcast offline gossip for them.
1096−pruneExpired : Int<Ms> -> IO {rw *ackedIds, rw *activePolls, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, rw *senderDeliveries, rw *senderExpirations, rw *sessions} [{expiresAt: Timestamp, pubkey: PubkeyHex, server: AnyServerName, timestamp: Timestamp}]
1110+pruneExpired : Int<Ms> -> IO {rw *activePolls, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, rw *senderDeliveries, rw *senderExpirations, rw *sessions} [{expiresAt: Timestamp, pubkey: PubkeyHex, server: AnyServerName, timestamp: Timestamp}]
10971111pruneExpired = \t -> atomic do
10981112 ch <- *challenges
10991113 *challenges = filter ch (\x -> x.expiresAt > t)
@@ -1103,8 +1117,6 @@ pruneExpired = \t -> atomic do
11031117 *messages = filter msgs (\m -> m.expiresAt > t)
11041118 fwds <- *forwards
11051119 *forwards = filter fwds (\f -> f.expiresAt > t)
1106− aids <- *ackedIds
1107− *ackedIds = filter aids (\a -> t - a.ackedAt < messageMaxAge)
11081120 dids <- *deliveredIds
11091121 *deliveredIds = filter dids (\d -> t - d.deliveredAt < messageMaxAge)
11101122 ff <- *forwardFailures
@@ -1140,26 +1152,20 @@ mkGossip = \etype pk t ttl -> {
11401152 seenBy: [serverHostN]
11411153}
11421154
1143−-- Build an acked entry for a federated message.
1144−mkAckedEntry : {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text | a} -> Timestamp -> AckedEntry
1145−mkAckedEntry = \m t -> {m.messageId, m.toKey, ackedAt: t, m.originServer, retries: 0}
1146−
1147−-- Record federation acks and sender deliveries for a batch of just-removed messages.
1148−-- Returns the federated-ack entries (the caller flushes them remotely later).
1149−recordAcksAndDeliveries : [{messageId: MessageIdHex, originServer: Text, senderKey: Text, toKey: PubkeyHex | a}] -> Int<Ms> -> IO {rw *ackedIds, rw *senderDeliveries} [{ackedAt: Timestamp, messageId: MessageIdHex, originServer: ServerName, retries: Int, toKey: PubkeyHex}]
1150−recordAcksAndDeliveries = \toAck t -> do
1151− let ackedFed = mapWhere toAck (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
1152− let myDel = mapWhere toAck (\m -> m.senderKey != "" && m.originServer == "") (\m -> {
1155+-- Record sender-side delivery rows for a batch of just-acked locally-originated
1156+-- messages. Federation-origin messages have `senderKey == ""` and are skipped;
1157+-- delivery confirmations for federated sends now travel as application-level
1158+-- receipts via the normal message flow, not server-to-server.
1159+recordSenderDeliveries : [{messageId: MessageIdHex, senderKey: Text, toKey: PubkeyHex | a}] -> Int<Ms> -> IO {rw *senderDeliveries} {}
1160+recordSenderDeliveries = \toAck t -> do
1161+ let myDel = mapWhere toAck (\m -> m.senderKey != "") (\m -> {
11531162 m.senderKey,
11541163 m.messageId,
11551164 m.toKey,
11561165 deliveredAt: t
11571166 })
1158− aids <- *ackedIds
1159− *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)
11601167 sds <- *senderDeliveries
11611168 *senderDeliveries = capList (union sds myDel) 10000 (\d -> d.deliveredAt)
1162− yield ackedFed
11631169
11641170-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.
11651171-- Records or clears federation-failure counters per call.
@@ -1189,12 +1195,10 @@ broadcastOfflineGossip = \t expiredLocal -> do
11891195 when (federationEnabled && count expiredLocal > 0) (sendBatchedGossip (knownPeerServersFromPresence presence) (pluck expiredLocal (\p -> mkGossip (GossipOffline {}) p.pubkey t offlineGossipMinTtl)))
11901196
11911197-- ============================================
1192−-- Forward retries and acks
1198+-- Forward retries
11931199-- ============================================
11941200-- `retryPendingForwards` walks queued cross-server forwards in batches per
11951201-- destination, increments retry counts, and gives up after `maxForwardRetries`.
1196−-- `sendAcksAndClean` flushes federated acks back to origin servers and prunes
1197−-- the local `ackedIds` rows once the remote confirms.
11981202-- `forwardQueuedToServer` is the catch-up path triggered when a remote key
11991203-- comes online: any messages already buffered for that pubkey are enqueued
12001204-- as forwards to the newly-known server.
@@ -1215,7 +1219,6 @@ retryPendingForwards = do
12151219 f.toKey,
12161220 f.encryptedBlob,
12171221 f.timestamp,
1218− f.originServer,
12191222 f.messageId,
12201223 fromServer: serverHostN,
12211224 requestTimestamp: t2,
@@ -1242,36 +1245,12 @@ retryPendingForwards = do
12421245 x.toKey,
12431246 x.encryptedBlob,
12441247 x.timestamp,
1245− x.originServer,
12461248 retries: x.retries + 1,
12471249 x.expiresAt,
12481250 x.senderKey
12491251 }
12501252 else x))))
12511253
1252−-- Flush federated ack receipts back to each origin server in one call, then
1253−-- prune the locally-recorded acked entries the peer confirmed.
1254−sendAcksAndClean : [{messageId: MessageIdHex, originServer: Text | a}] -> IO {} {}
1255−sendAcksAndClean = \pendingAcks -> forEach (collectPeers pendingAcks (\a -> a.originServer)) (\server -> do
1256− let mids = mapWhere pendingAcks (\a -> a.originServer == server) (\a -> a.messageId)
1257− t <- now
1258− auth <- generateFederationAuth "ack" t
1259− result <- fetch (federationUrl server) (FederationAck {
1260− authorization: auth,
1261− messageIds: mids,
1262− originServer: server,
1263− fromServer: serverHostN,
1264− requestTimestamp: t,
1265− correlationId: ""
1266− })
1267− case result of
1268− Err _ -> recordFailure server
1269− Ok _ -> do
1270− clearFailure server
1271− atomic (do
1272− aids <- *ackedIds
1273− *ackedIds = filter aids (\a -> not (a.originServer == server && elem mids a.messageId))))
1274−
12751254-- Forward queued messages to a server when a remote key comes online.
12761255forwardQueuedToServer : PubkeyHex -> ServerName -> Int<Ms> -> IO {rw *forwards, r *messages} {}
12771256forwardQueuedToServer = \pubkey server t -> atomic (do
@@ -1285,7 +1264,6 @@ forwardQueuedToServer = \pubkey server t -> atomic (do
12851264 m.toKey,
12861265 m.encryptedBlob,
12871266 m.timestamp,
1288− originServer: if m.originServer == "" then serverHostN else m.originServer,
12891267 retries: 0,
12901268 m.expiresAt,
12911269 m.senderKey
@@ -1300,16 +1278,14 @@ forwardQueuedToServer = \pubkey server t -> atomic (do
13001278-- local presence so peers don't see us go offline mid-session.
13011279
13021280-- One composite background tick: prune expired state, gossip offline for
1303−-- locally-expired keys, retry pending forwards, flush federation acks, and
1304−-- (occasionally) refresh local presence so peers don't see us drop off.
1281+-- locally-expired keys, retry pending forwards, and (occasionally) refresh
1282+-- local presence so peers don't see us drop off.
13051283pruneAndRetry = \t -> do
13061284 *lastPrunedAt = t
13071285 expiredLocal <- pruneExpired t
13081286 fork (broadcastOfflineGossip t expiredLocal)
13091287 when federationEnabled (do
13101288 fork retryPendingForwards
1311− aids <- *ackedIds
1312− fork (sendAcksAndClean (filter aids (\a -> isRemote a.originServer)))
13131289 lastRefresh <- *lastPresenceRefresh
13141290 when (t - lastRefresh >= presenceRefreshInterval) (do
13151291 *lastPresenceRefresh = t
@@ -1513,9 +1489,9 @@ pollOnce : Text -> Int -> Bool -> Text -> Int -> ({authorized: Bool, cursor: Tim
15131489pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t -> do
15141490 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
15151491 let clampedCursor = clamp 0 t cursor
1516− pendingAcks <- if clampedCursor > 0 then ackPolledMessages pubkey clampedCursor t else yield []
1492+ when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor t)
15171493 startWakeup <- *pollWakeup
1518− yield (Just {value: {pubkey, clampedCursor, startWakeup, pendingAcks}}))
1494+ yield (Just {value: {pubkey, clampedCursor, startWakeup}}))
15191495 case setup of
15201496 Nothing {} -> yield (respond (emptyPollResp 0 (False {}) (False {}) (False {}) serverGen 0 cursorWasReset))
15211497 Just {value: s} -> do
@@ -1531,7 +1507,6 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
15311507 else do
15321508 pollResult <- doPoll s t
15331509 atomic (do aps <- *activePolls; *activePolls = filter aps (\p -> p.pubkey != s.pubkey))
1534− when federationEnabled (fork (sendAcksAndClean s.pendingAcks))
15351510 extras <- atomic (do
15361511 let mine = \x -> x.senderKey == s.pubkey
15371512 seItems <- *senderExpirations
@@ -1562,7 +1537,7 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
15621537
15631538-- One pass: collect events at `t`. If empty, retry-wait on `pollWakeup`
15641539-- (bumped by send/heartbeat) and re-collect once at the new wall time.
1565−doPoll : {clampedCursor: Int, pendingAcks: [AckedEntry], pubkey: PubkeyHex, startWakeup: Timestamp} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{id: MessageIdHex, payload: {encryptedBlob: BlobHex, expiresAt: Timestamp, timestamp: Timestamp, toKey: PubkeyHex}}], hasMore: Bool}
1540+doPoll : {clampedCursor: Int, pubkey: PubkeyHex, startWakeup: Timestamp} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{id: MessageIdHex, payload: {encryptedBlob: BlobHex, expiresAt: Timestamp, timestamp: Timestamp, toKey: PubkeyHex}}], hasMore: Bool}
15661541doPoll = \s t -> do
15671542 let collectAt = \tnow -> atomic (do
15681543 msgs <- *messages
@@ -1581,9 +1556,9 @@ collectPollEvents : [Message] -> PubkeyHex -> Timestamp -> Timestamp -> {events:
15811556collectPollEvents = \messages pubkey cursor t -> (\myMsgs -> (\limited -> {events: pluck limited (\m -> {id: m.messageId, payload: {m.toKey, m.encryptedBlob, m.timestamp, m.expiresAt}}), cursor: fold (\acc m -> max acc m.receivedAt) cursor limited, hasMore: count myMsgs > count limited}) (takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs))) (filter messages (\m -> m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor))
15821557
15831558-- Treat the supplied cursor as an ack: drop all messages with
1584−-- receivedAt <= cursor, plus their pending forwards, and record federated
1585−-- acks/sender deliveries via `recordAcksAndDeliveries`.
1586−ackPolledMessages : PubkeyHex -> Int -> Int<Ms> -> IO {rw *ackedIds, rw *forwards, rw *messages, rw *senderDeliveries} [{ackedAt: Timestamp, messageId: MessageIdHex, originServer: ServerName, retries: Int, toKey: PubkeyHex}]
1559+-- receivedAt <= cursor, plus their pending forwards, and record sender-side
1560+-- deliveries for any locally-originated messages via `recordSenderDeliveries`.
1561+ackPolledMessages : PubkeyHex -> Int -> Int<Ms> -> IO {rw *forwards, rw *messages, rw *senderDeliveries} {}
15871562ackPolledMessages = \pubkey cursor t -> do
15881563 msgs <- *messages
15891564 let matches = \m -> m.toKey == pubkey && m.receivedAt <= cursor
@@ -1592,7 +1567,7 @@ ackPolledMessages = \pubkey cursor t -> do
15921567 *messages = removeWhere msgs matches
15931568 fwds <- *forwards
15941569 *forwards = filter fwds (\f -> not (f.toKey == pubkey && elem mids f.messageId))
1595− recordAcksAndDeliveries toAck t
1570+ recordSenderDeliveries toAck t
15961571
15971572-- ============================================
15981573-- Handler: send
@@ -1642,10 +1617,9 @@ handleSendMessage = \to encryptedBlob timestamp expiresInMs authorization client
16421617-- `seqCounter`, appends the message, and pre-stages forwards for any remote
16431618-- presence rows that already exist for `to`. Returns the chosen status and
16441619-- the list of remote servers the caller should kick forwards to.
1645−sendMessageCore : PubkeyHex -> PubkeyHex -> BlobHex -> Int<Ms> -> MessageIdHex -> Int -> Int<Ms> -> IO {r *ackedIds, r *deliveredIds, rw *forwards, rw *messages, rw *pairRates, rw *pollWakeup, r *presence, rw *recvRates, rw *sendRates, rw *seqCounter} {remoteServers: [Text], status: MessageStatus}
1620+sendMessageCore : PubkeyHex -> PubkeyHex -> BlobHex -> Int<Ms> -> MessageIdHex -> Int -> Int<Ms> -> IO {r *deliveredIds, rw *forwards, rw *messages, rw *pairRates, rw *pollWakeup, r *presence, rw *recvRates, rw *sendRates, rw *seqCounter} {remoteServers: [Text], status: MessageStatus}
16461621sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs t -> do
16471622 msgs <- *messages
1648− aids <- *ackedIds
16491623 dids <- *deliveredIds
16501624 sRates <- *sendRates
16511625 let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t
@@ -1662,7 +1636,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs t -
16621636 let reject = \s -> yield {status: s, remoteServers: []}
16631637 if senderKey == to
16641638 then reject (MsRejected {})
1665− else if isDuplicateMessage effMsgId to msgs aids dids
1639+ else if isDuplicateMessage effMsgId to msgs dids
16661640 then reject (MsDuplicate {})
16671641 else if srLim || prLim || rrLim
16681642 then reject (MsRateLimited {})
@@ -1678,7 +1652,6 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs t -
16781652 toKey: to,
16791653 encryptedBlob,
16801654 timestamp: clampedTs,
1681− originServer: "",
16821655 receivedAt: seqTs,
16831656 expiresAt,
16841657 senderKey
@@ -1697,7 +1670,6 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs t -
16971670 toKey: to,
16981671 encryptedBlob,
16991672 timestamp: clampedTs,
1700− originServer: serverHostN,
17011673 retries: 0,
17021674 expiresAt,
17031675 senderKey
@@ -1713,8 +1685,11 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs t -
17131685 }
17141686
17151687-- Fire one /federation/forward call per pre-staged server. On success the
1716−-- corresponding row in `forwards` is pruned (no retry needed); on failure
1717−-- it stays so `retryPendingForwards` can re-attempt with backoff.
1688+-- corresponding row in `forwards` is pruned, and if no other forwards remain
1689+-- for this messageId AND the recipient has no local presence, the backing
1690+-- `*messages` row is dropped too: the origin's job is done and the message
1691+-- now lives on the receiving peer's mailbox. On failure the forwards row
1692+-- stays so `retryPendingForwards` can re-attempt with backoff.
17181693forkForwardsTo : MessageIdHex -> PubkeyHex -> BlobHex -> Int<Ms> -> [Text] -> IO {} {}
17191694forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers -> forEach servers (\server -> do
17201695 fwds <- *forwards
@@ -1728,7 +1703,6 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers -> forEach servers
17281703 toKey: to,
17291704 encryptedBlob,
17301705 timestamp: clampedTs,
1731− originServer: serverHostN,
17321706 messageId: effMsgId,
17331707 fromServer: serverHostN,
17341708 requestTimestamp: t,
@@ -1740,7 +1714,16 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers -> forEach servers
17401714 Err _ -> recordFailure server
17411715 Ok {value: r} -> when r.ok (do
17421716 clearFailure server
1743− atomic (do fwds2 <- *forwards; *forwards = filter fwds2 (\x -> not same x))))
1717+ atomic (do
1718+ fwds2 <- *forwards
1719+ let remaining = filter fwds2 (\x -> not same x)
1720+ replace *forwards = remaining
1721+ pres <- *presence
1722+ let stillLocal = any pres (\p -> p.pubkey == to && isLocalServer p.server && p.expiresAt > t)
1723+ let othersPending = any remaining (\x -> x.messageId == effMsgId)
1724+ when (not stillLocal && not othersPending) (do
1725+ msgs <- *messages
1726+ *messages = filter msgs (\m -> m.messageId != effMsgId)))))
17441727
17451728-- ============================================
17461729-- Handler: lookup
@@ -1793,14 +1776,13 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
17931776 gateAuth (False {}) authorization clientIp t respond mkErr (\_ -> do
17941777 msgs <- *messages
17951778 fwds <- *forwards
1796− aids <- *ackedIds
17971779 dids <- *deliveredIds
17981780 let matches = \x -> x.messageId == messageId && x.toKey == to
17991781 let s = if any msgs matches
18001782 then "pending"
18011783 else if any fwds matches
18021784 then "forwarding"
1803− else if any aids matches || any dids matches then "delivered" else "unknown"
1785+ else if any dids matches then "delivered" else "unknown"
18041786 yield (respond {status: s, authorized: True {}}))
18051787
18061788-- ============================================
@@ -1809,8 +1791,8 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
18091791-- Mailbox mutation by the recipient (ack, list, delete).
18101792
18111793-- POST /messages/ack: explicit ack by message id (alternative to the implicit
1812−-- cursor-based ack in /poll). Drops mailbox+forward rows and forks federation
1813−-- ack flush for any messages that originated remotely.
1794+-- cursor-based ack in /poll). Drops mailbox+forward rows and records
1795+-- sender-side delivery for any locally-originated messages.
18141796handleAckMessages : [MessageIdHex] -> Text -> Text -> ({ackedCount: Int, authorized: Bool, error: Text, ok: Bool} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
18151797handleAckMessages = \messageIds authorization clientIp respond -> do
18161798 t <- now
@@ -1821,12 +1803,11 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
18211803 ackedCount: 0,
18221804 error: "rate_limited"
18231805 } (\pubkey -> do
1824− result <- atomic (do
1806+ ackedCount <- atomic (do
18251807 toAck <- removeForKey (\x -> x.toKey == pubkey && elem (dedupMessageIds messageIds) x.messageId)
1826− ackedFed <- recordAcksAndDeliveries toAck t
1827− yield {ackedCount: count toAck, pendingAcks: ackedFed})
1828− when federationEnabled (fork (sendAcksAndClean result.pendingAcks))
1829− yield (respond {ok: True {}, authorized: True {}, result.ackedCount, error: ""}))
1808+ recordSenderDeliveries toAck t
1809+ yield (count toAck))
1810+ yield (respond {ok: True {}, authorized: True {}, ackedCount, error: ""}))
18301811
18311812-- POST /messages/list: paginated peek at the caller's mailbox without acking.
18321813-- Returns metadata only (no ciphertext) plus the total count.
@@ -1875,14 +1856,10 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
18751856 deletedCount: 0,
18761857 error: "rate_limited"
18771858 } (\pubkey -> do
1878− result <- atomic (do
1859+ deletedCount <- atomic (do
18791860 toDel <- removeForKey (\x -> x.toKey == pubkey && elem (dedupMessageIds messageIds) x.messageId)
1880− let ackedFed = mapWhere toDel (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
1881− aids <- *ackedIds
1882− *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)
1883− yield {deletedCount: count toDel, pendingAcks: ackedFed})
1884− when federationEnabled (fork (sendAcksAndClean result.pendingAcks))
1885− yield (respond {ok: True {}, authorized: True {}, result.deletedCount, error: ""}))
1861+ yield (count toDel))
1862+ yield (respond {ok: True {}, authorized: True {}, deletedCount, error: ""}))
18861863
18871864-- POST /messages/batch: multi-recipient send in one call. Each entry runs
18881865-- through `sendMessageCore` separately so per-entry rate limits/errors don't
@@ -1910,8 +1887,6 @@ handleSendBatch = \messages authorization clientIp respond -> do
19101887-- `withFederationAuth` which checks: peer block/allow lists, request
19111888-- precondition (e.g. expected origin/server constraints), clock skew vs the
19121889-- request timestamp, Ed25519 signature auth, and per-peer rate limit.
1913−-- Receive paths never trust `originServer` for routing; they only use it for
1914−-- ack bookkeeping and SSRF-style filtering.
19151890
19161891-- GET /federation/pubkey: publish the Ed25519 public key peers should use to
19171892-- verify our `Signed ...` federation auth headers. Returns ok=False when the
@@ -1962,15 +1937,20 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
19621937-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.
19631938-- Returns "" on success (including duplicates), otherwise an error code:
19641939-- "no_presence" | "rate_limited" (only when checkRecvRate).
1965−storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> Text -> MessageIdHex -> Int<Ms> -> Bool -> Int<Ms> -> IO {r *ackedIds, r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter} Text
1966−storeFederated = \toKey encryptedBlob timestamp originServer messageId expiresAt checkRecvRate t -> do
1940+--
1941+-- The presence gate requires the recipient to be **currently locally online**
1942+-- (not just visible to gossip somewhere). Without that, an A → B forward based
1943+-- on stale gossip about R could land on B with no actively-polling client and
1944+-- get stuck. If B refuses, A retries via `retryPendingForwards` against the
1945+-- next presence target.
1946+storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> MessageIdHex -> Int<Ms> -> Bool -> Int<Ms> -> IO {r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter} Text
1947+storeFederated = \toKey encryptedBlob timestamp messageId expiresAt checkRecvRate t -> do
19671948 pres <- *presence
19681949 msgs <- *messages
1969− aids <- *ackedIds
19701950 dids <- *deliveredIds
1971− if not any pres (\p -> p.pubkey == toKey && p.expiresAt > t)
1951+ if not any pres (\p -> p.pubkey == toKey && isLocalServer p.server && p.expiresAt > t)
19721952 then yield "no_presence"
1973− else if isDuplicateMessage messageId toKey msgs aids dids
1953+ else if isDuplicateMessage messageId toKey msgs dids
19741954 then yield ""
19751955 else do
19761956 limited <- if checkRecvRate
@@ -1995,7 +1975,6 @@ storeFederated = \toKey encryptedBlob timestamp originServer messageId expiresAt
19951975 toKey,
19961976 encryptedBlob,
19971977 timestamp: clampTimestamp timestamp t,
1998− originServer,
19991978 receivedAt: seqTs,
20001979 expiresAt: cappedExpiry,
20011980 senderKey: ""
@@ -2007,95 +1986,33 @@ storeFederated = \toKey encryptedBlob timestamp originServer messageId expiresAt
20071986-- POST /federation/forward: receive a single federated message destined for
20081987-- a local key. Delegates the actual store decision to `storeFederated`,
20091988-- which returns "" on success (including the duplicate case).
2010−handleForwardMessage : PubkeyHex -> BlobHex -> Int<Ms> -> Text -> Text -> Text -> Int -> Int<Ms> -> Text -> ({error: Text, ok: Bool, retryAfterMs: Int} -> a) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, r *ackedIds, r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter, clock, network} a
2011−handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp expiresAt authorization respond -> do
1989+handleForwardMessage : PubkeyHex -> BlobHex -> Int<Ms> -> Text -> Text -> Int -> Int<Ms> -> Text -> ({error: Text, ok: Bool, retryAfterMs: Int} -> a) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter, clock, network} a
1990+handleForwardMessage = \toKey encryptedBlob timestamp messageId fromServer requestTimestamp expiresAt authorization respond -> do
20121991 t <- now
2013− withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (isRemote originServer && isRemote fromServer) (do
1992+ withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (isRemote fromServer) (do
20141993 if t - timestamp > messageMaxAge
20151994 then yield (respond {ok: False {}, error: "stale_timestamp", retryAfterMs: 0})
20161995 else do
2017− result <- atomic storeFederated toKey encryptedBlob timestamp originServer messageId expiresAt (True {}) t
1996+ result <- atomic storeFederated toKey encryptedBlob timestamp messageId expiresAt (True {}) t
20181997 yield (respond {ok: result == "", error: result, retryAfterMs: 0}))
20191998
2020−-- POST /federation/ack: peer reports it has acked messages we forwarded to it.
2021−-- We drop the matching `forwards` rows; if the recipient is no longer locally
2022−-- live and no other forward is still pending for the same id, also reap the
2023−-- backing `messages` row (after `ackGracePeriod` to dampen race with /poll).
2024−handleFederationAck : [MessageIdHex] -> Text -> Text -> Int -> Text -> ({ackedCount: Int, error: Text, ok: Bool, results: [{found: Bool, messageId: MessageIdHex}], retryAfterMs: Int} -> a) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, rw *forwards, rw *messages, r *presence, clock, network} a
2025−handleFederationAck = \messageIds originServer fromServer requestTimestamp authorization respond -> do
2026− t <- now
2027− let ackRespond = \r -> respond {r.ok, r.error, ackedCount: 0, results: [], r.retryAfterMs}
2028− withFederationAuth authorization t ackRespond fromServer "ack" requestTimestamp (isLocalServer originServer && isRemote fromServer) (do
2029− let dedupedIds = dedupMessageIds messageIds
2030− result <- atomic (do
2031− fwds <- *forwards
2032− let isAcked = \f -> f.toServer == fromServer && elem dedupedIds f.messageId
2033− let removed = filter fwds isAcked
2034− let remainingFwds = removeWhere fwds isAcked
2035− replace *forwards = remainingFwds
2036− msgs <- *messages
2037− let removedIds = pluck removed (\f -> f.messageId)
2038− pres <- *presence
2039− let toReap = filter msgs (\m -> elem removedIds m.messageId && t - m.receivedAt >= ackGracePeriod && none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t) && none remainingFwds (\f -> f.messageId == m.messageId && f.toKey == m.toKey))
2040− *messages = removeWhere msgs (\m -> hasMessage m.messageId m.toKey toReap)
2041− yield {
2042− ackedCount: count removed,
2043− results: pluck dedupedIds (\mid -> {
2044− messageId: mid,
2045− found: any removed (\f -> f.messageId == mid)
2046− })
2047− })
2048− yield (respond {ok: True {}, error: "", result.ackedCount, result.results, retryAfterMs: 0}))
2049−
20501999-- POST /federation/forward-batch: bulk variant of `handleForwardMessage`.
20512000-- Skips per-entry recv-rate charging (the per-server federation rate limit
20522001-- already gates the call as a whole).
2053−handleFederationBatchForward : [{encryptedBlob: BlobHex, expiresAt: Int<Ms>, messageId: MessageIdHex, originServer: Text, timestamp: Int<Ms>, toKey: PubkeyHex | a}] -> Text -> Int -> Text -> ({error: Text, ok: Bool, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} -> b) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, clock, network} b
2002+handleFederationBatchForward : [{encryptedBlob: BlobHex, expiresAt: Int<Ms>, messageId: MessageIdHex, timestamp: Int<Ms>, toKey: PubkeyHex | a}] -> Text -> Int -> Text -> ({error: Text, ok: Bool, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} -> b) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, clock, network} b
20542003handleFederationBatchForward = \forwards fromServer requestTimestamp authorization respond -> do
20552004 t <- now
20562005 let batchRespond = \r -> respond {r.ok, r.error, results: [], r.retryAfterMs}
20572006 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (isRemote fromServer && count forwards <= maxBatchSize) (do
20582007 results <- mapEach (takeRelation maxBatchSize forwards) (\e -> do
2059− ok <- if isLocalServer e.originServer || t - e.timestamp > messageMaxAge
2008+ ok <- if t - e.timestamp > messageMaxAge
20602009 then yield (False {})
20612010 else do
2062− result <- atomic storeFederated e.toKey e.encryptedBlob e.timestamp e.originServer e.messageId e.expiresAt (False {}) t
2011+ result <- atomic storeFederated e.toKey e.encryptedBlob e.timestamp e.messageId e.expiresAt (False {}) t
20632012 yield (result == "")
20642013 yield {e.messageId, ok})
20652014 yield (respond {ok: True {}, error: "", results, retryAfterMs: 0}))
20662015
2067−-- POST /federation/delivery-receipt: peer confirms delivery of messages whose
2068−-- sender is local to us; we record per-sender delivery rows so that sender's
2069−-- next /poll can include the deliveredAt timestamp.
2070−handleFederationDeliveryReceipt : [MessageIdHex] -> [PubkeyHex] -> Text -> Int -> Text -> ({error: Text, ok: Bool, receivedCount: Int, retryAfterMs: Int} -> a) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, r *messages, rw *senderDeliveries, clock, network} a
2071−handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp authorization respond -> do
2072− t <- now
2073− let receiptRespond = \r -> respond {r.ok, r.error, receivedCount: 0, r.retryAfterMs}
2074− withFederationAuth authorization t receiptRespond fromServer "delivery-receipt" requestTimestamp (isRemote fromServer && count messageIds <= maxMessageIds) (do
2075− if count messageIds != count toKeys
2076− then yield (respond {
2077− ok: False {},
2078− error: "mismatched_lengths",
2079− receivedCount: 0,
2080− retryAfterMs: 0
2081− })
2082− else do
2083− let pairs = do
2084− mp <- withIndex messageIds
2085− kp <- withIndex toKeys
2086− where mp.i == kp.i
2087− yield {messageId: mp.v, toKey: kp.v}
2088− atomic (do
2089− msgs <- *messages
2090− sds <- *senderDeliveries
2091− let newDels = do
2092− p <- pairs
2093− m <- msgs
2094− where m.messageId == p.messageId && m.toKey == p.toKey && m.senderKey != "" && m.originServer == ""
2095− yield {m.senderKey, p.messageId, p.toKey, deliveredAt: t}
2096− *senderDeliveries = capList (union sds newDels) 10000 (\d -> d.deliveredAt))
2097− yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
2098−
20992016-- ============================================
21002017-- Maintenance handlers
21012018-- ============================================
@@ -2148,7 +2065,7 @@ handleShutdown = \authorization respond -> handleDrain 2 authorization (\r -> re
21482065-- POST /maintenance/prune: trigger an out-of-band `pruneAndRetry` tick.
21492066-- Useful when the operator wants to force expiry/forward retries without
21502067-- waiting for the next `backgroundPrune` interval.
2151−handlePrune : Text -> ({ok: Bool} -> a) -> IO {r *runtimeMaintenanceSecret, rw *ackedIds, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *senderDeliveries, rw *senderExpirations, rw *sessions, clock} a
2068+handlePrune : Text -> ({ok: Bool} -> a) -> IO {r *runtimeMaintenanceSecret, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *senderDeliveries, rw *senderExpirations, rw *sessions, clock} a
21522069handlePrune = \authorization respond -> do
21532070 t <- now
21542071 authorized <- verifyMaintenanceAuth authorization
@@ -2319,14 +2236,13 @@ handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth autho
23192236 yield (respond {ok: True {}, revokedCount: revoked}))
23202237
23212238-- POST /maintenance/messages/purge: drop all queued/forwarding messages
2322−-- destined for `pubkey`, recording federation acks for any that originated
2323−-- elsewhere so peers learn we have nothing for the key.
2324−handleAdminPurgeMessages : PubkeyHex -> Text -> ({ok: Bool, purgedCount: Int} -> a) -> IO {rw *ackedIds, rw *forwards, rw *messages, r *runtimeMaintenanceSecret, clock} a
2239+-- destined for `pubkey`. With the federation ack flow gone, there's nothing
2240+-- to ack back — peers will discover the absence on their next forward attempt.
2241+handleAdminPurgeMessages : PubkeyHex -> Text -> ({ok: Bool, purgedCount: Int} -> a) -> IO {rw *forwards, rw *messages, r *runtimeMaintenanceSecret} a
23252242handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth authorization (yield (respond {
23262243 ok: False {},
23272244 purgedCount: 0
23282245})) (do
2329− t <- now
23302246 purged <- atomic (do
23312247 msgs <- *messages
23322248 let mine = \m -> m.toKey == pubkey
@@ -2334,8 +2250,6 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
23342250 *messages = removeWhere msgs mine
23352251 fwds <- *forwards
23362252 *forwards = filter fwds (\f -> f.toKey != pubkey)
2337− aids <- *ackedIds
2338− *ackedIds = capList (union aids (mapWhere purgedMsgs (\m -> m.originServer != "") (\m -> mkAckedEntry m t))) maxAckedIds (\a -> a.ackedAt)
23392253 yield (count purgedMsgs))
23402254 yield (respond {ok: True {}, purgedCount: purged}))
23412255
@@ -2433,7 +2347,7 @@ handleGetConfig = \authorization respond -> do
24332347-- GET /stats: capacity gauges + minimal federation/timing snapshot. Most of
24342348-- the historical metric fields are zero-filled (see `zeroStats`) — only the
24352349-- counts that come straight from the state refs are populated.
2436−handleStats : Text -> ({capacity: {ackOverflow: Int, ackOverflowMax: Int, ackedIds: Int, ackedIdsMax: Int, activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, senderDeliveries: Int, senderExpirations: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text} -> a) -> IO {r *ackedIds, r *activePolls, r *blockedPeers, r *challenges, r *deliveredIds, r *drainPhase, r *forwards, r *heartbeatCount, r *lastHeartbeatAt, r *lastPrunedAt, r *messages, r *presence, r *pruneCount, r *runtimeMaintenanceSecret, r *sessions, r *startedAt, clock} a
2350+handleStats : Text -> ({capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, senderDeliveries: Int, senderExpirations: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text} -> a) -> IO {r *activePolls, r *blockedPeers, r *challenges, r *deliveredIds, r *drainPhase, r *forwards, r *heartbeatCount, r *lastHeartbeatAt, r *lastPrunedAt, r *messages, r *presence, r *pruneCount, r *runtimeMaintenanceSecret, r *sessions, r *startedAt, clock} a
24372351handleStats = \authorization respond -> withMaintAuth authorization (yield (respond zeroStats)) (do
24382352 t <- now
24392353 sa <- *startedAt
@@ -2441,7 +2355,6 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
24412355 msgs <- *messages
24422356 pres <- *presence
24432357 fwds <- *forwards
2444− aids <- *ackedIds
24452358 dids <- *deliveredIds
24462359 aps <- *activePolls
24472360 chs <- *challenges
@@ -2468,10 +2381,6 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
24682381 forwards: fCount,
24692382 challenges: count chs,
24702383 challengesMax: maxChallenges,
2471− ackedIds: count aids,
2472− ackedIdsMax: maxAckedIds,
2473− ackOverflow: 0,
2474− ackOverflowMax: 0,
24752384 deliveredIds: count dids,
24762385 deliveredIdsMax: maxDeliveredIds,
24772386 forwardNonces: 0,
@@ -2555,7 +2464,7 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
25552464-- All-zero /stats payload returned to unauthorized callers. The shape is kept
25562465-- in lockstep with the response in `handleStats` so older operator tooling
25572466-- never sees a missing field.
2558−zeroStats : {capacity: {ackOverflow: Int, ackOverflowMax: Int, ackedIds: Int, ackedIdsMax: Int, activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, senderDeliveries: Int, senderExpirations: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text}
2467+zeroStats : {capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, senderDeliveries: Int, senderExpirations: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text}
25592468zeroStats = {
25602469 ok: False {},
25612470 version: serverBuildVersion,
@@ -2569,10 +2478,6 @@ zeroStats = {
25692478 forwards: 0,
25702479 challenges: 0,
25712480 challengesMax: 0,
2572− ackedIds: 0,
2573− ackedIdsMax: 0,
2574− ackOverflow: 0,
2575− ackOverflowMax: 0,
25762481 deliveredIds: 0,
25772482 deliveredIdsMax: 0,
25782483 forwardNonces: 0,
@@ -2816,10 +2721,8 @@ serve = \req -> case req of
28162721 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
28172722 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
28182723 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
2819− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, expiresAt, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp expiresAt authorization respond
2820− FederationAck {messageIds, originServer, fromServer, requestTimestamp, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp authorization respond
2724+ ForwardMessage {toKey, encryptedBlob, timestamp, messageId, fromServer, requestTimestamp, expiresAt, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp messageId fromServer requestTimestamp expiresAt authorization respond
28212725 FederationBatchForward {forwards, fromServer, requestTimestamp, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp authorization respond
2822− FederationDeliveryReceipt {messageIds, toKeys, fromServer, requestTimestamp, authorization, respond} -> handleFederationDeliveryReceipt messageIds toKeys fromServer requestTimestamp authorization respond
28232726 FederationPubkey {respond} -> handleFederationPubkey respond
28242727 Prune {authorization, respond} -> handlePrune authorization respond
28252728 Drain {phase, authorization, respond} -> handleDrain phase authorization respond
@@ -2863,7 +2766,7 @@ serve = \req -> case req of
28632766
28642767-- Endless tick: every `backgroundPruneInterval` run `pruneAndRetry` (skipped
28652768-- when drain phase >= 2) and bump `pruneCount` for /stats and /health.
2866−backgroundPrune : IO {r *drainPhase, rw *ackedIds, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *pruneCount, rw *senderDeliveries, rw *senderExpirations, rw *sessions, clock} a
2769+backgroundPrune : IO {r *drainPhase, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *pruneCount, rw *senderDeliveries, rw *senderExpirations, rw *sessions, clock} a
28672770backgroundPrune = do
28682771 sleep backgroundPruneInterval
28692772 dp <- *drainPhase
@@ -2903,9 +2806,8 @@ main = do
29032806 *federationPrivateKeyHex = bytesToHex kp.privateKey
29042807 *federationPublicKeyHex = bytesToHex kp.publicKey)
29052808 logInfo "Generated federation keypair on first startup")
2906− logInfo ("Skrepka relay server starting on port " ++ show listenPort ++ "...")
2809+ logInfo ("Skrepka relay server starting on " ++ bindHost ++ ":" ++ show listenPort ++ " ...")
29072810 logInfo (" federation: " ++ (if federationEnabled then "enabled" else "disabled"))
29082811 fork backgroundPrune
29092812 fork pollHeartbeat
2910− listen listenPort serve
2911− listen listenPort serve
2813+ listenOn bindHost listenPort serve