bfa417f · skrepka · git.iko.soy

git.iko.soy/skrepka

Improve reliability and observability: global message cap, per-peer federation secrets, per-key stats, config validation, pruneRunning recovery

bfa417fiko6 months ago

server.knot+82−26

compare with master

unifiedsplit

@@ -18,6 +18,10 @@ serverHost = "localhost:8080"

18federationSecret = "" -- must be set to enable federation

19maintenanceSecret = "" -- set to enable /maintenance/prune independently of federation

20federationScheme = "https://" -- scheme for federation HTTP calls

21+-- Per-peer federation secrets (overrides global federationSecret for specific peers)

22+-- Format: [{server: "peer.example.com", secret: "peer-specific-secret"}]

23+federationPeerSecrets : [{server: Text, secret: Text}]

24+federationPeerSecrets = []

21-- Message and data size limits

25-- Message and data size limits

22messageMaxAge : Int<Ms>

26messageMaxAge : Int<Ms>

23messageMaxAge = 2592000000 -- 30 days

27messageMaxAge = 2592000000 -- 30 days

@@ -31,6 +35,7 @@ maxGossipEvents = 100

31maxPollEvents = 50

35maxPollEvents = 50

32maxPresenceEntries = 10000

36maxPresenceEntries = 10000

33maxMessagesPerRecipient = 500

37maxMessagesPerRecipient = 500

38+maxMessages = 100000 -- global cap on total stored messages across all recipients

34-- Global capacity limits for in-memory collections

39-- Global capacity limits for in-memory collections

35clockSkewTolerance : Int<Ms>

40clockSkewTolerance : Int<Ms>

36clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers

41clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers

@@ -225,6 +230,7 @@ type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

225-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles

230-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles

226type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}

231type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}

227type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool}

232type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool}

233+type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}

228

234

229-- ============================================

235-- ============================================

230-- Persisted State

236-- Persisted State

@@ -380,11 +386,14 @@ route HealthApi where

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

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

381route StatsApi where

387route StatsApi where

382 GET /stats headers {authorization: Text}

388 GET /stats headers {authorization: Text}

383− -> {sessions: Int, sessionsMax: Int, messages: 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, ok: Bool}

389+ -> {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, ok: Bool}

384 = Stats

390 = Stats

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

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

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

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

387 = StatsPeers

393 = StatsPeers

394+ GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text}

395+ -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, activePolls: Int, ok: Bool}

396+ = StatsKey

388

397

390

399

@@ -639,6 +648,15 @@ verifyAuthWith = \check authorization ->

639verifyFederationAuth : Text -> Bool

648verifyFederationAuth : Text -> Bool

640verifyFederationAuth = verifyAuthWith (\token -> verifySecret token federationSecret)

649verifyFederationAuth = verifyAuthWith (\token -> verifySecret token federationSecret)

641

650

651+-- Per-peer federation auth: checks peer-specific secret first, falls back to global

652+verifyFederationAuthForServer : Text -> Text -> Bool

653+verifyFederationAuthForServer = \fromServer authorization ->

654+ verifyAuthWith (\token ->

655+ case headOf (do p <- federationPeerSecrets; where normalizeServerName p.server == fromServer; yield p) of

656+ Just {value: p} -> verifySecret token p.secret

657+ Nothing {} -> verifySecret token federationSecret

658+ ) authorization

659+

642verifyMaintenanceAuth : Text -> Bool

660verifyMaintenanceAuth : Text -> Bool

643verifyMaintenanceAuth = verifyAuthWith (\token -> verifySecret token federationSecret || verifySecret token maintenanceSecret)

661verifyMaintenanceAuth = verifyAuthWith (\token -> verifySecret token federationSecret || verifySecret token maintenanceSecret)

644

662

@@ -664,10 +682,11 @@ nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1

664-- Deduplicate message IDs, preserving order, capped at maxMessageIds

682-- Deduplicate message IDs, preserving order, capped at maxMessageIds

665dedupMessageIds : [MessageIdHex] -> [MessageIdHex]

