6718962 · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Improve relay resilience: drain poll wake, rate-limit field, stats timing, forward backoff, server cooldown

6718962iko6 months ago

server.knot+29−13

compare with master

unifiedsplit

@@ -81,6 +81,8 @@ retryBackoffBase = 30000 -- 30 seconds

81retryBackoffMax : Int<Ms>

82retryBackoffMax = 480000 -- 8 minutes

83maxForwardFailures = 1000 -- max tracked server failure records

84+maxServerDowntime : Int<Ms>

85+maxServerDowntime = 86400000 -- 24 hours; stop retrying a server after this duration of consecutive failures

85-- Background task intervals

87-- Background task intervals

86backgroundPruneInterval : Int<Ms>

88backgroundPruneInterval : Int<Ms>

@@ -178,7 +180,7 @@ type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp

178-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)

180-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)

179type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}

181type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}

180-- Tracks consecutive failures per federation peer for exponential backoff

182-- Tracks consecutive failures per federation peer for exponential backoff

181−type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}

183+type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFailedAt: Timestamp}

182-- Sliding-window rate limit state per pubkey

184-- Sliding-window rate limit state per pubkey

183type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

185type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

184type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

186type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

@@ -264,6 +266,10 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

264

266

265*heartbeatCount : Int

267*heartbeatCount : Int

266

268

269+*startedAt : Timestamp

270+

271+*lastHeartbeatAt : Timestamp

272+

267-- ============================================

273-- ============================================

268-- Routes

274-- Routes

269-- ============================================

275-- ============================================

@@ -285,7 +291,7 @@ route AuthApi where

285route ClientApi where

291route ClientApi where

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

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

287 POST {cursor: Timestamp} /poll headers {authorization: Text}

293 POST {cursor: Timestamp} /poll headers {authorization: Text}

288− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool}

294+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool}

289 = Poll

295 = Poll

290 /messages

296 /messages

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

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

@@ -328,7 +334,7 @@ route HealthApi where

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

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

329route StatsApi where

335route StatsApi where

330 GET /stats headers {authorization: Text}

336 GET /stats headers {authorization: Text}

331− -> {sessions: Int, messages: Int, presence: Int, forwards: Int, challenges: Int, ackedIds: Int, ackOverflow: Int, deliveredIds: Int, forwardFailures: Int, activePolls: Int, sendRates: Int, challengeRates: Int, federationRates: Int, pruneCount: Int, heartbeatCount: Int, ok: Bool}

337+ -> {sessions: Int, messages: Int, presence: Int, forwards: Int, challenges: Int, ackedIds: Int, ackOverflow: Int, deliveredIds: Int, forwardFailures: Int, activePolls: Int, sendRates: Int, challengeRates: Int, federationRates: Int, pruneCount: Int, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, ok: Bool}

332 = Stats

338 = Stats

333

339

@@ -803,7 +809,7 @@ pruneExpired = \t -> do

803 sendRates <- *sendRates

809 sendRates <- *sendRates

804 set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r

810 set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r

805 forwardFailures <- *forwardFailures

811 forwardFailures <- *forwardFailures

806− set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < retryBackoffMax * 2; yield f

812+ set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f

807 challengeRates <- *challengeRates

813 challengeRates <- *challengeRates

808 set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r

814 set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r

809 federationRates <- *federationRates

815 federationRates <- *federationRates

@@ -950,7 +956,7 @@ shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool

950shouldRetryServer = \server t failures ->

956shouldRetryServer = \server t failures ->

951 case headOf (do f <- failures; where f.server == server; yield f) of

957 case headOf (do f <- failures; where f.server == server; yield f) of

952 Nothing {} -> True {}

958 Nothing {} -> True {}

953− Just {value: f} -> t - f.failedAt >= retryBackoff f.failures

959+ Just {value: f} -> t - f.firstFailedAt < maxServerDowntime && t - f.failedAt >= retryBackoff f.failures

954

960

955recordFailure = \server -> do

961recordFailure = \server -> do

956 t <- now

962 t <- now

@@ -958,8 +964,8 @@ recordFailure = \server -> do

958 failures <- *forwardFailures

964 failures <- *forwardFailures

959 let existing = headOf (do f <- failures; where f.server == server; yield f)

965 let existing = headOf (do f <- failures; where f.server == server; yield f)

960 let updated = case existing of

966 let updated = case existing of

961− Nothing {} -> union failures [{server: server, failedAt: t, failures: 1}]

967+ Nothing {} -> union failures [{server: server, failedAt: t, failures: 1, firstFailedAt: t}]

962− Just {value: f} -> union (do x <- failures; where x.server != server; yield x) [{server: server, failedAt: t, failures: min (f.failures + 1) 10}]

968+ Just {value: f} -> union (do x <- failures; where x.server != server; yield x) [{server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}]

963 set *forwardFailures = if count updated > maxForwardFailures then evictOneOldest updated (\x -> x.failedAt) else updated

969 set *forwardFailures = if count updated > maxForwardFailures then evictOneOldest updated (\x -> x.failedAt) else updated

964

970

965clearFailure = \server -> atomic do

971clearFailure = \server -> atomic do

@@ -1012,7 +1018,9 @@ forwardQueuedMessages = \targets t -> do

1012 set *forwards = updatedForwards

1018 set *forwards = updatedForwards

1013 yield capped

1019 yield capped

