src/transfer/BackupCursor.bend

mylsm-lsm-store@0.6.0.1 · 345 lines · 27 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.1/src/scan/VisibleCursor.bend as VisibleCursor
4import mylsm-lsm-store@0.6.0.1/src/transfer/Backup.bend as Backup
5import mylsm-lsm-store@0.6.0.1/src/transfer/BackupCursorPolicy.bend as Policy
6import mylsm-lsm-store@0.6.0.1/src/coordinator/CursorCodec.bend as CursorCodec
7import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
8
9# Separates bounded record emission from footer emission and completion.
10type Phase is Data:
11  CursorBlocks{}
12  CursorFooter{}
13  CursorFinished{}
14
15# Tracks one visible cursor and no more than one pending export block.
16type Stream is Type:
17  Stream{phase: Phase, cursor: Maybe<&1, VisibleCursor.Cursor>,
18    pending: List<&1, ByteMemTable.Entry>, pending_count: Nat,
19    seen: Nat, expected: Nat, entry_count: U32, block_count: U32,
20    block_index: U32}
21
22# Returns a bounded backup chunk, terminal completion, or a validation error.
23type Step is Type:
24  Chunk{stream: Stream, bytes: Bytes.Bytes}
25  Continue{stream: Stream}
26  End{}
27  Failure{cursor: Maybe<&1, VisibleCursor.Cursor>, error: Backup.Error}
28
29# Initializes the bounded cursor stream after encoding its header.
30def start.ready(
31  cursor: Maybe<&1, VisibleCursor.Cursor>,
32  entry_count: U32,
33  expected: Nat,
34  block_count: U32,
35  header: Bytes.Bytes
36) -> Step:
37  Chunk{Stream{CursorBlocks{}, cursor, Nil{}, 0n, 0n, expected,
38    entry_count, block_count, 0}, header}
39
40def start.count(
41  cursor: Maybe<&1, VisibleCursor.Cursor>,
42  entries: U32 & U32,
43  block_count: U32,
44  header: Bytes.Bytes
45) -> Step:
46  match entries:
47    case (entry_count, expected):
48      start.ready(cursor, entry_count, U32.to_nat(expected),
49        block_count, header)
50
51def start.header(
52  cursor: Maybe<&1, VisibleCursor.Cursor>,
53  pair: Backup.Stream & Bytes.Bytes
54) -> Step:
55  match pair:
56    case (Backup.Stream{_, _, entry_count, block_count, _}, header):
57      start.count(cursor, Backup.u32.duplicate.pair(entry_count),
58        block_count, header)
59
60def start.counts(
61  cursor: Maybe<&1, VisibleCursor.Cursor>,
62  counts: Nat & Nat
63) -> Step:
64  match counts:
65    case (entry_count, block_input):
66      +encoded_entries = U32.from_nat(entry_count)
67      +encoded_blocks = Backup.blocks.count(block_input)
68      start.header(cursor, Backup.header.bytes(
69        Backup.Stream{Backup.Header{}, Nil{}, encoded_entries,
70          encoded_blocks, 0}))
71
72# Emits the format header and prepares the first incremental source read.
73def start(cursor: VisibleCursor.Cursor, entry_count: Nat) -> Step:
74  start.counts(Some{cursor}, Backup.nat.duplicate.pair(entry_count))
75
76# Initializes the same bounded encoder without a local cursor handle.
77def start.owner(entry_count: Nat) -> Step:
78  start.counts(None{}, Backup.nat.duplicate.pair(entry_count))
79
80def block.written(
81  result: Result<&1, &1, Backup.Error, Bytes.Bytes>,
82  cursor: Maybe<&1, VisibleCursor.Cursor>,
83  seen: Nat,
84  expected: Nat,
85  entry_count: U32,
86  block_count: U32,
87  next_index: U32,
88  next_phase: Phase
89) -> Step:
90  match result:
91    case Fail{error}: Failure{cursor, error}
92    case Done{bytes}:
93      Chunk{Stream{next_phase, cursor, Nil{}, 0n, seen,
94        expected, entry_count, block_count, next_index}, bytes}
95
96def block.index.pair(index: U32) -> U32 & U32:
97  +current = U32.to_nat(index)
98  (U32.from_nat(current), U32.from_nat(Policy.block.index.next(current)))
99
100def block.encoded(
101  cursor: Maybe<&1, VisibleCursor.Cursor>,
102  pending: List<&1, ByteMemTable.Entry>,
103  seen: Nat,
104  expected: Nat,
105  entry_count: U32,
106  block_count: U32,
107  indexes: U32 & U32,
108  count: Nat,
109  next_phase: Phase
110) -> Step:
111  match indexes:
112    case (encode_index, next_index):
113      block.written(Backup.block.encoded(
114        List.reverse(&1, ByteMemTable.Entry, pending), encode_index,
115        U32.from_nat(count)), cursor, seen, expected,
116        entry_count, block_count, next_index, next_phase)
117
118def item.full(
119  cursor: Maybe<&1, VisibleCursor.Cursor>,
120  pending: List<&1, ByteMemTable.Entry>,
121  seen: Nat,
122  expected: Nat,
123  entry_count: U32,
124  block_count: U32,
125  block_index: U32,
126  count: Nat,
127  next_phase: Phase
128) -> Step:
129  block.encoded(cursor, pending, seen, expected, entry_count, block_count,
130    block.index.pair(block_index), count, next_phase)
131
132def item.decided(
133  decision: Policy.AppendDecision,
134  cursor: Maybe<&1, VisibleCursor.Cursor>,
135  pending: List<&1, ByteMemTable.Entry>,
136  expected: Nat,
137  entry_count: U32,
138  block_count: U32,
139  block_index: U32
140) -> Step:
141  match decision:
142    case Policy.CursorRejected{}:
143      Failure{cursor, Backup.ResourceLimit{}}
144    case Policy.CursorBuffer{seen, pending_count}:
145      Continue{Stream{CursorBlocks{}, cursor, pending, pending_count,
146        seen, expected, entry_count, block_count, block_index}}
147    case Policy.CursorBlockReady{seen, pending_count}:
148      item.full(cursor, pending, seen, expected, entry_count, block_count,
149        block_index, pending_count, CursorBlocks{})
150
151def item.counted(
152  entry: ByteMemTable.Entry,
153  cursor: Maybe<&1, VisibleCursor.Cursor>,
154  pending: List<&1, ByteMemTable.Entry>,
155  seen: Nat,
156  pending_count: Nat,
157  entry_count: U32,
158  block_count: U32,
159  block_index: U32,
160  expected_copies: Nat & Nat
161) -> Step:
162  match expected_copies:
163    case (for_policy, retained):
164      item.decided(Policy.append(seen, pending_count, for_policy),
165        cursor, Con{entry, pending}, retained, entry_count,
166        block_count, block_index)
167
168# Adds one cursor item without exceeding the declared total or block capacity.
169def item(
170  entry: ByteMemTable.Entry,
171  cursor: Maybe<&1, VisibleCursor.Cursor>,
172  pending: List<&1, ByteMemTable.Entry>,
173  seen: Nat,
174  pending_count: Nat,
175  expected: Nat,
176  entry_count: U32,
177  block_count: U32,
178  block_index: U32
179) -> Step:
180  item.counted(entry, cursor, pending, seen, pending_count,
181    entry_count, block_count, block_index,
182    Backup.nat.duplicate.pair(expected))
183
184# Retains the pending block when the cursor advances over a tombstone.
185def skip(
186  cursor: Maybe<&1, VisibleCursor.Cursor>,
187  pending: List<&1, ByteMemTable.Entry>,
188  pending_count: Nat,
189  seen: Nat,
190  expected: Nat,
191  entry_count: U32,
192  block_count: U32,
193  block_index: U32
194) -> Step:
195  Continue{Stream{CursorBlocks{}, cursor, pending, pending_count, seen,
196    expected, entry_count, block_count, block_index}}
197
198def footer.ready(
199  cursor: Maybe<&1, VisibleCursor.Cursor>,
200  seen: Nat,
201  expected: Nat,
202  entry_count: U32,
203  block_count: U32,
204  block_index: U32,
205  encoded: Backup.Stream & Bytes.Bytes
206) -> Step:
207  match encoded:
208    case (_, bytes):
209      Chunk{Stream{CursorFinished{}, cursor, Nil{}, 0n, seen,
210        expected, entry_count, block_count, block_index}, bytes}
211
212def footer.counts(
213  cursor: Maybe<&1, VisibleCursor.Cursor>,
214  seen: Nat,
215  expected: Nat,
216  entries: U32 & U32,
217  blocks: U32 & U32,
218  indexes: U32 & U32
219) -> Step:
220  match entries:
221    case (encoded_entries, retained_entries):
222      match blocks:
223        case (encoded_blocks, retained_blocks):
224          match indexes:
225            case (encoded_index, retained_index):
226              footer.ready(cursor, seen, expected,
227                retained_entries, retained_blocks, retained_index,
228                Backup.footer.bytes(Backup.Stream{Backup.Footer{}, Nil{},
229                  encoded_entries, encoded_blocks, encoded_index}))
230
231# Emits the checksum footer while retaining only terminal cursor state.
232def footer(
233  cursor: Maybe<&1, VisibleCursor.Cursor>,
234  seen: Nat,
235  expected: Nat,
236  entry_count: U32,
237  block_count: U32,
238  block_index: U32
239) -> Step:
240  footer.counts(cursor, seen, expected,
241    Backup.u32.duplicate.pair(entry_count),
242    Backup.u32.duplicate.pair(block_count),
243    Backup.u32.duplicate.pair(block_index))
244
245def finish.decided(
246  decision: Policy.FinishDecision,
247  cursor: Maybe<&1, VisibleCursor.Cursor>,
248  pending: List<&1, ByteMemTable.Entry>,
249  seen: Nat,
250  expected: Nat,
251  entry_count: U32,
252  block_count: U32,
253  block_index: U32
254) -> Step:
255  match decision:
256    case Policy.CursorFinishRejected{}: Failure{cursor, Backup.ResourceLimit{}}
257    case Policy.CursorFooter{}:
258      footer(cursor, seen, expected, entry_count, block_count, block_index)
259    case Policy.CursorFinalBlock{pending_count}:
260      block.encoded(cursor, pending, seen, expected, entry_count,
261        block_count, block.index.pair(block_index), pending_count,
262        CursorFooter{})
263
264def finish.copied(
265  cursor: Maybe<&1, VisibleCursor.Cursor>,
266  pending: List<&1, ByteMemTable.Entry>,
267  pending_count: Nat,
268  seen_copies: Nat & Nat,
269  expected_copies: Nat & Nat,
270  entry_count: U32,
271  block_count: U32,
272  block_index: U32
273) -> Step:
274  match seen_copies:
275    case (seen_check, seen_retained):
276      match expected_copies:
277        case (expected_check, expected_retained):
278          finish.decided(Policy.finish(seen_check, expected_check, pending_count),
279            cursor, pending, seen_retained, expected_retained, entry_count,
280            block_count, block_index)
281
282# Rejects early cursor end; emits a final partial block before the footer.
283def finish(stream: Stream) -> Step:
284  match stream:
285    case Stream{CursorBlocks{}, cursor, pending, pending_count, seen, expected,
286        entry_count, block_count, block_index}:
287      finish.copied(cursor, pending, pending_count,
288        Backup.nat.duplicate.pair(seen),
289        Backup.nat.duplicate.pair(expected),
290        entry_count, block_count, block_index)
291    case Stream{_, cursor, _, _, _, _, _, _, _}: Failure{cursor, Backup.CodecFailure{}}
292
293def advance.active(
294  stream: Stream,
295  source_step: VisibleCursor.Step
296) -> Step:
297  match stream:
298    case Stream{CursorBlocks{}, cursor, pending, pending_count, seen, expected,
299        entry_count, block_count, block_index}:
300      match source_step:
301        case VisibleCursor.Item{entry, cursor}:
302          item(entry, Some{cursor}, pending, seen, pending_count, expected,
303            entry_count, block_count, block_index)
304        case VisibleCursor.Skip{cursor}:
305          skip(Some{cursor}, pending, pending_count, seen, expected,
306            entry_count, block_count, block_index)
307        case VisibleCursor.End{}:
308          finish(Stream{CursorBlocks{}, None{}, pending, pending_count, seen,
309            expected, entry_count, block_count, block_index})
310        case VisibleCursor.Error{error}:
311          match error:
312            case (code, message):
313              Failure{None{}, Backup.SourceFailure{code, message}}
314    case Stream{CursorFooter{}, cursor, pending, pending_count, seen,
315        expected, entry_count, block_count, block_index}:
316      footer(cursor, seen, expected, entry_count, block_count, block_index)
317    case Stream{CursorFinished{}, _, _, _, _, _, _, _, _}: End{}
318
319# Applies one result returned by the bounded visible cursor.
320def advance(stream: Stream, source_step: VisibleCursor.Step) -> Step:
321  advance.active(stream, source_step)
322
323# Applies one frame returned by a retained owner cursor.
324def advance.owner.decision(
325  stream: Stream,
326  decision: Policy.OwnerFrameDecision
327) -> Step:
328  match stream:
329    case Stream{CursorBlocks{}, _, pending, pending_count, seen, expected,
330        entry_count, block_count, block_index}:
331      match decision:
332        case Policy.OwnerCursorItem{key, value}:
333          item(ByteMemTable.Entry{key, Some{value}}, None{}, pending,
334            seen, pending_count, expected, entry_count, block_count,
335            block_index)
336        case Policy.OwnerCursorEnd{}:
337          finish(Stream{CursorBlocks{}, None{}, pending, pending_count,
338            seen, expected, entry_count, block_count, block_index})
339    case Stream{CursorFooter{}, cursor, pending, pending_count, seen,
340        expected, entry_count, block_count, block_index}:
341      footer(cursor, seen, expected, entry_count, block_count, block_index)
342    case Stream{CursorFinished{}, _, _, _, _, _, _, _, _}: End{}
343
344def advance.owner(stream: Stream, frame: CursorCodec.Frame) -> Step:
345  advance.owner.decision(stream, Policy.owner.frame(frame))