683dedupMessageIds : [MessageIdHex] -> [MessageIdHex]

666dedupMessageIds = \messageIds ->

684dedupMessageIds = \messageIds ->

667− fold (\acc mid ->

685+ let result = fold (\acc mid ->

668− if count acc >= maxMessageIds then acc

686+ if acc.n >= maxMessageIds then acc

669− else if any acc (\a -> a == mid) then acc

687+ else if any acc.items (\a -> a == mid) then acc

670− else union acc [mid]) [] messageIds

688+ else {n: acc.n + 1, items: union acc.items [mid]}) {n: 0, items: []} messageIds

689+ in result.items

671

690

672-- Remove the N oldest acked entries (by ackedAt timestamp)

691-- Remove the N oldest acked entries (by ackedAt timestamp)

673evictOldestAckedIds = \items n ->

692evictOldestAckedIds = \items n ->

@@ -1001,7 +1020,7 @@ checkPreAuth = \clientIp t -> do

1001 else yield ""

1020 else yield ""

1002

1021

1003withFederationAuth = \authorization t respond fromServer precondition action ->

1022withFederationAuth = \authorization t respond fromServer precondition action ->

1004− if not (verifyFederationAuth authorization)

1023+ if not (verifyFederationAuthForServer fromServer authorization)

1005 then do

1024 then do

1006 maybePruneAndBroadcast t

1025 maybePruneAndBroadcast t

1007 yield (respond {ok: False {}, error: "unauthorized"})

1026 yield (respond {ok: False {}, error: "unauthorized"})

@@ -1249,8 +1268,9 @@ hexCharIndex = \c ->

1249-- Map pubkey to one of 16 signal buckets based on first hex digit.

1268-- Map pubkey to one of 16 signal buckets based on first hex digit.

1250-- This partitions the signal space so storing a message for one pubkey

1269-- This partitions the signal space so storing a message for one pubkey

1251-- only wakes polls in the same bucket, not all active polls.

1270-- only wakes polls in the same bucket, not all active polls.

1271+-- Defensive clamp ensures an invalid first char never indexes out of bounds

1252pubkeyBucket : PubkeyHex -> Int

1272pubkeyBucket : PubkeyHex -> Int

1253−pubkeyBucket = \pubkey -> hexCharIndex (take 1 pubkey)

1273+pubkeyBucket = \pubkey -> clamp 0 15 (hexCharIndex (take 1 pubkey))

1254

1274

1255-- Update the signal timestamp for a pubkey's bucket, waking any blocked polls in that bucket

1275-- Update the signal timestamp for a pubkey's bucket, waking any blocked polls in that bucket

1256signalMessageForKey = \pubkey ts ->

1276signalMessageForKey = \pubkey ts ->

@@ -1452,7 +1472,7 @@ forwardCandidates = \targets messages forwards t ->

1452

1472

1453

1473

1454challengeMatch = \challenges pubkey challenge ip t ->

1474challengeMatch = \challenges pubkey challenge ip t ->

1455− any challenges (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && (ip == "" || c.ip == "" || c.ip == ip))

1475+ any challenges (\c -> c.pubkey == pubkey && safeCompare c.challenge challenge && c.expiresAt > t && (ip == "" || c.ip == "" || c.ip == ip))

1456

1476

1457challengesExcluding = \challenges pubkey challenge ->

1477challengesExcluding = \challenges pubkey challenge ->

1458 do c <- challenges; where c.pubkey != pubkey || c.challenge != challenge; yield c

1478 do c <- challenges; where c.pubkey != pubkey || c.challenge != challenge; yield c

@@ -1680,7 +1700,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond

1680 then yield {pr: pr, send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}}

1700 then yield {pr: pr, send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}}

1681 else do

1701 else do

1682 set *recvRates = recvResult.entries

1702 set *recvRates = recvResult.entries

1703+ if count messages >= maxMessages

1704+ then yield {pr: pr, send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}}

1683− if countMessagesFor messages to t >= maxMessagesPerRecipient

