104bf5e · skrepka · git.iko.soy

git.iko.soy/skrepka

Harden server.knot: IP binding, challenge consume-all, federation warnings, latency tracking, undrain, revoke sessions, cursor generation

104bf5eiko6 months ago

server.knot+123−28

compare with master

unifiedsplit

@@ -366,6 +366,14 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest

366-- Tracks outstanding federation fetches per server (detects hung connections)

367*federationInFlight : [FederationInFlightEntry]

368

369+-- Federation latency tracking (for /stats)

370+*federationLatencyTotal : Int<Ms>

371+*federationLatencyCount : Int

372+*federationLatencyMax : Int<Ms>

373+

374+-- Server instance generation (epoch timestamp, for cursor staleness detection)

375+*serverGeneration : Int

376+

369-- ============================================

377-- ============================================

370-- Routes

378-- Routes

371-- ============================================

379-- ============================================

@@ -382,12 +390,15 @@ route AuthApi where

382 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /verify headers {clientIp: Text}

390 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /verify headers {clientIp: Text}

383 -> {token: Text, expiresAt: Timestamp}

391 -> {token: Text, expiresAt: Timestamp}

384 = VerifyAuth

392 = VerifyAuth

393+ POST {targetToken: Text} /revoke headers {authorization: Text, clientIp: Text}

394+ -> {ok: Bool, authorized: Bool}

395+ = RevokeSession

385

396

386-- Client-facing API (requires session token in Authorization header)

397-- Client-facing API (requires session token in Authorization header)

387route ClientApi where

398route ClientApi where

388 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp

399 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp

389− POST {cursor: Timestamp} /poll headers {authorization: Text, clientIp: Text}

400+ POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text}

390− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool}

401+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int}

391 = Poll

402 = Poll

392 /messages

403 /messages

393 -- Send an E2E encrypted message to a recipient pubkey

404 -- Send an E2E encrypted message to a recipient pubkey

@@ -423,6 +434,9 @@ route MaintenanceApi where

423 POST {} /maintenance/drain headers {authorization: Text}

434 POST {} /maintenance/drain headers {authorization: Text}

424 -> {ok: Bool, drained: Int}

435 -> {ok: Bool, drained: Int}

425 = Drain

436 = Drain

437+ POST {} /maintenance/undrain headers {authorization: Text}

438+ -> {ok: Bool}

439+ = Undrain

426

440

427route HealthApi where

441route HealthApi where

428 GET /health -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool} = Health

442 GET /health -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool} = Health

@@ -430,7 +444,7 @@ route HealthApi where

430-- Returns counts of in-memory collections (requires federation or maintenance secret)

444-- Returns counts of in-memory collections (requires federation or maintenance secret)

431route StatsApi where

445route StatsApi where

432 GET /stats headers {authorization: Text}

446 GET /stats headers {authorization: Text}

433− -> {sessions: Int, sessionsMax: Int, messages: Int, messagesMax: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pruneCount: Int, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, federationEnabled: Bool, recentGossip: Int, totalMessageCount: Int, recipientCounts: Int, federationInFlight: Int, ok: Bool}

447+ -> {sessions: Int, sessionsMax: Int, messages: Int, messagesMax: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pruneCount: Int, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, federationEnabled: Bool, recentGossip: Int, totalMessageCount: Int, recipientCounts: Int, federationInFlight: Int, messagesPct: Int, presencePct: Int, forwardsPct: Int, sessionsPct: Int, federationAvgLatencyMs: Int, federationMaxLatencyMs: Int, federationCallCount: Int, ok: Bool}

434 = Stats

448 = Stats

435 GET /stats/peers headers {authorization: Text}

449 GET /stats/peers headers {authorization: Text}

436 -> {peers: [PeerStatus], ok: Bool}

450 -> {peers: [PeerStatus], ok: Bool}

@@ -665,6 +679,10 @@ isBadServerName = \sn ->

665 || containsSuffix ".arpa" sn

679 || containsSuffix ".arpa" sn

666 || containsSuffix ".onion" sn

