1	import Base
2	import ../ByteSstFile.bend as ByteSstFile
3	import ../ByteWal.bend as ByteWal
4	import ../Wal.bend as Wal
5	import ../ByteMemTable.bend as ByteMemTable
6	import ../StorageBytes.bend as StorageBytes
7	import ../ByteRecoveryPolicy.bend as RecoveryPolicy
8	import ../Fs.bend as Fs
9	import ../SstStreamIo.bend as SstStreamIo
10	import ../coordinator/QueuePolicy.bend as QueuePolicy
11	import ./BlockOrder.bend as BlockOrder
12	import ./BlockIndexPolicy.bend as BlockIndexPolicy
13	import ./BlockIndexFormatPolicy.bend as BlockIndexFormatPolicy
14	import ./BlockRoutePolicy.bend as BlockRoutePolicy
15	import ../hub_sha/sha256.bend as SHA
16	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
17	
18	# Retains bounded traversal metadata without materializing the table entries.
19	type State is Type:
20	  State{size: Nat, header: Bytes.Bytes, index_offset: Nat, used: Nat,
21	    total: U32, block_count: U32, next_block: U32, decoded: U32,
22	    root_hash: SHA.Stream, index_hash: SHA.Stream,
23	    last_key: Maybe<&1, Bytes.Bytes>, selected: Bool}
24	
25	# Owns the open file and its current stream position.
26	type Reader is Type:
27	  Reader{file: File, state: State}
28	
29	# Retains only the authenticated file's fixed-size table and index metadata.
30	type Metadata is Data:
31	  Metadata{size: Nat, total: U32, block_count: U32, index_offset: Nat,
32	    live_entries: Nat, payload_bytes: Nat}
33	
34	# Represents one decoded block, end of input, or a read failure.
35	type Step is Type:
36	  Block{reader: Reader, entries: List<&1, ByteMemTable.Entry>}
37	  End{}
38	
39	# Owns the active reader and the unread entries from its current block.
40	type EntryCursor is Type:
41	  EntryReady{reader: Reader, entries: List<&1, ByteMemTable.Entry>}
42	  EntryEnded{}
43	  EntryClosed{}
44	  EntryFailed{error: U32 & String}
45	
46	# Represents one entry, an empty block, end of file, or a terminal error.
47	type EntryStep is Type:
48	  EntryItem{entry: ByteMemTable.Entry, cursor: EntryCursor}
49	  EntrySkip{cursor: EntryCursor}
50	  EntryEnd{}
51	  EntryError{cursor: EntryCursor}
52	
53	
54	# Carries the block header while the bounded body read is in flight.
55	type BodyContext is Type:
56	  BodyContext{state: State, block_header: Bytes.Bytes, payload_length: U32}
57	
58	# Retains one bounded serialized-index page and its incremental validators.
59	type IndexPageState is Type:
60	  IndexPageNeed{file: File, reader: IndexMeta, offset: Nat, remaining: Nat,
61	    actual_hash: SHA.Stream, root_hash: SHA.Stream}
62	  IndexPageRead{pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>,
63	    reader: IndexMeta, offset: Nat, remaining: Nat,
64	    actual_hash: SHA.Stream, root_hash: SHA.Stream}
65	  IndexPageParsed{file: File, result: Result<&1, &1, U32 & String, IndexPageData>}
66	
67	# Carries page validation outputs without retaining serialized index bytes.
68	type IndexPageData is Type:
69	  IndexPageData{reader: IndexMeta, offset: Nat, remaining: Nat,
70	    actual_hash: SHA.Stream, root_hash: SHA.Stream}
71	
72	# Holds only footer fields required after index traversal completes.
73	type IndexMeta is Type:
74	  IndexMeta{total: U32, index_length: U32, expected_hash: SHA.Stream}
75	
76	# Carries the owned reader through positional separator search states.
77	type RouteContext is Data:
78	  RouteContext{size: Nat, index_offset: Nat, block_count: Nat}
79	
80	# Carries each positional read through the separator-search state machine.
81	type RouteState is Type:
82	  RouteNeed{file: File, state: State, context: RouteContext, key: Bytes.Bytes,
83	    search: BlockRoutePolicy.Search}
84	  RouteSelect{reader: Reader, block: Nat}
85	  RouteRecordRead{pair: File & Result<&1, &1, U32 & String,
86	    SstStreamIo.ExactResult>, state: State, context: RouteContext, key: Bytes.Bytes,
87	    search: BlockRoutePolicy.Search, mid: Nat}
88	  RouteSeparatorNeed{file: File, state: State, context: RouteContext, key: Bytes.Bytes,
89	    search: BlockRoutePolicy.Search, mid: Nat, offset: Nat, length: U32,
90	    separator_start: Nat}
91	  RouteSeparatorRead{pair: File & Result<&1, &1, U32 & String,
92	    SstStreamIo.ExactResult>, state: State, context: RouteContext, key: Bytes.Bytes,
93	    search: BlockRoutePolicy.Search, mid: Nat}
94	  RouteCompare{file: File, state: State, context: RouteContext, key: Bytes.Bytes,
95	    search: BlockRoutePolicy.Search, mid: Nat, separator: Bytes.Bytes}
96	  RouteFailed{file: File, error: U32 & String}
97	
98	# Computes enough transitions to visit blocks and stream their index pages.
99	def transition.budget.state(file: File, state: State) -> Nat & Reader:
100	  match state:
101	    case State{+size, header, +index_offset, used, total, +block_count,
102	        next_block, decoded, root_hash, index_hash, last_key, selected}:
103	      +blocks = U32.to_nat(block_count)
104	      +index_bytes = Nat.sub(size, Nat.add(index_offset, 44n))
105	      +page_bytes = BlockIndexPolicy.max_bytes()
106	      +pages = Nat.div(Nat.add(index_bytes, Nat.sub(page_bytes, 1n)), page_bytes)
107	      (Nat.add(Nat.add(Nat.mul(4n, blocks), Nat.mul(3n, pages)), 4n),
108	        Reader{file, State{size, header, index_offset, used, total, block_count,
109	          next_block, decoded, root_hash, index_hash, last_key, selected}})
110	
111	# Returns the block traversal budget while preserving the owned reader.
112	def transition_budget(reader: Reader) -> Nat & Reader:
113	  match reader:
114	    case Reader{file, state}: transition.budget.state(file, state)
115	
116	# Returns compact file metadata while preserving the owned reader.
117	def metadata(reader: Reader) -> Metadata & Reader:
118	  match reader:
119	    case Reader{file, state}:
120	      match state:
121	        case State{+size, header, +index_offset, used, +total,
122	            +block_count, next_block, decoded, root_hash, index_hash,
123	            last_key, selected}:
124	          (Metadata{size, total, block_count, index_offset, 0n, 0n},
125	            Reader{file, State{size, header, index_offset, used, total,
126	              block_count, next_block, decoded, root_hash, index_hash,
127	              last_key, selected}})
128	
129	def bytes.slice.result(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
130	  match pair:
131	    case (_, bytes): bytes
132	
133	def block.checksum.compare(
134	  comparison: Bytes.Bytes & Bytes.Bytes & Bool
135	) -> Bool:
136	  match comparison:
137	    case (_, _, valid): valid
138	
139	def block.checksum.valid.copies(
140	  copies: Bytes.Bytes & Bytes.Bytes,
141	  +payload_length: U32
142	) -> Bool:
143	  match copies:
144	    case (for_hash, for_checksum):
145	      block.checksum.compare(Bytes.eq(
146	        SHA.sha256_packed_bytes(bytes.slice.result(Bytes.slice(for_hash, 0,
147	          U32.add(8, payload_length)))), bytes.slice.result(Bytes.slice(
148	          for_checksum, U32.add(8, payload_length), 32))))
149	
150	def block.checksum.valid(
151	  block: Bytes.Bytes,
152	  +payload_length: U32
153	) -> Bool:
154	  block.checksum.valid.copies(ByteSstFile.clone.split(block), payload_length)
155	
156	def read_at.exact.complete(
157	  bytes: Bytes.Bytes,
158	  lengths: Nat & Nat,
159	  needed: U32
160	) -> Result<&1, &1, U32 & String, SstStreamIo.ExactResult>:
161	  match lengths:
162	    case (actual, _):
163	      Done{SstStreamIo.ExactResult{bytes,
164	        U32.is_eq(U32.from_nat(actual), needed)}}
165	
166	def read_at.exact.length(
167	  needed: U32,
168	  copies: Bytes.Bytes & Bytes.Bytes
169	) -> Result<&1, &1, U32 & String, SstStreamIo.ExactResult>:
170	  match copies:
171	    case (measured, retained):
172	      match measured:
173	        case Bytes.Bytes{len, _}:
174	          read_at.exact.complete(retained,
175	            QueuePolicy.nat.clone.pair(U32.to_nat(len)), needed)
176	
177	def read_at.exact.bytes(
178	  needed: U32,
179	  result: Result<&1, &1, U32 & String, Bytes.Bytes>
180	) -> Result<&1, &1, U32 & String, SstStreamIo.ExactResult>:
181	  match result:
182	    case Fail{error}: Fail{error}
183	    case Done{bytes}:
184	      read_at.exact.length(needed, ByteSstFile.clone.split(bytes))
185	
186	def read_at.exact.pair(
187	  needed: U32,
188	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
189	) -> File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>:
190	  match pair:
191	    case (file, result): (file, read_at.exact.bytes(needed, result))
192	
193	# Maps malformed SST content to the storage corruption domain.
194	def corruption(message: String) -> U32 & String:
195	  (4294967292, message)
196	
197	# Identifies the explicit resident index budget failure to the public API.
198	def resource_limit_error_code() -> U32:
199	  RecoveryPolicy.resource_limit_error_code()
200	
201	# Returns a typed host error for an index larger than the resident budget.
202	def resource_limit(message: String) -> U32 & String:
203	  (resource_limit_error_code(), message)
204	
205	def codec.error(error: ByteSstFile.Error) -> U32 & String:
206	  match error:
207	    case ByteSstFile.ChecksumMismatch{}:
208	      corruption("byte SST checksum mismatch")
209	    case ByteSstFile.InvalidOrder{}:
210	      corruption("byte SST keys are not strictly ordered")
211	    case _:
212	      corruption("malformed byte SST")
213	
214	def close.failed(
215	  file: File,
216	  error: U32 & String
217	) -> IO(Result<&1, &1, U32 & String, Step>):
218	  do IO<Result<&1, &1, U32 & String, Step>>:
219	    _closed : Unit <- File.close(file)
220	        return Fail{error}
221	
222	def open.failed(
223	  file: File,
224	  error: U32 & String
225	) -> IO(Result<&1, &1, U32 & String, Reader>):
226	  do IO<Result<&1, &1, U32 & String, Reader>>:
227	    _closed : Unit <- File.close(file)
228	    return Fail{error}
229	
230	def open.hash.index(
231	  file: File,
232	  +size: Nat,
233	  header: Bytes.Bytes,
234	  total: U32,
235	  block_count: U32,
236	  index_offset: Nat,
237	  root_hash: SHA.Stream,
238	  result: Result<&1, &1, SHA.StreamError, SHA.Stream>
239	) -> IO(Result<&1, &1, U32 & String, Reader>):
240	  match result:
241	    case Fail{_}:
242	      open.failed(file, resource_limit("SST metadata exceeds the SHA-256 input limit"))
243	    case Done{index_hash}:
244	      IO.pure(Result<&1, &1, U32 & String, Reader>, Done{Reader{file,
245	        State{size, header, index_offset, 0n, total, block_count, 0, 0,
246	          root_hash, index_hash, None{}, False{}}}})
247	
248	def open.hash.root(
249	  file: File,
250	  +size: Nat,
251	  header: Bytes.Bytes,
252	  total: U32,
253	  counts: Nat & Nat,
254	  index_offset: Nat,
255	  result: Result<&1, &1, SHA.StreamError, SHA.Stream>
256	) -> IO(Result<&1, &1, U32 & String, Reader>):
257	  match counts:
258	    case (for_reader, for_prefix):
259	      match result:
260	        case Fail{_}:
261	          open.failed(file, resource_limit("SST metadata exceeds the SHA-256 input limit"))
262	        case Done{root_hash}:
263	          open.hash.index(file, size, header, total, U32.from_nat(for_reader),
264	            index_offset, root_hash,
265	            SHA.stream.update(SHA.stream.initial(), Bytes.concat([
266	              ByteSstFile.index.magic(),
267	              ByteSstFile.u32.bytes(U32.from_nat(for_prefix))])))
268	
269	def open.hash.header(
270	  file: File,
271	  +size: Nat,
272	  copies: Bytes.Bytes & Bytes.Bytes,
273	  total: U32,
274	  block_count: U32,
275	  index_offset: Nat
276	) -> IO(Result<&1, &1, U32 & String, Reader>):
277	  match copies:
278	    case (header, for_hash):
279	      open.hash.root(file, size, header, total,
280	        QueuePolicy.nat.clone.pair(U32.to_nat(block_count)), index_offset,
281	        SHA.stream.update(SHA.stream.initial(), for_hash))
282	
283	def open.hashes(
284	  file: File,
285	  +size: Nat,
286	  header: Bytes.Bytes,
287	  total: U32,
288	  block_count: U32,
289	  index_offset: Nat
290	) -> IO(Result<&1, &1, U32 & String, Reader>):
291	  open.hash.header(file, size, ByteSstFile.clone.split(header), total,
292	    block_count, index_offset)
293	
294	def open.header.index(
295	  file: File,
296	  +size: Nat,
297	  header: Bytes.Bytes,
298	  total: U32,
299	  block_count: U32,
300	  index_offset: Nat,
301	  valid: Bool
302	) -> IO(Result<&1, &1, U32 & String, Reader>):
303	  match valid:
304	    case False{}:
305	      open.failed(file, corruption("byte SST index exceeds file bounds"))
306	    case True{}:
307	      open.hashes(file, size, header, total, block_count, index_offset)
308	
309	def open.header.index.length.decoded(
310	  file: File,
311	  +size: Nat,
312	  header: Bytes.Bytes,
313	  total: U32,
314	  block_count: U32,
315	  +block_count_nat: U32,
316	  result: Bytes.Bytes & Maybe<&2, U32>
317	) -> IO(Result<&1, &1, U32 & String, Reader>):
318	  match result:
319	    case (_, Some{index_length_word}):
320	      +index_length = U32.to_nat(index_length_word)
321	      +blocks = U32.to_nat(block_count_nat)
322	      +minimum_index_length = Nat.add(8n, Nat.mul(blocks, 16n))
323	      +index_size = Nat.add(44n, index_length)
324	      +index_offset = Nat.sub(size, index_size)
325	      open.header.index(file, size, header, total, block_count,
326	        index_offset,
327	        Nat.is_le(minimum_index_length, index_length) &&
328	          Nat.is_le(index_size, size) && Nat.is_le(20n, index_offset))
329	    case (_, None{}):
330	      open.failed(file, corruption("byte SST footer has no index length"))
331	
332	def open.header.index.length.bytes(
333	  file: File,
334	  +size: Nat,
335	  header: Bytes.Bytes,
336	  total: U32,
337	  block_count: U32,
338	  block_count_nat: U32,
339	  result: Result<&1, &1, U32 & String, Bytes.Bytes>
340	) -> IO(Result<&1, &1, U32 & String, Reader>):
341	  match result:
342	    case Fail{error}: open.failed(file, error)
343	    case Done{bytes}:
344	      open.header.index.length.decoded(file, size, header, total,
345	        block_count, block_count_nat, Bytes.get.u32be(bytes, 0))
346	
347	def open.header.index.length.exact(
348	  pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>,
349	  +size: Nat,
350	  header: Bytes.Bytes,
351	  total: U32,
352	  block_count: U32,
353	  block_count_nat: U32
354	) -> IO(Result<&1, &1, U32 & String, Reader>):
355	  match pair:
356	    case (file, Fail{error}): open.failed(file, error)
357	    case (file, Done{SstStreamIo.ExactResult{bytes, complete}}):
358	      match complete:
359	        case True{}:
360	          open.header.index.length.bytes(file, size, header, total,
361	            block_count, block_count_nat, Done{bytes})
362	        case False{}:
363	          open.failed(file, corruption("truncated byte SST footer index length"))
364	
365	def open.header.index.counts(
366	  file: File,
367	  +size: Nat,
368	  header: Bytes.Bytes,
369	  total: U32,
370	  counts: Nat & Nat
371	) -> IO(Result<&1, &1, U32 & String, Reader>):
372	  match counts:
373	    case (for_offset, for_reader):
374	      do IO<Result<&1, &1, U32 & String, Reader>>:
375	        footer_length : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
376	          Fs.read_at(file, U32.from_nat(Nat.sub(size, 36n)), 4)
377	        open.header.index.length.exact(
378	          read_at.exact.pair(4, footer_length), size, header, total,
379	          U32.from_nat(for_reader), U32.from_nat(for_offset))
380	
381	def open.header.valid(
382	  file: File,
383	  +size: Nat,
384	  header: Bytes.Bytes,
385	  total: U32,
386	  block_count: U32,
387	  size_valid: Bool
388	) -> IO(Result<&1, &1, U32 & String, Reader>):
389	  match size_valid:
390	    case False{}:
391	      open.failed(file, corruption("byte SST exceeds reader metadata bounds"))
392	    case True{}:
393	      open.header.index.counts(file, size, header, total,
394	        QueuePolicy.nat.clone.pair(U32.to_nat(block_count)))
395	
396	def open.header.parsed(
397	  file: File,
398	  +size: Nat,
399	  result: Result<&1, &1, ByteSstFile.Error, ByteSstFile.MultiHeader>
400	) -> IO(Result<&1, &1, U32 & String, Reader>):
401	  match result:
402	    case Fail{error}:
403	      open.failed(file, codec.error(error))
404	    case Done{ByteSstFile.MultiHeader{header, _, total, +block_count}}:
405	      open.header.valid(file, size, header, total, block_count,
406	        Nat.is_le(20n, size) && Nat.is_le(size, 4294967295n))
407	
408	def open.header.result(
409	  file: File,
410	  +size: Nat,
411	  result: Result<&1, &1, U32 & String, Bytes.Bytes>
412	) -> IO(Result<&1, &1, U32 & String, Reader>):
413	  match result:
414	    case Fail{error}:
415	      open.failed(file, error)
416	    case Done{header}:
417	      open.header.parsed(file, size, ByteSstFile.parse_stream_header(header))
418	
419	def open.header.read(
420	  +size: Nat,
421	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
422	) -> IO(Result<&1, &1, U32 & String, Reader>):
423	  match pair:
424	    case (file, result):
425	      open.header.result(file, size, result)
426	
427	def open.header.pair(
428	  +size: Nat,
429	  pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>
430	) -> IO(Result<&1, &1, U32 & String, Reader>):
431	  open.header.read(size, SstStreamIo.read.exact.state(pair))
432	
433	def open.file(
434	  +size: Nat,
435	  opened: Result<&1, &1, U32 & String, File>
436	) -> IO(Result<&1, &1, U32 & String, Reader>):
437	  match opened:
438	    case Fail{error}:
439	      IO.pure(Result<&1, &1, U32 & String, Reader>, Fail{error})
440	    case Done{file}:
441	      do IO<Result<&1, &1, U32 & String, Reader>>:
442	        pair : File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult> <-
443	          SstStreamIo.exact.start(file, 20n)
444	        open.header.pair(size, pair)
445	
446	def open.sized.valid(
447	  +path: String,
448	  +size: Nat,
449	  valid: Bool
450	) -> IO(Result<&1, &1, U32 & String, Reader>):
451	  match valid:
452	    case False{}:
453	      IO.pure(Result<&1, &1, U32 & String, Reader>,
454	        Fail{corruption("byte SST is truncated")})
455	    case True{}:
456	      do IO<Result<&1, &1, U32 & String, Reader>>:
457	        opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
458	        open.file(size, opened)
459	
460	
461	def open.sized(
462	  +path: String,
463	  result: Result<&1, &1, U32 & String, Nat>
464	) -> IO(Result<&1, &1, U32 & String, Reader>):
465	  match result:
466	    case Fail{error}:
467	      IO.pure(Result<&1, &1, U32 & String, Reader>, Fail{error})
468	    case Done{+size}:
469	      open.sized.valid(path, size, Nat.is_le(60n, size))
470	# Opens a generation-four SST and validates its bounded header metadata.
471	def open(+path: String) -> IO(Result<&1, &1, U32 & String, Reader>):
472	  do IO<Result<&1, &1, U32 & String, Reader>>:
473	    size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path)
474	    open.sized(path, size)
475	
476	def seek.block.invalid(file: File) -> IO(Result<&1, &1, U32 & String, Reader>):
477	  do IO<Result<&1, &1, U32 & String, Reader>>:
478	    _closed : Unit <- File.close(file)
479	    return Fail{corruption("byte SST block index is outside the file")}
480	
481	def seek.block.construct(
482	  file: File,
483	  size: Nat,
484	  header: Bytes.Bytes,
485	  index_offset: Nat,
486	  total: U32,
487	  block_count: U32,
488	  root_hash: SHA.Stream,
489	  index_hash: SHA.Stream,
490	  index: U32,
491	  offset: U32,
492	  valid: Bool
493	) -> IO(Result<&1, &1, U32 & String, Reader>):
494	  match valid:
495	    case False{}: seek.block.invalid(file)
496	    case True{}:
497	      IO.pure(Result<&1, &1, U32 & String, Reader>,
498	        Done{Reader{file, State{size, header, index_offset,
499	          Nat.sub(U32.to_nat(offset), 20n), total, block_count, index, 0,
500	          root_hash, index_hash, None{}, True{}}}})
501	
502	def seek.block.ready(
503	  file: File,
504	  state: State,
505	  index: U32,
506	  offset: U32,
507	  valid: Bool
508	) -> IO(Result<&1, &1, U32 & String, Reader>):
509	  match state:
510	    case State{size, header, index_offset, _, total, block_count, _, _,
511	        root_hash, index_hash, _, _}:
512	      seek.block.construct(file, size, header, index_offset, total,
513	        block_count, root_hash, index_hash, index, offset, valid)
514	
515	def seek.block.decoded.offsets(
516	  state: State,
517	  index: U32,
518	  file: File,
519	  index_offset: Nat,
520	  offsets: Maybe<&2, U32> & Maybe<&2, U32>
521	) -> IO(Result<&1, &1, U32 & String, Reader>):
522	  match offsets:
523	    case (Some{+offset}, Some{length}):
524	      seek.block.ready(file, state, index, offset,
525	        BlockIndexPolicy.entry_valid(U32.to_nat(offset),
526	          U32.to_nat(length), index_offset))
527	    case _:
528	      seek.block.invalid(file)
529	
530	def seek.block.decoded(
531	  state: State,
532	  index: U32,
533	  file: File,
534	  offsets: Maybe<&2, U32> & Maybe<&2, U32>
535	) -> IO(Result<&1, &1, U32 & String, Reader>):
536	  match state:
537	    case State{_, _, +index_offset, _, _, _, _, _, _, _, _, _}:
538	      seek.block.decoded.offsets(state, index, file, index_offset, offsets)
539	
540	def seek.block.word(pair: Bytes.Bytes & Maybe<&2, U32>) -> Maybe<&2, U32>:
541	  match pair:
542	    case (_, value): value
543	
544	def seek.block.words.copies(
545	  pair: Bytes.Bytes & Bytes.Bytes
546	) -> Maybe<&2, U32> & Maybe<&2, U32>:
547	  match pair:
548	    case (offset, length):
549	      (seek.block.word(Bytes.get.u32be(offset, 0)),
550	        seek.block.word(Bytes.get.u32be(length, 4)))
551	
552	def seek.block.words(bytes: Bytes.Bytes) -> Maybe<&2, U32> & Maybe<&2, U32>:
553	  seek.block.words.copies(ByteSstFile.clone.split(bytes))
554	
555	def seek.block.read(
556	  state: State,
557	  index: U32,
558	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
559	) -> IO(Result<&1, &1, U32 & String, Reader>):
560	  match pair:
561	    case (file, Fail{_}): seek.block.invalid(file)
562	    case (file, Done{bytes}):
563	      seek.block.decoded(state, index, file, seek.block.words(bytes))
564	
565	def seek.block.indices.quad.first(
566	  firsts: Nat & Nat,
567	  second: Nat,
568	  third: Nat
569	) -> U32 & (U32 & (U32 & U32)):
570	  match firsts:
571	    case (first, first_copy):
572	      (U32.from_nat(first), (U32.from_nat(first_copy),
573	        (U32.from_nat(second), U32.from_nat(third))))
574	
575	def seek.block.indices.quad.result(
576	  triple: Nat & Nat & Nat
577	) -> U32 & (U32 & (U32 & U32)):
578	  match triple:
579	    case (first, second, third):
580	      seek.block.indices.quad.first(QueuePolicy.nat.clone.pair(first),
581	        second, third)
582	
583	def seek.block.indices.quad(value: U32) -> U32 & (U32 & (U32 & U32)):
584	  seek.block.indices.quad.result(Fs.read_at.nat.clone.triple(U32.to_nat(value)))
585	
586	def seek.block.u32.clone.pair.result(pair: Nat & Nat) -> U32 & U32:
587	  match pair:
588	    case (first, second): (U32.from_nat(first), U32.from_nat(second))
589	
590	def seek.block.u32.clone.pair(value: U32) -> U32 & U32:
591	  seek.block.u32.clone.pair.result(QueuePolicy.nat.clone.pair(U32.to_nat(value)))
592	
593	def seek.block.index.read(
594	  file: File,
595	  state: State,
596	  index_offset: Nat,
597	  indices: U32 & U32
598	) -> IO(Result<&1, &1, U32 & String, Reader>):
599	  match indices:
600	    case (address_index, read_index):
601	      do IO<Result<&1, &1, U32 & String, Reader>>:
602	        pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
603	          Fs.read_at(file, U32.from_nat(Nat.add(index_offset,
604	            Nat.add(8n, Nat.mul(U32.to_nat(address_index), 16n)))), 8)
605	        exact : File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult> =
606	          read_at.exact.pair(8, pair)
607	        seek.block.read(state, read_index, SstStreamIo.read.exact.state(exact))
608	
609	def seek.block.offsets(
610	  file: File,
611	  size: Nat,
612	  header: Bytes.Bytes,
613	  offsets: Nat & Nat,
614	  total: U32,
615	  block_count: U32,
616	  root_hash: SHA.Stream,
617	  index_hash: SHA.Stream,
618	  indices: U32 & U32 & U32,
619	  valid: Bool
620	) -> IO(Result<&1, &1, U32 & String, Reader>):
621	  match offsets:
622	    case (state_offset, read_offset):
623	      match indices:
624	        case (state_index, (address_index, read_index)):
625	          match valid:
626	            case False{}: seek.block.invalid(file)
627	            case True{}:
628	              seek.block.index.read(file,
629	                State{size, header, state_offset, 0n, total, block_count,
630	                  state_index, 0, root_hash, index_hash, None{}, True{}},
631	                read_offset, (address_index, read_index))
632	
633	def seek.block.validated(
634	  file: File,
635	  size: Nat,
636	  header: Bytes.Bytes,
637	  index_offset: Nat,
638	  total: U32,
639	  block_count: U32,
640	  root_hash: SHA.Stream,
641	  index_hash: SHA.Stream,
642	  indices: U32 & U32 & U32,
643	  valid: Bool
644	) -> IO(Result<&1, &1, U32 & String, Reader>):
645	  seek.block.offsets(file, size, header, QueuePolicy.nat.clone.pair(index_offset),
646	    total, block_count, root_hash, index_hash, indices, valid)
647	
648	def seek.block.checked.bounds(
649	  file: File,
650	  size: Nat,
651	  header: Bytes.Bytes,
652	  index_offset: Nat,
653	  total: U32,
654	  block_counts: U32 & U32,
655	  root_hash: SHA.Stream,
656	  index_hash: SHA.Stream,
657	  indices: U32 & (U32 & (U32 & U32))
658	) -> IO(Result<&1, &1, U32 & String, Reader>):
659	  match block_counts:
660	    case (result_count, check_count):
661	      match indices:
662	        case (compare_index, (state_index, (address_index, read_index))):
663	          seek.block.validated(file, size, header, index_offset, total,
664	            result_count, root_hash, index_hash,
665	            (state_index, (address_index, read_index)),
666	            BlockIndexPolicy.contains(U32.to_nat(check_count),
667	              U32.to_nat(compare_index)))
668	
669	def seek.block.checked.indices.ready(
670	  file: File,
671	  size: Nat,
672	  header: Bytes.Bytes,
673	  index_offset: Nat,
674	  total: U32,
675	  root_hash: SHA.Stream,
676	  index_hash: SHA.Stream,
677	  indices: U32 & (U32 & (U32 & U32)),
678	  block_count: U32
679	) -> IO(Result<&1, &1, U32 & String, Reader>):
680	  match indices:
681	    case (compare_index, (state_index, (address_index, read_index))):
682	      seek.block.checked.bounds(file, size, header, index_offset, total,
683	        seek.block.u32.clone.pair(block_count), root_hash, index_hash,
684	        (compare_index, (state_index, (address_index, read_index))))
685	
686	def seek.block.checked.indices(
687	  file: File,
688	  size: Nat,
689	  header: Bytes.Bytes,
690	  index_offset: Nat,
691	  total: U32,
692	  block_count: U32,
693	  root_hash: SHA.Stream,
694	  index_hash: SHA.Stream,
695	  indices: U32 & (U32 & (U32 & U32))
696	) -> IO(Result<&1, &1, U32 & String, Reader>):
697	  match block_count:
698	    case _:
699	      seek.block.checked.indices.ready(file, size, header, index_offset,
700	        total, root_hash, index_hash, indices, block_count)
701	
702	def seek.block.checked(
703	  file: File,
704	  state: State,
705	  indices: U32 & (U32 & (U32 & U32))
706	) -> IO(Result<&1, &1, U32 & String, Reader>):
707	  match state:
708	    case State{size, header, index_offset, _, total, block_count, _, _,
709	        root_hash, index_hash, _, _}:
710	      seek.block.checked.indices(file, size, header, index_offset, total,
711	        block_count, root_hash, index_hash, indices)
712	
713	# Positions within an SST whose complete index and root were validated by an earlier pass.
714	def seek_block_after_validation(
715	  reader: Reader,
716	  index: U32
717	) -> IO(Result<&1, &1, U32 & String, Reader>):
718	  match reader:
719	    case Reader{file, state}:
720	      seek.block.checked(file, state, seek.block.indices.quad(index))
721	
722	# Closes the file owned by a reader that has not reached its terminal state.
723	def close(reader: Reader) -> IO(Result<&1, &1, U32 & String, Unit>):
724	  match reader:
725	    case Reader{file, _}:
726	      do IO<Result<&1, &1, U32 & String, Unit>>:
727	        _closed : Unit <- File.close(file)
728	        return Done{Unit{}}
729	
730	def route.close.failed(
731	  file: File,
732	  error: U32 & String
733	) -> IO(Result<&1, &1, U32 & String, Reader>):
734	  do IO<Result<&1, &1, U32 & String, Reader>>:
735	    _closed : Unit <- File.close(file)
736	    return Fail{error}
737	
738	def route.record.bounds(
739	  file: File,
740	  state: State,
741	  +context: RouteContext,
742	  key: Bytes.Bytes,
743	  +search: BlockRoutePolicy.Search,
744	  mid: Nat,
745	  +offset: U32,
746	  +length: U32,
747	  separator_start: Nat,
748	  valid: Bool
749	) -> RouteState:
750	  match valid:
751	    case True{}:
752	      RouteSeparatorNeed{file, state, context, key, search, mid,
753	        U32.to_nat(offset), length, separator_start}
754	    case False{}:
755	      RouteFailed{file, corruption("byte SST separator exceeds the index region")}
756	
757	def route.record.words(
758	  file: File,
759	  state: State,
760	  context: RouteContext,
761	  key: Bytes.Bytes,
762	  search: BlockRoutePolicy.Search,
763	  mid: Nat,
764	  words: Maybe<&2, U32> & Maybe<&2, U32>
765	) -> RouteState:
766	  match context:
767	    case RouteContext{+size, +index_offset, +block_count}:
768	      match words:
769	        case (Some{+offset}, Some{+length}):
770	          +index_length = Nat.sub(size, Nat.add(index_offset, 44n))
771	          +fixed_bytes = Nat.add(8n, Nat.mul(block_count, 16n))
772	          +region_length = Nat.sub(index_length, fixed_bytes)
773	          +separator_start = Nat.add(index_offset, fixed_bytes)
774	          route.record.bounds(file, state, context, key, search, mid,
775	            offset, length, separator_start,
776	            BlockIndexFormatPolicy.separator_valid(U32.to_nat(offset),
777	              U32.to_nat(length), region_length) &&
778	              U32.is_le(length, StorageBytes.MAX_BLOCK_BYTES()))
779	        case _:
780	          RouteFailed{file, corruption("byte SST separator fields are missing")}
781	
782	def route.record.slice(
783	  file: File,
784	  state: State,
785	  context: RouteContext,
786	  key: Bytes.Bytes,
787	  search: BlockRoutePolicy.Search,
788	  mid: Nat,
789	  pair: Bytes.Bytes & Bytes.Bytes
790	) -> RouteState:
791	  match pair:
792	    case (_, metadata):
793	      route.record.words(file, state, context, key, search, mid,
794	        seek.block.words(metadata))
795	
796	def route.record.parsed(
797	  state: State,
798	  context: RouteContext,
799	  key: Bytes.Bytes,
800	  search: BlockRoutePolicy.Search,
801	  mid: Nat,
802	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
803	) -> RouteState:
804	  match pair:
805	    case (file, Fail{error}): RouteFailed{file, error}
806	    case (file, Done{record}):
807	      route.record.slice(file, state, context, key, search, mid,
808	        Bytes.slice(record, 8, 8))
809	
810	def route.separator.parsed(
811	  state: State,
812	  context: RouteContext,
813	  key: Bytes.Bytes,
814	  search: BlockRoutePolicy.Search,
815	  mid: Nat,
816	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
817	) -> RouteState:
818	  match pair:
819	    case (file, Fail{error}): RouteFailed{file, error}
820	    case (file, Done{separator}):
821	      RouteCompare{file, state, context, key, search, mid, separator}
822	
823	def route.compare.copies(
824	  file: File,
825	  state: State,
826	  context: RouteContext,
827	  search: BlockRoutePolicy.Search,
828	  mid: Nat,
829	  separator: Bytes.Bytes,
830	  copies: Bytes.Bytes & Bytes.Bytes
831	) -> RouteState:
832	  match copies:
833	    case (target, retained):
834	      RouteNeed{file, state, context, retained,
835	        BlockRoutePolicy.advance(search,
836	          BlockRoutePolicy.move(target, separator), mid)}
837	
838	def route.exhausted(state: RouteState) -> IO(Result<&1, &1, U32 & String, Reader>):
839	  match state:
840	    case RouteNeed{file, _, _, _, _}:
841	      route.close.failed(file, resource_limit("SST key seek budget exhausted"))
842	    case RouteSelect{reader, _}:
843	      match reader:
844	        case Reader{file, _}:
845	          route.close.failed(file,
846	            resource_limit("SST key seek budget exhausted"))
847	    case RouteRecordRead{pair, _, _, _, _, _}:
848	      match pair:
849	        case (file, _):
850	          route.close.failed(file, resource_limit("SST key seek budget exhausted"))
851	    case RouteSeparatorNeed{file, _, _, _, _, _, _, _, _}:
852	      route.close.failed(file, resource_limit("SST key seek budget exhausted"))
853	    case RouteSeparatorRead{pair, _, _, _, _, _}:
854	      match pair:
855	        case (file, _):
856	          route.close.failed(file, resource_limit("SST key seek budget exhausted"))
857	    case RouteCompare{file, _, _, _, _, _, _}:
858	      route.close.failed(file, resource_limit("SST key seek budget exhausted"))
859	    case RouteFailed{file, error}: route.close.failed(file, error)
860	
861	def route.midpoint(search: BlockRoutePolicy.Search) -> Nat:
862	  match search:
863	    case BlockRoutePolicy.Search{low, high, _}:
864	      BlockRoutePolicy.midpoint(low, high)
865	
866	def route.need(
867	  file: File,
868	  state: State,
869	  +context: RouteContext,
870	  key: Bytes.Bytes,
871	  +search: BlockRoutePolicy.Search,
872	  active: Bool
873	) -> IO(RouteState):
874	  match context:
875	    case RouteContext{_, index_offset, _}:
876	      match active:
877	        case False{}:
878	          IO.pure(RouteState, RouteSelect{Reader{file, state},
879	            BlockRoutePolicy.selected_block(search)})
880	        case True{}:
881	          +mid = route.midpoint(search)
882	          do IO<RouteState>:
883	            raw : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
884	              Fs.read_at(file, U32.from_nat(Nat.add(index_offset,
885	                Nat.add(8n, Nat.mul(mid, 16n)))), 16)
886	            return RouteRecordRead{
887	              read_at.exact.pair(16, raw), state, context, key, search, mid}
888	
889	def route.loop(fuel: Nat, pending: RouteState) -> IO(Result<&1, &1, U32 & String, Reader>):
890	  match fuel:
891	    case 0n: route.exhausted(pending)
892	    case 1n+rest:
893	      match pending:
894	        case RouteNeed{file, state, context, key, +search}:
895	          do IO<Result<&1, &1, U32 & String, Reader>>:
896	            next : RouteState <- route.need(file, state, context, key, search,
897	              BlockRoutePolicy.active(search))
898	            route.loop(rest, next)
899	        case RouteSelect{reader, block}:
900	          seek_block_after_validation(reader, U32.from_nat(block))
901	        case RouteRecordRead{pair, state, context, key, search, mid}:
902	          route.loop(rest, route.record.parsed(state, context, key, search,
903	            mid, SstStreamIo.read.exact.state(pair)))
904	        case RouteSeparatorNeed{file, state, context, key, search, mid,
905	            offset, +length, separator_start}:
906	          do IO<Result<&1, &1, U32 & String, Reader>>:
907	            raw : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
908	              Fs.read_at(file, U32.from_nat(Nat.add(separator_start, offset)),
909	                length)
910	            route.loop(rest, RouteSeparatorRead{
911	              read_at.exact.pair(length, raw), state, context, key, search,
912	              mid})
913	        case RouteSeparatorRead{pair, state, context, key, search, mid}:
914	          route.loop(rest, route.separator.parsed(state, context, key,
915	            search, mid, SstStreamIo.read.exact.state(pair)))
916	        case RouteCompare{file, state, context, key, search, mid, separator}:
917	          route.loop(rest, route.compare.copies(file, state, context, search,
918	            mid, separator, ByteMemTable.bytes.clone(key)))
919	        case RouteFailed{file, error}: route.close.failed(file, error)
920	
921	# Routes a lower key to the last block whose separator does not exceed it.
922	def seek_key_after_validation(
923	  reader: Reader,
924	  key: Bytes.Bytes
925	) -> IO(Result<&1, &1, U32 & String, Reader>):
926	  match reader:
927	    case Reader{file, state}:
928	      match state:
929	        case State{+size, header, +index_offset, used, total,
930	            +block_count, next_block, decoded, root_hash, index_hash,
931	            last_key, selected}:
932	          +count = U32.to_nat(block_count)
933	          +fuel = Nat.add(Nat.mul(8n, count), 8n)
934	          context : RouteContext = RouteContext{size, index_offset, count}
935	          retained_state : State = State{size, header, index_offset, used, total, block_count,
936	            next_block, decoded, root_hash, index_hash, last_key, selected}
937	          route.loop(fuel, RouteNeed{file, retained_state, context, key,
938	            BlockRoutePolicy.Search{0n, count, 0n}})
939	
940	# Handles measure key result.
941	def measure_key_result(
942	  result: Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>
943	) -> Result<&1, &1, U32 & String, ByteWal.ValueMetadata>:
944	  match result:
945	    case Done{value}: Done{value}
946	    case Fail{Wal.ChecksumMismatch{}}:
947	      Fail{corruption("byte SST block checksum mismatch")}
948	    case Fail{Wal.Malformed{}}:
949	      Fail{corruption("malformed byte SST metadata")}
950	    case Fail{Wal.TooLarge{}}:
951	      Fail{corruption("byte SST metadata exceeds decoder bounds")}
952	    case Fail{Wal.InvalidUtf8{}}:
953	      Fail{corruption("byte SST metadata contains invalid text")}
954	
955	# Handles measure key closed.
956	def measure_key_closed(
957	  file: File,
958	  result: Result<&1, &1, Wal.Error, ByteWal.ValueMetadata>
959	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
960	  do IO<Result<&1, &1, U32 & String, ByteWal.ValueMetadata>>:
961	    _closed : Unit <- File.close(file)
962	    return measure_key_result(result)
963	
964	# Handles measure key body result.
965	def measure_key_body_result(
966	  key: Bytes.Bytes,
967	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
968	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
969	  match pair:
970	    case (file, Fail{error}):
971	      do IO<Result<&1, &1, U32 & String, ByteWal.ValueMetadata>>:
972	        _closed : Unit <- File.close(file)
973	        return Fail{error}
974	    case (file, Done{block}):
975	      measure_key_closed(file, ByteSstFile.measure_block(block, key))
976	
977	# Handles measure key body read.
978	def measure_key_body_read(
979	  +used: Nat,
980	  key: Bytes.Bytes,
981	  +payload_length: U32,
982	  file: File
983	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
984	  do IO<Result<&1, &1, U32 & String, ByteWal.ValueMetadata>>:
985	    block : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
986	      Fs.read_at(file, U32.from_nat(Nat.add(20n, used)),
987	        U32.add(payload_length, 40))
988	    measure_key_body_result(key, block)
989	
990	# Handles measure key bounds.
991	def measure_key_bounds(
992	  +index_offset: Nat,
993	  +used: Nat,
994	  +payload_length: U32
995	) -> Bool:
996	  U32.is_le(payload_length, StorageBytes.MAX_BLOCK_BYTES()) &&
997	    Nat.is_le(Nat.add(20n, Nat.add(used,
998	      Nat.add(U32.to_nat(payload_length), 40n))), index_offset)
999	
1000	# Handles measure key header valid.
1001	def measure_key_header_valid(
1002	  +_index_offset: Nat,
1003	  +used: Nat,
1004	  key: Bytes.Bytes,
1005	  +payload_length: U32,
1006	  file: File,
1007	  valid: Bool
1008	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
1009	  match valid:
1010	    case True{}: measure_key_body_read(used, key, payload_length, file)
1011	    case False{}:
1012	      do IO<Result<&1, &1, U32 & String, ByteWal.ValueMetadata>>:
1013	        _closed : Unit <- File.close(file)
1014	        return Fail{corruption("byte SST block exceeds file bounds")}
1015	
1016	# Handles measure key header length.
1017	def measure_key_header_length(
1018	  +index_offset: Nat,
1019	  +used: Nat,
1020	  key: Bytes.Bytes,
1021	  pair: Bytes.Bytes & Maybe<&2, U32>,
1022	  file: File
1023	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
1024	  match pair:
1025	    case (_, None{}):
1026	      do IO<Result<&1, &1, U32 & String, ByteWal.ValueMetadata>>:
1027	        _closed : Unit <- File.close(file)
1028	        return Fail{corruption("invalid byte SST block length")}
1029	    case (_, Some{+payload_length}):
1030	      measure_key_header_valid(index_offset, used, key, payload_length,
1031	        file, measure_key_bounds(index_offset, used, payload_length))
1032	
1033	# Handles measure key header exact.
1034	def measure_key_header_exact(
1035	  +index_offset: Nat,
1036	  +used: Nat,
1037	  key: Bytes.Bytes,
1038	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
1039	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
1040	  match pair:
1041	    case (file, Fail{error}):
1042	      do IO<Result<&1, &1, U32 & String, ByteWal.ValueMetadata>>:
1043	        _closed : Unit <- File.close(file)
1044	        return Fail{error}
1045	    case (file, Done{header}):
1046	      measure_key_header_length(index_offset, used, key,
1047	        Bytes.get.u32be(header, 0), file)
1048	
1049	# Handles measure key header.
1050	def measure_key_header(
1051	  +index_offset: Nat,
1052	  +used: Nat,
1053	  key: Bytes.Bytes,
1054	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
1055	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
1056	  measure_key_header_exact(index_offset, used, key,
1057	    SstStreamIo.read.exact.state(read_at.exact.pair(8, pair)))
1058	
1059	def measure.key.block(
1060	  reader: Reader,
1061	  key: Bytes.Bytes
1062	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
1063	  match reader:
1064	    case Reader{file, state}:
1065	      match state:
1066	        case State{_, _, +index_offset, +used, _, _, _, _, _, _, _, _}:
1067	          do IO<Result<&1, &1, U32 & String, ByteWal.ValueMetadata>>:
1068	            header : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
1069	              Fs.read_at(file, U32.from_nat(Nat.add(20n, used)), 8)
1070	            measure_key_header(index_offset, used, key, header)
1071	
1072	# Handles measure key routed.
1073	def measure_key_routed(
1074	  key: Bytes.Bytes,
1075	  result: Result<&1, &1, U32 & String, Reader>
1076	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
1077	  match result:
1078	    case Fail{error}:
1079	      IO.pure(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>,
1080	        Fail{error})
1081	    case Done{reader}: measure.key.block(reader, key)
1082	
1083	def measure_key.route(
1084	  reader: Reader,
1085	  route_key: Bytes.Bytes,
1086	  query_key: Bytes.Bytes
1087	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
1088	  do IO<Result<&1, &1, U32 & String, ByteWal.ValueMetadata>>:
1089	    routed : Result<&1, &1, U32 & String, Reader> <-
1090	      seek_key_after_validation(reader, route_key)
1091	    measure_key_routed(query_key, routed)
1092	
1093	def measure_key.copied(
1094	  reader: Reader,
1095	  keys: Bytes.Bytes & Bytes.Bytes
1096	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
1097	  match keys:
1098	    case (route_key, query_key):
1099	      measure_key.route(reader, route_key, query_key)
1100	
1101	# Handles measure key.
1102	def measure_key(
1103	  reader: Reader,
1104	  key: Bytes.Bytes
1105	) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
1106	  measure_key.copied(reader, ByteMemTable.bytes.clone(key))
1107	
1108	def next.body.output.index.hash(
1109	  file: File,
1110	  +size: Nat,
1111	  header: Bytes.Bytes,
1112	  +index_offset: Nat,
1113	  +next_used: Nat,
1114	  total: U32,
1115	  block_count: U32,
1116	  next_number: U32,
1117	  next_decoded: U32,
1118	  last_key: Maybe<&1, Bytes.Bytes>,
1119	  entries: List<&1, ByteMemTable.Entry>,
1120	  root_hash: SHA.Stream,
1121	  index_hash: SHA.Stream,
1122	  selected: Bool
1123	) -> IO(Result<&1, &1, U32 & String, Step>):
1124	  IO.pure(Result<&1, &1, U32 & String, Step>,
1125	    Done{Block{Reader{file, State{size, header, index_offset, next_used,
1126	      total, block_count, next_number, next_decoded, root_hash, index_hash,
1127	      last_key, selected}}, entries}})
1128	
1129	
1130	
1131	def next.body.output.hash.results(
1132	  file: File,
1133	  +size: Nat,
1134	  header: Bytes.Bytes,
1135	  +index_offset: Nat,
1136	  +next_used: Nat,
1137	  total: U32,
1138	  block_count: U32,
1139	  next_number: U32,
1140	  next_decoded: U32,
1141	  last_key: Maybe<&1, Bytes.Bytes>,
1142	  entries: List<&1, ByteMemTable.Entry>,
1143	  root_result: Result<&1, &1, SHA.StreamError, SHA.Stream>,
1144	  index_hash: SHA.Stream,
1145	  selected: Bool
1146	) -> IO(Result<&1, &1, U32 & String, Step>):
1147	  match root_result:
1148	    case Fail{_}:
1149	      close.failed(file, resource_limit("SST root hash exceeds the SHA-256 input limit"))
1150	    case Done{root_hash}:
1151	      next.body.output.index.hash(file, size, header, index_offset,
1152	        next_used, total, block_count, next_number, next_decoded,
1153	        last_key, entries, root_hash, index_hash, selected)
1154	
1155	
1156	def next.body.output.hashes(
1157	  file: File,
1158	  +size: Nat,
1159	  header: Bytes.Bytes,
1160	  +index_offset: Nat,
1161	  +next_used: Nat,
1162	  total: U32,
1163	  block_count: U32,
1164	  next_number: U32,
1165	  next_decoded: U32,
1166	  last_key: Maybe<&1, Bytes.Bytes>,
1167	  entries: List<&1, ByteMemTable.Entry>,
1168	  checksum: Bytes.Bytes,
1169	  root_hash: SHA.Stream,
1170	  index_hash: SHA.Stream,
1171	  selected: Bool
1172	) -> IO(Result<&1, &1, U32 & String, Step>):
1173	  next.body.output.hash.results(file, size, header, index_offset,
1174	    next_used, total, block_count, next_number, next_decoded,
1175	    last_key, entries, SHA.stream.update(root_hash, checksum),
1176	    index_hash, selected)
1177	
1178	def next.body.output.count(
1179	  file: File,
1180	  +size: Nat,
1181	  header: Bytes.Bytes,
1182	  +index_offset: Nat,
1183	  +used: Nat,
1184	  total: U32,
1185	  block_count: U32,
1186	  next_block: U32,
1187	  decoded: U32,
1188	  root_hash: SHA.Stream,
1189	  index_hash: SHA.Stream,
1190	  block_bytes: Bytes.Bytes,
1191	  +payload_length: U32,
1192	  entry_count: U32,
1193	  entries: List<&1, ByteMemTable.Entry>,
1194	  last_key: Maybe<&1, Bytes.Bytes>,
1195	  valid: Bool,
1196	  selected: Bool
1197	) -> IO(Result<&1, &1, U32 & String, Step>):
1198	  match valid:
1199	    case False{}:
1200	      close.failed(file, corruption("byte SST keys are not strictly ordered"))
1201	    case True{}:
1202	      +block_length = U32.add(40, payload_length)
1203	      +checksum_offset = U32.add(8, payload_length)
1204	      block_checksum = bytes.slice.result(Bytes.slice(block_bytes,
1205	        checksum_offset, 32))
1206	      +next_used = Nat.add(used, U32.to_nat(block_length))
1207	      +next_decoded = U32.add(decoded, entry_count)
1208	      +next_number = U32.add(next_block, 1)
1209	      next.body.output.hashes(file, size, header, index_offset, next_used,
1210	        total, block_count, next_number, next_decoded, last_key, entries,
1211	        block_checksum, root_hash, index_hash, selected)
1212	
1213	def next.body.output.order(
1214	  file: File,
1215	  +size: Nat,
1216	  header: Bytes.Bytes,
1217	  +index_offset: Nat,
1218	  +used: Nat,
1219	  total: U32,
1220	  block_count: U32,
1221	  next_block: U32,
1222	  decoded: U32,
1223	  root_hash: SHA.Stream,
1224	  index_hash: SHA.Stream,
1225	  block_bytes: Bytes.Bytes,
1226	  +payload_length: U32,
1227	  raw_count: Nat,
1228	  entries: List<&1, ByteMemTable.Entry>,
1229	  order: BlockOrder.State,
1230	  selected: Bool
1231	) -> IO(Result<&1, &1, U32 & String, Step>):
1232	  match order:
1233	    case BlockOrder.State{valid, last_key, +count}:
1234	      next.body.output.count(file, size, header, index_offset, used, total,
1235	        block_count, next_block, decoded,
1236	        root_hash, index_hash,
1237	        block_bytes, payload_length, U32.from_nat(count), entries, last_key,
1238	        valid && Nat.is_eq(raw_count, count), selected)
1239	
1240	def next.body.output(
1241	  file: File,
1242	  state: State,
1243	  block_bytes: Bytes.Bytes,
1244	  +payload_length: U32,
1245	  raw_count: Nat,
1246	  copies: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
1247	) -> IO(Result<&1, &1, U32 & String, Step>):
1248	  match state:
1249	    case State{+size, header, +index_offset, +used, total, block_count,
1250	        next_block, decoded, root_hash, index_hash, last_key, selected}:
1251	      match copies:
1252	        case (for_order, retained):
1253	          next.body.output.order(file, size, header, index_offset, used,
1254	            total, block_count, next_block, decoded,
1255	            root_hash, index_hash,
1256	            block_bytes, payload_length, raw_count, retained,
1257	            BlockOrder.go(for_order, BlockOrder.empty(last_key)), selected)
1258	
1259	def next.body.ordered(
1260	  file: File,
1261	  state: State,
1262	  block_bytes: Bytes.Bytes,
1263	  +payload_length: U32,
1264	  measured: List<&1, ByteMemTable.Entry> & Nat
1265	) -> IO(Result<&1, &1, U32 & String, Step>):
1266	  match measured:
1267	    case (entries, raw_count):
1268	      next.body.output(file, state, block_bytes, payload_length, raw_count,
1269	        ByteMemTable.entries.clone(ByteMemTable.ordered_entries(
1270	          ByteMemTable.Table{entries})))
1271	
1272	def next.body.parsed(
1273	  file: File,
1274	  state: State,
1275	  block_bytes: Bytes.Bytes,
1276	  +payload_length: U32,
1277	  valid_checksum: Bool,
1278	  result: Result<&1, &1, ByteSstFile.Error, ByteMemTable.Table>
1279	) -> IO(Result<&1, &1, U32 & String, Step>):
1280	  match valid_checksum:
1281	    case False{}:
1282	      close.failed(file, corruption("byte SST block checksum mismatch"))
1283	    case True{}:
1284	      match result:
1285	        case Fail{error}:
1286	          close.failed(file, codec.error(error))
1287	        case Done{ByteMemTable.Table{entries}}:
1288	          next.body.ordered(file, state, block_bytes, payload_length,
1289	            ByteMemTable.count.keep(entries))
1290	
1291	def next.body.copies.checked(
1292	  file: File,
1293	  state: State,
1294	  block_bytes: Bytes.Bytes,
1295	  +payload_length: U32,
1296	  copies: Bytes.Bytes & Bytes.Bytes
1297	) -> IO(Result<&1, &1, U32 & String, Step>):
1298	  match copies:
1299	    case (for_checksum, for_parser):
1300	      next.body.parsed(file, state, block_bytes, payload_length,
1301	        block.checksum.valid(for_checksum, payload_length),
1302	        ByteSstFile.parse_block(for_parser))
1303	
1304	def next.body.copies(
1305	  file: File,
1306	  state: State,
1307	  +payload_length: U32,
1308	  copies: Bytes.Bytes & Bytes.Bytes
1309	) -> IO(Result<&1, &1, U32 & String, Step>):
1310	  match copies:
1311	    case (for_parse, retained):
1312	      next.body.copies.checked(file, state, retained, payload_length,
1313	        ByteSstFile.clone.split(for_parse))
1314	
1315	def next.body.exact(
1316	  context: BodyContext,
1317	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
1318	) -> IO(Result<&1, &1, U32 & String, Step>):
1319	  match context:
1320	    case BodyContext{state, block_header, +payload_length}:
1321	      match pair:
1322	        case (read_file, Fail{error}):
1323	          close.failed(read_file, error)
1324	        case (read_file, Done{body}):
1325	          next.body.copies(read_file, state, payload_length,
1326	            ByteSstFile.clone.split(Bytes.concat([block_header, body])))
1327	
1328	def next.body.result(
1329	  context: BodyContext,
1330	  pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>
1331	) -> IO(Result<&1, &1, U32 & String, Step>):
1332	  match context:
1333	    case BodyContext{state, block_header, +payload_length}:
1334	      match pair:
1335	        case (file, result):
1336	          next.body.exact(BodyContext{state, block_header, payload_length},
1337	            SstStreamIo.read.exact.state((file, result)))
1338	
1339	def next.body.valid(
1340	  file: File,
1341	  state: State,
1342	  block_header: Bytes.Bytes,
1343	  +used: Nat,
1344	  +payload_length: U32,
1345	  valid: Bool
1346	) -> IO(Result<&1, &1, U32 & String, Step>):
1347	  match valid:
1348	    case False{}:
1349	      close.failed(file, corruption("byte SST block exceeds file bounds"))
1350	    case True{}:
1351	      do IO<Result<&1, &1, U32 & String, Step>>:
1352	        raw : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
1353	          Fs.read_at(file,
1354	            U32.from_nat(Nat.add(Nat.add(20n, used), 8n)),
1355	            U32.add(payload_length, 32))
1356	        next.body.result(BodyContext{state, block_header, payload_length},
1357	          read_at.exact.pair(U32.add(payload_length, 32), raw))
1358	
1359	def next.block.length.state(
1360	  file: File,
1361	  state: State,
1362	  block_header: Bytes.Bytes,
1363	  +payload_length: U32
1364	) -> IO(Result<&1, &1, U32 & String, Step>):
1365	  match state:
1366	    case State{_, _, +index_offset, +used, _, _, _, _, _, _, _, _}:
1367	      next.body.valid(file, state, block_header, used, payload_length,
1368	        U32.is_le(payload_length, StorageBytes.MAX_BLOCK_BYTES()) &&
1369	          Nat.is_le(Nat.add(used,
1370	            U32.to_nat(U32.add(payload_length, 40))), index_offset))
1371	
1372	
1373	
1374	def next.block.length(
1375	  file: File,
1376	  state: State,
1377	  block_header: Bytes.Bytes,
1378	  result: Bytes.Bytes & Maybe<&2, U32>
1379	) -> IO(Result<&1, &1, U32 & String, Step>):
1380	  match result:
1381	    case (_, None{}):
1382	      close.failed(file, corruption("invalid byte SST block length"))
1383	    case (_, Some{payload_length}):
1384	      next.block.length.state(file, state, block_header, payload_length)
1385	def next.block.header.copies(
1386	  file: File,
1387	  state: State,
1388	  copies: Bytes.Bytes & Bytes.Bytes
1389	) -> IO(Result<&1, &1, U32 & String, Step>):
1390	  match copies:
1391	    case (for_length, retained):
1392	      next.block.length(file, state, retained, Bytes.get.u32be(for_length, 0))
1393	
1394	def next.block.header(
1395	  file: File,
1396	  state: State,
1397	  result: Result<&1, &1, U32 & String, Bytes.Bytes>
1398	) -> IO(Result<&1, &1, U32 & String, Step>):
1399	  match result:
1400	    case Fail{error}:
1401	      close.failed(file, error)
1402	    case Done{block_header}:
1403	      next.block.header.copies(file, state,
1404	        ByteSstFile.clone.split(block_header))
1405	def next.block.header.bytes(
1406	  state: State,
1407	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
1408	) -> IO(Result<&1, &1, U32 & String, Step>):
1409	  match pair:
1410	    case (file, result):
1411	      next.block.header(file, state, result)
1412	
1413	def next.block.header.pair(
1414	  state: State,
1415	  pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>
1416	) -> IO(Result<&1, &1, U32 & String, Step>):
1417	  next.block.header.bytes(state, SstStreamIo.read.exact.state(pair))
1418	
1419	
1420	def next.block.read(
1421	  file: File,
1422	  state: State
1423	) -> IO(Result<&1, &1, U32 & String, Step>):
1424	  match state:
1425	    case State{_, _, _, +used, _, _, _, _, _, _, _, _}:
1426	      do IO<Result<&1, &1, U32 & String, Step>>:
1427	        raw : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
1428	          Fs.read_at(file, U32.from_nat(Nat.add(20n, used)), 8)
1429	        next.block.header.pair(state, read_at.exact.pair(8, raw))
1430	
1431	def next.trailer.finish.footer(
1432	  file: File,
1433	  _footer: Bytes.Bytes,
1434	  _expected: Bytes.Bytes,
1435	  valid: Bool
1436	) -> IO(Result<&1, &1, U32 & String, Step>):
1437	  match valid:
1438	    case False{}:
1439	      close.failed(file, corruption("byte SST root checksum mismatch"))
1440	    case True{}:
1441	      do IO<Result<&1, &1, U32 & String, Step>>:
1442	        _closed : Unit <- File.close(file)
1443	        return Done{End{}}
1444	
1445	def next.trailer.finish.root.compare.footer(
1446	  file: File,
1447	  footers: Bytes.Bytes & Bytes.Bytes,
1448	  expected: Bytes.Bytes,
1449	  for_compare: Bytes.Bytes
1450	) -> IO(Result<&1, &1, U32 & String, Step>):
1451	  match footers:
1452	    case (for_result, footer):
1453	      next.trailer.finish.footer(file, for_result, expected,
1454	        block.checksum.compare(Bytes.eq(footer, for_compare)))
1455	
1456	
1457	
1458	def next.trailer.finish.root.compare(
1459	  file: File,
1460	  footer: Bytes.Bytes,
1461	  copies: Bytes.Bytes & Bytes.Bytes
1462	) -> IO(Result<&1, &1, U32 & String, Step>):
1463	  match copies:
1464	    case (for_result, for_compare):
1465	      next.trailer.finish.root.compare.footer(file,
1466	        ByteSstFile.clone.split(footer), for_result, for_compare)
1467	
1468	def next.trailer.finish.root.expected(
1469	  file: File,
1470	  footer: Bytes.Bytes,
1471	  expected: Bytes.Bytes
1472	) -> IO(Result<&1, &1, U32 & String, Step>):
1473	  next.trailer.finish.root.compare(file, footer,
1474	    ByteSstFile.clone.split(expected))
1475	
1476	
1477	def next.trailer.finish.root.digest(
1478	  file: File,
1479	  +total: U32,
1480	  +index_length: U32,
1481	  footer: Bytes.Bytes,
1482	  digest: Bytes.Bytes
1483	) -> IO(Result<&1, &1, U32 & String, Step>):
1484	  next.trailer.finish.root.expected(file, footer,
1485	    Bytes.concat([ByteSstFile.footer.magic(), ByteSstFile.u32.bytes(total),
1486	      ByteSstFile.u32.bytes(index_length), digest]))
1487	
1488	
1489	def index.page.footer.finish(
1490	  file: File,
1491	  +total: U32,
1492	  +index_length: U32,
1493	  footer: Bytes.Bytes,
1494	  root_hash: SHA.Stream,
1495	  valid_index: Bool
1496	) -> IO(Result<&1, &1, U32 & String, Step>):
1497	  match valid_index:
1498	    case False{}:
1499	      close.failed(file, corruption("byte SST serialized index mismatch"))
1500	    case True{}:
1501	      next.trailer.finish.root.digest(file, total, index_length, footer,
1502	        SHA.stream.finish(root_hash))
1503	
1504	def index.page.footer.pair(
1505	  file: File,
1506	  reader: IndexMeta,
1507	  footer: Bytes.Bytes,
1508	  root_hash: SHA.Stream
1509	) -> IO(Result<&1, &1, U32 & String, Step>):
1510	  match reader:
1511	    case IndexMeta{+total, +index_length, _}:
1512	      index.page.footer.finish(file, total, index_length, footer, root_hash,
1513	        True{})
1514	
1515	def index.page.footer.exact(
1516	  pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>,
1517	  reader: IndexMeta,
1518	  root_hash: SHA.Stream
1519	) -> IO(Result<&1, &1, U32 & String, Step>):
1520	  match pair:
1521	    case (file, Fail{error}): close.failed(file, error)
1522	    case (file, Done{SstStreamIo.ExactResult{footer, complete}}):
1523	      match complete:
1524	        case False{}: close.failed(file, corruption("truncated byte SST footer"))
1525	        case True{}:
1526	          index.page.footer.pair(file, reader, footer, root_hash)
1527	
1528	def index.page.footer.read(
1529	  file: File,
1530	  reader: IndexMeta,
1531	  offset: Nat,
1532	  root_hash: SHA.Stream
1533	) -> IO(Result<&1, &1, U32 & String, Step>):
1534	  do IO<Result<&1, &1, U32 & String, Step>>:
1535	    raw : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
1536	      Fs.read_at(file, U32.from_nat(offset), 44)
1537	    index.page.footer.exact(read_at.exact.pair(44, raw), reader, root_hash)
1538	
1539	def index.page.decode.root(
1540	  reader: IndexMeta,
1541	  +offset: Nat,
1542	  +remaining: Nat,
1543	  actual_hash: SHA.Stream,
1544	  root_result: Result<&1, &1, SHA.StreamError, SHA.Stream>
1545	) -> Result<&1, &1, U32 & String, IndexPageData>:
1546	  match root_result:
1547	    case Fail{_}:
1548	      Fail{resource_limit("SST root hash exceeds the SHA-256 input limit")}
1549	    case Done{next_root_hash}:
1550	      +page_size = BlockIndexPolicy.read_bytes(remaining)
1551	      Done{IndexPageData{reader, Nat.add(offset, page_size),
1552	        Nat.sub(remaining, page_size), actual_hash, next_root_hash}}
1553	
1554	def index.page.decode.actual(
1555	  reader: IndexMeta,
1556	  +offset: Nat,
1557	  +remaining: Nat,
1558	  root_hash: SHA.Stream,
1559	  actual_result: Result<&1, &1, SHA.StreamError, SHA.Stream>,
1560	  for_root: Bytes.Bytes
1561	) -> Result<&1, &1, U32 & String, IndexPageData>:
1562	  match actual_result:
1563	    case Fail{_}:
1564	      Fail{resource_limit("SST index hash exceeds the SHA-256 input limit")}
1565	    case Done{next_actual_hash}:
1566	      index.page.decode.root(reader, offset, remaining, next_actual_hash,
1567	        SHA.stream.update(root_hash, for_root))
1568	
1569	def index.page.decode.copies(
1570	  reader: IndexMeta,
1571	  +offset: Nat,
1572	  +remaining: Nat,
1573	  actual_hash: SHA.Stream,
1574	  root_hash: SHA.Stream,
1575	  copies: Bytes.Bytes & Bytes.Bytes,
1576	  complete: Bool
1577	) -> Result<&1, &1, U32 & String, IndexPageData>:
1578	  match copies:
1579	    case (for_hash, for_root):
1580	      match complete:
1581	        case False{}: Fail{corruption("truncated byte SST serialized index")}
1582	        case True{}:
1583	          index.page.decode.actual(reader, offset, remaining, root_hash,
1584	            SHA.stream.update(actual_hash, for_hash), for_root)
1585	
1586	def index.page.decode(
1587	  reader: IndexMeta,
1588	  +offset: Nat,
1589	  +remaining: Nat,
1590	  actual_hash: SHA.Stream,
1591	  root_hash: SHA.Stream,
1592	  bytes: Bytes.Bytes,
1593	  complete: Bool
1594	) -> Result<&1, &1, U32 & String, IndexPageData>:
1595	  index.page.decode.copies(reader, offset, remaining, actual_hash, root_hash,
1596	    ByteSstFile.clone.split(bytes), complete)
1597	
1598	def index.page.loop(fuel: Nat, state: IndexPageState) -> IO(Result<&1, &1, U32 & String, Step>):
1599	  match fuel:
1600	    case 0n:
1601	      match state:
1602	        case IndexPageNeed{file, _, _, _, _, _}:
1603	          close.failed(file, resource_limit("SST index page traversal limit exceeded"))
1604	        case IndexPageRead{pair, _, _, _, _, _}:
1605	          match pair:
1606	            case (file, _):
1607	              close.failed(file, resource_limit("SST index page traversal limit exceeded"))
1608	        case IndexPageParsed{file, _}:
1609	          close.failed(file, resource_limit("SST index page traversal limit exceeded"))
1610	    case 1n+rest:
1611	      match state:
1612	        case IndexPageNeed{file, reader, +offset, +remaining,
1613	            actual_hash, root_hash}:
1614	          match remaining:
1615	            case 0n:
1616	              index.page.footer.read(file, reader, offset, root_hash)
1617	            case 1n+_:
1618	              +needed = U32.from_nat(BlockIndexPolicy.read_bytes(remaining))
1619	              do IO<Result<&1, &1, U32 & String, Step>>:
1620	                raw : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
1621	                  Fs.read_at(file, U32.from_nat(offset), needed)
1622	                index.page.loop(rest, IndexPageRead{
1623	                  read_at.exact.pair(needed, raw), reader, offset, remaining,
1624	                  actual_hash, root_hash})
1625	        case IndexPageRead{pair, reader, +offset, +remaining,
1626	            actual_hash, root_hash}:
1627	          match pair:
1628	            case (file, Fail{error}): close.failed(file, error)
1629	            case (file, Done{SstStreamIo.ExactResult{bytes, complete}}):
1630	              index.page.loop(rest, IndexPageParsed{file,
1631	                index.page.decode(reader, offset, remaining,
1632	                  actual_hash, root_hash, bytes, complete)})
1633	        case IndexPageParsed{file, result}:
1634	          match result:
1635	            case Fail{error}: close.failed(file, error)
1636	            case Done{IndexPageData{reader, offset, remaining,
1637	                actual_hash, root_hash}}:
1638	              index.page.loop(rest, IndexPageNeed{file, reader, offset,
1639	                remaining, actual_hash, root_hash})
1640	            case Done{_}:
1641	              close.failed(file, corruption("invalid SST index traversal state"))
1642	
1643	def index.page.start(file: File, state: State) -> IO(Result<&1, &1, U32 & String, Step>):
1644	  match state:
1645	    case State{+size, _, +index_offset, _, +total, _, _, _, root_hash,
1646	        index_hash, _, _}:
1647	      +remaining = Nat.sub(size, Nat.add(index_offset, 44n))
1648	      +index_length = U32.from_nat(remaining)
1649	      +page_bytes = BlockIndexPolicy.max_bytes()
1650	      +pages = Nat.div(Nat.add(remaining, Nat.sub(page_bytes, 1n)), page_bytes)
1651	      +fuel = Nat.add(Nat.mul(3n, pages), 1n)
1652	      index.page.loop(fuel, IndexPageNeed{file,
1653	        IndexMeta{total, index_length, index_hash}, index_offset, remaining,
1654	        SHA.stream.initial(), root_hash})
1655	
1656	def next.trailer.read(
1657	  file: File,
1658	  state: State
1659	) -> IO(Result<&1, &1, U32 & String, Step>):
1660	  index.page.start(file, state)
1661	
1662	def next.trailer.ready(
1663	  file: File,
1664	  state: State,
1665	  valid: Bool
1666	) -> IO(Result<&1, &1, U32 & String, Step>):
1667	  match valid:
1668	    case False{}:
1669	      close.failed(file, corruption("byte SST data count or index offset mismatch"))
1670	    case True{}:
1671	      next.trailer.read(file, state)
1672	
1673	def next.selected.end(file: File) -> IO(Result<&1, &1, U32 & String, Step>):
1674	  do IO<Result<&1, &1, U32 & String, Step>>:
1675	    _closed : Unit <- File.close(file)
1676	    return Done{End{}}
1677	
1678	def next.trailer.state(file: File, state: State) -> IO(Result<&1, &1, U32 & String, Step>):
1679	  match state:
1680	    case State{_, _, +index_offset, +used, +total, _, _, +decoded, _, _, _, selected}:
1681	      match selected:
1682	        case True{}: next.selected.end(file)
1683	        case False{}:
1684	          next.trailer.ready(file, state,
1685	            Nat.is_eq(Nat.add(20n, used), index_offset) && U32.is_eq(total, decoded))
1686	
1687	
1688	def next.state.block.finished(
1689	  file: File,
1690	  state: State,
1691	  finished: Bool
1692	) -> IO(Result<&1, &1, U32 & String, Step>):
1693	  match finished:
1694	    case False{}:
1695	      next.block.read(file, state)
1696	    case True{}:
1697	      next.trailer.state(file, state)
1698	
1699	def next.state.block(
1700	  file: File,
1701	  state: State
1702	) -> IO(Result<&1, &1, U32 & String, Step>):
1703	  match state:
1704	    case State{_, _, _, _, _, +block_count, +next_block, _, _, _, _, _}:
1705	      next.state.block.finished(file, state,
1706	        U32.is_eq(next_block, block_count))
1707	
1708	def next.state(reader: Reader) -> IO(Result<&1, &1, U32 & String, Step>):
1709	  match reader:
1710	    case Reader{file, state}:
1711	      next.state.block(file, state)
1712	# Reads one bounded block or validates the trailer and returns End.
1713	def next(reader: Reader) -> IO(Result<&1, &1, U32 & String, Step>):
1714	  next.state(reader)
1715	
1716	# Starts incremental entry consumption from an open block reader.
1717	def entry.open(reader: Reader) -> EntryCursor:
1718	  EntryReady{reader, Nil{}}
1719	
1720	def entry.ready(
1721	  reader: Reader,
1722	  entries: List<&1, ByteMemTable.Entry>
1723	) -> EntryStep:
1724	  match entries:
1725	    case Nil{}: EntrySkip{EntryReady{reader, Nil{}}}
1726	    case Con{entry, tail}:
1727	      EntryItem{entry, EntryReady{reader, tail}}
1728	
1729	def entry.result(
1730	  result: Result<&1, &1, U32 & String, Step>
1731	) -> IO(EntryStep):
1732	  match result:
1733	    case Fail{error}:
1734	      IO.pure(EntryStep, EntryError{EntryFailed{error}})
1735	    case Done{End{}}:
1736	      IO.pure(EntryStep, EntryEnd{})
1737	    case Done{Block{reader, entries}}:
1738	      IO.pure(EntryStep, entry.ready(reader, entries))
1739	
1740	# Returns one entry and reads another SST block only when the current block is empty.
1741	def entry.next(cursor: EntryCursor) -> IO(EntryStep):
1742	  match cursor:
1743	    case EntryReady{reader, entries}:
1744	      match entries:
1745	        case Con{entry, tail}:
1746	          IO.pure(EntryStep, EntryItem{entry, EntryReady{reader, tail}})
1747	        case Nil{}:
1748	          do IO<EntryStep>:
1749	            result : Result<&1, &1, U32 & String, Step> <- next(reader)
1750	            entry.result(result)
1751	    case EntryEnded{}: IO.pure(EntryStep, EntryEnd{})
1752	    case EntryClosed{}: IO.pure(EntryStep, EntryEnd{})
1753	    case EntryFailed{error}:
1754	      IO.pure(EntryStep, EntryError{EntryFailed{error}})
1755	
1756	
1757	# Closes the active SST reader and preserves terminal cursor states.
1758	def entry.close(cursor: EntryCursor) -> IO(Result<&1, &1, U32 & String, Unit>):
1759	  match cursor:
1760	    case EntryReady{reader, _}: close(reader)
1761	    case EntryEnded{}:
1762	      IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
1763	    case EntryClosed{}:
1764	      IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
1765	    case EntryFailed{_}:
1766	      IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})