@@ -63,6 +63,8 @@ pollTimeoutMs : Int<Ms>
6363pollTimeoutMs = 25000 -- long-poll heartbeat interval
6464presenceRefreshInterval : Int<Ms>
6565presenceRefreshInterval = 300000 -- 5 minutes
66+presenceRefreshThreshold : Int<Ms>
67+presenceRefreshThreshold = presenceRefreshInterval * 2 -- refresh entries expiring within this window
6668federationEnabled = federationSecret != ""
6769federationAuth = "Bearer " ++ federationSecret
6870federationUrl = \server -> federationScheme ++ server
@@ -129,8 +131,7 @@ type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp
129131type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}
130132type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
131133data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
132−data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName], forwardServers: [ServerName]} | SendDuplicate {}
133−data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp, pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
134+data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName]} | SendDuplicate {}
134135data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
135136
136137-- ============================================
@@ -849,7 +850,7 @@ pruneAndRetry = \t -> do
849850 lastRefresh <- *lastPresenceRefresh
850851 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval
851852 presence <- *presence
852− let localActive = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p) else []
853+ let localActive = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []
853854 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []
854855 when shouldRefresh (set *lastPresenceRefresh = t)
855856 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks, localActive: localActive, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
@@ -910,30 +911,35 @@ readMessageSignal = \pubkey ->
910911 else if b == 14 then do s <- *messageSignal14; yield s
911912 else do s <- *messageSignal15; yield s
912913
913−pollEventsAndDelete = \pubkey clampedCursor t -> do
914+ackPolledMessages = \pubkey cursor t -> do
914915 messages <- *messages
915− let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages messages pubkey clampedCursor t))
916− if count limitedMsgs > 0
916+ let toAck = do m <- messages; where m.toKey == pubkey && m.receivedAt <= cursor; yield m
917+ if count toAck > 0
917918 then do
918− set *messages = diff messages limitedMsgs
919− let polledMsgIds = do m <- limitedMsgs; yield m.messageId
919+ set *messages = diff messages toAck
920+ let ackedMsgIds = do m <- toAck; yield m.messageId
920921 forwards <- *forwards
921− set *forwards = do f <- forwards; where none polledMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
922− let ackedFederated = federatedAckEntries limitedMsgs t
922+ set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
923+ let ackedFederated = federatedAckEntries toAck t
923924 ackedIds <- *ackedIds
924925 let combinedAckedIds = union ackedIds ackedFederated
925− let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
926− if excessCount <= 0 then [] else takeRelation excessCount (sortBy (\a -> a.ackedAt) combinedAckedIds)
926+ let excessCount = count combinedAckedIds - maxAckedIds
927+ let evictedEntries = if excessCount <= 0 then [] else takeRelation excessCount (sortBy (\a -> a.ackedAt) combinedAckedIds)
927928 set *ackedIds = diff combinedAckedIds evictedEntries
928− let evts = do
929− m <- limitedMsgs
930− yield {id: m.messageId, eventType: "message",
931− payload: {toKey: m.toKey,
932− encryptedBlob: m.encryptedBlob,
933− timestamp: m.timestamp}}
934− let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
935− yield (PollOk {events: evts, cursor: lastReceivedAt, pendingAcks: ackedFederated, evictedEntries: evictedEntries})
936− else yield (PollOk {events: [], cursor: clampedCursor, pendingAcks: [], evictedEntries: []})
929+ yield {pendingAcks: ackedFederated, evictedEntries: evictedEntries}
930+ else yield {pendingAcks: [], evictedEntries: []}
931+
932+pollNewEvents = \pubkey clampedCursor t -> do
933+ messages <- *messages
934+ let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages messages pubkey clampedCursor t))
935+ let evts = do
936+ m <- limitedMsgs
937+ yield {id: m.messageId, eventType: "message",
938+ payload: {toKey: m.toKey,
939+ encryptedBlob: m.encryptedBlob,
940+ timestamp: m.timestamp}}
941+ let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
942+ yield {events: evts, cursor: lastReceivedAt}
937943
938944pollOnce = \token cursor -> do
939945 t <- now
@@ -944,17 +950,22 @@ pollOnce = \token cursor -> do
944950 Authorized {pubkey} -> do
945951 seqCounter <- *seqCounter
946952 let clampedCursor = clamp 0 (max seqCounter t) cursor
953+ ackResult <- if clampedCursor > 0
954+ then ackPolledMessages pubkey clampedCursor t
955+ else yield {pendingAcks: [], evictedEntries: []}
947956 startWakeup <- *pollWakeup
948957 startSignal <- readMessageSignal pubkey
949− yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, startSignal: startSignal}})
958+ yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, startSignal: startSignal, pendingAcks: ackResult.pendingAcks, evictedEntries: ackResult.evictedEntries}})
950959 case setup of
951− Nothing {} -> yield (PollUnauthorized {})
960+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
952961 Just {value: s} -> do
953962 -- Phase 1: check for existing messages (no blocking)
954963 initialMessages <- *messages
955964 let initial = pollableMessages initialMessages s.pubkey s.clampedCursor t
956965 if count initial > 0
957− then atomic do pollEventsAndDelete s.pubkey s.clampedCursor t
966+ then do
967+ result <- atomic do pollNewEvents s.pubkey s.clampedCursor t
968+ yield {authorized: True {}, events: result.events, cursor: result.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
958969 else do
959970 -- Phase 2: wait for signal or heartbeat (no *messages dependency)
960971 atomic do
@@ -963,9 +974,16 @@ pollOnce = \token cursor -> do
963974 if wakeup > s.startWakeup || signal > s.startSignal
964975 then yield {}
965976 else retry
966− -- Phase 3: read messages after waking, delete atomically
977+ -- Phase 3: re-authenticate and read messages after waking
967978 t2 <- now
968− atomic do pollEventsAndDelete s.pubkey s.clampedCursor t2
979+ result <- atomic do
980+ sessions2 <- *sessions
981+ case authenticateSession sessions2 token t2 of
982+ Unauthorized {} -> yield {authorized: False {}, events: [], cursor: s.clampedCursor}
983+ Authorized {} -> do
984+ r <- pollNewEvents s.pubkey s.clampedCursor t2
985+ yield {authorized: True {}, events: r.events, cursor: r.cursor}
986+ yield {authorized: result.authorized, events: result.events, cursor: result.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
969987
970988-- ============================================
971989-- Derived Relations
@@ -1067,22 +1085,25 @@ serve = \req -> case req of
10671085 t <- now
10681086 result <- atomic do
10691087 pr <- maybePrune t
1070− rateTs <- *challengeRateTs
1071− rateCount <- *challengeRateCount
1072− let newWindow = t - rateTs >= challengeRateWindowMs
1073− let currentCount = if newWindow then 0 else rateCount
1074− if currentCount >= challengeRateLimit
1075− then yield {pr: pr, accepted: False {}, rateLimited: True {}}
1076− else do
1077− set *challengeRateTs = if newWindow then t else rateTs
1078− set *challengeRateCount = currentCount + 1
1079− challenges <- *challenges
1080− let perKeyCapped = capPerKey challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t
1081− if count perKeyCapped >= maxChallenges
1082− then yield {pr: pr, accepted: False {}, rateLimited: False {}}
1083− else do
1084− set *challenges = union perKeyCapped [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
1085− yield {pr: pr, accepted: True {}, rateLimited: False {}}
1088+ challenges <- *challenges
1089+ let active = do c <- challenges; where c.expiresAt > t; yield c
1090+ let activeForKey = count (do c <- active; where c.pubkey == pubkey; yield c)
1091+ if activeForKey >= maxChallengesPerKey
1092+ then yield {pr: pr, accepted: False {}, rateLimited: False {}}
1093+ else if count active >= maxChallenges
1094+ then yield {pr: pr, accepted: False {}, rateLimited: False {}}
1095+ else do
1096+ rateTs <- *challengeRateTs
1097+ rateCount <- *challengeRateCount
1098+ let newWindow = t - rateTs >= challengeRateWindowMs
1099+ let currentCount = if newWindow then 0 else rateCount
1100+ if currentCount >= challengeRateLimit
1101+ then yield {pr: pr, accepted: False {}, rateLimited: True {}}
1102+ else do
1103+ set *challengeRateTs = if newWindow then t else rateTs
1104+ set *challengeRateCount = currentCount + 1
1105+ set *challenges = union active [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
1106+ yield {pr: pr, accepted: True {}, rateLimited: False {}}
10861107 when result.rateLimited
10871108 (println ("challenge: rate limited (" ++ show challengeRateLimit ++ "/s), rejecting for " ++ pubkey))
10881109 when (not result.accepted && not result.rateLimited)
@@ -1130,19 +1151,14 @@ serve = \req -> case req of
11301151
11311152 Poll {cursor, authorization, respond} -> do
11321153 result <- case extractToken authorization of
1133− Nothing {} -> yield (PollUnauthorized {})
1154+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
11341155 Just {value: token} -> pollOnce token cursor
11351156 tAfterPoll <- now
11361157 maybePruneAndBroadcast tAfterPoll
1137− case result of
1138− PollOk {pendingAcks, evictedEntries} ->
1139− fork (when federationEnabled (do
1140− sendAcksAndClean pendingAcks
1141− retryEvictedAcks evictedEntries pendingAcks))
1142− PollUnauthorized {} -> yield {}
1143− yield (respond (case result of
1144− PollUnauthorized {} -> {events: [], cursor: 0, authorized: False {}}
1145− PollOk {events: evts, cursor: c} -> {events: evts, cursor: c, authorized: True {}}))
1158+ fork (when federationEnabled (do
1159+ sendAcksAndClean result.pendingAcks
1160+ retryEvictedAcks result.evictedEntries result.pendingAcks))
1161+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
11461162
11471163 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, respond} -> do
11481164 t <- now
@@ -1153,61 +1169,57 @@ serve = \req -> case req of
11531169 withSessionAuth authorization t
11541170 (yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}})
11551171 (\senderKey -> do
1156− sendRates <- *sendRates
1157− let senderEntry = headOf (do r <- sendRates; where r.pubkey == senderKey; yield r)
1158− let isNewWindow = case senderEntry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= sendRateWindowMs
1159− let rateCount = case senderEntry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count
1160− if rateCount >= maxSendRatePerKey
1161− then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
1172+ messages <- *messages
1173+ ackedIds <- *ackedIds
1174+ if messageId != "" && (hasMessage messageId to messages || hasMessage messageId to ackedIds)
1175+ then yield {pr: pr, send: SendDuplicate {}}
11621176 else do
1163− let newRates = if isNewWindow
1164− then union (do r <- sendRates; where r.pubkey != senderKey; yield r) [{pubkey: senderKey, windowStart: t, count: 1}]
1165− else do r <- sendRates; yield (if r.pubkey == senderKey then {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1} else r)
1166− set *sendRates = if count newRates > maxSendRateEntries then (do r <- newRates; where t - r.windowStart < sendRateWindowMs; yield r) else newRates
1167− messages <- *messages
1168− ackedIds <- *ackedIds
1169− if messageId != "" && (hasMessage messageId to messages || hasMessage messageId to ackedIds)
1170− then yield {pr: pr, send: SendDuplicate {}}
1171− else if countMessagesFor messages to t >= maxMessagesPerRecipient
1172− then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
1173− else do
1174− presence <- *presence
1175− seqCounter <- *seqCounter
1176− let seqTs = nextSeqTs t seqCounter
1177− set *seqCounter = seqTs
1178− set *messages = union messages
1179− [{messageId: effectiveMsgId, toKey: to,
1180− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
1181− signalMessageForKey to seqTs
1182− let toPresence = presenceForKey presence to t
1183− let localOnline = hasLocalPresence toPresence
1184− let remoteServers = collectPeerServers toPresence (\p -> p.server)
1185− forwards <- *forwards
1186− let forwardServers = takeRelation (maxForwardsEntries - count forwards) remoteServers
1187− let shouldForward = federationEnabled && count forwardServers > 0
1188− let delivery = if localOnline then DsDelivered {}
1189− else if shouldForward then DsFederated {}
1190− else DsQueued {}
1191− let updatedForwards = if shouldForward
1192− then union forwards (do s <- forwardServers; yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs})
1193− else forwards
1194− set *forwards = updatedForwards
1195− yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers}})
1177+ sendRates <- *sendRates
1178+ let senderEntry = headOf (do r <- sendRates; where r.pubkey == senderKey; yield r)
1179+ let isNewWindow = case senderEntry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= sendRateWindowMs
1180+ let rateCount = case senderEntry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count
1181+ if rateCount >= maxSendRatePerKey
1182+ then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
1183+ else do
1184+ let newRates = if isNewWindow
1185+ then union (do r <- sendRates; where r.pubkey != senderKey; yield r) [{pubkey: senderKey, windowStart: t, count: 1}]
1186+ else do r <- sendRates; yield (if r.pubkey == senderKey then {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1} else r)
1187+ set *sendRates = if count newRates > maxSendRateEntries then (do r <- newRates; where t - r.windowStart < sendRateWindowMs; yield r) else newRates
1188+ if countMessagesFor messages to t >= maxMessagesPerRecipient
1189+ then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
1190+ else do
1191+ presence <- *presence
1192+ seqCounter <- *seqCounter
1193+ let seqTs = nextSeqTs t seqCounter
1194+ set *seqCounter = seqTs
1195+ set *messages = union messages
1196+ [{messageId: effectiveMsgId, toKey: to,
1197+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
1198+ signalMessageForKey to seqTs
1199+ let toPresence = presenceForKey presence to t
1200+ let localOnline = hasLocalPresence toPresence
1201+ let remoteServers = collectPeerServers toPresence (\p -> p.server)
1202+ let delivery = if localOnline then DsDelivered {}
1203+ else if federationEnabled && count remoteServers > 0 then DsFederated {}
1204+ else DsQueued {}
1205+ yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers}})
11961206 case result.send of
1197− SendDone {delivery, remoteServers, forwardServers} ->
1207+ SendDone {delivery, remoteServers} ->
11981208 fork (do
1199− when (count remoteServers > 0 && federationEnabled) (do
1209+ when (count remoteServers > 0 && federationEnabled) (
12001210 sendToEach remoteServers (\server -> do
12011211 fwdResult <- fetch (federationUrl server) (ForwardMessage {
12021212 toKey: to, encryptedBlob: encryptedBlob,
12031213 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
12041214 authorization: federationAuth})
12051215 case fwdResult of
1206− Err _ -> println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)
1207− Ok _ -> when (any forwardServers (\s -> s == server))
1208− (removeForward effectiveMsgId server))
1209− when (count forwardServers < count remoteServers)
1210− (println ("sendMessage: " ++ show (count remoteServers - count forwardServers) ++ " remote server(s) not tracked for " ++ effectiveMsgId ++ " (forwards at capacity " ++ show maxForwardsEntries ++ ")")))
1216+ Err _ -> do
1217+ println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)
1218+ atomic do
1219+ forwards <- *forwards
1220+ when (count forwards < maxForwardsEntries)
1221+ (set *forwards = union forwards [{messageId: effectiveMsgId, toServer: server, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}])
1222+ Ok _ -> yield {}))
12111223 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
12121224 _ -> forkBroadcastOffline t result.pr
12131225 yield (case result.send of