680 || containsSuffix ".onion" sn

667 || containsSuffix ".i2p" sn

681 || containsSuffix ".i2p" sn

682+ || containsSuffix ".corp" sn

683+ || containsSuffix ".home" sn

684+ || containsSuffix ".lan" sn

685+ || containsSuffix ".intranet" sn

668

686

669-- Look up a non-expired session matching the given token (constant-time comparison).

687-- Look up a non-expired session matching the given token (constant-time comparison).

670-- Caller passes the appropriate session bucket (sharded by token prefix).

688-- Caller passes the appropriate session bucket (sharded by token prefix).

@@ -885,8 +903,16 @@ fetchFederation = \server request -> do

885 result <- fetch (federationUrl server) request

903 result <- fetch (federationUrl server) request

886 clearFederationInFlight server

904 clearFederationInFlight server

887 endTime <- now

905 endTime <- now

906+ let latency = endTime - startTime

888− when (endTime - startTime > federationFetchTimeoutMs)

907+ when (latency > federationFetchTimeoutMs)

889− (println ("federation: slow response from " ++ server ++ " (" ++ show (endTime - startTime) ++ "ms)"))

908+ (println ("federation: slow response from " ++ server ++ " (" ++ show latency ++ "ms)"))

909+ atomic do

910+ total <- *federationLatencyTotal

911+ cnt <- *federationLatencyCount

912+ mx <- *federationLatencyMax

913+ set *federationLatencyTotal = total + latency

914+ set *federationLatencyCount = cnt + 1

915+ set *federationLatencyMax = if latency > mx then latency else mx

890 yield result

916 yield result

891

917

892clampTtl : Int<S> -> Int<S>

918clampTtl : Int<S> -> Int<S>

@@ -1282,6 +1308,10 @@ withFederationAuth = \authorization t respond fromServer precondition action ->

1282 println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)

1308 println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)

1283 yield (respond {ok: False {}, error: "rate_limited"})

1309 yield (respond {ok: False {}, error: "rate_limited"})

1284 else do

1310 else do

1311+ let hasPeerSecrets = count federationPeerSecrets > 0

1312+ let hasSpecificSecret = any federationPeerSecrets (\p -> normalizeServerName p.server == fromServer)

1313+ when (hasPeerSecrets && not hasSpecificSecret)

1314+ (println ("federation: warning: " ++ fromServer ++ " using global secret (no per-peer secret configured)"))

1285 clearFailure fromServer

1315 clearFailure fromServer

1286 action

1316 action

1287

1317

@@ -1294,9 +1324,11 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do

1294 case authenticateSession sessions token t of

1324 case authenticateSession sessions token t of

1295 Unauthorized {} -> onUnauth

1325 Unauthorized {} -> onUnauth

1296 Authorized {pubkey, ip} -> do

1326 Authorized {pubkey, ip} -> do

1297− when (clientIp != "" && ip != "" && clientIp != ip)

1327+ if clientIp != "" && ip != "" && clientIp != ip

1328+ then do

1298− (println ("session: IP mismatch for " ++ pubkey ++ " (session: " ++ ip ++ ", request: " ++ clientIp ++ ")"))

1329+ println ("session: IP mismatch for " ++ pubkey ++ " (session: " ++ ip ++ ", request: " ++ clientIp ++ "), rejecting")

1330+ onUnauth

1299− onAuth pubkey

1331+ else onAuth pubkey

1300

1332

1301removeForward = \mid srv -> atomic do

1333removeForward = \mid srv -> atomic do

1302 forwards <- *forwards

1334 forwards <- *forwards

@@ -1463,6 +1495,8 @@ retryEvictedAcks = \evictedEntries pendingAcks ->

1463 let deliveredExcess = count combinedDelivered - maxDeliveredIds

1495 let deliveredExcess = count combinedDelivered - maxDeliveredIds

1464 set *deliveredIds = if deliveredExcess <= 0 then combinedDelivered

1496 set *deliveredIds = if deliveredExcess <= 0 then combinedDelivered

1465 else (evictSmallest combinedDelivered (max 0 deliveredExcess) (\d -> d.deliveredAt)).kept