1705+ else if countMessagesFor messages to t >= maxMessagesPerRecipient

1684 then yield {pr: pr, send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}}

1706 then yield {pr: pr, send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}}

1685 else do

1707 else do

1686 presence <- *presence

1708 presence <- *presence

@@ -1817,6 +1839,8 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

1817 deliveredIds <- *deliveredIds

1839 deliveredIds <- *deliveredIds

1818 if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds

1840 if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds

1819 then yield {pr: pr, forward: ForwardDuplicate {}}

1841 then yield {pr: pr, forward: ForwardDuplicate {}}

1842+ else if count messages >= maxMessages

1843+ then yield {pr: pr, forward: ForwardRejected {}}

1820 else if countMessagesFor messages toKey t >= maxMessagesPerRecipient

1844 else if countMessagesFor messages toKey t >= maxMessagesPerRecipient

1821 then yield {pr: pr, forward: ForwardRejected {}}

1845 then yield {pr: pr, forward: ForwardRejected {}}

1822 else do

1846 else do

@@ -1919,9 +1943,11 @@ handleHealth = \respond -> do

1919 forwards <- *forwards

1943 forwards <- *forwards

1920 sessions <- *sessions

1944 sessions <- *sessions

1921 activePolls <- *activePolls

1945 activePolls <- *activePolls

1946+ messages <- *messages

1922 let warmup = t - sa < healthPruneMaxAge

1947 let warmup = t - sa < healthPruneMaxAge

1923 -- Capacity pressure: flag when any capped collection exceeds 90% full

1948 -- Capacity pressure: flag when any capped collection exceeds 90% full

1949+ let capacityOk = count messages * 10 <= maxMessages * 9

1924− let capacityOk = count presence * 10 <= maxPresenceEntries * 9

1950+ && count presence * 10 <= maxPresenceEntries * 9

1925 && count forwards * 10 <= maxForwardsEntries * 9

1951 && count forwards * 10 <= maxForwardsEntries * 9

1926 && count sessions * 10 <= maxSessions * 9

1952 && count sessions * 10 <= maxSessions * 9

1927 && count activePolls * 10 <= maxActivePolls * 9

1953 && count activePolls * 10 <= maxActivePolls * 9

@@ -1962,8 +1988,8 @@ handleStats = \authorization respond -> do

1962 lp <- *lastPrunedAt

1988 lp <- *lastPrunedAt

1963 lh <- *lastHeartbeatAt

1989 lh <- *lastHeartbeatAt

1964 yield {sessions: count s, messages: count m, presence: count p, forwards: count f, challenges: count c, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff, activePolls: count ap, sendRates: count sr, challengeRates: count cr, federationRates: count fr, globalRateCount: grc, ipRates: count ir, lookupRates: count lr, recvRates: count rr, pruneCount: pc, heartbeatCount: hc, uptime: t - sa, lastPruneAt: lp, lastHeartbeatAt: lh}

1990 yield {sessions: count s, messages: count m, presence: count p, forwards: count f, challenges: count c, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff, activePolls: count ap, sendRates: count sr, challengeRates: count cr, federationRates: count fr, globalRateCount: grc, ipRates: count ir, lookupRates: count lr, recvRates: count rr, pruneCount: pc, heartbeatCount: hc, uptime: t - sa, lastPruneAt: lp, lastHeartbeatAt: lh}

1965− yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, 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, ok: True {}})

1991+ 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, ok: True {}})

1966− else yield (respond {sessions: 0, sessionsMax: 0, messages: 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, ok: False {}})

1992+ else yield (respond {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, ok: False {}})

1967

1993

1968handleStatsPeers = \authorization respond -> do

1994handleStatsPeers = \authorization respond -> do

1969 let authorized = verifyMaintenanceAuth authorization

1995 let authorized = verifyMaintenanceAuth authorization

@@ -1975,6 +2001,26 @@ handleStatsPeers = \authorization respond -> do

1975 yield (respond {peers: peers, ok: True {}})

2001 yield (respond {peers: peers, ok: True {}})

