src/ByteSstFile.bend
mylsm-lsm-store@0.6.0.0 · 1,968 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/StorageBytes.bend as StorageBytes
5import mylsm-lsm-store@0.6.0.0/src/scan/ByteBlockBuilder.bend as ByteBlockBuilder
6import mylsm-lsm-store@0.6.0.0/src/coordinator/QueuePolicy.bend as QueuePolicy
7import mylsm-lsm-store@0.6.0.0/src/Wal.bend as Wal
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# Errors produced by the generation-four byte SST codec.
12type Error is Data:
13 InvalidLevel{}
14 InvalidOrder{}
15 TooLarge{}
16 Malformed{}
17 ChecksumMismatch{}
18
19# Carries validated one-block envelope metadata after the fixed header.
20type PayloadWords is Type:
21 PayloadWords{cursor: Bytes.Cursor, size: U32, level: U32, total: U32,
22 payload_len: U32, entries: U32}
23
24# Carries a byte-bounded page under construction and completed pages in reverse order.
25type PageBuild is Type:
26 PageBuild{current: List<&1, ByteMemTable.Entry>, current_bytes: Nat,
27 current_count: Nat, completed: List<&1, List<&1, ByteMemTable.Entry>>,
28 completed_count: Nat}
29
30# Holds a size-bounded page sequence and its exact count.
31type PagePlan is Type:
32 PagePlan{pages: List<&1, List<&1, ByteMemTable.Entry>>, count: Nat}
33
34# Carries the shared header metadata used to parse one multi-block SST.
35type MultiHeader is Type:
36 MultiHeader{file: Bytes.Bytes, level: U32, total: U32, block_count: U32}
37
38# Carries the state while decoding blocks in file order.
39type MultiRead is Type:
40 MultiRead{file: Bytes.Bytes, index_offset: Nat, next_block: Nat,
41 block_count: Nat, level: U32, total: Nat, decoded: Nat,
42 table: ByteMemTable.Table}
43
44# Carries an encoded page block and its record count.
45type EncodedBlock is Type:
46 EncodedBlock{bytes: Bytes.Bytes, byte_length: U32, checksum: Bytes.Bytes,
47 count: U32, separator: Bytes.Bytes, separator_length: U32}
48
49# Holds every encoded block and the exact number of pages produced.
50type EncodedPages is Type:
51 EncodedPages{blocks: List<&1, EncodedBlock>, count: U32}
52
53# Holds concatenated data blocks and the metadata used by the index and root hash.
54type BlocksCollected is Type:
55 BlocksCollected{bytes: Bytes.Bytes, checksums: Bytes.Bytes,
56 index: Bytes.Bytes, block_count: U32, entry_count: U32}
57
58# Reifies two independently owned copies of a byte buffer.
59def clone.pair(+len: U32, copied: Array<U32> & Array<U32>) -> Bytes.Bytes & Bytes.Bytes:
60 match copied:
61 case (left, right):
62 (Bytes.Bytes{len, left}, Bytes.Bytes{len, right})
63
64# Clones bytes when output and hashing both need the same contents.
65def clone.split(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes:
66 match bytes:
67 case Bytes.Bytes{+len, buf}:
68 clone.pair(len, Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
69
70# Encodes an unsigned 32-bit word in network byte order.
71def u32.bytes(value: U32) -> Bytes.Bytes:
72 Bytes.set.u32be(Bytes.new(4), 0, value)
73
74# Unwraps fixed known format markers.
75def marker(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
76 match value:
77 case Some{bytes}:
78 bytes
79 case None{}:
80 Bytes.new(0)
81
82# Returns the generation-four SST signature.
83def header.magic() -> Bytes.Bytes:
84 marker(Bytes.from_hex("4d594c534d345300"))
85
86# Returns the generation-four index signature.
87def index.magic() -> Bytes.Bytes:
88 marker(Bytes.from_hex("49445834"))
89
90# Returns the generation-four SST footer signature.
91def footer.magic() -> Bytes.Bytes:
92 marker(Bytes.from_hex("454e4434"))
93
94# Converts byte table entries to the shared byte WAL mutation representation.
95def mutations(entries: List<&1, ByteMemTable.Entry>) -> List<&1, ByteWal.Mutation>:
96 match entries:
97 case Nil{}:
98 Nil{}
99 case Con{ByteMemTable.Entry{key, val}, tail}:
100 match val:
101 case None{}:
102 Con{ByteWal.Delete{key}, mutations(tail)}
103 case Some{value}:
104 Con{ByteWal.Put{key, value}, mutations(tail)}
105
106# Rebuilds two entries from independently owned key and value copies.
107def entry.clone.parts(
108 keys: Bytes.Bytes & Bytes.Bytes,
109 values: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>
110) -> ByteMemTable.Entry & ByteMemTable.Entry:
111 match keys:
112 case (first_key, second_key):
113 match values:
114 case (first_value, second_value):
115 (ByteMemTable.Entry{first_key, first_value},
116 ByteMemTable.Entry{second_key, second_value})
117
118# Clones one affine entry for independent validation and encoding passes.
119def entry.clone(entry: ByteMemTable.Entry) -> ByteMemTable.Entry & ByteMemTable.Entry:
120 match entry:
121 case ByteMemTable.Entry{key, value}:
122 entry.clone.parts(ByteMemTable.bytes.clone(key), ByteMemTable.value.clone(value))
123
124# Prepends two cloned entries to their matching ordered lists.
125def entries.clone.push(
126 head: ByteMemTable.Entry & ByteMemTable.Entry,
127 tail: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
128) -> List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>:
129 match head:
130 case (first, second):
131 match tail:
132 case (first_tail, second_tail):
133 (Con{first, first_tail}, Con{second, second_tail})
134
135# Clones an entire affine entry list without changing its order.
136def entries.clone(entries: List<&1, ByteMemTable.Entry>) -> List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>:
137 match entries:
138 case Nil{}:
139 (Nil{}, Nil{})
140 case Con{entry, tail}:
141 entries.clone.push(entry.clone(entry), entries.clone(tail))
142
143# Emits all authenticated file sections with a completed root digest.
144def file.emit(
145 header: Bytes.Bytes,
146 block: Bytes.Bytes,
147 index: Bytes.Bytes,
148 footer: Bytes.Bytes
149) -> Result<&1, &1, Error, Bytes.Bytes>:
150 Done{Bytes.concat([header, block, index, footer])}
151
152# Adds the root footer after hashing header, block digest, and index metadata.
153def file.root(
154 header_output: Bytes.Bytes,
155 header_hash: Bytes.Bytes,
156 block: Bytes.Bytes,
157 block_root: Bytes.Bytes,
158 index_output: Bytes.Bytes,
159 index_hash: Bytes.Bytes,
160 index_length: U32,
161 +total: U32
162) -> Result<&1, &1, Error, Bytes.Bytes>:
163 file.emit(header_output, block, index_output,
164 Bytes.concat([footer.magic(), u32.bytes(total), u32.bytes(index_length),
165 SHA.sha256_packed_bytes(Bytes.concat([header_hash, block_root, index_hash]))]))
166
167# Separates the index copies used by output and root authentication.
168def file.index.length(
169 header_output: Bytes.Bytes,
170 header_hash: Bytes.Bytes,
171 block: Bytes.Bytes,
172 block_root: Bytes.Bytes,
173 measured: Bytes.Bytes & U32,
174 index_hash: Bytes.Bytes,
175 +total: U32
176) -> Result<&1, &1, Error, Bytes.Bytes>:
177 match measured:
178 case (output, index_length):
179 file.root(header_output, header_hash, block, block_root,
180 output, index_hash, index_length, total)
181
182def file.index.copies(
183 header_output: Bytes.Bytes,
184 header_hash: Bytes.Bytes,
185 block: Bytes.Bytes,
186 block_root: Bytes.Bytes,
187 copies: Bytes.Bytes & Bytes.Bytes,
188 +total: U32
189) -> Result<&1, &1, Error, Bytes.Bytes>:
190 match copies:
191 case (index_output, index_hash):
192 file.index.length(header_output, header_hash, block, block_root,
193 Bytes.length(index_output), index_hash, total)
194
195# Builds the one-block index page and clones it for root authentication.
196def file.index(
197 header_output: Bytes.Bytes,
198 header_hash: Bytes.Bytes,
199 block: Bytes.Bytes,
200 block_root: Bytes.Bytes,
201 +total: U32,
202 +payload_len: U32
203) -> Result<&1, &1, Error, Bytes.Bytes>:
204 file.index.copies(header_output, header_hash, block, block_root,
205 clone.split(Bytes.concat([index.magic(), u32.bytes(1),
206 u32.bytes(20), u32.bytes((40 + payload_len : U32))])), total)
207
208# Separates header copies needed by output and root authentication.
209def file.header.copies(
210 copies: Bytes.Bytes & Bytes.Bytes,
211 block: Bytes.Bytes,
212 block_root: Bytes.Bytes,
213 +total: U32,
214 +payload_len: U32
215) -> Result<&1, &1, Error, Bytes.Bytes>:
216 match copies:
217 case (header_output, header_hash):
218 file.index(header_output, header_hash, block, block_root, total, payload_len)
219
220# Creates the one-block header and gives its copy to the root calculation.
221def file.header(
222 block: Bytes.Bytes,
223 block_root: Bytes.Bytes,
224 +level: U32,
225 +total: U32,
226 +payload_len: U32
227) -> Result<&1, &1, Error, Bytes.Bytes>:
228 file.header.copies(clone.split(Bytes.concat([header.magic(),
229 u32.bytes(level), u32.bytes(total), u32.bytes(1)])), block,
230 block_root, total, payload_len)
231
232# Splits a block digest for the stored block and root calculation.
233def file.digest.copies(
234 block: Bytes.Bytes,
235 copies: Bytes.Bytes & Bytes.Bytes,
236 +level: U32,
237 +total: U32,
238 +payload_len: U32
239) -> Result<&1, &1, Error, Bytes.Bytes>:
240 match copies:
241 case (block_digest, block_root):
242 file.header(Bytes.concat([block, block_digest]), block_root,
243 level, total, payload_len)
244
245# Adds a block checksum to the encoded block header and payload.
246def file.digest(
247 block_header: Bytes.Bytes,
248 frame: Bytes.Bytes,
249 digest: Bytes.Bytes,
250 +level: U32,
251 +total: U32,
252 +payload_len: U32
253) -> Result<&1, &1, Error, Bytes.Bytes>:
254 file.digest.copies(Bytes.concat([block_header, frame]),
255 clone.split(digest), level, total, payload_len)
256
257# Unpacks both copies before hashing and constructing the block.
258def file.block.parts(
259 headers: Bytes.Bytes & Bytes.Bytes,
260 frames: Bytes.Bytes & Bytes.Bytes,
261 +level: U32,
262 +total: U32,
263 +payload_len: U32
264) -> Result<&1, &1, Error, Bytes.Bytes>:
265 match headers:
266 case (block_output, block_hash):
267 match frames:
268 case (frame_output, frame_hash):
269 file.digest(block_output, frame_output,
270 SHA.sha256_packed_bytes(Bytes.concat([block_hash, frame_hash])),
271 level, total, payload_len)
272
273# Hashes cloned block header and frame bytes before building the file.
274def file.block.copies(
275 headers: Bytes.Bytes & Bytes.Bytes,
276 frames: Bytes.Bytes & Bytes.Bytes,
277 +level: U32,
278 +total: U32,
279 +payload_len: U32
280) -> Result<&1, &1, Error, Bytes.Bytes>:
281 match headers:
282 case header_pair:
283 match frames:
284 case frame_pair:
285 file.block.parts(header_pair, frame_pair, level, total, payload_len)
286
287# Creates fixed block header copies and payload copies for hashing.
288def file.block(
289 frame: Bytes.Bytes,
290 +level: U32,
291 +total: U32,
292 +payload_len: U32
293) -> Result<&1, &1, Error, Bytes.Bytes>:
294 file.block.copies(clone.split(Bytes.concat([u32.bytes(payload_len),
295 u32.bytes(total)])), clone.split(frame), level, total, payload_len)
296
297# Rejects an encoded WAL frame larger than the resident block limit.
298def file.bounded(
299 frame: Bytes.Bytes,
300 +level: U32,
301 +total: U32,
302 +payload_len: U32,
303 allowed: Bool
304) -> Result<&1, &1, Error, Bytes.Bytes>:
305 match allowed:
306 case False{}:
307 Fail{TooLarge{}}
308 case True{}:
309 file.block(frame, level, total, payload_len)
310
311# Unpacks frame bytes and their measured length.
312def file.measured(
313 pair: Bytes.Bytes & U32,
314 +level: U32,
315 +total: U32
316) -> Result<&1, &1, Error, Bytes.Bytes>:
317 match pair:
318 case (frame, +payload_len):
319 file.bounded(frame, level, total, payload_len,
320 U32.is_le(payload_len, StorageBytes.MAX_BLOCK_BYTES()))
321
322# Encodes a checksummed one-block generation-four SST file.
323def file(
324 frame: Bytes.Bytes,
325 +level: U32,
326 +total: U32
327) -> Result<&1, &1, Error, Bytes.Bytes>:
328 file.measured(Bytes.length(frame), level, total)
329
330# Measures one put record without allocating its serialized bytes.
331def record.size.put(+key_len: U32, value: Bytes.Bytes & U32) -> Nat:
332 match value:
333 case (_, +value_len):
334 Nat.add(9n, Nat.add(U32.to_nat(key_len), U32.to_nat(value_len)))
335
336# Measures a put or tombstone record from exact raw byte lengths.
337def record.size(pair: Bytes.Bytes & U32, value: Maybe<&1, Bytes.Bytes>) -> Nat:
338 match pair:
339 case (_, +key_len):
340 match value:
341 case None{}:
342 Nat.add(5n, U32.to_nat(key_len))
343 case Some{bytes}:
344 record.size.put(key_len, Bytes.length(bytes))
345
346# Sums exact record bytes while preserving a finite structural recursion bound.
347def records.size.go(
348 fuel: Nat,
349 entries: List<&1, ByteMemTable.Entry>,
350 total: Nat
351) -> Nat:
352 match fuel:
353 case 0n:
354 total
355 case 1n+rest:
356 match entries:
357 case Nil{}:
358 total
359 case Con{ByteMemTable.Entry{key, value}, tail}:
360 +entry_size = record.size(Bytes.length(key), value)
361 records.size.go(rest, tail, Nat.add(total, entry_size))
362
363# Measures all encoded records before frame allocation.
364def records.size(entries: List<&1, ByteMemTable.Entry>, +count: Nat) -> Nat:
365 records.size.go(Nat.add(count, 1n), entries, 0n)
366
367def page.record.required.value(record_bytes: Nat, has_current_records: Bool) -> Nat:
368 match has_current_records:
369 case True{}: record_bytes
370 case False{}: Nat.add(40n, record_bytes)
371
372def page.record.required(
373 entry: ByteMemTable.Entry,
374 has_current_records: Bool
375) -> Nat:
376 match entry:
377 case ByteMemTable.Entry{key, value}:
378 page.record.required.value(record.size(Bytes.length(key), value),
379 has_current_records)
380
381def page.finish(
382 pages: List<&1, List<&1, ByteMemTable.Entry>>,
383 current: List<&1, ByteMemTable.Entry>,
384 completed_count: Nat
385) -> PagePlan:
386 match current:
387 case Nil{}:
388 PagePlan{List.reverse(&1, List<&1, ByteMemTable.Entry>, pages), completed_count}
389 case _:
390 PagePlan{List.reverse(&1, List<&1, ByteMemTable.Entry>, Con{
391 List.reverse(&1, ByteMemTable.Entry, current), pages}),
392 Nat.add(completed_count, 1n)}
393
394def page.rollover(
395 entry: ByteMemTable.Entry,
396 current: List<&1, ByteMemTable.Entry>,
397 pages: List<&1, List<&1, ByteMemTable.Entry>>,
398 completed_count: Nat,
399 required_bytes: Nat
400) -> PageBuild:
401 PageBuild{Con{entry, Nil{}}, required_bytes, 1n,
402 Con{List.reverse(&1, ByteMemTable.Entry, current), pages},
403 Nat.add(completed_count, 1n)}
404
405def page.next.cloned(
406 tail: List<&1, ByteMemTable.Entry>,
407 current_count: Nat,
408 current_bytes: Nat,
409 +max_bytes: Nat,
410 copies: ByteMemTable.Entry & ByteMemTable.Entry
411) -> List<&1, ByteMemTable.Entry> & ByteBlockBuilder.Decision:
412 match copies:
413 case (for_decision, for_remaining):
414 (Con{for_remaining, tail}, ByteBlockBuilder.decide.target(current_bytes,
415 page.record.required(for_decision, Nat.is_gt(current_count, 0n)),
416 Nat.min(U32.to_nat(StorageBytes.TARGET_BLOCK_BYTES()), max_bytes), max_bytes))
417
418def page.next.prepare(
419 entries: List<&1, ByteMemTable.Entry>,
420 current_count: Nat,
421 current_bytes: Nat,
422 max_bytes: Nat
423) -> List<&1, ByteMemTable.Entry> & ByteBlockBuilder.Decision:
424 match entries:
425 case Nil{}: (Nil{}, ByteBlockBuilder.Append{current_bytes})
426 case Con{entry, tail}:
427 page.next.cloned(tail, current_count, current_bytes, max_bytes,
428 ByteMemTable.entry.clone(entry))
429
430def page.build.go(
431 fuel: Nat,
432 builder: PageBuild,
433 pending: List<&1, ByteMemTable.Entry> & ByteBlockBuilder.Decision,
434 +max_bytes: Nat
435) -> Result<&1, &1, Error, PagePlan>:
436 match fuel:
437 case 0n:
438 match builder:
439 case PageBuild{current, _, _, completed, completed_count}:
440 match pending:
441 case (Nil{}, _):
442 Done{page.finish(completed, current, completed_count)}
443 case _: Fail{TooLarge{}}
444 case 1n+rest:
445 match builder:
446 case PageBuild{current, current_bytes, +current_count, completed,
447 completed_count}:
448 match pending:
449 case (Nil{}, _):
450 Done{page.finish(completed, current, completed_count)}
451 case (Con{entry, tail}, decision):
452 match decision:
453 case ByteBlockBuilder.RecordTooLarge{}: Fail{TooLarge{}}
454 case ByteBlockBuilder.Append{+used_bytes}:
455 page.build.go(rest,
456 PageBuild{Con{entry, current}, used_bytes,
457 Nat.add(current_count, 1n), completed, completed_count},
458 page.next.prepare(tail, Nat.add(current_count, 1n), used_bytes,
459 max_bytes), max_bytes)
460 case ByteBlockBuilder.StartNextBlock{+record_bytes}:
461 page.build.go(rest,
462 page.rollover(entry, current, completed, completed_count,
463 record_bytes),
464 page.next.prepare(tail, 1n, record_bytes, max_bytes), max_bytes)
465
466def page.build.measured(
467 measured: List<&1, ByteMemTable.Entry> & Nat,
468 +max_bytes: Nat
469) -> Result<&1, &1, Error, PagePlan>:
470 match measured:
471 case (entries, count):
472 page.build.go(count, PageBuild{Nil{}, 0n, 0n, Nil{}, 0n},
473 page.next.prepare(entries, 0n, 0n, max_bytes), max_bytes)
474
475def page.build.limit(
476 entries: List<&1, ByteMemTable.Entry>,
477 max_bytes: Nat
478) ->
479 Result<&1, &1, Error, PagePlan>:
480 page.build.measured(ByteMemTable.count.keep(entries), max_bytes)
481
482def pages.count(result: Result<&1, &1, Error, PagePlan>) -> Nat:
483 match result:
484 case Fail{_}: 0n
485 case Done{PagePlan{_, count}}: count
486
487# Projects the byte builder's admission result to the codec preflight contract.
488def block.payload.decision(decision: ByteBlockBuilder.Decision) -> Bool:
489 match decision:
490 case ByteBlockBuilder.Append{_}: True{}
491 case _: False{}
492
493# Checks record payload plus WAL framing through the shared byte block policy.
494def block.payload.allowed(record_bytes: Nat) -> Bool:
495 block.payload.decision(ByteBlockBuilder.decide(0n,
496 Nat.add(40n, record_bytes), U32.to_nat(StorageBytes.MAX_BLOCK_BYTES())))
497
498def file.block.raw.parts(
499 headers: Bytes.Bytes & Bytes.Bytes,
500 frames: Bytes.Bytes & Bytes.Bytes
501) -> Result<&1, &1, Error, Bytes.Bytes>:
502 match headers:
503 case (header_output, header_hash):
504 match frames:
505 case (frame_output, frame_hash):
506 Done{Bytes.concat([header_output, frame_output,
507 SHA.sha256_packed_bytes(Bytes.concat([header_hash, frame_hash]))])}
508
509def file.block.raw(
510 frame: Bytes.Bytes,
511 count: U32,
512 payload_len: U32
513) -> Result<&1, &1, Error, Bytes.Bytes>:
514 file.block.raw.parts(clone.split(Bytes.concat([u32.bytes(payload_len),
515 u32.bytes(count)])), clone.split(frame))
516
517def encode.page.frame.block(
518 count: U32,
519 lengths: Bytes.Bytes & U32
520) -> Result<&1, &1, Error, Bytes.Bytes>:
521 match lengths:
522 case (frame, +frame_length):
523 file.block.raw(frame, count, frame_length)
524
525def encode.page.frame.result(
526 frame: Result<&1, &1, Wal.Error, Bytes.Bytes>,
527 count: U32
528) -> Result<&1, &1, Error, Bytes.Bytes>:
529 match frame:
530 case Fail{_}: Fail{TooLarge{}}
531 case Done{bytes}:
532 encode.page.frame.block(count, Bytes.length(bytes))
533
534def encode.page.block.measured(
535 measured: Bytes.Bytes & U32,
536 checksum: Bytes.Bytes,
537 count: U32,
538 separator: Bytes.Bytes,
539 separator_length: U32
540) -> Result<&1, &1, Error, EncodedBlock>:
541 match measured:
542 case (bytes, +byte_length):
543 Done{EncodedBlock{bytes, byte_length, checksum, count, separator,
544 separator_length}}
545
546def encode.page.block.output(
547 copies: Bytes.Bytes & Bytes.Bytes,
548 byte_length: U32,
549 count: U32,
550 separator: Bytes.Bytes,
551 separator_length: U32
552) -> Result<&1, &1, Error, EncodedBlock>:
553 match copies:
554 case (output, checksum):
555 encode.page.block.measured((output, byte_length), checksum, count,
556 separator, separator_length)
557
558def encode.page.block.digest(
559 measured: Bytes.Bytes & U32,
560 count: U32,
561 separator: Bytes.Bytes,
562 separator_length: U32
563) -> Result<&1, &1, Error, EncodedBlock>:
564 match measured:
565 case (bytes, +byte_length):
566 encode.page.block.output(Bytes.slice(bytes, (byte_length - 32 : U32), 32),
567 byte_length, count, separator, separator_length)
568
569def encode.page.block(
570 result: Result<&1, &1, Error, Bytes.Bytes>,
571 count: U32,
572 separator: Bytes.Bytes,
573 separator_length: U32
574) -> Result<&1, &1, Error, EncodedBlock>:
575 match result:
576 case Fail{error}: Fail{error}
577 case Done{bytes}:
578 encode.page.block.digest(Bytes.length(bytes), count, separator,
579 separator_length)
580
581def encode.page.frame.counted(
582 encoded: List<&1, ByteMemTable.Entry>,
583 +count: Nat,
584 separator: Bytes.Bytes,
585 separator_length: U32
586) -> Result<&1, &1, Error, EncodedBlock>:
587 encode.page.block(encode.page.frame.result(
588 ByteWal.encode(ByteWal.Batch{mutations(encoded)}),
589 U32.from_nat(count)), U32.from_nat(count), separator, separator_length)
590
591def encode.page.frame.key.measured(
592 separator: Bytes.Bytes,
593 encoded: List<&1, ByteMemTable.Entry>,
594 +count: Nat,
595 measured: Bytes.Bytes & U32
596) -> Result<&1, &1, Error, EncodedBlock>:
597 match measured:
598 case (_, separator_length):
599 encode.page.frame.counted(encoded, count, separator, separator_length)
600
601def encode.page.frame.key.copied(
602 copied_key: Bytes.Bytes & Bytes.Bytes,
603 encoded: List<&1, ByteMemTable.Entry>,
604 +count: Nat
605) -> Result<&1, &1, Error, EncodedBlock>:
606 match copied_key:
607 case (separator, measurement):
608 encode.page.frame.key.measured(separator, encoded, count,
609 Bytes.length(measurement))
610
611def encode.page.frame(
612 counted: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
613) -> Result<&1, &1, Error, EncodedBlock>:
614 match counted:
615 case (sized, encoded):
616 match sized:
617 case Con{ByteMemTable.Entry{key, _}, tail}:
618 encode.page.frame.key.copied(ByteMemTable.bytes.clone(key), encoded,
619 Nat.add(List.length(&1, ByteMemTable.Entry, tail), 1n))
620 case Nil{}: Fail{Malformed{}}
621
622def encode.page(page: List<&1, ByteMemTable.Entry>) -> Result<&1, &1, Error, EncodedBlock>:
623 encode.page.frame(ByteMemTable.entries.clone(page))
624
625def encode.pages.prepare.cloned(
626 tail: List<&1, List<&1, ByteMemTable.Entry>>,
627 copies: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
628) -> List<&1, List<&1, ByteMemTable.Entry>> &
629 Result<&1, &1, Error, EncodedBlock>:
630 match copies:
631 case (to_encode, to_retain):
632 (Con{to_retain, tail}, encode.page(to_encode))
633
634def encode.pages.prepare(
635 pages: List<&1, List<&1, ByteMemTable.Entry>>
636) -> List<&1, List<&1, ByteMemTable.Entry>> &
637 Result<&1, &1, Error, EncodedBlock>:
638 match pages:
639 case Nil{}: (Nil{}, Done{EncodedBlock{Bytes.new(0), 0, Bytes.new(0), 0,
640 Bytes.new(0), 0}})
641 case Con{page, tail}:
642 encode.pages.prepare.cloned(tail, ByteMemTable.entries.clone(page))
643
644def encode.blocks.go(
645 fuel: Nat,
646 pending: List<&1, List<&1, ByteMemTable.Entry>> & Result<&1, &1, Error, EncodedBlock>,
647 reversed: List<&1, EncodedBlock>
648) -> Result<&1, &1, Error, List<&1, EncodedBlock>>:
649 match fuel:
650 case 0n:
651 match pending:
652 case (Nil{}, _): Done{List.reverse(&1, EncodedBlock, reversed)}
653 case _: Fail{Malformed{}}
654 case 1n+rest:
655 match pending:
656 case (Nil{}, _): Done{List.reverse(&1, EncodedBlock, reversed)}
657 case (Con{_, _}, Fail{error}): Fail{error}
658 case (Con{_, tail}, Done{block}):
659 encode.blocks.go(rest, encode.pages.prepare(tail),
660 Con{block, reversed})
661
662def encode.blocks.result(
663 count: U32,
664 result: Result<&1, &1, Error, List<&1, EncodedBlock>>
665) -> Result<&1, &1, Error, EncodedPages>:
666 match result:
667 case Fail{error}: Fail{error}
668 case Done{blocks}: Done{EncodedPages{blocks, count}}
669
670def encode.blocks(plan: PagePlan) -> Result<&1, &1, Error, EncodedPages>:
671 match plan:
672 case PagePlan{pages, +count}:
673 encode.blocks.result(U32.from_nat(count),
674 encode.blocks.go(count, encode.pages.prepare(pages), Nil{}))
675
676def blocks.collect.go(
677 fuel: Nat,
678 fits: Bool,
679 remaining: List<&1, EncodedBlock>,
680 parts: List<&1, Bytes.Bytes>,
681 checksums: List<&1, Bytes.Bytes>,
682 index_entries: List<&1, Bytes.Bytes>,
683 separators: List<&1, Bytes.Bytes>,
684 +offset: Nat,
685 +separator_offset: Nat,
686 block_count: U32,
687 entry_count: U32
688) -> Result<&1, &1, Error, BlocksCollected>:
689 match fuel:
690 case 0n:
691 match fits:
692 case False{}: Fail{TooLarge{}}
693 case True{}:
694 match remaining:
695 case Nil{}:
696 +count = U32.to_nat(block_count)
697 Done{BlocksCollected{
698 Bytes.concat(List.reverse(&1, Bytes.Bytes, parts)),
699 Bytes.concat(List.reverse(&1, Bytes.Bytes, checksums)),
700 Bytes.concat([index.magic(), u32.bytes(U32.from_nat(count)),
701 Bytes.concat([Bytes.concat(List.reverse(&1, Bytes.Bytes,
702 index_entries)), Bytes.concat(List.reverse(&1, Bytes.Bytes,
703 separators))])]),
704 U32.from_nat(count), entry_count}}
705 case _: Fail{Malformed{}}
706 case 1n+rest:
707 match fits:
708 case False{}: Fail{TooLarge{}}
709 case True{}:
710 match remaining:
711 case Nil{}:
712 +count = U32.to_nat(block_count)
713 Done{BlocksCollected{
714 Bytes.concat(List.reverse(&1, Bytes.Bytes, parts)),
715 Bytes.concat(List.reverse(&1, Bytes.Bytes, checksums)),
716 Bytes.concat([index.magic(), u32.bytes(U32.from_nat(count)),
717 Bytes.concat([Bytes.concat(List.reverse(&1, Bytes.Bytes,
718 index_entries)), Bytes.concat(List.reverse(&1, Bytes.Bytes,
719 separators))])]),
720 U32.from_nat(count), entry_count}}
721 case Con{EncodedBlock{bytes, +byte_length, checksum, count,
722 separator, +separator_length}, tail}:
723 +length_nat = U32.to_nat(byte_length)
724 +separator_length_nat = U32.to_nat(separator_length)
725 blocks.collect.go(rest,
726 fits && Nat.is_le(Nat.add(offset, length_nat), 4294967295n) &&
727 Nat.is_le(Nat.add(separator_offset, separator_length_nat),
728 4294967295n), tail, Con{bytes, parts},
729 Con{checksum, checksums},
730 Con{Bytes.concat([u32.bytes(U32.from_nat(offset)),
731 u32.bytes(byte_length),
732 u32.bytes(U32.from_nat(separator_offset)),
733 u32.bytes(separator_length)]), index_entries},
734 Con{separator, separators}, Nat.add(offset, length_nat),
735 Nat.add(separator_offset, separator_length_nat),
736 (block_count + 1 : U32), (entry_count + count : U32))
737
738def blocks.collect(
739 encoded: EncodedPages
740) -> Result<&1, &1, Error, BlocksCollected>:
741 match encoded:
742 case EncodedPages{blocks, count}:
743 blocks.collect.go(U32.to_nat(count), True{}, blocks, Nil{}, Nil{}, Nil{},
744 Nil{}, 20n, 0n, 0, 0)
745
746def file.multi.output.measured(
747 header_output: Bytes.Bytes,
748 blocks: Bytes.Bytes,
749 checksums: Bytes.Bytes,
750 measured_index: Bytes.Bytes & U32,
751 index_hash: Bytes.Bytes,
752 header_hash: Bytes.Bytes,
753 +total: U32
754) -> Result<&1, &1, Error, Bytes.Bytes>:
755 match measured_index:
756 case (index_output, index_length):
757 Done{Bytes.concat([header_output, blocks, index_output,
758 footer.magic(), u32.bytes(total), u32.bytes(index_length),
759 SHA.sha256_packed_bytes(Bytes.concat([header_hash, checksums,
760 index_hash]))])}
761
762def file.multi.output(
763 header_parts: Bytes.Bytes & Bytes.Bytes,
764 blocks: Bytes.Bytes,
765 checksums: Bytes.Bytes,
766 index_parts: Bytes.Bytes & Bytes.Bytes,
767 +total: U32
768) -> Result<&1, &1, Error, Bytes.Bytes>:
769 match header_parts:
770 case (header_output, header_hash):
771 match index_parts:
772 case (index_output, index_hash):
773 file.multi.output.measured(header_output, blocks, checksums,
774 Bytes.length(index_output), index_hash, header_hash, total)
775
776def file.multi.finish(
777 level: U32,
778 collected: BlocksCollected
779) -> Result<&1, &1, Error, Bytes.Bytes>:
780 match collected:
781 case BlocksCollected{blocks, checksums, index_bytes, block_count, entry_count}:
782 +total = U32.to_nat(entry_count)
783 file.multi.output(clone.split(Bytes.concat([header.magic(),
784 u32.bytes(level), u32.bytes(U32.from_nat(total)),
785 u32.bytes(block_count)])), blocks, checksums,
786 clone.split(index_bytes), U32.from_nat(total))
787
788def encode.multi.collected(
789 result: Result<&1, &1, Error, BlocksCollected>,
790 level: U32
791) -> Result<&1, &1, Error, Bytes.Bytes>:
792 match result:
793 case Fail{error}: Fail{error}
794 case Done{collected}: file.multi.finish(level, collected)
795
796def encode.multi.pages(
797 result: Result<&1, &1, Error, EncodedPages>,
798 level: U32
799) -> Result<&1, &1, Error, Bytes.Bytes>:
800 match result:
801 case Fail{error}: Fail{error}
802 case Done{pages}:
803 encode.multi.collected(blocks.collect(pages), level)
804
805def encode.multi.plan(
806 result: Result<&1, &1, Error, PagePlan>,
807 level: U32
808) -> Result<&1, &1, Error, Bytes.Bytes>:
809 match result:
810 case Fail{error}: Fail{error}
811 case Done{plan}:
812 encode.multi.pages(encode.blocks(plan), level)
813
814def encode.multi.level(
815 entries: List<&1, ByteMemTable.Entry>,
816 level: U32,
817 valid: Bool,
818 max_bytes: Nat
819) -> Result<&1, &1, Error, Bytes.Bytes>:
820 match valid:
821 case False{}: Fail{InvalidLevel{}}
822 case True{}:
823 encode.multi.plan(page.build.limit(entries, max_bytes), level)
824
825def encode.multi.order(
826 entries: List<&1, ByteMemTable.Entry>,
827 level: U32,
828 ordered: Bool,
829 max_bytes: Nat
830) -> Result<&1, &1, Error, Bytes.Bytes>:
831 match ordered:
832 case False{}: Fail{InvalidOrder{}}
833 case True{}:
834 +level_nat = U32.to_nat(level)
835 encode.multi.level(entries, U32.from_nat(level_nat),
836 Nat.is_le(level_nat, 255n), max_bytes)
837
838def encode.multi.validated(
839 copies: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>,
840 level: U32,
841 max_bytes: Nat
842) -> Result<&1, &1, Error, Bytes.Bytes>:
843 match copies:
844 case (to_validate, to_encode):
845 encode.multi.order(to_encode, level,
846 ByteMemTable.ordered.strict(to_validate), max_bytes)
847
848def encode_file.limit(
849 entries: List<&1, ByteMemTable.Entry>,
850 level: U32,
851 max_bytes: Nat
852) -> Result<&1, &1, Error, Bytes.Bytes>:
853 encode.multi.validated(ByteMemTable.entries.clone(entries), level, max_bytes)
854
855# Encodes the validated byte table in the canonical single-block format.
856def encode_file(
857 entries: List<&1, ByteMemTable.Entry>,
858 level: U32
859) -> Result<&1, &1, Error, Bytes.Bytes>:
860 encode_file.limit(entries, level, U32.to_nat(StorageBytes.MAX_BLOCK_BYTES()))
861
862# Applies each decoded byte mutation while rejecting count mismatches.
863def decode.apply(
864 fuel: Nat,
865 table: ByteMemTable.Table,
866 mutations: List<&1, ByteWal.Mutation>
867) -> Result<&1, &1, Error, ByteMemTable.Table>:
868 match fuel:
869 case 0n:
870 match mutations:
871 case Nil{}:
872 Done{table}
873 case Con{_, _}:
874 Fail{Malformed{}}
875 case 1n+rest:
876 match mutations:
877 case Nil{}:
878 Fail{Malformed{}}
879 case Con{mutation, tail}:
880 match mutation:
881 case ByteWal.Put{key, value}:
882 decode.apply(rest, ByteMemTable.put(table, key, value), tail)
883 case ByteWal.Delete{key}:
884 decode.apply(rest, ByteMemTable.del(table, key), tail)
885
886# Decodes the inner checksummed frame and applies exactly its declared count.
887def decode.frame(
888 total: U32,
889 result: Result<&1, &1, Wal.Error, ByteWal.Batch>
890) -> Result<&1, &1, Error, ByteMemTable.Table>:
891 match result:
892 case Fail{Wal.ChecksumMismatch{}}:
893 Fail{ChecksumMismatch{}}
894 case Fail{_}:
895 Fail{Malformed{}}
896 case Done{ByteWal.Batch{mutations}}:
897 decode.apply(U32.to_nat(total), ByteMemTable.empty(), mutations)
898
899# Converts a root comparison over raw header, checksum, and index bytes into a result.
900def parse.header.entries.valid(
901 cursor: Bytes.Cursor,
902 +size: U32,
903 +level: U32,
904 +total: U32,
905 +payload_len: U32,
906 +entries: U32,
907 valid: Bool
908) -> Result<&1, &1, Error, PayloadWords>:
909 match valid:
910 case True{}:
911 Done{PayloadWords{cursor, size, level, total, payload_len, entries}}
912 case False{}:
913 Fail{Malformed{}}
914def parse.header.entries(
915 pair: Bytes.Cursor & Maybe<&2, U32>,
916 +size: U32,
917 +level: U32,
918 +total: U32,
919 +payload_len: U32
920) -> Result<&1, &1, Error, PayloadWords>:
921 match pair:
922 case (cursor, Some{+entries}):
923 parse.header.entries.valid(cursor, size, level, total, payload_len,
924 entries, U32.is_eq(entries, total) &&
925 U32.is_le(payload_len, StorageBytes.MAX_BLOCK_BYTES()) &&
926 U32.is_eq(size, (116 + payload_len : U32)))
927 case _:
928 Fail{Malformed{}}
929def parse.header.payload(
930 pair: Bytes.Cursor & Maybe<&2, U32>,
931 size: U32,
932 level: U32,
933 total: U32
934) -> Result<&1, &1, Error, PayloadWords>:
935 match pair:
936 case (cursor, Some{payload_len}):
937 parse.header.entries(Bytes.Cursor.u32be(cursor), size, level, total,
938 payload_len)
939 case _:
940 Fail{Malformed{}}
941def parse.header.blocks(
942 pair: Bytes.Cursor & Maybe<&2, U32>,
943 size: U32,
944 level: U32,
945 total: U32
946) -> Result<&1, &1, Error, PayloadWords>:
947 match pair:
948 case (cursor, Some{1}):
949 parse.header.payload(Bytes.Cursor.u32be(cursor), size, level, total)
950 case _:
951 Fail{Malformed{}}
952def parse.header.total(
953 pair: Bytes.Cursor & Maybe<&2, U32>,
954 size: U32,
955 level: U32
956) -> Result<&1, &1, Error, PayloadWords>:
957 match pair:
958 case (cursor, Some{total}):
959 parse.header.blocks(Bytes.Cursor.u32be(cursor), size, level, total)
960 case _:
961 Fail{Malformed{}}
962def parse.header.level.valid(
963 pair: Bytes.Cursor & Maybe<&2, U32>,
964 size: U32,
965 level: U32,
966 valid: Bool
967) -> Result<&1, &1, Error, PayloadWords>:
968 match valid:
969 case False{}:
970 Fail{InvalidLevel{}}
971 case True{}:
972 parse.header.total(pair, size, level)
973def parse.header.level(
974 pair: Bytes.Cursor & Maybe<&2, U32>,
975 size: U32
976) -> Result<&1, &1, Error, PayloadWords>:
977 match pair:
978 case (cursor, Some{+level}):
979 parse.header.level.valid(Bytes.Cursor.u32be(cursor), size, level,
980 U32.is_le(level, 255))
981 case _:
982 Fail{Malformed{}}
983def parse.header.magic.second(
984 pair: Bytes.Cursor & Maybe<&2, U32>,
985 size: U32
986) -> Result<&1, &1, Error, PayloadWords>:
987 match pair:
988 case (cursor, Some{1295274752}):
989 parse.header.level(Bytes.Cursor.u32be(cursor), size)
990 case _:
991 Fail{Malformed{}}
992def parse.header.words(
993 pair: Bytes.Cursor & Maybe<&2, U32>,
994 size: U32
995) -> Result<&1, &1, Error, PayloadWords>:
996 match pair:
997 case (cursor, Some{1297697875}):
998 parse.header.magic.second(Bytes.Cursor.u32be(cursor), size)
999 case _:
1000 Fail{Malformed{}}
1001def parse.header.length(pair: Bytes.Bytes & U32) -> Result<&1, &1, Error, PayloadWords>:
1002 match pair:
1003 case (source, size):
1004 parse.header.words(Bytes.Cursor.u32be(Bytes.Cursor.new(source)), size)
1005def parse.header(file: Bytes.Bytes) -> Result<&1, &1, Error, PayloadWords>:
1006 parse.header.length(Bytes.length(file))
1007
1008def parse.canonical.comparison(
1009 entries: List<&1, ByteMemTable.Entry>,
1010 comparison: Bytes.Bytes & Bytes.Bytes & Bool
1011) -> Result<&1, &1, Error, ByteMemTable.Table>:
1012 match comparison:
1013 case (_, _, True{}):
1014 Done{ByteMemTable.Table{entries}}
1015 case (_, _, False{}):
1016 Fail{ChecksumMismatch{}}
1017
1018def parse.canonical.encoded(
1019 source: Bytes.Bytes,
1020 entries: List<&1, ByteMemTable.Entry>,
1021 result: Result<&1, &1, Error, Bytes.Bytes>
1022) -> Result<&1, &1, Error, ByteMemTable.Table>:
1023 match result:
1024 case Fail{error}:
1025 Fail{error}
1026 case Done{expected}:
1027 parse.canonical.comparison(entries, Bytes.eq(source, expected))
1028
1029def parse.canonical.encode(
1030 source: Bytes.Bytes,
1031 pair: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>,
1032 +level: U32
1033) -> Result<&1, &1, Error, ByteMemTable.Table>:
1034 match pair:
1035 case (entries, encoded):
1036 parse.canonical.encoded(source, entries, encode_file(encoded, level))
1037
1038def parse.canonical.table(
1039 source: Bytes.Bytes,
1040 table: ByteMemTable.Table,
1041 +level: U32
1042) -> Result<&1, &1, Error, ByteMemTable.Table>:
1043 match table:
1044 case ByteMemTable.Table{entries}:
1045 parse.canonical.encode(source, entries.clone(List.reverse(&1,
1046 ByteMemTable.Entry, entries)), level)
1047def parse.block.frame.decoded(
1048 result: Result<&1, &1, Error, ByteMemTable.Table>,
1049 source: Bytes.Bytes,
1050 +level: U32
1051) -> Result<&1, &1, Error, ByteMemTable.Table>:
1052 match result:
1053 case Fail{error}:
1054 Fail{error}
1055 case Done{table}:
1056 parse.canonical.table(source, table, level)
1057def parse.block.frame(
1058 pair: Bytes.Bytes & Bytes.Bytes,
1059 total: U32,
1060 +level: U32
1061) -> Result<&1, &1, Error, ByteMemTable.Table>:
1062 match pair:
1063 case (source, frame):
1064 parse.block.frame.decoded(decode.frame(total, ByteWal.decode(frame)),
1065 source, level)
1066
1067def parse.block.payload(
1068 cursor: Bytes.Cursor,
1069 level: U32,
1070 total: U32,
1071 payload_len: U32
1072) -> Result<&1, &1, Error, ByteMemTable.Table>:
1073 match cursor:
1074 case Bytes.Cursor{bytes, +pos, _, _}:
1075 parse.block.frame(Bytes.slice(bytes, pos, payload_len), total, level)
1076
1077def parse.file.header(
1078 result: Result<&1, &1, Error, PayloadWords>
1079) -> Result<&1, &1, Error, ByteMemTable.Table>:
1080 match result:
1081 case Fail{error}:
1082 Fail{error}
1083 case Done{PayloadWords{cursor, _, level, total, payload_len, _}}:
1084 parse.block.payload(cursor, level, total, payload_len)
1085def parse.file(file: Bytes.Bytes) -> Result<&1, &1, Error, ByteMemTable.Table>:
1086 parse.file.header(parse.header(file))
1087
1088def parse.block.envelope.comparison(
1089 table: ByteMemTable.Table,
1090 comparison: Bytes.Bytes & Bytes.Bytes & Bool
1091) -> Result<&1, &1, Error, ByteMemTable.Table>:
1092 match comparison:
1093 case (_, _, True{}): Done{table}
1094 case (_, _, False{}): Fail{ChecksumMismatch{}}
1095
1096def parse.block.envelope.copies(
1097 table: Result<&1, &1, Error, ByteMemTable.Table>,
1098 block: Bytes.Bytes,
1099 encoded: Result<&1, &1, Error, Bytes.Bytes>
1100) -> Result<&1, &1, Error, ByteMemTable.Table>:
1101 match table:
1102 case Fail{error}: Fail{error}
1103 case Done{table}:
1104 match encoded:
1105 case Fail{error}: Fail{error}
1106 case Done{expected}:
1107 parse.block.envelope.comparison(table, Bytes.eq(block, expected))
1108
1109def parse.block.envelope(
1110 table: Result<&1, &1, Error, ByteMemTable.Table>,
1111 block: Bytes.Bytes,
1112 frame: Bytes.Bytes,
1113 count: U32,
1114 payload_length: U32
1115) -> Result<&1, &1, Error, ByteMemTable.Table>:
1116 parse.block.envelope.copies(table, block,
1117 file.block.raw(frame, count, payload_length))
1118
1119def parse.block.frame.encoded(
1120 source: Bytes.Bytes,
1121 block: Bytes.Bytes,
1122 frame: Bytes.Bytes,
1123 count: U32,
1124 payload_length: U32,
1125 entries: List<&1, ByteMemTable.Entry>,
1126 encoded: Result<&1, &1, Error, Bytes.Bytes>
1127) -> Result<&1, &1, Error, ByteMemTable.Table>:
1128 parse.block.envelope(parse.canonical.encoded(source, entries, encoded),
1129 block, frame, count, payload_length)
1130
1131def parse.block.frame.wal(
1132 source: Bytes.Bytes,
1133 block: Bytes.Bytes,
1134 frame: Bytes.Bytes,
1135 count: U32,
1136 payload_length: U32,
1137 entries: List<&1, ByteMemTable.Entry>,
1138 encoded: Result<&1, &1, Wal.Error, Bytes.Bytes>
1139) -> Result<&1, &1, Error, ByteMemTable.Table>:
1140 match encoded:
1141 case Fail{_}: Fail{Malformed{}}
1142 case Done{bytes}:
1143 parse.block.frame.encoded(source, block, frame, count, payload_length,
1144 entries, Done{bytes})
1145
1146def parse.block.frame.order(
1147 ordered: Bool,
1148 frames: Bytes.Bytes & Bytes.Bytes,
1149 block: Bytes.Bytes,
1150 count: U32,
1151 payload_length: U32,
1152 copies: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
1153) -> Result<&1, &1, Error, ByteMemTable.Table>:
1154 match ordered:
1155 case False{}: Fail{InvalidOrder{}}
1156 case True{}:
1157 match frames:
1158 case (frame_source, frame_block):
1159 match copies:
1160 case (for_validation, for_encoding):
1161 parse.block.frame.wal(frame_source, block, frame_block,
1162 count, payload_length, for_validation,
1163 ByteWal.encode(ByteWal.Batch{mutations(for_encoding)}))
1164
1165def parse.block.frame.copies(
1166 ordered: Bool,
1167 frames: Bytes.Bytes & Bytes.Bytes,
1168 block: Bytes.Bytes,
1169 count: U32,
1170 payload_length: U32,
1171 copies: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
1172) -> Result<&1, &1, Error, ByteMemTable.Table>:
1173 parse.block.frame.order(ordered, frames, block, count, payload_length, copies)
1174
1175def parse.block.table.ordered(
1176 copies: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>,
1177 block: Bytes.Bytes,
1178 frames: Bytes.Bytes & Bytes.Bytes,
1179 count: U32,
1180 payload_length: U32
1181) -> Result<&1, &1, Error, ByteMemTable.Table>:
1182 match copies:
1183 case (for_order, for_encoding):
1184 parse.block.frame.copies(ByteMemTable.ordered.strict(for_order), frames,
1185 block, count, payload_length,
1186 ByteMemTable.entries.clone(for_encoding))
1187
1188def parse.block.table(
1189 result: Result<&1, &1, Error, ByteMemTable.Table>,
1190 block: Bytes.Bytes,
1191 frames: Bytes.Bytes & Bytes.Bytes,
1192 count: U32,
1193 payload_length: U32
1194) -> Result<&1, &1, Error, ByteMemTable.Table>:
1195 match result:
1196 case Fail{error}: Fail{error}
1197 case Done{ByteMemTable.Table{entries}}:
1198 parse.block.table.ordered(
1199 ByteMemTable.entries.clone(List.reverse(&1, ByteMemTable.Entry, entries)),
1200 block, frames, count, payload_length)
1201
1202def parse.block.decoded(
1203 result: Result<&1, &1, Wal.Error, ByteWal.Batch>,
1204 block: Bytes.Bytes,
1205 frames: Bytes.Bytes & Bytes.Bytes,
1206 counts: Nat & Nat,
1207 payload_length: U32
1208) -> Result<&1, &1, Error, ByteMemTable.Table>:
1209 match result:
1210 case Fail{Wal.ChecksumMismatch{}}: Fail{ChecksumMismatch{}}
1211 case Fail{_}: Fail{Malformed{}}
1212 case Done{ByteWal.Batch{mutations}}:
1213 match counts:
1214 case (decode_count, stored_count):
1215 parse.block.table(decode.apply(decode_count,
1216 ByteMemTable.empty(), mutations), block, frames,
1217 U32.from_nat(stored_count), payload_length)
1218
1219def parse.block.frames.decode(
1220 frames: Bytes.Bytes & Bytes.Bytes,
1221 block: Bytes.Bytes,
1222 count: U32,
1223 payload_length: U32
1224) -> Result<&1, &1, Error, ByteMemTable.Table>:
1225 match frames:
1226 case (for_decode, for_compare):
1227 parse.block.decoded(ByteWal.decode(for_decode), block,
1228 clone.split(for_compare),
1229 QueuePolicy.nat.clone.pair(U32.to_nat(count)), payload_length)
1230
1231
1232def parse.block.frames(
1233 pair: Bytes.Bytes & Bytes.Bytes,
1234 count: U32,
1235 payload_length: U32
1236) -> Result<&1, &1, Error, ByteMemTable.Table>:
1237 match pair:
1238 case (block, frame):
1239 parse.block.frames.decode(clone.split(frame), block, count,
1240 payload_length)
1241def parse.block.count.copy(
1242 block: Bytes.Bytes,
1243 count: U32,
1244 pair: Nat & Nat
1245) -> Result<&1, &1, Error, ByteMemTable.Table>:
1246 match pair:
1247 case (slice_length, check_length):
1248 parse.block.frames(Bytes.slice(block, 8, U32.from_nat(slice_length)),
1249 count, U32.from_nat(check_length))
1250
1251def parse.block.count(
1252 pair: Bytes.Bytes & Maybe<&2, U32>,
1253 payload_length: U32
1254) -> Result<&1, &1, Error, ByteMemTable.Table>:
1255 match pair:
1256 case (block, Some{count}):
1257 parse.block.count.copy(block, count,
1258 QueuePolicy.nat.clone.pair(U32.to_nat(payload_length)))
1259 case (_, None{}):
1260 Fail{Malformed{}}
1261
1262def parse.block.length.valid(
1263 block: Bytes.Bytes,
1264 payload_length: U32,
1265 valid: Bool
1266) -> Result<&1, &1, Error, ByteMemTable.Table>:
1267 match valid:
1268 case False{}: Fail{Malformed{}}
1269 case True{}:
1270 parse.block.count(Bytes.get.u32be(block, 4), payload_length)
1271
1272def parse.block.length.check(
1273 block: Bytes.Bytes,
1274 size: U32,
1275 copies: Nat & Nat
1276) -> Result<&1, &1, Error, ByteMemTable.Table>:
1277 match copies:
1278 case (count_length, validation_length):
1279 parse.block.length.valid(block, U32.from_nat(count_length),
1280 U32.is_eq(size, (40 + U32.from_nat(validation_length) : U32)))
1281
1282
1283def parse.block.length(
1284 measured: Bytes.Bytes & U32,
1285 payload: Bytes.Bytes & Maybe<&2, U32>
1286) -> Result<&1, &1, Error, ByteMemTable.Table>:
1287 match measured:
1288 case (block, +size):
1289 match payload:
1290 case (_, Some{payload_length}):
1291 parse.block.length.check(block, size,
1292 QueuePolicy.nat.clone.pair(U32.to_nat(payload_length)))
1293 case _:
1294 Fail{Malformed{}}
1295def parse_block.payload.copies(
1296 copies: Bytes.Bytes & Bytes.Bytes,
1297 payload_length: U32
1298) -> Result<&1, &1, Error, ByteMemTable.Table>:
1299 match copies:
1300 case (measured, block):
1301 parse.block.length(Bytes.length(measured), (block, Some{payload_length}))
1302
1303def parse_block.payload(
1304 pair: Bytes.Bytes & Maybe<&2, U32>
1305) -> Result<&1, &1, Error, ByteMemTable.Table>:
1306 match pair:
1307 case (block, Some{payload_length}):
1308 parse_block.payload.copies(clone.split(block), payload_length)
1309 case _:
1310 Fail{Malformed{}}
1311
1312def measure.block.checksum.comparison(
1313 frame: Bytes.Bytes,
1314 comparison: Bytes.Bytes & Bytes.Bytes & Bool
1315) -> Bytes.Bytes & Bool:
1316 match comparison:
1317 case (_, _, valid): (frame, valid)
1318
1319def measure.block.checksum.claimed(
1320 pair: Bytes.Bytes & Bytes.Bytes,
1321 expected: Bytes.Bytes
1322) -> Bytes.Bytes & Bool:
1323 match pair:
1324 case (frame, claimed):
1325 measure.block.checksum.comparison(frame, Bytes.eq(claimed, expected))
1326
1327def measure.block.checksum.prefix(
1328 pair: Bytes.Bytes & Bytes.Bytes,
1329 +digest_offset: U32
1330) -> Bytes.Bytes & Bool:
1331 match pair:
1332 case (frame, prefix):
1333 measure.block.checksum.claimed(
1334 Bytes.slice(frame, digest_offset, 32),
1335 SHA.sha256_packed_bytes(prefix))
1336
1337def measure.block.checksum(
1338 block: Bytes.Bytes,
1339 +payload_length: U32
1340) -> Bytes.Bytes & Bool:
1341 +digest_offset = U32.add(8, payload_length)
1342 measure.block.checksum.prefix(Bytes.slice(block, 0, digest_offset),
1343 digest_offset)
1344
1345def measure.block.inner.valid(
1346 frame: Bytes.Bytes,
1347 query: Bytes.Bytes,
1348 +frame_length: U32,
1349 +record_count: U32,
1350 +payload_length: U32,
1351 valid: Bool
1352) -> Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>:
1353 match valid:
1354 case False{}: Fail{Wal.Malformed{}}
1355 case True{}:
1356 ByteWal.measure.lookup.bounded(frame, query, 8, frame_length,
1357 record_count, payload_length)
1358
1359def measure.block.inner.count(
1360 pair: Bytes.Bytes & Maybe<&2, U32>,
1361 +frame_length: U32,
1362 +record_count: U32,
1363 +payload_length: U32,
1364 query: Bytes.Bytes
1365) -> Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>:
1366 match pair:
1367 case (frame, Some{inner_count}):
1368 measure.block.inner.valid(frame, query, frame_length, record_count,
1369 payload_length, U32.is_eq(inner_count, record_count))
1370 case (_, None{}): Fail{Wal.Malformed{}}
1371
1372def measure.block.inner.length(
1373 pair: Bytes.Bytes & Maybe<&2, U32>,
1374 +record_count: U32,
1375 +payload_length: U32,
1376 query: Bytes.Bytes
1377) -> Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>:
1378 match pair:
1379 case (frame, Some{frame_length}):
1380 measure.block.inner.count(Bytes.get.u32be(frame, 12), frame_length,
1381 record_count, payload_length, query)
1382 case (_, None{}): Fail{Wal.Malformed{}}
1383
1384def measure.block.inner(
1385 frame: Bytes.Bytes,
1386 query: Bytes.Bytes,
1387 +record_count: U32,
1388 +payload_length: U32
1389) -> Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>:
1390 measure.block.inner.length(Bytes.get.u32be(frame, 8), record_count,
1391 payload_length, query)
1392
1393def measure.block.checksum.result(
1394 pair: Bytes.Bytes & Bool,
1395 query: Bytes.Bytes,
1396 +record_count: U32,
1397 +payload_length: U32
1398) -> Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>:
1399 match pair:
1400 case (block, valid):
1401 match valid:
1402 case True{}:
1403 measure.block.inner(block, query, record_count, payload_length)
1404 case False{}: Fail{Wal.ChecksumMismatch{}}
1405
1406def measure.block.valid(
1407 frame: Bytes.Bytes,
1408 query: Bytes.Bytes,
1409 +record_count: U32,
1410 +payload_length: U32,
1411 +_total: U32,
1412 valid: Bool
1413) -> Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>:
1414 match valid:
1415 case False{}: Fail{Wal.Malformed{}}
1416 case True{}:
1417 measure.block.checksum.result(
1418 measure.block.checksum(frame, payload_length), query, record_count,
1419 payload_length)
1420
1421def measure.block.count(
1422 pair: Bytes.Bytes & Maybe<&2, U32>,
1423 +payload_length: U32,
1424 +total: U32,
1425 query: Bytes.Bytes
1426) -> Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>:
1427 match pair:
1428 case (frame, Some{record_count}):
1429 measure.block.valid(frame, query, record_count, payload_length, total,
1430 U32.is_le(payload_length, StorageBytes.MAX_BLOCK_BYTES()) &&
1431 U32.is_eq(total, U32.add(40, payload_length)))
1432 case (_, None{}): Fail{Wal.Malformed{}}
1433
1434def measure.block.payload(
1435 pair: Bytes.Bytes & Maybe<&2, U32>,
1436 +total: U32,
1437 query: Bytes.Bytes
1438) -> Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>:
1439 match pair:
1440 case (block, Some{payload_length}):
1441 measure.block.count(Bytes.get.u32be(block, 4), payload_length, total,
1442 query)
1443 case (_, None{}): Fail{Wal.Malformed{}}
1444
1445def measure.block.total(
1446 pair: Bytes.Bytes & U32,
1447 query: Bytes.Bytes
1448) -> Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>:
1449 match pair:
1450 case (block, total):
1451 measure.block.payload(Bytes.get.u32be(block, 0), total, query)
1452
1453# Handles measure block.
1454def measure_block(
1455 block: Bytes.Bytes,
1456 query: Bytes.Bytes
1457) -> Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>:
1458 measure.block.total(Bytes.length(block), query)
1459
1460# Validates one complete generation-four page without requiring its file index.
1461def parse_block(block: Bytes.Bytes) -> Result<&1, &1, Error, ByteMemTable.Table>:
1462 parse_block.payload(Bytes.get.u32be(block, 0))
1463
1464def parse.multi.header.blocks(
1465 pair: Bytes.Bytes & Maybe<&2, U32>,
1466 level: Nat,
1467 total: Nat
1468) -> Result<&1, &1, Error, MultiHeader>:
1469 match pair:
1470 case (file, Some{block_count}):
1471 Done{MultiHeader{file, U32.from_nat(level), U32.from_nat(total),
1472 block_count}}
1473 case _:
1474 Fail{Malformed{}}
1475
1476def parse.multi.header.total(
1477 pair: Bytes.Bytes & Maybe<&2, U32>,
1478 level: Nat
1479) -> Result<&1, &1, Error, MultiHeader>:
1480 match pair:
1481 case (file, Some{total}):
1482 parse.multi.header.blocks(Bytes.get.u32be(file, 16), level,
1483 U32.to_nat(total))
1484 case _:
1485 Fail{Malformed{}}
1486
1487def parse.multi.header.level(
1488 pair: Bytes.Bytes & Maybe<&2, U32>
1489) -> Result<&1, &1, Error, MultiHeader>:
1490 match pair:
1491 case (file, Some{level}):
1492 parse.multi.header.total(Bytes.get.u32be(file, 12), U32.to_nat(level))
1493 case _:
1494 Fail{Malformed{}}
1495
1496def parse.multi.header(
1497 file: Bytes.Bytes
1498) -> Result<&1, &1, Error, MultiHeader>:
1499 parse.multi.header.level(Bytes.get.u32be(file, 8))
1500
1501def parse.stream.header.rules(
1502 header: MultiHeader,
1503 valid_magic: Bool,
1504 valid_fields: Bool
1505) -> Result<&1, &1, Error, MultiHeader>:
1506 match valid_magic:
1507 case False{}: Fail{Malformed{}}
1508 case True{}:
1509 match valid_fields:
1510 case False{}: Fail{Malformed{}}
1511 case True{}: Done{header}
1512
1513def parse.stream.header.fields(
1514 header: MultiHeader,
1515 valid_magic: Bool
1516) -> Result<&1, &1, Error, MultiHeader>:
1517 match header:
1518 case MultiHeader{_, +level, _, +block_count}:
1519 parse.stream.header.rules(header, valid_magic,
1520 U32.is_le(level, 255) && U32.is_lt(0, block_count))
1521
1522
1523def parse.stream.header.valid(
1524 result: Result<&1, &1, Error, MultiHeader>,
1525 valid: Bool
1526) -> Result<&1, &1, Error, MultiHeader>:
1527 match result:
1528 case Fail{error}: Fail{error}
1529 case Done{header}:
1530 parse.stream.header.fields(header, valid)
1531
1532
1533def parse.stream.header.magic(
1534 source: Bytes.Bytes,
1535 comparison: Bytes.Bytes & Bytes.Bytes & Bool
1536) -> Result<&1, &1, Error, MultiHeader>:
1537 match comparison:
1538 case (_, _, valid):
1539 parse.stream.header.valid(parse.multi.header(source), valid)
1540
1541def parse.stream.header.slice(
1542 source: Bytes.Bytes,
1543 signature: Bytes.Bytes & Bytes.Bytes
1544) -> Result<&1, &1, Error, MultiHeader>:
1545 match signature:
1546 case (_, candidate):
1547 parse.stream.header.magic(source, Bytes.eq(header.magic(), candidate))
1548
1549def parse.stream.header.copies(
1550 copies: Bytes.Bytes & Bytes.Bytes
1551) -> Result<&1, &1, Error, MultiHeader>:
1552 match copies:
1553 case (source, input):
1554 parse.stream.header.slice(source, Bytes.slice(input, 0, 8))
1555
1556# Validates the fixed generation-four header before opening a block stream.
1557def parse_stream_header(source: Bytes.Bytes) -> Result<&1, &1, Error, MultiHeader>:
1558 parse.stream.header.copies(clone.split(source))
1559
1560def parse_stream_header.accepted(
1561 result: Result<&1, &1, Error, MultiHeader>
1562) -> Bool:
1563 match result:
1564 case Done{_}: True{}
1565 case Fail{_}: False{}
1566
1567def parse.stream.finish.footer.compare(
1568 comparison: Bytes.Bytes & Bytes.Bytes & Bool
1569) -> Result<&1, &1, Error, Unit>:
1570 match comparison:
1571 case (_, _, True{}): Done{Unit{}}
1572 case (_, _, False{}): Fail{ChecksumMismatch{}}
1573
1574def parse.stream.finish.footer(
1575 index_valid: Bool,
1576 header_hash: Bytes.Bytes,
1577 checksums: Bytes.Bytes,
1578 index_hash: Bytes.Bytes,
1579 footer: Bytes.Bytes,
1580 +total: U32
1581) -> Result<&1, &1, Error, Unit>:
1582 match index_valid:
1583 case False{}: Fail{Malformed{}}
1584 case True{}:
1585 parse.stream.finish.footer.compare(Bytes.eq(footer,
1586 Bytes.concat([footer.magic(), u32.bytes(total),
1587 SHA.sha256_packed_bytes(Bytes.concat([header_hash, checksums,
1588 index_hash]))])))
1589
1590def parse.stream.finish.index.root(
1591 header: Bytes.Bytes,
1592 checksums: Bytes.Bytes,
1593 footer: Bytes.Bytes,
1594 +total: U32,
1595 index_hash: Bytes.Bytes,
1596 comparison: Bytes.Bytes & Bytes.Bytes & Bool
1597) -> Result<&1, &1, Error, Unit>:
1598 match comparison:
1599 case (_, _, valid):
1600 parse.stream.finish.footer(valid, header, checksums, index_hash, footer,
1601 total)
1602
1603def parse.stream.finish.index.copies(
1604 header: Bytes.Bytes,
1605 checksums: Bytes.Bytes,
1606 entries: Bytes.Bytes,
1607 index_copies: Bytes.Bytes & Bytes.Bytes,
1608 footer: Bytes.Bytes,
1609 +total: U32,
1610 +block_count: U32
1611) -> Result<&1, &1, Error, Unit>:
1612 match index_copies:
1613 case (for_compare, for_hash):
1614 parse.stream.finish.index.root(header, checksums, footer, total,
1615 for_hash, Bytes.eq(for_compare,
1616 Bytes.concat([index.magic(), u32.bytes(block_count), entries])))
1617
1618def parse.stream.finish.index(
1619 header: Bytes.Bytes,
1620 checksums: Bytes.Bytes,
1621 entries: Bytes.Bytes,
1622 index: Bytes.Bytes,
1623 footer: Bytes.Bytes,
1624 +total: U32,
1625 +block_count: U32
1626) -> Result<&1, &1, Error, Unit>:
1627 parse.stream.finish.index.copies(header, checksums, entries,
1628 clone.split(index), footer, total, block_count)
1629# Verifies serialized index bytes and the generation-four root footer.
1630def parse_stream_finish(
1631 header: Bytes.Bytes,
1632 checksums: Bytes.Bytes,
1633 entries: Bytes.Bytes,
1634 index: Bytes.Bytes,
1635 footer: Bytes.Bytes,
1636 total: U32,
1637 block_count: U32
1638) -> Result<&1, &1, Error, Unit>:
1639 parse.stream.finish.index(header, checksums, entries, index, footer,
1640 total, block_count)
1641
1642def parse_stream_finish.accepted(
1643 result: Result<&1, &1, Error, Unit>
1644) -> Bool:
1645 match result:
1646 case Done{Unit{}}: True{}
1647 case Fail{_}: False{}
1648
1649def parse.multi.table.updated(
1650 result: Result<&1, &1, Error, ByteMemTable.Table>,
1651 file: Bytes.Bytes,
1652 index_offset: Nat,
1653 next_block: Nat,
1654 block_count: Nat,
1655 level: U32,
1656 total: Nat,
1657 decoded: Nat,
1658 record_count: Nat
1659) -> Result<&1, &1, Error, MultiRead>:
1660 match result:
1661 case Fail{error}: Fail{error}
1662 case Done{table}:
1663 Done{MultiRead{file, index_offset, Nat.add(next_block, 1n),
1664 block_count, level, total, Nat.add(decoded, record_count), table}}
1665
1666def parse.multi.block.apply(
1667 record_counts: Nat & Nat,
1668 table: ByteMemTable.Table,
1669 mutations: List<&1, ByteWal.Mutation>,
1670 file: Bytes.Bytes,
1671 index_offset: Nat,
1672 next_block: Nat,
1673 block_count: Nat,
1674 level: U32,
1675 total: Nat,
1676 decoded: Nat
1677) -> Result<&1, &1, Error, MultiRead>:
1678 match record_counts:
1679 case (for_decode, for_total):
1680 parse.multi.table.updated(decode.apply(for_decode, table, mutations),
1681 file, index_offset, next_block, block_count, level, total, decoded,
1682 for_total)
1683
1684def parse.multi.block.decoded(
1685 result: Result<&1, &1, Wal.Error, ByteWal.Batch>,
1686 file: Bytes.Bytes,
1687 index_offset: Nat,
1688 next_block: Nat,
1689 block_count: Nat,
1690 level: U32,
1691 total: Nat,
1692 decoded: Nat,
1693 table: ByteMemTable.Table,
1694 record_count: Nat
1695) -> Result<&1, &1, Error, MultiRead>:
1696 match result:
1697 case Fail{Wal.ChecksumMismatch{}}: Fail{ChecksumMismatch{}}
1698 case Fail{_}: Fail{Malformed{}}
1699 case Done{ByteWal.Batch{mutations}}:
1700 parse.multi.block.apply(QueuePolicy.nat.clone.pair(record_count),
1701 table, mutations, file, index_offset, next_block, block_count, level,
1702 total, decoded)
1703
1704def parse.multi.block.frame(
1705 pair: Bytes.Bytes & Bytes.Bytes,
1706 file: Bytes.Bytes,
1707 index_offset: Nat,
1708 next_block: Nat,
1709 block_count: Nat,
1710 level: U32,
1711 total: Nat,
1712 decoded: Nat,
1713 table: ByteMemTable.Table,
1714 record_count: Nat
1715) -> Result<&1, &1, Error, MultiRead>:
1716 match pair:
1717 case (_, frame):
1718 parse.multi.block.decoded(ByteWal.decode(frame), file, index_offset,
1719 next_block, block_count, level, total, decoded, table, record_count)
1720
1721def parse.multi.block.count(
1722 pair: Bytes.Bytes & Maybe<&2, U32>,
1723 file: Bytes.Bytes,
1724 index_offset: Nat,
1725 next_block: Nat,
1726 block_count: Nat,
1727 level: U32,
1728 total: Nat,
1729 decoded: Nat,
1730 table: ByteMemTable.Table,
1731 payload_length: U32
1732) -> Result<&1, &1, Error, MultiRead>:
1733 match pair:
1734 case (block, Some{record_count}):
1735 +payload_nat = U32.to_nat(payload_length)
1736 +record_nat = U32.to_nat(record_count)
1737 parse.multi.block.frame(Bytes.slice(block, 8, U32.from_nat(payload_nat)),
1738 file, index_offset, next_block, block_count, level, total, decoded,
1739 table, record_nat)
1740 case _:
1741 Fail{Malformed{}}
1742
1743def parse.multi.block.payload.length(
1744 pair: Bytes.Bytes & Maybe<&2, U32>,
1745 file: Bytes.Bytes,
1746 index_offset: Nat,
1747 next_block: Nat,
1748 block_count: Nat,
1749 level: U32,
1750 total: Nat,
1751 decoded: Nat,
1752 table: ByteMemTable.Table
1753) -> Result<&1, &1, Error, MultiRead>:
1754 match pair:
1755 case (block, Some{payload_length}):
1756 parse.multi.block.count(Bytes.get.u32be(block, 4), file,
1757 index_offset, next_block, block_count, level, total, decoded, table,
1758 payload_length)
1759 case _:
1760 Fail{Malformed{}}
1761
1762def parse.multi.block.payload(
1763 pair: Bytes.Bytes & Bytes.Bytes,
1764 index_offset: Nat,
1765 next_block: Nat,
1766 block_count: Nat,
1767 level: U32,
1768 total: Nat,
1769 decoded: Nat,
1770 table: ByteMemTable.Table
1771) -> Result<&1, &1, Error, MultiRead>:
1772 match pair:
1773 case (file, block):
1774 parse.multi.block.payload.length(Bytes.get.u32be(block, 0), file,
1775 index_offset, next_block, block_count, level, total, decoded, table)
1776
1777def parse.multi.block.offset(
1778 pair: Bytes.Bytes & Maybe<&2, U32>,
1779 index_offset: Nat,
1780 next_block: Nat,
1781 block_count: Nat,
1782 level: U32,
1783 total: Nat,
1784 decoded: Nat,
1785 table: ByteMemTable.Table
1786) -> Result<&1, &1, Error, MultiRead>:
1787 match pair:
1788 case (file, Some{block_offset}):
1789 parse.multi.block.payload(Bytes.slice(file, block_offset, 4294967295),
1790 index_offset, next_block, block_count, level, total, decoded, table)
1791 case _:
1792 Fail{Malformed{}}
1793
1794def parse.multi.step.address(
1795 offsets: Nat & Nat,
1796 positions: Nat & Nat,
1797 file: Bytes.Bytes,
1798 block_count: Nat,
1799 level: U32,
1800 total: Nat,
1801 decoded: Nat,
1802 table: ByteMemTable.Table
1803) -> Result<&1, &1, Error, MultiRead>:
1804 match offsets:
1805 case (index_offset, retained_offset):
1806 match positions:
1807 case (next_block, retained_block):
1808 parse.multi.block.offset(Bytes.get.u32be(file,
1809 U32.from_nat(Nat.add(Nat.add(index_offset, 8n),
1810 Nat.mul(next_block, 16n)))), retained_offset, retained_block,
1811 block_count, level, total, decoded, table)
1812
1813def parse.multi.step(state: MultiRead) -> Result<&1, &1, Error, MultiRead>:
1814 match state:
1815 case MultiRead{file, index_offset, next_block, block_count, level,
1816 total, decoded, table}:
1817 parse.multi.step.address(QueuePolicy.nat.clone.pair(index_offset),
1818 QueuePolicy.nat.clone.pair(next_block), file, block_count, level,
1819 total, decoded, table)
1820
1821def parse.multi.complete(
1822 source: Bytes.Bytes,
1823 table: ByteMemTable.Table,
1824 level: U32,
1825 count_matches: Bool
1826) -> Result<&1, &1, Error, ByteMemTable.Table>:
1827 match count_matches:
1828 case False{}: Fail{Malformed{}}
1829 case True{}: parse.canonical.table(source, table, level)
1830
1831def parse.multi.go(
1832 fuel: Nat,
1833 source: Bytes.Bytes,
1834 pending: Result<&1, &1, Error, MultiRead>
1835) -> Result<&1, &1, Error, ByteMemTable.Table>:
1836 match fuel:
1837 case 0n:
1838 match pending:
1839 case Fail{error}: Fail{error}
1840 case Done{state}:
1841 match state:
1842 case MultiRead{_, _, _, _, level, total, decoded, table}:
1843 parse.multi.complete(source, table, level,
1844 Nat.is_eq(total, decoded))
1845 case 1n+rest:
1846 match pending:
1847 case Fail{error}: Fail{error}
1848 case Done{state}:
1849 parse.multi.go(rest, source, parse.multi.step(state))
1850
1851def parse.multi.sized.start(
1852 block_counts: Nat & Nat,
1853 source: Bytes.Bytes,
1854 file: Bytes.Bytes,
1855 index_offset: Nat,
1856 entries: Nat,
1857 level: Nat
1858) -> Result<&1, &1, Error, ByteMemTable.Table>:
1859 match block_counts:
1860 case (fuel, block_count):
1861 parse.multi.go(fuel, source, Done{MultiRead{file, index_offset,
1862 0n, block_count, U32.from_nat(level), entries, 0n,
1863 ByteMemTable.empty()}})
1864
1865def parse.multi.sized.valid(
1866 source: Bytes.Bytes,
1867 file: Bytes.Bytes,
1868 +blocks: Nat,
1869 +entries: Nat,
1870 +level: Nat,
1871 +size: Nat,
1872 index_length: Nat,
1873 valid: Bool
1874) -> Result<&1, &1, Error, ByteMemTable.Table>:
1875 match valid:
1876 case False{}: Fail{Malformed{}}
1877 case True{}:
1878 +index_size = Nat.add(44n, index_length)
1879 +index_offset = Nat.sub(size, index_size)
1880 parse.multi.sized.start(QueuePolicy.nat.clone.pair(blocks), source,
1881 file, index_offset, entries, level)
1882
1883def parse.multi.sized.index_length(
1884 source: Bytes.Bytes,
1885 file: Bytes.Bytes,
1886 +blocks: Nat,
1887 +entries: Nat,
1888 +level: Nat,
1889 +size: Nat,
1890 result: Bytes.Bytes & Maybe<&2, U32>
1891) -> Result<&1, &1, Error, ByteMemTable.Table>:
1892 match result:
1893 case (_, Some{length}):
1894 +index_length = U32.to_nat(length)
1895 +minimum_index_length = Nat.add(8n, Nat.mul(blocks, 16n))
1896 +index_size = Nat.add(44n, index_length)
1897 parse.multi.sized.valid(source, file, blocks, entries, level, size,
1898 index_length,
1899 Nat.is_le(minimum_index_length, index_length) &&
1900 Nat.is_le(index_size, size) &&
1901 Nat.is_le(20n, Nat.sub(size, index_size)))
1902 case (_, None{}): Fail{Malformed{}}
1903
1904def parse.multi.sized.index.read(
1905 source: Bytes.Bytes,
1906 files: Bytes.Bytes & Bytes.Bytes,
1907 +blocks: Nat,
1908 +entries: Nat,
1909 +level: Nat,
1910 +size: Nat
1911) -> Result<&1, &1, Error, ByteMemTable.Table>:
1912 match files:
1913 case (file, index_file):
1914 parse.multi.sized.index_length(source, file, blocks, entries, level,
1915 size, Bytes.get.u32be(index_file,
1916 U32.from_nat(Nat.sub(size, 36n))))
1917
1918def parse.multi.sized(
1919 source: Bytes.Bytes,
1920 measured: Bytes.Bytes & U32,
1921 level: U32,
1922 total: U32,
1923 block_count: U32
1924) -> Result<&1, &1, Error, ByteMemTable.Table>:
1925 match measured:
1926 case (file, +size):
1927 +blocks = U32.to_nat(block_count)
1928 +entries = U32.to_nat(total)
1929 +level_value = U32.to_nat(level)
1930 +size_nat = U32.to_nat(size)
1931 parse.multi.sized.index.read(source, clone.split(file), blocks,
1932 entries, level_value, size_nat)
1933
1934def parse.multi.header.level.valid(
1935 source: Bytes.Bytes,
1936 file: Bytes.Bytes,
1937 level: U32,
1938 total: U32,
1939 block_count: U32,
1940 valid: Bool
1941) -> Result<&1, &1, Error, ByteMemTable.Table>:
1942 match valid:
1943 case False{}: Fail{InvalidLevel{}}
1944 case True{}:
1945 parse.multi.sized(source, Bytes.length(file), level, total, block_count)
1946
1947def parse.multi.header.valid(
1948 source: Bytes.Bytes,
1949 result: Result<&1, &1, Error, MultiHeader>
1950) -> Result<&1, &1, Error, ByteMemTable.Table>:
1951 match result:
1952 case Fail{error}: Fail{error}
1953 case Done{MultiHeader{file, level, total, block_count}}:
1954 +level_nat = U32.to_nat(level)
1955 parse.multi.header.level.valid(source, file,
1956 U32.from_nat(level_nat), total, block_count,
1957 Nat.is_le(level_nat, 255n))
1958
1959def parse.multi.copies(
1960 copies: Bytes.Bytes & Bytes.Bytes
1961) -> Result<&1, &1, Error, ByteMemTable.Table>:
1962 match copies:
1963 case (source, input):
1964 parse.multi.header.valid(source, parse.multi.header(input))
1965
1966# Parses and validates one canonical generation-four SST block.
1967def parse_file(file: Bytes.Bytes) -> Result<&1, &1, Error, ByteMemTable.Table>:
1968 parse.multi.copies(clone.split(file))