1497 else (evictSmallest combinedDelivered (max 0 deliveredExcess) (\d -> d.deliveredAt)).kept

1498+ when (deliveredExcess > 0)

1499+ (println ("ackMessages: deliveredIds at capacity (" ++ show maxDeliveredIds ++ "), " ++ show deliveredExcess ++ " oldest dedup entries permanently removed for " ++ server))

1466 yield overflowExcess

1500 yield overflowExcess

1467 when (dropped > 0)

1501 when (dropped > 0)

1468 (println ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries moved to delivered set for " ++ server)))

1502 (println ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries moved to delivered set for " ++ server)))

@@ -1494,12 +1528,14 @@ retryOverflowAcks = \overflow -> do

1494pruneAndRetry = \t -> do

1528pruneAndRetry = \t -> do

1495 atomic do set *lastPrunedAt = t

1529 atomic do set *lastPrunedAt = t

1496 pr <- pruneExpired t

1530 pr <- pruneExpired t

1497− result <- atomic do

1531+ pendingForwards <- atomic do f <- *forwards; yield f

1498− forwards <- *forwards

1532+ ackState <- atomic do

1499 ackedIds2 <- *ackedIds

1533 ackedIds2 <- *ackedIds

1500 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a

1534 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a

1501 ackOverflow <- *ackOverflow

1535 ackOverflow <- *ackOverflow

1502 let pendingOverflow = do a <- ackOverflow; where not (isLocalServer a.originServer); yield a

1536 let pendingOverflow = do a <- ackOverflow; where not (isLocalServer a.originServer); yield a

1537+ yield {pendingAcks: pendingAcks, pendingOverflow: pendingOverflow}

1538+ refreshResult <- atomic do

1503 lastRefresh <- *lastPresenceRefresh

1539 lastRefresh <- *lastPresenceRefresh

1504 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval

1540 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval

1505 presence <- *presence

1541 presence <- *presence

@@ -1512,7 +1548,8 @@ pruneAndRetry = \t -> do

1512 set *presence = do pp <- presence; yield (if any toExtend (\e -> e.pubkey == pp.pubkey && isLocalServer pp.server) then {pubkey: pp.pubkey, server: pp.server, expiresAt: t + sessionExpiry, timestamp: t} else pp))

1548 set *presence = do pp <- presence; yield (if any toExtend (\e -> e.pubkey == pp.pubkey && isLocalServer pp.server) then {pubkey: pp.pubkey, server: pp.server, expiresAt: t + sessionExpiry, timestamp: t} else pp))

1513 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []

1549 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []

1514 when shouldRefresh (set *lastPresenceRefresh = t)

1550 when shouldRefresh (set *lastPresenceRefresh = t)

1515− yield {pendingForwards: forwards, pendingAcks: pendingAcks, pendingOverflow: pendingOverflow, extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}

1551+ yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}

1552+ let result = {pendingForwards: pendingForwards, pendingAcks: ackState.pendingAcks, pendingOverflow: ackState.pendingOverflow, extendedActive: refreshResult.extendedActive, nearExpiryActive: refreshResult.nearExpiryActive, refreshPeers: refreshResult.refreshPeers, shouldRefresh: refreshResult.shouldRefresh}

1516 -- Fork federation I/O so hung peers don't block next prune cycle

1553 -- Fork federation I/O so hung peers don't block next prune cycle

