@@ -36,8 +36,8 @@ federationScheme = "https://" -- scheme for federation HTTP calls
36-- nginx: client_max_body_size 256k;
36-- nginx: client_max_body_size 256k;
37-- HAProxy: tune.maxrewrite 256k
37-- HAProxy: tune.maxrewrite 256k
38-- Without this, the server is vulnerable to large-payload DoS attacks. The maxBlobLen
38-- Without this, the server is vulnerable to large-payload DoS attacks. The maxBlobLen
39−-- limit (64KB hex = 32KB binary) bounds legitimate message payloads; snapshots and
39+-- limit (64KB hex = 32KB binary) bounds legitimate message payloads; batch requests
40−-- batch requests may be larger but are authenticated via maintenance secret.
40+-- may be larger but are authenticated via maintenance secret.
41-- TODO: If Knot adds HTTP-layer configuration, enforce a request body size limit natively.
41-- TODO: If Knot adds HTTP-layer configuration, enforce a request body size limit natively.
42messageMaxAge : Int<Ms>
42messageMaxAge : Int<Ms>
43messageMaxAge = 2592000000 -- 30 days
43messageMaxAge = 2592000000 -- 30 days
@@ -182,9 +182,6 @@ reconciliationDriftThreshold = 10
182federationLatencyWindowMs : Int<Ms>
182federationLatencyWindowMs : Int<Ms>
183federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
183federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
185−-- Snapshot format version (increment on schema changes to prevent corrupt restores)
186−snapshotVersion = 14 -- bumped for rate limit override persistence in snapshots
188-- Server build version: included in /health and /stats responses for deployment verification.
185-- Server build version: included in /health and /stats responses for deployment verification.
189-- Update this string when deploying a new build so operators can confirm which version is running.
186-- Update this string when deploying a new build so operators can confirm which version is running.
190serverBuildVersion = "2026-04-26c"
187serverBuildVersion = "2026-04-26c"
@@ -195,33 +192,6 @@ serverBuildVersion = "2026-04-26c"
195-- drops unknown keys). 0 = pre-versioning peer (field missing or unset).
192-- drops unknown keys). 0 = pre-versioning peer (field missing or unset).
196federationProtocolVersion = 12
193federationProtocolVersion = 12
198−-- Snapshot checksum: deterministic hash of entry counts for integrity verification.
199−-- Uses multiplicative hash to distinguish permutations of the same counts
200−-- (e.g., {presence:1, ackedIds:0} vs {presence:0, ackedIds:1}).
201−-- Not cryptographic — catches truncation/corruption, not tampering.
202−snapshotChecksum : Int -> Int -> Int -> Int -> Int -> Int
203−snapshotChecksum = \presenceCount ackedCount deliveredCount forwardCount nonceCount ->
204− presenceCount * 100003 + ackedCount * 100019 + deliveredCount * 100043 + forwardCount * 100049 + nonceCount * 100057
206−-- Snapshot integrity tag: combines checksum with maintenance secret to detect tampering.
207−-- Not cryptographic (no HMAC primitive in Knot), but uses a double-hash scheme with
208−-- independent seeds and multipliers to resist inversion attacks. An attacker who
209−-- captures one valid (snapshot, tag) pair cannot derive the secretHash components
210−-- independently because the final mixing step entangles them non-linearly.
211−-- Returns 0 when no maintenance secret is configured (no integrity protection).
212−-- SECURITY NOTE: This is NOT a substitute for cryptographic signing. The hash constants
213−-- are public (in this source), so an attacker with access to the maintenance secret can
214−-- forge valid tags. For high-security deployments, GPG-sign or HMAC snapshot lines
215−-- externally (e.g., via a sidecar that signs [SNAPSHOT_DATA] lines before archiving).
216−-- TODO: Replace with HMAC when Knot adds a cryptographic hash primitive.
217−snapshotIntegrityTag : Int -> Text -> Int
218−snapshotIntegrityTag = \checksum secret ->
219− if secret == "" then 0
220− else let secretChars = chars (bytesToHex (textToBytes secret))
221− in let hash1 = fold (\acc c -> acc * 31 + hexCharIndex c) 7 secretChars
222− in let hash2 = fold (\acc c -> acc * 37 + hexCharIndex c) 13 secretChars
223− in (checksum * 100069 + hash1 * 100081) * 100003 + (checksum * 100057 + hash2 * 100019)
225-- Max seenBy list length in gossip events (bounds memory and bandwidth per event)
195-- Max seenBy list length in gossip events (bounds memory and bandwidth per event)
226maxSeenByLen = 20
196maxSeenByLen = 20
@@ -243,21 +213,6 @@ drainTimeoutMs = 120000 -- 2 minutes; covers forward retries + ack sends + gossi
243-- Background task intervals
213-- Background task intervals
244backgroundPruneInterval : Int<Ms>
214backgroundPruneInterval : Int<Ms>
245backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
215backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
246−autoSnapshotInterval : Int<Ms>
247−autoSnapshotInterval = 300000 -- 5 minutes; periodic state summary to stdout (0 to disable)
248−-- Full snapshot auto-export interval: emit [SNAPSHOT_DATA] JSON to stdout for sidecar capture.
249−-- 0 to disable. When enabled, a crash loses at most this interval of federation state.
250−-- SECURITY: auto-exported snapshots contain unencrypted federation state (presence, acked
251−-- message IDs, peer failure records, forward nonces). The data does not include message
252−-- content (E2E encrypted blobs are ephemeral) but does reveal metadata: which pubkeys are
253−-- online, which servers they federate with, and delivery timing. Ensure the log pipeline
254−-- is access-controlled and encrypted at rest. Set to 0 in high-security deployments
255−-- and use manual GET /maintenance/snapshot with authenticated access instead.
256−-- Only emits when there's meaningful state to persist (presence/forwards/ackedIds > 0).
257−autoSnapshotExportInterval : Int<Ms>
258−autoSnapshotExportInterval = 3600000 -- 1 hour; full state export to stdout (0 to disable)
259−snapshotStalenessWarning : Int<Ms>
260−snapshotStalenessWarning = 3600000 -- 1 hour; warn if no snapshot exported via /maintenance/snapshot
261pollTimeoutMs : Int<Ms>
216pollTimeoutMs : Int<Ms>
262pollTimeoutMs = 25000 -- long-poll heartbeat interval (how long a poll blocks before returning empty)
217pollTimeoutMs = 25000 -- long-poll heartbeat interval (how long a poll blocks before returning empty)
263maxPollsPerKey = 3 -- max concurrent long-poll connections per pubkey
218maxPollsPerKey = 3 -- max concurrent long-poll connections per pubkey
@@ -278,14 +233,6 @@ federationEnabled = True {}
278-- TODO: If Knot adds a resolved-IP introspection API (e.g., fetchWithResolvedIp),
233-- TODO: If Knot adds a resolved-IP introspection API (e.g., fetchWithResolvedIp),
279-- validate the resolved IP against isBadServerName before completing the request.
234-- validate the resolved IP against isBadServerName before completing the request.
280federationUrl = \server -> federationScheme ++ server
235federationUrl = \server -> federationScheme ++ server
281−-- Bootstrap peer: on startup, fetch a snapshot from this server and auto-restore.
282−-- Set to a trusted peer's hostname (e.g., "peer1.example.com:8080") to bootstrap
283−-- federation state from an existing server. Empty = no bootstrap (default).
284−-- Requires the bootstrap peer to accept unauthenticated /maintenance/snapshot (or
285−-- set bootstrapPeerSecret to the peer's maintenance secret).
286−-- TODO: If Knot adds file I/O, add local file-based auto-load as an alternative.
287−bootstrapPeerUrl = "" -- set to a peer hostname to bootstrap state on startup
288−bootstrapPeerSecret = "" -- maintenance secret of the bootstrap peer
289-- Callback verification: outbound federation requests include a random single-use token;
236-- Callback verification: outbound federation requests include a random single-use token;
290-- the receiver calls back to /federation/verify to confirm the sender actually issued it.
237-- the receiver calls back to /federation/verify to confirm the sender actually issued it.
291maxPendingVerifications = 10000
238maxPendingVerifications = 10000
@@ -349,10 +296,6 @@ maxPendingCancels = 10000
349-- Max retries per cancel entry before dropping (uses same backoff as forward retries)
296-- Max retries per cancel entry before dropping (uses same backoff as forward retries)
350maxCancelRetries = 10
297maxCancelRetries = 10
352−-- Forward storm prevention: after snapshot restore, ramp up forward retries over this many
353−-- prune cycles (each cycle retries 1/N of maxForwardRetriesPerServer, increasing to full rate).
354−-- Prevents overwhelming peers when thousands of queued forwards are restored at once.
355−restoreForwardRampCycles = 5
356-- Verify callback caching: cache successful verifications per-server to reduce
299-- Verify callback caching: cache successful verifications per-server to reduce
357-- outbound HTTP calls. A peer sending N requests/second forces N verify callbacks;
300-- outbound HTTP calls. A peer sending N requests/second forces N verify callbacks;
358-- caching the success for a short window amortizes the cost.
301-- caching the success for a short window amortizes the cost.
@@ -454,7 +397,6 @@ type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Times
454-- Queued federation forward: message waiting to be delivered to a remote server.
397-- Queued federation forward: message waiting to be delivered to a remote server.
455-- senderKey: pubkey of the authenticated sender (locally-originated only; "" for federation relays).
398-- senderKey: pubkey of the authenticated sender (locally-originated only; "" for federation relays).
456-- Used by fedDeliveryReceipts to generate delivery receipts even if the original message was pruned.
399-- Used by fedDeliveryReceipts to generate delivery receipts even if the original message was pruned.
457−-- Pre-v12 snapshots without senderKey default to "" (Knot ignores missing Text fields).
458type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int, expiresAt: Timestamp, priority: Int, senderKey: Text}
400type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int, expiresAt: Timestamp, priority: Int, senderKey: Text}
459-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
401-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
460type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}
402type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}
@@ -489,10 +431,7 @@ type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
489-- Tracks unique pubkeys authenticated per IP (key-rotation abuse detection)
431-- Tracks unique pubkeys authenticated per IP (key-rotation abuse detection)
490type UniqueKeyIpEntry = {ip: Text, keys: [PubkeyHex], windowStart: Timestamp}
432type UniqueKeyIpEntry = {ip: Text, keys: [PubkeyHex], windowStart: Timestamp}
492−-- Forward nonce for extended replay protection (persisted in snapshots)
434+-- Forward nonce for extended replay protection
493−-- originServer was removed in v9: nonce checks ignore originServer (a messageId+toKey
494−-- pair is globally unique), so storing it wasted memory. Pre-v9 snapshots with the
495−-- extra field are accepted during restore (Knot's JSON parser ignores unknown fields).
496type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, seenAt: Timestamp}
435type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, seenAt: Timestamp}
498-- Batch send types
437-- Batch send types
@@ -847,12 +786,6 @@ maxPeerLatencyEntries = 100
847-- Federation verify per-IP rate limiting (tighter than general IP rate)
786-- Federation verify per-IP rate limiting (tighter than general IP rate)
848*verifyIpRates : [IpRateEntry]
787*verifyIpRates : [IpRateEntry]
850−-- Tracks when /maintenance/snapshot was last exported (for staleness warnings)
851−*lastSnapshotExportAt : Timestamp
853−-- Tracks when the last auto-export [SNAPSHOT_DATA] was emitted to stdout
854−*lastAutoSnapshotExportAt : Timestamp
856-- Per-peer gossip event volume rate limiting
789-- Per-peer gossip event volume rate limiting
857*gossipEventRates : [GossipEventRateEntry]
790*gossipEventRates : [GossipEventRateEntry]
@@ -942,21 +875,6 @@ maxPeerLatencyEntries = 100
942-- Last time the background watchdog loop ran (for health check liveness detection)
875-- Last time the background watchdog loop ran (for health check liveness detection)
943*lastWatchdogAt : Timestamp
876*lastWatchdogAt : Timestamp
945−-- Fingerprint of key state counters at last auto-snapshot export.
946−-- backgroundSnapshot compares current counters against this to skip redundant exports.
947−*lastSnapshotFingerprint : Int
949−-- Prune cycles since last snapshot export (wall-clock agnostic staleness tracking).
950−-- Incremented each prune cycle; reset to 0 on snapshot export (manual, auto, or drain).
951−-- More robust than wall-clock comparison: survives clock jumps without false alarms.
952−*prunesSinceSnapshot : Int
954−-- Prune cycles since last snapshot restore (forward storm prevention).
955−-- After restore, forward retries are throttled for restoreForwardRampCycles prune cycles
956−-- to avoid overwhelming peers with thousands of queued forwards simultaneously.
957−-- 0 = not in ramp-up phase. Decremented each prune cycle; forwards are throttled while > 0.
958−*restoreForwardRampRemaining : Int
960-- Cached health result: returned when /health is rate-limited instead of false all-healthy.
878-- Cached health result: returned when /health is rate-limited instead of false all-healthy.
961-- Updated on every successful (non-rate-limited) health check.
879-- Updated on every successful (non-rate-limited) health check.
962*cachedHealthOk : Bool
880*cachedHealthOk : Bool
@@ -964,7 +882,6 @@ maxPeerLatencyEntries = 100
964*cachedHealthHeartbeat : Bool
882*cachedHealthHeartbeat : Bool
965*cachedHealthCapacity : Bool
883*cachedHealthCapacity : Bool
966*cachedHealthFederation : Bool
884*cachedHealthFederation : Bool
967−*cachedHealthSnapshot : Bool
968*cachedHealthWatchdog : Bool
885*cachedHealthWatchdog : Bool
969*cachedHealthNonce : Bool
886*cachedHealthNonce : Bool
970*cachedHealthCapacityWarning : Bool
887*cachedHealthCapacityWarning : Bool
@@ -1122,18 +1039,6 @@ route MaintenanceApi where
1122 POST {} /maintenance/shutdown headers {authorization: Text}
1039 POST {} /maintenance/shutdown headers {authorization: Text}
1123 -> {ok: Bool, drained: Int}
1040 -> {ok: Bool, drained: Int}
1124 = Shutdown
1041 = Shutdown
1125− -- Snapshot: export federation metadata for persistence across restarts.
1126− -- Messages are excluded (ephemeral by design; clients handle re-delivery).
1127− -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
1128− GET /maintenance/snapshot headers {authorization: Text}
1129− -> {ok: Bool, version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}], senderExpirations: [SenderExpiration], senderDeliveries: [SenderDelivery], rateLimitOverrides: {globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}}
1131− -- Restore: import a previously-exported snapshot. Server must be in drain mode
1132− -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
1133− -- checksum: integrity check (0 = skip verification, for pre-v6 snapshots)
1134− POST {version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}], senderExpirations: [SenderExpiration], senderDeliveries: [SenderDelivery], rateLimitOverrides: {globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}} /maintenance/restore headers {authorization: Text}
1135− -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
1137 -- Block a federation peer (rejects all inbound/outbound federation with this server)
1042 -- Block a federation peer (rejects all inbound/outbound federation with this server)
1138 POST {server: ServerName} /maintenance/block headers {authorization: Text}
1043 POST {server: ServerName} /maintenance/block headers {authorization: Text}
1139 -> {ok: Bool, blocked: Int}
1044 -> {ok: Bool, blocked: Int}
@@ -1209,7 +1114,7 @@ route MaintenanceApi where
1209 = RotateSecret
1114 = RotateSecret
1211route HealthApi where
1116route HealthApi where
1212− GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, draining: Bool, degradedPeers: Int} = Health
1117+ GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, draining: Bool, degradedPeers: Int} = Health
1214-- Returns counts of in-memory collections (requires maintenance secret)
1119-- Returns counts of in-memory collections (requires maintenance secret)
1215-- TODO: If Knot adds raw text response support, add a /metrics endpoint
1120-- TODO: If Knot adds raw text response support, add a /metrics endpoint
@@ -1219,7 +1124,7 @@ route HealthApi where
1219-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1124-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1220route StatsApi where
1125route StatsApi where
1221 GET /stats headers {authorization: Text}
1126 GET /stats headers {authorization: Text}
1222− -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, activePolls: Int, activePollsMax: Int, recipientCounts: 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}]}, 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, prunesSinceSnapshot: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int}}
1127+ -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, activePolls: Int, activePollsMax: Int, recipientCounts: 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}]}, 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, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int}}
1224 GET /stats/peers headers {authorization: Text}
1129 GET /stats/peers headers {authorization: Text}
1225 -> {peers: [PeerStatus], ok: Bool}
1130 -> {peers: [PeerStatus], ok: Bool}
@@ -1380,8 +1285,7 @@ tooLong : Text -> Int -> Bool
1380tooLong = \s maxLen -> drop maxLen s != ""
1285tooLong = \s maxLen -> drop maxLen s != ""
1382-- Sanitize correlationId: truncate to maxCorrelationIdLen and strip non-printable characters
1287-- Sanitize correlationId: truncate to maxCorrelationIdLen and strip non-printable characters
1383−-- to prevent log injection attacks. A malicious peer could embed "\n[SNAPSHOT_DATA] ..."
1288+-- to prevent log injection attacks. A malicious peer could inject fake [ERROR] entries or
1384−-- to spoof snapshot lines parsed by sidecar processes, inject fake [ERROR] entries, or
1385-- use ANSI escape codes to manipulate terminal output. Only printable ASCII (0x20-0x7E)
1289-- use ANSI escape codes to manipulate terminal output. Only printable ASCII (0x20-0x7E)
1386-- is kept; all other bytes (null, control chars, escape sequences, DEL, high bytes) are
1290-- is kept; all other bytes (null, control chars, escape sequences, DEL, high bytes) are
1387-- stripped. This is stricter than the original \n/\r/\t filter.
1291-- stripped. This is stricter than the original \n/\r/\t filter.
@@ -4163,17 +4067,7 @@ retryPendingForwardsFromBuckets = do
4163 let onlineServers = collectPeerServers onlineForwards (\x -> x.toServer)
4067 let onlineServers = collectPeerServers onlineForwards (\x -> x.toServer)
4164 let offlineOnlyServers = do s <- eligibleServers; where none onlineServers (\os -> os == s); yield s
4068 let offlineOnlyServers = do s <- eligibleServers; where none onlineServers (\os -> os == s); yield s
4165 let prioritizedServers = union onlineServers offlineOnlyServers
4069 let prioritizedServers = union onlineServers offlineOnlyServers
4166− -- Forward storm prevention: after restore, limit per-server retries to a fraction that
4167− -- increases each prune cycle. Prevents thousands of restored forwards from overwhelming
4168− -- peers in the first cycle. rampRemaining counts down from restoreForwardRampCycles to 0.
4169− rampRemaining <- *restoreForwardRampRemaining
4170− let effectiveMaxRetries = if rampRemaining > 0
4171− then max 1 (maxForwardRetriesPerServer * (restoreForwardRampCycles - rampRemaining + 1) / restoreForwardRampCycles)
4172− else maxForwardRetriesPerServer
4173− when (rampRemaining > 0) (do
4174− logInfo ("federation: post-restore ramp " ++ show (restoreForwardRampCycles - rampRemaining + 1) ++ "/" ++ show restoreForwardRampCycles ++ " (max " ++ show effectiveMaxRetries ++ " retries/server)")
4175− atomic do set *restoreForwardRampRemaining = rampRemaining - 1)
4176− sendForwardsPartitioned prioritizedForwards prioritizedServers effectiveMaxRetries
4070+ sendForwardsPartitioned prioritizedForwards prioritizedServers maxForwardRetriesPerServer
4178-- When a remote key comes online, create forward entries for queued messages and send them
4072-- When a remote key comes online, create forward entries for queued messages and send them
4179forwardQueuedMessages = \targets t -> do
4073forwardQueuedMessages = \targets t -> do
@@ -4489,8 +4383,6 @@ waitForInFlightDrain = \deadline -> do
4489pruneAndRetry = \t -> do
4383pruneAndRetry = \t -> do
4490 atomic do
4384 atomic do
4491 set *lastPrunedAt = t
4385 set *lastPrunedAt = t
4492− pss <- *prunesSinceSnapshot
4493− set *prunesSinceSnapshot = pss + 1
4494 pr <- pruneExpired t
4386 pr <- pruneExpired t
4495 -- Warn if forward nonces are near capacity (non-expired eviction creates a replay window).
4387 -- Warn if forward nonces are near capacity (non-expired eviction creates a replay window).
4496 -- Operators should increase maxForwardNonces or investigate high message volume.
4388 -- Operators should increase maxForwardNonces or investigate high message volume.
@@ -5227,7 +5119,7 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
5227 globalRateLimited <- checkGlobalRate t
5119 globalRateLimited <- checkGlobalRate t
5228 if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0})
5120 if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0})
5230− -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
5122+ -- Capture presenceVersion before long-poll wait to skip unchanged lookups
5231 startPresenceVersion <- *presenceVersion
5123 startPresenceVersion <- *presenceVersion
5232 result <- case extractToken authorization of
5124 result <- case extractToken authorization of
5233 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: [], pubkey: "", pollSlotFull: False {}, hasMore: False {}}
5125 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: [], pubkey: "", pollSlotFull: False {}, hasMore: False {}}
@@ -6461,37 +6353,6 @@ handleDrain = \phase authorization respond -> do
6461 yield {forwards: f, pendingAcks: pa}
6353 yield {forwards: f, pendingAcks: pa}
6462 let pendingForwardCount = count flushState.forwards
6354 let pendingForwardCount = count flushState.forwards
6463 let pendingAckCount = count flushState.pendingAcks
6355 let pendingAckCount = count flushState.pendingAcks
6464− -- Auto-snapshot on drain: export federation state so a crash during drain doesn't lose it.
6465− -- Output as structured JSON with [SNAPSHOT_DATA] prefix for log-shipping pipelines.
6466− -- This is the last chance to persist state before the server stops.
6467− drainSnapshot <- atomic do
6470− d <- readAllDelivered
6471− ff <- *forwardFailures
6472− fn <- readAllNonces
6473− bp <- *blockedPeers
6474− ap <- *allowedPeers
6475− pac <- *pendingAckCleanups
6476− se <- *senderExpirations
6477− sdl <- *senderDeliveries
6478− rlGlobal <- *rateOverrideGlobalRequest; rlIp <- *rateOverrideIpRate; rlSend <- *rateOverrideSendRate
6479− rlFed <- *rateOverrideFederationRate; rlAuth <- *rateOverrideAuthGlobal; rlRecv <- *rateOverrideRecvRate
6480− rlChal <- *rateOverrideChallengeRate; rlPair <- *rateOverridePairRate; rlLookup <- *rateOverrideLookupRate
6481− rlWrite <- *rateOverrideClientWriteRate
6482− let drainRlo = {globalRequest: rlGlobal, ipRate: rlIp, sendRate: rlSend, federationRate: rlFed, authGlobal: rlAuth, recvRate: rlRecv, challengeRate: rlChal, pairRate: rlPair, lookupRate: rlLookup, clientWriteRate: rlWrite}
6483− let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
6484− yield {version: snapshotVersion, checksum: drainCs, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se, senderDeliveries: sdl, rateLimitOverrides: drainRlo}
6485− -- Sentinel-wrapped: sidecar processes should verify [SNAPSHOT_END] follows to detect truncation.
6486− -- Integrity tag appended when maintenance secret is configured (enables tamper detection on restore).
6487− runtimeSec <- *runtimeMaintenanceSecret
6488− let effectiveSecret = if runtimeSec != "" then runtimeSec else maintenanceSecret
6489− let drainTag = snapshotIntegrityTag drainSnapshot.checksum effectiveSecret
6490− println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot ++ (if drainTag != 0 then " [INTEGRITY:" ++ show drainTag ++ "]" else ""))
6491− println "[SNAPSHOT_END]"
6492− atomic do set *prunesSinceSnapshot = 0
6493− logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
6494− logEvent "warn" "snapshot_unencrypted" "drain snapshot exported to stdout — ensure log pipeline is secured"
6495 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
6356 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
6496 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
6357 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
6497 atomic do set *drainCompletedAt = 0
6358 atomic do set *drainCompletedAt = 0
@@ -6640,11 +6501,6 @@ computeAndCacheHealth = \t -> do
6640 let fedOk = not federationEnabled || warmup
6501 let fedOk = not federationEnabled || warmup
6641 || count forwardFailures == 0
6502 || count forwardFailures == 0
6642 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
6503 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
6643− pss <- *prunesSinceSnapshot
6644− let staleThresholdCycles = if backgroundPruneInterval > 0 then snapshotStalenessWarning / backgroundPruneInterval else 0
6645− let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup
6646− || staleThresholdCycles == 0 || pss <= staleThresholdCycles
6647− || (tmc == 0 && presenceCount == 0 && forwardCount == 0)
6648 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
6504 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
6649 nonceCount <- countAllNonces 0
6505 nonceCount <- countAllNonces 0
6650 let nonceOk = nonceCount * 10 <= maxForwardNonces * 9
6506 let nonceOk = nonceCount * 10 <= maxForwardNonces * 9
@@ -6655,12 +6511,11 @@ computeAndCacheHealth = \t -> do
6655 capacityOk: capacityOk,
6511 capacityOk: capacityOk,
6656 capacityWarning: capacityWarn,
6512 capacityWarning: capacityWarn,
6657 federationOk: fedOk,
6513 federationOk: fedOk,
6658− snapshotOk: snapOk,
6659 watchdogOk: wdOk,
6514 watchdogOk: wdOk,
6660 nonceOk: nonceOk,
6515 nonceOk: nonceOk,
6661 isDraining: dp > 0,
6516 isDraining: dp > 0,
6662 degradedPeers: degradedPeerCount}
6517 degradedPeers: degradedPeerCount}
6663− let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk && result.nonceOk
6518+ let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk
6664 atomic do
6519 atomic do
6665 set *cachedHealthOk = allOk
6520 set *cachedHealthOk = allOk
6666 set *cachedHealthPrune = result.pruneOk
6521 set *cachedHealthPrune = result.pruneOk
@@ -6668,7 +6523,6 @@ computeAndCacheHealth = \t -> do
6668 set *cachedHealthCapacity = result.capacityOk
6523 set *cachedHealthCapacity = result.capacityOk
6669 set *cachedHealthCapacityWarning = result.capacityWarning
6524 set *cachedHealthCapacityWarning = result.capacityWarning
6670 set *cachedHealthFederation = result.federationOk
6525 set *cachedHealthFederation = result.federationOk
6671− set *cachedHealthSnapshot = result.snapshotOk
6672 set *cachedHealthWatchdog = result.watchdogOk
6526 set *cachedHealthWatchdog = result.watchdogOk
6673 set *cachedHealthNonce = result.nonceOk
6527 set *cachedHealthNonce = result.nonceOk
6674 set *cachedHealthDegradedPeers = result.degradedPeers
6528 set *cachedHealthDegradedPeers = result.degradedPeers
@@ -6688,18 +6542,17 @@ handleHealth = \authorization respond -> do
6688 cH <- *cachedHealthHeartbeat
6542 cH <- *cachedHealthHeartbeat
6689 cC <- *cachedHealthCapacity
6543 cC <- *cachedHealthCapacity
6690 cF <- *cachedHealthFederation
6544 cF <- *cachedHealthFederation
6691− cS <- *cachedHealthSnapshot
6692 cW <- *cachedHealthWatchdog
6545 cW <- *cachedHealthWatchdog
6693 cN <- *cachedHealthNonce
6546 cN <- *cachedHealthNonce
6694 cCW <- *cachedHealthCapacityWarning
6547 cCW <- *cachedHealthCapacityWarning
6695 cDP <- *cachedHealthDegradedPeers
6548 cDP <- *cachedHealthDegradedPeers
6696 dp <- *drainPhase
6549 dp <- *drainPhase
6697− yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, capacityWarning: cCW, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, nonceHealthy: cN, draining: dp > 0, degradedPeers: cDP}
6550+ yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, capacityWarning: cCW, federationHealthy: cF, watchdogHealthy: cW, nonceHealthy: cN, draining: dp > 0, degradedPeers: cDP}
6698 yield (respond cached)
6551 yield (respond cached)
6700 isAuthed <- verifyMaintenanceAuth authorization
6553 isAuthed <- verifyMaintenanceAuth authorization
6701 result <- computeAndCacheHealth t
6554 result <- computeAndCacheHealth t
6702− let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk && result.nonceOk
6555+ let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk
6703 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
6556 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
6704 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
6557 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
6705 -- Draining status is always visible (load balancers need it for routing decisions).
6558 -- Draining status is always visible (load balancers need it for routing decisions).
@@ -6709,7 +6562,6 @@ handleHealth = \authorization respond -> do
6709 capacityHealthy: result.capacityOk,
6562 capacityHealthy: result.capacityOk,
6710 capacityWarning: result.capacityWarning,
6563 capacityWarning: result.capacityWarning,
6711 federationHealthy: result.federationOk,
6564 federationHealthy: result.federationOk,
6712− snapshotHealthy: result.snapshotOk,
6713 watchdogHealthy: result.watchdogOk,
6565 watchdogHealthy: result.watchdogOk,
6714 nonceHealthy: result.nonceOk,
6566 nonceHealthy: result.nonceOk,
6715 draining: result.isDraining,
6567 draining: result.isDraining,
@@ -6719,13 +6571,12 @@ handleHealth = \authorization respond -> do
6719 capacityHealthy: True {},
6571 capacityHealthy: True {},
6720 capacityWarning: False {},
6572 capacityWarning: False {},
6721 federationHealthy: True {},
6573 federationHealthy: True {},
6722− snapshotHealthy: True {},
6723 watchdogHealthy: True {},
6574 watchdogHealthy: True {},
6724 nonceHealthy: True {},
6575 nonceHealthy: True {},
6725 draining: result.isDraining,
6576 draining: result.isDraining,
6726 degradedPeers: 0})
6577 degradedPeers: 0})
6728−zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: []}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, prunesSinceSnapshot: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0}}
6579+zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: []}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0}}
6730handleStats = \authorization respond -> do
6581handleStats = \authorization respond -> do
6731 t <- now
6582 t <- now
@@ -6781,7 +6632,6 @@ handleStats = \authorization respond -> do
6781 lp <- *lastPrunedAt
6632 lp <- *lastPrunedAt
6782 lh <- *lastHeartbeatAt
6633 lh <- *lastHeartbeatAt
6783 dr <- *drainPhase
6634 dr <- *drainPhase
6784− pss <- *prunesSinceSnapshot
6785 flt <- *federationLatencyTotal
6635 flt <- *federationLatencyTotal
6786 flc <- *federationLatencyCount
6636 flc <- *federationLatencyCount
6787 flm <- *federationLatencyMax
6637 flm <- *federationLatencyMax
@@ -6801,7 +6651,7 @@ handleStats = \authorization respond -> do
6801 cslt <- *clientSendLatencyTotal
6651 cslt <- *clientSendLatencyTotal
6802 cslc <- *clientSendLatencyCount
6652 cslc <- *clientSendLatencyCount
6803 cslm <- *clientSendLatencyMax
6653 cslm <- *clientSendLatencyMax
6804− yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, pss: pss, flt: flt, flc: flc, flm: flm, svc: svc, vfc: vfc, dca: dca, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf, arq: arq, parq: parq, cplt: cplt, cplc: cplc, cplm: cplm, cslt: cslt, cslc: cslc, cslm: cslm}
6654+ yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, svc: svc, vfc: vfc, dca: dca, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf, arq: arq, parq: parq, cplt: cplt, cplc: cplc, cplm: cplm, cslt: cslt, cslc: cslc, cslm: cslm}
6805 -- Compute derived metrics
6655 -- Compute derived metrics
6806 let sessions = core.sc
6656 let sessions = core.sc
6807 let messages = core.tmc
6657 let messages = core.tmc
@@ -6829,7 +6679,7 @@ handleStats = \authorization respond -> do
6829 capacity: {sessions: sessions, sessionsMax: maxSessions, sessionsPct: sessPct, messages: messages, messagesMax: maxMessages, messagesPct: msgPct, totalMessageCount: messages, presence: presence, presenceMax: maxPresenceEntries, presencePct: presPct, forwards: fwds, forwardsMax: maxForwardsEntries, forwardsPct: fwdPct, challenges: count core.c, challengesMax: maxChallenges, ackedIds: core.aCount, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: core.dCount, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
6679 capacity: {sessions: sessions, sessionsMax: maxSessions, sessionsPct: sessPct, messages: messages, messagesMax: maxMessages, messagesPct: msgPct, totalMessageCount: messages, presence: presence, presenceMax: maxPresenceEntries, presencePct: presPct, forwards: fwds, forwardsMax: maxForwardsEntries, forwardsPct: fwdPct, challenges: count core.c, challengesMax: maxChallenges, ackedIds: core.aCount, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: core.dCount, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
6830 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, slowVerifies: counters.svc, verifyFailures: counters.vfc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca, forwardServers: fwdServerCount, maxForwardQueueDepth: maxFwdDepth, topFailingPeers: topFailingPeers},
6680 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, slowVerifies: counters.svc, verifyFailures: counters.vfc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca, forwardServers: fwdServerCount, maxForwardQueueDepth: maxFwdDepth, topFailingPeers: topFailingPeers},
6831 rates: {send: count counters.sr, challenge: count counters.cr, federation: count counters.fr, federationPair: count counters.fpr, global: counters.grc, ip: count counters.ir, lookup: count counters.lr, recv: count counters.rr, pair: count counters.prr, gossipEvent: count core.ger, verifyIp: count core.vir, uniqueKeyIp: count core.ukir, pendingAckCleanups: count core.pac, senderExpirations: count core.se, senderDeliveries: count core.sdl, rejectedGlobal: counters.rlg, rejectedIp: counters.rli, rejectedAuth: counters.rla, rejectedPerKey: counters.rlk, rejectedFederation: counters.rlf},
6681 rates: {send: count counters.sr, challenge: count counters.cr, federation: count counters.fr, federationPair: count counters.fpr, global: counters.grc, ip: count counters.ir, lookup: count counters.lr, recv: count counters.rr, pair: count counters.prr, gossipEvent: count core.ger, verifyIp: count core.vir, uniqueKeyIp: count core.ukir, pendingAckCleanups: count core.pac, senderExpirations: count core.se, senderDeliveries: count core.sdl, rejectedGlobal: counters.rlg, rejectedIp: counters.rli, rejectedAuth: counters.rla, rejectedPerKey: counters.rlk, rejectedFederation: counters.rlf},
6832− timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0, drainPhase: counters.dr, prunesSinceSnapshot: counters.pss, totalEntries: sessions + messages + presence + fwds + count core.c + core.aCount + count core.ao + core.dCount + core.fnCount + count core.ap + count core.fi, activeRequests: counters.arq, peakActiveRequests: counters.parq, pollAvgMs: pollAvg, pollMaxMs: counters.cplm, pollCount: counters.cplc, sendAvgMs: sendAvg, sendMaxMs: counters.cslm, sendCount: counters.cslc}})
6682+ timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0, drainPhase: counters.dr, totalEntries: sessions + messages + presence + fwds + count core.c + core.aCount + count core.ao + core.dCount + core.fnCount + count core.ap + count core.fi, activeRequests: counters.arq, peakActiveRequests: counters.parq, pollAvgMs: pollAvg, pollMaxMs: counters.cplm, pollCount: counters.cplc, sendAvgMs: sendAvg, sendMaxMs: counters.cslm, sendCount: counters.cslc}})
6833 else yield (respond zeroStats)
6683 else yield (respond zeroStats)
6835handleStatsPeers = \authorization respond -> do
6685handleStatsPeers = \authorization respond -> do
@@ -6918,179 +6768,6 @@ handleReady = \respond -> do
6918 let isReady = lastPrune > 0 && lastPrune >= sa
6768 let isReady = lastPrune > 0 && lastPrune >= sa
6919 yield (respond {ready: isReady})
6769 yield (respond {ready: isReady})
6921−zeroRateLimitOverrides = {globalRequest: 0, ipRate: 0, sendRate: 0, federationRate: 0, authGlobal: 0, recvRate: 0, challengeRate: 0, pairRate: 0, lookupRate: 0, clientWriteRate: 0}
6923−emptySnapshot = {ok: False {}, version: 0, checksum: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: [], allowedPeers: [], pendingAckCleanups: [], senderExpirations: [], senderDeliveries: [], rateLimitOverrides: zeroRateLimitOverrides}
6925−-- GET /maintenance/snapshot: Export federation metadata for persistence.
6926−-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
6927−-- and pending forwards in a single atomic snapshot. Forward retry counters are
6928−-- preserved so they survive restarts (prevents re-trying permanently-failing forwards).
6929−handleSnapshot = \authorization respond -> do
6930− authorized <- verifyMaintenanceAuth authorization
6931− if not authorized then yield (respond emptySnapshot)
6934− result <- atomic do
6935− p <- readAllPresence
6938− d <- readAllDelivered
6939− ff <- *forwardFailures
6940− f <- readAllForwards
6941− fn <- readAllNonces
6942− bp <- *blockedPeers
6943− ap <- *allowedPeers
6944− pac <- *pendingAckCleanups
6945− se <- *senderExpirations
6946− sdl <- *senderDeliveries
6947− -- Read rate limit overrides for snapshot persistence (v14+)
6948− rlGlobal <- *rateOverrideGlobalRequest; rlIp <- *rateOverrideIpRate; rlSend <- *rateOverrideSendRate
6949− rlFed <- *rateOverrideFederationRate; rlAuth <- *rateOverrideAuthGlobal; rlRecv <- *rateOverrideRecvRate
6950− rlChal <- *rateOverrideChallengeRate; rlPair <- *rateOverridePairRate; rlLookup <- *rateOverrideLookupRate
6951− rlWrite <- *rateOverrideClientWriteRate
6952− let rlo = {globalRequest: rlGlobal, ipRate: rlIp, sendRate: rlSend, federationRate: rlFed, authGlobal: rlAuth, recvRate: rlRecv, challengeRate: rlChal, pairRate: rlPair, lookupRate: rlLookup, clientWriteRate: rlWrite}
6953− yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se, senderDeliveries: sdl, rateLimitOverrides: rlo}
6955− set *lastSnapshotExportAt = t
6956− set *prunesSinceSnapshot = 0
6957− let cs = snapshotChecksum (count result.presence) (count result.ackedIds) (count result.deliveredIds) (count result.forwards) (count result.forwardNonces)
6958− logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards, " ++ show (count result.forwardNonces) ++ " forwardNonces, " ++ show (count result.blockedPeers) ++ " blockedPeers, " ++ show (count result.allowedPeers) ++ " allowedPeers, " ++ show (count result.pendingAckCleanups) ++ " pendingAckCleanups, " ++ show (count result.senderExpirations) ++ " senderExpirations, " ++ show (count result.senderDeliveries) ++ " senderDeliveries, checksum=" ++ show cs)
6959− yield (respond {ok: True {}, version: snapshotVersion, checksum: cs, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces, blockedPeers: result.blockedPeers, allowedPeers: result.allowedPeers, pendingAckCleanups: result.pendingAckCleanups, senderExpirations: result.senderExpirations, senderDeliveries: result.senderDeliveries, rateLimitOverrides: result.rateLimitOverrides})
6961−-- POST /maintenance/restore: Import a previously-exported snapshot.
6962−-- Requires drain mode to prevent conflicts with live state. Merges imported data
6963−-- with existing state (dedup entries are unioned, presence is merged by freshness).
6964−-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
6965−-- trigger federation forward creation from messages + restored presence.
6966−handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData blockedPeersData allowedPeersData pendingAckCleanupsData senderExpirationsData senderDeliveriesData rateLimitOverridesData authorization respond -> do
6967− authorized <- verifyMaintenanceAuth authorization
6968− if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6970− logEvent "audit" "restore" ("version=" ++ show version ++ " presence=" ++ show (count presenceData) ++ " forwards=" ++ show (count forwardsData))
6971− -- Minimum version: v6 (first version with checksum integrity verification).
6972− -- Pre-v6 snapshots (0-5) have no checksum and could be corrupted without detection.
6973− -- Versions 6-10 are accepted with backward compatibility (missing fields default to []/0).
6974− if version < 6 then do
6975− logError ("restore: snapshot version " ++ show version ++ " is too old (minimum: 6, current: " ++ show snapshotVersion ++ "). Pre-v6 snapshots lack checksum integrity verification.")
6976− yield (respond {ok: False {}, error: "version_too_old", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6977− else if version > snapshotVersion then do
6978− logError ("restore: snapshot version " ++ show version ++ " is newer than server (current: " ++ show snapshotVersion ++ ")")
6979− yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6981− -- Checksum verification (required for v6+)
6982− let expectedCs = snapshotChecksum (count presenceData) (count ackedIdsData) (count deliveredIdsData) (count forwardsData) (count forwardNoncesData)
6983− if checksumIn != 0 && checksumIn != expectedCs then do
6984− logError ("restore: checksum mismatch (got " ++ show checksumIn ++ ", expected " ++ show expectedCs ++ ") — snapshot may be truncated or corrupted")
6985− yield (respond {ok: False {}, error: "checksum_mismatch", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6989− if dp < 2 then yield (respond {ok: False {}, error: "server_must_be_draining", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6991− -- Validate imported data: reject entries with invalid or self-referencing server names.
6992− -- originServer in ForwardEntry is untyped Text (not ServerName), so not validated by the type system.
6993− -- Also reject entries pointing to the local server (nonsensical: we don't forward to ourselves
6994− -- or ack to ourselves).
6995− let validForwards = do f <- forwardsData; where not (isLocalServer f.toServer) && (f.originServer == "" || not (isBadServerName f.originServer)) && not (isLocalServer f.originServer); yield f
6996− let validAcked = do a <- ackedIdsData; where not (isLocalServer a.originServer); yield a
6997− let validOverflow = do a <- ackOverflowData; where not (isLocalServer a.originServer); yield a
6998− -- Validate presence entries: reject entries with bad server names (SSRF risk) or
6999− -- self-referencing local server (nonsensical in a snapshot import). The server field
7000− -- in PresenceEntry uses AnyServerName (no isBadServerName check), so a corrupted
7001− -- snapshot could inject presence pointing to private IPs, causing forward attempts
7002− -- to those addresses.
7003− let validPresence = do p <- presenceData; where not (isLocalServer p.server) && not (isBadServerName p.server); yield p
7004− -- DeliveredEntry has no originServer field (only messageId, toKey, deliveredAt),
7005− -- so no server-name validation is needed here (unlike AckedEntry/ForwardEntry).
7006− let validDeliveredIds = deliveredIdsData
7007− let validFailures = do f <- forwardFailuresData; where not (isBadServerName f.server) && not (isLocalServer f.server); yield f
7008− let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow) + (count presenceData - count validPresence) + (count deliveredIdsData - count validDeliveredIds) + (count forwardFailuresData - count validFailures)
7009− when (rejectedCount > 0) (logWarn ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))
7010− result <- atomic do
7011− -- Merge presence (existing entries win on conflict via freshness)
7012− allPresence <- readAllPresence
7013− let mergedPresence = mergeByPubkeyServer allPresence (do p <- validPresence; where p.expiresAt > t; yield p)
7014− let cappedPresence = takeRelation maxPresenceEntries mergedPresence
7015− writeAllPresence cappedPresence
7016− set *totalPresenceCount = count cappedPresence
7017− set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)
7018− set *presenceVersion = t
7019− -- Union dedup state, respecting capacity limits (using validated data)
7020− ackedIds <- readAllAcked
7021− let mergedAcked = union ackedIds validAcked
7022− let cappedAcked = takeRelation maxAckedIds mergedAcked
7023− writeAllAcked cappedAcked
7024− ackOverflow <- *ackOverflow
7025− let mergedOverflow = union ackOverflow validOverflow
7026− set *ackOverflow = takeRelation maxAckOverflow mergedOverflow
7027− deliveredIds <- readAllDelivered
7028− let mergedDelivered = union deliveredIds validDeliveredIds
7029− let cappedDelivered = takeRelation maxDeliveredIds mergedDelivered
7030− writeAllDelivered cappedDelivered
7031− -- Restore peer failure records
7032− forwardFailures <- *forwardFailures
7033− let mergedFailures = mergeByKey (\a b -> a.server == b.server) (\x -> x.failedAt) forwardFailures validFailures
7034− let cappedFailures = takeRelation maxForwardFailures mergedFailures
7035− set *forwardFailures = cappedFailures
7036− -- Restore forward entries (dedup by messageId + toServer, keep fresher on conflict; using validated data)
7037− existingForwards <- readAllForwards
7038− let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) existingForwards validForwards
7039− let cappedForwards = takeRelation maxForwardsEntries mergedForwards
7040− writeAllForwards cappedForwards
7041− set *totalForwardCount = count cappedForwards
7042− -- Restore forward nonces (dedup by messageId + toKey, keep fresher on conflict)
7043− existingNonces <- readAllNonces
7044− let validNonces = do n <- forwardNoncesData; where n.seenAt <= t && t - n.seenAt < forwardNonceRetention; yield n
7045− let mergedNonces = mergeByKey (\a b -> a.messageId == b.messageId && a.toKey == b.toKey) (\x -> x.seenAt) existingNonces validNonces
7046− let cappedNonces = takeRelation maxForwardNonces mergedNonces
7047− writeAllNonces cappedNonces
7048− -- Restore blocked peers (union with existing, cap at maxBlockedPeers)
7049− blocked <- *blockedPeers
7050− let mergedBlocked = fold (\acc b -> if any acc (\x -> x == b) then acc else union [b] acc) blocked blockedPeersData
7051− let cappedBlocked = takeRelation maxBlockedPeers mergedBlocked
7052− set *blockedPeers = cappedBlocked
7053− -- Restore allowed peers (union with existing, cap at maxAllowedPeers)
7054− allowed <- *allowedPeers
7055− let mergedAllowed = fold (\acc a -> if any acc (\x -> x == a) then acc else union [a] acc) allowed allowedPeersData
7056− let cappedAllowed = takeRelation maxAllowedPeers mergedAllowed
7057− set *allowedPeers = cappedAllowed
7058− -- Restore pending ack cleanups (union with existing, cap at maxPendingAckCleanups)
7059− -- pre-v7 snapshots have empty pendingAckCleanups (field defaults to [])
7060− existingCleanups <- *pendingAckCleanups
7061− let mergedCleanups = union existingCleanups pendingAckCleanupsData
7062− set *pendingAckCleanups = takeRelation maxPendingAckCleanups mergedCleanups
7063− -- Restore sender expiration notifications (union with existing, cap at maxSenderExpirations)
7064− -- pre-v8 snapshots have empty senderExpirations (field defaults to [])
7065− existingSenderExp <- *senderExpirations
7066− let mergedSenderExp = union existingSenderExp senderExpirationsData
7067− set *senderExpirations = takeRelation maxSenderExpirations mergedSenderExp
7068− -- Restore sender delivery receipts (union with existing, cap at maxSenderDeliveries)
7069− -- pre-v9 snapshots have empty senderDeliveries (field defaults to [])
7070− existingSenderDel <- *senderDeliveries
7071− let mergedSenderDel = union existingSenderDel senderDeliveriesData
7072− set *senderDeliveries = takeRelation maxSenderDeliveries mergedSenderDel
7073− -- Restore rate limit overrides (v14+; pre-v14 snapshots have all-zero fields via Knot defaults)
7074− -- Non-zero values from the snapshot override compile-time defaults; 0 = keep current.
7075− let rlo = rateLimitOverridesData
7076− when (rlo.globalRequest > 0) (set *rateOverrideGlobalRequest = rlo.globalRequest)
7077− when (rlo.ipRate > 0) (set *rateOverrideIpRate = rlo.ipRate)
7078− when (rlo.sendRate > 0) (set *rateOverrideSendRate = rlo.sendRate)
7079− when (rlo.federationRate > 0) (set *rateOverrideFederationRate = rlo.federationRate)
7080− when (rlo.authGlobal > 0) (set *rateOverrideAuthGlobal = rlo.authGlobal)
7081− when (rlo.recvRate > 0) (set *rateOverrideRecvRate = rlo.recvRate)
7082− when (rlo.challengeRate > 0) (set *rateOverrideChallengeRate = rlo.challengeRate)
7083− when (rlo.pairRate > 0) (set *rateOverridePairRate = rlo.pairRate)
7084− when (rlo.lookupRate > 0) (set *rateOverrideLookupRate = rlo.lookupRate)
7085− when (rlo.clientWriteRate > 0) (set *rateOverrideClientWriteRate = rlo.clientWriteRate)
7086− yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
7087− -- Activate forward storm prevention: ramp up retries over several prune cycles
7088− -- so restored forwards don't overwhelm peers in the first cycle.
7089− when (count validForwards > 0) (atomic do
7090− set *restoreForwardRampRemaining = restoreForwardRampCycles)
7091− logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
7092− yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
7094-- POST /messages/ack: Explicitly acknowledge received messages by ID.
6771-- POST /messages/ack: Explicitly acknowledge received messages by ID.
7095-- Alternative to the implicit cursor-based ack in poll. Benefits:
6772-- Alternative to the implicit cursor-based ack in poll. Benefits:
7096-- 1. Reduces the window for duplicate delivery (ack immediately without waiting for next poll)
6773-- 1. Reduces the window for duplicate delivery (ack immediately without waiting for next poll)
@@ -7881,8 +7558,6 @@ requestPath = \req -> case req of
7881 Drain {} -> "POST /maintenance/drain"
7558 Drain {} -> "POST /maintenance/drain"
7882 Shutdown {} -> "POST /maintenance/shutdown"
7559 Shutdown {} -> "POST /maintenance/shutdown"
7883 Undrain {} -> "POST /maintenance/undrain"
7560 Undrain {} -> "POST /maintenance/undrain"
7884− Snapshot {} -> "GET /maintenance/snapshot"
7885− Restore {} -> "POST /maintenance/restore"
7886 BlockPeer {} -> "POST /maintenance/block"
7561 BlockPeer {} -> "POST /maintenance/block"
7887 UnblockPeer {} -> "POST /maintenance/unblock"
7562 UnblockPeer {} -> "POST /maintenance/unblock"
7888 ListBlocked {} -> "GET /maintenance/blocked"
7563 ListBlocked {} -> "GET /maintenance/blocked"
@@ -7935,8 +7610,6 @@ requestClientIp = \req -> case req of
7935 Drain {} -> ""
7610 Drain {} -> ""
7936 Shutdown {} -> ""
7611 Shutdown {} -> ""
7937 Undrain {} -> ""
7612 Undrain {} -> ""
7940 BlockPeer {} -> ""
7613 BlockPeer {} -> ""
7941 UnblockPeer {} -> ""
7614 UnblockPeer {} -> ""
7942 ListBlocked {} -> ""
7615 ListBlocked {} -> ""
@@ -7998,8 +7671,6 @@ serveInner = \req -> case req of
7998 Drain {phase, authorization, respond} -> handleDrain phase authorization respond
7671 Drain {phase, authorization, respond} -> handleDrain phase authorization respond
7999 Shutdown {authorization, respond} -> handleShutdown authorization respond
7672 Shutdown {authorization, respond} -> handleShutdown authorization respond
8000 Undrain {authorization, respond} -> handleUndrain authorization respond
7673 Undrain {authorization, respond} -> handleUndrain authorization respond
8001− Snapshot {authorization, respond} -> handleSnapshot authorization respond
8002− Restore {version, checksum, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, blockedPeers, allowedPeers, pendingAckCleanups, senderExpirations, senderDeliveries, rateLimitOverrides, authorization, respond} -> handleRestore version checksum presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces blockedPeers allowedPeers pendingAckCleanups senderExpirations senderDeliveries rateLimitOverrides authorization respond
8003 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
7674 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
8004 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
7675 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
8005 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
7676 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
@@ -8034,8 +7705,6 @@ isOverloadExempt = \req -> case req of
8034 Drain {} -> True {}
7705 Drain {} -> True {}
8035 Shutdown {} -> True {}
7706 Shutdown {} -> True {}
8036 Undrain {} -> True {}
7707 Undrain {} -> True {}
8037− Snapshot {} -> True {}
8038− Restore {} -> True {}
8039 Reconcile {} -> True {}
7708 Reconcile {} -> True {}
8040 Stats {} -> True {}
7709 Stats {} -> True {}
8041 StatsPeers {} -> True {}
7710 StatsPeers {} -> True {}
@@ -8163,7 +7832,7 @@ serve = \req -> do
8163-- Knot compiles tail calls to loops, so stack growth is bounded. If Knot's tail-call
7832-- Knot compiles tail calls to loops, so stack growth is bounded. If Knot's tail-call
8164-- optimization is ever disabled, these loops will overflow the stack over time.
7833-- optimization is ever disabled, these loops will overflow the stack over time.
8165-- The backgroundWatchdog can restart backgroundPrune and pollHeartbeat if they die,
7834-- The backgroundWatchdog can restart backgroundPrune and pollHeartbeat if they die,
8166−-- but cannot restart itself or backgroundSnapshot. A TCO failure in any loop would
7835+-- but cannot restart itself. A TCO failure in any loop would
8167-- eventually surface as a health check failure (pruneHealthy/heartbeatHealthy/watchdogHealthy).
7836-- eventually surface as a health check failure (pruneHealthy/heartbeatHealthy/watchdogHealthy).
8168-- TODO: If Knot adds an iterative construct (e.g., loop/while), migrate background loops
7837-- TODO: If Knot adds an iterative construct (e.g., loop/while), migrate background loops
8169-- to avoid the TCO dependency entirely.
7838-- to avoid the TCO dependency entirely.
@@ -8281,110 +7950,6 @@ pollHeartbeat = do
8281 set *heartbeatCount = hc + 1
7950 set *heartbeatCount = hc + 1
8282 pollHeartbeat
7951 pollHeartbeat
8284−-- Background loop: periodically logs a state summary to stdout.
8285−-- Operators can capture stdout to detect state drift; use /maintenance/snapshot for full export.
8286−-- When autoSnapshotExportInterval > 0, periodically emits [SNAPSHOT_DATA] JSON to stdout
8287−-- for sidecar capture, so a crash outside drain loses at most one export interval of state.
8288−-- TODO: If Knot adds file I/O primitives, auto-export snapshots to a file
8289−-- (e.g., /var/lib/skrepka/snapshot.json) in addition to logging.
8290−backgroundSnapshot = do
8291− sleep autoSnapshotInterval
8293− shutdownReq <- *shutdownRequested
8294− dca <- *drainCompletedAt
8295− if shutdownReq && dca > 0 then logInfo "backgroundSnapshot: shutdown complete, exiting loop"
8296− else if dp < 2 then do
8297− -- Split into two reads: cached counters (cheap, no bucket scans) and per-bucket
8298− -- counts (scan 16 buckets each). Reduces atomic block width vs one large block.
8299− cached <- atomic do
8300− tmc <- *totalMessageCount
8301− sc <- *totalSessionCount
8302− pc <- *totalPresenceCount
8304− ff <- *forwardFailures
8305− pac <- *pendingAckCleanups
8306− yield {messages: tmc, sessions: sc, presence: pc, ackOverflow: count ao, forwardFailures: count ff, pendingAckCleanups: count pac}
8307− bucketCounts <- atomic do
8308− fwdCount <- countAllForwards 0
8309− ackedCount <- countAllAcked 0
8310− deliveredCount <- countAllDelivered 0
8311− yield {forwards: fwdCount, ackedIds: ackedCount, deliveredIds: deliveredCount}
8312− let result = {messages: cached.messages, sessions: cached.sessions, presence: cached.presence, forwards: bucketCounts.forwards, ackedIds: bucketCounts.ackedIds, ackOverflow: cached.ackOverflow, deliveredIds: bucketCounts.deliveredIds, forwardFailures: cached.forwardFailures, pendingAckCleanups: cached.pendingAckCleanups}
8313− logInfo ("snapshot: messages=" ++ show result.messages ++ " sessions=" ++ show result.sessions ++ " presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " ackOverflow=" ++ show result.ackOverflow ++ " deliveredIds=" ++ show result.deliveredIds ++ " forwardFailures=" ++ show result.forwardFailures ++ " pendingAckCleanups=" ++ show result.pendingAckCleanups)
8315− -- Periodic full snapshot auto-export: emit [SNAPSHOT_DATA] JSON to stdout so a
8316− -- sidecar process can persist it. Only when there's meaningful state to save.
8317− when (autoSnapshotExportInterval > 0) (do
8318− lastAutoExport <- *lastAutoSnapshotExportAt
8319− let hasState = result.presence > 0 || result.forwards > 0 || result.ackedIds > 0 || result.deliveredIds > 0 || result.ackOverflow > 0
8320− -- Fingerprint: skip export when key counters haven't changed since last export.
8321− -- Uses multiplicative hash to distinguish {messages:1,sessions:0} from {messages:0,sessions:1}.
8322− -- Includes ackOverflow and forwardFailures so state changes in those collections
8323− -- (e.g., acks moving to overflow during a federation outage) trigger an export.
8324− let fingerprint = result.messages + result.sessions * 100003 + result.presence * 100019 + result.forwards * 100043 + result.ackedIds * 100049 + result.deliveredIds * 100057 + result.ackOverflow * 100069 + result.forwardFailures * 100081
8325− lastFp <- *lastSnapshotFingerprint
8326− let stateChanged = lastFp == 0 || fingerprint != lastFp
8327− when (hasState && stateChanged && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do
8328− -- NOTE: TOCTOU between the fingerprint/hasState check above and the atomic snapshot
8329− -- below. State could change between the two reads (e.g., a message arrives, presence
8330− -- is pruned). This is acceptable: worst case is one extra export (fingerprint changed
8331− -- after we decided to export) or one skipped export (state became empty after we
8332− -- decided to export). The next backgroundSnapshot cycle corrects either case.
8333− -- Single atomic read for consistent [SNAPSHOT_DATA] output (used for restore).
8334− -- Previous per-collection reads caused inconsistencies (e.g., ackedIds referencing
8335− -- messageIds from a different snapshot window). The contention cost is acceptable
8336− -- since auto-export runs at most once per hour.
8337− exportSnapshot <- atomic do
8338− p <- readAllPresence
8341− d <- readAllDelivered
8342− ff <- *forwardFailures
8343− f <- readAllForwards
8344− fn <- readAllNonces
8345− bp <- *blockedPeers
8346− ap <- *allowedPeers
8347− pac <- *pendingAckCleanups
8348− se <- *senderExpirations
8349− sdl <- *senderDeliveries
8350− rlGlobal <- *rateOverrideGlobalRequest; rlIp <- *rateOverrideIpRate; rlSend <- *rateOverrideSendRate
8351− rlFed <- *rateOverrideFederationRate; rlAuth <- *rateOverrideAuthGlobal; rlRecv <- *rateOverrideRecvRate
8352− rlChal <- *rateOverrideChallengeRate; rlPair <- *rateOverridePairRate; rlLookup <- *rateOverrideLookupRate
8353− rlWrite <- *rateOverrideClientWriteRate
8354− let autoRlo = {globalRequest: rlGlobal, ipRate: rlIp, sendRate: rlSend, federationRate: rlFed, authGlobal: rlAuth, recvRate: rlRecv, challengeRate: rlChal, pairRate: rlPair, lookupRate: rlLookup, clientWriteRate: rlWrite}
8355− yield {p: p, a: a, ao: ao, d: d, ff: ff, f: f, fn: fn, bp: bp, ap: ap, pac: pac, se: se, sdl: sdl, rlo: autoRlo}
8356− let p = exportSnapshot.p
8357− let a = exportSnapshot.a
8358− let ao = exportSnapshot.ao
8359− let d = exportSnapshot.d
8360− let ff = exportSnapshot.ff
8361− let f = exportSnapshot.f
8362− let fn = exportSnapshot.fn
8363− let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
8364− let exportData = {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: exportSnapshot.bp, allowedPeers: exportSnapshot.ap, pendingAckCleanups: exportSnapshot.pac, senderExpirations: exportSnapshot.se, senderDeliveries: exportSnapshot.sdl, rateLimitOverrides: exportSnapshot.rlo}
8365− autoRtSec <- *runtimeMaintenanceSecret
8366− let autoEffSec = if autoRtSec != "" then autoRtSec else maintenanceSecret
8367− let autoTag = snapshotIntegrityTag exportCs autoEffSec
8368− println ("[SNAPSHOT_DATA] " ++ toJson exportData ++ (if autoTag != 0 then " [INTEGRITY:" ++ show autoTag ++ "]" else ""))
8369− println "[SNAPSHOT_END]"
8370− logEvent "warn" "snapshot_unencrypted" "snapshot exported to stdout — ensure log pipeline is secured"
8372− set *lastAutoSnapshotExportAt = t
8373− set *lastSnapshotFingerprint = fingerprint
8374− set *prunesSinceSnapshot = 0
8375− logInfo ("snapshot: auto-exported full state (" ++ show result.presence ++ " presence, " ++ show result.forwards ++ " forwards)")))
8376− -- Warn if federation state hasn't been exported recently (will be lost on restart).
8377− -- Uses prune-cycle counter instead of wall-clock comparison to survive clock jumps.
8378− -- Threshold: snapshotStalenessWarning / backgroundPruneInterval prune cycles.
8379− pss <- *prunesSinceSnapshot
8380− let staleCycles = if backgroundPruneInterval > 0 then snapshotStalenessWarning / backgroundPruneInterval else 0
8381− when (maintenanceSecret != "" && snapshotStalenessWarning > 0
8382− && staleCycles > 0 && pss > staleCycles
8383− && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))
8384− (logError ("no snapshot exported in " ++ show pss ++ " prune cycles (~" ++ show (pss * backgroundPruneInterval / 1000) ++ "s) — federation state will be lost on restart (presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " deliveredIds=" ++ show result.deliveredIds ++ "). ACTION REQUIRED: run GET /maintenance/snapshot to persist state before restart"))
8388-- Background watchdog: monitors prune and heartbeat loops, restarts if they appear dead.
7953-- Background watchdog: monitors prune and heartbeat loops, restarts if they appear dead.
8389-- Knot lacks try/catch, so panicked loops can't self-recover. This watchdog detects
7954-- Knot lacks try/catch, so panicked loops can't self-recover. This watchdog detects
8390-- liveness by checking counter freshness and forks a new loop instance as needed.
7955-- liveness by checking counter freshness and forks a new loop instance as needed.
@@ -8528,7 +8093,7 @@ main = do
8528 when (maintenanceSecret != "" && not (tooLong maintenanceSecret 31))
8093 when (maintenanceSecret != "" && not (tooLong maintenanceSecret 31))
8529 (logWarn "maintenanceSecret is shorter than 32 characters — 32+ character secrets recommended for production (128+ bits of entropy)")
8094 (logWarn "maintenanceSecret is shorter than 32 characters — 32+ character secrets recommended for production (128+ bits of entropy)")
8530 when (maintenanceSecret != "" && federationEnabled && not (tooLong maintenanceSecret 15))
8095 when (maintenanceSecret != "" && federationEnabled && not (tooLong maintenanceSecret 15))
8531− (logError "SECURITY: maintenanceSecret is weak and federation is enabled — maintenance endpoints control drain, snapshot, and peer blocking. Use a secret of at least 32 characters.")
8096+ (logError "SECURITY: maintenanceSecret is weak and federation is enabled — maintenance endpoints control drain and peer blocking. Use a secret of at least 32 characters.")
8532 when (matchesHostnameOrSubdomain serverHostN "localhost")
8097 when (matchesHostnameOrSubdomain serverHostN "localhost")
8533 (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
8098 (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
8534 when (federationEnabled && isBadServerName serverHostN)
8099 when (federationEnabled && isBadServerName serverHostN)
@@ -8558,78 +8123,10 @@ main = do
8558 when (federationCircuitBreakerThreshold > 0)
8123 when (federationCircuitBreakerThreshold > 0)
8559 (logInfo (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
8124 (logInfo (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
8560 logInfo (" federation allowlist: " ++ show maxAllowedPeers ++ " max (empty = open federation)")
8125 logInfo (" federation allowlist: " ++ show maxAllowedPeers ++ " max (empty = open federation)")
8561− when (autoSnapshotInterval > 0)
8562− (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
8563− when (autoSnapshotExportInterval > 0) (do
8564− logInfo (" auto-snapshot export: " ++ show autoSnapshotExportInterval ++ "ms (full [SNAPSHOT_DATA] to stdout)")
8565− logWarn "SECURITY: auto-snapshot exports unencrypted federation state to stdout — ensure log pipeline is secured and access-restricted")
8566− -- Bootstrap from peer: fetch snapshot from a trusted peer and auto-restore.
8567− -- Runs before listen so the server starts with federation state.
8568− when (bootstrapPeerUrl != "") (do
8569− logInfo ("startup: bootstrapping from peer " ++ bootstrapPeerUrl ++ "...")
8570− let authHeader = if bootstrapPeerSecret != "" then "Bearer " ++ bootstrapPeerSecret else ""
8571− bootstrapResult <- fetch (federationUrl bootstrapPeerUrl) (Snapshot {authorization: authHeader})
8572− case bootstrapResult of
8573− Err e -> logWarn ("startup: bootstrap from " ++ bootstrapPeerUrl ++ " failed: " ++ show e ++ " (continuing without snapshot)")
8575− if not snap.ok then logWarn ("startup: bootstrap from " ++ bootstrapPeerUrl ++ " returned ok=false (peer may require auth), continuing without snapshot")
8577− logInfo ("startup: restoring bootstrap snapshot (version " ++ show snap.version ++ ", " ++ show (count snap.presence) ++ " presence, " ++ show (count snap.forwards) ++ " forwards)")
8578− -- Inline restore logic (can't call handleRestore since we're not listening yet).
8579− -- Only restore if version is compatible; skip checksum verification for simplicity.
8580− if snap.version < 6 || snap.version > snapshotVersion
8581− then logWarn ("startup: bootstrap snapshot version " ++ show snap.version ++ " incompatible (need 6-" ++ show snapshotVersion ++ "), skipping")
8584− set *drainPhase = 2 -- temporarily drain for restore safety
8585− -- Validate and merge (simplified version of handleRestore)
8586− let validPresence = do p <- snap.presence; where not (isLocalServer p.server) && not (isBadServerName p.server); yield p
8587− let validForwards = do f <- snap.forwards; where not (isLocalServer f.toServer) && (f.originServer == "" || not (isBadServerName f.originServer)) && not (isLocalServer f.originServer); yield f
8588− let validAcked = do a <- snap.ackedIds; where not (isLocalServer a.originServer); yield a
8589− let validOverflow = do a <- snap.ackOverflow; where not (isLocalServer a.originServer); yield a
8592− let cappedPresence = takeRelation maxPresenceEntries (do p <- validPresence; where p.expiresAt > bt; yield p)
8593− writeAllPresence cappedPresence
8594− set *totalPresenceCount = count cappedPresence
8595− set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)
8596− set *presenceVersion = bt
8597− writeAllAcked (takeRelation maxAckedIds validAcked)
8598− set *ackOverflow = takeRelation maxAckOverflow validOverflow
8599− writeAllDelivered (takeRelation maxDeliveredIds snap.deliveredIds)
8600− let validFailures = do f <- snap.forwardFailures; where not (isBadServerName f.server) && not (isLocalServer f.server); yield f
8601− set *forwardFailures = takeRelation maxForwardFailures validFailures
8602− let cappedForwards = takeRelation maxForwardsEntries validForwards
8603− writeAllForwards cappedForwards
8604− set *totalForwardCount = count cappedForwards
8605− let validNonces = do n <- snap.forwardNonces; where n.seenAt <= bt && bt - n.seenAt < forwardNonceRetention; yield n
8606− writeAllNonces (takeRelation maxForwardNonces validNonces)
8607− set *blockedPeers = takeRelation maxBlockedPeers snap.blockedPeers
8608− set *allowedPeers = takeRelation maxAllowedPeers snap.allowedPeers
8609− set *pendingAckCleanups = takeRelation maxPendingAckCleanups snap.pendingAckCleanups
8610− set *senderExpirations = takeRelation maxSenderExpirations snap.senderExpirations
8611− set *senderDeliveries = takeRelation maxSenderDeliveries snap.senderDeliveries
8612− -- Restore rate limit overrides (v14+; pre-v14 defaults to all zeros)
8613− let rlo = snap.rateLimitOverrides
8614− when (rlo.globalRequest > 0) (set *rateOverrideGlobalRequest = rlo.globalRequest)
8615− when (rlo.ipRate > 0) (set *rateOverrideIpRate = rlo.ipRate)
8616− when (rlo.sendRate > 0) (set *rateOverrideSendRate = rlo.sendRate)
8617− when (rlo.federationRate > 0) (set *rateOverrideFederationRate = rlo.federationRate)
8618− when (rlo.authGlobal > 0) (set *rateOverrideAuthGlobal = rlo.authGlobal)
8619− when (rlo.recvRate > 0) (set *rateOverrideRecvRate = rlo.recvRate)
8620− when (rlo.challengeRate > 0) (set *rateOverrideChallengeRate = rlo.challengeRate)
8621− when (rlo.pairRate > 0) (set *rateOverridePairRate = rlo.pairRate)
8622− when (rlo.lookupRate > 0) (set *rateOverrideLookupRate = rlo.lookupRate)
8623− when (rlo.clientWriteRate > 0) (set *rateOverrideClientWriteRate = rlo.clientWriteRate)
8624− set *drainPhase = 0 -- undrain after restore
8625− when (count cappedForwards > 0)
8626− (atomic do set *restoreForwardRampRemaining = restoreForwardRampCycles)
8627− logInfo ("startup: bootstrap complete (" ++ show (count validPresence) ++ " presence, " ++ show (count validForwards) ++ " forwards)"))
8628 fork backgroundPrune
8126 fork backgroundPrune
8629 fork pollHeartbeat
8127 fork pollHeartbeat
8630 fork backgroundWatchdog
8128 fork backgroundWatchdog
8631 fork backgroundMetaWatchdog
8129 fork backgroundMetaWatchdog
8632− when (autoSnapshotInterval > 0) (fork backgroundSnapshot)
8633 -- Startup peer connectivity check: probe known peers after a short delay
8130 -- Startup peer connectivity check: probe known peers after a short delay
8634 -- to surface federation misconfigurations immediately (DNS, firewall, TLS).
8131 -- to surface federation misconfigurations immediately (DNS, firewall, TLS).
8635 -- Runs as a forked task so it doesn't delay server startup.
8132 -- Runs as a forked task so it doesn't delay server startup.