1 import Base 2 import ./Client.bend as Client 3 import ./ClientPolicy.bend as Policy 4 import ./Protocol.bend as Protocol 5 import ./CursorCodec.bend as CursorCodec 6 import ./OwnerCursorCodec.bend as OwnerCursorCodec 7 import ./CursorCountPolicy.bend as CursorCountPolicy 8 import ./CursorOpenCodec.bend as CursorOpenCodec 9 import ./StatsCodec.bend as StatsCodec 10 import ./OptionsCodec.bend as OptionsCodec 11 import ../ByteWal.bend as ByteWal 12 import ../Wal.bend as Wal 13 import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes 14 15 # Carries a bounded remote cursor count across incremental owner reads. 16 type SnapshotCountState is Type: 17 SnapshotCountNeed{session: Client.Session, cursor_ids: U32 & U32, 18 count: U32} 19 SnapshotCountFrame{pair: 20 Client.Session & Result<&1, &1, Client.Error, CursorCodec.Frame>, 21 cursor_id: U32, count: U32} 22 SnapshotCountItem{session: Client.Session, cursor_id: U32, 23 decision: CursorCountPolicy.Decision} 24 25 def describe.result( 26 pair: Client.Session & Result<&1, &1, Client.Error, Protocol.Response> 27 ) -> Client.Session & Result<&1, &1, Client.Error, OptionsCodec.Decision>: 28 match pair: 29 case (session, Fail{error}): (session, Fail{error}) 30 case (session, Done{Protocol.Response{_, _, _, status, payload}}): 31 match status: 32 case Protocol.Success{}: (session, Done{OptionsCodec.decode(payload)}) 33 case _: (session, Fail{Client.OwnerRejected{status}}) 34 35 # Handles describe owner. 36 def describe_owner( 37 session: Client.Session 38 ) -> IO(Client.Session & Result<&1, &1, Client.Error, OptionsCodec.Decision>): 39 do IO>: 40 pair : Client.Session & 41 Result<&1, &1, Client.Error, Protocol.Response> <- 42 Client.session.describe_owner(session) 43 return describe.result(pair) 44 def result.failure( 45 error: Client.Error 46 ) -> Result<&1, &1, Client.Error, Maybe<&1, Bytes.Bytes>>: 47 Fail{error} 48 49 def result.value( 50 value: Maybe<&1, Bytes.Bytes> 51 ) -> Result<&1, &1, Client.Error, Maybe<&1, Bytes.Bytes>>: 52 Done{value} 53 54 def get.outcome( 55 outcome: Policy.GetStatus, 56 payload: Bytes.Bytes 57 ) -> Result<&1, &1, Client.Error, Maybe<&1, Bytes.Bytes>>: 58 match outcome: 59 case Policy.GetValueStatus{}: result.value(Some{payload}) 60 case Policy.GetMissingStatus{}: result.value(None{}) 61 case Policy.GetFailureStatus{status}: 62 result.failure(Client.OwnerRejected{status}) 63 64 def get.status( 65 status: Protocol.ResponseStatus, 66 payload: Bytes.Bytes 67 ) -> Result<&1, &1, Client.Error, Maybe<&1, Bytes.Bytes>>: 68 get.outcome(Policy.response.get.status(status), payload) 69 70 def get.response( 71 result: Result<&1, &1, Client.Error, Protocol.Response> 72 ) -> Result<&1, &1, Client.Error, Maybe<&1, Bytes.Bytes>>: 73 match result: 74 case Fail{error}: result.failure(error) 75 case Done{response}: 76 match response: 77 case Protocol.Response{_, _, _, status, payload}: 78 get.status(status, payload) 79 80 def get.result( 81 pair: Client.Session & Result<&1, &1, Client.Error, Protocol.Response> 82 ) -> Client.Session & Result<&1, &1, Client.Error, Maybe<&1, Bytes.Bytes>>: 83 match pair: 84 case (session, result): (session, get.response(result)) 85 86 # Reads one raw key and maps owner NotFound to an absent value. 87 def get( 88 session: Client.Session, 89 key: Bytes.Bytes 90 ) -> IO(Client.Session & Result<&1, &1, Client.Error, Maybe<&1, Bytes.Bytes>>): 91 do IO>>: 92 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 93 Client.session.get(session, key) 94 return get.result(pair) 95 96 def write.payload( 97 is_empty: Bool 98 ) -> Result<&1, &1, Client.Error, Unit>: 99 match is_empty: 100 case True{}: Done{Unit{}} 101 case False{}: Fail{Client.MalformedResponse{}} 102 103 def write.status( 104 status: Protocol.ResponseStatus, 105 payload: Bytes.Bytes & U32 106 ) -> Result<&1, &1, Client.Error, Unit>: 107 match status: 108 case Protocol.Success{}: 109 match payload: 110 case (_, length): write.payload(U32.is_eq(length, 0)) 111 case _: Fail{Client.OwnerRejected{status}} 112 113 def write.response( 114 result: Result<&1, &1, Client.Error, Protocol.Response> 115 ) -> Result<&1, &1, Client.Error, Unit>: 116 match result: 117 case Fail{error}: Fail{error} 118 case Done{response}: 119 match response: 120 case Protocol.Response{_, _, _, status, payload}: 121 write.status(status, Bytes.length(payload)) 122 123 def write.result( 124 pair: Client.Session & Result<&1, &1, Client.Error, Protocol.Response> 125 ) -> Client.Session & Result<&1, &1, Client.Error, Unit>: 126 match pair: 127 case (session, result): (session, write.response(result)) 128 129 # Writes one raw key/value pair through the serialized owner session. 130 def put( 131 session: Client.Session, 132 key: Bytes.Bytes, 133 value: Bytes.Bytes 134 ) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>): 135 do IO>: 136 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 137 Client.session.put(session, key, value) 138 return write.result(pair) 139 140 # Deletes one raw key through the serialized owner session. 141 def delete( 142 session: Client.Session, 143 key: Bytes.Bytes 144 ) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>): 145 do IO>: 146 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 147 Client.session.delete(session, key) 148 return write.result(pair) 149 150 # Flushes the owner's durable WAL state through the serialized session. 151 def flush( 152 session: Client.Session 153 ) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>): 154 do IO>: 155 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 156 Client.session.flush(session) 157 return write.result(pair) 158 159 # Compacts the owner's durable state through the serialized session. 160 def compact( 161 session: Client.Session 162 ) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>): 163 do IO>: 164 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 165 Client.session.compact(session) 166 return write.result(pair) 167 168 def write_batch.encoded( 169 session: Client.Session, 170 encoded: Result<&1, &1, Wal.Error, Bytes.Bytes> 171 ) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>): 172 match encoded: 173 case Fail{_}: 174 IO.pure(Client.Session & Result<&1, &1, Client.Error, Unit>, 175 (session, Fail{Client.FrameTooLarge{}})) 176 case Done{payload}: 177 do IO>: 178 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 179 Client.session.call(session, 8, payload) 180 return write.result(pair) 181 182 # Commits raw byte mutations as one owner-side atomic batch. 183 def write_batch( 184 session: Client.Session, 185 mutations: List<&1, ByteWal.Mutation> 186 ) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>): 187 write_batch.encoded(session, ByteWal.encode(ByteWal.Batch{mutations})) 188 189 def stats.payload( 190 decision: StatsCodec.Decision 191 ) -> Result<&1, &1, Client.Error, StatsCodec.Stats>: 192 match decision: 193 case StatsCodec.Parsed{stats}: Done{stats} 194 case StatsCodec.Malformed{}: Fail{Client.MalformedResponse{}} 195 196 def stats.status( 197 status: Protocol.ResponseStatus, 198 payload: Bytes.Bytes 199 ) -> Result<&1, &1, Client.Error, StatsCodec.Stats>: 200 match status: 201 case Protocol.Success{}: stats.payload(StatsCodec.parse(payload)) 202 case _: Fail{Client.OwnerRejected{status}} 203 204 def stats.response( 205 response: Protocol.Response 206 ) -> Result<&1, &1, Client.Error, StatsCodec.Stats>: 207 match response: 208 case Protocol.Response{_, _, _, status, payload}: 209 stats.status(status, payload) 210 211 def stats.result( 212 pair: Client.Session & Result<&1, &1, Client.Error, Protocol.Response> 213 ) -> Client.Session & Result<&1, &1, Client.Error, StatsCodec.Stats>: 214 match pair: 215 case (session, Fail{error}): (session, Fail{error}) 216 case (session, Done{response}): (session, stats.response(response)) 217 218 # Reads owner entry and payload counters from a typed statistics response. 219 def stats( 220 session: Client.Session 221 ) -> IO(Client.Session & Result<&1, &1, Client.Error, StatsCodec.Stats>): 222 do IO>: 223 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 224 Client.session.stats(session) 225 return stats.result(pair) 226 227 def snapshot.open.parsed( 228 parsed: Maybe<&2, U32> 229 ) -> Result<&1, &1, Client.Error, U32>: 230 match parsed: 231 case Some{snapshot_id}: Done{snapshot_id} 232 case None{}: Fail{Client.MalformedResponse{}} 233 234 def snapshot.open.status( 235 status: Protocol.ResponseStatus, 236 payload: Bytes.Bytes 237 ) -> Result<&1, &1, Client.Error, U32>: 238 match status: 239 case Protocol.Success{}: 240 snapshot.open.parsed(Protocol.snapshot.id.read(payload)) 241 case _: Fail{Client.OwnerRejected{status}} 242 243 def snapshot.open.response( 244 result: Result<&1, &1, Client.Error, Protocol.Response> 245 ) -> Result<&1, &1, Client.Error, U32>: 246 match result: 247 case Fail{error}: Fail{error} 248 case Done{response}: 249 match response: 250 case Protocol.Response{_, _, _, status, payload}: 251 snapshot.open.status(status, payload) 252 253 def snapshot.open.result( 254 pair: Client.Session & Result<&1, &1, Client.Error, Protocol.Response> 255 ) -> Client.Session & Result<&1, &1, Client.Error, U32>: 256 match pair: 257 case (session, result): (session, snapshot.open.response(result)) 258 259 # Requests a retained snapshot from the serialized owner and returns its ID. 260 def snapshot.open( 261 session: Client.Session 262 ) -> IO(Client.Session & Result<&1, &1, Client.Error, U32>): 263 do IO>: 264 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 265 Client.session.call(session, 9, Bytes.new(0)) 266 return snapshot.open.result(pair) 267 268 # Prefixes a byte key with its fixed-width snapshot token. 269 def snapshot.get.payload(snapshot_id: U32, key: Bytes.Bytes) -> Bytes.Bytes: 270 Bytes.concat([Protocol.snapshot.id.bytes(snapshot_id), key]) 271 272 # Reads one key through a retained owner snapshot without changing its token. 273 def snapshot.get( 274 session: Client.Session, 275 snapshot_id: U32, 276 key: Bytes.Bytes 277 ) -> IO(Client.Session & Result<&1, &1, Client.Error, Maybe<&1, Bytes.Bytes>>): 278 do IO>>: 280 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 281 Client.session.call(session, 10, snapshot.get.payload(snapshot_id, key)) 282 return get.result(pair) 283 284 # Releases an owner snapshot token after its final cursor has closed. 285 def snapshot.close( 286 session: Client.Session, 287 snapshot_id: U32 288 ) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>): 289 do IO>: 290 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 291 Client.session.call(session, 11, Protocol.snapshot.id.bytes(snapshot_id)) 292 return write.result(pair) 293 294 # Maps one pure open-response decision into a client result. 295 def snapshot.cursor.open.decision( 296 decision: OwnerCursorCodec.OpenDecision 297 ) -> Result<&1, &1, Client.Error, U32>: 298 match decision: 299 case OwnerCursorCodec.Opened{cursor_id}: Done{cursor_id} 300 case OwnerCursorCodec.OpenRejected{status}: 301 Fail{Client.OwnerRejected{status}} 302 case OwnerCursorCodec.OpenMalformed{}: 303 Fail{Client.MalformedResponse{}} 304 305 def snapshot.cursor.open.status( 306 status: Protocol.ResponseStatus, 307 payload: Bytes.Bytes 308 ) -> Result<&1, &1, Client.Error, U32>: 309 snapshot.cursor.open.decision(OwnerCursorCodec.open(status, payload)) 310 311 def snapshot.cursor.open.response( 312 result: Result<&1, &1, Client.Error, Protocol.Response> 313 ) -> Result<&1, &1, Client.Error, U32>: 314 match result: 315 case Fail{error}: Fail{error} 316 case Done{response}: 317 match response: 318 case Protocol.Response{_, _, _, status, payload}: 319 snapshot.cursor.open.status(status, payload) 320 321 def snapshot.cursor.open.result( 322 pair: Client.Session & Result<&1, &1, Client.Error, Protocol.Response> 323 ) -> Client.Session & Result<&1, &1, Client.Error, U32>: 324 match pair: 325 case (session, result): (session, snapshot.cursor.open.response(result)) 326 327 def snapshot.cursor.open.request( 328 session: Client.Session, 329 payload: Bytes.Bytes 330 ) -> IO(Client.Session & Result<&1, &1, Client.Error, U32>): 331 do IO>: 332 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 333 Client.session.call(session, 12, payload) 334 return snapshot.cursor.open.result(pair) 335 336 # Opens an owner cursor over one half-open byte range. 337 def snapshot.cursor.open.range( 338 session: Client.Session, 339 snapshot_id: U32, 340 lower: Maybe<&1, Bytes.Bytes>, 341 upper: Maybe<&1, Bytes.Bytes> 342 ) -> IO(Client.Session & Result<&1, &1, Client.Error, U32>): 343 snapshot.cursor.open.request(session, 344 CursorOpenCodec.encode.range(snapshot_id, lower, upper)) 345 346 # Opens an owner cursor over one exact raw-byte prefix. 347 def snapshot.cursor.open.prefix( 348 session: Client.Session, 349 snapshot_id: U32, 350 prefix: Bytes.Bytes 351 ) -> IO(Client.Session & Result<&1, &1, Client.Error, U32>): 352 snapshot.cursor.open.request(session, 353 CursorOpenCodec.encode.prefix(snapshot_id, prefix)) 354 355 # Opens a full-range owner cursor over a retained snapshot. 356 def snapshot.cursor.open( 357 session: Client.Session, 358 snapshot_id: U32 359 ) -> IO(Client.Session & Result<&1, &1, Client.Error, U32>): 360 snapshot.cursor.open.range(session, snapshot_id, None{}, None{}) 361 362 def snapshot.cursor.seek.decision( 363 decision: OwnerCursorCodec.CloseDecision 364 ) -> Result<&1, &1, Client.Error, Unit>: 365 match decision: 366 case OwnerCursorCodec.CloseSucceeded{}: Done{Unit{}} 367 case OwnerCursorCodec.CloseRejected{status}: 368 Fail{Client.OwnerRejected{status}} 369 case OwnerCursorCodec.CloseMalformed{}: 370 Fail{Client.MalformedResponse{}} 371 372 def snapshot.cursor.seek.status( 373 status: Protocol.ResponseStatus, 374 payload: Bytes.Bytes 375 ) -> Result<&1, &1, Client.Error, Unit>: 376 snapshot.cursor.seek.decision(OwnerCursorCodec.close(status, payload)) 377 378 def snapshot.cursor.seek.response( 379 result: Result<&1, &1, Client.Error, Protocol.Response> 380 ) -> Result<&1, &1, Client.Error, Unit>: 381 match result: 382 case Fail{error}: Fail{error} 383 case Done{Protocol.Response{_, _, _, status, payload}}: 384 snapshot.cursor.seek.status(status, payload) 385 386 def snapshot.cursor.seek.result( 387 pair: Client.Session & Result<&1, &1, Client.Error, Protocol.Response> 388 ) -> Client.Session & Result<&1, &1, Client.Error, Unit>: 389 match pair: 390 case (session, result): (session, snapshot.cursor.seek.response(result)) 391 392 def snapshot.cursor.seek( 393 session: Client.Session, 394 cursor_id: U32, 395 lower: Maybe<&1, Bytes.Bytes> 396 ) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>): 397 do IO>: 398 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 399 Client.session.call(session, 15, 400 CursorOpenCodec.encode.range(cursor_id, lower, None{})) 401 return snapshot.cursor.seek.result(pair) 402 403 # Maps one pure next-response decision into a client result. 404 def snapshot.cursor.next.decision( 405 decision: OwnerCursorCodec.NextDecision 406 ) -> Result<&1, &1, Client.Error, CursorCodec.Frame>: 407 match decision: 408 case OwnerCursorCodec.NextFrame{frame}: Done{frame} 409 case OwnerCursorCodec.NextRejected{status}: 410 Fail{Client.OwnerRejected{status}} 411 case OwnerCursorCodec.NextMalformed{}: 412 Fail{Client.MalformedResponse{}} 413 414 def snapshot.cursor.next.status( 415 status: Protocol.ResponseStatus, 416 payload: Bytes.Bytes 417 ) -> Result<&1, &1, Client.Error, CursorCodec.Frame>: 418 snapshot.cursor.next.decision(OwnerCursorCodec.next(status, payload)) 419 420 def snapshot.cursor.next.response( 421 result: Result<&1, &1, Client.Error, Protocol.Response> 422 ) -> Result<&1, &1, Client.Error, CursorCodec.Frame>: 423 match result: 424 case Fail{error}: Fail{error} 425 case Done{response}: 426 match response: 427 case Protocol.Response{_, _, _, status, payload}: 428 snapshot.cursor.next.status(status, payload) 429 430 def snapshot.cursor.next.result( 431 pair: Client.Session & Result<&1, &1, Client.Error, Protocol.Response> 432 ) -> Client.Session & Result<&1, &1, Client.Error, CursorCodec.Frame>: 433 match pair: 434 case (session, result): (session, snapshot.cursor.next.response(result)) 435 436 # Reads one raw-byte item or the terminal marker from an owner cursor. 437 def snapshot.cursor.next( 438 session: Client.Session, 439 cursor_id: U32 440 ) -> IO(Client.Session & Result<&1, &1, Client.Error, CursorCodec.Frame>): 441 do IO>: 442 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 443 Client.session.call(session, 13, Protocol.snapshot.id.bytes(cursor_id)) 444 return snapshot.cursor.next.result(pair) 445 446 # Releases one owner cursor token through the serialized session. 447 # Maps one pure close-response decision into a client result. 448 def snapshot.cursor.close.decision( 449 decision: OwnerCursorCodec.CloseDecision 450 ) -> Result<&1, &1, Client.Error, Unit>: 451 match decision: 452 case OwnerCursorCodec.CloseSucceeded{}: Done{Unit{}} 453 case OwnerCursorCodec.CloseRejected{status}: 454 Fail{Client.OwnerRejected{status}} 455 case OwnerCursorCodec.CloseMalformed{}: 456 Fail{Client.MalformedResponse{}} 457 458 # Decodes the owner close status and requires its exact empty success body. 459 def snapshot.cursor.close.status( 460 status: Protocol.ResponseStatus, 461 payload: Bytes.Bytes 462 ) -> Result<&1, &1, Client.Error, Unit>: 463 snapshot.cursor.close.decision(OwnerCursorCodec.close(status, payload)) 464 465 # Maps transport or owner response errors into one close result. 466 def snapshot.cursor.close.response( 467 result: Result<&1, &1, Client.Error, Protocol.Response> 468 ) -> Result<&1, &1, Client.Error, Unit>: 469 match result: 470 case Fail{error}: Fail{error} 471 case Done{response}: 472 match response: 473 case Protocol.Response{_, _, _, status, payload}: 474 snapshot.cursor.close.status(status, payload) 475 476 # Preserves the live client session after one close request. 477 def snapshot.cursor.close.result( 478 pair: Client.Session & Result<&1, &1, Client.Error, Protocol.Response> 479 ) -> Client.Session & Result<&1, &1, Client.Error, Unit>: 480 match pair: 481 case (session, result): (session, snapshot.cursor.close.response(result)) 482 483 def snapshot.cursor.close( 484 session: Client.Session, 485 cursor_id: U32 486 ) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>): 487 do IO>: 488 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <- 489 Client.session.call(session, 14, Protocol.snapshot.id.bytes(cursor_id)) 490 return snapshot.cursor.close.result(pair) 491 492 def snapshot.cursor.count.close.result( 493 pair: Client.Session & Result<&1, &1, Client.Error, Unit>, 494 error: Client.Error 495 ) -> Client.Session & Result<&1, &1, Client.Error, U32>: 496 match pair: 497 case (session, _): (session, Fail{error}) 498 499 def snapshot.cursor.count.close.resource( 500 session: Client.Session, 501 cursor_id: U32, 502 error: Client.Error 503 ) -> IO(Client.Session & Result<&1, &1, Client.Error, U32>): 504 do IO>: 505 closed : Client.Session & Result<&1, &1, Client.Error, Unit> <- 506 snapshot.cursor.close(session, cursor_id) 507 return snapshot.cursor.count.close.result(closed, error) 508 509 def snapshot.cursor.count.close.pair( 510 pair: Client.Session & Result<&1, &1, Client.Error, CursorCodec.Frame>, 511 cursor_id: U32, 512 error: Client.Error 513 ) -> IO(Client.Session & Result<&1, &1, Client.Error, U32>): 514 match pair: 515 case (session, _): 516 snapshot.cursor.count.close.resource(session, cursor_id, error) 517 518 def snapshot.cursor.count.close.state( 519 state: SnapshotCountState, 520 error: Client.Error 521 ) -> IO(Client.Session & Result<&1, &1, Client.Error, U32>): 522 match state: 523 case SnapshotCountNeed{session, cursor_ids, _}: 524 match cursor_ids: 525 case (cursor_id, _): 526 snapshot.cursor.count.close.resource(session, cursor_id, error) 527 case SnapshotCountFrame{pair, cursor_id, _}: 528 snapshot.cursor.count.close.pair(pair, cursor_id, error) 529 case SnapshotCountItem{session, cursor_id, _}: 530 snapshot.cursor.count.close.resource(session, cursor_id, error) 531 532 def snapshot.cursor.count.loop( 533 fuel: Nat, 534 state: SnapshotCountState 535 ) -> IO(Client.Session & Result<&1, &1, Client.Error, U32>): 536 match fuel: 537 case 0n: 538 snapshot.cursor.count.close.state(state, 539 Client.OwnerRejected{Protocol.ResponseResourceLimit{}}) 540 case 1n+rest: 541 match state: 542 case SnapshotCountFrame{pair, cursor_id, count}: 543 match pair: 544 case (session, Fail{error}): 545 snapshot.cursor.count.close.resource(session, cursor_id, error) 546 case (session, Done{CursorCodec.CursorEnd{}}): 547 IO.pure(Client.Session & Result<&1, &1, Client.Error, U32>, 548 (session, Done{count})) 549 case (session, Done{CursorCodec.CursorItem{_, _}}): 550 snapshot.cursor.count.loop(rest, SnapshotCountItem{session, 551 cursor_id, CursorCountPolicy.next(count)}) 552 case SnapshotCountItem{session, cursor_id, decision}: 553 match decision: 554 case CursorCountPolicy.CountLimit{}: 555 snapshot.cursor.count.close.resource(session, cursor_id, 556 Client.OwnerRejected{Protocol.ResponseResourceLimit{}}) 557 case CursorCountPolicy.CountContinue{next_count}: 558 snapshot.cursor.count.loop(rest, 559 SnapshotCountNeed{session, 560 CursorCountPolicy.u32.pair(cursor_id), next_count}) 561 case SnapshotCountNeed{session, cursor_ids, count}: 562 match cursor_ids: 563 case (wire_id, retained_id): 564 do IO>: 566 next : Client.Session & 567 Result<&1, &1, Client.Error, CursorCodec.Frame> <- 568 snapshot.cursor.next(session, wire_id) 569 snapshot.cursor.count.loop(rest, 570 SnapshotCountFrame{next, retained_id, count}) 571 572 # Counts one retained owner cursor without materializing its records. 573 def snapshot.cursor.count( 574 session: Client.Session, 575 cursor_id: U32 576 ) -> IO(Client.Session & Result<&1, &1, Client.Error, U32>): 577 snapshot.cursor.count.loop( 578 Nat.mul(3n, Nat.add(4294967295n, 1n)), 579 SnapshotCountNeed{session, CursorCountPolicy.u32.pair(cursor_id), 0})