@@ -303,7 +303,6 @@ type PresenceEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Timestam
303303-- Identity is derived from `encryptedBlob` (see `Message`).
304304type ForwardEntry = {
305305 toServer: ServerName,
306− createdAt: Timestamp,
307306 toKey: PubkeyHex,
308307 encryptedBlob: BlobHex,
309308 retries: Int
@@ -458,9 +457,9 @@ type RecipientCount = {pubkey: PubkeyHex, count: Int}
458457route StatsApi where
459458 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
460459 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
462461 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
464463 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
465464
466465route MetricsApi where
@@ -972,7 +971,7 @@ knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.serve
972971-- One sweep over every state ref: drops expired rows in a single atomic block.
973972-- Returns the locally-expired presence rows whose pubkey has no other live
974973-- 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}]
976975pruneExpired = \t -> atomic do
977976 ch <- *challenges
978977 *challenges = filter ch (\x -> x.expiresAt > t)
@@ -980,8 +979,6 @@ pruneExpired = \t -> atomic do
980979 *sessions = filter ses (\x -> x.expiresAt > t)
981980 msgs <- *messages
982981 *messages = filter msgs (\m -> t - m.receivedAt < messageMaxAge)
983− fwds <- *forwards
984− *forwards = filter fwds (\f -> t - f.createdAt < messageMaxAge)
985982 dids <- *deliveredIds
986983 *deliveredIds = filter dids (\d -> t - d.deliveredAt < messageMaxAge)
987984 ff <- *forwardFailures
@@ -1074,18 +1071,17 @@ retryPendingForwards = do
10741071 atomic (do
10751072 cur <- *forwards
10761073 *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}
10781075 else x))))
10791076
10801077-- Forward queued messages to a server when a remote key comes online.
1081−forwardQueuedToServer : PubkeyHex -> ServerName -> Int<Ms> -> IO {rw *forwards, r *messages} {}
1082−forwardQueuedToServer = \pubkey server t -> atomic (do
1078+forwardQueuedToServer : PubkeyHex -> ServerName -> IO {rw *forwards, r *messages} {}
1079+forwardQueuedToServer = \pubkey server -> atomic (do
10831080 msgs <- *messages
10841081 fwds <- *forwards
10851082 let candidates = filter msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.encryptedBlob == m.encryptedBlob && f.toServer == server))
10861083 let newFwds = pluck (takeRelation maxForwardsPerServer candidates) (\m -> {
10871084 toServer: server,
1088− createdAt: t,
10891085 m.toKey,
10901086 m.encryptedBlob,
10911087 retries: 0
@@ -1441,7 +1437,6 @@ sendMessageCore = \senderKey to encryptedBlob t -> do
14411437 fwds <- *forwards
14421438 let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s)) (\s -> {
14431439 toServer: s,
1444− createdAt: t,
14451440 toKey: to,
14461441 encryptedBlob,
14471442 retries: 0
@@ -1659,7 +1654,7 @@ handleRecvGossip = \events fromServer requestTimestamp authorization respond ->
16591654 let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys
16601655 *presence = union kept added
16611656 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))
16631658 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
16641659
16651660-- 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
17801775-- POST /maintenance/prune: trigger an out-of-band `pruneAndRetry` tick.
17811776-- Useful when the operator wants to force expiry/forward retries without
17821777-- 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
17841779handlePrune = \authorization respond -> do
17851780 t <- now
17861781 authorized <- verifyMaintenanceAuth authorization
@@ -2294,7 +2289,7 @@ handleStatsPeers = \authorization respond -> withMaintAuth authorization (yield
22942289
22952290-- GET /stats/key/{pubkey}: per-key footprint — message count, presence rows,
22962291-- 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
22982293handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization (yield (respond {
22992294 messages: 0,
23002295 presence: [],
@@ -2319,7 +2314,7 @@ handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization (y
23192314 }),
23202315 sessions: countWhere ses (\x -> x.pubkey == pubkey),
23212316 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}),
23232318 activePolls: countWhere aps (\x -> x.pubkey == pubkey),
23242319 ok: True {}
23252320 }))
@@ -2341,18 +2336,17 @@ handleTopRecipients = \limit authorization respond -> withMaintAuth authorizatio
23412336 let top = takeRelation cap (sortBy (\c -> 0 - c.count) counts)
23422337 yield (respond {recipients: top, ok: True {}, total: count counts}))
23432338
2344−-- GET /stats/forwards: per-destination forward queue summary — count, age of
2345−-- oldest pending entry, and 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
2339+-- GET /stats/forwards: per-destination forward queue summary — count and
2340+-- total accumulated retries.
2341+handleStatsForwards : Text -> ({ok: Bool, servers: [{count: Int<u>, server: ServerName, totalRetries: Int}], total: Int} -> a) -> IO {r *forwards, r *runtimeMaintenanceSecret} a
23472342handleStatsForwards = \authorization respond -> withMaintAuth authorization (yield (respond {
23482343 servers: [],
23492344 ok: False {},
23502345 total: 0
23512346})) (do
2352− t <- now
23532347 fwds <- *forwards
23542348 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))),
23562350 ok: True {},
23572351 total: count fwds
23582352 }))
@@ -2472,7 +2466,7 @@ serve = \req -> case req of
24722466
24732467-- Endless tick: every `backgroundPruneInterval` run `pruneAndRetry` (skipped
24742468-- 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
24762470backgroundPrune = do
24772471 sleep backgroundPruneInterval
24782472 dp <- *drainPhase