src/SstFile.bend source

src/SstFile.bend on the hub · documented module

import Baseimport ./MemTable.bend as MemTableimport ./Keys.bend as Keysimport ./Sstable.bend as Sstableimport ./StorageBytes.bend as StorageBytesimport ./hub_sha/sha256.bend as SHAimport bend-kit-bytes@0.3.2.0/bytes.bend as Bytes# Errors returned by the v3 encoder and strict parser.type Error is Data:  InvalidLevel{}  InvalidOrder{}  TooLarge{}  Malformed{}  InvalidUtf8{}  ChecksumMismatch{}# Encoder progression through the header, blocks, and footer.type Phase is Data:  Header{}  Blocks{}  Finished{}# Serialized block paired with its stored digest.type EncodedBlock is Type:  Block{bytes: Bytes.Bytes, digest: Bytes.Bytes}# State that yields the encoded file in bounded chunks.type Encoder is Type:  Enc{    phase: Phase,    chunks: List<&1, Bytes.Bytes>,    level: U32,    total: U32,    block_count: U32,    footer: Bytes.Bytes  }# Validated level and counts read from the v3 header.type SstHeader is Data:  SstHdr{level: U32, total: U32, block_count: U32}# Streaming parser stage. A completed stage carries only decoded table state.# Bounded chunk accumulator; the final parser rejects aggregate overflow before joining.type Decoder is Type:  Dec{chunks: List<&1, Bytes.Bytes>, total: U32, invalid: Bool}# Result of applying one validated block to file-level counts.type BlockDecision is Type:  BlockRejected{error: Error}  BlockOutOfBounds{}  BlockContinue{cursor: Bytes.Cursor, entries_left: U32,    previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>,    hashes: List<&1, Bytes.Bytes>}  BlockFinish{cursor: Bytes.Cursor, entries: List<&2, MemTable.Entry>,    hashes: List<&1, Bytes.Bytes>}def bytes.clone.pair(  +len: U32,  copied: Array<U32> & Array<U32>) -> Bytes.Bytes & Bytes.Bytes:  match copied:    case (left, right):      (Bytes.Bytes{len, left}, Bytes.Bytes{len, right})def bytes.clone.split(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes:  match bytes:    case Bytes.Bytes{+len, buf}:      bytes.clone.pair(len, Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))def bytes.clone.first(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:  match pair:    case (left, _):      leftdef bytes.clone(bytes: Bytes.Bytes) -> Bytes.Bytes:  bytes.clone.first(bytes.clone.split(bytes))# Construct a packed u32 in network byte order.def u32_bytes(value: U32) -> Bytes.Bytes:  Bytes.set.u32be(Bytes.new(4), 0, value)# Construct a packed tag byte.def u8_bytes(value: U32) -> Bytes.Bytes:  Bytes.set(Bytes.new(1), 0, value)# Unwrap a fixed valid hexadecimal constant.def hex_bytes.fin(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:  match value:    case Some{bytes}:      bytes    case None{}:      Bytes.new(0)# Encode one mutation as a tagged, length-delimited record.def record.value.finish.checked(  key_bytes: Bytes.Bytes,  +key_len: U32,  value_bytes: Bytes.Bytes,  +value_len: U32,  valid: Bool) -> Result<&1, &1, Error, Bytes.Bytes>:  match valid:    case False{}:      Fail{TooLarge{}}    case True{}:      Done{Bytes.concat([u8_bytes(0), u32_bytes(key_len), key_bytes,        u32_bytes(value_len), value_bytes])}def record.value.finish.valid(  key_bytes: Bytes.Bytes,  +key_len: U32,  value_bytes: Bytes.Bytes,  +value_len: U32,  key_fits: Bool) -> Result<&1, &1, Error, Bytes.Bytes>:  match key_fits:    case False{}:      Fail{TooLarge{}}    case True{}:      record.value.finish.checked(key_bytes, key_len, value_bytes, value_len,        U32.is_le(value_len, (StorageBytes.MAX_RECORD_BYTES() - 9 - key_len : U32)))def record.value.finish.val(  key_bytes: Bytes.Bytes,  +key_len: U32,  value_pair: Bytes.Bytes & U32) -> Result<&1, &1, Error, Bytes.Bytes>:  match value_pair:    case (value_bytes, +value_len):      record.value.finish.valid(key_bytes, key_len, value_bytes, value_len,        U32.is_le(key_len, (StorageBytes.MAX_RECORD_BYTES() - 9 : U32)))def record.value.finish(  key_pair: Bytes.Bytes & U32,  value_pair: Bytes.Bytes & U32) -> Result<&1, &1, Error, Bytes.Bytes>:  match key_pair:    case (key_bytes, key_len):      record.value.finish.val(key_bytes, key_len, value_pair)def record.value.fin(  key_pair: Bytes.Bytes & U32,  result: Result<&1, &1, U32 & String, Bytes.Bytes>) -> Result<&1, &1, Error, Bytes.Bytes>:  match result:    case Fail{_}:      Fail{TooLarge{}}    case Done{value_bytes}:      record.value.finish(key_pair, Bytes.length(value_bytes))def record.value(  key_bytes: Bytes.Bytes,  value_result: Result<&1, &1, U32 & String, Bytes.Bytes>) -> Result<&1, &1, Error, Bytes.Bytes>:  record.value.fin(Bytes.length(key_bytes), value_result)def record.key.delete.checked(  key_bytes: Bytes.Bytes,  +key_len: U32,  valid: Bool) -> Result<&1, &1, Error, Bytes.Bytes>:  match valid:    case False{}:      Fail{TooLarge{}}    case True{}:      Done{Bytes.concat([u8_bytes(1), u32_bytes(key_len), key_bytes])}def record.key.delete(key_pair: Bytes.Bytes & U32) -> Result<&1, &1, Error, Bytes.Bytes>:  match key_pair:    case (key_bytes, +key_len):      record.key.delete.checked(key_bytes, key_len,        U32.is_le(key_len, (StorageBytes.MAX_RECORD_BYTES() - 5 : U32)))def record.key(  key_result: Result<&1, &1, U32 & String, Bytes.Bytes>,  val: Maybe<&2, String>) -> Result<&1, &1, Error, Bytes.Bytes>:  match key_result val:    case Fail{_} _:      Fail{TooLarge{}}    case Done{key_bytes} None{}:      record.key.delete(Bytes.length(key_bytes))    case Done{key_bytes} Some{value}:      record.value(key_bytes, StorageBytes.from_string(value))# Encode one checked put or delete record.def encode_record(entry: MemTable.Entry) -> Result<&1, &1, Error, Bytes.Bytes>:  match entry:    case MemTable.Entry{key, val}:      record.key(StorageBytes.from_string(key), val)def entries.strict.tail(  tail: List<&2, MemTable.Entry>,  current: String,  ordered: Bool) -> Bool:  match tail ordered:    case Nil{} True{}:      True{}    case Nil{} False{}:      False{}    case Con{_entry, _rest} False{}:      False{}    case Con{MemTable.Entry{+key, val}, rest} True{}:      entries.strict.tail(rest, key, Keys.lt(current, key))def entries.strict(entries: List<&2, MemTable.Entry>) -> Bool:  match entries:    case Nil{}:      True{}    case Con{MemTable.Entry{key, val}, rest}:      entries.strict.tail(rest, key, True{})# Result of measuring a record against the target block size.type BlockMeasure is Data:  InvalidRecord{}  Measured{length: U32, fits: Bool}# Measure an already encoded record for byte-bounded block packing.def blocks.measure.length(+length: U32, +size: U32) -> BlockMeasure:  Measured{length, U32.is_eq(size, 0) ||    U32.is_le((size + length : U32), StorageBytes.MAX_BLOCK_BYTES())}def blocks.measure.valid(length: U32, +size: U32, valid: Bool) -> BlockMeasure:  match valid:    case False{}:      InvalidRecord{}    case True{}:      blocks.measure.length(length, size)def blocks.measure.delete(key_length: Maybe<&2, U32>, +size: U32) -> BlockMeasure:  match key_length:    case None{}:      InvalidRecord{}    case Some{+length}:      blocks.measure.valid((5 + length : U32), size,        U32.is_le(length, (StorageBytes.MAX_RECORD_BYTES() - 5 : U32)))def blocks.measure.put.value(  +key_length: U32,  value_length: Maybe<&2, U32>,  +size: U32) -> BlockMeasure:  match value_length:    case None{}:      InvalidRecord{}    case Some{+length}:      blocks.measure.valid((9 + key_length + length : U32), size,        U32.is_le(key_length, (StorageBytes.MAX_RECORD_BYTES() - 9 : U32)) &&        U32.is_le(length, (StorageBytes.MAX_RECORD_BYTES() - 9 - key_length : U32)))def blocks.measure.put(  key_length: Maybe<&2, U32>,  value_length: Maybe<&2, U32>,  +size: U32) -> BlockMeasure:  match key_length:    case None{}:      InvalidRecord{}    case Some{+length}:      blocks.measure.put.value(length, value_length, size)def blocks.measure.entry(entry: MemTable.Entry, +size: U32) -> BlockMeasure:  match entry:    case MemTable.Entry{key, value}:      match value:        case None{}:          blocks.measure.delete(StorageBytes.from_string.size.begin(key), size)        case Some{text}:          blocks.measure.put(StorageBytes.from_string.size.begin(key),            StorageBytes.from_string.size.begin(text), size)def blocks.measure.next(  entries: List<&2, MemTable.Entry>,  +size: U32) -> BlockMeasure:  match entries:    case Nil{}:      InvalidRecord{}    case Con{MemTable.Entry{key, val}, _}:      blocks.measure.entry(MemTable.Entry{key, val}, size)# Finish the current block list, which is stored in reverse order.def blocks.pack.finish(  current: List<&2, MemTable.Entry>,  acc: List<&2, List<&2, MemTable.Entry>>) -> List<&2, List<&2, MemTable.Entry>>:  match current:    case Nil{}:      List.reverse(&2, List<&2, MemTable.Entry>, acc)    case Con{_, _}:      List.reverse(&2, List<&2, MemTable.Entry>,        Con{List.reverse(&2, MemTable.Entry, current), acc})# Pack records into blocks capped by encoded payload bytes, preserving order.def blocks.pack.go(  entries: List<&2, MemTable.Entry>,  current: List<&2, MemTable.Entry>,  +size: U32,  acc: List<&2, List<&2, MemTable.Entry>>,  measured: BlockMeasure) -> List<&2, List<&2, MemTable.Entry>>:  match entries:    case Nil{}:      blocks.pack.finish(current, acc)    case Con{entry, +rest}:      match current:        case Nil{}:          match measured:            case InvalidRecord{}:              blocks.pack.go(rest, Con{entry, Nil{}}, StorageBytes.MAX_BLOCK_BYTES(), acc,                blocks.measure.next(rest, StorageBytes.MAX_BLOCK_BYTES()))            case Measured{+length, fits}:              blocks.pack.go(rest, Con{entry, Nil{}}, length, acc,                blocks.measure.next(rest, length))        case Con{_, _}:          match measured:            case InvalidRecord{}:              blocks.pack.go(rest, Con{entry, Nil{}}, StorageBytes.MAX_BLOCK_BYTES(),                Con{List.reverse(&2, MemTable.Entry, current), acc},                blocks.measure.next(rest, StorageBytes.MAX_BLOCK_BYTES()))            case Measured{+length, False{}}:              blocks.pack.go(rest, Con{entry, Nil{}}, length,                Con{List.reverse(&2, MemTable.Entry, current), acc},                blocks.measure.next(rest, length))            case Measured{+length, True{}}:              blocks.pack.go(rest, Con{entry, current}, (size + length : U32), acc,                blocks.measure.next(rest, (size + length : U32)))def blocks.pack(entries: List<&2, MemTable.Entry>) -> List<&2, List<&2, MemTable.Entry>>:  match entries:    case Nil{}:      Nil{}    case Con{+entry, _}:      blocks.pack.go(entries, Nil{}, 0, Nil{},        blocks.measure.entry(entry, 0))# Continue encoding a record list after the current record has been checked.def records.step(  rest: List<&2, MemTable.Entry>,  current: Result<&1, &1, Error, Bytes.Bytes>,  acc: List<&1, Bytes.Bytes>) -> Result<&1, &1, Error, List<&1, Bytes.Bytes>>:  match rest current:    case Nil{} Fail{error}:      Fail{error}    case Nil{} Done{bytes}:      Done{List.reverse(&1, Bytes.Bytes, bytes <> acc)}    case Con{_entry, _tail} Fail{error}:      Fail{error}    case Con{entry, tail} Done{bytes}:      records.step(tail, encode_record(entry), bytes <> acc)# Encode each entry independently, retaining only the record fragments for one block.def records(entries: List<&2, MemTable.Entry>) -> Result<&1, &1, Error, List<&1, Bytes.Bytes>>:  match entries:    case Nil{}:      Done{Nil{}}    case Con{entry, +rest}:          records.step(rest, encode_record(entry), Nil{})# Build one v3 block and its digest over the two header words and payload.def block.fin.digest.parts(  +count: U32,  payload: Bytes.Bytes,  +payload_len: U32,  pair: Bytes.Bytes & Bytes.Bytes) -> EncodedBlock:  match pair:    case (output_digest, stored_digest):      Block{Bytes.concat([u32_bytes(payload_len), u32_bytes(count), payload,        output_digest]), stored_digest}def block.fin.digest(  +count: U32,  pair: Bytes.Bytes & Bytes.Bytes,  payload: Bytes.Bytes,  +payload_len: U32) -> EncodedBlock:  match pair:    case (_, _):      block.fin.digest.parts(count, payload, payload_len, pair)def block.fin.payload(  +count: U32,  +payload_len: U32,  pair: Bytes.Bytes & Bytes.Bytes) -> EncodedBlock:  match pair:    case (output_payload, hash_payload):      block.fin.digest(count, bytes.clone.split(        SHA.sha256_packed_bytes(Bytes.concat([u32_bytes(payload_len),          u32_bytes(count), hash_payload]))), output_payload, payload_len)def block.fin(  +count: U32,  +payload_len: U32,  payload: Bytes.Bytes) -> EncodedBlock:  block.fin.payload(count, payload_len, bytes.clone.split(payload))def block.payload.check(  +count: U32,  +payload_len: U32,  payload: Bytes.Bytes,  within: Bool) -> Result<&1, &1, Error, EncodedBlock>:  match within:    case True{}:      Done{block.fin(count, payload_len, payload)}    case False{}:      Fail{TooLarge{}}def block.payload.total(  +count: U32,  payload_pair: Bytes.Bytes & U32) -> Result<&1, &1, Error, EncodedBlock>:  match payload_pair:    case (payload, +payload_len):      block.payload.check(count, payload_len, payload,        U32.is_le(payload_len, StorageBytes.MAX_BLOCK_BYTES()) ||        U32.is_eq(count, 1) && U32.is_le(payload_len, StorageBytes.MAX_RECORD_BYTES()))def block.payload(  count: U32,  result: Result<&1, &1, Error, List<&1, Bytes.Bytes>>) -> Result<&1, &1, Error, EncodedBlock>:  match result:    case Fail{error}:      Fail{error}    case Done{fragments}:      block.payload.total(count, Bytes.length(Bytes.concat(fragments)))# Encode records and compute the block digest.def encode_block(+entries: List<&2, MemTable.Entry>) -> Result<&1, &1, Error, EncodedBlock>:  block.payload(U32.from_nat(List.length(&2, MemTable.Entry, entries)), records(entries))# Store a block digest and continue with the remaining blocks.def blocks.prepare.step(  rest: List<&2, List<&2, MemTable.Entry>>,  current: Result<&1, &1, Error, EncodedBlock>,  blocks: List<&1, Bytes.Bytes>,  digests: List<&1, Bytes.Bytes>) -> Result<&1, &1, Error, List<&1, Bytes.Bytes> & List<&1, Bytes.Bytes>>:  match rest current:    case Nil{} Fail{error}:      Fail{error}    case Nil{} Done{Block{bytes, digest}}:      Done{(List.reverse(&1, Bytes.Bytes, bytes <> blocks),        List.reverse(&1, Bytes.Bytes, digest <> digests))}    case Con{_entries, _tail} Fail{error}:      Fail{error}    case Con{entries, tail} Done{Block{bytes, digest}}:      blocks.prepare.step(tail, encode_block(entries), bytes <> blocks, digest <> digests)def blocks.prepare(  +chunks: List<&2, List<&2, MemTable.Entry>>) -> Result<&1, &1, Error, List<&1, Bytes.Bytes> & List<&1, Bytes.Bytes>>:  match chunks:    case Nil{}:      Done{(Nil{}, Nil{})}    case Con{entries, rest}:      blocks.prepare.step(rest, encode_block(entries), Nil{}, Nil{})# Encode fixed-width v3 header fields in network byte order.def header_bytes(  +level: U32,  +total: U32,  +block_count: U32) -> Bytes.Bytes:  Bytes.concat([hex_bytes.fin(Bytes.from_hex("4d594c534d335300")),    u32_bytes(level), u32_bytes(total), u32_bytes(block_count)])# Propagate the block-summary result into an encoder header and footer.def encoder.build(  +entries: List<&2, MemTable.Entry>,  +level: U32,  +block_count: U32,  prepared: Result<&1, &1, Error, List<&1, Bytes.Bytes> & List<&1, Bytes.Bytes>>) -> Result<&1, &1, Error, Encoder>:  match prepared:    case Fail{error}:      Fail{error}    case Done{(chunks, hashes)}:      +total = U32.from_nat(List.length(&2, MemTable.Entry, entries))      root = SHA.sha256_packed_bytes(Bytes.concat(header_bytes(level, total, block_count) <> hashes))      footer = Bytes.concat([hex_bytes.fin(Bytes.from_hex("454e4433")), u32_bytes(total), root])      Done{Enc{Header{}, chunks, level, total, block_count, footer}}# Create an encoder only for supported levels and bounded records.def new_encoder.ready(  +entries: List<&2, MemTable.Entry>,  +level: U32,  +chunks: List<&2, List<&2, MemTable.Entry>>) -> Result<&1, &1, Error, Encoder>:  encoder.build(entries, level,    U32.from_nat(List.length(&2, List<&2, MemTable.Entry>, chunks)), blocks.prepare(chunks))def new_encoder.order(  +entries: List<&2, MemTable.Entry>,  +level: U32,  ordered: Bool) -> Result<&1, &1, Error, Encoder>:  match ordered:    case False{}:      Fail{InvalidOrder{}}    case True{}:      new_encoder.ready(entries, level, blocks.pack(entries))def new_encoder.valid(  +entries: List<&2, MemTable.Entry>,  +level: U32,  valid: Bool) -> Result<&1, &1, Error, Encoder>:  match valid:    case False{}:      Fail{InvalidLevel{}}    case True{}:      new_encoder.order(entries, level, entries.strict(entries))# Validate input order and level before encoding.def new_encoder(  +entries: List<&2, MemTable.Entry>,  +level: U32) -> Result<&1, &1, Error, Encoder>:  new_encoder.valid(entries, level, U32.is_le(level, 255))# Yield the next serialized part and advance encoder state.def next_chunk(encoder: Encoder) -> Encoder & Maybe<&1, Bytes.Bytes>:  match encoder:    case Enc{Header{}, chunks, +level, +total, +block_count, footer}:      (Enc{Blocks{}, chunks, level, total, block_count, footer}, Some{header_bytes(level, total, block_count)})    case Enc{Blocks{}, Nil{}, level, total, block_count, footer}:      (Enc{Finished{}, Nil{}, level, total, block_count, Bytes.new(0)}, Some{footer})    case Enc{Blocks{}, Con{bytes, rest}, level, total, block_count, footer}:      (Enc{Blocks{}, rest, level, total, block_count, footer}, Some{bytes})    case Enc{Finished{}, chunks, level, total, block_count, footer}:      (Enc{Finished{}, chunks, level, total, block_count, footer}, None{})# Translate the shared strict UTF-8 result into the SST error domain.type TextRead is Type:  Text{cursor: Bytes.Cursor, byte: Maybe<&2, U32>}def parse.text.wrap(pair: Bytes.Cursor & Maybe<&2, U32>) -> TextRead:  match pair:    case (cursor, byte):      Text{cursor, byte}def parse.text.finish(  cursor: Bytes.Cursor,  result: Result<&1, &1, U32 & String, String>) -> Bytes.Cursor & Result<&1, &1, Error, String>:  match result:    case Fail{_}:      (cursor, Fail{InvalidUtf8{}})    case Done{text}:      (cursor, Done{text})# Read exactly n UTF-8 bytes from a bounded cursor without materializing a slice.def parse.text.go(  remaining: Nat,  read: TextRead,  state: StorageBytes.Utf8State) -> Bytes.Cursor & Result<&1, &1, Error, String>:  match remaining read:    case 0n Text{cursor, _}:      parse.text.finish(cursor, StorageBytes.utf8.scan.fin(state))    case 1n+0n Text{cursor, Some{+byte}}:      parse.text.finish(cursor, StorageBytes.utf8.scan.fin(        StorageBytes.utf8.step(state, byte)))    case 1n+1n+rest Text{cursor, Some{+byte}}:      parse.text.go(1n+rest, parse.text.wrap(Bytes.Cursor.u8(cursor)),        StorageBytes.utf8.step(state, byte))    case 1n+_ Text{cursor, None{}}:      (cursor, Fail{Malformed{}})def parse.text.start(  cursor: Bytes.Cursor,  +length: U32,  empty: Bool) -> Bytes.Cursor & Result<&1, &1, Error, String>:  match empty:    case True{}:      parse.text.go(0n, Text{cursor, Some{0}},        StorageBytes.Utf8State{0, 128, 191, 0, SNil{}, True{}})    case False{}:      parse.text.go(U32.to_nat(length), parse.text.wrap(Bytes.Cursor.u8(cursor)),        StorageBytes.Utf8State{0, 128, 191, 0, SNil{}, True{}})def parse.text(  cursor: Bytes.Cursor,  +length: U32) -> Bytes.Cursor & Result<&1, &1, Error, String>:  parse.text.start(cursor, length, U32.is_eq(length, 0))# Parsed mutation paired with its encoded byte length.type ParsedRecord is Type:  Rec{entry: MemTable.Entry, bytes_used: U32}# Decode one length-delimited key and optional value from the cursor.def parse.record.value.decoded(  pair: Bytes.Cursor & Result<&1, &1, Error, String>,  +key_len: U32,  +key: String,  +value_len: U32) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:  match pair:    case (cursor, Fail{error}):      (cursor, Fail{error})    case (cursor, Done{value}):      (cursor, Done{Rec{MemTable.Entry{key, Some{value}},        (9 + key_len + value_len : U32)}})def parse.record.value.length.checked(  cursor: Bytes.Cursor,  +key_len: U32,  +key: String,  +value_len: U32,  valid: Bool) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:  match valid:    case False{}:      (cursor, Fail{TooLarge{}})    case True{}:      parse.record.value.decoded(parse.text(cursor, value_len), key_len,        key, value_len)def parse.record.value.length(  pair: Bytes.Cursor & Maybe<&2, U32>,  +key_len: U32,  +key: String) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:  match pair:    case (cursor, Some{+value_len}):      parse.record.value.length.checked(cursor, key_len, key, value_len,        U32.is_le(value_len, (StorageBytes.MAX_RECORD_BYTES() - 9 - key_len : U32)))    case (cursor, None{}):      (cursor, Fail{Malformed{}})def parse.record.key.ready(  cursor: Bytes.Cursor,  tag: U32,  +key_len: U32,  +key: String) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:  match tag:    case 1:      (cursor, Done{Rec{MemTable.Entry{key, None{}}, (5 + key_len : U32)}})    case 0:      parse.record.value.length(Bytes.Cursor.u32be(cursor), key_len, key)    case _:      (cursor, Fail{Malformed{}})def parse.record.key.decoded(  pair: Bytes.Cursor & Result<&1, &1, Error, String>,  tag: U32,  +key_len: U32) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:  match pair:    case (cursor, Fail{error}):      (cursor, Fail{error})    case (cursor, Done{key}):      parse.record.key.ready(cursor, tag, key_len, key)def parse.record.key.limit(  tag: U32,  +key_len: U32) -> Bool:  match tag:    case 1:      U32.is_le(key_len, (StorageBytes.MAX_RECORD_BYTES() - 5 : U32))    case _:      U32.is_le(key_len, (StorageBytes.MAX_RECORD_BYTES() - 9 : U32))def parse.record.key.length.checked(  cursor: Bytes.Cursor,  tag: U32,  +key_len: U32,  valid: Bool) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:  match valid:    case False{}:      (cursor, Fail{TooLarge{}})    case True{}:      parse.record.key.decoded(parse.text(cursor, key_len), tag, key_len)def parse.record.key.length(  pair: Bytes.Cursor & Maybe<&2, U32>,  +tag: U32) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:  match pair:    case (cursor, Some{+key_len}):      parse.record.key.length.checked(cursor, tag, key_len,        parse.record.key.limit(tag, key_len))    case (cursor, None{}):      (cursor, Fail{Malformed{}})def parse.record.tag(  pair: Bytes.Cursor & Maybe<&2, U32>) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:  match pair:    case (cursor, Some{0}):      parse.record.key.length(Bytes.Cursor.u32be(cursor), 0)    case (cursor, Some{1}):      parse.record.key.length(Bytes.Cursor.u32be(cursor), 1)    case (cursor, Some{_}):      (cursor, Fail{Malformed{}})    case (cursor, None{}):      (cursor, Fail{Malformed{}})def parse.record(  cursor: Bytes.Cursor) -> Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>:  parse.record.tag(Bytes.Cursor.u8(cursor))# Parsed entries in reverse order with the last strict-order key.type ParsedEntries is Type:  Entries{reversed: List<&2, MemTable.Entry>, previous: Maybe<&2, String>}# Record result annotated with block bounds and key order.type RecordRead is Type:  Record{cursor: Bytes.Cursor,    result: Result<&1, &1, Error, ParsedRecord>,    within: Bool, ordered: Bool, last: Bool, complete: Bool, key: String}# Decision for a record at its declared block position.type RecordDecision is Type:  Rejected{cursor: Bytes.Cursor, error: Error}  OutOfBounds{cursor: Bytes.Cursor}  OutOfOrder{cursor: Bytes.Cursor}  LastMalformed{cursor: Bytes.Cursor}  LastRecord{cursor: Bytes.Cursor, entry: MemTable.Entry, key: String}  MoreRecords{cursor: Bytes.Cursor, entry: MemTable.Entry, key: String, used: U32}def parse.entries.ordered(  previous: Maybe<&2, String>,  +key: String) -> Bool:  match previous:    case None{}:      True{}    case Some{current}:      Keys.lt(current, key)def parse.record.read.entry(  cursor: Bytes.Cursor,  +entry: MemTable.Entry,  +bytes_used: U32,  +payload_left: U32,  previous: Maybe<&2, String>,  last: Bool) -> RecordRead:  match entry:    case MemTable.Entry{+key, _}:      Record{cursor, Done{Rec{entry, bytes_used}},        U32.is_le(bytes_used, payload_left), parse.entries.ordered(previous, key),        last, U32.is_eq(bytes_used, payload_left), key}def parse.record.read.pair(  pair: Bytes.Cursor & Result<&1, &1, Error, ParsedRecord>,  payload_left: U32,  previous: Maybe<&2, String>,  last: Bool) -> RecordRead:  match pair:    case (cursor, Fail{error}):      Record{cursor, Fail{error}, False{}, False{}, last, False{}, ""}    case (cursor, Done{Rec{entry, +bytes_used}}):      parse.record.read.entry(cursor, entry, bytes_used, payload_left, previous, last)def parse.record.read(  cursor: Bytes.Cursor,  +payload_left: U32,  previous: Maybe<&2, String>,  last: Bool) -> RecordRead:  parse.record.read.pair(parse.record(cursor), payload_left, previous, last)def parse.records.decision(read: RecordRead) -> RecordDecision:  match read:    case Record{cursor, Fail{error}, _, _, _, _, _}:      Rejected{cursor, error}    case Record{cursor, Done{_}, False{}, _, _, _, _}:      OutOfBounds{cursor}    case Record{cursor, Done{_}, True{}, False{}, _, _, _}:      OutOfOrder{cursor}    case Record{cursor, Done{_}, True{}, True{}, True{}, False{}, _}:      LastMalformed{cursor}    case Record{cursor, Done{Rec{entry, _}}, True{}, True{}, True{}, True{}, +key}:      LastRecord{cursor, entry, key}    case Record{cursor, Done{Rec{entry, +used}}, True{}, True{}, False{}, _, +key}:      MoreRecords{cursor, entry, key, used}# Parse a counted sequence and enforce block boundaries and strict key order.def parse.records.empty(  decision: RecordDecision) -> Bytes.Cursor & Result<&1, &1, Error, ParsedEntries>:  match decision:    case Rejected{cursor, _}:      (cursor, Fail{Malformed{}})    case OutOfBounds{cursor}:      (cursor, Fail{Malformed{}})    case OutOfOrder{cursor}:      (cursor, Fail{Malformed{}})    case LastMalformed{cursor}:      (cursor, Fail{Malformed{}})    case LastRecord{cursor, _, _}:      (cursor, Fail{Malformed{}})    case MoreRecords{cursor, _, _, _}:      (cursor, Fail{Malformed{}})def parse.records.go(  remaining: Nat,  decision: RecordDecision,  +payload_left: U32,  acc: List<&2, MemTable.Entry>) -> Bytes.Cursor & Result<&1, &1, Error, ParsedEntries>:  match remaining:    case 0n:      parse.records.empty(decision)    case 1n++rest:      match decision:        case Rejected{cursor, error}:          (cursor, Fail{error})        case OutOfBounds{cursor}:          (cursor, Fail{Malformed{}})        case OutOfOrder{cursor}:          (cursor, Fail{InvalidOrder{}})        case LastMalformed{cursor}:          (cursor, Fail{Malformed{}})        case LastRecord{cursor, entry, +key}:          (cursor, Done{Entries{Con{entry, acc}, Some{key}}})        case MoreRecords{cursor, entry, +key, +used}:          +payload_after = (payload_left - used : U32)          parse.records.go(rest,            parse.records.decision(parse.record.read(cursor, payload_after,              Some{key}, Nat.is_eq(rest, 1n))),            payload_after, Con{entry, acc})def parse.records.zero(  cursor: Bytes.Cursor,  previous: Maybe<&2, String>,  empty: Bool) -> Bytes.Cursor & Result<&1, &1, Error, ParsedEntries>:  match empty:    case True{}:      (cursor, Done{Entries{Nil{}, previous}})    case False{}:      (cursor, Fail{Malformed{}})def parse.records.start(  cursor: Bytes.Cursor,  +count: U32,  +payload_len: U32,  previous: Maybe<&2, String>) -> Bytes.Cursor & Result<&1, &1, Error, ParsedEntries>:  match count:    case 0:      parse.records.zero(cursor, previous, U32.is_eq(payload_len, 0))    case _:      parse.records.go(U32.to_nat(count),        parse.records.decision(parse.record.read(cursor, payload_len, previous, U32.is_eq(count, 1))),        payload_len, Nil{})# Cursor state while assembling a fixed-width digest.type HashRead is Type:  HashInput{cursor: Bytes.Cursor, value: Maybe<&2, U32>}# Validated records, count, and digest for one block.type ParsedBlock is Type:  ParsedBlock{records: ParsedEntries, count: U32, digest: Bytes.Bytes}# Represent StreamState data used by the packed SSTable codec.type StreamState is Type:  StreamState{level: U32, total: U32, block_count: U32,    remaining: U32, entries_left: U32,    previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>,    hashes: List<&1, Bytes.Bytes>}def parse.hash.wrap(pair: Bytes.Cursor & Maybe<&2, U32>) -> HashRead:  match pair:    case (cursor, value):      HashInput{cursor, value}def parse.hash.result(  pair: Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:  pair# Read exactly eight words into a digest without consuming a byte after it.def parse.hash.words.go(  remaining: Nat,  read: HashRead,  digest: Bytes.Bytes,  +offset: U32) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:  match remaining read:    case 0n HashInput{cursor, _}:      (cursor, Done{digest})    case 1n+0n HashInput{cursor, Some{word}}:      (cursor, Done{Bytes.set.u32be(digest, offset, word)})    case 1n+1n+rest HashInput{cursor, Some{word}}:      parse.hash.words.go(1n+rest, parse.hash.wrap(Bytes.Cursor.u32be(cursor)),        Bytes.set.u32be(digest, offset, word), (offset + 4 : U32))    case 1n+_ HashInput{cursor, None{}}:      (cursor, Fail{Malformed{}})def parse.hash.words(  cursor: Bytes.Cursor) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:  parse.hash.words.go(8n, parse.hash.wrap(Bytes.Cursor.u32be(cursor)), Bytes.new(32), 0)def parse.block.verify.equal(  cursor: Bytes.Cursor,  records: ParsedEntries,  +count: U32,  comparison: Bytes.Bytes & Bytes.Bytes & Bool) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:  match comparison:    case (claimed, actual, False{}):      (cursor, Fail{ChecksumMismatch{}})    case (claimed, digest, True{}):      (cursor, Done{ParsedBlock{records, count, digest}})def parse.block.verify.copy(  pair: Bytes.Bytes & Bytes.Bytes,  +pos: U32,  +start: U32,  +end: U32,  records: ParsedEntries,  +count: U32,  claimed: Bytes.Bytes) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:  match pair:    case (source, content):      parse.block.verify.equal(Bytes.Cursor{source, pos, start, end}, records,        count, Bytes.eq(claimed, SHA.sha256_packed_bytes(content)))def parse.block.verify(  cursor: Bytes.Cursor,  records: ParsedEntries,  +payload_len: U32,  +count: U32,  claimed: Bytes.Bytes) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:  match cursor:    case Bytes.Cursor{bytes, +pos, +start, +end}:      parse.block.verify.copy(Bytes.slice(bytes,        (pos - (payload_len + 40 : U32) : U32), (payload_len + 8 : U32)),        pos, start, end, records, count, claimed)def parse.block.digest(  pair: Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>,  records: ParsedEntries,  +payload_len: U32,  +count: U32) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:  match pair:    case (cursor, Fail{error}):      (cursor, Fail{error})    case (cursor, Done{claimed}):      parse.block.verify(cursor, records, payload_len, count, claimed)def parse.block.records(  pair: Bytes.Cursor & Result<&1, &1, Error, ParsedEntries>,  +payload_len: U32,  +count: U32) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:  match pair:    case (cursor, Fail{error}):      (cursor, Fail{error})    case (cursor, Done{records}):      parse.block.digest(parse.hash.words(cursor), records, payload_len, count)def parse.block.count.valid(  cursor: Bytes.Cursor,  +payload_len: U32,  +count: U32,  previous: Maybe<&2, String>,  valid: Bool) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:  match valid:    case False{}:      (cursor, Fail{Malformed{}})    case True{}:      parse.block.records(parse.records.start(cursor, count, payload_len, previous),        payload_len, count)def parse.block.count(  pair: Bytes.Cursor & Maybe<&2, U32>,  +payload_len: U32,  previous: Maybe<&2, String>) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:  match pair:    case (cursor, Some{+count}):      parse.block.count.valid(cursor, payload_len, count, previous,        U32.is_lt(0, count) && (U32.is_le(payload_len, StorageBytes.MAX_BLOCK_BYTES()) ||          U32.is_eq(count, 1) && U32.is_le(payload_len, StorageBytes.MAX_RECORD_BYTES())))    case (cursor, None{}):      (cursor, Fail{Malformed{}})def parse.block.length(  pair: Bytes.Cursor & Maybe<&2, U32>,  previous: Maybe<&2, String>) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:  match pair:    case (cursor, Some{+payload_len}):      parse.block.count(Bytes.Cursor.u32be(cursor), payload_len, previous)    case (cursor, None{}):      (cursor, Fail{Malformed{}})# Parse, bound, and authenticate one complete SST block.def parse.block(  cursor: Bytes.Cursor,  previous: Maybe<&2, String>) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:  parse.block.length(Bytes.Cursor.u32be(cursor), previous)def parse.footer.end(  +level: U32,  entries: List<&2, MemTable.Entry>,  at_end: Bool) -> Result<&1, &1, Error, Sstable.Table>:  match at_end:    case False{}:      Fail{Malformed{}}    case True{}:      Done{Sstable.from_sorted_unique(        List.reverse(&2, MemTable.Entry, entries), U32.to_nat(level))}def parse.footer.check(  +level: U32,  entries: List<&2, MemTable.Entry>,  at_end: Bool,  comparison: Bytes.Bytes & Bytes.Bytes & Bool) -> Result<&1, &1, Error, Sstable.Table>:  match comparison:    case (claimed_back, expected_back, False{}):      Fail{ChecksumMismatch{}}    case (claimed_back, expected_back, True{}):      parse.footer.end(level, entries, at_end)def parse.footer.verify(  cursor: Bytes.Cursor,  claimed: Bytes.Bytes,  +level: U32,  +total: U32,  +block_count: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>) -> Result<&1, &1, Error, Sstable.Table>:  match cursor:    case Bytes.Cursor{bytes, +pos, _, +end}:      expected = SHA.sha256_packed_bytes(Bytes.concat(        header_bytes(level, total, block_count) <> List.reverse(&1, Bytes.Bytes, hashes)))      parse.footer.check(level, entries, U32.is_eq(pos, end),        Bytes.eq(claimed, expected))def parse.footer.root(  pair: Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>,  +level: U32,  +total: U32,  +block_count: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>) -> Result<&1, &1, Error, Sstable.Table>:  match pair:    case (_, Fail{error}):      Fail{error}    case (cursor, Done{claimed}):      parse.footer.verify(cursor, claimed, level, total, block_count, entries, hashes)def parse.footer.count.checked(  cursor: Bytes.Cursor,  +level: U32,  +total: U32,  +block_count: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>,  valid: Bool) -> Result<&1, &1, Error, Sstable.Table>:  match valid:    case False{}:      Fail{Malformed{}}    case True{}:      parse.footer.root(parse.hash.words(cursor), level, total, block_count,        entries, hashes)def parse.footer.count(  pair: Bytes.Cursor & Maybe<&2, U32>,  +level: U32,  +total: U32,  +block_count: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>) -> Result<&1, &1, Error, Sstable.Table>:  match pair:    case (cursor, Some{+footer_total}):      parse.footer.count.checked(cursor, level, total, block_count, entries,        hashes, U32.is_eq(total, footer_total))    case (_, None{}):      Fail{Malformed{}}def parse.footer.magic(  pair: Bytes.Cursor & Maybe<&2, U32>,  +level: U32,  +total: U32,  +block_count: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>) -> Result<&1, &1, Error, Sstable.Table>:  match pair:    case (cursor, Some{1162757171}):      parse.footer.count(Bytes.Cursor.u32be(cursor), level, total,        block_count, entries, hashes)    case (_, _):      Fail{Malformed{}}# Block result paired with its ending cursor.type BlockRead is Type:  BlockInput{cursor: Bytes.Cursor, result: Result<&1, &1, Error, ParsedBlock>}def parse.block.read(pair: Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>) -> BlockRead:  match pair:    case (cursor, result):      BlockInput{cursor, result}def parse.blocks.count.valid.last(  last: Bool,  +entries_left: U32,  +count: U32) -> Bool:  match last:    case True{}:      U32.is_eq(entries_left, count)    case False{}:      U32.is_lt(count, entries_left)def parse.blocks.count.valid(  +remaining: Nat,  +entries_left: U32,  +count: U32) -> Bool:  parse.blocks.count.valid.last(Nat.is_eq(remaining, 1n), entries_left, count)def parse.blocks.accumulate.records.last(  cursor: Bytes.Cursor,  +entries_left: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>,  reversed: List<&2, MemTable.Entry>,  previous: Maybe<&2, String>,  last: Bool) -> BlockDecision:  match last:    case True{}:      BlockFinish{cursor, List.append(&2, MemTable.Entry, reversed, entries), hashes}    case False{}:      BlockContinue{cursor, entries_left, previous,        List.append(&2, MemTable.Entry, reversed, entries), hashes}def parse.blocks.accumulate.records(  cursor: Bytes.Cursor,  +remaining: Nat,  +entries_left: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>,  records: ParsedEntries) -> BlockDecision:  match records:    case Entries{reversed, previous}:      parse.blocks.accumulate.records.last(cursor, entries_left, entries,        hashes, reversed, previous, Nat.is_eq(remaining, 1n))def parse.blocks.content.valid(  cursor: Bytes.Cursor,  +remaining: Nat,  +entries_left: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>,  +count: U32,  records: ParsedEntries,  digest: Bytes.Bytes,  valid: Bool) -> BlockDecision:  match valid:    case False{}:      BlockOutOfBounds{}    case True{}:      parse.blocks.accumulate.records(cursor, remaining,        (entries_left - count : U32), entries, digest <> hashes, records)def parse.blocks.content.fit(  cursor: Bytes.Cursor,  +remaining: Nat,  +entries_left: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>,  +count: U32,  records: ParsedEntries,  digest: Bytes.Bytes,  fits: Bool) -> BlockDecision:  match fits:    case False{}:      BlockOutOfBounds{}    case True{}:      parse.blocks.content.valid(cursor, remaining, entries_left,        entries, hashes, count, records, digest,        parse.blocks.count.valid(remaining, entries_left, count))def parse.blocks.content(  cursor: Bytes.Cursor,  +remaining: Nat,  +entries_left: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>,  block: ParsedBlock) -> BlockDecision:  match block:    case ParsedBlock{records, +count, digest}:      parse.blocks.content.fit(cursor, remaining, entries_left,        entries, hashes, count, records, digest, U32.is_le(count, entries_left))def parse.blocks.decision(  read: BlockRead,  +remaining: Nat,  +entries_left: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>) -> BlockDecision:  match read:    case BlockInput{_, Fail{error}}:      BlockRejected{error}    case BlockInput{cursor, Done{block}}:      parse.blocks.content(cursor, remaining, entries_left, entries,        hashes, block)def parse.blocks.go(  remaining: Nat,  decision: BlockDecision,  +level: U32,  +total: U32,  +block_count: U32) -> Result<&1, &1, Error, Sstable.Table>:  match remaining:    case 0n:      Fail{Malformed{}}    case 1n++rest:      match decision:        case BlockRejected{error}:          Fail{error}        case BlockOutOfBounds{}:          Fail{Malformed{}}        case BlockFinish{cursor, entries, hashes}:          parse.footer.magic(Bytes.Cursor.u32be(cursor), level, total,            block_count, entries, hashes)        case BlockContinue{cursor, entries_left, previous, entries, hashes}:          parse.blocks.go(rest,            parse.blocks.decision(              parse.block.read(parse.block(cursor, previous)), rest,              entries_left, entries, hashes),            level, total, block_count)def parse.blocks.start(  cursor: Bytes.Cursor,  valid: Bool,  +count: U32,  +total: U32,  +level: U32) -> Result<&1, &1, Error, Sstable.Table>:  match valid:    case False{}:      Fail{Malformed{}}    case True{}:      match count:        case 0:          parse.footer.magic(Bytes.Cursor.u32be(cursor), level, total,            count, Nil{}, Nil{})        case _:          parse.blocks.go(U32.to_nat(count),            parse.blocks.decision(              parse.block.read(parse.block(cursor, None{})), U32.to_nat(count),            total, Nil{}, Nil{}),            level, total, count)def stream.block.records(  +level: U32,  +total: U32,  +block_count: U32,  +remaining: U32,  +entries_left: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>,  records: ParsedEntries,  +count: U32,  digest: Bytes.Bytes) -> Result<&1, &1, Error, StreamState>:  match records:    case Entries{reversed, previous}:      Done{StreamState{level, total, block_count,        (remaining - 1 : U32), (entries_left - count : U32),        previous, List.append(&2, MemTable.Entry, reversed, entries),        digest <> hashes}}def stream.block.count(  +level: U32,  +total: U32,  +block_count: U32,  +remaining: U32,  +entries_left: U32,  entries: List<&2, MemTable.Entry>,  hashes: List<&1, Bytes.Bytes>,  records: ParsedEntries,  +count: U32,  digest: Bytes.Bytes,  valid: Bool) -> Result<&1, &1, Error, StreamState>:  match valid:    case False{}:      Fail{Malformed{}}    case True{}:      stream.block.records(level, total, block_count, remaining,        entries_left, entries, hashes, records, count, digest)def stream.block.valid(  state: StreamState,  records: ParsedEntries,  +count: U32,  digest: Bytes.Bytes,  exact: Bool) -> Result<&1, &1, Error, StreamState>:  match state:    case StreamState{+level, +total, +block_count, +remaining, +entries_left, _, entries, hashes}:      stream.block.count(level, total, block_count, remaining,        entries_left, entries, hashes, records, count, digest,        exact && U32.is_le(count, entries_left) &&          parse.blocks.count.valid(U32.to_nat(remaining), entries_left, count))def stream.block.complete(  state: StreamState,  block: ParsedBlock,  exact: Bool) -> Result<&1, &1, Error, StreamState>:  match block:    case ParsedBlock{records, +count, digest}:      stream.block.valid(state, records, count, digest, exact)def stream.block.consumed(  pair: Bytes.Cursor & U32,  state: StreamState,  block: ParsedBlock) -> Result<&1, &1, Error, StreamState>:  match pair:    case (_, remaining):      stream.block.complete(state, block, U32.is_eq(remaining, 0))def stream.block.result(  state: StreamState,  pair: Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>) -> Result<&1, &1, Error, StreamState>:  match pair:    case (_, Fail{error}):      Fail{error}    case (cursor, Done{block}):      stream.block.consumed(Bytes.Cursor.remaining(cursor), state, block)def stream.block(state: StreamState, bytes: Bytes.Bytes) -> Result<&1, &1, Error, StreamState>:  match state:    case StreamState{_, _, _, 0, _, _, _, _}:      Fail{Malformed{}}    case StreamState{+level, +total, +block_count, +remaining,        +entries_left, +previous, entries, hashes}:      stream.block.result(        StreamState{level, total, block_count, remaining, entries_left,          previous, entries, hashes},        parse.block(Bytes.Cursor.new(bytes), previous))def stream.footer(state: StreamState, bytes: Bytes.Bytes) -> Result<&1, &1, Error, Sstable.Table>:  match state:    case StreamState{+level, +total, +block_count, 0, 0, _, entries, hashes}:      parse.footer.magic(        Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)), level, total,        block_count, entries, hashes)    case _:      Fail{Malformed{}}def stream.block.tail.size(  pair: Bytes.Cursor & Maybe<&2, U32>,  +payload_len: U32) -> Maybe<&2, U32>:  match pair:    case (_, Some{+count}):      Bool.pick(Maybe<&2, U32>,        U32.is_lt(0, count) && (U32.is_le(payload_len, StorageBytes.MAX_BLOCK_BYTES()) ||          U32.is_eq(count, 1) && U32.is_le(payload_len, StorageBytes.MAX_RECORD_BYTES())) &&          U32.is_le(payload_len, (4294967263 : U32)),        Some{(payload_len + 32 : U32)}, None{})    case (_, None{}):      None{}def stream.block.tail(cursor: Bytes.Cursor & Maybe<&2, U32>) -> Maybe<&2, U32>:  match cursor:    case (next, Some{+payload_len}):      stream.block.tail.size(Bytes.Cursor.u32be(next), payload_len)    case (_, None{}):      None{}def stream.block.tail_size(bytes: Bytes.Bytes) -> Maybe<&2, U32>:  stream.block.tail(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))def parse.header.blocks.valid(  cursor: Bytes.Cursor,  +level: U32,  +total: U32,  +count: U32,  valid: Bool) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:  match valid:    case False{}:      (cursor, Fail{Malformed{}})    case True{}:      (cursor, Done{SstHdr{level, total, count}})def parse.header.blocks(  pair: Bytes.Cursor & Maybe<&2, U32>,  +level: U32,  +total: U32) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:  match pair:    case (cursor, Some{+count}):      parse.header.blocks.valid(cursor, level, total, count,        (U32.is_eq(total, 0) && U32.is_eq(count, 0)) ||          (U32.is_lt(0, total) && U32.is_lt(0, count)))    case (cursor, None{}):      (cursor, Fail{Malformed{}})def parse.header.total(  pair: Bytes.Cursor & Maybe<&2, U32>,  +level: U32) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:  match pair:    case (cursor, Some{+total}):      parse.header.blocks(Bytes.Cursor.u32be(cursor), level, total)    case (cursor, None{}):      (cursor, Fail{Malformed{}})def parse.header.level.checked(  cursor: Bytes.Cursor,  +level: U32,  valid: Bool) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:  match valid:    case False{}:      (cursor, Fail{InvalidLevel{}})    case True{}:      parse.header.total(Bytes.Cursor.u32be(cursor), level)def parse.header.level(  pair: Bytes.Cursor & Maybe<&2, U32>) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:  match pair:    case (cursor, Some{+level}):      parse.header.level.checked(cursor, level, U32.is_le(level, 255))    case (cursor, None{}):      (cursor, Fail{Malformed{}})def parse.header.magic.two(  pair: Bytes.Cursor & Maybe<&2, U32>) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:  match pair:    case (cursor, Some{1295209216}):      parse.header.level(Bytes.Cursor.u32be(cursor))    case (cursor, _):      (cursor, Fail{Malformed{}})def parse.header.magic.one(  pair: Bytes.Cursor & Maybe<&2, U32>) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:  match pair:    case (cursor, Some{1297697875}):      parse.header.magic.two(Bytes.Cursor.u32be(cursor))    case (cursor, _):      (cursor, Fail{Malformed{}})def stream.header.remaining(  pair: Bytes.Cursor & U32,  +level: U32,  +total: U32,  +block_count: U32) -> Result<&1, &1, Error, StreamState>:  match pair:    case (_, 0):      Done{StreamState{level, total, block_count, block_count,        total, None{}, Nil{}, Nil{}}}    case (_, _):      Fail{Malformed{}}def stream.header.done(  pair: Bytes.Cursor & Result<&1, &1, Error, SstHeader>) -> Result<&1, &1, Error, StreamState>:  match pair:    case (_, Fail{error}):      Fail{error}    case (cursor, Done{SstHdr{+level, +total, +block_count}}):      stream.header.remaining(Bytes.Cursor.remaining(cursor), level, total, block_count)def stream.header(bytes: Bytes.Bytes) -> Result<&1, &1, Error, StreamState>:  stream.header.done(parse.header.magic.one(    Bytes.Cursor.u32be(Bytes.Cursor.new(bytes))))# Retain bounded file chunks until the strict parser is applied.def new_decoder() -> Decoder:  Dec{Nil{}, 0, False{}}def feed.chunk.add(  chunks: List<&1, Bytes.Bytes>,  +total: U32,  invalid: Bool,  chunk: Bytes.Bytes,  checked: Maybe<&2, U32>) -> Decoder:  match checked:    case None{}:      Dec{chunks, total, True{}}    case Some{next_total}:      Dec{chunk <> chunks, next_total, invalid}def feed.chunk.result(  chunks: List<&1, Bytes.Bytes>,  +total: U32,  invalid: Bool,  pair: Bytes.Bytes & U32) -> Decoder:  match pair:    case (chunk, +len):      feed.chunk.add(chunks, total, invalid, chunk, StorageBytes.checked_add(total, len))def feed.chunk(decoder: Decoder, pair: Bytes.Bytes & U32) -> Decoder:  match decoder:    case Dec{chunks, total, invalid}:      feed.chunk.result(chunks, total, invalid, pair)def feed.if.valid(  invalid: Bool,  chunks: List<&1, Bytes.Bytes>,  total: U32,  chunk: Bytes.Bytes) -> Decoder:  match invalid:    case True{}:      Dec{chunks, total, True{}}    case False{}:      feed.chunk(Dec{chunks, total, False{}}, Bytes.length(chunk))# Buffer streaming chunks until the bounded pure parser is applied.def feed(decoder: Decoder, chunk: Bytes.Bytes) -> Decoder:  match decoder:    case Dec{chunks, total, invalid}:      feed.if.valid(invalid, chunks, total, chunk)def parse.file.header(  pair: Bytes.Cursor & Result<&1, &1, Error, SstHeader>) -> Result<&1, &1, Error, Sstable.Table>:  match pair:    case (_, Fail{error}):      Fail{error}    case (cursor, Done{SstHdr{+level, +total, +block_count}}):      parse.blocks.start(cursor, True{}, block_count, total, level)def parse.file(bytes: Bytes.Bytes) -> Result<&1, &1, Error, Sstable.Table>:  parse.file.header(parse.header.magic.one(    Bytes.Cursor.u32be(Bytes.Cursor.new(bytes))))def finish.concat.check(  bytes: Bytes.Bytes,  valid: Bool) -> Result<&1, &1, Error, Sstable.Table>:  match valid:    case True{}:      parse.file(bytes)    case False{}:      Fail{Malformed{}}def finish.concat.pair(  +total: U32,  pair: Bytes.Bytes & U32) -> Result<&1, &1, Error, Sstable.Table>:  match pair:    case (bytes, +len):      finish.concat.check(bytes, U32.is_eq(total, len))def finish.concat(  +total: U32,  bytes: Bytes.Bytes) -> Result<&1, &1, Error, Sstable.Table>:  finish.concat.pair(total, Bytes.length(bytes))# Parse and authenticate the complete accumulated SST file.def finish(decoder: Decoder) -> Result<&1, &1, Error, Sstable.Table>:  match decoder:    case Dec{chunks, total, False{}}:      finish.concat(total, Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks)))    case Dec{_, _, True{}}:      Fail{TooLarge{}}# Pure whole-file encoder for fixtures and small in-memory callers. Production# persistence uses next_chunk directly and never joins the complete file.def encode.chunks(  fuel: Nat,  pair: Encoder & Maybe<&1, Bytes.Bytes>,  acc: List<&1, Bytes.Bytes>) -> Result<&1, &1, Error, Bytes.Bytes>:  match fuel:    case 0n:      Fail{TooLarge{}}    case 1n+rest:      match pair:        case (_, None{}):          Done{Bytes.concat(List.reverse(&1, Bytes.Bytes, acc))}        case (encoder, Some{chunk}):          encode.chunks(rest, next_chunk(encoder), Con{chunk, acc})def encode.ready(  +entries: List<&2, MemTable.Entry>,  result: Result<&1, &1, Error, Encoder>) -> Result<&1, &1, Error, Bytes.Bytes>:  match result:    case Fail{error}:      Fail{error}    case Done{encoder}:      encode.chunks(Nat.add(List.length(&2, MemTable.Entry, entries), 3n), next_chunk(encoder), Nil{})# Encode one complete table for small in-memory callers and fixtures.def encode_file(  +entries: List<&2, MemTable.Entry>,  +level: U32) -> Result<&1, &1, Error, Bytes.Bytes>:  encode.ready(entries, new_encoder(entries, level))# Parse packed bytes into a validated immutable SST table.def parse(bytes: Bytes.Bytes) -> Result<&1, &1, Error, Sstable.Table>:  parse.file(bytes)# Return true because the parser always returns an explicit Result value.def is_decided(_result: Result<&1, &1, Error, Sstable.Table>) -> Bool:  True{}