bd17e0a · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Remove generation field from /poll protocol

bd17e0aiko5 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; Generation: int64 }

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

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

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

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

95 return raise (ServerRejected "Message rejected by server")

96 }

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

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

99− let body = JsonSerializer.Serialize({| cursor = cursor; generation = generation |})

99+ let body = JsonSerializer.Serialize({| cursor = cursor |})

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

101

client/App.fs+8−10

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

132 Messages = Map.empty

133 Error = None

134 PollCursor = 0L

135− ServerGeneration = 0L

136 Profile = None

135 Profile = None

137 FlushingOutbox = false

136 FlushingOutbox = false

138 PollRetries = 0

137 PollRetries = 0

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

182 Error = None

181 Error = None

183 PollStatus = "polling..."

182 PollStatus = "polling..."

184 FlushingOutbox = true },

183 FlushingOutbox = true },

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

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

186 | _ -> model, []

185 | _ -> model, []

187

186

188 | AuthErr err ->

187 | AuthErr err ->

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

262 else

261 else

263 { model with FlushingOutbox = false }, []

262 { model with FlushingOutbox = false }, []

264

263

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

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

266 let retries =

265 let retries =

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

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

268 model.PollRetries + 1

267 model.PollRetries + 1

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

276 { model with

275 { model with

277 PollStatus = status

276 PollStatus = status

278 PollCursor = newCursor

277 PollCursor = newCursor

279− ServerGeneration = newGeneration

280 PollRetries = retries }

278 PollRetries = retries }

281

279

282 model',

280 model',

283 [ match trySession model' with

281 [ match trySession model' with

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

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

285 | None -> ()

283 | None -> ()

286 if not events.IsEmpty then

284 if not events.IsEmpty then

287 saveCmdMsg model' ]

285 saveCmdMsg model' ]

@@ -593,13 +591,13 @@ module App =

593 }

591 }

594 |> Async.Start)

592 |> Async.Start)

595

593

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

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

597 Cmd.ofEffect (fun dispatch ->

595 Cmd.ofEffect (fun dispatch ->

598 async {

596 async {

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

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

600

598

601 try

599 try

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

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

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

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

604

602

605 let results =

603 let results =

@@ -647,11 +645,11 @@ module App =

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

645 | Some e -> $" [{e}]"

648 | None -> "")

646 | None -> "")

649

647

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

648+ dispatch (PollResult(events, status, response.Cursor))

651 with

649 with

652 | :? TimeoutException ->

650 | :? TimeoutException ->

653 log "poll timeout"

651 log "poll timeout"

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

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

655 | Unauthorized ->

653 | Unauthorized ->

656 log "poll unauthorized, reauthenticating"

654 log "poll unauthorized, reauthenticating"

657 dispatch TokenExpired

655 dispatch TokenExpired

@@ -659,7 +657,7 @@ module App =

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

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

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

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

661 do! Async.Sleep delay

659 do! Async.Sleep delay

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

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

663 }

661 }

664 |> Async.Start)

662 |> Async.Start)

665

663

client/AppTypes.fs+2−3

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

57 Messages: Map<string, ChatMessage list>

58 Error: string option

59 PollCursor: int64

60− ServerGeneration: int64

61 Profile: Profile option

60 Profile: Profile option

62 FlushingOutbox: bool

61 FlushingOutbox: bool

63 PollRetries: int

62 PollRetries: int

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

77 | DoDisconnect

76 | DoDisconnect

78 | DoSend

77 | DoSend

79 | DoSaveContact

78 | DoSaveContact

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

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

81 | CopyPubKey

80 | CopyPubKey

82 | DismissError

81 | DismissError

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

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

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

99 | CmdConnect of url: string * identity: Identity

98 | CmdConnect of url: string * identity: Identity

100 | CmdEnqueue of recipientHex: string * envelope: Envelope

99 | CmdEnqueue of recipientHex: string * envelope: Envelope

101 | CmdFlushOutbox of session: Session

100 | CmdFlushOutbox of session: Session

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

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

103 | CmdCopyToClipboard of string

102 | CmdCopyToClipboard of string

104 | CmdLoadState

103 | CmdLoadState

105 | CmdSaveIdentity of Identity

104 | CmdSaveIdentity of Identity

server.knot+13−26

@@ -305,9 +305,6 @@ type RateEntry = {key: Text, windowStart: Timestamp, count: Int}

305*lastPresenceRefresh : Timestamp

306-- Wall time of process start; gates /ready (must be ≤ lastPrunedAt).

307*startedAt : Timestamp

308−-- Random 64-bit-ish identity stamped at startup. Clients echo it back on /poll;

309−-- any mismatch means the server restarted and the client must reset its cursor.

310−*serverGeneration : Int

311

308

312-- ============================================

309-- ============================================

313-- Routes

