1	import Base
2	import ./MemTable.bend as MemTable
3	import ./Keys.bend as Keys
4	import ./Sstable.bend as Sstable
5	import ./StorageBytes.bend as StorageBytes
6	import ./hub_sha/sha256.bend as SHA
7	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
8	
9	# Errors returned by the v3 encoder and strict parser.
10	type 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.
19	type Phase is Data:
20	  Header{}
21	  Blocks{}
22	  Finished{}
23	
24	# Serialized block paired with its stored digest.
25	type EncodedBlock is Type:
26	  Block{bytes: Bytes.Bytes, digest: Bytes.Bytes}
27	
28	# State that yields the encoded file in bounded chunks.
29	type 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.
40	type 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.
45	type 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.
49	type 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	
58	def 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	
66	def 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	
71	def bytes.clone.first(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
72	  match pair:
73	    case (left, _):
74	      left
75	
76	def 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.
80	def u32_bytes(value: U32) -> Bytes.Bytes:
81	  Bytes.set.u32be(Bytes.new(4), 0, value)
82	
83	# Construct a packed tag byte.
84	def u8_bytes(value: U32) -> Bytes.Bytes:
85	  Bytes.set(Bytes.new(1), 0, value)
86	
87	# Unwrap a fixed valid hexadecimal constant.
88	def 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.
96	def 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	
110	def 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	
124	def 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	
134	def 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	
142	def 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	
152	def 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	
158	def 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	
169	def 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	
175	def 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.
188	def 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	
193	def 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	
208	def 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.
216	type BlockMeasure is Data:
217	  InvalidRecord{}
218	  Measured{length: U32, fits: Bool}
219	
220	# Measure an already encoded record for byte-bounded block packing.
221	def 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	
225	def 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	
232	def 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	
240	def 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	
253	def 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	
264	def 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	
274	def 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.
285	def 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.
297	def 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	
331	def 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.
340	def 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.
356	def 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.
364	def 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	
375	def 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	
385	def 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	
396	def 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	
403	def 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	
415	def 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	
425	def 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.
436	def 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.
440	def 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	
457	def 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.
467	def 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.
476	def 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.
492	def 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	
500	def 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	
511	def 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.
523	def 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.
530	def 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.
542	type TextRead is Type:
543	  Text{cursor: Bytes.Cursor, byte: Maybe<&2, U32>}
544	
545	def parse.text.wrap(pair: Bytes.Cursor & Maybe<&2, U32>) -> TextRead:
546	  match pair:
547	    case (cursor, byte):
548	      Text{cursor, byte}
549	
550	def 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.
561	def 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 Text{cursor, _}:
568	      parse.text.finish(cursor, StorageBytes.utf8.scan.fin(state))
569	    case 1n+0n Text{cursor, Some{+byte}}:
570	      parse.text.finish(cursor, StorageBytes.utf8.scan.fin(
571	        StorageBytes.utf8.step(state, byte)))
572	    case 1n+1n+rest Text{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+_ Text{cursor, None{}}:
576	      (cursor, Fail{Malformed{}})
577	
578	def 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, Text{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	
591	def 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.
598	type ParsedRecord is Type:
599	  Rec{entry: MemTable.Entry, bytes_used: U32}
600	
601	# Decode one length-delimited key and optional value from the cursor.
602	def 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	
615	def 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	
629	def 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	
641	def 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	
655	def 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	
666	def 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	
676	def 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	
688	def 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	
699	def 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	
712	def 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.
718	type 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.
722	type 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.
728	type 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	
736	def 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	
746	def 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	
760	def 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	
772	def 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	
780	def 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.
796	def 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	
813	def 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	
841	def 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	
852	def 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.
867	type HashRead is Type:
868	  HashInput{cursor: Bytes.Cursor, value: Maybe<&2, U32>}
869	
870	# Validated records, count, and digest for one block.
871	type ParsedBlock is Type:
872	  ParsedBlock{records: ParsedEntries, count: U32, digest: Bytes.Bytes}
873	
874	# Represent StreamState data used by the packed SSTable codec.
875	type 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	def parse.hash.wrap(pair: Bytes.Cursor & Maybe<&2, U32>) -> HashRead:
882	  match pair:
883	    case (cursor, value):
884	      HashInput{cursor, value}
885	
886	def parse.hash.result(
887	  pair: Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>
888	) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:
889	  pair
890	
891	# Read exactly eight words into a digest without consuming a byte after it.
892	def parse.hash.words.go(
893	  remaining: Nat,
894	  read: HashRead,
895	  digest: Bytes.Bytes,
896	  +offset: U32
897	) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:
898	  match remaining read:
899	    case 0n HashInput{cursor, _}:
900	      (cursor, Done{digest})
901	    case 1n+0n HashInput{cursor, Some{word}}:
902	      (cursor, Done{Bytes.set.u32be(digest, offset, word)})
903	    case 1n+1n+rest HashInput{cursor, Some{word}}:
904	      parse.hash.words.go(1n+rest, parse.hash.wrap(Bytes.Cursor.u32be(cursor)),
905	        Bytes.set.u32be(digest, offset, word), (offset + 4 : U32))
906	    case 1n+_ HashInput{cursor, None{}}:
907	      (cursor, Fail{Malformed{}})
908	
909	def parse.hash.words(
910	  cursor: Bytes.Cursor
911	) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:
912	  parse.hash.words.go(8n, parse.hash.wrap(Bytes.Cursor.u32be(cursor)), Bytes.new(32), 0)
913	
914	def parse.block.verify.equal(
915	  cursor: Bytes.Cursor,
916	  records: ParsedEntries,
917	  +count: U32,
918	  comparison: Bytes.Bytes & Bytes.Bytes & Bool
919	) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
920	  match comparison:
921	    case (claimed, actual, False{}):
922	      (cursor, Fail{ChecksumMismatch{}})
923	    case (claimed, digest, True{}):
924	      (cursor, Done{ParsedBlock{records, count, digest}})
925	
926	def parse.block.verify.copy(
927	  pair: Bytes.Bytes & Bytes.Bytes,
928	  +pos: U32,
929	  +start: U32,
930	  +end: U32,
931	  records: ParsedEntries,
932	  +count: U32,
933	  claimed: Bytes.Bytes
934	) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
935	  match pair:
936	    case (source, content):
937	      parse.block.verify.equal(Bytes.Cursor{source, pos, start, end}, records,
938	        count, Bytes.eq(claimed, SHA.sha256_packed_bytes(content)))
939	
940	def parse.block.verify(
941	  cursor: Bytes.Cursor,
942	  records: ParsedEntries,
943	  +payload_len: U32,
944	  +count: U32,
945	  claimed: Bytes.Bytes
946	) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
947	  match cursor:
948	    case Bytes.Cursor{bytes, +pos, +start, +end}:
949	      parse.block.verify.copy(Bytes.slice(bytes,
950	        (pos - (payload_len + 40 : U32) : U32), (payload_len + 8 : U32)),
951	        pos, start, end, records, count, claimed)
952	
953	def parse.block.digest(
954	  pair: Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>,
955	  records: ParsedEntries,
956	  +payload_len: U32,
957	  +count: U32
958	) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
959	  match pair:
960	    case (cursor, Fail{error}):
961	      (cursor, Fail{error})
962	    case (cursor, Done{claimed}):
963	      parse.block.verify(cursor, records, payload_len, count, claimed)
964	
965	def parse.block.records(
966	  pair: Bytes.Cursor & Result<&1, &1, Error, ParsedEntries>,
967	  +payload_len: U32,
968	  +count: U32
969	) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
970	  match pair:
971	    case (cursor, Fail{error}):
972	      (cursor, Fail{error})
973	    case (cursor, Done{records}):
974	      parse.block.digest(parse.hash.words(cursor), records, payload_len, count)
975	
976	def parse.block.count.valid(
977	  cursor: Bytes.Cursor,
978	  +payload_len: U32,
979	  +count: U32,
980	  previous: Maybe<&2, String>,
981	  valid: Bool
982	) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
983	  match valid:
984	    case False{}:
985	      (cursor, Fail{Malformed{}})
986	    case True{}:
987	      parse.block.records(parse.records.start(cursor, count, payload_len, previous),
988	        payload_len, count)
989	
990	def parse.block.count(
991	  pair: Bytes.Cursor & Maybe<&2, U32>,
992	  +payload_len: U32,
993	  previous: Maybe<&2, String>
994	) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
995	  match pair:
996	    case (cursor, Some{+count}):
997	      parse.block.count.valid(cursor, payload_len, count, previous,
998	        U32.is_lt(0, count) && (U32.is_le(payload_len, StorageBytes.MAX_BLOCK_BYTES()) ||
999	          U32.is_eq(count, 1) && U32.is_le(payload_len, StorageBytes.MAX_RECORD_BYTES())))
1000	    case (cursor, None{}):
1001	      (cursor, Fail{Malformed{}})
1002	
1003	def parse.block.length(
1004	  pair: Bytes.Cursor & Maybe<&2, U32>,
1005	  previous: Maybe<&2, String>
1006	) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
1007	  match pair:
1008	    case (cursor, Some{+payload_len}):
1009	      parse.block.count(Bytes.Cursor.u32be(cursor), payload_len, previous)
1010	    case (cursor, None{}):
1011	      (cursor, Fail{Malformed{}})
1012	
1013	# Parse, bound, and authenticate one complete SST block.
1014	def parse.block(
1015	  cursor: Bytes.Cursor,
1016	  previous: Maybe<&2, String>
1017	) -> Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>:
1018	  parse.block.length(Bytes.Cursor.u32be(cursor), previous)
1019	
1020	def parse.footer.end(
1021	  +level: U32,
1022	  entries: List<&2, MemTable.Entry>,
1023	  at_end: Bool
1024	) -> Result<&1, &1, Error, Sstable.Table>:
1025	  match at_end:
1026	    case False{}:
1027	      Fail{Malformed{}}
1028	    case True{}:
1029	      Done{Sstable.from_sorted_unique(
1030	        List.reverse(&2, MemTable.Entry, entries), U32.to_nat(level))}
1031	
1032	def parse.footer.check(
1033	  +level: U32,
1034	  entries: List<&2, MemTable.Entry>,
1035	  at_end: Bool,
1036	  comparison: Bytes.Bytes & Bytes.Bytes & Bool
1037	) -> Result<&1, &1, Error, Sstable.Table>:
1038	  match comparison:
1039	    case (claimed_back, expected_back, False{}):
1040	      Fail{ChecksumMismatch{}}
1041	    case (claimed_back, expected_back, True{}):
1042	      parse.footer.end(level, entries, at_end)
1043	
1044	def parse.footer.verify(
1045	  cursor: Bytes.Cursor,
1046	  claimed: Bytes.Bytes,
1047	  +level: U32,
1048	  +total: U32,
1049	  +block_count: U32,
1050	  entries: List<&2, MemTable.Entry>,
1051	  hashes: List<&1, Bytes.Bytes>
1052	) -> Result<&1, &1, Error, Sstable.Table>:
1053	  match cursor:
1054	    case Bytes.Cursor{bytes, +pos, _, +end}:
1055	      expected = SHA.sha256_packed_bytes(Bytes.concat(
1056	        header_bytes(level, total, block_count) <> List.reverse(&1, Bytes.Bytes, hashes)))
1057	      parse.footer.check(level, entries, U32.is_eq(pos, end),
1058	        Bytes.eq(claimed, expected))
1059	
1060	def parse.footer.root(
1061	  pair: Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>,
1062	  +level: U32,
1063	  +total: U32,
1064	  +block_count: U32,
1065	  entries: List<&2, MemTable.Entry>,
1066	  hashes: List<&1, Bytes.Bytes>
1067	) -> Result<&1, &1, Error, Sstable.Table>:
1068	  match pair:
1069	    case (_, Fail{error}):
1070	      Fail{error}
1071	    case (cursor, Done{claimed}):
1072	      parse.footer.verify(cursor, claimed, level, total, block_count, entries, hashes)
1073	
1074	def parse.footer.count.checked(
1075	  cursor: Bytes.Cursor,
1076	  +level: U32,
1077	  +total: U32,
1078	  +block_count: U32,
1079	  entries: List<&2, MemTable.Entry>,
1080	  hashes: List<&1, Bytes.Bytes>,
1081	  valid: Bool
1082	) -> Result<&1, &1, Error, Sstable.Table>:
1083	  match valid:
1084	    case False{}:
1085	      Fail{Malformed{}}
1086	    case True{}:
1087	      parse.footer.root(parse.hash.words(cursor), level, total, block_count,
1088	        entries, hashes)
1089	
1090	def parse.footer.count(
1091	  pair: Bytes.Cursor & Maybe<&2, U32>,
1092	  +level: U32,
1093	  +total: U32,
1094	  +block_count: U32,
1095	  entries: List<&2, MemTable.Entry>,
1096	  hashes: List<&1, Bytes.Bytes>
1097	) -> Result<&1, &1, Error, Sstable.Table>:
1098	  match pair:
1099	    case (cursor, Some{+footer_total}):
1100	      parse.footer.count.checked(cursor, level, total, block_count, entries,
1101	        hashes, U32.is_eq(total, footer_total))
1102	    case (_, None{}):
1103	      Fail{Malformed{}}
1104	
1105	def parse.footer.magic(
1106	  pair: Bytes.Cursor & Maybe<&2, U32>,
1107	  +level: U32,
1108	  +total: U32,
1109	  +block_count: U32,
1110	  entries: List<&2, MemTable.Entry>,
1111	  hashes: List<&1, Bytes.Bytes>
1112	) -> Result<&1, &1, Error, Sstable.Table>:
1113	  match pair:
1114	    case (cursor, Some{1162757171}):
1115	      parse.footer.count(Bytes.Cursor.u32be(cursor), level, total,
1116	        block_count, entries, hashes)
1117	    case (_, _):
1118	      Fail{Malformed{}}
1119	
1120	# Block result paired with its ending cursor.
1121	type BlockRead is Type:
1122	  BlockInput{cursor: Bytes.Cursor, result: Result<&1, &1, Error, ParsedBlock>}
1123	
1124	def parse.block.read(pair: Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>) -> BlockRead:
1125	  match pair:
1126	    case (cursor, result):
1127	      BlockInput{cursor, result}
1128	
1129	def parse.blocks.count.valid.last(
1130	  last: Bool,
1131	  +entries_left: U32,
1132	  +count: U32
1133	) -> Bool:
1134	  match last:
1135	    case True{}:
1136	      U32.is_eq(entries_left, count)
1137	    case False{}:
1138	      U32.is_lt(count, entries_left)
1139	
1140	def parse.blocks.count.valid(
1141	  +remaining: Nat,
1142	  +entries_left: U32,
1143	  +count: U32
1144	) -> Bool:
1145	  parse.blocks.count.valid.last(Nat.is_eq(remaining, 1n), entries_left, count)
1146	
1147	def parse.blocks.accumulate.records.last(
1148	  cursor: Bytes.Cursor,
1149	  +entries_left: U32,
1150	  entries: List<&2, MemTable.Entry>,
1151	  hashes: List<&1, Bytes.Bytes>,
1152	  reversed: List<&2, MemTable.Entry>,
1153	  previous: Maybe<&2, String>,
1154	  last: Bool
1155	) -> BlockDecision:
1156	  match last:
1157	    case True{}:
1158	      BlockFinish{cursor, List.append(&2, MemTable.Entry, reversed, entries), hashes}
1159	    case False{}:
1160	      BlockContinue{cursor, entries_left, previous,
1161	        List.append(&2, MemTable.Entry, reversed, entries), hashes}
1162	
1163	def parse.blocks.accumulate.records(
1164	  cursor: Bytes.Cursor,
1165	  +remaining: Nat,
1166	  +entries_left: U32,
1167	  entries: List<&2, MemTable.Entry>,
1168	  hashes: List<&1, Bytes.Bytes>,
1169	  records: ParsedEntries
1170	) -> BlockDecision:
1171	  match records:
1172	    case Entries{reversed, previous}:
1173	      parse.blocks.accumulate.records.last(cursor, entries_left, entries,
1174	        hashes, reversed, previous, Nat.is_eq(remaining, 1n))
1175	
1176	def parse.blocks.content.valid(
1177	  cursor: Bytes.Cursor,
1178	  +remaining: Nat,
1179	  +entries_left: U32,
1180	  entries: List<&2, MemTable.Entry>,
1181	  hashes: List<&1, Bytes.Bytes>,
1182	  +count: U32,
1183	  records: ParsedEntries,
1184	  digest: Bytes.Bytes,
1185	  valid: Bool
1186	) -> BlockDecision:
1187	  match valid:
1188	    case False{}:
1189	      BlockOutOfBounds{}
1190	    case True{}:
1191	      parse.blocks.accumulate.records(cursor, remaining,
1192	        (entries_left - count : U32), entries, digest <> hashes, records)
1193	
1194	def parse.blocks.content.fit(
1195	  cursor: Bytes.Cursor,
1196	  +remaining: Nat,
1197	  +entries_left: U32,
1198	  entries: List<&2, MemTable.Entry>,
1199	  hashes: List<&1, Bytes.Bytes>,
1200	  +count: U32,
1201	  records: ParsedEntries,
1202	  digest: Bytes.Bytes,
1203	  fits: Bool
1204	) -> BlockDecision:
1205	  match fits:
1206	    case False{}:
1207	      BlockOutOfBounds{}
1208	    case True{}:
1209	      parse.blocks.content.valid(cursor, remaining, entries_left,
1210	        entries, hashes, count, records, digest,
1211	        parse.blocks.count.valid(remaining, entries_left, count))
1212	
1213	def parse.blocks.content(
1214	  cursor: Bytes.Cursor,
1215	  +remaining: Nat,
1216	  +entries_left: U32,
1217	  entries: List<&2, MemTable.Entry>,
1218	  hashes: List<&1, Bytes.Bytes>,
1219	  block: ParsedBlock
1220	) -> BlockDecision:
1221	  match block:
1222	    case ParsedBlock{records, +count, digest}:
1223	      parse.blocks.content.fit(cursor, remaining, entries_left,
1224	        entries, hashes, count, records, digest, U32.is_le(count, entries_left))
1225	
1226	def parse.blocks.decision(
1227	  read: BlockRead,
1228	  +remaining: Nat,
1229	  +entries_left: U32,
1230	  entries: List<&2, MemTable.Entry>,
1231	  hashes: List<&1, Bytes.Bytes>
1232	) -> BlockDecision:
1233	  match read:
1234	    case BlockInput{_, Fail{error}}:
1235	      BlockRejected{error}
1236	    case BlockInput{cursor, Done{block}}:
1237	      parse.blocks.content(cursor, remaining, entries_left, entries,
1238	        hashes, block)
1239	
1240	def parse.blocks.go(
1241	  remaining: Nat,
1242	  decision: BlockDecision,
1243	  +level: U32,
1244	  +total: U32,
1245	  +block_count: U32
1246	) -> Result<&1, &1, Error, Sstable.Table>:
1247	  match remaining:
1248	    case 0n:
1249	      Fail{Malformed{}}
1250	    case 1n++rest:
1251	      match decision:
1252	        case BlockRejected{error}:
1253	          Fail{error}
1254	        case BlockOutOfBounds{}:
1255	          Fail{Malformed{}}
1256	        case BlockFinish{cursor, entries, hashes}:
1257	          parse.footer.magic(Bytes.Cursor.u32be(cursor), level, total,
1258	            block_count, entries, hashes)
1259	        case BlockContinue{cursor, entries_left, previous, entries, hashes}:
1260	          parse.blocks.go(rest,
1261	            parse.blocks.decision(
1262	              parse.block.read(parse.block(cursor, previous)), rest,
1263	              entries_left, entries, hashes),
1264	            level, total, block_count)
1265	
1266	def parse.blocks.start(
1267	  cursor: Bytes.Cursor,
1268	  valid: Bool,
1269	  +count: U32,
1270	  +total: U32,
1271	  +level: U32
1272	) -> Result<&1, &1, Error, Sstable.Table>:
1273	  match valid:
1274	    case False{}:
1275	      Fail{Malformed{}}
1276	    case True{}:
1277	      match count:
1278	        case 0:
1279	          parse.footer.magic(Bytes.Cursor.u32be(cursor), level, total,
1280	            count, Nil{}, Nil{})
1281	        case _:
1282	          parse.blocks.go(U32.to_nat(count),
1283	            parse.blocks.decision(
1284	              parse.block.read(parse.block(cursor, None{})), U32.to_nat(count),
1285	            total, Nil{}, Nil{}),
1286	            level, total, count)
1287	
1288	def stream.block.records(
1289	  +level: U32,
1290	  +total: U32,
1291	  +block_count: U32,
1292	  +remaining: U32,
1293	  +entries_left: U32,
1294	  entries: List<&2, MemTable.Entry>,
1295	  hashes: List<&1, Bytes.Bytes>,
1296	  records: ParsedEntries,
1297	  +count: U32,
1298	  digest: Bytes.Bytes
1299	) -> Result<&1, &1, Error, StreamState>:
1300	  match records:
1301	    case Entries{reversed, previous}:
1302	      Done{StreamState{level, total, block_count,
1303	        (remaining - 1 : U32), (entries_left - count : U32),
1304	        previous, List.append(&2, MemTable.Entry, reversed, entries),
1305	        digest <> hashes}}
1306	
1307	def stream.block.count(
1308	  +level: U32,
1309	  +total: U32,
1310	  +block_count: U32,
1311	  +remaining: U32,
1312	  +entries_left: U32,
1313	  entries: List<&2, MemTable.Entry>,
1314	  hashes: List<&1, Bytes.Bytes>,
1315	  records: ParsedEntries,
1316	  +count: U32,
1317	  digest: Bytes.Bytes,
1318	  valid: Bool
1319	) -> Result<&1, &1, Error, StreamState>:
1320	  match valid:
1321	    case False{}:
1322	      Fail{Malformed{}}
1323	    case True{}:
1324	      stream.block.records(level, total, block_count, remaining,
1325	        entries_left, entries, hashes, records, count, digest)
1326	
1327	def stream.block.valid(
1328	  state: StreamState,
1329	  records: ParsedEntries,
1330	  +count: U32,
1331	  digest: Bytes.Bytes,
1332	  exact: Bool
1333	) -> Result<&1, &1, Error, StreamState>:
1334	  match state:
1335	    case StreamState{+level, +total, +block_count, +remaining, +entries_left, _, entries, hashes}:
1336	      stream.block.count(level, total, block_count, remaining,
1337	        entries_left, entries, hashes, records, count, digest,
1338	        exact && U32.is_le(count, entries_left) &&
1339	          parse.blocks.count.valid(U32.to_nat(remaining), entries_left, count))
1340	
1341	def stream.block.complete(
1342	  state: StreamState,
1343	  block: ParsedBlock,
1344	  exact: Bool
1345	) -> Result<&1, &1, Error, StreamState>:
1346	  match block:
1347	    case ParsedBlock{records, +count, digest}:
1348	      stream.block.valid(state, records, count, digest, exact)
1349	
1350	def stream.block.consumed(
1351	  pair: Bytes.Cursor & U32,
1352	  state: StreamState,
1353	  block: ParsedBlock
1354	) -> Result<&1, &1, Error, StreamState>:
1355	  match pair:
1356	    case (_, remaining):
1357	      stream.block.complete(state, block, U32.is_eq(remaining, 0))
1358	
1359	def stream.block.result(
1360	  state: StreamState,
1361	  pair: Bytes.Cursor & Result<&1, &1, Error, ParsedBlock>
1362	) -> Result<&1, &1, Error, StreamState>:
1363	  match pair:
1364	    case (_, Fail{error}):
1365	      Fail{error}
1366	    case (cursor, Done{block}):
1367	      stream.block.consumed(Bytes.Cursor.remaining(cursor), state, block)
1368	
1369	def stream.block(state: StreamState, bytes: Bytes.Bytes) -> Result<&1, &1, Error, StreamState>:
1370	  match state:
1371	    case StreamState{_, _, _, 0, _, _, _, _}:
1372	      Fail{Malformed{}}
1373	    case StreamState{+level, +total, +block_count, +remaining,
1374	        +entries_left, +previous, entries, hashes}:
1375	      stream.block.result(
1376	        StreamState{level, total, block_count, remaining, entries_left,
1377	          previous, entries, hashes},
1378	        parse.block(Bytes.Cursor.new(bytes), previous))
1379	
1380	def stream.footer(state: StreamState, bytes: Bytes.Bytes) -> Result<&1, &1, Error, Sstable.Table>:
1381	  match state:
1382	    case StreamState{+level, +total, +block_count, 0, 0, _, entries, hashes}:
1383	      parse.footer.magic(
1384	        Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)), level, total,
1385	        block_count, entries, hashes)
1386	    case _:
1387	      Fail{Malformed{}}
1388	
1389	def stream.block.tail.size(
1390	  pair: Bytes.Cursor & Maybe<&2, U32>,
1391	  +payload_len: U32
1392	) -> Maybe<&2, U32>:
1393	  match pair:
1394	    case (_, Some{+count}):
1395	      Bool.pick(Maybe<&2, U32>,
1396	        U32.is_lt(0, count) && (U32.is_le(payload_len, StorageBytes.MAX_BLOCK_BYTES()) ||
1397	          U32.is_eq(count, 1) && U32.is_le(payload_len, StorageBytes.MAX_RECORD_BYTES())) &&
1398	          U32.is_le(payload_len, (4294967263 : U32)),
1399	        Some{(payload_len + 32 : U32)}, None{})
1400	    case (_, None{}):
1401	      None{}
1402	
1403	def stream.block.tail(cursor: Bytes.Cursor & Maybe<&2, U32>) -> Maybe<&2, U32>:
1404	  match cursor:
1405	    case (next, Some{+payload_len}):
1406	      stream.block.tail.size(Bytes.Cursor.u32be(next), payload_len)
1407	    case (_, None{}):
1408	      None{}
1409	
1410	def stream.block.tail_size(bytes: Bytes.Bytes) -> Maybe<&2, U32>:
1411	  stream.block.tail(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))
1412	
1413	def parse.header.blocks.valid(
1414	  cursor: Bytes.Cursor,
1415	  +level: U32,
1416	  +total: U32,
1417	  +count: U32,
1418	  valid: Bool
1419	) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1420	  match valid:
1421	    case False{}:
1422	      (cursor, Fail{Malformed{}})
1423	    case True{}:
1424	      (cursor, Done{SstHdr{level, total, count}})
1425	
1426	def parse.header.blocks(
1427	  pair: Bytes.Cursor & Maybe<&2, U32>,
1428	  +level: U32,
1429	  +total: U32
1430	) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1431	  match pair:
1432	    case (cursor, Some{+count}):
1433	      parse.header.blocks.valid(cursor, level, total, count,
1434	        (U32.is_eq(total, 0) && U32.is_eq(count, 0)) ||
1435	          (U32.is_lt(0, total) && U32.is_lt(0, count)))
1436	    case (cursor, None{}):
1437	      (cursor, Fail{Malformed{}})
1438	
1439	def parse.header.total(
1440	  pair: Bytes.Cursor & Maybe<&2, U32>,
1441	  +level: U32
1442	) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1443	  match pair:
1444	    case (cursor, Some{+total}):
1445	      parse.header.blocks(Bytes.Cursor.u32be(cursor), level, total)
1446	    case (cursor, None{}):
1447	      (cursor, Fail{Malformed{}})
1448	
1449	def parse.header.level.checked(
1450	  cursor: Bytes.Cursor,
1451	  +level: U32,
1452	  valid: Bool
1453	) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1454	  match valid:
1455	    case False{}:
1456	      (cursor, Fail{InvalidLevel{}})
1457	    case True{}:
1458	      parse.header.total(Bytes.Cursor.u32be(cursor), level)
1459	
1460	def parse.header.level(
1461	  pair: Bytes.Cursor & Maybe<&2, U32>
1462	) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1463	  match pair:
1464	    case (cursor, Some{+level}):
1465	      parse.header.level.checked(cursor, level, U32.is_le(level, 255))
1466	    case (cursor, None{}):
1467	      (cursor, Fail{Malformed{}})
1468	
1469	def parse.header.magic.two(
1470	  pair: Bytes.Cursor & Maybe<&2, U32>
1471	) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1472	  match pair:
1473	    case (cursor, Some{1295209216}):
1474	      parse.header.level(Bytes.Cursor.u32be(cursor))
1475	    case (cursor, _):
1476	      (cursor, Fail{Malformed{}})
1477	
1478	def parse.header.magic.one(
1479	  pair: Bytes.Cursor & Maybe<&2, U32>
1480	) -> Bytes.Cursor & Result<&1, &1, Error, SstHeader>:
1481	  match pair:
1482	    case (cursor, Some{1297697875}):
1483	      parse.header.magic.two(Bytes.Cursor.u32be(cursor))
1484	    case (cursor, _):
1485	      (cursor, Fail{Malformed{}})
1486	
1487	def stream.header.remaining(
1488	  pair: Bytes.Cursor & U32,
1489	  +level: U32,
1490	  +total: U32,
1491	  +block_count: U32
1492	) -> Result<&1, &1, Error, StreamState>:
1493	  match pair:
1494	    case (_, 0):
1495	      Done{StreamState{level, total, block_count, block_count,
1496	        total, None{}, Nil{}, Nil{}}}
1497	    case (_, _):
1498	      Fail{Malformed{}}
1499	
1500	def stream.header.done(
1501	  pair: Bytes.Cursor & Result<&1, &1, Error, SstHeader>
1502	) -> Result<&1, &1, Error, StreamState>:
1503	  match pair:
1504	    case (_, Fail{error}):
1505	      Fail{error}
1506	    case (cursor, Done{SstHdr{+level, +total, +block_count}}):
1507	      stream.header.remaining(Bytes.Cursor.remaining(cursor), level, total, block_count)
1508	
1509	def stream.header(bytes: Bytes.Bytes) -> Result<&1, &1, Error, StreamState>:
1510	  stream.header.done(parse.header.magic.one(
1511	    Bytes.Cursor.u32be(Bytes.Cursor.new(bytes))))
1512	
1513	# Retain bounded file chunks until the strict parser is applied.
1514	def new_decoder() -> Decoder:
1515	  Dec{Nil{}, 0, False{}}
1516	
1517	def feed.chunk.add(
1518	  chunks: List<&1, Bytes.Bytes>,
1519	  +total: U32,
1520	  invalid: Bool,
1521	  chunk: Bytes.Bytes,
1522	  checked: Maybe<&2, U32>
1523	) -> Decoder:
1524	  match checked:
1525	    case None{}:
1526	      Dec{chunks, total, True{}}
1527	    case Some{next_total}:
1528	      Dec{chunk <> chunks, next_total, invalid}
1529	
1530	def feed.chunk.result(
1531	  chunks: List<&1, Bytes.Bytes>,
1532	  +total: U32,
1533	  invalid: Bool,
1534	  pair: Bytes.Bytes & U32
1535	) -> Decoder:
1536	  match pair:
1537	    case (chunk, +len):
1538	      feed.chunk.add(chunks, total, invalid, chunk, StorageBytes.checked_add(total, len))
1539	
1540	def feed.chunk(decoder: Decoder, pair: Bytes.Bytes & U32) -> Decoder:
1541	  match decoder:
1542	    case Dec{chunks, total, invalid}:
1543	      feed.chunk.result(chunks, total, invalid, pair)
1544	
1545	def feed.if.valid(
1546	  invalid: Bool,
1547	  chunks: List<&1, Bytes.Bytes>,
1548	  total: U32,
1549	  chunk: Bytes.Bytes
1550	) -> Decoder:
1551	  match invalid:
1552	    case True{}:
1553	      Dec{chunks, total, True{}}
1554	    case False{}:
1555	      feed.chunk(Dec{chunks, total, False{}}, Bytes.length(chunk))
1556	
1557	# Buffer streaming chunks until the bounded pure parser is applied.
1558	def feed(decoder: Decoder, chunk: Bytes.Bytes) -> Decoder:
1559	  match decoder:
1560	    case Dec{chunks, total, invalid}:
1561	      feed.if.valid(invalid, chunks, total, chunk)
1562	
1563	def parse.file.header(
1564	  pair: Bytes.Cursor & Result<&1, &1, Error, SstHeader>
1565	) -> Result<&1, &1, Error, Sstable.Table>:
1566	  match pair:
1567	    case (_, Fail{error}):
1568	      Fail{error}
1569	    case (cursor, Done{SstHdr{+level, +total, +block_count}}):
1570	      parse.blocks.start(cursor, True{}, block_count, total, level)
1571	
1572	def parse.file(bytes: Bytes.Bytes) -> Result<&1, &1, Error, Sstable.Table>:
1573	  parse.file.header(parse.header.magic.one(
1574	    Bytes.Cursor.u32be(Bytes.Cursor.new(bytes))))
1575	
1576	def finish.concat.check(
1577	  bytes: Bytes.Bytes,
1578	  valid: Bool
1579	) -> Result<&1, &1, Error, Sstable.Table>:
1580	  match valid:
1581	    case True{}:
1582	      parse.file(bytes)
1583	    case False{}:
1584	      Fail{Malformed{}}
1585	
1586	def finish.concat.pair(
1587	  +total: U32,
1588	  pair: Bytes.Bytes & U32
1589	) -> Result<&1, &1, Error, Sstable.Table>:
1590	  match pair:
1591	    case (bytes, +len):
1592	      finish.concat.check(bytes, U32.is_eq(total, len))
1593	
1594	def finish.concat(
1595	  +total: U32,
1596	  bytes: Bytes.Bytes
1597	) -> Result<&1, &1, Error, Sstable.Table>:
1598	  finish.concat.pair(total, Bytes.length(bytes))
1599	
1600	# Parse and authenticate the complete accumulated SST file.
1601	def finish(decoder: Decoder) -> Result<&1, &1, Error, Sstable.Table>:
1602	  match decoder:
1603	    case Dec{chunks, total, False{}}:
1604	      finish.concat(total, Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks)))
1605	    case Dec{_, _, True{}}:
1606	      Fail{TooLarge{}}
1607	
1608	# Pure whole-file encoder for fixtures and small in-memory callers. Production
1609	# persistence uses next_chunk directly and never joins the complete file.
1610	def encode.chunks(
1611	  fuel: Nat,
1612	  pair: Encoder & Maybe<&1, Bytes.Bytes>,
1613	  acc: List<&1, Bytes.Bytes>
1614	) -> Result<&1, &1, Error, Bytes.Bytes>:
1615	  match fuel:
1616	    case 0n:
1617	      Fail{TooLarge{}}
1618	    case 1n+rest:
1619	      match pair:
1620	        case (_, None{}):
1621	          Done{Bytes.concat(List.reverse(&1, Bytes.Bytes, acc))}
1622	        case (encoder, Some{chunk}):
1623	          encode.chunks(rest, next_chunk(encoder), Con{chunk, acc})
1624	
1625	def encode.ready(
1626	  +entries: List<&2, MemTable.Entry>,
1627	  result: Result<&1, &1, Error, Encoder>
1628	) -> Result<&1, &1, Error, Bytes.Bytes>:
1629	  match result:
1630	    case Fail{error}:
1631	      Fail{error}
1632	    case Done{encoder}:
1633	      encode.chunks(Nat.add(List.length(&2, MemTable.Entry, entries), 3n), next_chunk(encoder), Nil{})
1634	
1635	# Encode one complete table for small in-memory callers and fixtures.
1636	def encode_file(
1637	  +entries: List<&2, MemTable.Entry>,
1638	  +level: U32
1639	) -> Result<&1, &1, Error, Bytes.Bytes>:
1640	  encode.ready(entries, new_encoder(entries, level))
1641	
1642	# Parse packed bytes into a validated immutable SST table.
1643	def parse(bytes: Bytes.Bytes) -> Result<&1, &1, Error, Sstable.Table>:
1644	  parse.file(bytes)
1645	
1646	# Return true because the parser always returns an explicit Result value.
1647	def is_decided(_result: Result<&1, &1, Error, Sstable.Table>) -> Bool:
1648	  True{}