1014 let servers = collectPeerServers newForwards (\x -> x.toServer)

1020 let servers = collectPeerServers newForwards (\x -> x.toServer)

1021+ failures <- *forwardFailures

1022+ let eligibleServers = do s <- servers; where shouldRetryServer s t failures; yield s

1015− sendToEach servers (\server ->

1023+ sendToEach eligibleServers (\server ->

1016 retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)

1024 retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)

1017

1025

1018fetchFederationAck = \origin messageIds ->

1026fetchFederationAck = \origin messageIds ->

@@ -1512,7 +1520,7 @@ verifyAuthInner = \pubkey challenge signature respond t -> do

1512handlePoll = \cursor authorization respond -> do

1520handlePoll = \cursor authorization respond -> do

1513 t <- now

1521 t <- now

1514 globalRateLimited <- checkGlobalRate t

1522 globalRateLimited <- checkGlobalRate t

1515− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}})

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

1516 else do

1524 else do

1517 result <- case extractToken authorization of

1525 result <- case extractToken authorization of

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

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

@@ -1522,7 +1530,7 @@ handlePoll = \cursor authorization respond -> do

1522 fork (when federationEnabled (do

1530 fork (when federationEnabled (do

1523 sendAcksAndClean result.pendingAcks

1531 sendAcksAndClean result.pendingAcks

1524 retryEvictedAcks result.evictedEntries result.pendingAcks))

1532 retryEvictedAcks result.evictedEntries result.pendingAcks))

1525− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})

1533+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}})

1526

1534

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

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

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

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

@@ -1722,6 +1730,7 @@ handleDrain = \authorization respond -> do

1722 presence <- *presence

1730 presence <- *presence

1723 let localEntries = do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p

1731 let localEntries = do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p

1724 set *presence = do p <- presence; where not (isLocalServer p.server) || p.expiresAt <= t; yield p

1732 set *presence = do p <- presence; where not (isLocalServer p.server) || p.expiresAt <= t; yield p

1733+ set *pollWakeup = t

1725 yield {localEntries: localEntries, peerServers: collectPeerServers presence (\p -> p.server)}

1734 yield {localEntries: localEntries, peerServers: collectPeerServers presence (\p -> p.server)}

1726 let drainedCount = count drainResult.localEntries

1735 let drainedCount = count drainResult.localEntries

1727 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do

1736 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do

@@ -1744,6 +1753,7 @@ handleStats = \authorization respond -> do

1744 let authorized = verifyMaintenanceAuth authorization

1753 let authorized = verifyMaintenanceAuth authorization

1745 if authorized

1754 if authorized

1746 then do

1755 then do

1756+ t <- now

1747 result <- atomic do

1757 result <- atomic do

1748 s <- *sessions

1758 s <- *sessions

1749 m <- *messages

1759 m <- *messages

@@ -1760,9 +1770,12 @@ handleStats = \authorization respond -> do

1760 fr <- *federationRates

1770 fr <- *federationRates

1761 pc <- *pruneCount

1771 pc <- *pruneCount

1762 hc <- *heartbeatCount

1772 hc <- *heartbeatCount

1773+ sa <- *startedAt

1774+ lp <- *lastPrunedAt

1775+ lh <- *lastHeartbeatAt

1763− 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, pruneCount: pc, heartbeatCount: hc}

1776+ 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, pruneCount: pc, heartbeatCount: hc, uptime: t - sa, lastPruneAt: lp, lastHeartbeatAt: lh}

1764− yield (respond {sessions: result.sessions, messages: result.messages, presence: result.presence, forwards: result.forwards, challenges: result.challenges, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, ok: True {}})

1777+ yield (respond {sessions: result.sessions, messages: result.messages, presence: result.presence, forwards: result.forwards, challenges: result.challenges, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, ok: True {}})

1765− else yield (respond {sessions: 0, messages: 0, presence: 0, forwards: 0, challenges: 0, ackedIds: 0, ackOverflow: 0, deliveredIds: 0, forwardFailures: 0, activePolls: 0, sendRates: 0, challengeRates: 0, federationRates: 0, pruneCount: 0, heartbeatCount: 0, ok: False {}})

1778+ else yield (respond {sessions: 0, messages: 0, presence: 0, forwards: 0, challenges: 0, ackedIds: 0, ackOverflow: 0, deliveredIds: 0, forwardFailures: 0, activePolls: 0, sendRates: 0, challengeRates: 0, federationRates: 0, pruneCount: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, ok: False {}})

1766

1779

1767serve = \req -> case req of

1780serve = \req -> case req of

1768 ReqChallenge {pubkey, respond} -> handleChallenge pubkey respond

1781 ReqChallenge {pubkey, respond} -> handleChallenge pubkey respond

@@ -1802,11 +1815,14 @@ pollHeartbeat = do

1802 t <- now

1815 t <- now

1803 atomic do

1816 atomic do

1804 set *pollWakeup = t

1817 set *pollWakeup = t

1818+ set *lastHeartbeatAt = t

1805 hc <- *heartbeatCount

1819 hc <- *heartbeatCount

1806 set *heartbeatCount = hc + 1

1820 set *heartbeatCount = hc + 1

1807 pollHeartbeat

1821 pollHeartbeat

1808

1822

1809main = do

1823main = do

1824+ t <- now

1825+ atomic do set *startedAt = t

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

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

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

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

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

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