1976 else yield (respond {peers: [], ok: False {}})

2002 else yield (respond {peers: [], ok: False {}})

1977

2003

2004+handleStatsKey = \pubkey authorization respond -> do

2005+ let authorized = verifyMaintenanceAuth authorization

2006+ if authorized

2007+ then do

2008+ t <- now

2009+ result <- atomic do

2010+ messages <- *messages

2011+ presence <- *presence

2012+ sessions <- *sessions

2013+ forwards <- *forwards

2014+ activePolls <- *activePolls

2015+ let keyMessages = count (do m <- messages; where m.toKey == pubkey; yield m)

2016+ let keyPresence = do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield {server: p.server, expiresAt: p.expiresAt}

2017+ let keySessions = count (do s <- sessions; where s.pubkey == pubkey; yield s)

2018+ let keyForwards = count (do f <- forwards; where f.toKey == pubkey; yield f)

2019+ let keyPolls = count (do p <- activePolls; where p.pubkey == pubkey; yield p)

2020+ yield {messages: keyMessages, presence: keyPresence, sessions: keySessions, forwards: keyForwards, activePolls: keyPolls}

2021+ yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})

2022+ else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})

2023+

1978serve = \req -> case req of

2024serve = \req -> case req of

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

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

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

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

@@ -1989,6 +2035,7 @@ serve = \req -> case req of

1989 Health {respond} -> handleHealth respond

2035 Health {respond} -> handleHealth respond

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

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

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

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

2038+ StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond

1992

2039

1993-- ============================================

2040-- ============================================

1994-- Entry Point

2041-- Entry Point

@@ -2003,16 +2050,18 @@ backgroundPrune = do

2003 if draining then println "backgroundPrune: stopping (server draining)"

2050 if draining then println "backgroundPrune: stopping (server draining)"

2004 else do

2051 else do

2005 t <- now

2052 t <- now

2006− canRun <- atomic do

2053+ status <- atomic do

2007 running <- *pruneRunning

2054 running <- *pruneRunning

2008− if running > 0 && t - running < backgroundPruneInterval * 5

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

2009− then yield (False {})

2056+ then yield {canRun: False {}, recovered: False {}}

2010 else do

2057 else do

2011 set *pruneRunning = t

2058 set *pruneRunning = t

2012 pc <- *pruneCount

2059 pc <- *pruneCount

2013 set *pruneCount = pc + 1

2060 set *pruneCount = pc + 1

2014− yield (True {})

2061+ yield {canRun: True {}, recovered: running > 0}

2062+ when status.recovered

2063+ (println "backgroundPrune: previous cycle appears stuck, forcing new cycle")

2015− if canRun

2064+ if status.canRun

2016 then fork (do

2065 then fork (do

2017 pruneAndRetry t

2066 pruneAndRetry t

2018 atomic do set *pruneRunning = 0)

2067 atomic do set *pruneRunning = 0)

@@ -2039,12 +2088,19 @@ pollHeartbeat = do

2039 pollHeartbeat

2088 pollHeartbeat

2040

2089

2041main = do

2090main = do

2091+ let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0

2092+ && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0

2093+ && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0

2094+ && maxPresenceTtl > 0 && maxForwardsEntries > 0 && maxAckedIds > 0

2095+ if not configOk

2096+ then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"

2097+ else do

2042− t <- now

2098+ t <- now

2043− atomic do set *startedAt = t

2099+ atomic do set *startedAt = t

2044− println "Skrepka relay server starting on port 8080..."

2100+ println "Skrepka relay server starting on port 8080..."

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

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

2046− println (" prune interval: " ++ show backgroundPruneInterval ++ "ms")

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

2047− println (" poll timeout: " ++ show pollTimeoutMs ++ "ms")

2103+ println (" poll timeout: " ++ show pollTimeoutMs ++ "ms")

2048− fork backgroundPrune

2104+ fork backgroundPrune

2049− fork pollHeartbeat

2105+ fork pollHeartbeat

2050− listen 8080 serve

2106+ listen 8080 serve