src/ByteSstFile.bend

mylsm-lsm-store@0.6.0.1 · 1,968 lines · 179 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.1/src/ByteWal.bend as ByteWal
4import mylsm-lsm-store@0.6.0.1/src/StorageBytes.bend as StorageBytes
5import mylsm-lsm-store@0.6.0.1/src/scan/ByteBlockBuilder.bend as ByteBlockBuilder
6import mylsm-lsm-store@0.6.0.1/src/coordinator/QueuePolicy.bend as QueuePolicy
7import mylsm-lsm-store@0.6.0.1/src/Wal.bend as Wal
8import mylsm-lsm-store@0.6.0.1/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))