7af7b29 · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Add generation tracking to poll protocol to prevent unnecessary cursor resets

7af7b29iko5 months ago

client/ApiClient.fs+3−3

compare with master

unifiedsplit

@@ -17,7 +17,7 @@ module ApiClient =

17 type PollEvent = { EncryptedBlob: string }

19 [<CLIMutable>]

20− type PollResponse = { Cursor: int64; Events: PollEvent array; Authorized: bool; Heartbeat: bool }

20+ type PollResponse = { Cursor: int64; Events: PollEvent array; Generation: int64; HasMore: bool }

22 /// Awaits a Task without converting TaskCanceledException to F# async

23 /// cancellation (which bypasses try...with). Re-raises it as a regular exception.

@@ -98,7 +98,7 @@ module ApiClient =

98 return raise (ApiError $"{serverUrl}/messages: {entryErr}")

99 }

100

101− let poll (serverUrl: string) (token: string) (cursor: int64) =

101+ let poll (serverUrl: string) (token: string) (cursor: int64) (generation: int64) =

102− let body = JsonSerializer.Serialize({| cursor = cursor |})

102+ let body = JsonSerializer.Serialize({| cursor = cursor; generation = generation |})

103 postJson<PollResponse> pollClient $"{serverUrl}/poll" body (Some token)

104

client/App.fs+54−54

@@ -132,6 +132,7 @@ module App =

132 Messages = Map.empty

133 Error = None

134 PollCursor = 0L

135+ ServerGeneration = 0L

135 Profile = None

136 Profile = None

136 FlushingOutbox = false

137 FlushingOutbox = false

137 PollRetries = 0

138 PollRetries = 0

@@ -181,7 +182,7 @@ module App =

181 Error = None

182 Error = None

182 PollStatus = "polling..."

183 PollStatus = "polling..."

183 FlushingOutbox = true },

184 FlushingOutbox = true },

184− [ CmdPoll(session, model.PollCursor, 0); CmdFlushOutbox session ]

185+ [ CmdPoll(session, model.PollCursor, model.ServerGeneration, 0); CmdFlushOutbox session ]

185 | _ -> model, []

186 | _ -> model, []

186

187

187 | AuthErr err ->

188 | AuthErr err ->

@@ -261,7 +262,7 @@ module App =

261 else

262 else

262 { model with FlushingOutbox = false }, []

263 { model with FlushingOutbox = false }, []

263

264

264− | PollResult(events, status, newCursor) ->

265+ | PollResult(events, status, newCursor, newGeneration) ->

265 let retries =

266 let retries =

266 if events.IsEmpty && status.StartsWith("poll error") then

267 if events.IsEmpty && status.StartsWith("poll error") then

267 model.PollRetries + 1

268 model.PollRetries + 1

@@ -275,11 +276,12 @@ module App =

275 { model with

276 { model with

276 PollStatus = status

277 PollStatus = status

277 PollCursor = newCursor

278 PollCursor = newCursor

279+ ServerGeneration = newGeneration

278 PollRetries = retries }

280 PollRetries = retries }

279

281

280 model',

282 model',

281 [ match trySession model' with

283 [ match trySession model' with

282− | Some session -> CmdPoll(session, model'.PollCursor, model'.PollRetries)

284+ | Some session -> CmdPoll(session, model'.PollCursor, model'.ServerGeneration, model'.PollRetries)

283 | None -> ()

285 | None -> ()

284 if not events.IsEmpty then

286 if not events.IsEmpty then

285 saveCmdMsg model' ]

287 saveCmdMsg model' ]

@@ -591,75 +593,73 @@ module App =

591 }

593 }

592 |> Async.Start)

594 |> Async.Start)

593

595

594− | CmdPoll(session, cursor, retries) ->

596+ | CmdPoll(session, cursor, generation, retries) ->

