@@ -303,7 +303,6 @@ type PresenceEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Timestam
303-- Identity is derived from `encryptedBlob` (see `Message`).
303-- Identity is derived from `encryptedBlob` (see `Message`).
304type ForwardEntry = {
304type ForwardEntry = {
305 toServer: ServerName,
305 toServer: ServerName,
306− createdAt: Timestamp,
307 toKey: PubkeyHex,
306 toKey: PubkeyHex,
308 encryptedBlob: BlobHex,
307 encryptedBlob: BlobHex,
309 retries: Int
308 retries: Int
@@ -458,9 +457,9 @@ type RecipientCount = {pubkey: PubkeyHex, count: Int}
458route StatsApi where
457route StatsApi where
459 GET /stats headers {authorization: Text} -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, totalMessageCount: Int, presence: Int, forwards: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, nonceEvictions: Int, activePolls: Int, recipientCounts: Int, messageBucketMax: Int, messageBucketMin: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}], gossipSuccess: Int, gossipFail: Int, latencyBuckets: {lt100: Int, lt500: Int, lt2000: Int, gte2000: Int}}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, pollLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int, sendLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, authAvgMs: Int, authMaxMs: Int, authCount: Int}} = Stats
458 GET /stats headers {authorization: Text} -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, totalMessageCount: Int, presence: Int, forwards: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, nonceEvictions: Int, activePolls: Int, recipientCounts: Int, messageBucketMax: Int, messageBucketMin: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}], gossipSuccess: Int, gossipFail: Int, latencyBuckets: {lt100: Int, lt500: Int, lt2000: Int, gte2000: Int}}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, pollLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int, sendLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, authAvgMs: Int, authMaxMs: Int, authCount: Int}} = Stats
460 GET /stats/peers headers {authorization: Text} -> {peers: [PeerStatus], ok: Bool} = StatsPeers
459 GET /stats/peers headers {authorization: Text} -> {peers: [PeerStatus], ok: Bool} = StatsPeers
461− GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text} -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, forwardDetails: [{server: Text, retries: Int, createdAt: Timestamp}], activePolls: Int, ok: Bool} = StatsKey
460+ GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text} -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, forwardDetails: [{server: Text, retries: Int}], activePolls: Int, ok: Bool} = StatsKey
462 POST {limit: Int} /stats/top-recipients headers {authorization: Text} -> {recipients: [RecipientCount], ok: Bool, total: Int} = TopRecipients
461 POST {limit: Int} /stats/top-recipients headers {authorization: Text} -> {recipients: [RecipientCount], ok: Bool, total: Int} = TopRecipients
463− GET /stats/forwards headers {authorization: Text} -> {servers: [{server: Text, count: Int, oldestAge: Int, totalRetries: Int}], ok: Bool, total: Int} = StatsForwards
462+ GET /stats/forwards headers {authorization: Text} -> {servers: [{server: Text, count: Int, totalRetries: Int}], ok: Bool, total: Int} = StatsForwards
464 GET /stats/rates headers {authorization: Text} -> {ok: Bool, scalePct: Int, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int} = StatsRates
463 GET /stats/rates headers {authorization: Text} -> {ok: Bool, scalePct: Int, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int} = StatsRates
466route MetricsApi where
465route MetricsApi where
@@ -972,7 +971,7 @@ knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.serve
972-- One sweep over every state ref: drops expired rows in a single atomic block.
971-- One sweep over every state ref: drops expired rows in a single atomic block.
973-- Returns the locally-expired presence rows whose pubkey has no other live
972-- Returns the locally-expired presence rows whose pubkey has no other live
974-- local session, so the caller can broadcast offline gossip for them.
973-- local session, so the caller can broadcast offline gossip for them.
975−pruneExpired : Int<Ms> -> IO {rw *activePolls, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, rw *sessions} [{expiresAt: Timestamp, pubkey: PubkeyHex, server: ServerName}]
974+pruneExpired : Int<Ms> -> IO {rw *activePolls, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *messages, rw *presence, rw *sessions} [{expiresAt: Timestamp, pubkey: PubkeyHex, server: ServerName}]
976pruneExpired = \t -> atomic do
975pruneExpired = \t -> atomic do
977 ch <- *challenges
976 ch <- *challenges
978 *challenges = filter ch (\x -> x.expiresAt > t)
977 *challenges = filter ch (\x -> x.expiresAt > t)
@@ -980,8 +979,6 @@ pruneExpired = \t -> atomic do
980 *sessions = filter ses (\x -> x.expiresAt > t)
979 *sessions = filter ses (\x -> x.expiresAt > t)
981 msgs <- *messages
980 msgs <- *messages
982 *messages = filter msgs (\m -> t - m.receivedAt < messageMaxAge)
981 *messages = filter msgs (\m -> t - m.receivedAt < messageMaxAge)
984− *forwards = filter fwds (\f -> t - f.createdAt < messageMaxAge)
985 dids <- *deliveredIds
982 dids <- *deliveredIds
986 *deliveredIds = filter dids (\d -> t - d.deliveredAt < messageMaxAge)
983 *deliveredIds = filter dids (\d -> t - d.deliveredAt < messageMaxAge)
987 ff <- *forwardFailures
984 ff <- *forwardFailures
@@ -1074,18 +1071,17 @@ retryPendingForwards = do
1074 atomic (do
1071 atomic (do
1075 cur <- *forwards
1072 cur <- *forwards
1076 *forwards = pluck (removeWhere cur isDone) (\x -> if isMatch x
1073 *forwards = pluck (removeWhere cur isDone) (\x -> if isMatch x
1077− then {x.toServer, x.createdAt, x.toKey, x.encryptedBlob, retries: x.retries + 1}
1074+ then {x.toServer, x.toKey, x.encryptedBlob, retries: x.retries + 1}
1078 else x))))
1075 else x))))
1080-- Forward queued messages to a server when a remote key comes online.
1077-- Forward queued messages to a server when a remote key comes online.
1081−forwardQueuedToServer : PubkeyHex -> ServerName -> Int<Ms> -> IO {rw *forwards, r *messages} {}
1078+forwardQueuedToServer : PubkeyHex -> ServerName -> IO {rw *forwards, r *messages} {}
1082−forwardQueuedToServer = \pubkey server t -> atomic (do
1079+forwardQueuedToServer = \pubkey server -> atomic (do
1083 msgs <- *messages
1080 msgs <- *messages
1084 fwds <- *forwards
1081 fwds <- *forwards
1085 let candidates = filter msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.encryptedBlob == m.encryptedBlob && f.toServer == server))
1082 let candidates = filter msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.encryptedBlob == m.encryptedBlob && f.toServer == server))
1086 let newFwds = pluck (takeRelation maxForwardsPerServer candidates) (\m -> {
1083 let newFwds = pluck (takeRelation maxForwardsPerServer candidates) (\m -> {
1087 toServer: server,
1084 toServer: server,
1089 m.toKey,
1085 m.toKey,
1090 m.encryptedBlob,
1086 m.encryptedBlob,
1091 retries: 0
1087 retries: 0
@@ -1441,7 +1437,6 @@ sendMessageCore = \senderKey to encryptedBlob t -> do
1441 fwds <- *forwards
1437 fwds <- *forwards
1442 let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s)) (\s -> {
1438 let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s)) (\s -> {
1443 toServer: s,
1439 toServer: s,
1445 toKey: to,
1440 toKey: to,
1446 encryptedBlob,
1441 encryptedBlob,
1447 retries: 0
1442 retries: 0
@@ -1659,7 +1654,7 @@ handleRecvGossip = \events fromServer requestTimestamp authorization respond ->
1659 let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys
1654 let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys
1660 *presence = union kept added
1655 *presence = union kept added
1661 yield (filter onlinePubkeys (\pk -> none pres (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t))))
1656 yield (filter onlinePubkeys (\pk -> none pres (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t))))
1662− fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer t))
1657+ fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer))
1663 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
1658 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
1665-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.
1660-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.
@@ -1780,7 +1775,7 @@ handleShutdown = \authorization respond -> handleDrain 2 authorization (\r -> re
1780-- POST /maintenance/prune: trigger an out-of-band `pruneAndRetry` tick.
1775-- POST /maintenance/prune: trigger an out-of-band `pruneAndRetry` tick.
1781-- Useful when the operator wants to force expiry/forward retries without
1776-- Useful when the operator wants to force expiry/forward retries without
1782-- waiting for the next `backgroundPrune` interval.
1777-- waiting for the next `backgroundPrune` interval.
1783−handlePrune : Text -> ({ok: Bool} -> a) -> IO {r *runtimeMaintenanceSecret, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *sessions, clock} a
1778+handlePrune : Text -> ({ok: Bool} -> a) -> IO {r *runtimeMaintenanceSecret, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, r *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *sessions, clock} a
1784handlePrune = \authorization respond -> do
1779handlePrune = \authorization respond -> do
1785 t <- now
1780 t <- now
1786 authorized <- verifyMaintenanceAuth authorization
1781 authorized <- verifyMaintenanceAuth authorization
@@ -2294,7 +2289,7 @@ handleStatsPeers = \authorization respond -> withMaintAuth authorization (yield
2295-- GET /stats/key/{pubkey}: per-key footprint — message count, presence rows,
2290-- GET /stats/key/{pubkey}: per-key footprint — message count, presence rows,
2296-- session count, queued forwards (with details), and active long-poll count.
2291-- session count, queued forwards (with details), and active long-poll count.
2297−handleStatsKey : PubkeyHex -> Text -> ({activePolls: Int, forwardDetails: [{createdAt: Int<Ms>, retries: Int, server: ServerName}], forwards: Int, messages: Int, ok: Bool, presence: [{expiresAt: Int<Ms>, server: ServerName}], sessions: Int} -> a) -> IO {r *activePolls, r *forwards, r *messages, r *presence, r *runtimeMaintenanceSecret, r *sessions, clock} a
2292+handleStatsKey : PubkeyHex -> Text -> ({activePolls: Int, forwardDetails: [{retries: Int, server: ServerName}], forwards: Int, messages: Int, ok: Bool, presence: [{expiresAt: Int<Ms>, server: ServerName}], sessions: Int} -> a) -> IO {r *activePolls, r *forwards, r *messages, r *presence, r *runtimeMaintenanceSecret, r *sessions, clock} a
2298handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization (yield (respond {
2293handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization (yield (respond {
2299 messages: 0,
2294 messages: 0,
2300 presence: [],
2295 presence: [],
@@ -2319,7 +2314,7 @@ handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization (y
2320 sessions: countWhere ses (\x -> x.pubkey == pubkey),
2315 sessions: countWhere ses (\x -> x.pubkey == pubkey),
2321 forwards: count myFwds,
2316 forwards: count myFwds,
2322− forwardDetails: pluck myFwds (\x -> {server: x.toServer, x.retries, x.createdAt}),
2317+ forwardDetails: pluck myFwds (\x -> {server: x.toServer, x.retries}),
2323 activePolls: countWhere aps (\x -> x.pubkey == pubkey),
2318 activePolls: countWhere aps (\x -> x.pubkey == pubkey),
2324 ok: True {}
2319 ok: True {}
@@ -2341,18 +2336,17 @@ handleTopRecipients = \limit authorization respond -> withMaintAuth authorizatio
2341 let top = takeRelation cap (sortBy (\c -> 0 - c.count) counts)
2336 let top = takeRelation cap (sortBy (\c -> 0 - c.count) counts)
2342 yield (respond {recipients: top, ok: True {}, total: count counts}))
2337 yield (respond {recipients: top, ok: True {}, total: count counts}))
2344−-- GET /stats/forwards: per-destination forward queue summary — count, age of
2339+-- GET /stats/forwards: per-destination forward queue summary — count and
2345−-- oldest pending entry, and total accumulated retries.
2340+-- total accumulated retries.
2346−handleStatsForwards : Text -> ({ok: Bool, servers: [{count: Int<u>, oldestAge: Int<Ms>, server: ServerName, totalRetries: Int}], total: Int} -> a) -> IO {r *forwards, r *runtimeMaintenanceSecret, clock} a
2341+handleStatsForwards : Text -> ({ok: Bool, servers: [{count: Int<u>, server: ServerName, totalRetries: Int}], total: Int} -> a) -> IO {r *forwards, r *runtimeMaintenanceSecret} a
2347handleStatsForwards = \authorization respond -> withMaintAuth authorization (yield (respond {
2342handleStatsForwards = \authorization respond -> withMaintAuth authorization (yield (respond {
2348 servers: [],
2343 servers: [],
2349 ok: False {},
2344 ok: False {},
2350 total: 0
2345 total: 0
2353 fwds <- *forwards
2347 fwds <- *forwards
2354 yield (respond {
2348 yield (respond {
2355− servers: pluck (collectPeers fwds (\f -> f.toServer)) (\s -> (\svFwds -> {server: s, count: count svFwds, oldestAge: t - fold (\acc f -> min acc f.createdAt) t svFwds, totalRetries: fold (\acc f -> acc + f.retries) 0 svFwds}) (filter fwds (\f -> f.toServer == s))),
2349+ servers: pluck (collectPeers fwds (\f -> f.toServer)) (\s -> (\svFwds -> {server: s, count: count svFwds, totalRetries: fold (\acc f -> acc + f.retries) 0 svFwds}) (filter fwds (\f -> f.toServer == s))),
2356 ok: True {},
2350 ok: True {},
2357 total: count fwds
2351 total: count fwds
@@ -2472,7 +2466,7 @@ serve = \req -> case req of
2473-- Endless tick: every `backgroundPruneInterval` run `pruneAndRetry` (skipped
2467-- Endless tick: every `backgroundPruneInterval` run `pruneAndRetry` (skipped
2474-- when drain phase >= 2) and bump `pruneCount` for /stats and /health.
2468-- when drain phase >= 2) and bump `pruneCount` for /stats and /health.
2475−backgroundPrune : IO {r *drainPhase, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *pruneCount, rw *sessions, clock} a
2469+backgroundPrune : IO {r *drainPhase, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, r *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *pruneCount, rw *sessions, clock} a
2476backgroundPrune = do
2470backgroundPrune = do
2477 sleep backgroundPruneInterval
2471 sleep backgroundPruneInterval
2478 dp <- *drainPhase
2472 dp <- *drainPhase