5cc044c · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Use row-level STM retry with per-request timeout race for long-poll

5cc044ciko5 months ago

server.knot+23−36

compare with master

unifiedsplit

@@ -227,8 +227,6 @@ type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}

227

228-- Strictly-monotonic sequence used as `Message.receivedAt`; guarantees poll-cursor ordering.

229*seqCounter : Timestamp

230−-- Bumped by send/heartbeat to wake all blocked /poll holders.

231−*pollWakeup : Timestamp

232

230

233-- ============================================

231-- ============================================

234-- Routes

232-- Routes

@@ -413,16 +411,15 @@ collectPeers : [a] -> (a -> Text) -> [Text]

413collectPeers = \items getServer -> filter (\s -> not isLocalServer s) (map getServer items)

411collectPeers = \items getServer -> filter (\s -> not isLocalServer s) (map getServer items)

414

412

415-- Append one ciphertext to the local mailbox under a fresh monotonic

413-- Append one ciphertext to the local mailbox under a fresh monotonic

416−-- sequence and bump pollWakeup so blocked /poll holders unblock. Caller

414+-- sequence. Row-level STM retry on `*messages` wakes any parked /poll

417−-- wraps in `atomic`.

415+-- transaction whose filter matches the new row. Caller wraps in `atomic`.

418−appendMessage : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWakeup, rw *seqCounter} {}

416+appendMessage : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *seqCounter} {}

419appendMessage = \toKey encryptedBlob t -> do

417appendMessage = \toKey encryptedBlob t -> do

420 msgs <- *messages

418 msgs <- *messages

421 seq <- *seqCounter

419 seq <- *seqCounter

422 let seqTs = max t (seq + 1)

420 let seqTs = max t (seq + 1)

423 *seqCounter = seqTs

421 *seqCounter = seqTs

424 *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs}]

422 *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs}]

425− *pollWakeup = seqTs

426

423

427-- True when `f` is the queued forward of `blob` to `server`. Used to dedupe

424-- True when `f` is the queued forward of `blob` to `server`. Used to dedupe

428-- per-(blob, server) entries on the forwards outbox.

425-- per-(blob, server) entries on the forwards outbox.

@@ -600,14 +597,15 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do

600-- ============================================

597-- ============================================

601-- Long-poll. The client supplies a cursor (last-seen `receivedAt`). We:

598-- Long-poll. The client supplies a cursor (last-seen `receivedAt`). We:

602-- * Ack-and-drop everything <= the cursor (treating poll progress as ack).

599-- * Ack-and-drop everything <= the cursor (treating poll progress as ack).

603−-- * Wait on `pollWakeup` (bumped by send/heartbeat) and return new events

600+-- * Race a row-level STM wait on `*messages` against `pollTimeoutMs` and

604−-- plus piggybacked sender-side expirations/deliveries.

601+-- return new events plus piggybacked sender-side expirations/deliveries.

605

602

606-- POST /poll: long-poll body. Treats the cursor advance as an implicit ack

603-- POST /poll: long-poll body. Treats the cursor advance as an implicit ack

607-- (`ackPolledMessages`), refreshes the caller's local presence row so peers

604-- (`ackPolledMessages`), refreshes the caller's local presence row so peers

608-- don't time us out mid-session, then runs `doPoll` to collect events (or

605-- don't time us out mid-session, then runs `doPoll` to collect events (or

606+-- block until a matching message arrives or the timeout elapses) and

609−-- block until wakeup) and assembles the response.

607+-- 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]})

608+handlePoll : Int -> Text -> Text -> IO {r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, network, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})

611handlePoll = \cursor authorization clientIp -> do

609handlePoll = \cursor authorization clientIp -> do

612 t <- now

610 t <- now

613 setup <- atomic (do

611 setup <- atomic (do

@@ -617,11 +615,10 @@ handlePoll = \cursor authorization clientIp -> do

617 Just {value: pubkey} -> do

615 Just {value: pubkey} -> do

618 let clampedCursor = max 0 (min t cursor)

616 let clampedCursor = max 0 (min t cursor)

619 when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)

617 when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)

620− startWakeup <- *pollWakeup

621 pres <- *presence

618 pres <- *presence

622 let stale = not any (\p -> p.pubkey == pubkey && isLocalServer p.server && p.expiresAt - t >= sessionExpiry / 2) pres

619 let stale = not any (\p -> p.pubkey == pubkey && isLocalServer p.server && p.expiresAt - t >= sessionExpiry / 2) pres