595 Cmd.ofEffect (fun dispatch ->

597 Cmd.ofEffect (fun dispatch ->

596 async {

598 async {

597 log $"poll start cursor={cursor}"

599 log $"poll start cursor={cursor}"

598

600

599 try

601 try

600− let! response = poll session.Url session.Token cursor

602+ let! response = poll session.Url session.Token cursor generation

601 log $"poll response: {response.Events.Length} events, cursor={response.Cursor}"

603 log $"poll response: {response.Events.Length} events, cursor={response.Cursor}"

602

604

603− if not response.Authorized then

605+ let results =

604− log "poll unauthorized, reauthenticating"

606+ response.Events

605− dispatch TokenExpired

607+ |> Array.map (fun evt -> decryptEvent session.Identity.PrivKey evt)

606− else

607

608

608− let results =

609+ let events = results |> Array.choose Result.toOption |> Array.toList

609− response.Events

610− |> Array.map (fun evt -> decryptEvent session.Identity.PrivKey evt)

611

610

612− let events = results |> Array.choose Result.toOption |> Array.toList

611+ let errors =

612+ results

613+ |> Array.choose (function

614+ | Error e -> Some e

615+ | _ -> None)

616+ |> Array.toList

613

617

614− let errors =

615− results

616− |> Array.choose (function

617− | Error e -> Some e

618− | _ -> None)

619− |> Array.toList

620−

621− for e in errors do

618+ for e in errors do

622− log $"decrypt error: {e}"

619+ log $"decrypt error: {e}"

623−

620+

624− if not events.IsEmpty then

621+ if not events.IsEmpty then

625− let acksBySender =

622+ let acksBySender =

626− events

627− |> List.choose (fun evt ->

628− match evt.Envelope with

629− | Envelope.TextMessage(id, _) -> Some(evt.Sender, id)

630− | _ -> None)

631− |> List.groupBy fst

632− |> List.map (fun (sender, pairs) -> sender, List.map snd pairs)

633−

634− for sender, msgIds in acksBySender do

635− try

636− do! sendEnvelope session sender (Envelope.DeliveryAck msgIds)

637− with ex ->

638− log $"delivery ack error: {ex.Message}"

639−

640− let chatCount =

641 events

623 events

624+ |> List.choose (fun evt ->

625+ match evt.Envelope with

626+ | Envelope.TextMessage(id, _) -> Some(evt.Sender, id)

627+ | _ -> None)

628+ |> List.groupBy fst

629+ |> List.map (fun (sender, pairs) -> sender, List.map snd pairs)

630+

631+ for sender, msgIds in acksBySender do

632+ try

633+ do! sendEnvelope session sender (Envelope.DeliveryAck msgIds)

634+ with ex ->

635+ log $"delivery ack error: {ex.Message}"

636+

637+ let chatCount =

638+ events

642− |> List.sumBy (fun e ->

639+ |> List.sumBy (fun e ->

643− match e.Envelope with

640+ match e.Envelope with

644− | Envelope.TextMessage _ -> 1

641+ | Envelope.TextMessage _ -> 1

645− | _ -> 0)

642+ | _ -> 0)

646−

643+

647− let status =

644+ let status =

648− $"evts:{response.Events.Length} msgs:{chatCount} errs:{errors.Length}"

645+ $"evts:{response.Events.Length} msgs:{chatCount} errs:{errors.Length}"

649− + (match List.tryHead errors with

646+ + (match List.tryHead errors with

650− | Some e -> $" [{e}]"

647+ | Some e -> $" [{e}]"

651− | None -> "")

648+ | None -> "")

652−

649+

653− dispatch (PollResult(events, status, response.Cursor))

650+ dispatch (PollResult(events, status, response.Cursor, response.Generation))

654 with

651 with

655 | :? TimeoutException ->

652 | :? TimeoutException ->

656 log "poll timeout"

653 log "poll timeout"

657− dispatch (PollResult([], "polling...", cursor))

654+ dispatch (PollResult([], "polling...", cursor, generation))

655+ | Unauthorized ->

656+ log "poll unauthorized, reauthenticating"

657+ dispatch TokenExpired

658 | ex ->

659 log $"poll error: {ex.Message}"

660 let delay = min (Constants.pollRetryBaseMs * (pown 2 retries)) 30000

661 do! Async.Sleep delay

662− dispatch (PollResult([], $"poll error: {ex.Message}", cursor))

662+ dispatch (PollResult([], $"poll error: {ex.Message}", cursor, generation))

663 }

664 |> Async.Start)

665

client/AppTypes.fs+3−2

@@ -57,6 +57,7 @@ module AppTypes =

57 Messages: Map<string, ChatMessage list>

58 Error: string option

59 PollCursor: int64

60+ ServerGeneration: int64

60 Profile: Profile option

61 Profile: Profile option

61 FlushingOutbox: bool

62 FlushingOutbox: bool

62 PollRetries: int

63 PollRetries: int

@@ -76,7 +77,7 @@ module AppTypes =

76 | DoDisconnect

77 | DoDisconnect

77 | DoSend

78 | DoSend

78 | DoSaveContact

79 | DoSaveContact

79− | PollResult of events: IncomingEvent list * status: string * cursor: int64

80+ | PollResult of events: IncomingEvent list * status: string * cursor: int64 * generation: int64

80 | CopyPubKey

81 | CopyPubKey

81 | DismissError

82 | DismissError

82 | StateLoaded of Identity option * Data option * Profile option

83 | StateLoaded of Identity option * Data option * Profile option

@@ -98,7 +99,7 @@ module AppTypes =

98 | CmdConnect of url: string * identity: Identity

99 | CmdConnect of url: string * identity: Identity

99 | CmdEnqueue of recipientHex: string * envelope: Envelope

100 | CmdEnqueue of recipientHex: string * envelope: Envelope

100 | CmdFlushOutbox of session: Session

101 | CmdFlushOutbox of session: Session

101− | CmdPoll of session: Session * cursor: int64 * retries: int

102+ | CmdPoll of session: Session * cursor: int64 * generation: int64 * retries: int

102 | CmdCopyToClipboard of string

103 | CmdCopyToClipboard of string

103 | CmdLoadState

104 | CmdLoadState

104 | CmdSaveIdentity of Identity

105 | CmdSaveIdentity of Identity

server.knot+12−43

@@ -325,7 +325,7 @@ route AuthApi where

325 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /auth/verify headers {clientIp: Text} -> {token: Text, expiresAt: Timestamp} = VerifyAuth

326

327route ClientApi where

328− POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int} = Poll

328+ POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, generation: Int, hasMore: Bool} = Poll

