src/SstFile.bend

mylsm-lsm-store@0.6.0.1 · 1,811 lines · 173 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/MemTable.bend as MemTable
3import mylsm-lsm-store@0.6.0.1/src/Keys.bend as Keys
4import mylsm-lsm-store@0.6.0.1/src/Sstable.bend as Sstable
5import mylsm-lsm-store@0.6.0.1/src/StorageBytes.bend as StorageBytes
6import mylsm-lsm-store@0.6.0.1/src/hub_sha/sha256.bend as SHA
7import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
8
9# Errors returned by the v3 encoder and strict parser.
10type Error is Data:
11  InvalidLevel{}
12  InvalidOrder{}
13  TooLarge{}
14  Malformed{}
15  InvalidUtf8{}
16  ChecksumMismatch{}
17
18# Encoder progression through the header, blocks, and footer.
19type Phase is Data:
20  Header{}
21  Blocks{}
22  Finished{}
23
24# Serialized block paired with its stored digest.
25type EncodedBlock is Type:
26  Block{bytes: Bytes.Bytes, digest: Bytes.Bytes}
27
28# State that yields the encoded file in bounded chunks.
29type Encoder is Type:
30  Enc{
31    phase: Phase,
32    chunks: List<&1, Bytes.Bytes>,
33    level: U32,
34    total: U32,
35    block_count: U32,
36    footer: Bytes.Bytes
37  }
38
39# Validated level and counts read from the v3 header.
40type SstHeader is Data:
41  SstHdr{level: U32, total: U32, block_count: U32}
42
43# Streaming parser stage. A completed stage carries only decoded table state.
44# Bounded chunk accumulator; the final parser rejects aggregate overflow before joining.
45type Decoder is Type:
46  Dec{chunks: List<&1, Bytes.Bytes>, total: U32, invalid: Bool}
47
48# Result of applying one validated block to file-level counts.
49type BlockDecision is Type:
50  BlockRejected{error: Error}
51  BlockOutOfBounds{}
52  BlockContinue{cursor: Bytes.Cursor, entries_left: U32,
53    previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>,
54    hashes: List<&1, Bytes.Bytes>}
55  BlockFinish{cursor: Bytes.Cursor, entries: List<&2, MemTable.Entry>,
56    hashes: List<&1, Bytes.Bytes>}
57
58def bytes.clone.pair(
59  +len: U32,
60  copied: Array<U32> & Array<U32>
61) -> Bytes.Bytes & Bytes.Bytes:
62  match copied:
63    case (left, right):
64      (Bytes.Bytes{len, left}, Bytes.Bytes{len, right})
65
66def bytes.clone.split(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes:
67  match bytes:
68    case Bytes.Bytes{+len, buf}:
69      bytes.clone.pair(len, Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
70
71def bytes.clone.first(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
72  match pair:
73    case (left, _):
74      left
75
76def bytes.clone(bytes: Bytes.Bytes) -> Bytes.Bytes:
77  bytes.clone.first(bytes.clone.split(bytes))
78
79# Construct a packed u32 in network byte order.
80def u32_bytes(value: U32) -> Bytes.Bytes:
81  Bytes.set.u32be(Bytes.new(4), 0, value)
82
83# Construct a packed tag byte.
84def u8_bytes(value: U32) -> Bytes.Bytes:
85  Bytes.set(Bytes.new(1), 0, value)
86
87# Unwrap a fixed valid hexadecimal constant.
88def hex_bytes.fin(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
89  match value:
90    case Some{bytes}:
91      bytes
92    case None{}:
93      Bytes.new(0)
94
95# Encode one mutation as a tagged, length-delimited record.
96def record.value.finish.checked(
97  key_bytes: Bytes.Bytes,
98  +key_len: U32,
99  value_bytes: Bytes.Bytes,
100  +value_len: U32,
101  valid: Bool
102) -> Result<&1, &1, Error, Bytes.Bytes>:
103  match valid:
104    case False{}:
105      Fail{TooLarge{}}
106    case True{}:
107      Done{Bytes.concat([u8_bytes(0), u32_bytes(key_len), key_bytes,
108        u32_bytes(value_len), value_bytes])}
109
110def record.value.finish.valid(
111  key_bytes: Bytes.Bytes,
112  +key_len: U32,
113  value_bytes: Bytes.Bytes,
114  +value_len: U32,
115  key_fits: Bool
116) -> Result<&1, &1, Error, Bytes.Bytes>:
117  match key_fits:
118    case False{}:
119      Fail{TooLarge{}}
120    case True{}:
121      record.value.finish.checked(key_bytes, key_len, value_bytes, value_len,
122        U32.is_le(value_len, (StorageBytes.MAX_RECORD_BYTES() - 9 - key_len : U32)))
123
124def record.value.finish.val(
125  key_bytes: Bytes.Bytes,
126  +key_len: U32,
127  value_pair: Bytes.Bytes & U32
128) -> Result<&1, &1, Error, Bytes.Bytes>:
129  match value_pair:
130    case (value_bytes, +value_len):
131      record.value.finish.valid(key_bytes, key_len, value_bytes, value_len,
132        U32.is_le(key_len, (StorageBytes.MAX_RECORD_BYTES() - 9 : U32)))
133
134def record.value.finish(
135  key_pair: Bytes.Bytes & U32,
136  value_pair: Bytes.Bytes & U32
137) -> Result<&1, &1, Error, Bytes.Bytes>:
138  match key_pair:
139    case (key_bytes, key_len):
140      record.value.finish.val(key_bytes, key_len, value_pair)
141
142def record.value.fin(
143  key_pair: Bytes.Bytes & U32,
144  result: Result<&1, &1, U32 & String, Bytes.Bytes>
145) -> Result<&1, &1, Error, Bytes.Bytes>:
146  match result:
147    case Fail{_}:
148      Fail{TooLarge{}}
149    case Done{value_bytes}:
150      record.value.finish(key_pair, Bytes.length(value_bytes))
151
152def record.value(
153  key_bytes: Bytes.Bytes,
154  value_result: Result<&1, &1, U32 & String, Bytes.Bytes>
155) -> Result<&1, &1, Error, Bytes.Bytes>:
156  record.value.fin(Bytes.length(key_bytes), value_result)
157
158def record.key.delete.checked(
159  key_bytes: Bytes.Bytes,
160  +key_len: U32,
161  valid: Bool
162) -> Result<&1, &1, Error, Bytes.Bytes>:
163  match valid:
164    case False{}:
165      Fail{TooLarge{}}
166    case True{}:
167      Done{Bytes.concat([u8_bytes(1), u32_bytes(key_len), key_bytes])}
168
169def record.key.delete(key_pair: Bytes.Bytes & U32) -> Result<&1, &1, Error, Bytes.Bytes>:
170  match key_pair:
171    case (key_bytes, +key_len):
172      record.key.delete.checked(key_bytes, key_len,
173        U32.is_le(key_len, (StorageBytes.MAX_RECORD_BYTES() - 5 : U32)))
174
175def record.key(
176  key_result: Result<&1, &1, U32 & String, Bytes.Bytes>,
177  val: Maybe<&2, String>
178) -> Result<&1, &1, Error, Bytes.Bytes>:
179  match key_result val:
180    case Fail{_} _:
181      Fail{TooLarge{}}
182    case Done{key_bytes} None{}:
183      record.key.delete(Bytes.length(key_bytes))
184    case Done{key_bytes} Some{value}:
185      record.value(key_bytes, StorageBytes.from_string(value))
186
187# Encode one checked put or delete record.
188def encode_record(entry: MemTable.Entry) -> Result<&1, &1, Error, Bytes.Bytes>:
189  match entry:
190    case MemTable.Entry{key, val}:
191      record.key(StorageBytes.from_string(key), val)
192
193def entries.strict.tail(
194  tail: List<&2, MemTable.Entry>,
195  current: String,
196  ordered: Bool
197) -> Bool:
198  match tail ordered:
199    case Nil{} True{}:
200      True{}
201    case Nil{} False{}:
202      False{}
203    case Con{_entry, _rest} False{}:
204      False{}
205    case Con{MemTable.Entry{+key, val}, rest} True{}:
206      entries.strict.tail(rest, key, Keys.lt(current, key))
207
208def entries.strict(entries: List<&2, MemTable.Entry>) -> Bool:
209  match entries:
210    case Nil{}:
211      True{}
212    case Con{MemTable.Entry{key, val}, rest}:
213      entries.strict.tail(rest, key, True{})
214
215# Result of measuring a record against the target block size.
216type BlockMeasure is Data:
217  InvalidRecord{}
218  Measured{length: U32, fits: Bool}
219
220# Measure an already encoded record for byte-bounded block packing.
221def blocks.measure.length(+length: U32, +size: U32) -> BlockMeasure:
222  Measured{length, U32.is_eq(size, 0) ||
223    U32.is_le((size + length : U32), StorageBytes.MAX_BLOCK_BYTES())}
224
225def blocks.measure.valid(length: U32, +size: U32, valid: Bool) -> BlockMeasure:
226  match valid:
227    case False{}:
228      InvalidRecord{}
229    case True{}:
230      blocks.measure.length(length, size)
231
232def blocks.measure.delete(key_length: Maybe<&2, U32>, +size: U32) -> BlockMeasure:
233  match key_length:
234    case None{}:
235      InvalidRecord{}
236    case Some{+length}:
237      blocks.measure.valid((5 + length : U32), size,
238        U32.is_le(length, (StorageBytes.MAX_RECORD_BYTES() - 5 : U32)))
239
240def blocks.measure.put.value(
241  +key_length: U32,
242  value_length: Maybe<&2, U32>,
243  +size: U32
244) -> BlockMeasure:
245  match value_length:
246    case None{}:
247      InvalidRecord{}
248    case Some{+length}:
249      blocks.measure.valid((9 + key_length + length : U32), size,
250        U32.is_le(key_length, (StorageBytes.MAX_RECORD_BYTES() - 9 : U32)) &&
251        U32.is_le(length, (StorageBytes.MAX_RECORD_BYTES() - 9 - key_length : U32)))
252
253def blocks.measure.put(
254  key_length: Maybe<&2, U32>,
255  value_length: Maybe<&2, U32>,
256  +size: U32
257) -> BlockMeasure:
258  match key_length:
259    case None{}:
260      InvalidRecord{}
261    case Some{+length}:
262      blocks.measure.put.value(length, value_length, size)
263
264def blocks.measure.entry(entry: MemTable.Entry, +size: U32) -> BlockMeasure:
265  match entry:
266    case MemTable.Entry{key, value}:
267      match value:
268        case None{}:
269          blocks.measure.delete(StorageBytes.from_string.size.begin(key), size)
270        case Some{text}:
271          blocks.measure.put(StorageBytes.from_string.size.begin(key),
272            StorageBytes.from_string.size.begin(text), size)
273
274def blocks.measure.next(
275  entries: List<&2, MemTable.Entry>,
276  +size: U32
277) -> BlockMeasure:
278  match entries:
279    case Nil{}:
280      InvalidRecord{}
281    case Con{MemTable.Entry{key, val}, _}:
282      blocks.measure.entry(MemTable.Entry{key, val}, size)
283
284# Finish the current block list, which is stored in reverse order.
285def blocks.pack.finish(
286  current: List<&2, MemTable.Entry>,
287  acc: List<&2, List<&2, MemTable.Entry>>
288) -> List<&2, List<&2, MemTable.Entry>>:
289  match current:
290    case Nil{}:
291      List.reverse(&2, List<&2, MemTable.Entry>, acc)
292    case Con{_, _}:
293      List.reverse(&2, List<&2, MemTable.Entry>,
294        Con{List.reverse(&2, MemTable.Entry, current), acc})
295
296# Pack records into blocks capped by encoded payload bytes, preserving order.
297def blocks.pack.go(
298  entries: List<&2, MemTable.Entry>,
299  current: List<&2, MemTable.Entry>,
300  +size: U32,
301  acc: List<&2, List<&2, MemTable.Entry>>,
302  measured: BlockMeasure
303) -> List<&2, List<&2, MemTable.Entry>>:
304  match entries:
305    case Nil{}:
306      blocks.pack.finish(current, acc)
307    case Con{entry, +rest}:
308      match current:
309        case Nil{}:
310          match measured:
311            case InvalidRecord{}:
312              blocks.pack.go(rest, Con{entry, Nil{}}, StorageBytes.MAX_BLOCK_BYTES(), acc,
313                blocks.measure.next(rest, StorageBytes.MAX_BLOCK_BYTES()))
314            case Measured{+length, fits}:
315              blocks.pack.go(rest, Con{entry, Nil{}}, length, acc,
316                blocks.measure.next(rest, length))
317        case Con{_, _}:
318          match measured:
319            case InvalidRecord{}:
320              blocks.pack.go(rest, Con{entry, Nil{}}, StorageBytes.MAX_BLOCK_BYTES(),
321                Con{List.reverse(&2, MemTable.Entry, current), acc},
322                blocks.measure.next(rest, StorageBytes.MAX_BLOCK_BYTES()))
323            case Measured{+length, False{}}:
324              blocks.pack.go(rest, Con{entry, Nil{}}, length,
325                Con{List.reverse(&2, MemTable.Entry, current), acc},
326                blocks.measure.next(rest, length))
327            case Measured{+length, True{}}:
328              blocks.pack.go(rest, Con{entry, current}, (size + length : U32), acc,
329                blocks.measure.next(rest, (size + length : U32)))
330
331def blocks.pack(entries: List<&2, MemTable.Entry>) -> List<&2, List<&2, MemTable.Entry>>:
332  match entries:
333    case Nil{}:
334      Nil{}
335    case Con{+entry, _}:
336      blocks.pack.go(entries, Nil{}, 0, Nil{},
337        blocks.measure.entry(entry, 0))
338
339# Continue encoding a record list after the current record has been checked.
340def records.step(
341  rest: List<&2, MemTable.Entry>,
342  current: Result<&1, &1, Error, Bytes.Bytes>,
343  acc: List<&1, Bytes.Bytes>
344) -> Result<&1, &1, Error, List<&1, Bytes.Bytes>>:
345  match rest current:
346    case Nil{} Fail{error}:
347      Fail{error}
348    case Nil{} Done{bytes}:
349      Done{List.reverse(&1, Bytes.Bytes, bytes <> acc)}
350    case Con{_entry, _tail} Fail{error}:
351      Fail{error}
352    case Con{entry, tail} Done{bytes}:
353      records.step(tail, encode_record(entry), bytes <> acc)
354
355# Encode each entry independently, retaining only the record fragments for one block.
356def records(entries: List<&2, MemTable.Entry>) -> Result<&1, &1, Error, List<&1, Bytes.Bytes>>:
357  match entries:
358    case Nil{}:
359      Done{Nil{}}
360    case Con{entry, +rest}:
361          records.step(rest, encode_record(entry), Nil{})
362
363# Build one v3 block and its digest over the two header words and payload.
364def block.fin.digest.parts(
365  +count: U32,
366  payload: Bytes.Bytes,
367  +payload_len: U32,
368  pair: Bytes.Bytes & Bytes.Bytes
369) -> EncodedBlock:
370  match pair:
371    case (output_digest, stored_digest):
372      Block{Bytes.concat([u32_bytes(payload_len), u32_bytes(count), payload,
373        output_digest]), stored_digest}
374
375def block.fin.digest(
376  +count: U32,
377  pair: Bytes.Bytes & Bytes.Bytes,
378  payload: Bytes.Bytes,
379  +payload_len: U32
380) -> EncodedBlock:
381  match pair:
382    case (_, _):
383      block.fin.digest.parts(count, payload, payload_len, pair)
384
385def block.fin.payload(
386  +count: U32,
387  +payload_len: U32,
388  pair: Bytes.Bytes & Bytes.Bytes
389) -> EncodedBlock:
390  match pair:
391    case (output_payload, hash_payload):
392      block.fin.digest(count, bytes.clone.split(
393        SHA.sha256_packed_bytes(Bytes.concat([u32_bytes(payload_len),
394          u32_bytes(count), hash_payload]))), output_payload, payload_len)
395
396def block.fin(
397  +count: U32,
398  +payload_len: U32,
399  payload: Bytes.Bytes
400) -> EncodedBlock:
401  block.fin.payload(count, payload_len, bytes.clone.split(payload))
402
403def block.payload.check(
404  +count: U32,
405  +payload_len: U32,
406  payload: Bytes.Bytes,
407  within: Bool
408) -> Result<&1, &1, Error, EncodedBlock>:
409  match within:
410    case True{}:
411      Done{block.fin(count, payload_len, payload)}
412    case False{}:
413      Fail{TooLarge{}}
414
415def block.payload.total(
416  +count: U32,
417  payload_pair: Bytes.Bytes & U32
418) -> Result<&1, &1, Error, EncodedBlock>:
419  match payload_pair:
420    case (payload, +payload_len):
421      block.payload.check(count, payload_len, payload,
422        U32.is_le(payload_len, StorageBytes.MAX_BLOCK_BYTES()) ||
423        U32.is_eq(count, 1) && U32.is_le(payload_len, StorageBytes.MAX_RECORD_BYTES()))
424
425def block.payload(
426  count: U32,
427  result: Result<&1, &1, Error, List<&1, Bytes.Bytes>>
428) -> Result<&1, &1, Error, EncodedBlock>:
429  match result:
430    case Fail{error}:
431      Fail{error}
432    case Done{fragments}:
433      block.payload.total(count, Bytes.length(Bytes.concat(fragments)))
434
435# Encode records and compute the block digest.
436def encode_block(+entries: List<&2, MemTable.Entry>) -> Result<&1, &1, Error, EncodedBlock>:
437  block.payload(U32.from_nat(List.length(&2, MemTable.Entry, entries)), records(entries))
438
439# Store a block digest and continue with the remaining blocks.
440def blocks.prepare.step(
441  rest: List<&2, List<&2, MemTable.Entry>>,
442  current: Result<&1, &1, Error, EncodedBlock>,
443  blocks: List<&1, Bytes.Bytes>,
444  digests: List<&1, Bytes.Bytes>
445) -> Result<&1, &1, Error, List<&1, Bytes.Bytes> & List<&1, Bytes.Bytes>>:
446  match rest current:
447    case Nil{} Fail{error}:
448      Fail{error}
449    case Nil{} Done{Block{bytes, digest}}:
450      Done{(List.reverse(&1, Bytes.Bytes, bytes <> blocks),
451        List.reverse(&1, Bytes.Bytes, digest <> digests))}
452    case Con{_entries, _tail} Fail{error}:
453      Fail{error}
454    case Con{entries, tail} Done{Block{bytes, digest}}:
455      blocks.prepare.step(tail, encode_block(entries), bytes <> blocks, digest <> digests)
456
457def blocks.prepare(
458  +chunks: List<&2, List<&2, MemTable.Entry>>
459) -> Result<&1, &1, Error, List<&1, Bytes.Bytes> & List<&1, Bytes.Bytes>>:
460  match chunks:
461    case Nil{}:
462      Done{(Nil{}, Nil{})}
463    case Con{entries, rest}:
464      blocks.prepare.step(rest, encode_block(entries), Nil{}, Nil{})
465
466# Encode fixed-width v3 header fields in network byte order.
467def header_bytes(
468  +level: U32,
469  +total: U32,
470  +block_count: U32
471) -> Bytes.Bytes:
472  Bytes.concat([hex_bytes.fin(Bytes.from_hex("4d594c534d335300")),
473    u32_bytes(level), u32_bytes(total), u32_bytes(block_count)])
474
475# Propagate the block-summary result into an encoder header and footer.
476def encoder.build(
477  +entries: List<&2, MemTable.Entry>,
478  +level: U32,
479  +block_count: U32,
480  prepared: Result<&1, &1, Error, List<&1, Bytes.Bytes> & List<&1, Bytes.Bytes>>
481) -> Result<&1, &1, Error, Encoder>:
482  match prepared:
483    case Fail{error}:
484      Fail{error}
485    case Done{(chunks, hashes)}:
486      +total = U32.from_nat(List.length(&2, MemTable.Entry, entries))
487      root = SHA.sha256_packed_bytes(Bytes.concat(header_bytes(level, total, block_count) <> hashes))
488      footer = Bytes.concat([hex_bytes.fin(Bytes.from_hex("454e4433")), u32_bytes(total), root])
489      Done{Enc{Header{}, chunks, level, total, block_count, footer}}
490
491# Create an encoder only for supported levels and bounded records.
492def new_encoder.ready(
493  +entries: List<&2, MemTable.Entry>,
494  +level: U32,
495  +chunks: List<&2, List<&2, MemTable.Entry>>
496) -> Result<&1, &1, Error, Encoder>:
497  encoder.build(entries, level,
498    U32.from_nat(List.length(&2, List<&2, MemTable.Entry>, chunks)), blocks.prepare(chunks))
499
500def new_encoder.order(
501  +entries: List<&2, MemTable.Entry>,
502  +level: U32,
503  ordered: Bool
504) -> Result<&1, &1, Error, Encoder>:
505  match ordered:
506    case False{}:
507      Fail{InvalidOrder{}}
508    case True{}:
509      new_encoder.ready(entries, level, blocks.pack(entries))
510
511def new_encoder.valid(
512  +entries: List<&2, MemTable.Entry>,
513  +level: U32,
514  valid: Bool
515) -> Result<&1, &1, Error, Encoder>:
516  match valid:
517    case False{}:
518      Fail{InvalidLevel{}}
519    case True{}:
520      new_encoder.order(entries, level, entries.strict(entries))
521
522# Validate input order and level before encoding.
523def new_encoder(
524  +entries: List<&2, MemTable.Entry>,
525  +level: U32
526) -> Result<&1, &1, Error, Encoder>:
527  new_encoder.valid(entries, level, U32.is_le(level, 255))
528
529# Yield the next serialized part and advance encoder state.
530def next_chunk(encoder: Encoder) -> Encoder & Maybe<&1, Bytes.Bytes>:
531  match encoder:
532    case Enc{Header{}, chunks, +level, +total, +block_count, footer}:
533      (Enc{Blocks{}, chunks, level, total, block_count, footer}, Some{header_bytes(level, total, block_count)})
534    case Enc{Blocks{}, Nil{}, level, total, block_count, footer}:
535      (Enc{Finished{}, Nil{}, level, total, block_count, Bytes.new(0)}, Some{footer})
536    case Enc{Blocks{}, Con{bytes, rest}, level, total, block_count, footer}:
537      (Enc{Blocks{}, rest, level, total, block_count, footer}, Some{bytes})
538    case Enc{Finished{}, chunks, level, total, block_count, footer}:
539      (Enc{Finished{}, chunks, level, total, block_count, footer}, None{})
540
541# Translate the shared strict UTF-8 result into the SST error domain.
542type TextRead is Type:
543  DecodedText{cursor: Bytes.Cursor, byte: Maybe<&2, U32>}
544
545def parse.text.wrap(pair: Bytes.Cursor & Maybe<&2, U32>) -> TextRead:
546  match pair:
547    case (cursor, byte):
548      DecodedText{cursor, byte}
549
550def parse.text.finish(
551  cursor: Bytes.Cursor,
552  result: Result<&1, &1, U32 & String, String>
553) -> Bytes.Cursor & Result<&1, &1, Error, String>:
554  match result:
555    case Fail{_}:
556      (cursor, Fail{InvalidUtf8{}})
557    case Done{text}:
558      (cursor, Done{text})
559
560# Read exactly n UTF-8 bytes from a bounded cursor without materializing a slice.
561def parse.text.go(
562  remaining: Nat,
563  read: TextRead,
564  state: StorageBytes.Utf8State
565) -> Bytes.Cursor & Result<&1, &1, Error, String>:
566  match remaining read:
567    case 0n DecodedText{cursor, _}:
568      parse.text.finish(cursor, StorageBytes.utf8.scan.fin(state))
569    case 1n+0n DecodedText{cursor, Some{+byte}}:
570      parse.text.finish(cursor, StorageBytes.utf8.scan.fin(
571        StorageBytes.utf8.step(state, byte)))
572    case 1n+1n+rest DecodedText{cursor, Some{+byte}}:
573      parse.text.go(1n+rest, parse.text.wrap(Bytes.Cursor.u8(cursor)),
574        StorageBytes.utf8.step(state, byte))
575    case 1n+_ DecodedText{cursor, None{}}:
576      (cursor, Fail{Malformed{}})
577
578def parse.text.start(
579  cursor: Bytes.Cursor,
580  +length: U32,
581  empty: Bool
582) -> Bytes.Cursor & Result<&1, &1, Error, String>:
583  match empty:
584    case True{}:
585      parse.text.go(0n, DecodedText{cursor, Some{0}},
586        StorageBytes.Utf8State{0, 128, 191, 0, SNil{}, True{}})
587    case False{}:
588      parse.text.go(U32.to_nat(length), parse.text.wrap(Bytes.Cursor.u8(cursor)),
589        StorageBytes.Utf8State{0, 128, 191, 0, SNil{}, True{}})
590
591def parse.text(
592  cursor: Bytes.Cursor,
593  +length: U32
594) -> Bytes.Cursor & Result<&1, &1, Error, String>:
595  parse.text.start(cursor, length, U32.is_eq(length, 0))
596
597# Parsed mutation paired with its encoded byte length.
598type ParsedRecord is Type:
599  Rec{entry: MemTable.Entry, bytes_used: U32}
600
601# Decode one length-delimited key and optional value from the cursor.
602def parse.record.value.decoded(
603  pair: Bytes.Cursor & Result<&1, &1, Error, String>,
604  +key_len: U32,
605  +key: String,
606  +value_len: U32
607) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:
608  match pair:
609    case (cursor, Fail{error}):
610      (cursor, Fail{error})
611    case (cursor, Done{value}):
612      (cursor, Done{Rec{MemTable.Entry{key, Some{value}},
613        (9 + key_len + value_len : U32)}})
614
615def parse.record.value.length.checked(
616  cursor: Bytes.Cursor,
617  +key_len: U32,
618  +key: String,
619  +value_len: U32,
620  valid: Bool
621) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:
622  match valid:
623    case False{}:
624      (cursor, Fail{TooLarge{}})
625    case True{}:
626      parse.record.value.decoded(parse.text(cursor, value_len), key_len,
627        key, value_len)
628
629def parse.record.value.length(
630  pair: Bytes.Cursor & Maybe<&2, U32>,
631  +key_len: U32,
632  +key: String
633) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:
634  match pair:
635    case (cursor, Some{+value_len}):
636      parse.record.value.length.checked(cursor, key_len, key, value_len,
637        U32.is_le(value_len, (StorageBytes.MAX_RECORD_BYTES() - 9 - key_len : U32)))
638    case (cursor, None{}):
639      (cursor, Fail{Malformed{}})
640
641def parse.record.key.ready(
642  cursor: Bytes.Cursor,
643  tag: U32,
644  +key_len: U32,
645  +key: String
646) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:
647  match tag:
648    case 1:
649      (cursor, Done{Rec{MemTable.Entry{key, None{}}, (5 + key_len : U32)}})
650    case 0:
651      parse.record.value.length(Bytes.Cursor.u32be(cursor), key_len, key)
652    case _:
653      (cursor, Fail{Malformed{}})
654
655def parse.record.key.decoded(
656  pair: Bytes.Cursor & Result<&1, &1, Error, String>,
657  tag: U32,
658  +key_len: U32
659) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:
660  match pair:
661    case (cursor, Fail{error}):
662      (cursor, Fail{error})
663    case (cursor, Done{key}):
664      parse.record.key.ready(cursor, tag, key_len, key)
665
666def parse.record.key.limit(
667  tag: U32,
668  +key_len: U32
669) -> Bool:
670  match tag:
671    case 1:
672      U32.is_le(key_len, (StorageBytes.MAX_RECORD_BYTES() - 5 : U32))
673    case _:
674      U32.is_le(key_len, (StorageBytes.MAX_RECORD_BYTES() - 9 : U32))
675
676def parse.record.key.length.checked(
677  cursor: Bytes.Cursor,
678  tag: U32,
679  +key_len: U32,
680  valid: Bool
681) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:
682  match valid:
683    case False{}:
684      (cursor, Fail{TooLarge{}})
685    case True{}:
686      parse.record.key.decoded(parse.text(cursor, key_len), tag, key_len)
687
688def parse.record.key.length(
689  pair: Bytes.Cursor & Maybe<&2, U32>,
690  +tag: U32
691) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:
692  match pair:
693    case (cursor, Some{+key_len}):
694      parse.record.key.length.checked(cursor, tag, key_len,
695        parse.record.key.limit(tag, key_len))
696    case (cursor, None{}):
697      (cursor, Fail{Malformed{}})
698
699def parse.record.tag(
700  pair: Bytes.Cursor & Maybe<&2, U32>
701) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:
702  match pair:
703    case (cursor, Some{0}):
704      parse.record.key.length(Bytes.Cursor.u32be(cursor), 0)
705    case (cursor, Some{1}):
706      parse.record.key.length(Bytes.Cursor.u32be(cursor), 1)
707    case (cursor, Some{_}):
708      (cursor, Fail{Malformed{}})
709    case (cursor, None{}):
710      (cursor, Fail{Malformed{}})
711
712def parse.record(
713  cursor: Bytes.Cursor
714) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:
715  parse.record.tag(Bytes.Cursor.u8(cursor))
716
717# Parsed entries in reverse order with the last strict-order key.
718type ParsedEntries is Type:
719  Entries{reversed: List<&2, MemTable.Entry>, previous: Maybe<&2, String>}
720
721# Record result annotated with block bounds and key order.
722type RecordRead is Type:
723  Record{cursor: Bytes.Cursor,
724    result: Result<&1, &1, Error, ParsedRecord>,
725    within: Bool, ordered: Bool, last: Bool, complete: Bool, key: String}
726
727# Decision for a record at its declared block position.
728type RecordDecision is Type:
729  Rejected{cursor: Bytes.Cursor, error: Error}
730  OutOfBounds{cursor: Bytes.Cursor}
731  OutOfOrder{cursor: Bytes.Cursor}
732  LastMalformed{cursor: Bytes.Cursor}
733  LastRecord{cursor: Bytes.Cursor, entry: MemTable.Entry, key: String}
734  MoreRecords{cursor: Bytes.Cursor, entry: MemTable.Entry, key: String, used: U32}
735
736def parse.entries.ordered(
737  previous: Maybe<&2, String>,
738  +key: String
739) -> Bool:
740  match previous:
741    case None{}:
742      True{}
743    case Some{current}:
744      Keys.lt(current, key)
745
746def parse.record.read.entry(
747  cursor: Bytes.Cursor,
748  +entry: MemTable.Entry,
749  +bytes_used: U32,
750  +payload_left: U32,
751  previous: Maybe<&2, String>,
752  last: Bool
753) -> RecordRead:
754  match entry:
755    case MemTable.Entry{+key, _}:
756      Record{cursor, Done{Rec{entry, bytes_used}},
757        U32.is_le(bytes_used, payload_left), parse.entries.ordered(previous, key),
758        last, U32.is_eq(bytes_used, payload_left), key}
759
760def parse.record.read.pair(
761  pair: Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>,
762  payload_left: U32,
763  previous: Maybe<&2, String>,
764  last: Bool
765) -> RecordRead:
766  match pair:
767    case (cursor, Fail{error}):
768      Record{cursor, Fail{error}, False{}, False{}, last, False{}, ""}
769    case (cursor, Done{Rec{entry, +bytes_used}}):
770      parse.record.read.entry(cursor, entry, bytes_used, payload_left, previous, last)
771
772def parse.record.read(
773  cursor: Bytes.Cursor,
774  +payload_left: U32,
775  previous: Maybe<&2, String>,
776  last: Bool
777) -> RecordRead:
778  parse.record.read.pair(parse.record(cursor), payload_left, previous, last)
779
780def parse.records.decision(read: RecordRead) -> RecordDecision:
781  match read:
782    case Record{cursor, Fail{error}, _, _, _, _, _}:
783      Rejected{cursor, error}
784    case Record{cursor, Done{_}, False{}, _, _, _, _}:
785      OutOfBounds{cursor}
786    case Record{cursor, Done{_}, True{}, False{}, _, _, _}:
787      OutOfOrder{cursor}
788    case Record{cursor, Done{_}, True{}, True{}, True{}, False{}, _}:
789      LastMalformed{cursor}
790    case Record{cursor, Done{Rec{entry, _}}, True{}, True{}, True{}, True{}, +key}:
791      LastRecord{cursor, entry, key}
792    case Record{cursor, Done{Rec{entry, +used}}, True{}, True{}, False{}, _, +key}:
793      MoreRecords{cursor, entry, key, used}
794
795# Parse a counted sequence and enforce block boundaries and strict key order.
796def parse.records.empty(
797  decision: RecordDecision
798) -> Bytes.Cursor & Result<&1, &1, Error, ParsedEntries>:
799  match decision:
800    case Rejected{cursor, _}:
801      (cursor, Fail{Malformed{}})
802    case OutOfBounds{cursor}:
803      (cursor, Fail{Malformed{}})
804    case OutOfOrder{cursor}:
805      (cursor, Fail{Malformed{}})
806    case LastMalformed{cursor}:
807      (cursor, Fail{Malformed{}})
808    case LastRecord{cursor, _, _}:
809      (cursor, Fail{Malformed{}})
810    case MoreRecords{cursor, _, _, _}:
811      (cursor, Fail{Malformed{}})
812
813def parse.records.go(
814  remaining: Nat,
815  decision: RecordDecision,
816  +payload_left: U32,
817  acc: List<&2, MemTable.Entry>
818) -> Bytes.Cursor & Result<&1, &1, Error, ParsedEntries>:
819  match remaining:
820    case 0n:
821      parse.records.empty(decision)
822    case 1n++rest:
823      match decision:
824        case Rejected{cursor, error}:
825          (cursor, Fail{error})
826        case OutOfBounds{cursor}:
827          (cursor, Fail{Malformed{}})
828        case OutOfOrder{cursor}:
829          (cursor, Fail{InvalidOrder{}})
830        case LastMalformed{cursor}:
831          (cursor, Fail{Malformed{}})
832        case LastRecord{cursor, entry, +key}:
833          (cursor, Done{Entries{Con{entry, acc}, Some{key}}})
834        case MoreRecords{cursor, entry, +key, +used}:
835          +payload_after = (payload_left - used : U32)
836          parse.records.go(rest,
837            parse.records.decision(parse.record.read(cursor, payload_after,
838              Some{key}, Nat.is_eq(rest, 1n))),
839            payload_after, Con{entry, acc})
840
841def parse.records.zero(
842  cursor: Bytes.Cursor,
843  previous: Maybe<&2, String>,
844  empty: Bool
845) -> Bytes.Cursor & Result<&1, &1, Error, ParsedEntries>:
846  match empty:
847    case True{}:
848      (cursor, Done{Entries{Nil{}, previous}})
849    case False{}:
850      (cursor, Fail{Malformed{}})
851
852def parse.records.start(
853  cursor: Bytes.Cursor,
854  +count: U32,
855  +payload_len: U32,
856  previous: Maybe<&2, String>
857) -> Bytes.Cursor & Result<&1, &1, Error, ParsedEntries>:
858  match count:
859    case 0:
860      parse.records.zero(cursor, previous, U32.is_eq(payload_len, 0))
861    case _:
862      parse.records.go(U32.to_nat(count),
863        parse.records.decision(parse.record.read(cursor, payload_len, previous, U32.is_eq(count, 1))),
864        payload_len, Nil{})
865
866# Cursor state while assembling a fixed-width digest.
867type HashRead is Type:
868  HashInput{cursor: Bytes.Cursor, value: Maybe<&2, U32>}
869
870# Validated records, count, and digest for one block.
871type ParsedBlock is Type:
872  ParsedBlock{records: ParsedEntries, count: U32, digest: Bytes.Bytes}
873
874# Represent StreamState data used by the packed SSTable codec.
875type StreamState is Type:
876  StreamState{level: U32, total: U32, block_count: U32,
877    remaining: U32, entries_left: U32,
878    previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>,
879    hashes: List<&1, Bytes.Bytes>}
880
881# Holds bounded SST validation state without retaining prior table entries.
882type BlockStreamState is Type:
883  BlockStreamState{level: U32, total: U32, block_count: U32,
884    remaining: U32, entries_left: U32,
885    previous: Maybe<&2, String>, hashes: List<&1, Bytes.Bytes>}
886
887# Returns one validated block and the state for the next block.
888type BlockStreamOutput is Type:
889  BlockStreamOutput{state: BlockStreamState, entries: List<&2, MemTable.Entry>}
890
891def parse.hash.wrap(pair: Bytes.Cursor & Maybe<&2, U32>) -> HashRead:
892  match pair:
893    case (cursor, value):
894      HashInput{cursor, value}
895
896# Read exactly eight words into a digest without consuming a byte after it.
897def parse.hash.words.go(
898  remaining: Nat,
899  read: HashRead,
900  digest: Bytes.Bytes,
901  +offset: U32
902) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:
903  match remaining read:
904    case 0n HashInput{cursor, _}:
905      (cursor, Done{digest})
906    case 1n+0n HashInput{cursor, Some{word}}:
907      (cursor, Done{Bytes.set.u32be(digest, offset, word)})
908    case 1n+1n+rest HashInput{cursor, Some{word}}:
909      parse.hash.words.go(1n+rest, parse.hash.wrap(Bytes.Cursor.u32be(cursor)),
910        Bytes.set.u32be(digest, offset, word), (offset + 4 : U32))
911    case 1n+_ HashInput{cursor, None{}}:
912      (cursor, Fail{Malformed{}})
913
914def parse.hash.words(
915  cursor: Bytes.Cursor
916) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:
917  parse.hash.words.go(8n, parse.hash.wrap(Bytes.Cursor.u32be(cursor)), Bytes.new(32), 0)
918
919def parse.block.verify.equal(
920  cursor: Bytes.Cursor,
921  records: ParsedEntries,
922  +count: U32,
923  comparison: Bytes.Bytes & Bytes.Bytes & Bool
924) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
925  match comparison:
926    case (claimed, actual, False{}):
927      (cursor, Fail{ChecksumMismatch{}})
928    case (claimed, digest, True{}):
929      (cursor, Done{ParsedBlock{records, count, digest}})
930
931def parse.block.verify.copy(
932  pair: Bytes.Bytes & Bytes.Bytes,
933  +pos: U32,
934  +start: U32,
935  +end: U32,
936  records: ParsedEntries,
937  +count: U32,
938  claimed: Bytes.Bytes
939) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
940  match pair:
941    case (source, content):
942      parse.block.verify.equal(Bytes.Cursor{source, pos, start, end}, records,
943        count, Bytes.eq(claimed, SHA.sha256_packed_bytes(content)))
944
945def parse.block.verify(
946  cursor: Bytes.Cursor,
947  records: ParsedEntries,
948  +payload_len: U32,
949  +count: U32,
950  claimed: Bytes.Bytes
951) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
952  match cursor:
953    case Bytes.Cursor{bytes, +pos, +start, +end}:
954      parse.block.verify.copy(Bytes.slice(bytes,
955        (pos - (payload_len + 40 : U32) : U32), (payload_len + 8 : U32)),
956        pos, start, end, records, count, claimed)
957
958def parse.block.digest(
959  pair: Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>,
960  records: ParsedEntries,
961  +payload_len: U32,
962  +count: U32
963) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
964  match pair:
965    case (cursor, Fail{error}):
966      (cursor, Fail{error})
967    case (cursor, Done{claimed}):
968      parse.block.verify(cursor, records, payload_len, count, claimed)
969
970def parse.block.records(
971  pair: Bytes.Cursor & Result<&1, &1, Error, ParsedEntries>,
972  +payload_len: U32,
973  +count: U32
974) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
975  match pair:
976    case (cursor, Fail{error}):
977      (cursor, Fail{error})
978    case (cursor, Done{records}):
979      parse.block.digest(parse.hash.words(cursor), records, payload_len, count)
980
981def parse.block.count.valid(
982  cursor: Bytes.Cursor,
983  +payload_len: U32,
984  +count: U32,
985  previous: Maybe<&2, String>,
986  valid: Bool
987) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
988  match valid:
989    case False{}:
990      (cursor, Fail{Malformed{}})
991    case True{}:
992      parse.block.records(parse.records.start(cursor, count, payload_len, previous),
993        payload_len, count)
994
995def parse.block.count(
996  pair: Bytes.Cursor & Maybe<&2, U32>,
997  +payload_len: U32,
998  previous: Maybe<&2, String>
999) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
1000  match pair:
1001    case (cursor, Some{+count}):
1002      parse.block.count.valid(cursor, payload_len, count, previous,
1003        U32.is_lt(0, count) && (U32.is_le(payload_len, StorageBytes.MAX_BLOCK_BYTES()) ||
1004          U32.is_eq(count, 1) && U32.is_le(payload_len, StorageBytes.MAX_RECORD_BYTES())))
1005    case (cursor, None{}):
1006      (cursor, Fail{Malformed{}})
1007
1008def parse.block.length(
1009  pair: Bytes.Cursor & Maybe<&2, U32>,
1010  previous: Maybe<&2, String>
1011) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
1012  match pair:
1013    case (cursor, Some{+payload_len}):
1014      parse.block.count(Bytes.Cursor.u32be(cursor), payload_len, previous)
1015    case (cursor, None{}):
1016      (cursor, Fail{Malformed{}})
1017
1018# Parse, bound, and authenticate one complete SST block.
1019def parse.block(
1020  cursor: Bytes.Cursor,
1021  previous: Maybe<&2, String>
1022) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
1023  parse.block.length(Bytes.Cursor.u32be(cursor), previous)
1024
1025def parse.footer.end(
1026  +level: U32,
1027  entries: List<&2, MemTable.Entry>,
1028  at_end: Bool
1029) -> Result<&1, &1, Error, Sstable.Table>:
1030  match at_end:
1031    case False{}:
1032      Fail{Malformed{}}
1033    case True{}:
1034      Done{Sstable.from_sorted_unique(
1035        List.reverse(&2, MemTable.Entry, entries), U32.to_nat(level))}
1036
1037def parse.footer.check(
1038  +level: U32,
1039  entries: List<&2, MemTable.Entry>,
1040  at_end: Bool,
1041  comparison: Bytes.Bytes & Bytes.Bytes & Bool
1042) -> Result<&1, &1, Error, Sstable.Table>:
1043  match comparison:
1044    case (claimed_back, expected_back, False{}):
1045      Fail{ChecksumMismatch{}}
1046    case (claimed_back, expected_back, True{}):
1047      parse.footer.end(level, entries, at_end)
1048
1049def parse.footer.verify(
1050  cursor: Bytes.Cursor,
1051  claimed: Bytes.Bytes,
1052  +level: U32,
1053  +total: U32,
1054  +block_count: U32,
1055  entries: List<&2, MemTable.Entry>,
1056  hashes: List<&1, Bytes.Bytes>
1057) -> Result<&1, &1, Error, Sstable.Table>:
1058  match cursor:
1059    case Bytes.Cursor{bytes, +pos, _, +end}:
1060      expected = SHA.sha256_packed_bytes(Bytes.concat(
1061        header_bytes(level, total, block_count) <> List.reverse(&1, Bytes.Bytes, hashes)))
1062      parse.footer.check(level, entries, U32.is_eq(pos, end),
1063        Bytes.eq(claimed, expected))
1064
1065def parse.footer.root(
1066  pair: Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>,
1067  +level: U32,
1068  +total: U32,
1069  +block_count: U32,
1070  entries: List<&2, MemTable.Entry>,
1071  hashes: List<&1, Bytes.Bytes>
1072) -> Result<&1, &1, Error, Sstable.Table>:
1073  match pair:
1074    case (_, Fail{error}):
1075      Fail{error}
1076    case (cursor, Done{claimed}):
1077      parse.footer.verify(cursor, claimed, level, total, block_count, entries, hashes)
1078
1079def parse.footer.count.checked(
1080  cursor: Bytes.Cursor,
1081  +level: U32,
1082  +total: U32,
1083  +block_count: U32,
1084  entries: List<&2, MemTable.Entry>,
1085  hashes: List<&1, Bytes.Bytes>,
1086  valid: Bool
1087) -> Result<&1, &1, Error, Sstable.Table>:
1088  match valid:
1089    case False{}:
1090      Fail{Malformed{}}
1091    case True{}:
1092      parse.footer.root(parse.hash.words(cursor), level, total, block_count,
1093        entries, hashes)
1094
1095def parse.footer.count(
1096  pair: Bytes.Cursor & Maybe<&2, U32>,
1097  +level: U32,
1098  +total: U32,
1099  +block_count: U32,
1100  entries: List<&2, MemTable.Entry>,
1101  hashes: List<&1, Bytes.Bytes>
1102) -> Result<&1, &1, Error, Sstable.Table>:
1103  match pair:
1104    case (cursor, Some{+footer_total}):
1105      parse.footer.count.checked(cursor, level, total, block_count, entries,
1106        hashes, U32.is_eq(total, footer_total))
1107    case (_, None{}):
1108      Fail{Malformed{}}
1109
1110def parse.footer.magic(
1111  pair: Bytes.Cursor & Maybe<&2, U32>,
1112  +level: U32,
1113  +total: U32,
1114  +block_count: U32,
1115  entries: List<&2, MemTable.Entry>,
1116  hashes: List<&1, Bytes.Bytes>
1117) -> Result<&1, &1, Error, Sstable.Table>:
1118  match pair:
1119    case (cursor, Some{1162757171}):
1120      parse.footer.count(Bytes.Cursor.u32be(cursor), level, total,
1121        block_count, entries, hashes)
1122    case (_, _):
1123      Fail{Malformed{}}
1124
1125# Block result paired with its ending cursor.
1126type BlockRead is Type:
1127  BlockInput{cursor: Bytes.Cursor, result: Result<&1, &1, Error, ParsedBlock>}
1128
1129def parse.block.read(pair: Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>) -> BlockRead:
1130  match pair:
1131    case (cursor, result):
1132      BlockInput{cursor, result}
1133
1134def parse.blocks.count.valid.last(
1135  last: Bool,
1136  +entries_left: U32,
1137  +count: U32
1138) -> Bool:
1139  match last:
1140    case True{}:
1141      U32.is_eq(entries_left, count)
1142    case False{}:
1143      U32.is_lt(count, entries_left)
1144
1145def parse.blocks.count.valid(
1146  +remaining: Nat,
1147  +entries_left: U32,
1148  +count: U32
1149) -> Bool:
1150  parse.blocks.count.valid.last(Nat.is_eq(remaining, 1n), entries_left, count)
1151
1152# Accepts a block only when it fits the file's remaining block and entry counts.
1153def stream.blocks.count.valid(
1154  exact: Bool,
1155  +remaining: U32,
1156  +entries_left: U32,
1157  +count: U32
1158) -> Bool:
1159  exact && U32.is_lt(0, remaining) && U32.is_le(count, entries_left) &&
1160    parse.blocks.count.valid(U32.to_nat(remaining), entries_left, count)
1161
1162# Updates constant-size file state and returns this block's ordered entries.
1163def stream.blocks.updated(
1164  state: BlockStreamState,
1165  records: ParsedEntries,
1166  +count: U32,
1167  digest: Bytes.Bytes
1168) -> BlockStreamOutput:
1169  match state:
1170    case BlockStreamState{+level, +total, +block_count, +remaining,
1171        +entries_left, _, hashes}:
1172      match records:
1173        case Entries{reversed, previous}:
1174          BlockStreamOutput{
1175            BlockStreamState{level, total, block_count,
1176              (remaining - 1 : U32), (entries_left - count : U32),
1177              previous, Con{digest, hashes}},
1178            List.reverse(&2, MemTable.Entry, reversed)}
1179
1180# Converts the bounded count decision into a validated block output.
1181def stream.blocks.decide(
1182  state: BlockStreamState,
1183  records: ParsedEntries,
1184  +count: U32,
1185  digest: Bytes.Bytes,
1186  valid: Bool
1187) -> Result<&1, &1, Error, BlockStreamOutput>:
1188  match valid:
1189    case False{}:
1190      Fail{Malformed{}}
1191    case True{}:
1192      Done{stream.blocks.updated(state, records, count, digest)}
1193
1194# Applies count, order, and exact-block-consumption rules to one parsed block.
1195def stream.blocks.valid(
1196  state: BlockStreamState,
1197  records: ParsedEntries,
1198  +count: U32,
1199  digest: Bytes.Bytes,
1200  exact: Bool
1201) -> Result<&1, &1, Error, BlockStreamOutput>:
1202  match state:
1203    case BlockStreamState{_, _, _, +remaining, +entries_left, _, _}:
1204      stream.blocks.decide(state, records, count, digest,
1205        stream.blocks.count.valid(exact, remaining, entries_left, count))
1206
1207# Validates one block's checksum and returns its bounded entry list.
1208def stream.blocks.complete(
1209  state: BlockStreamState,
1210  block: ParsedBlock,
1211  exact: Bool
1212) -> Result<&1, &1, Error, BlockStreamOutput>:
1213  match block:
1214    case ParsedBlock{records, +count, digest}:
1215      stream.blocks.valid(state, records, count, digest, exact)
1216
1217# Requires the block parser to consume its complete input slice.
1218def stream.blocks.consumed(
1219  pair: Bytes.Cursor & U32,
1220  state: BlockStreamState,
1221  block: ParsedBlock
1222) -> Result<&1, &1, Error, BlockStreamOutput>:
1223  match pair:
1224    case (_, remaining):
1225      stream.blocks.complete(state, block, U32.is_eq(remaining, 0))
1226
1227# Propagates parse failures without returning a partial block.
1228def stream.blocks.result(
1229  state: BlockStreamState,
1230  pair: Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>
1231) -> Result<&1, &1, Error, BlockStreamOutput>:
1232  match pair:
1233    case (_, Fail{error}):
1234      Fail{error}
1235    case (cursor, Done{block}):
1236      stream.blocks.consumed(Bytes.Cursor.remaining(cursor), state, block)
1237
1238# Parses and authenticates exactly one SST block at a time.
1239def stream.blocks.next(
1240  state: BlockStreamState,
1241  bytes: Bytes.Bytes
1242) -> Result<&1, &1, Error, BlockStreamOutput>:
1243  match state:
1244    case BlockStreamState{_, _, _, _, _, +previous, _}:
1245      stream.blocks.result(state, parse.block(Bytes.Cursor.new(bytes), previous))
1246
1247# Projects successful footer validation to a unit result.
1248def stream.blocks.footer.result(
1249  result: Result<&1, &1, Error, Sstable.Table>
1250) -> Result<&1, &1, Error, Unit>:
1251  match result:
1252    case Fail{error}:
1253      Fail{error}
1254    case Done{_}:
1255      Done{Unit{}}
1256
1257# Parses a footer only after the block and entry counts reach zero.
1258def stream.blocks.footer.parse(
1259  +level: U32,
1260  +total: U32,
1261  +block_count: U32,
1262  hashes: List<&1, Bytes.Bytes>,
1263  bytes: Bytes.Bytes,
1264  valid: Bool
1265) -> Result<&1, &1, Error, Unit>:
1266  match valid:
1267    case False{}:
1268      Fail{Malformed{}}
1269    case True{}:
1270      stream.blocks.footer.result(parse.footer.magic(
1271        Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)), level, total,
1272        block_count, Nil{}, hashes))
1273
1274# Validates the exact footer after every block has been consumed.
1275def stream.blocks.footer.valid(
1276  state: BlockStreamState,
1277  bytes: Bytes.Bytes,
1278  valid: Bool
1279) -> Result<&1, &1, Error, Unit>:
1280  match state:
1281    case BlockStreamState{+level, +total, +block_count, _, _, _, hashes}:
1282      stream.blocks.footer.parse(level, total, block_count, hashes, bytes, valid)
1283
1284# Checks block and entry exhaustion before validating the SST footer.
1285def stream.blocks.footer(
1286  state: BlockStreamState,
1287  bytes: Bytes.Bytes
1288) -> Result<&1, &1, Error, Unit>:
1289  match state:
1290    case BlockStreamState{_, _, _, +remaining, +entries_left, _, _}:
1291      stream.blocks.footer.valid(state, bytes,
1292        U32.is_eq(remaining, 0) && U32.is_eq(entries_left, 0))
1293
1294def parse.blocks.accumulate.records.last(
1295  cursor: Bytes.Cursor,
1296  +entries_left: U32,
1297  entries: List<&2, MemTable.Entry>,
1298  hashes: List<&1, Bytes.Bytes>,
1299  reversed: List<&2, MemTable.Entry>,
1300  previous: Maybe<&2, String>,
1301  last: Bool
1302) -> BlockDecision:
1303  match last:
1304    case True{}:
1305      BlockFinish{cursor, List.append(&2, MemTable.Entry, reversed, entries), hashes}
1306    case False{}:
1307      BlockContinue{cursor, entries_left, previous,
1308        List.append(&2, MemTable.Entry, reversed, entries), hashes}
1309
1310def parse.blocks.accumulate.records(
1311  cursor: Bytes.Cursor,
1312  +remaining: Nat,
1313  +entries_left: U32,
1314  entries: List<&2, MemTable.Entry>,
1315  hashes: List<&1, Bytes.Bytes>,
1316  records: ParsedEntries
1317) -> BlockDecision:
1318  match records:
1319    case Entries{reversed, previous}:
1320      parse.blocks.accumulate.records.last(cursor, entries_left, entries,
1321        hashes, reversed, previous, Nat.is_eq(remaining, 1n))
1322
1323def parse.blocks.content.valid(
1324  cursor: Bytes.Cursor,
1325  +remaining: Nat,
1326  +entries_left: U32,
1327  entries: List<&2, MemTable.Entry>,
1328  hashes: List<&1, Bytes.Bytes>,
1329  +count: U32,
1330  records: ParsedEntries,
1331  digest: Bytes.Bytes,
1332  valid: Bool
1333) -> BlockDecision:
1334  match valid:
1335    case False{}:
1336      BlockOutOfBounds{}
1337    case True{}:
1338      parse.blocks.accumulate.records(cursor, remaining,
1339        (entries_left - count : U32), entries, digest <> hashes, records)
1340
1341def parse.blocks.content.fit(
1342  cursor: Bytes.Cursor,
1343  +remaining: Nat,
1344  +entries_left: U32,
1345  entries: List<&2, MemTable.Entry>,
1346  hashes: List<&1, Bytes.Bytes>,
1347  +count: U32,
1348  records: ParsedEntries,
1349  digest: Bytes.Bytes,
1350  fits: Bool
1351) -> BlockDecision:
1352  match fits:
1353    case False{}:
1354      BlockOutOfBounds{}
1355    case True{}:
1356      parse.blocks.content.valid(cursor, remaining, entries_left,
1357        entries, hashes, count, records, digest,
1358        parse.blocks.count.valid(remaining, entries_left, count))
1359
1360def parse.blocks.content(
1361  cursor: Bytes.Cursor,
1362  +remaining: Nat,
1363  +entries_left: U32,
1364  entries: List<&2, MemTable.Entry>,
1365  hashes: List<&1, Bytes.Bytes>,
1366  block: ParsedBlock
1367) -> BlockDecision:
1368  match block:
1369    case ParsedBlock{records, +count, digest}:
1370      parse.blocks.content.fit(cursor, remaining, entries_left,
1371        entries, hashes, count, records, digest, U32.is_le(count, entries_left))
1372
1373def parse.blocks.decision(
1374  read: BlockRead,
1375  +remaining: Nat,
1376  +entries_left: U32,
1377  entries: List<&2, MemTable.Entry>,
1378  hashes: List<&1, Bytes.Bytes>
1379) -> BlockDecision:
1380  match read:
1381    case BlockInput{_, Fail{error}}:
1382      BlockRejected{error}
1383    case BlockInput{cursor, Done{block}}:
1384      parse.blocks.content(cursor, remaining, entries_left, entries,
1385        hashes, block)
1386
1387def parse.blocks.go(
1388  remaining: Nat,
1389  decision: BlockDecision,
1390  +level: U32,
1391  +total: U32,
1392  +block_count: U32
1393) -> Result<&1, &1, Error, Sstable.Table>:
1394  match remaining:
1395    case 0n:
1396      Fail{Malformed{}}
1397    case 1n++rest:
1398      match decision:
1399        case BlockRejected{error}:
1400          Fail{error}
1401        case BlockOutOfBounds{}:
1402          Fail{Malformed{}}
1403        case BlockFinish{cursor, entries, hashes}:
1404          parse.footer.magic(Bytes.Cursor.u32be(cursor), level, total,
1405            block_count, entries, hashes)
1406        case BlockContinue{cursor, entries_left, previous, entries, hashes}:
1407          parse.blocks.go(rest,
1408            parse.blocks.decision(
1409              parse.block.read(parse.block(cursor, previous)), rest,
1410              entries_left, entries, hashes),
1411            level, total, block_count)
1412
1413def parse.blocks.start(
1414  cursor: Bytes.Cursor,
1415  valid: Bool,
1416  +count: U32,
1417  +total: U32,
1418  +level: U32
1419) -> Result<&1, &1, Error, Sstable.Table>:
1420  match valid:
1421    case False{}:
1422      Fail{Malformed{}}
1423    case True{}:
1424      match count:
1425        case 0:
1426          parse.footer.magic(Bytes.Cursor.u32be(cursor), level, total,
1427            count, Nil{}, Nil{})
1428        case _:
1429          parse.blocks.go(U32.to_nat(count),
1430            parse.blocks.decision(
1431              parse.block.read(parse.block(cursor, None{})), U32.to_nat(count),
1432            total, Nil{}, Nil{}),
1433            level, total, count)
1434
1435def stream.block.records(
1436  +level: U32,
1437  +total: U32,
1438  +block_count: U32,
1439  +remaining: U32,
1440  +entries_left: U32,
1441  entries: List<&2, MemTable.Entry>,
1442  hashes: List<&1, Bytes.Bytes>,
1443  records: ParsedEntries,
1444  +count: U32,
1445  digest: Bytes.Bytes
1446) -> Result<&1, &1, Error, StreamState>:
1447  match records:
1448    case Entries{reversed, previous}:
1449      Done{StreamState{level, total, block_count,
1450        (remaining - 1 : U32), (entries_left - count : U32),
1451        previous, List.append(&2, MemTable.Entry, reversed, entries),
1452        digest <> hashes}}
1453
1454def stream.block.count(
1455  +level: U32,
1456  +total: U32,
1457  +block_count: U32,
1458  +remaining: U32,
1459  +entries_left: U32,
1460  entries: List<&2, MemTable.Entry>,
1461  hashes: List<&1, Bytes.Bytes>,
1462  records: ParsedEntries,
1463  +count: U32,
1464  digest: Bytes.Bytes,
1465  valid: Bool
1466) -> Result<&1, &1, Error, StreamState>:
1467  match valid:
1468    case False{}:
1469      Fail{Malformed{}}
1470    case True{}:
1471      stream.block.records(level, total, block_count, remaining,
1472        entries_left, entries, hashes, records, count, digest)
1473
1474def stream.block.valid(
1475  state: StreamState,
1476  records: ParsedEntries,
1477  +count: U32,
1478  digest: Bytes.Bytes,
1479  exact: Bool
1480) -> Result<&1, &1, Error, StreamState>:
1481  match state:
1482    case StreamState{+level, +total, +block_count, +remaining, +entries_left, _, entries, hashes}:
1483      stream.block.count(level, total, block_count, remaining,
1484        entries_left, entries, hashes, records, count, digest,
1485        exact && U32.is_le(count, entries_left) &&
1486          parse.blocks.count.valid(U32.to_nat(remaining), entries_left, count))
1487
1488def stream.block.complete(
1489  state: StreamState,
1490  block: ParsedBlock,
1491  exact: Bool
1492) -> Result<&1, &1, Error, StreamState>:
1493  match block:
1494    case ParsedBlock{records, +count, digest}:
1495      stream.block.valid(state, records, count, digest, exact)
1496
1497def stream.block.consumed(
1498  pair: Bytes.Cursor & U32,
1499  state: StreamState,
1500  block: ParsedBlock
1501) -> Result<&1, &1, Error, StreamState>:
1502  match pair:
1503    case (_, remaining):
1504      stream.block.complete(state, block, U32.is_eq(remaining, 0))
1505
1506def stream.block.result(
1507  state: StreamState,
1508  pair: Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>
1509) -> Result<&1, &1, Error, StreamState>:
1510  match pair:
1511    case (_, Fail{error}):
1512      Fail{error}
1513    case (cursor, Done{block}):
1514      stream.block.consumed(Bytes.Cursor.remaining(cursor), state, block)
1515
1516def stream.block(state: StreamState, bytes: Bytes.Bytes) -> Result<&1, &1, Error, StreamState>:
1517  match state:
1518    case StreamState{_, _, _, 0, _, _, _, _}:
1519      Fail{Malformed{}}
1520    case StreamState{+level, +total, +block_count, +remaining,
1521        +entries_left, +previous, entries, hashes}:
1522      stream.block.result(
1523        StreamState{level, total, block_count, remaining, entries_left,
1524          previous, entries, hashes},
1525        parse.block(Bytes.Cursor.new(bytes), previous))
1526
1527def stream.footer(state: StreamState, bytes: Bytes.Bytes) -> Result<&1, &1, Error, Sstable.Table>:
1528  match state:
1529    case StreamState{+level, +total, +block_count, 0, 0, _, entries, hashes}:
1530      parse.footer.magic(
1531        Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)), level, total,
1532        block_count, entries, hashes)
1533    case _:
1534      Fail{Malformed{}}
1535
1536def stream.block.tail.size(
1537  pair: Bytes.Cursor & Maybe<&2, U32>,
1538  +payload_len: U32
1539) -> Maybe<&2, U32>:
1540  match pair:
1541    case (_, Some{+count}):
1542      Bool.pick(Maybe<&2, U32>,
1543        U32.is_lt(0, count) && (U32.is_le(payload_len, StorageBytes.MAX_BLOCK_BYTES()) ||
1544          U32.is_eq(count, 1) && U32.is_le(payload_len, StorageBytes.MAX_RECORD_BYTES())) &&
1545          U32.is_le(payload_len, (4294967263 : U32)),
1546        Some{(payload_len + 32 : U32)}, None{})
1547    case (_, None{}):
1548      None{}
1549
1550def stream.block.tail(cursor: Bytes.Cursor & Maybe<&2, U32>) -> Maybe<&2, U32>:
1551  match cursor:
1552    case (next, Some{+payload_len}):
1553      stream.block.tail.size(Bytes.Cursor.u32be(next), payload_len)
1554    case (_, None{}):
1555      None{}
1556
1557def stream.block.tail_size(bytes: Bytes.Bytes) -> Maybe<&2, U32>:
1558  stream.block.tail(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))
1559
1560def parse.header.blocks.valid(
1561  cursor: Bytes.Cursor,
1562  +level: U32,
1563  +total: U32,
1564  +count: U32,
1565  valid: Bool
1566) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1567  match valid:
1568    case False{}:
1569      (cursor, Fail{Malformed{}})
1570    case True{}:
1571      (cursor, Done{SstHdr{level, total, count}})
1572
1573def parse.header.blocks(
1574  pair: Bytes.Cursor & Maybe<&2, U32>,
1575  +level: U32,
1576  +total: U32
1577) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1578  match pair:
1579    case (cursor, Some{+count}):
1580      parse.header.blocks.valid(cursor, level, total, count,
1581        (U32.is_eq(total, 0) && U32.is_eq(count, 0)) ||
1582          (U32.is_lt(0, total) && U32.is_lt(0, count)))
1583    case (cursor, None{}):
1584      (cursor, Fail{Malformed{}})
1585
1586def parse.header.total(
1587  pair: Bytes.Cursor & Maybe<&2, U32>,
1588  +level: U32
1589) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1590  match pair:
1591    case (cursor, Some{+total}):
1592      parse.header.blocks(Bytes.Cursor.u32be(cursor), level, total)
1593    case (cursor, None{}):
1594      (cursor, Fail{Malformed{}})
1595
1596def parse.header.level.checked(
1597  cursor: Bytes.Cursor,
1598  +level: U32,
1599  valid: Bool
1600) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1601  match valid:
1602    case False{}:
1603      (cursor, Fail{InvalidLevel{}})
1604    case True{}:
1605      parse.header.total(Bytes.Cursor.u32be(cursor), level)
1606
1607def parse.header.level(
1608  pair: Bytes.Cursor & Maybe<&2, U32>
1609) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1610  match pair:
1611    case (cursor, Some{+level}):
1612      parse.header.level.checked(cursor, level, U32.is_le(level, 255))
1613    case (cursor, None{}):
1614      (cursor, Fail{Malformed{}})
1615
1616def parse.header.magic.two(
1617  pair: Bytes.Cursor & Maybe<&2, U32>
1618) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1619  match pair:
1620    case (cursor, Some{1295209216}):
1621      parse.header.level(Bytes.Cursor.u32be(cursor))
1622    case (cursor, _):
1623      (cursor, Fail{Malformed{}})
1624
1625def parse.header.magic.one(
1626  pair: Bytes.Cursor & Maybe<&2, U32>
1627) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1628  match pair:
1629    case (cursor, Some{1297697875}):
1630      parse.header.magic.two(Bytes.Cursor.u32be(cursor))
1631    case (cursor, _):
1632      (cursor, Fail{Malformed{}})
1633
1634def stream.header.remaining(
1635  pair: Bytes.Cursor & U32,
1636  +level: U32,
1637  +total: U32,
1638  +block_count: U32
1639) -> Result<&1, &1, Error, StreamState>:
1640  match pair:
1641    case (_, 0):
1642      Done{StreamState{level, total, block_count, block_count,
1643        total, None{}, Nil{}, Nil{}}}
1644    case (_, _):
1645      Fail{Malformed{}}
1646
1647def stream.header.done(
1648  pair: Bytes.Cursor & Result<&1, &1, Error, SstHeader>
1649) -> Result<&1, &1, Error, StreamState>:
1650  match pair:
1651    case (_, Fail{error}):
1652      Fail{error}
1653    case (cursor, Done{SstHdr{+level, +total, +block_count}}):
1654      stream.header.remaining(Bytes.Cursor.remaining(cursor), level, total, block_count)
1655
1656def stream.header(bytes: Bytes.Bytes) -> Result<&1, &1, Error, StreamState>:
1657  stream.header.done(parse.header.magic.one(
1658    Bytes.Cursor.u32be(Bytes.Cursor.new(bytes))))
1659
1660# Drops the legacy materialized-entry field when opening a block reader.
1661def stream.blocks.header.result(
1662  result: Result<&1, &1, Error, StreamState>
1663) -> Result<&1, &1, Error, BlockStreamState>:
1664  match result:
1665    case Fail{error}:
1666      Fail{error}
1667    case Done{StreamState{level, total, block_count, remaining,
1668        entries_left, previous, _, hashes}}:
1669      Done{BlockStreamState{level, total, block_count, remaining,
1670        entries_left, previous, hashes}}
1671
1672# Opens a block stream from the fixed SST header without allocating entries.
1673def stream.blocks.header(bytes: Bytes.Bytes) -> Result<&1, &1, Error, BlockStreamState>:
1674  stream.blocks.header.result(stream.header(bytes))
1675
1676# Retain bounded file chunks until the strict parser is applied.
1677def new_decoder() -> Decoder:
1678  Dec{Nil{}, 0, False{}}
1679
1680def feed.chunk.add(
1681  chunks: List<&1, Bytes.Bytes>,
1682  +total: U32,
1683  invalid: Bool,
1684  chunk: Bytes.Bytes,
1685  checked: Maybe<&2, U32>
1686) -> Decoder:
1687  match checked:
1688    case None{}:
1689      Dec{chunks, total, True{}}
1690    case Some{next_total}:
1691      Dec{chunk <> chunks, next_total, invalid}
1692
1693def feed.chunk.result(
1694  chunks: List<&1, Bytes.Bytes>,
1695  +total: U32,
1696  invalid: Bool,
1697  pair: Bytes.Bytes & U32
1698) -> Decoder:
1699  match pair:
1700    case (chunk, +len):
1701      feed.chunk.add(chunks, total, invalid, chunk, StorageBytes.checked_add(total, len))
1702
1703def feed.chunk(decoder: Decoder, pair: Bytes.Bytes & U32) -> Decoder:
1704  match decoder:
1705    case Dec{chunks, total, invalid}:
1706      feed.chunk.result(chunks, total, invalid, pair)
1707
1708def feed.if.valid(
1709  invalid: Bool,
1710  chunks: List<&1, Bytes.Bytes>,
1711  total: U32,
1712  chunk: Bytes.Bytes
1713) -> Decoder:
1714  match invalid:
1715    case True{}:
1716      Dec{chunks, total, True{}}
1717    case False{}:
1718      feed.chunk(Dec{chunks, total, False{}}, Bytes.length(chunk))
1719
1720# Buffer streaming chunks until the bounded pure parser is applied.
1721def feed(decoder: Decoder, chunk: Bytes.Bytes) -> Decoder:
1722  match decoder:
1723    case Dec{chunks, total, invalid}:
1724      feed.if.valid(invalid, chunks, total, chunk)
1725
1726def parse.file.header(
1727  pair: Bytes.Cursor & Result<&1, &1, Error, SstHeader>
1728) -> Result<&1, &1, Error, Sstable.Table>:
1729  match pair:
1730    case (_, Fail{error}):
1731      Fail{error}
1732    case (cursor, Done{SstHdr{+level, +total, +block_count}}):
1733      parse.blocks.start(cursor, True{}, block_count, total, level)
1734
1735def parse.file(bytes: Bytes.Bytes) -> Result<&1, &1, Error, Sstable.Table>:
1736  parse.file.header(parse.header.magic.one(
1737    Bytes.Cursor.u32be(Bytes.Cursor.new(bytes))))
1738
1739def finish.concat.check(
1740  bytes: Bytes.Bytes,
1741  valid: Bool
1742) -> Result<&1, &1, Error, Sstable.Table>:
1743  match valid:
1744    case True{}:
1745      parse.file(bytes)
1746    case False{}:
1747      Fail{Malformed{}}
1748
1749def finish.concat.pair(
1750  +total: U32,
1751  pair: Bytes.Bytes & U32
1752) -> Result<&1, &1, Error, Sstable.Table>:
1753  match pair:
1754    case (bytes, +len):
1755      finish.concat.check(bytes, U32.is_eq(total, len))
1756
1757def finish.concat(
1758  +total: U32,
1759  bytes: Bytes.Bytes
1760) -> Result<&1, &1, Error, Sstable.Table>:
1761  finish.concat.pair(total, Bytes.length(bytes))
1762
1763# Parse and authenticate the complete accumulated SST file.
1764def finish(decoder: Decoder) -> Result<&1, &1, Error, Sstable.Table>:
1765  match decoder:
1766    case Dec{chunks, total, False{}}:
1767      finish.concat(total, Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks)))
1768    case Dec{_, _, True{}}:
1769      Fail{TooLarge{}}
1770
1771# Pure whole-file encoder for fixtures and small in-memory callers. Production
1772# persistence uses next_chunk directly and never joins the complete file.
1773def encode.chunks(
1774  fuel: Nat,
1775  pair: Encoder & Maybe<&1, Bytes.Bytes>,
1776  acc: List<&1, Bytes.Bytes>
1777) -> Result<&1, &1, Error, Bytes.Bytes>:
1778  match fuel:
1779    case 0n:
1780      Fail{TooLarge{}}
1781    case 1n+rest:
1782      match pair:
1783        case (_, None{}):
1784          Done{Bytes.concat(List.reverse(&1, Bytes.Bytes, acc))}
1785        case (encoder, Some{chunk}):
1786          encode.chunks(rest, next_chunk(encoder), Con{chunk, acc})
1787
1788def encode.ready(
1789  +entries: List<&2, MemTable.Entry>,
1790  result: Result<&1, &1, Error, Encoder>
1791) -> Result<&1, &1, Error, Bytes.Bytes>:
1792  match result:
1793    case Fail{error}:
1794      Fail{error}
1795    case Done{encoder}:
1796      encode.chunks(Nat.add(List.length(&2, MemTable.Entry, entries), 3n), next_chunk(encoder), Nil{})
1797
1798# Encode one complete table for small in-memory callers and fixtures.
1799def encode_file(
1800  +entries: List<&2, MemTable.Entry>,
1801  +level: U32
1802) -> Result<&1, &1, Error, Bytes.Bytes>:
1803  encode.ready(entries, new_encoder(entries, level))
1804
1805# Parse packed bytes into a validated immutable SST table.
1806def parse(bytes: Bytes.Bytes) -> Result<&1, &1, Error, Sstable.Table>:
1807  parse.file(bytes)
1808
1809# Return true because the parser always returns an explicit Result value.
1810def is_decided(_result: Result<&1, &1, Error, Sstable.Table>) -> Bool:
1811  True{}