1517 fork (do

1554 fork (do

1518 broadcastOfflineGossip t pr.peerServers pr.expiredLocal

1555 broadcastOfflineGossip t pr.peerServers pr.expiredLocal

@@ -1632,8 +1669,11 @@ pollOnce = \token cursor clientIp -> do

1632 case authenticateSession sessions token t of

1669 case authenticateSession sessions token t of

1633 Unauthorized {} -> yield (Nothing {})

1670 Unauthorized {} -> yield (Nothing {})

1634 Authorized {pubkey, ip} -> do

1671 Authorized {pubkey, ip} -> do

1635− when (clientIp != "" && ip != "" && clientIp != ip)

1672+ if clientIp != "" && ip != "" && clientIp != ip

1673+ then do

1636− (println ("poll: session IP mismatch for " ++ pubkey ++ " (session: " ++ ip ++ ", request: " ++ clientIp ++ ")"))

1674+ println ("poll: session IP mismatch for " ++ pubkey ++ " (session: " ++ ip ++ ", request: " ++ clientIp ++ "), rejecting")

1675+ yield (Nothing {})

1676+ else do

1637 -- Refresh session expiry (sliding window keeps active sessions alive)

1677 -- Refresh session expiry (sliding window keeps active sessions alive)

1638 writeSessionBucket bucket (do s <- sessions; yield (if safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip} else s))

1678 writeSessionBucket bucket (do s <- sessions; yield (if safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip} else s))

1639 seqCounter <- *seqCounter

1679 seqCounter <- *seqCounter

@@ -1885,7 +1925,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->

1885 let hasMatch = challengeMatch challenges pubkey challenge clientIp t

1925 let hasMatch = challengeMatch challenges pubkey challenge clientIp t

1886 -- Only consume challenge on IP match (rate limiter handles brute-force from same IP;

1926 -- Only consume challenge on IP match (rate limiter handles brute-force from same IP;

1887 -- not consuming on IP mismatch prevents remote attackers from burning challenges)

1927 -- not consuming on IP mismatch prevents remote attackers from burning challenges)

1888− when hasMatch (set *challenges = challengesExcluding challenges pubkey challenge)

1928+ when hasMatch (set *challenges = do c <- challenges; where c.pubkey != pubkey; yield c)

1889 if not hasMatch || not sigValid

1929 if not hasMatch || not sigValid

1890 then yield {valid: False {}}

1930 then yield {valid: False {}}

1891 else yield {valid: True {}}

1931 else yield {valid: True {}}

@@ -1943,21 +1983,23 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->

1943-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor

1983-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor

1944-- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.

1984-- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.

1945-- Acks for federated messages are sent to origin servers in the background.

1985-- Acks for federated messages are sent to origin servers in the background.

1946−handlePoll = \cursor authorization clientIp respond -> do

1986+handlePoll = \cursor generation authorization clientIp respond -> do

1947 t <- now

1987 t <- now

1988+ serverGen <- *serverGeneration

1989+ let effectiveCursor = if generation == serverGen then cursor else 0

1948 draining <- *draining

1990 draining <- *draining

1949− if draining then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}})

1991+ if draining then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen})

1950 else do

1992 else do

1951 globalRateLimited <- checkGlobalRate t

1993 globalRateLimited <- checkGlobalRate t

1952− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}})

1994+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen})

1953 else do

1995 else do

1954 result <- case extractToken authorization of

1996 result <- case extractToken authorization of

1955 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}

1997 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}

1956− Just {value: token} -> pollOnce token cursor clientIp

1998+ Just {value: token} -> pollOnce token effectiveCursor clientIp

1957 fork (when federationEnabled (do

1999 fork (when federationEnabled (do

1958 sendAcksAndClean result.pendingAcks

2000 sendAcksAndClean result.pendingAcks

1959 retryEvictedAcks result.evictedEntries result.pendingAcks))

2001 retryEvictedAcks result.evictedEntries result.pendingAcks))

1960− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}})

2002+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen})

1961

2003

1962-- POST /messages: Send an encrypted message to a recipient.

2004-- POST /messages: Send an encrypted message to a recipient.

1963-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.

2005-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.

@@ -2097,6 +2139,29 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do

2097 yield {server: maskedServer, authorized: True {}})

2139 yield {server: maskedServer, authorized: True {}})

2098 yield (respond {server: result.server, authorized: result.authorized})

2140 yield (respond {server: result.server, authorized: result.authorized})

2099

2141

2142+-- POST /auth/revoke: Revoke a specific session token.

2143+-- Caller authenticates via Authorization header, then specifies the target token to remove.

2144+-- Only removes the target if it belongs to the same pubkey as the caller (prevents cross-user revocation).

