src/transfer/LegacyWal.bend
mylsm-lsm-store@0.6.0.1 · 398 lines · 33 definitions
1import Base
2import mylsm-lsm-store@0.6.0.1/src/coordinator/QueuePolicy.bend as QueuePolicy
3import mylsm-lsm-store@0.6.0.1/src/Wal.bend as Wal
4import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
5
6
7type VerifyResult is Data:
8 WalValid{frames: Nat}
9 WalInvalid{}
10
11
12type ReplayResult is Data:
13 ReplayValid{batches: List<&2, Wal.Batch>}
14 ReplayInvalid{}
15
16
17type ReplayState is Type:
18 ReplayNeed{bytes: Bytes.Bytes, offset: Nat, batches: List<&2, Wal.Batch>}
19 ReplayMeasured{bytes: Bytes.Bytes, offset: Nat, batches: List<&2, Wal.Batch>,
20 size: U32}
21 ReplayBound{bytes: Bytes.Bytes, offset: Nat, batches: List<&2, Wal.Batch>,
22 total: Nat, at_end: Bool, has_length: Bool}
23 ReplayLength{bytes: Bytes.Bytes, next_offset: Nat, total: Nat,
24 batches: List<&2, Wal.Batch>, length: Maybe<&2, U32>}
25 ReplayFrame{bytes: Bytes.Bytes, next_offset: Nat, frame_end: Nat,
26 batches: List<&2, Wal.Batch>, length: U32, fits: Bool}
27 ReplayDecoded{bytes: Bytes.Bytes, next_offset: Nat,
28 batches: List<&2, Wal.Batch>, result: Result<&1, &1, Wal.Error, Wal.Batch>}
29
30
31type WalkState is Type:
32 Header{pair: Bytes.Bytes & Bytes.Bytes}
33 HeaderDecoded{source: Bytes.Bytes, result: Result<&1, &1, Wal.Error, Unit>}
34 HeaderMeasured{pair: Bytes.Bytes & U32}
35 HeaderSeek{pair: Bytes.Cursor & Bool}
36 Read{cursor: Bytes.Cursor, frames: Nat}
37 Remaining{pair: Bytes.Cursor & U32, frames: Nat}
38 Empty{cursor: Bytes.Cursor, remaining: Nat, frames: Nat}
39 EmptyChecked{cursor: Bytes.Cursor, remaining: Nat, frames: Nat, empty: Bool}
40 HasLength{cursor: Bytes.Cursor, remaining: Nat, frames: Nat, has_length: Bool}
41 Length{pair: Bytes.Cursor & Maybe<&2, U32>, remaining: U32, frames: Nat}
42 Bounds{cursor: Bytes.Cursor, length: U32, frames: Nat, valid: Bool}
43 Position{cursor: Bytes.Cursor, positions: Nat & Nat,
44 frame_bytes: U32, frame_length: U32, frames: Nat}
45 Frame{pair: Bytes.Bytes & Bytes.Bytes, end: U32, frames: Nat}
46 Decoded{result: Result<&1, &1, Wal.Error, Wal.Batch>, source: Bytes.Bytes, end: U32, frames: Nat}
47 Seek{pair: Bytes.Cursor & Bool, frames: Nat}
48 Finish{frames: Nat}
49 Invalid{}
50
51
52def walk.empty.checked(
53 cursor: Bytes.Cursor,
54 frames: Nat,
55 copies: Nat & Nat
56) -> WalkState:
57 match copies:
58 case (for_check, retained):
59 EmptyChecked{cursor, retained, frames, Nat.is_eq(for_check, 0n)}
60
61
62def walk.length.checked(
63 cursor: Bytes.Cursor,
64 frames: Nat,
65 copies: Nat & Nat
66) -> WalkState:
67 match copies:
68 case (for_check, retained):
69 HasLength{cursor, retained, frames, Nat.is_le(4n, for_check)}
70
71def walk.nat.clone.remainder(
72 first: Nat,
73 pair: Nat & Nat
74) -> Nat & Nat & Nat:
75 match pair:
76 case (second, third): (first, second, third)
77
78def walk.nat.clone.pair(pair: Nat & Nat) -> Nat & Nat & Nat:
79 match pair:
80 case (first, remainder):
81 walk.nat.clone.remainder(first,
82 QueuePolicy.nat.clone.pair(remainder))
83
84def walk.nat.clone.triple(value: Nat) -> Nat & Nat & Nat:
85 walk.nat.clone.pair(QueuePolicy.nat.clone.pair(value))
86
87def walk.length.copies.for.cursor(
88 cursor: Bytes.Cursor,
89 remaining: U32,
90 frames: Nat,
91 pair: Nat & Nat & Nat
92) -> WalkState:
93 match pair:
94 case (minimum_check, (limit_check, retained_length)):
95 +valid = Nat.is_le(36n, minimum_check) &&
96 Nat.is_le(limit_check, U32.to_nat(remaining))
97 Bounds{cursor, U32.from_nat(retained_length), frames, valid}
98
99def walk.length.validated(
100 cursor: Bytes.Cursor,
101 remaining: U32,
102 frames: Nat,
103 length: U32
104) -> WalkState:
105 walk.length.copies.for.cursor(cursor, remaining, frames,
106 walk.nat.clone.triple(U32.to_nat(length)))
107
108
109def walk.position.measured(
110 pair: Bytes.Cursor & U32,
111 frames: Nat,
112 lengths: Nat & Nat
113) -> WalkState:
114 match pair:
115 case (cursor, position):
116 match lengths:
117 case (frame_length, endpoint_length):
118 Position{cursor, QueuePolicy.nat.clone.pair(U32.to_nat(position)),
119 U32.from_nat(Nat.add(frame_length, 4n)),
120 U32.from_nat(endpoint_length), frames}
121
122def walk.position.state(
123 cursor: Bytes.Cursor,
124 frames: Nat,
125 lengths: Nat & Nat
126) -> WalkState:
127 walk.position.measured(Bytes.Cursor.position(cursor), frames, lengths)
128
129
130def walk(fuel: Nat, state: WalkState) -> VerifyResult:
131 match fuel:
132 case 0n: WalInvalid{}
133 case 1n+rest:
134 match state:
135 case Header{pair}:
136 match pair:
137 case (source, header):
138 walk(rest, HeaderDecoded{source, Wal.decode_header(header)})
139 case HeaderDecoded{source, result}:
140 match result:
141 case Fail{_}: WalInvalid{}
142 case Done{Unit{}}:
143 walk(rest, HeaderMeasured{Bytes.length(source)})
144 case HeaderMeasured{pair}:
145 match pair:
146 case (source, size):
147 walk(rest, HeaderSeek{
148 Bytes.Cursor.seek(Bytes.Cursor.new(source), 8)})
149 case HeaderSeek{pair}:
150 match pair:
151 case (_, False{}): WalInvalid{}
152 case (cursor, True{}): walk(rest, Read{cursor, 0n})
153 case Read{cursor, frames}:
154 walk(rest, Remaining{Bytes.Cursor.remaining(cursor), frames})
155 case Remaining{pair, frames}:
156 match pair:
157 case (cursor, remaining):
158 walk(rest, Empty{cursor, U32.to_nat(remaining), frames})
159 case Empty{cursor, remaining, frames}:
160 walk(rest, walk.empty.checked(cursor, frames,
161 QueuePolicy.nat.clone.pair(remaining)))
162 case EmptyChecked{cursor, remaining, frames, empty}:
163 match empty:
164 case True{}: walk(rest, Finish{frames})
165 case False{}:
166 walk(rest, walk.length.checked(cursor, frames,
167 QueuePolicy.nat.clone.pair(remaining)))
168 case HasLength{cursor, remaining, frames, has_length}:
169 match has_length:
170 case False{}: WalInvalid{}
171 case True{}:
172 walk(rest, Length{Bytes.Cursor.u32be(cursor),
173 U32.from_nat(Nat.sub(remaining, 4n)), frames})
174 case Length{pair, remaining, frames}:
175 match pair:
176 case (_, None{}): WalInvalid{}
177 case (cursor, Some{length}):
178 walk(rest, walk.length.validated(cursor, remaining, frames, length))
179 case Bounds{cursor, length, frames, valid}:
180 match valid:
181 case False{}: WalInvalid{}
182 case True{}:
183 walk(rest, walk.position.state(cursor, frames,
184 QueuePolicy.nat.clone.pair(U32.to_nat(length))))
185 case Position{cursor, positions, frame_bytes, frame_length, frames}:
186 match cursor:
187 case Bytes.Cursor{bytes, _, _, _}:
188 match positions:
189 case (start_nat, end_nat):
190 +start = U32.from_nat(Nat.sub(start_nat, 4n))
191 +end = U32.add(U32.from_nat(end_nat), frame_length)
192 walk(rest, Frame{Bytes.slice(bytes, start, frame_bytes),
193 end, frames})
194 case Frame{pair, end, frames}:
195 match pair:
196 case (source, frame):
197 walk(rest, Decoded{Wal.decode_frame(frame), source, end, frames})
198 case Decoded{result, source, end, frames}:
199 match result:
200 case Fail{_}: WalInvalid{}
201 case Done{_}:
202 walk(rest, Seek{Bytes.Cursor.seek(Bytes.Cursor.new(source), end),
203 Nat.add(frames, 1n)})
204 case Seek{pair, frames}:
205 match pair:
206 case (_, False{}): WalInvalid{}
207 case (cursor, True{}): walk(rest, Read{cursor, frames})
208 case Finish{frames}: WalValid{frames}
209 case Invalid{}: WalInvalid{}
210
211
212def verify.measured(pair: Bytes.Bytes & U32) -> VerifyResult:
213 match pair:
214 case (bytes, size):
215 walk(Nat.add(5n,
216 Nat.mul(12n, Nat.add(Nat.div(U32.to_nat(size), 40n), 1n))),
217 Header{Bytes.slice(bytes, 0, 8)})
218
219
220def verify(bytes: Bytes.Bytes) -> VerifyResult:
221 verify.measured(Bytes.length(bytes))
222
223def replay.measure(
224 measured: Bytes.Bytes & U32,
225 offset: Nat,
226 batches: List<&2, Wal.Batch>
227) -> ReplayState:
228 match measured:
229 case (bytes, size): ReplayMeasured{bytes, offset, batches, size}
230
231def replay.nat.clone.remainder(first: Nat, pair: Nat & Nat) -> Nat & Nat & Nat:
232 match pair:
233 case (second, third): (first, second, third)
234
235def replay.nat.clone.pair(pair: Nat & Nat) -> Nat & Nat & Nat:
236 match pair:
237 case (first, remainder):
238 replay.nat.clone.remainder(first, QueuePolicy.nat.clone.pair(remainder))
239
240def replay.nat.clone.triple(value: Nat) -> Nat & Nat & Nat:
241 replay.nat.clone.pair(QueuePolicy.nat.clone.pair(value))
242
243def replay.bound.measured(
244 bytes: Bytes.Bytes,
245 batches: List<&2, Wal.Batch>,
246 offsets: Nat & Nat & Nat,
247 totals: Nat & Nat & Nat
248) -> ReplayState:
249 match offsets:
250 case (retained_offset, (end_offset, length_offset)):
251 match totals:
252 case (retained_total, (end_total, length_total)):
253 ReplayBound{bytes, retained_offset, batches, retained_total,
254 Nat.is_eq(end_offset, end_total),
255 Nat.is_le(Nat.add(length_offset, 4n), length_total)}
256
257def replay.bound.state(
258 bytes: Bytes.Bytes,
259 offset: Nat,
260 batches: List<&2, Wal.Batch>,
261 total: Nat
262) -> ReplayState:
263 replay.bound.measured(bytes, batches,
264 replay.nat.clone.triple(offset), replay.nat.clone.triple(total))
265
266def replay.length.pair(
267 measured: Bytes.Bytes & Maybe<&2, U32>,
268 next_offset: Nat,
269 total: Nat,
270 batches: List<&2, Wal.Batch>
271) -> ReplayState:
272 match measured:
273 case (bytes, length):
274 ReplayLength{bytes, next_offset, total, batches, length}
275
276def replay.length.indexed(
277 bytes: Bytes.Bytes,
278 offsets: Nat & Nat,
279 total: Nat,
280 batches: List<&2, Wal.Batch>
281) -> ReplayState:
282 match offsets:
283 case (next_offset, read_offset):
284 replay.length.pair(Bytes.get.u32be(bytes, U32.from_nat(read_offset)),
285 Nat.add(next_offset, 4n), total, batches)
286
287def replay.length.state(
288 bytes: Bytes.Bytes,
289 offset: Nat,
290 total: Nat,
291 batches: List<&2, Wal.Batch>
292) -> ReplayState:
293 replay.length.indexed(bytes, QueuePolicy.nat.clone.pair(offset),
294 total, batches)
295
296def replay.frame.measured(
297 bytes: Bytes.Bytes,
298 batches: List<&2, Wal.Batch>,
299 total: Nat,
300 offsets: Nat & Nat,
301 lengths: Nat & Nat
302) -> ReplayState:
303 match offsets:
304 case (slice_offset, end_offset):
305 match lengths:
306 case (end_length, slice_length):
307 +frame_end = Nat.add(end_offset, end_length)
308 +frame_start = Nat.sub(slice_offset, 4n)
309 +serialized_length = Nat.add(slice_length, 4n)
310 ReplayFrame{bytes, frame_start, frame_end, batches,
311 U32.from_nat(serialized_length), Nat.is_le(frame_end, total)}
312
313def replay.frame.state(
314 bytes: Bytes.Bytes,
315 next_offset: Nat,
316 total: Nat,
317 batches: List<&2, Wal.Batch>,
318 length: U32
319) -> ReplayState:
320 replay.frame.measured(bytes, batches, total,
321 QueuePolicy.nat.clone.pair(next_offset),
322 QueuePolicy.nat.clone.pair(U32.to_nat(length)))
323
324def replay.frame.decoded(
325 pair: Bytes.Bytes & Bytes.Bytes,
326 frame_end: Nat,
327 batches: List<&2, Wal.Batch>
328) -> ReplayState:
329 match pair:
330 case (bytes, frame):
331 ReplayDecoded{bytes, frame_end, batches, Wal.decode_frame(frame)}
332
333def replay.loop(fuel: Nat, state: ReplayState) -> ReplayResult:
334 match fuel:
335 case 0n: ReplayInvalid{}
336 case 1n+rest:
337 match state:
338 case ReplayNeed{bytes, offset, batches}:
339 replay.loop(rest, replay.measure(Bytes.length(bytes), offset, batches))
340 case ReplayMeasured{bytes, offset, batches, size}:
341 +total = U32.to_nat(size)
342 replay.loop(rest, replay.bound.state(bytes, offset, batches, total))
343 case ReplayBound{bytes, offset, batches, total, at_end, has_length}:
344 match at_end:
345 case True{}:
346 ReplayValid{batches}
347 case False{}:
348 match has_length:
349 case True{}:
350 replay.loop(rest,
351 replay.length.state(bytes, offset, total, batches))
352 case False{}: ReplayInvalid{}
353 case ReplayLength{bytes, next_offset, total, batches, length}:
354 match length:
355 case None{}: ReplayInvalid{}
356 case Some{frame_length}:
357 replay.loop(rest, replay.frame.state(bytes, next_offset,
358 total, batches, frame_length))
359 case ReplayFrame{bytes, next_offset, frame_end, batches, length, fits}:
360 match fits:
361 case True{}:
362 replay.loop(rest, replay.frame.decoded(Bytes.slice(bytes,
363 U32.from_nat(next_offset), length), frame_end, batches))
364 case False{}: ReplayInvalid{}
365 case ReplayDecoded{bytes, next_offset, batches, result}:
366 match result:
367 case Fail{_}: ReplayInvalid{}
368 case Done{batch}:
369 replay.loop(rest, ReplayNeed{bytes, next_offset,
370 List.append(&2, Wal.Batch, batches, Con{batch, Nil{}})})
371
372def replay.header.validated(
373 source: Bytes.Bytes,
374 fuel: Nat,
375 validation: Result<&1, &1, Wal.Error, Unit>
376) -> ReplayResult:
377 match validation:
378 case Fail{_}: ReplayInvalid{}
379 case Done{Unit{}}: replay.loop(fuel, ReplayNeed{source, 8n, Nil{}})
380
381def replay.header.pair(
382 pair: Bytes.Bytes & Bytes.Bytes,
383 fuel: Nat
384) -> ReplayResult:
385 match pair:
386 case (source, header):
387 replay.header.validated(source, fuel, Wal.decode_header(header))
388
389
390def replay.measured(pair: Bytes.Bytes & U32) -> ReplayResult:
391 match pair:
392 case (bytes, size):
393 replay.header.pair(Bytes.slice(bytes, 0, 8),
394 Nat.add(3n, Nat.mul(6n, Nat.div(U32.to_nat(size), 40n))))
395
396
397def replay(bytes: Bytes.Bytes) -> ReplayResult:
398 replay.measured(Bytes.length(bytes))