329 POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, clientIp: Text} -> {results: [{error: Text}]} = SendMessages

330

331route FederationApi where

@@ -404,22 +404,6 @@ capList = \items maxN getKey -> dropRelation (max 0 (count items - maxN)) (sortB

404capPerKey : [a] -> (a -> Bool) -> Int -> (a -> b) -> [a]

405capPerKey = \items pred maxN getKey -> union (removeWhere items pred) (capList (filter items pred) (maxN - 1) getKey)

406

407−-- Shared empty/error shape for /poll. Only a few fields ever vary.

408−emptyPollResp : Timestamp -> Bool -> Bool -> Int -> Int -> Bool -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int}

409−emptyPollResp = \cursor authorized rateLimited gen retryAfterMs cursorWasReset -> {

410− events: [],

411− cursor,

412− authorized,

413− rateLimited,

414− generation: gen,

415− heartbeat: False {},

416− retryAfterMs,

417− cursorReset: cursorWasReset,

418− hasMore: False {},

419− cursorStale: False {},

420− loadPct: 0

421−}

422−

423-- Sequence an IO action over each item, in order. Recurses through the list

407-- Sequence an IO action over each item, in order. Recurses through the list

424-- one element at a time; this is the closest equivalent of `mapM_` here.

408-- one element at a time; this is the closest equivalent of `mapM_` here.

425forEach : [a] -> (a -> IO {} {}) -> IO {} {}

409forEach : [a] -> (a -> IO {} {}) -> IO {} {}

@@ -937,50 +921,35 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do

937

921

938-- POST /poll: top-level dispatch. Validates the global rate gate, detects a