2145+handleRevoke = \targetToken authorization clientIp respond -> do

2146+ t <- now

2147+ globalRateLimited <- checkGlobalRate t

2148+ if globalRateLimited then yield (respond {ok: False {}, authorized: False {}})

2149+ else

2150+ withSessionAuth authorization t clientIp

2151+ (yield (respond {ok: False {}, authorized: False {}}))

2152+ (\callerPubkey -> do

2153+ let targetBucket = tokenBucket targetToken

2154+ removed <- atomic do

2155+ sessions <- readSessionBucket targetBucket

2156+ let hasTarget = any sessions (\s -> safeCompare s.token targetToken && s.pubkey == callerPubkey)

2157+ if hasTarget

2158+ then do

2159+ writeSessionBucket targetBucket (do s <- sessions; where not (safeCompare s.token targetToken && s.pubkey == callerPubkey); yield s)

2160+ yield (True {})

2161+ else yield (False {})

2162+ when removed (println ("revoke: session revoked for " ++ callerPubkey))

2163+ yield (respond {ok: removed, authorized: True {}}))

2164+

2100-- POST /federation/gossip: Receive presence gossip from a peer server.

2165-- POST /federation/gossip: Receive presence gossip from a peer server.

2101-- Validates events, merges into presence table (last-write-wins), and re-gossips

2166-- Validates events, merges into presence table (last-write-wins), and re-gossips

2102-- to other known peers with decremented hop count. Triggers message forwarding

2167-- to other known peers with decremented hop count. Triggers message forwarding

@@ -2263,6 +2328,20 @@ handleDrain = \authorization respond -> do

2263 (println ("drain: warning: " ++ show pendingAckCount ++ " pending acks will not be sent"))

2328 (println ("drain: warning: " ++ show pendingAckCount ++ " pending acks will not be sent"))

2264 yield (respond {ok: True {}, drained: drainedCount})

2329 yield (respond {ok: True {}, drained: drainedCount})

2265

2330

2331+-- POST /maintenance/undrain: Resume accepting requests after a drain.

2332+-- Background loops (prune, heartbeat) automatically resume on next iteration.

2333+handleUndrain = \authorization respond -> do

2334+ let authorized = verifyMaintenanceAuth authorization

2335+ if not authorized then yield (respond {ok: False {}})

2336+ else do

2337+ wasDraining <- atomic do

2338+ d <- *draining

2339+ set *draining = False {}

2340+ yield d

2341+ when wasDraining (println "undrain: server resumed, accepting new requests")

2342+ when (not wasDraining) (println "undrain: server was not draining")

2343+ yield (respond {ok: True {}})

2344+

2266-- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).

2345-- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).

2267-- Accepts either federation or maintenance secret.

2346-- Accepts either federation or maintenance secret.

2268handlePrune = \authorization respond -> do

2347handlePrune = \authorization respond -> do

@@ -2307,7 +2386,7 @@ handleHealth = \respond -> do

2307 capacityHealthy: result.capacityOk,

2386 capacityHealthy: result.capacityOk,

2308 federationHealthy: result.federationOk})

2387 federationHealthy: result.federationOk})

2309

2388

2310−zeroStats = {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pruneCount: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, ok: False {}}

2389+zeroStats = {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pruneCount: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, ok: False {}}

2311

2390

2312handleStats = \authorization respond -> do

2391handleStats = \authorization respond -> do

2313 let authorized = verifyMaintenanceAuth authorization

2392 let authorized = verifyMaintenanceAuth authorization

@@ -2343,9 +2422,17 @@ handleStats = \authorization respond -> do

2343 tmc <- *totalMessageCount

2422 tmc <- *totalMessageCount

2344 rc <- *recipientMsgCounts

2423 rc <- *recipientMsgCounts

2345 fi <- *federationInFlight

2424 fi <- *federationInFlight

2425+ flt <- *federationLatencyTotal

2426+ flc <- *federationLatencyCount

2427+ flm <- *federationLatencyMax

