@@ -227,8 +227,6 @@ type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}
227227
228228-- Strictly-monotonic sequence used as `Message.receivedAt`; guarantees poll-cursor ordering.
229229*seqCounter : Timestamp
230−-- Bumped by send/heartbeat to wake all blocked /poll holders.
231−*pollWakeup : Timestamp
232230
233231-- ============================================
234232-- Routes
@@ -413,16 +411,15 @@ collectPeers : [a] -> (a -> Text) -> [Text]
413411collectPeers = \items getServer -> filter (\s -> not isLocalServer s) (map getServer items)
414412
415413-- Append one ciphertext to the local mailbox under a fresh monotonic
416−-- sequence and bump pollWakeup so blocked /poll holders unblock. Caller
417−-- wraps in `atomic`.
418−appendMessage : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWakeup, rw *seqCounter} {}
414+-- sequence. Row-level STM retry on `*messages` wakes any parked /poll
415+-- transaction whose filter matches the new row. Caller wraps in `atomic`.
416+appendMessage : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *seqCounter} {}
419417appendMessage = \toKey encryptedBlob t -> do
420418 msgs <- *messages
421419 seq <- *seqCounter
422420 let seqTs = max t (seq + 1)
423421 *seqCounter = seqTs
424422 *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs}]
425− *pollWakeup = seqTs
426423
427424-- True when `f` is the queued forward of `blob` to `server`. Used to dedupe
428425-- per-(blob, server) entries on the forwards outbox.
@@ -600,14 +597,15 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
600597-- ============================================
601598-- Long-poll. The client supplies a cursor (last-seen `receivedAt`). We:
602599-- * Ack-and-drop everything <= the cursor (treating poll progress as ack).
603−-- * Wait on `pollWakeup` (bumped by send/heartbeat) and return new events
604−-- plus piggybacked sender-side expirations/deliveries.
600+-- * Race a row-level STM wait on `*messages` against `pollTimeoutMs` and
601+-- return new events plus piggybacked sender-side expirations/deliveries.
605602
606603-- POST /poll: long-poll body. Treats the cursor advance as an implicit ack
607604-- (`ackPolledMessages`), refreshes the caller's local presence row so peers
608605-- don't time us out mid-session, then runs `doPoll` to collect events (or
609−-- block until wakeup) and assembles the response.
610−handlePoll : Int -> Text -> Text -> IO {r *pollWakeup, r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, network, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
606+-- block until a matching message arrives or the timeout elapses) and
607+-- assembles the response.
608+handlePoll : Int -> Text -> Text -> IO {r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, network, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
611609handlePoll = \cursor authorization clientIp -> do
612610 t <- now
613611 setup <- atomic (do
@@ -617,11 +615,10 @@ handlePoll = \cursor authorization clientIp -> do
617615 Just {value: pubkey} -> do
618616 let clampedCursor = max 0 (min t cursor)
619617 when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)
620− startWakeup <- *pollWakeup
621618 pres <- *presence
622619 let stale = not any (\p -> p.pubkey == pubkey && isLocalServer p.server && p.expiresAt - t >= sessionExpiry / 2) pres
623620 refreshPeers <- if stale then refreshLocalPresence pubkey (t + sessionExpiry) else yield []
624− yield (Just {value: {pubkey, clampedCursor, startWakeup, refreshPeers}}))
621+ yield (Just {value: {pubkey, clampedCursor, refreshPeers}}))
625622 case setup of
626623 Nothing {} -> yield (Err {error: {status: 401, message: "unauthorized"}})
627624 Just {value: s} -> do
@@ -634,13 +631,11 @@ handlePoll = \cursor authorization clientIp -> do
634631-- max `receivedAt` in the page (or the input cursor when the page is empty),
635632-- which the client echoes on the next /poll to advance — and which the
636633-- server treats as an implicit ack.
637−-- If the page is empty, retry-wait on `pollWakeup` (bumped by send/heartbeat)
638−-- and re-collect once at the new wall time. The retry uses Knot's STM `retry`
639−-- primitive: the transaction blocks until some other transaction commits a
640−-- change to `*pollWakeup`. Because `pollHeartbeat` bumps `*pollWakeup` every
641−-- `pollTimeoutMs`, no caller waits longer than one heartbeat period before
642−-- getting at least an empty response.
643−doPoll : {clampedCursor: Int, pubkey: PubkeyHex, startWakeup: Timestamp | a} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{encryptedBlob: BlobHex}]}
634+-- If the page is empty, race a row-level STM wait against `pollTimeoutMs`:
635+-- the wait transaction reads `*messages` and `retry`s until a row matching
636+-- this pubkey/cursor is committed; the timer wins on idle. Either way we
637+-- re-collect at the new wall time, so a timeout returns an empty page.
638+doPoll : {clampedCursor: Int, pubkey: PubkeyHex | a} -> Int<Ms> -> IO {r *messages, clock} {cursor: Int<Ms>, events: [{encryptedBlob: BlobHex}]}
644639doPoll = \s t -> do
645640 let collectAt = \tnow -> atomic (do
646641 msgs <- *messages
@@ -654,7 +649,12 @@ doPoll = \s t -> do
654649 if count initial.events > 0
655650 then yield initial
656651 else do
657− atomic (do wakeup <- *pollWakeup; if wakeup > s.startWakeup then yield {} else retry)
652+ race
653+ (atomic (do
654+ msgs <- *messages
655+ let fresh = filter (\m -> m.toKey == s.pubkey && m.receivedAt > s.clampedCursor) msgs
656+ if count fresh > 0 then yield {} else retry))
657+ (sleep pollTimeoutMs)
658658 t2 <- now
659659 collectAt t2
660660
@@ -685,7 +685,7 @@ ackPolledMessages = \pubkey cursor -> do
685685-- rejected up-front; the remaining batch is committed inside one atomic
686686-- block (all-or-nothing) which reads presence to pre-stage federation
687687-- forwards. Federation forwards are kicked per message after commit.
688−handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> IO {r *presence, r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *pollWakeup, rw *seqCounter, network, clock} (Result HttpError {})
688+handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> IO {r *presence, r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *seqCounter, network, clock} (Result HttpError {})
689689handleSendMessages = \messages authorization clientIp -> do
690690 t <- now
691691 auth <- authedPubkey authorization t clientIp
@@ -781,7 +781,7 @@ handleRecvGossip = \events fromServer -> do
781781-- A → B forward based on stale gossip about R could land on B with no
782782-- actively-polling client and get stuck. If B refuses, A retries via
783783-- `retryPendingForwards` against the next presence target.
784−handleForwardMessage : PubkeyHex -> BlobHex -> IO {rw *messages, rw *pollWakeup, r *presence, rw *seqCounter, clock} (Result HttpError {})
784+handleForwardMessage : PubkeyHex -> BlobHex -> IO {rw *messages, r *presence, rw *seqCounter, clock} (Result HttpError {})
785785handleForwardMessage = \toKey encryptedBlob -> do
786786 t <- now
787787 atomic (do
@@ -814,9 +814,6 @@ api = serve Api where
814814-- Background loops
815815-- ============================================
816816-- `backgroundPrune` runs the prune/retry sweep every `backgroundPruneInterval` ms.
817−-- `pollHeartbeat` bumps `pollWakeup` every `pollTimeoutMs` so long-poll
818−-- holders unblock and return an empty heartbeat response, keeping connections
819−-- fresh.
820817
821818-- Endless tick: prune expired state, gossip offline for locally-expired keys,
822819-- and retry pending forwards. First sweep happens immediately so stale rows
@@ -833,24 +830,14 @@ backgroundPrune = do
833830 sleep backgroundPruneInterval
834831 backgroundPrune
835832
836−-- Endless tick that bumps `pollWakeup` every `pollTimeoutMs` so any blocked
837−-- /poll holders unblock and return an empty heartbeat response.
838−pollHeartbeat : IO {rw *pollWakeup, clock} a
839−pollHeartbeat = do
840− sleep pollTimeoutMs
841− t <- now
842− *pollWakeup = t
843− pollHeartbeat
844−
845833-- ============================================
846834-- Entry Point
847835-- ============================================
848−-- Fork the background loops (the prune loop runs an immediate sweep on its
836+-- Fork the background prune loop (which runs an immediate sweep on its
849837-- first tick) and start the HTTP listener.
850838main : IO _ {}
851839main = do
852840 logInfo ("Skrepka relay server starting on " ++ bindHost ++ ":" ++ show listenPort ++ " ...")
853841 logInfo (" federation: " ++ (if federationEnabled then "enabled" else "disabled"))
854842 fork backgroundPrune
855− fork pollHeartbeat
856843 listenOn bindHost listenPort api