922-- POST /poll: top-level dispatch. Validates the global rate gate, detects a

939-- generation mismatch (server restart) to force a cursor reset, then hands

923-- generation mismatch (server restart) to force a cursor reset, then hands

940−-- off to `pollOnce` for the actual long-poll body.

924+-- off to `pollOnce` for the actual long-poll body. Clients detect a server

925+-- restart by comparing the `generation` they sent to the one returned.

941−handlePoll : Int -> Int -> Text -> Text -> IO {r *pollWakeup, r *serverGeneration, r *sessions, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, clock} (Result HttpError {authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int})

926+handlePoll : Int -> Int -> Text -> Text -> IO {r *pollWakeup, r *serverGeneration, r *sessions, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent], generation: Int, hasMore: Bool})

942handlePoll = \cursor generation authorization clientIp -> do

927handlePoll = \cursor generation authorization clientIp -> do

943 t <- now

928 t <- now

944 serverGen <- *serverGeneration

929 serverGen <- *serverGeneration

945 g <- checkGlobalRate t

930 g <- checkGlobalRate t

946− let cursorWasReset = generation != serverGen && cursor > 0

947− let er = \auth rate retryRate -> yield (Ok {

948− value: emptyPollResp 0 auth rate serverGen retryRate cursorWasReset

949− })

950 if g

931 if g

951− then er (False {}) (True {}) globalRateWindowMs

932+ then yield (Err {error: {status: 429, message: "rate_limited"}})

933+ else do

934+ let cursorWasReset = generation != serverGen && cursor > 0

952− else pollOnce authorization (if cursorWasReset then 0 else cursor) cursorWasReset clientIp serverGen t

935+ pollOnce authorization (if cursorWasReset then 0 else cursor) clientIp serverGen t

953

936

954-- Body of a single poll. Treats the cursor advance as an implicit ack

937-- Body of a single poll. Treats the cursor advance as an implicit ack

955-- (`ackPolledMessages`), then runs `doPoll` to collect events (or block

938-- (`ackPolledMessages`), then runs `doPoll` to collect events (or block

956-- until wakeup) and assembles the response.

939-- until wakeup) and assembles the response.

957−pollOnce : Text -> Int -> Bool -> Text -> Int -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *forwards, rw *messages, clock} (Result HttpError {authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int})

940+pollOnce : Text -> Int -> Text -> Int -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *forwards, rw *messages, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent], generation: Int, hasMore: Bool})

958−pollOnce = \authorization cursor cursorWasReset clientIp serverGen t -> do

941+pollOnce = \authorization cursor clientIp serverGen t -> do

959 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do

942 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do

960 let clampedCursor = clamp 0 t cursor

943 let clampedCursor = clamp 0 t cursor

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

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

962 startWakeup <- *pollWakeup

945 startWakeup <- *pollWakeup

963 yield (Just {value: {pubkey, clampedCursor, startWakeup}}))

946 yield (Just {value: {pubkey, clampedCursor, startWakeup}}))

964 case setup of

947 case setup of

965− Nothing {} -> yield (Ok {

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

966− value: emptyPollResp 0 (False {}) (False {}) serverGen 0 cursorWasReset

967− })

968 Just {value: s} -> do

949 Just {value: s} -> do

969 pollResult <- doPoll s t

950 pollResult <- doPoll s t

970 yield (Ok {

951 yield (Ok {

971− value: {

972− pollResult.events,

973− pollResult.cursor,

974− authorized: True {},

975− rateLimited: False {},

976− generation: serverGen,

952+ value: {pollResult.events, pollResult.cursor, generation: serverGen, pollResult.hasMore}

977− heartbeat: count pollResult.events == 0,

978− retryAfterMs: 0,

979− cursorReset: cursorWasReset,

980− pollResult.hasMore,

981− cursorStale: False {},

982− loadPct: 0

983− }

984 })

953 })

985

954

986-- One pass: collect events at `t`. If empty, retry-wait on `pollWakeup`

955-- One pass: collect events at `t`. If empty, retry-wait on `pollWakeup`