2346− yield {sc: sc, m: m, p: p, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, sr: sr, cr: cr, fr: fr, grc: grc, ir: ir, lr: lr, rr: rr, pc: pc, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, rg: rg, tmc: tmc, rc: rc, fi: fi}

2428+ yield {sc: sc, m: m, p: p, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, sr: sr, cr: cr, fr: fr, grc: grc, ir: ir, lr: lr, rr: rr, pc: pc, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, rg: rg, tmc: tmc, rc: rc, fi: fi, flt: flt, flc: flc, flm: flm}

2347− let result = {sessions: snapshot.sc, messages: count snapshot.m, presence: count snapshot.p, forwards: count snapshot.f, challenges: count snapshot.c, ackedIds: count snapshot.a, ackOverflow: count snapshot.ao, deliveredIds: count snapshot.d, forwardFailures: count snapshot.ff, activePolls: count snapshot.ap, sendRates: count snapshot.sr, challengeRates: count snapshot.cr, federationRates: count snapshot.fr, globalRateCount: snapshot.grc, ipRates: count snapshot.ir, lookupRates: count snapshot.lr, recvRates: count snapshot.rr, pruneCount: snapshot.pc, heartbeatCount: snapshot.hc, uptime: t - snapshot.sa, lastPruneAt: snapshot.lp, lastHeartbeatAt: snapshot.lh, draining: snapshot.dr, recentGossip: count snapshot.rg, totalMessageCount: snapshot.tmc, recipientCounts: count snapshot.rc, federationInFlight: count snapshot.fi}

2429+ let result = {sessions: snapshot.sc, messages: count snapshot.m, presence: count snapshot.p, forwards: count snapshot.f, challenges: count snapshot.c, ackedIds: count snapshot.a, ackOverflow: count snapshot.ao, deliveredIds: count snapshot.d, forwardFailures: count snapshot.ff, activePolls: count snapshot.ap, sendRates: count snapshot.sr, challengeRates: count snapshot.cr, federationRates: count snapshot.fr, globalRateCount: snapshot.grc, ipRates: count snapshot.ir, lookupRates: count snapshot.lr, recvRates: count snapshot.rr, pruneCount: snapshot.pc, heartbeatCount: snapshot.hc, uptime: t - snapshot.sa, lastPruneAt: snapshot.lp, lastHeartbeatAt: snapshot.lh, draining: snapshot.dr, recentGossip: count snapshot.rg, totalMessageCount: snapshot.tmc, recipientCounts: count snapshot.rc, federationInFlight: count snapshot.fi, federationLatencyTotal: snapshot.flt, federationLatencyCount: snapshot.flc, federationLatencyMax: snapshot.flm}

2430+ let msgPct = if maxMessages > 0 then result.messages * 100 / maxMessages else 0

2431+ let presPct = if maxPresenceEntries > 0 then result.presence * 100 / maxPresenceEntries else 0

2432+ let fwdPct = if maxForwardsEntries > 0 then result.forwards * 100 / maxForwardsEntries else 0

2433+ let sessPct = if maxSessions > 0 then result.sessions * 100 / maxSessions else 0

2434+ let fedAvg = if result.federationLatencyCount > 0 then result.federationLatencyTotal / result.federationLatencyCount else 0

2348− yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, draining: result.draining, federationEnabled: federationEnabled, recentGossip: result.recentGossip, totalMessageCount: result.totalMessageCount, recipientCounts: result.recipientCounts, federationInFlight: result.federationInFlight, ok: True {}})

2435+ yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, draining: result.draining, federationEnabled: federationEnabled, recentGossip: result.recentGossip, totalMessageCount: result.totalMessageCount, recipientCounts: result.recipientCounts, federationInFlight: result.federationInFlight, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, ok: True {}})

2349 else yield (respond zeroStats)

2436 else yield (respond zeroStats)

2350

2437

2351handleStatsPeers = \authorization respond -> do

2438handleStatsPeers = \authorization respond -> do

@@ -2381,7 +2468,8 @@ handleStatsKey = \pubkey authorization respond -> do

2381serve = \req -> case req of

2468serve = \req -> case req of

