src/transfer/BackupCursorPolicy.bend
mylsm-lsm-store@0.6.0.0 · 76 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/coordinator/QueuePolicy.bend as QueuePolicy
3import mylsm-lsm-store@0.6.0.0/src/coordinator/CursorCodec.bend as CursorCodec
4import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
5
6# Reports whether one item is buffered, completes a block, or exceeds bounds.
7type AppendDecision is Data:
8 CursorBuffer{seen: Nat, pending: Nat}
9 CursorBlockReady{seen: Nat, pending: Nat}
10 CursorRejected{}
11
12# Reports footer readiness, a final partial block, or an invalid end count.
13type FinishDecision is Data:
14 CursorFooter{}
15 CursorFinalBlock{pending: Nat}
16 CursorFinishRejected{}
17
18# Carries an owner end marker or one unchanged raw-byte record.
19type OwnerFrameDecision is Type:
20 OwnerCursorEnd{}
21 OwnerCursorItem{key: Bytes.Bytes, value: Bytes.Bytes}
22
23def block.index.next(index: Nat) -> Nat:
24 Nat.add(index, 1n)
25
26# Preserves raw bytes while classifying one owner cursor response.
27def owner.frame(frame: CursorCodec.Frame) -> OwnerFrameDecision:
28 match frame:
29 case CursorCodec.CursorEnd{}: OwnerCursorEnd{}
30 case CursorCodec.CursorItem{key, value}: OwnerCursorItem{key, value}
31
32# Selects buffering, a full block, or rejection after the count check.
33def append.capacity(seen: Nat, pending: Nat) -> AppendDecision:
34 match pending:
35 case 0n: CursorBuffer{Nat.add(seen, 1n), 1n}
36 case 1n+rest:
37 match rest:
38 case 14n: CursorBlockReady{Nat.add(seen, 1n), 16n}
39 case 15n+more: CursorRejected{}
40 case _: CursorBuffer{Nat.add(seen, 1n), Nat.add(pending, 1n)}
41
42def append.checked(
43 seen: Nat,
44 pending: Nat,
45 within_expected_count: Bool
46) -> AppendDecision:
47 match within_expected_count:
48 case True{}: append.capacity(seen, pending)
49 case False{}: CursorRejected{}
50
51def append.copies(
52 seen: Nat & Nat,
53 pending: Nat,
54 expected: Nat
55) -> AppendDecision:
56 match seen:
57 case (output, checked):
58 append.checked(output, pending, Nat.is_lt(checked, expected))
59
60# Adds one cursor item without exceeding the declared total or block capacity.
61def append(seen: Nat, pending: Nat, expected: Nat) -> AppendDecision:
62 append.copies(QueuePolicy.nat.clone.pair(seen), pending, expected)
63
64def finish.buffer(pending: Nat) -> FinishDecision:
65 match pending:
66 case 0n: CursorFooter{}
67 case 1n+rest: CursorFinalBlock{pending}
68
69def finish.pending(pending: Nat, count_matches: Bool) -> FinishDecision:
70 match count_matches:
71 case False{}: CursorFinishRejected{}
72 case True{}: finish.buffer(pending)
73
74# Emits a partial final block only after the cursor returned its exact count.
75def finish(seen: Nat, expected: Nat, pending: Nat) -> FinishDecision:
76 finish.pending(pending, Nat.is_eq(seen, expected))