@@ -339,17 +339,13 @@ type Session = {
339339 createdAt: Timestamp
340340}
341341-- Mailbox row: a stored message awaiting recipient pickup or federation forward.
342−-- `senderKey` is "" for messages received via federation (we don't see the
343−-- remote sender's pubkey); non-empty for messages submitted by a local sender.
344−-- That predicate is the local-vs-federation discriminator throughout the code.
345342type Message = {
346343 messageId: MessageIdHex,
347344 toKey: PubkeyHex,
348345 encryptedBlob: BlobHex,
349346 timestamp: Timestamp,
350347 receivedAt: Timestamp,
351− expiresAt: Timestamp,
352− senderKey: Text
348+ expiresAt: Timestamp
353349}
354350-- Last-known location of a pubkey, either local (`server == serverHostN`) or
355351-- learned from a peer via gossip. Multiple entries per pubkey are allowed.
@@ -369,8 +365,7 @@ type ForwardEntry = {
369365 encryptedBlob: BlobHex,
370366 timestamp: Timestamp,
371367 retries: Int,
372− expiresAt: Timestamp,
373− senderKey: Text
368+ expiresAt: Timestamp
374369}
375370-- Replay-protection record: a message id we already delivered locally.
376371type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
@@ -382,22 +377,6 @@ type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
382377-- A live long-poll slot held by a pubkey; used to cap concurrency per key.
383378type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
384379
385−-- Pending notification to a sender that their queued message expired unacked.
386−-- Drained in the sender's next /poll response.
387−type SenderExpiration = {
388− senderKey: Text,
389− messageId: MessageIdHex,
390− toKey: PubkeyHex,
391− expiredAt: Timestamp
392−}
393−-- Pending notification to a sender that their message was delivered to the
394−-- recipient. Drained in the sender's next /poll response.
395−type SenderDelivery = {
396− senderKey: Text,
397− messageId: MessageIdHex,
398− toKey: PubkeyHex,
399− deliveredAt: Timestamp
400−}
401380-- One entry of a /messages/batch request body.
402381type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, expiresInMs: Int}
403382-- Per-entry result returned by /messages/batch.
@@ -430,6 +409,19 @@ migrate *forwards
430409 to {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, retries: Int, expiresAt: Timestamp, senderKey: Text}
431410 using (\old -> {old.messageId, old.toServer, old.createdAt, old.toKey, old.encryptedBlob, old.timestamp, old.retries, old.expiresAt, old.senderKey})
432411
412+-- Drop `senderKey`. Server-side delivery/expiration receipts are gone (clients
413+-- exchange them as encrypted app-level envelopes), so the only thing the field
414+-- still gated -- self-send rejection -- now reads the value from the session.
415+migrate *messages
416+ from {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, receivedAt: Timestamp, expiresAt: Timestamp, senderKey: Text}
417+ to {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, receivedAt: Timestamp, expiresAt: Timestamp}
418+ using (\old -> {old.messageId, old.toKey, old.encryptedBlob, old.timestamp, old.receivedAt, old.expiresAt})
419+
420+migrate *forwards
421+ from {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, retries: Int, expiresAt: Timestamp, senderKey: Text}
422+ to {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, retries: Int, expiresAt: Timestamp}
423+ using (\old -> {old.messageId, old.toServer, old.createdAt, old.toKey, old.encryptedBlob, old.timestamp, old.retries, old.expiresAt})
424+
433425*forwardFailures : [ForwardFailure]
434426*activePolls : [ActivePoll]
435427
@@ -458,9 +450,6 @@ migrate *forwards
458450*drainPhase : Int
459451*drainCompletedAt : Timestamp
460452
461−*senderExpirations : [SenderExpiration]
462−*senderDeliveries : [SenderDelivery]
463−
464453*runtimeMaintenanceSecret : Text
465454
466455*federationPrivateKeyHex : Text
@@ -492,7 +481,7 @@ route AuthApi where
492481 POST /auth/logout headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool} = Logout
493482
494483route ClientApi where
495− POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [DeliveredEntry], droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int, loadPct: Int} = Poll
484+ POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int} = Poll
496485 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, expiresInMs: Int} /messages headers {authorization: Text, clientIp: Text} -> {status: MessageStatus, messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int} = SendMessage
497486 POST {to: PubkeyHex, messageId: MessageIdHex} /messages/status headers {authorization: Text, clientIp: Text} -> {status: Text, authorized: Bool} = MessageStatus
498487 POST {messageIds: [MessageIdHex]} /messages/ack headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, ackedCount: Int, error: Text} = AckMessages
@@ -558,7 +547,7 @@ type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
558547type RecipientCount = {pubkey: PubkeyHex, count: Int}
559548
560549route StatsApi where
561− 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, senderExpirations: Int, senderDeliveries: 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
550+ 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
562551 GET /stats/peers headers {authorization: Text} -> {peers: [PeerStatus], ok: Bool} = StatsPeers
563552 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
564553 POST {limit: Int} /stats/top-recipients headers {authorization: Text} -> {recipients: [RecipientCount], ok: Bool, total: Int} = TopRecipients
@@ -638,7 +627,7 @@ pct : Int -> Int -> Int
638627pct = \cur maxN -> if maxN > 0 then cur * 100 / maxN else 0
639628
640629-- Common ack/delete tail: pull matching messages, drop them and their forwards.
641−removeForKey : (forall r. {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, expiresAt: Timestamp, senderKey: Text | r} -> Bool) -> IO {rw *forwards, rw *messages} [Message]
630+removeForKey : (forall r. {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, expiresAt: Timestamp | r} -> Bool) -> IO {rw *forwards, rw *messages} [Message]
642631removeForKey = \pred -> do
643632 msgs <- *messages
644633 fwds <- *forwards
@@ -651,7 +640,7 @@ capPerKey : [a] -> (a -> Bool) -> Int -> (a -> b) -> [a]
651640capPerKey = \items pred maxN getKey -> union (removeWhere items pred) (capList (filter items pred) (maxN - 1) getKey)
652641
653642-- Shared empty/error shape for /poll. Only a few fields ever vary.
654−emptyPollResp : Timestamp -> Bool -> Bool -> Bool -> Int -> Int -> Bool -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [DeliveredEntry], droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int, loadPct: Int}
643+emptyPollResp : Timestamp -> Bool -> Bool -> Bool -> Int -> Int -> Bool -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int}
655644emptyPollResp = \cursor authorized rateLimited draining gen retryAfterMs cursorWasReset -> {
656645 events: [],
657646 cursor,
@@ -664,10 +653,6 @@ emptyPollResp = \cursor authorized rateLimited draining gen retryAfterMs cursorW
664653 cursorReset: cursorWasReset,
665654 hasMore: False {},
666655 cursorStale: False {},
667− messageExpirations: [],
668− messageDeliveries: [],
669− droppedDeliveryReceipts: 0,
670− droppedExpirationNotifs: 0,
671656 loadPct: 0
672657}
673658
@@ -1107,7 +1092,7 @@ mergePresence = \existing incoming -> fold (\acc e -> (\same -> if any acc (\x -
11071092-- One sweep over every state ref: drops expired rows in a single atomic block.
11081093-- Returns the locally-expired presence rows whose pubkey has no other live
11091094-- local session, so the caller can broadcast offline gossip for them.
1110−pruneExpired : Int<Ms> -> IO {rw *activePolls, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, rw *senderDeliveries, rw *senderExpirations, rw *sessions} [{expiresAt: Timestamp, pubkey: PubkeyHex, server: AnyServerName, timestamp: Timestamp}]
1095+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: AnyServerName, timestamp: Timestamp}]
11111096pruneExpired = \t -> atomic do
11121097 ch <- *challenges
11131098 *challenges = filter ch (\x -> x.expiresAt > t)
@@ -1123,10 +1108,6 @@ pruneExpired = \t -> atomic do
11231108 *forwardFailures = filter ff (\f -> t - f.failedAt < 86400000)
11241109 aps <- *activePolls
11251110 *activePolls = filter aps (\p -> t - p.startedAt < pollTimeoutMs * 3)
1126− se <- *senderExpirations
1127− *senderExpirations = filter se (\x -> t - x.expiredAt < unackedMessageMaxAge)
1128− sd <- *senderDeliveries
1129− *senderDeliveries = filter sd (\x -> t - x.deliveredAt < unackedMessageMaxAge)
11301111 -- presence: remember locally-expired keys for offline gossip
11311112 pres <- *presence
11321113 *presence = filter pres (\p -> p.expiresAt > t)
@@ -1152,21 +1133,6 @@ mkGossip = \etype pk t ttl -> {
11521133 seenBy: [serverHostN]
11531134}
11541135
1155−-- Record sender-side delivery rows for a batch of just-acked locally-originated
1156−-- messages. Federation-origin messages have `senderKey == ""` and are skipped;
1157−-- delivery confirmations for federated sends now travel as application-level
1158−-- receipts via the normal message flow, not server-to-server.
1159−recordSenderDeliveries : [{messageId: MessageIdHex, senderKey: Text, toKey: PubkeyHex | a}] -> Int<Ms> -> IO {rw *senderDeliveries} {}
1160−recordSenderDeliveries = \toAck t -> do
1161− let myDel = mapWhere toAck (\m -> m.senderKey != "") (\m -> {
1162− m.senderKey,
1163− m.messageId,
1164− m.toKey,
1165− deliveredAt: t
1166− })
1167− sds <- *senderDeliveries
1168− *senderDeliveries = capList (union sds myDel) 10000 (\d -> d.deliveredAt)
1169−
11701136-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.
11711137-- Records or clears federation-failure counters per call.
11721138sendBatchedGossip : [Text] -> [GossipEvent] -> IO {} {}
@@ -1246,8 +1212,7 @@ retryPendingForwards = do
12461212 x.encryptedBlob,
12471213 x.timestamp,
12481214 retries: x.retries + 1,
1249− x.expiresAt,
1250− x.senderKey
1215+ x.expiresAt
12511216 }
12521217 else x))))
12531218
@@ -1265,8 +1230,7 @@ forwardQueuedToServer = \pubkey server t -> atomic (do
12651230 m.encryptedBlob,
12661231 m.timestamp,
12671232 retries: 0,
1268− m.expiresAt,
1269− m.senderKey
1233+ m.expiresAt
12701234 })
12711235 *forwards = union fwds newFwds)
12721236
@@ -1467,7 +1431,7 @@ handleLogout = \authorization clientIp respond -> do
14671431-- POST /poll: top-level dispatch. Validates drain/global gates, detects a
14681432-- generation mismatch (server restart) to force a cursor reset, then hands
14691433-- off to `pollOnce` for the actual long-poll body.
1470−handlePoll : Int -> Int -> Text -> Text -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, draining: Bool, droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, messageDeliveries: [DeliveredEntry], messageExpirations: [{expiredAt: Timestamp, messageId: MessageIdHex, toKey: PubkeyHex}], rateLimited: Bool, retryAfterMs: Int} -> Response) -> IO {rw *activePolls, r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *pollWakeup, rw *senderDeliveries, rw *senderExpirations, r *serverGeneration, r *sessions, clock} Response
1434+handlePoll : Int -> Int -> Text -> Text -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, draining: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int} -> Response) -> IO {rw *activePolls, r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *pollWakeup, r *serverGeneration, r *sessions, clock} Response
14711435handlePoll = \cursor generation authorization clientIp respond -> do
14721436 t <- now
14731437 serverGen <- *serverGeneration
@@ -1483,13 +1447,12 @@ handlePoll = \cursor generation authorization clientIp respond -> do
14831447
14841448-- Body of a single poll. Treats the cursor advance as an implicit ack
14851449-- (`ackPolledMessages`), acquires a per-pubkey poll slot, runs `doPoll` to
1486−-- collect events (or block until wakeup), then assembles the response with
1487−-- piggybacked expiration/delivery metadata.
1488−pollOnce : Text -> Int -> Bool -> Text -> Int -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, draining: Bool, droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, messageDeliveries: [DeliveredEntry], messageExpirations: [{expiredAt: Timestamp, messageId: MessageIdHex, toKey: PubkeyHex}], rateLimited: Bool, retryAfterMs: Int} -> Response) -> Int<Ms> -> IO {rw *activePolls, r *pollWakeup, rw *senderDeliveries, rw *senderExpirations, r *sessions, clock} Response
1450+-- collect events (or block until wakeup), then assembles the response.
1451+pollOnce : Text -> Int -> Bool -> Text -> Int -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, draining: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int} -> Response) -> Int<Ms> -> IO {rw *activePolls, r *pollWakeup, r *sessions, clock} Response
14891452pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t -> do
14901453 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
14911454 let clampedCursor = clamp 0 t cursor
1492− when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor t)
1455+ when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)
14931456 startWakeup <- *pollWakeup
14941457 yield (Just {value: {pubkey, clampedCursor, startWakeup}}))
14951458 case setup of
@@ -1507,15 +1470,6 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
15071470 else do
15081471 pollResult <- doPoll s t
15091472 atomic (do aps <- *activePolls; *activePolls = filter aps (\p -> p.pubkey != s.pubkey))
1510− extras <- atomic (do
1511− let mine = \x -> x.senderKey == s.pubkey
1512− seItems <- *senderExpirations
1513− *senderExpirations = removeWhere seItems mine
1514− let myExp = mapWhere seItems mine (\x -> {x.messageId, x.toKey, x.expiredAt})
1515− sdItems <- *senderDeliveries
1516− *senderDeliveries = removeWhere sdItems mine
1517− let myDel = mapWhere sdItems mine (\x -> {x.messageId, x.toKey, x.deliveredAt})
1518− yield {expirations: myExp, deliveries: myDel})
15191473 yield (respond {
15201474 pollResult.events,
15211475 pollResult.cursor,
@@ -1528,10 +1482,6 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
15281482 cursorReset: cursorWasReset,
15291483 pollResult.hasMore,
15301484 cursorStale: False {},
1531− messageExpirations: extras.expirations,
1532− messageDeliveries: extras.deliveries,
1533− droppedDeliveryReceipts: 0,
1534− droppedExpirationNotifs: 0,
15351485 loadPct: 0
15361486 })
15371487
@@ -1556,10 +1506,9 @@ collectPollEvents : [Message] -> PubkeyHex -> Timestamp -> Timestamp -> {events:
15561506collectPollEvents = \messages pubkey cursor t -> (\myMsgs -> (\limited -> {events: pluck limited (\m -> {id: m.messageId, payload: {m.toKey, m.encryptedBlob, m.timestamp, m.expiresAt}}), cursor: fold (\acc m -> max acc m.receivedAt) cursor limited, hasMore: count myMsgs > count limited}) (takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs))) (filter messages (\m -> m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor))
15571507
15581508-- Treat the supplied cursor as an ack: drop all messages with
1559−-- receivedAt <= cursor, plus their pending forwards, and record sender-side
1560−-- deliveries for any locally-originated messages via `recordSenderDeliveries`.
1561−ackPolledMessages : PubkeyHex -> Int -> Int<Ms> -> IO {rw *forwards, rw *messages, rw *senderDeliveries} {}
1562−ackPolledMessages = \pubkey cursor t -> do
1509+-- receivedAt <= cursor, plus their pending forwards.
1510+ackPolledMessages : PubkeyHex -> Int -> IO {rw *forwards, rw *messages} {}
1511+ackPolledMessages = \pubkey cursor -> do
15631512 msgs <- *messages
15641513 let matches = \m -> m.toKey == pubkey && m.receivedAt <= cursor
15651514 let toAck = filter msgs matches
@@ -1567,7 +1516,6 @@ ackPolledMessages = \pubkey cursor t -> do
15671516 *messages = removeWhere msgs matches
15681517 fwds <- *forwards
15691518 *forwards = filter fwds (\f -> not (f.toKey == pubkey && elem mids f.messageId))
1570− recordSenderDeliveries toAck t
15711519
15721520-- ============================================
15731521-- Handler: send
@@ -1653,8 +1601,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs t -
16531601 encryptedBlob,
16541602 timestamp: clampedTs,
16551603 receivedAt: seqTs,
1656− expiresAt,
1657− senderKey
1604+ expiresAt
16581605 }
16591606 ]
16601607 *pollWakeup = seqTs
@@ -1671,8 +1618,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs t -
16711618 encryptedBlob,
16721619 timestamp: clampedTs,
16731620 retries: 0,
1674− expiresAt,
1675− senderKey
1621+ expiresAt
16761622 })
16771623 *forwards = union fwds newFwds
16781624 yield (count newFwds > 0)
@@ -1791,8 +1737,7 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
17911737-- Mailbox mutation by the recipient (ack, list, delete).
17921738
17931739-- POST /messages/ack: explicit ack by message id (alternative to the implicit
1794−-- cursor-based ack in /poll). Drops mailbox+forward rows and records
1795−-- sender-side delivery for any locally-originated messages.
1740+-- cursor-based ack in /poll). Drops mailbox+forward rows.
17961741handleAckMessages : [MessageIdHex] -> Text -> Text -> ({ackedCount: Int, authorized: Bool, error: Text, ok: Bool} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
17971742handleAckMessages = \messageIds authorization clientIp respond -> do
17981743 t <- now
@@ -1805,7 +1750,6 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
18051750 } (\pubkey -> do
18061751 ackedCount <- atomic (do
18071752 toAck <- removeForKey (\x -> x.toKey == pubkey && elem (dedupMessageIds messageIds) x.messageId)
1808− recordSenderDeliveries toAck t
18091753 yield (count toAck))
18101754 yield (respond {ok: True {}, authorized: True {}, ackedCount, error: ""}))
18111755
@@ -1976,8 +1920,7 @@ storeFederated = \toKey encryptedBlob timestamp messageId expiresAt checkRecvRat
19761920 encryptedBlob,
19771921 timestamp: clampTimestamp timestamp t,
19781922 receivedAt: seqTs,
1979− expiresAt: cappedExpiry,
1980− senderKey: ""
1923+ expiresAt: cappedExpiry
19811924 }
19821925 ]
19831926 *pollWakeup = seqTs
@@ -2065,7 +2008,7 @@ handleShutdown = \authorization respond -> handleDrain 2 authorization (\r -> re
20652008-- POST /maintenance/prune: trigger an out-of-band `pruneAndRetry` tick.
20662009-- Useful when the operator wants to force expiry/forward retries without
20672010-- waiting for the next `backgroundPrune` interval.
2068−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 *senderDeliveries, rw *senderExpirations, rw *sessions, clock} a
2011+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
20692012handlePrune = \authorization respond -> do
20702013 t <- now
20712014 authorized <- verifyMaintenanceAuth authorization
@@ -2347,7 +2290,7 @@ handleGetConfig = \authorization respond -> do
23472290-- GET /stats: capacity gauges + minimal federation/timing snapshot. Most of
23482291-- the historical metric fields are zero-filled (see `zeroStats`) — only the
23492292-- counts that come straight from the state refs are populated.
2350−handleStats : Text -> ({capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, senderDeliveries: Int, senderExpirations: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text} -> a) -> IO {r *activePolls, r *blockedPeers, r *challenges, r *deliveredIds, r *drainPhase, r *forwards, r *heartbeatCount, r *lastHeartbeatAt, r *lastPrunedAt, r *messages, r *presence, r *pruneCount, r *runtimeMaintenanceSecret, r *sessions, r *startedAt, clock} a
2293+handleStats : Text -> ({capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text} -> a) -> IO {r *activePolls, r *blockedPeers, r *challenges, r *deliveredIds, r *drainPhase, r *forwards, r *heartbeatCount, r *lastHeartbeatAt, r *lastPrunedAt, r *messages, r *presence, r *pruneCount, r *runtimeMaintenanceSecret, r *sessions, r *startedAt, clock} a
23512294handleStats = \authorization respond -> withMaintAuth authorization (yield (respond zeroStats)) (do
23522295 t <- now
23532296 sa <- *startedAt
@@ -2427,8 +2370,6 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
24272370 verifyIp: 0,
24282371 uniqueKeyIp: 0,
24292372 pendingAckCleanups: 0,
2430− senderExpirations: 0,
2431− senderDeliveries: 0,
24322373 rejectedGlobal: 0,
24332374 rejectedIp: 0,
24342375 rejectedAuth: 0,
@@ -2464,7 +2405,7 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
24642405-- All-zero /stats payload returned to unauthorized callers. The shape is kept
24652406-- in lockstep with the response in `handleStats` so older operator tooling
24662407-- never sees a missing field.
2467−zeroStats : {capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, senderDeliveries: Int, senderExpirations: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text}
2408+zeroStats : {capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text}
24682409zeroStats = {
24692410 ok: False {},
24702411 version: serverBuildVersion,
@@ -2524,8 +2465,6 @@ zeroStats = {
25242465 verifyIp: 0,
25252466 uniqueKeyIp: 0,
25262467 pendingAckCleanups: 0,
2527− senderExpirations: 0,
2528− senderDeliveries: 0,
25292468 rejectedGlobal: 0,
25302469 rejectedIp: 0,
25312470 rejectedAuth: 0,
@@ -2766,7 +2705,7 @@ serve = \req -> case req of
27662705
27672706-- Endless tick: every `backgroundPruneInterval` run `pruneAndRetry` (skipped
27682707-- when drain phase >= 2) and bump `pruneCount` for /stats and /health.
2769−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 *senderDeliveries, rw *senderExpirations, rw *sessions, clock} a
2708+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
27702709backgroundPrune = do
27712710 sleep backgroundPruneInterval
27722711 dp <- *drainPhase