2382 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond

2469 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond

2383 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond

2470 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond

2471+ RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond

2384− Poll {cursor, authorization, clientIp, respond} -> handlePoll cursor authorization clientIp respond

2472+ Poll {cursor, generation, authorization, clientIp, respond} -> handlePoll cursor generation authorization clientIp respond

2385 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond

2473 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond

2386 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond

2474 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond

2387 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer authorization respond

2475 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer authorization respond

@@ -2389,6 +2477,7 @@ serve = \req -> case req of

2389 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> handleFederationAck messageIds originServer fromServer authorization respond

2477 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> handleFederationAck messageIds originServer fromServer authorization respond

2390 Prune {authorization, respond} -> handlePrune authorization respond

2478 Prune {authorization, respond} -> handlePrune authorization respond

2391 Drain {authorization, respond} -> handleDrain authorization respond

2479 Drain {authorization, respond} -> handleDrain authorization respond

2480+ Undrain {authorization, respond} -> handleUndrain authorization respond

2392 Health {respond} -> handleHealth respond

2481 Health {respond} -> handleHealth respond

2393 Stats {authorization, respond} -> handleStats authorization respond

2482 Stats {authorization, respond} -> handleStats authorization respond

2394 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond

2483 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond

@@ -2404,7 +2493,9 @@ backgroundPrune = do

2404 jitter <- randomInt 5000

2493 jitter <- randomInt 5000

2405 sleep (backgroundPruneInterval + jitter)

2494 sleep (backgroundPruneInterval + jitter)

2406 draining <- *draining

2495 draining <- *draining

2496+ if draining then do

2497+ sleep backgroundPruneInterval

2407− if draining then println "backgroundPrune: stopping (server draining)"

2498+ backgroundPrune

2408 else do

2499 else do

2409 t <- now

2500 t <- now

2410 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead

2501 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead

@@ -2417,7 +2508,7 @@ backgroundPrune = do

2417 set *lastHeartbeatAt = t)

2508 set *lastHeartbeatAt = t)

2418 status <- atomic do

2509 status <- atomic do

2419 running <- *pruneRunning

2510 running <- *pruneRunning

2420− if running > 0 && t - running < backgroundPruneInterval * 2

2511+ if running > 0 && t - running < backgroundPruneInterval * 3

2421 then yield {canRun: False {}, recovered: False {}, gen: 0}

2512 then yield {canRun: False {}, recovered: False {}, gen: 0}

2422 else do

2513 else do

2423 set *pruneRunning = t

2514 set *pruneRunning = t

@@ -2449,7 +2540,9 @@ pollHeartbeat = do

2449 if draining then do

2540 if draining then do

2450 t <- now

2541 t <- now

2451 atomic do set *pollWakeup = t

2542 atomic do set *pollWakeup = t

2543+ -- Don't exit; sleep and retry so undrain can resume the loop

2544+ sleep pollTimeoutMs

2452− println "pollHeartbeat: stopping (server draining)"

2545+ pollHeartbeat

2453 else do

2546 else do

2454 t <- now

2547 t <- now

2455 atomic do

2548 atomic do

@@ -2471,7 +2564,9 @@ main = do

2471 then println "FATAL: invalid federationPeerSecrets — server names must pass validation and secrets must be non-empty"

2564 then println "FATAL: invalid federationPeerSecrets — server names must pass validation and secrets must be non-empty"

2472 else do

2565 else do

2473 t <- now

2566 t <- now

2567+ atomic do

2474− atomic do set *startedAt = t

2568+ set *startedAt = t

2569+ set *serverGeneration = t

2475 println "Skrepka relay server starting on port 8080..."

2570 println "Skrepka relay server starting on port 8080..."

2476 println (" federation: " ++ if federationEnabled then "enabled" else "disabled")

2571 println (" federation: " ++ if federationEnabled then "enabled" else "disabled")

2477 println (" prune interval: " ++ show backgroundPruneInterval ++ "ms")

2572 println (" prune interval: " ++ show backgroundPruneInterval ++ "ms")