623 refreshPeers <- if stale then refreshLocalPresence pubkey (t + sessionExpiry) else yield []

620 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}}))

625 case setup of

622 case setup of

626 Nothing {} -> yield (Err {error: {status: 401, message: "unauthorized"}})

623 Nothing {} -> yield (Err {error: {status: 401, message: "unauthorized"}})

627 Just {value: s} -> do

624 Just {value: s} -> do

@@ -634,13 +631,11 @@ handlePoll = \cursor authorization clientIp -> do

634-- max `receivedAt` in the page (or the input cursor when the page is empty),

631-- max `receivedAt` in the page (or the input cursor when the page is empty),

635-- which the client echoes on the next /poll to advance — and which the

632-- which the client echoes on the next /poll to advance — and which the

636-- server treats as an implicit ack.

633-- server treats as an implicit ack.

637−-- If the page is empty, retry-wait on `pollWakeup` (bumped by send/heartbeat)

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

638−-- and re-collect once at the new wall time. The retry uses Knot's STM `retry`

637+-- re-collect at the new wall time, so a timeout returns an empty page.

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}]}

638+doPoll : {clampedCursor: Int, pubkey: PubkeyHex | a} -> Int<Ms> -> IO {r *messages, clock} {cursor: Int<Ms>, events: [{encryptedBlob: BlobHex}]}

644doPoll = \s t -> do

639doPoll = \s t -> do

645 let collectAt = \tnow -> atomic (do

640 let collectAt = \tnow -> atomic (do

646 msgs <- *messages

641 msgs <- *messages

@@ -654,7 +649,12 @@ doPoll = \s t -> do

654 if count initial.events > 0

649 if count initial.events > 0

655 then yield initial

650 then yield initial

656 else do

651 else do

652+ race

653+ (atomic (do

654+ msgs <- *messages

655+ let fresh = filter (\m -> m.toKey == s.pubkey && m.receivedAt > s.clampedCursor) msgs

657− atomic (do wakeup <- *pollWakeup; if wakeup > s.startWakeup then yield {} else retry)

656+ if count fresh > 0 then yield {} else retry))

657+ (sleep pollTimeoutMs)

658 t2 <- now

659 collectAt t2

660

@@ -685,7 +685,7 @@ ackPolledMessages = \pubkey cursor -> do

685-- rejected up-front; the remaining batch is committed inside one atomic

686-- block (all-or-nothing) which reads presence to pre-stage federation

687-- 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 {})

689handleSendMessages = \messages authorization clientIp -> do

690 t <- now

691 auth <- authedPubkey authorization t clientIp

@@ -781,7 +781,7 @@ handleRecvGossip = \events fromServer -> do

781-- A → B forward based on stale gossip about R could land on B with no

782-- actively-polling client and get stuck. If B refuses, A retries via

783-- `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 {})

785handleForwardMessage = \toKey encryptedBlob -> do

786 t <- now

787 atomic (do

@@ -814,9 +814,6 @@ api = serve Api where

814-- Background loops

815-- ============================================

816-- `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.

820

817

821-- Endless tick: prune expired state, gossip offline for locally-expired keys,

818-- Endless tick: prune expired state, gossip offline for locally-expired keys,

822-- and retry pending forwards. First sweep happens immediately so stale rows

819-- and retry pending forwards. First sweep happens immediately so stale rows

@@ -833,24 +830,14 @@ backgroundPrune = do

833 sleep backgroundPruneInterval

830 sleep backgroundPruneInterval

834 backgroundPrune

831 backgroundPrune

835

832

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−

845-- ============================================

833-- ============================================

846-- Entry Point

834-- Entry Point

847-- ============================================

835-- ============================================

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

849-- first tick) and start the HTTP listener.

837-- first tick) and start the HTTP listener.

850main : IO _ {}

838main : IO _ {}

851main = do

839main = do

852 logInfo ("Skrepka relay server starting on " ++ bindHost ++ ":" ++ show listenPort ++ " ...")

840 logInfo ("Skrepka relay server starting on " ++ bindHost ++ ":" ++ show listenPort ++ " ...")

853 logInfo (" federation: " ++ (if federationEnabled then "enabled" else "disabled"))

841 logInfo (" federation: " ++ (if federationEnabled then "enabled" else "disabled"))

854 fork backgroundPrune

842 fork backgroundPrune

855− fork pollHeartbeat

856 listenOn bindHost listenPort api

843 listenOn bindHost listenPort api