src/coordinator/ByteClient.bend
mylsm-lsm-store@0.6.0.0 · 579 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/coordinator/Client.bend as Client
3import mylsm-lsm-store@0.6.0.0/src/coordinator/ClientPolicy.bend as Policy
4import mylsm-lsm-store@0.6.0.0/src/coordinator/Protocol.bend as Protocol
5import mylsm-lsm-store@0.6.0.0/src/coordinator/CursorCodec.bend as CursorCodec
6import mylsm-lsm-store@0.6.0.0/src/coordinator/OwnerCursorCodec.bend as OwnerCursorCodec
7import mylsm-lsm-store@0.6.0.0/src/coordinator/CursorCountPolicy.bend as CursorCountPolicy
8import mylsm-lsm-store@0.6.0.0/src/coordinator/CursorOpenCodec.bend as CursorOpenCodec
9import mylsm-lsm-store@0.6.0.0/src/coordinator/StatsCodec.bend as StatsCodec
10import mylsm-lsm-store@0.6.0.0/src/coordinator/OptionsCodec.bend as OptionsCodec
11import mylsm-lsm-store@0.6.0.0/src/ByteWal.bend as ByteWal
12import mylsm-lsm-store@0.6.0.0/src/Wal.bend as Wal
13import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
14
15# Carries a bounded remote cursor count across incremental owner reads.
16type 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
25def 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.
36def describe_owner(
37 session: Client.Session
38) -> IO(Client.Session & Result<&1, &1, Client.Error, OptionsCodec.Decision>):
39 do IO<Client.Session & Result<&1, &1, Client.Error, OptionsCodec.Decision>>:
40 pair : Client.Session &
41 Result<&1, &1, Client.Error, Protocol.Response> <-
42 Client.session.describe_owner(session)
43 return describe.result(pair)
44def result.failure(
45 error: Client.Error
46) -> Result<&1, &1, Client.Error, Maybe<&1, Bytes.Bytes>>:
47 Fail{error}
48
49def result.value(
50 value: Maybe<&1, Bytes.Bytes>
51) -> Result<&1, &1, Client.Error, Maybe<&1, Bytes.Bytes>>:
52 Done{value}
53
54def 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
64def 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
70def 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
80def 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.
87def 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<Client.Session & Result<&1, &1, Client.Error, Maybe<&1, Bytes.Bytes>>>:
92 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <-
93 Client.session.get(session, key)
94 return get.result(pair)
95
96def 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
103def 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
113def 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
123def 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.
130def 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<Client.Session & Result<&1, &1, Client.Error, Unit>>:
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.
141def delete(
142 session: Client.Session,
143 key: Bytes.Bytes
144) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>):
145 do IO<Client.Session & Result<&1, &1, Client.Error, Unit>>:
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.
151def flush(
152 session: Client.Session
153) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>):
154 do IO<Client.Session & Result<&1, &1, Client.Error, Unit>>:
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.
160def compact(
161 session: Client.Session
162) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>):
163 do IO<Client.Session & Result<&1, &1, Client.Error, Unit>>:
164 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <-
165 Client.session.compact(session)
166 return write.result(pair)
167
168def 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<Client.Session & Result<&1, &1, Client.Error, Unit>>:
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.
183def 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
189def 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
196def 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
204def 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
211def 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.
219def stats(
220 session: Client.Session
221) -> IO(Client.Session & Result<&1, &1, Client.Error, StatsCodec.Stats>):
222 do IO<Client.Session & Result<&1, &1, Client.Error, StatsCodec.Stats>>:
223 pair : Client.Session & Result<&1, &1, Client.Error, Protocol.Response> <-
224 Client.session.stats(session)
225 return stats.result(pair)
226
227def 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
234def 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
243def 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
253def 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.
260def snapshot.open(
261 session: Client.Session
262) -> IO(Client.Session & Result<&1, &1, Client.Error, U32>):
263 do IO<Client.Session & Result<&1, &1, Client.Error, U32>>:
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.
269def 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.
273def 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<Client.Session & Result<&1, &1, Client.Error,
279 Maybe<&1, Bytes.Bytes>>>:
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.
285def snapshot.close(
286 session: Client.Session,
287 snapshot_id: U32
288) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>):
289 do IO<Client.Session & Result<&1, &1, Client.Error, Unit>>:
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.
295def 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
305def 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
311def 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
321def 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
327def 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<Client.Session & Result<&1, &1, Client.Error, U32>>:
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.
337def 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.
347def 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.
356def 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
362def 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
372def 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
378def 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
386def 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
392def 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<Client.Session & Result<&1, &1, Client.Error, Unit>>:
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.
404def 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
414def 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
420def 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
430def 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.
437def 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<Client.Session & Result<&1, &1, Client.Error, CursorCodec.Frame>>:
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.
448def 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.
459def 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.
466def 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.
477def 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
483def snapshot.cursor.close(
484 session: Client.Session,
485 cursor_id: U32
486) -> IO(Client.Session & Result<&1, &1, Client.Error, Unit>):
487 do IO<Client.Session & Result<&1, &1, Client.Error, Unit>>:
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
492def 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
499def 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<Client.Session & Result<&1, &1, Client.Error, U32>>:
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
509def 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
518def 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
532def 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<Client.Session &
565 Result<&1, &1, Client.Error, U32>>:
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.
573def 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})