src/transfer/Restore.bend
mylsm-lsm-store@0.6.0.0 · 938 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.0/src/ByteWal.bend as ByteWal
4import mylsm-lsm-store@0.6.0.0/src/transfer/RestorePolicy.bend as RestorePolicy
5import mylsm-lsm-store@0.6.0.0/src/StorageBytes.bend as StorageBytes
6import mylsm-lsm-store@0.6.0.0/src/Wal.bend as Wal
7import mylsm-lsm-store@0.6.0.0/src/coordinator/QueuePolicy.bend as QueuePolicy
8import mylsm-lsm-store@0.6.0.0/src/hub_sha/sha256.bend as SHA
9import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
10
11# Distinguishes malformed exchange data from integrity and resource failures.
12type Error is Data:
13 InvalidFormat{}
14 ResourceLimit{}
15 InvalidOrder{}
16 ChecksumMismatch{}
17 CountMismatch{}
18
19# Carries the verified logical records produced by restore validation.
20type Decoded is Type:
21 Decoded{entries: List<&1, ByteMemTable.Entry>}
22
23# Carries the fixed header metadata shared by all restore paths.
24type Header is Type:
25 Header{input: Bytes.Bytes, entry_count: U32, block_count: U32}
26
27# Tracks block traversal and the reverse-ordered decoded entries.
28type State is Type:
29 State{input: Bytes.Bytes & Bytes.Bytes, offset: Nat, entry_count: U32,
30 block_count: U32, remaining_blocks: Nat, block_index: Nat,
31 decoded_count: Nat, reversed: List<&1, ByteMemTable.Entry>}
32
33# Retains the source slice and decoded entries for one validated block.
34type BlockMeta is Type:
35 BlockMeta{input: Bytes.Bytes, offset: Nat, frame_offset: Nat,
36 frame_length: Nat, record_count: Nat, block_length: Nat,
37 entries: List<&1, ByteMemTable.Entry>}
38
39# Holds validated block-header fields while checksum and frame decoding run.
40type BlockHeader is Type:
41 BlockHeader{input: Bytes.Bytes, offset: Nat, index: U32,
42 record_count: U32, frame_length: U32, block_length: Nat}
43
44# Pairs decoded entries with their observed count.
45type EntryBatch is Type:
46 EntryBatch{entries: List<&1, ByteMemTable.Entry>, count: Nat}
47
48# Projects byte equality from the packed-byte comparison result.
49def equal(pair: Bytes.Bytes & Bytes.Bytes & Bool) -> Bool:
50 match pair:
51 case (_, _, same): same
52
53def slice.result(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
54 match pair:
55 case (_, bytes): bytes
56
57def u32.field.is(result: Maybe<&2, U32>, expected: U32) -> Bool:
58 match result:
59 case None{}: False{}
60 case Some{field}: U32.is_eq(field, expected)
61
62def clone.pair(+len: U32, copied: Array<U32> & Array<U32>) -> Bytes.Bytes & Bytes.Bytes:
63 match copied:
64 case (left, right):
65 (Bytes.Bytes{len, left}, Bytes.Bytes{len, right})
66
67def clone.split(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes:
68 match bytes:
69 case Bytes.Bytes{+len, buf}:
70 clone.pair(len, Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
71
72def clone.triple.left(
73 second: Bytes.Bytes,
74 copies: Bytes.Bytes & Bytes.Bytes
75) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
76 match copies:
77 case (left, right): (left, right, second)
78
79def clone.triple.step(copies: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
80 match copies:
81 case (first, second):
82 clone.triple.left(second, clone.split(first))
83
84def clone.triple(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
85 clone.triple.step(clone.split(bytes))
86
87def bytes.size(pair: Bytes.Bytes & U32) -> U32:
88 match pair:
89 case (_, size): size
90
91# Unwraps a fixed exchange-format marker.
92def magic(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
93 match value:
94 case Some{bytes}: bytes
95 case None{}: Bytes.new(0)
96
97def magic.header() -> Bytes.Bytes:
98 magic(Bytes.from_hex("4d594c534d424b31"))
99
100def footer.magic() -> Bytes.Bytes:
101 magic(Bytes.from_hex("454e4431"))
102
103def block.magic() -> Bytes.Bytes:
104 magic(Bytes.from_hex("424c4b31"))
105
106def digest.matches(
107 claimed: Bytes.Bytes,
108 input: Bytes.Bytes
109) -> Bool:
110 equal(Bytes.eq(claimed, SHA.sha256_packed_bytes(input)))
111
112def header.ready.block(pair: Bytes.Bytes & Maybe<&2, U32>, entry_count: U32) -> Result<&1, &1, Error, Header>:
113 match pair:
114 case (input, value):
115 match value:
116 case None{}: Fail{InvalidFormat{}}
117 case Some{block_count}: Done{Header{input, entry_count, block_count}}
118
119def header.entry_count.read(pair: Bytes.Bytes & Maybe<&2, U32>) -> Result<&1, &1, Error, Header>:
120 match pair:
121 case (input, value):
122 match value:
123 case None{}: Fail{InvalidFormat{}}
124 case Some{entry_count}: header.ready.block(Bytes.get.u32be(input, 16), entry_count)
125
126def header.block_count.checked(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, Header>:
127 match valid:
128 case False{}: Fail{ResourceLimit{}}
129 case True{}: header.entry_count.read(Bytes.get.u32be(input, 12))
130
131def header.block_count(pair: Bytes.Bytes & Maybe<&2, U32>) -> Result<&1, &1, Error, Header>:
132 match pair:
133 case (input, value):
134 match value:
135 case None{}: Fail{InvalidFormat{}}
136 case Some{block_count}:
137 header.block_count.checked(input, U32.is_le(block_count, 65536))
138
139def header.entry_count.checked(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, Header>:
140 match valid:
141 case False{}: Fail{ResourceLimit{}}
142 case True{}: header.block_count(Bytes.get.u32be(input, 16))
143
144def header.entry_count(pair: Bytes.Bytes & Maybe<&2, U32>) -> Result<&1, &1, Error, Header>:
145 match pair:
146 case (input, value):
147 match value:
148 case None{}: Fail{InvalidFormat{}}
149 case Some{entry_count}:
150 header.entry_count.checked(input, U32.is_le(entry_count, 1048576))
151
152def header.version.checked(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, Header>:
153 match valid:
154 case False{}: Fail{InvalidFormat{}}
155 case True{}: header.entry_count(Bytes.get.u32be(input, 12))
156
157def header.version(pair: Bytes.Bytes & Maybe<&2, U32>) -> Result<&1, &1, Error, Header>:
158 match pair:
159 case (input, value):
160 match value:
161 case None{}: Fail{InvalidFormat{}}
162 case Some{version}: header.version.checked(input, U32.is_eq(version, 1))
163
164def header.magic.checked(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, Header>:
165 match valid:
166 case False{}: Fail{InvalidFormat{}}
167 case True{}: header.version(Bytes.get.u32be(input, 8))
168
169def header.marker(pair: Bytes.Bytes & Bytes.Bytes) -> Result<&1, &1, Error, Header>:
170 match pair:
171 case (input, marker):
172 header.magic.checked(input, equal(Bytes.eq(marker, magic.header())))
173
174def header.parse(input: Bytes.Bytes) -> Result<&1, &1, Error, Header>:
175 header.marker(Bytes.slice(input, 0, 8))
176
177def block.checksum.lengths(
178 lengths: U32 & U32,
179 copies: Bytes.Bytes & Bytes.Bytes
180) -> Bool:
181 match lengths:
182 case (hash_length, claim_offset):
183 match copies:
184 case (for_hash, for_claim):
185 digest.matches(slice.result(Bytes.slice(for_claim, claim_offset, 32)),
186 slice.result(Bytes.slice(for_hash, 0, hash_length)))
187
188def block.checksum.copies.lengths(
189 lengths: Nat & Nat,
190 copies: Bytes.Bytes & Bytes.Bytes
191) -> Bool:
192 match lengths:
193 case (hash_length, claim_offset):
194 block.checksum.lengths((U32.from_nat(hash_length),
195 U32.from_nat(claim_offset)), copies)
196
197def block.checksum.copies(
198 block_length: U32,
199 copies: Bytes.Bytes & Bytes.Bytes
200) -> Bool:
201 block.checksum.copies.lengths(
202 QueuePolicy.nat.clone.pair(U32.to_nat(U32.sub(block_length, 32))), copies)
203
204def block.checksum.sized.validated(
205 block: Bytes.Bytes,
206 hash_length: U32,
207 valid: Bool
208) -> Bool:
209 match valid:
210 case False{}: False{}
211 case True{}:
212 block.checksum.copies(hash_length, clone.split(block))
213
214def block.checksum.sized.measure(
215 block: Bytes.Bytes,
216 compare_length: U32,
217 hash_length: U32,
218 measured: Bytes.Bytes & U32
219) -> Bool:
220 match measured:
221 case (_, actual):
222 block.checksum.sized.validated(block, hash_length,
223 U32.is_eq(compare_length, actual))
224
225def block.checksum.sized.lengths(
226 block: Bytes.Bytes,
227 lengths: Nat & Nat,
228 measured: Bytes.Bytes & U32
229) -> Bool:
230 match lengths:
231 case (for_compare, for_hash):
232 block.checksum.sized.measure(block, U32.from_nat(for_compare),
233 U32.from_nat(for_hash), measured)
234
235def block.checksum.sized(
236 block: Bytes.Bytes,
237 block_length: U32,
238 measured: Bytes.Bytes & U32
239) -> Bool:
240 block.checksum.sized.lengths(block,
241 QueuePolicy.nat.clone.pair(U32.to_nat(block_length)), measured)
242
243def block.checksum.input.cloned(
244 block_length: U32,
245 copies: Bytes.Bytes & Bytes.Bytes
246) -> Bool:
247 match copies:
248 case (for_length, for_hash):
249 block.checksum.sized(for_hash, block_length, Bytes.length(for_length))
250
251def block.checksum.input(
252 block_length: U32,
253 block: Bytes.Bytes & Bytes.Bytes
254) -> Bool:
255 match block:
256 case (_, bytes):
257 block.checksum.input.cloned(block_length, clone.split(bytes))
258
259def mutations.entries.loop(
260 fuel: Nat,
261 mutations: List<&1, ByteWal.Mutation>,
262 reversed: List<&1, ByteMemTable.Entry>,
263 count: Nat
264) -> Result<&1, &1, Error, EntryBatch>:
265 match fuel:
266 case 0n: Fail{ResourceLimit{}}
267 case 1n+rest:
268 match mutations:
269 case Nil{}:
270 Done{EntryBatch{List.reverse(&1, ByteMemTable.Entry, reversed), count}}
271 case Con{mutation, tail}:
272 match mutation:
273 case ByteWal.Put{key, value}:
274 mutations.entries.loop(rest, tail,
275 Con{ByteMemTable.Entry{key, Some{value}}, reversed},
276 Nat.add(count, 1n))
277 case ByteWal.Delete{_}: Fail{InvalidFormat{}}
278
279def mutations.entries(
280 result: Result<&1, &1, Wal.Error, ByteWal.Batch>,
281 count: U32
282) -> Result<&1, &1, Error, EntryBatch>:
283 match result:
284 case Fail{_}: Fail{InvalidFormat{}}
285 case Done{ByteWal.Batch{mutations}}:
286 mutations.entries.loop(Nat.add(U32.to_nat(count), 1n), mutations, Nil{}, 0n)
287
288def footer.expected_prefix(entry_count: Nat, block_count: Nat) -> Bytes.Bytes:
289 Bytes.concat([footer.magic(),
290 Bytes.set.u32be(Bytes.new(4), 0, U32.from_nat(entry_count)),
291 Bytes.set.u32be(Bytes.new(4), 0, U32.from_nat(block_count))])
292
293def footer.valid.digest(copies: Bytes.Bytes & Bytes.Bytes) -> Bool:
294 match copies:
295 case (for_claim, for_hash):
296 digest.matches(slice.result(Bytes.slice(for_claim, 12, 32)),
297 slice.result(Bytes.slice(for_hash, 0, 12)))
298
299def footer.valid.copies(
300 copies: Bytes.Bytes & Bytes.Bytes,
301 entry_count: Nat,
302 block_count: Nat
303) -> Bool:
304 match copies:
305 case (for_fields, for_digest):
306 equal(Bytes.eq(slice.result(Bytes.slice(for_fields, 0, 12)),
307 footer.expected_prefix(entry_count, block_count))) &&
308 footer.valid.digest(clone.split(for_digest))
309
310def footer.valid.pair(
311 pair: Bytes.Bytes & Bytes.Bytes,
312 entry_count: Nat,
313 block_count: Nat
314) -> Bool:
315 match pair:
316 case (_, footer): footer.valid.copies(clone.split(footer), entry_count, block_count)
317
318def footer.valid(
319 input: Bytes.Bytes,
320 offset: U32,
321 entry_count: Nat,
322 block_count: Nat
323) -> Bool:
324 footer.valid.pair(Bytes.slice(input, offset, 44), entry_count, block_count)
325
326def footer.checked.checksum(
327 entries: List<&1, ByteMemTable.Entry>,
328 valid: Bool
329) -> Result<&1, &1, Error, Decoded>:
330 match valid:
331 case False{}: Fail{ChecksumMismatch{}}
332 case True{}: Done{Decoded{entries}}
333
334def footer.checked(
335 input: Bytes.Bytes,
336 offset: U32,
337 count: Nat,
338 block_count: Nat,
339 entries: List<&1, ByteMemTable.Entry>,
340 size_valid: Bool
341) -> Result<&1, &1, Error, Decoded>:
342 match size_valid:
343 case False{}: Fail{InvalidFormat{}}
344 case True{}:
345 footer.checked.checksum(entries, footer.valid(input, offset, count, block_count))
346
347def footer.result(
348 input: Bytes.Bytes,
349 offset: U32,
350 count: Nat,
351 block_count: Nat,
352 entries: List<&1, ByteMemTable.Entry>,
353 size_valid: Bool
354) -> Result<&1, &1, Error, Decoded>:
355 footer.checked(input, offset, count, block_count, entries, size_valid)
356
357def empty.footer.valid.copies(copies: Bytes.Bytes & Bytes.Bytes) -> Bool:
358 match copies:
359 case (for_footer, for_digest):
360 digest.matches(slice.result(Bytes.slice(for_footer, 32, 32)),
361 slice.result(Bytes.slice(for_digest, 20, 12)))
362
363def empty.footer.valid(input: Bytes.Bytes) -> Bool:
364 empty.footer.valid.copies(clone.split(input))
365
366def empty.header.blocks(pair: Bytes.Bytes & Maybe<&2, U32>) -> Bool:
367 match pair:
368 case (_, count): u32.field.is(count, 0)
369
370def empty.header.entries.checked(input: Bytes.Bytes, valid: Bool) -> Bool:
371 match valid:
372 case False{}: False{}
373 case True{}: empty.header.blocks(Bytes.get.u32be(input, 16))
374
375def empty.header.entries(pair: Bytes.Bytes & Maybe<&2, U32>) -> Bool:
376 match pair:
377 case (input, count):
378 empty.header.entries.checked(input, u32.field.is(count, 0))
379
380def empty.header.version.checked(input: Bytes.Bytes, valid: Bool) -> Bool:
381 match valid:
382 case False{}: False{}
383 case True{}: empty.header.entries(Bytes.get.u32be(input, 12))
384
385def empty.header.version(pair: Bytes.Bytes & Maybe<&2, U32>) -> Bool:
386 match pair:
387 case (input, version):
388 empty.header.version.checked(input, u32.field.is(version, 1))
389
390def empty.header.magic.checked(input: Bytes.Bytes, valid: Bool) -> Bool:
391 match valid:
392 case False{}: False{}
393 case True{}: empty.header.version(Bytes.get.u32be(input, 8))
394
395def empty.header.magic(pair: Bytes.Bytes & Bytes.Bytes) -> Bool:
396 match pair:
397 case (input, marker):
398 empty.header.magic.checked(input, equal(Bytes.eq(marker, magic.header())))
399
400def empty.header.valid(input: Bytes.Bytes) -> Bool:
401 empty.header.magic(Bytes.slice(input, 0, 8))
402
403def decode.empty(valid: Bool) -> Result<&1, &1, Error, Decoded>:
404 match valid:
405 case False{}: Fail{InvalidFormat{}}
406 case True{}: Done{Decoded{Nil{}}}
407
408def empty.footer.fields(input: Bytes.Bytes) -> Bool:
409 equal(Bytes.eq(slice.result(Bytes.slice(input, 20, 12)),
410 Bytes.concat([footer.magic(), Bytes.new(8)])))
411
412def decode.empty.footer.checked(
413 input: Bytes.Bytes,
414 checksum_valid: Bool
415) -> Result<&1, &1, Error, Decoded>:
416 match checksum_valid:
417 case False{}: Fail{ChecksumMismatch{}}
418 case True{}: decode.empty(empty.footer.fields(input))
419
420def decode.empty.footer.copies(
421 copies: Bytes.Bytes & Bytes.Bytes
422) -> Result<&1, &1, Error, Decoded>:
423 match copies:
424 case (for_fields, for_digest):
425 decode.empty.footer.checked(for_fields, empty.footer.valid(for_digest))
426
427def decode.empty.header.checked(
428 input: Bytes.Bytes,
429 header_valid: Bool
430) -> Result<&1, &1, Error, Decoded>:
431 match header_valid:
432 case False{}: Fail{InvalidFormat{}}
433 case True{}: decode.empty.footer.copies(clone.split(input))
434
435def decode.empty.header.copies(
436 copies: Bytes.Bytes & Bytes.Bytes
437) -> Result<&1, &1, Error, Decoded>:
438 match copies:
439 case (for_footer, for_header):
440 decode.empty.header.checked(for_footer, empty.header.valid(for_header))
441
442def decode.empty.size.validated(
443 input: Bytes.Bytes,
444 size_valid: Bool
445) -> Result<&1, &1, Error, Decoded>:
446 match size_valid:
447 case False{}: Fail{InvalidFormat{}}
448 case True{}: decode.empty.header.copies(clone.split(input))
449
450def decode.empty.size(input: Bytes.Bytes, size: U32) -> Result<&1, &1, Error, Decoded>:
451 decode.empty.size.validated(input, U32.is_eq(size, 64))
452
453def decode.length(pair: Bytes.Bytes & U32) -> Result<&1, &1, Error, Decoded>:
454 match pair:
455 case (input, size): decode.empty.size(input, size)
456
457def restore.block.frame.bounded(
458 input: Bytes.Bytes,
459 +offset: Nat,
460 index: U32,
461 +record_count: U32,
462 +frame_length: U32,
463 +block_length: Nat,
464 valid: Bool
465) -> Result<&1, &1, Error, BlockHeader>:
466 match valid:
467 case False{}: Fail{InvalidFormat{}}
468 case True{}:
469 Done{BlockHeader{input, offset, index, record_count,
470 frame_length, block_length}}
471def restore.block.frame.bounds(
472 input: Bytes.Bytes,
473 +offset: Nat,
474 index: U32,
475 record_count: U32,
476 +frame_length: U32,
477 +block_length: Nat,
478 +input_size: U32
479) -> Result<&1, &1, Error, BlockHeader>:
480 restore.block.frame.bounded(input, offset, index, record_count,
481 frame_length, block_length,
482 Nat.is_le(Nat.add(offset, block_length), U32.to_nat(input_size)))
483def restore.block.frame.limit(
484 input: Bytes.Bytes,
485 +offset: Nat,
486 index: U32,
487 record_count: U32,
488 +frame_length: U32,
489 +input_size: U32,
490 valid: Bool
491) -> Result<&1, &1, Error, BlockHeader>:
492 match valid:
493 case False{}: Fail{ResourceLimit{}}
494 case True{}:
495 +block_length = Nat.add(U32.to_nat(frame_length), 48n)
496 restore.block.frame.bounds(input, offset, index, record_count,
497 frame_length, block_length, input_size)
498def restore.block.frame.length.checked(
499 measured: Bytes.Bytes & U32,
500 +offset: Nat,
501 index: U32,
502 record_count: U32,
503 +frame_length: U32,
504 valid: Bool
505) -> Result<&1, &1, Error, BlockHeader>:
506 match measured:
507 case (input, input_size):
508 restore.block.frame.limit(input, offset, index, record_count,
509 frame_length, input_size, valid)
510
511def restore.block.frame.field(
512 pair: Bytes.Bytes & Maybe<&2, U32>,
513 +offset: Nat,
514 index: U32,
515 record_count: U32
516) -> Result<&1, &1, Error, BlockHeader>:
517 match pair:
518 case (input, result):
519 match result:
520 case None{}: Fail{InvalidFormat{}}
521 case Some{+frame_length}:
522 restore.block.frame.length.checked(Bytes.length(input), offset,
523 index, record_count, frame_length,
524 U32.is_le(frame_length, StorageBytes.MAX_BLOCK_BYTES()))
525
526def restore.block.count.valid(
527 input: Bytes.Bytes,
528 +offset: Nat,
529 index: U32,
530 record_count: U32,
531 within_limit: Bool,
532 empty: Bool
533) -> Result<&1, &1, Error, BlockHeader>:
534 match within_limit:
535 case False{}: Fail{ResourceLimit{}}
536 case True{}:
537 match empty:
538 case True{}: Fail{InvalidFormat{}}
539 case False{}:
540 restore.block.frame.field(Bytes.get.u32be(input,
541 U32.from_nat(Nat.add(offset, 12n))), offset,
542 index, record_count)
543def restore.block.count.field(
544 pair: Bytes.Bytes & Maybe<&2, U32>,
545 +offset: Nat,
546 index: U32
547) -> Result<&1, &1, Error, BlockHeader>:
548 match pair:
549 case (input, result):
550 match result:
551 case None{}: Fail{InvalidFormat{}}
552 case Some{+record_count}:
553 restore.block.count.valid(input, offset, index, record_count,
554 U32.is_le(record_count, 16), U32.is_eq(record_count, 0))
555def restore.block.index.checked(
556 input: Bytes.Bytes,
557 +offset: Nat,
558 index: U32,
559 valid: Bool
560) -> Result<&1, &1, Error, BlockHeader>:
561 match valid:
562 case False{}: Fail{InvalidFormat{}}
563 case True{}:
564 restore.block.count.field(Bytes.get.u32be(input,
565 U32.from_nat(Nat.add(offset, 8n))), offset, index)
566def restore.block.index.field(
567 pair: Bytes.Bytes & Maybe<&2, U32>,
568 +offset: Nat,
569 expected_index: U32
570) -> Result<&1, &1, Error, BlockHeader>:
571 match pair:
572 case (input, result):
573 match result:
574 case None{}: Fail{InvalidFormat{}}
575 case Some{+index}:
576 restore.block.index.checked(input, offset, index,
577 U32.is_eq(index, expected_index))
578def restore.block.index.valid(
579 pair: Bytes.Bytes & Maybe<&2, U32>,
580 +offset: Nat,
581 expected_index: U32
582) -> Result<&1, &1, Error, BlockHeader>:
583 restore.block.index.field(pair, offset, expected_index)
584def restore.block.magic.valid(
585 input: Bytes.Bytes,
586 +offset: Nat,
587 expected_index: U32,
588 valid: Bool
589) -> Result<&1, &1, Error, BlockHeader>:
590 match valid:
591 case False{}: Fail{InvalidFormat{}}
592 case True{}:
593 restore.block.index.valid(Bytes.get.u32be(input,
594 U32.from_nat(Nat.add(offset, 4n))), offset, expected_index)
595
596
597def restore.block.magic(
598 pair: Bytes.Bytes & Bytes.Bytes,
599 +offset: Nat,
600 expected_index: U32
601) -> Result<&1, &1, Error, BlockHeader>:
602 match pair:
603 case (input, marker):
604 restore.block.magic.valid(input, offset, expected_index,
605 equal(Bytes.eq(marker, block.magic())))
606
607def restore.block.header(
608 input: Bytes.Bytes,
609 +offset: Nat,
610 expected_index: U32
611) -> Result<&1, &1, Error, BlockHeader>:
612 restore.block.magic(Bytes.slice(input, U32.from_nat(offset), 4),
613 offset, expected_index)
614
615def restore.block.batch.checked(
616 input: Bytes.Bytes,
617 +offset: Nat,
618 index: U32,
619 record_count: U32,
620 frame_length: U32,
621 +block_length: Nat,
622 batch: EntryBatch,
623 valid: Bool
624) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
625 match batch:
626 case EntryBatch{entries, _}:
627 match valid:
628 case False{}: Fail{CountMismatch{}}
629 case True{}:
630 Done{(BlockHeader{input, offset, index, record_count,
631 frame_length, block_length}, entries)}
632
633
634def restore.block.batch.measure(
635 input: Bytes.Bytes,
636 +offset: Nat,
637 index: U32,
638 +record_count: U32,
639 frame_length: U32,
640 +block_length: Nat,
641 batch: EntryBatch
642) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
643 match batch:
644 case EntryBatch{_, +count}:
645 restore.block.batch.checked(input, offset, index, record_count,
646 frame_length, block_length, batch,
647 Nat.is_eq(count, U32.to_nat(record_count)))
648
649
650
651def restore.block.frame.entries(
652 header: BlockHeader,
653 result: Result<&1, &1, Error, EntryBatch>
654) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
655 match header:
656 case BlockHeader{input, offset, index, record_count, frame_length,
657 block_length}:
658 match result:
659 case Fail{error}: Fail{error}
660 case Done{batch}:
661 restore.block.batch.measure(input, offset, index, record_count,
662 frame_length, block_length, batch)
663
664def restore.block.frame.decoded(
665 header: BlockHeader,
666 result: Result<&1, &1, Wal.Error, ByteWal.Batch>
667) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
668 match header:
669 case BlockHeader{_, _, _, +record_count, _, _}:
670 restore.block.frame.entries(header, mutations.entries(result, record_count))
671
672
673def restore.block.checksum.validated(
674 header: BlockHeader,
675 decoded: Result<&1, &1, Wal.Error, ByteWal.Batch>,
676 valid: Bool
677) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
678 match valid:
679 case False{}: Fail{ChecksumMismatch{}}
680 case True{}: restore.block.frame.decoded(header, decoded)
681
682
683def restore.block.checksum.copies(
684 +offset: Nat,
685 index: U32,
686 record_count: U32,
687 +frame_length: U32,
688 +block_length: Nat,
689 copies: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes
690) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
691 match copies:
692 case (for_hash, for_frame, for_input):
693 restore.block.checksum.validated(
694 BlockHeader{for_input, offset, index, record_count,
695 frame_length, block_length},
696 ByteWal.decode(slice.result(Bytes.slice(for_frame,
697 U32.from_nat(Nat.add(offset, 16n)), frame_length))),
698 block.checksum.input(U32.from_nat(block_length),
699 Bytes.slice(for_hash, U32.from_nat(offset),
700 U32.from_nat(block_length))))
701
702
703def restore.block.checksum.valid(
704 header: BlockHeader
705) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
706 match header:
707 case BlockHeader{input, +offset, index, record_count, +frame_length,
708 +block_length}:
709 restore.block.checksum.copies(offset, index, record_count,
710 frame_length, block_length, clone.triple(input))
711
712def restore.block.checked(
713 result: Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>
714) -> Result<&1, &1, Error, BlockMeta>:
715 match result:
716 case Fail{error}: Fail{error}
717 case Done{(verified, entries)}:
718 match verified:
719 case BlockHeader{input, +offset, _, +record_count,
720 +frame_length, +block_length}:
721 Done{BlockMeta{input, offset, Nat.add(offset, 16n),
722 U32.to_nat(frame_length), U32.to_nat(record_count), block_length,
723 entries}}
724
725def restore.block.open.result(
726 result: Result<&1, &1, Error, BlockHeader>
727) -> Result<&1, &1, Error, BlockMeta>:
728 match result:
729 case Fail{error}: Fail{error}
730 case Done{header}:
731 restore.block.checked(restore.block.checksum.valid(header))
732
733
734def restore.block.open(
735 input: Bytes.Bytes,
736 +offset: Nat,
737 expected_index: U32
738) -> Result<&1, &1, Error, BlockMeta>:
739 restore.block.open.result(restore.block.header(input, offset, expected_index))
740
741
742def restore.prepend.entries(
743 entries: List<&1, ByteMemTable.Entry>,
744 reversed: List<&1, ByteMemTable.Entry>
745) -> List<&1, ByteMemTable.Entry>:
746 match entries:
747 case Nil{}: reversed
748 case Con{entry, tail}:
749 restore.prepend.entries(tail, Con{entry, reversed})
750
751def restore.finish.order(
752 input: Bytes.Bytes,
753 +offset: Nat,
754 +entry_count: U32,
755 +block_count: U32,
756 entries: List<&1, ByteMemTable.Entry>,
757 ordered: Bool
758) -> Result<&1, &1, Error, Decoded>:
759 match ordered:
760 case False{}: Fail{InvalidOrder{}}
761 case True{}:
762 footer.checked.checksum(entries,
763 footer.valid(input, U32.from_nat(offset), U32.to_nat(entry_count),
764 U32.to_nat(block_count)))
765
766
767def restore.finish.footer.copies(
768 input: Bytes.Bytes,
769 +offset: Nat,
770 entry_count: U32,
771 block_count: U32,
772 copies: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
773) -> Result<&1, &1, Error, Decoded>:
774 match copies:
775 case (for_order, entries):
776 restore.finish.order(input, offset, entry_count, block_count, entries,
777 ByteMemTable.ordered.strict(for_order))
778
779
780
781def restore.finish.footer(
782 input: Bytes.Bytes,
783 +offset: Nat,
784 entry_count: U32,
785 block_count: U32,
786 entries: List<&1, ByteMemTable.Entry>
787) -> Result<&1, &1, Error, Decoded>:
788 restore.finish.footer.copies(input, offset, entry_count, block_count,
789 ByteMemTable.entries.clone(entries))
790
791def restore.finish.size.valid(
792 input: Bytes.Bytes,
793 +offset: Nat,
794 entry_count: U32,
795 block_count: U32,
796 reversed: List<&1, ByteMemTable.Entry>,
797 valid: Bool
798) -> Result<&1, &1, Error, Decoded>:
799 match valid:
800 case False{}: Fail{InvalidFormat{}}
801 case True{}:
802 restore.finish.footer(input, offset, entry_count, block_count,
803 List.reverse(&1, ByteMemTable.Entry, reversed))
804
805
806def restore.finish.count.valid(
807 input: Bytes.Bytes,
808 +offset: Nat,
809 entry_count: U32,
810 block_count: U32,
811 reversed: List<&1, ByteMemTable.Entry>,
812 +size: U32,
813 valid: Bool
814) -> Result<&1, &1, Error, Decoded>:
815 match valid:
816 case False{}: Fail{CountMismatch{}}
817 case True{}:
818 restore.finish.size.valid(input, offset, entry_count, block_count,
819 reversed, Nat.is_eq(Nat.add(offset, 44n), U32.to_nat(size)))
820
821
822def restore.finish(
823 measured: Bytes.Bytes & U32,
824 +offset: Nat,
825 +entry_count: U32,
826 +block_count: U32,
827 +decoded_count: Nat,
828 reversed: List<&1, ByteMemTable.Entry>
829) -> Result<&1, &1, Error, Decoded>:
830 match measured:
831 case (input, size):
832 restore.finish.count.valid(input, offset, entry_count, block_count,
833 reversed, size,
834 U32.is_eq(U32.from_nat(decoded_count), entry_count))
835
836def restore.loop(
837 fuel: Nat,
838 state: State,
839 current: Result<&1, &1, Error, BlockMeta>
840) -> Result<&1, &1, Error, Decoded>:
841 match fuel:
842 case 0n: Fail{InvalidFormat{}}
843 case 1n+fuel_rest:
844 match state:
845 case State{input, +offset, entry_count, block_count, +remaining_blocks,
846 +block_index, +decoded_count, reversed}:
847 match input:
848 case (for_size, retained):
849 match remaining_blocks:
850 case 0n: Fail{InvalidFormat{}}
851 case 1n+rest:
852 match rest:
853 case 0n:
854 match current:
855 case Fail{error}: Fail{error}
856 case Done{block}:
857 match block:
858 case BlockMeta{_, _, _, _, +record_count, +block_length, entries}:
859 +next_offset = Nat.add(offset, block_length)
860 +next_count = Nat.add(decoded_count, record_count)
861 next_reversed = restore.prepend.entries(entries, reversed)
862 restore.finish(Bytes.length(for_size), next_offset,
863 entry_count, block_count, next_count,
864 next_reversed)
865 case 1n+more:
866 match current:
867 case Fail{error}: Fail{error}
868 case Done{block}:
869 match block:
870 case BlockMeta{_, _, _, _, +record_count, +block_length, entries}:
871 +next_offset = Nat.add(offset, block_length)
872 +next_index = Nat.add(block_index, 1n)
873 +next_count = Nat.add(decoded_count, record_count)
874 next_reversed = restore.prepend.entries(entries, reversed)
875 restore.loop(fuel_rest,
876 State{clone.split(retained), next_offset, entry_count,
877 block_count,
878 RestorePolicy.remaining(
879 RestorePolicy.after_block(remaining_blocks)),
880 next_index, next_count,
881 next_reversed},
882 restore.block.open(for_size, next_offset,
883 U32.from_nat(next_index)))
884
885def restore.loop.start(
886 copies: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes,
887 entry_count: U32,
888 +block_count: U32
889) -> Result<&1, &1, Error, Decoded>:
890 match copies:
891 case (for_state, for_retained, for_open):
892 restore.loop(U32.to_nat(block_count),
893 State{(for_state, for_retained), 20n, entry_count, block_count,
894 U32.to_nat(block_count), 0n, 0n, Nil{}},
895 restore.block.open(for_open, 20n, 0))
896
897
898def decode.header.blocks(input: Bytes.Bytes, header: Header) -> Result<&1, &1, Error, Decoded>:
899 match header:
900 case Header{_, entry_count, block_count}:
901 restore.loop.start(clone.triple(input), entry_count, block_count)
902
903def decode.header.route(input: Bytes.Bytes, header: Header, empty: Bool) -> Result<&1, &1, Error, Decoded>:
904 match empty:
905 case True{}: decode.length(Bytes.length(input))
906 case False{}: decode.header.blocks(input, header)
907
908def decode.header.ready.count(
909 input: Bytes.Bytes,
910 header_input: Bytes.Bytes,
911 entry_count: U32,
912 counts: Nat & Nat
913) -> Result<&1, &1, Error, Decoded>:
914 match counts:
915 case (for_check, for_header):
916 decode.header.route(input,
917 Header{header_input, entry_count, U32.from_nat(for_header)},
918 U32.is_eq(U32.from_nat(for_check), 0))
919
920def decode.header.ready(input: Bytes.Bytes, header: Header) -> Result<&1, &1, Error, Decoded>:
921 match header:
922 case Header{header_input, entry_count, block_count}:
923 decode.header.ready.count(input, header_input, entry_count,
924 QueuePolicy.nat.clone.pair(U32.to_nat(block_count)))
925
926def decode.header(input: Bytes.Bytes, result: Result<&1, &1, Error, Header>) -> Result<&1, &1, Error, Decoded>:
927 match result:
928 case Fail{error}: Fail{error}
929 case Done{header}: decode.header.ready(input, header)
930
931def decode.copies(copies: Bytes.Bytes & Bytes.Bytes) -> Result<&1, &1, Error, Decoded>:
932 match copies:
933 case (for_restore, for_header):
934 decode.header(for_restore, header.parse(for_header))
935
936# Validates a bounded exchange and decodes the currently supported block layout.
937def decode(input: Bytes.Bytes) -> Result<&1, &1, Error, Decoded>:
938 decode.copies(clone.split(input))