310-- Routes

@@ -326,7 +323,7 @@ route AuthApi where

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

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

327

324

328route ClientApi where

325route ClientApi where

329− POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, generation: Int} = Poll

326+ POST {cursor: Timestamp} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp} = Poll

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

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

331

328

332route FederationApi where

329route FederationApi where

@@ -913,33 +910,26 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do

913-- ============================================

910-- ============================================

914-- Handler: poll

911-- Handler: poll

915-- ============================================

912-- ============================================

916−-- Long-poll. The client supplies a cursor (last-seen `receivedAt`) and the

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

917−-- server's last-known `generation` (server identity for cursor validity). We:

918−-- * If the generation mismatches, reset the cursor and signal `cursorReset`.

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

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

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

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

921-- plus piggybacked sender-side expirations/deliveries.

916-- plus piggybacked sender-side expirations/deliveries.

922

917

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

918+-- POST /poll: top-level dispatch. Validates the global rate gate, then hands

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

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

919+-- off to `pollOnce` for the actual long-poll body.

926−-- restart by comparing the `generation` they sent to the one returned.

927−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})

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

928−handlePoll = \cursor generation authorization clientIp -> do

921+handlePoll = \cursor authorization clientIp -> do

929 t <- now

922 t <- now

930− serverGen <- *serverGeneration

931 g <- checkGlobalRate t

923 g <- checkGlobalRate t

932 if g

924 if g

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

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

934− else do

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

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

926+ else pollOnce authorization cursor clientIp t

937

927

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

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

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

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

940-- until wakeup) and assembles the response.

930-- until wakeup) and assembles the response.

941−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})

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

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

932+pollOnce = \authorization cursor clientIp t -> do

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

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

944 let clampedCursor = clamp 0 t cursor

934 let clampedCursor = clamp 0 t cursor

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

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

@@ -950,7 +940,7 @@ pollOnce = \authorization cursor clientIp serverGen t -> do

950 Just {value: s} -> do

940 Just {value: s} -> do

951 pollResult <- doPoll s t

941 pollResult <- doPoll s t

952 yield (Ok {

942 yield (Ok {

953− value: {pollResult.events, pollResult.cursor, generation: serverGen}

943+ value: {pollResult.events, pollResult.cursor}

954 })

944 })

955

945

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

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

@@ -1230,7 +1220,7 @@ api : Server Api _

1230api = serve Api where

1220api = serve Api where

1231 ReqChallenge = \{pubkey, clientIp} -> handleChallenge pubkey clientIp

1221 ReqChallenge = \{pubkey, clientIp} -> handleChallenge pubkey clientIp

1232 VerifyAuth = \{pubkey, challenge, signature, revokeOthers, clientIp} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp

1222 VerifyAuth = \{pubkey, challenge, signature, revokeOthers, clientIp} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp

1233− Poll = \{cursor, generation, authorization, clientIp} -> handlePoll cursor generation authorization clientIp

1223+ Poll = \{cursor, authorization, clientIp} -> handlePoll cursor authorization clientIp

1234 SendMessages = \{messages, authorization, clientIp} -> handleSendMessages messages authorization clientIp

1224 SendMessages = \{messages, authorization, clientIp} -> handleSendMessages messages authorization clientIp

1235 RecvGossip = \{events, fromServer} -> handleRecvGossip events fromServer

1225 RecvGossip = \{events, fromServer} -> handleRecvGossip events fromServer

1236 ForwardMessage = \{toKey, encryptedBlob, fromServer} -> handleForwardMessage toKey encryptedBlob fromServer

1226 ForwardMessage = \{toKey, encryptedBlob, fromServer} -> handleForwardMessage toKey encryptedBlob fromServer

@@ -1265,17 +1255,14 @@ pollHeartbeat = do

1265-- ============================================

1255-- ============================================

1266-- Entry Point

1256-- Entry Point

1267-- ============================================

1257-- ============================================

1268−-- Stamp `startedAt` and a fresh `serverGeneration` (used by clients to detect

1269−-- server restarts and reset their poll cursor), spawn the two background

1270−-- loops, and start the listener.

1258+-- Stamp `startedAt`, spawn the two background loops, and start the listener.

1271

1259

1272-- Entry point: stamp startup metadata, fork the background loops, and start

1260-- Entry point: stamp startup metadata, fork the background loops, and start

1273-- the HTTP listener.

1261-- the HTTP listener.

1274main : IO _ {}

1262main : IO _ {}

1275main = do

1263main = do

1276 t <- now

1264 t <- now

1277− genJitter <- randomInt 1000000

1278− atomic (do *startedAt = t; *serverGeneration = t * 1000 + genJitter)

1265+ atomic (do *startedAt = t)

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

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

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

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

1281 fork backgroundPrune

1268 fork backgroundPrune