src/scan/ByteBlockReader.bend
mylsm-lsm-store@0.6.0.1 · 1,766 lines · 144 definitions
1import Base
2import mylsm-lsm-store@0.6.0.1/src/ByteSstFile.bend as ByteSstFile
3import mylsm-lsm-store@0.6.0.1/src/ByteWal.bend as ByteWal
4import mylsm-lsm-store@0.6.0.1/src/Wal.bend as Wal
5import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
6import mylsm-lsm-store@0.6.0.1/src/StorageBytes.bend as StorageBytes
7import mylsm-lsm-store@0.6.0.1/src/ByteRecoveryPolicy.bend as RecoveryPolicy
8import mylsm-lsm-store@0.6.0.1/src/Fs.bend as Fs
9import mylsm-lsm-store@0.6.0.1/src/SstStreamIo.bend as SstStreamIo
10import mylsm-lsm-store@0.6.0.1/src/coordinator/QueuePolicy.bend as QueuePolicy
11import mylsm-lsm-store@0.6.0.1/src/scan/BlockOrder.bend as BlockOrder
12import mylsm-lsm-store@0.6.0.1/src/scan/BlockIndexPolicy.bend as BlockIndexPolicy
13import mylsm-lsm-store@0.6.0.1/src/scan/BlockIndexFormatPolicy.bend as BlockIndexFormatPolicy
14import mylsm-lsm-store@0.6.0.1/src/scan/BlockRoutePolicy.bend as BlockRoutePolicy
15import mylsm-lsm-store@0.6.0.1/src/hub_sha/sha256.bend as SHA
16import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
17
18
19type 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
26type Reader is Type:
27 Reader{file: File, state: State}
28
29
30type Metadata is Data:
31 Metadata{size: Nat, total: U32, block_count: U32, index_offset: Nat,
32 live_entries: Nat, payload_bytes: Nat}
33
34
35type Step is Type:
36 Block{reader: Reader, entries: List<&1, ByteMemTable.Entry>}
37 End{}
38
39
40type EntryCursor is Type:
41 EntryReady{reader: Reader, entries: List<&1, ByteMemTable.Entry>}
42 EntryEnded{}
43 EntryClosed{}
44 EntryFailed{error: U32 & String}
45
46
47type EntryStep is Type:
48 EntryItem{entry: ByteMemTable.Entry, cursor: EntryCursor}
49 EntrySkip{cursor: EntryCursor}
50 EntryEnd{}
51 EntryError{cursor: EntryCursor}
52
53
54
55type BodyContext is Type:
56 BodyContext{state: State, block_header: Bytes.Bytes, payload_length: U32}
57
58
59type 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
68type IndexPageData is Type:
69 IndexPageData{reader: IndexMeta, offset: Nat, remaining: Nat,
70 actual_hash: SHA.Stream, root_hash: SHA.Stream}
71
72
73type IndexMeta is Type:
74 IndexMeta{total: U32, index_length: U32, expected_hash: SHA.Stream}
75
76
77type RouteContext is Data:
78 RouteContext{size: Nat, index_offset: Nat, block_count: Nat}
79
80
81type 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
99def 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
112def transition_budget(reader: Reader) -> Nat & Reader:
113 match reader:
114 case Reader{file, state}: transition.budget.state(file, state)
115
116
117def 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
129def bytes.slice.result(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
130 match pair:
131 case (_, bytes): bytes
132
133def block.checksum.compare(
134 comparison: Bytes.Bytes & Bytes.Bytes & Bool
135) -> Bool:
136 match comparison:
137 case (_, _, valid): valid
138
139def 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
150def 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
156def 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
166def 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
177def 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
186def 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
194def corruption(message: String) -> U32 & String:
195 (4294967292, message)
196
197
198def resource_limit_error_code() -> U32:
199 RecoveryPolicy.resource_limit_error_code()
200
201
202def resource_limit(message: String) -> U32 & String:
203 (resource_limit_error_code(), message)
204
205def 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
214def 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
222def 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
230def 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
248def 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
269def 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
283def 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
294def 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
309def 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
332def 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
347def 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
365def 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
381def 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
396def 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
408def 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
419def 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
427def 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
433def 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
446def 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
461def 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
471def 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
476def 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
481def 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
502def 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
515def 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
530def 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
540def seek.block.word(pair: Bytes.Bytes & Maybe<&2, U32>) -> Maybe<&2, U32>:
541 match pair:
542 case (_, value): value
543
544def 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
552def seek.block.words(bytes: Bytes.Bytes) -> Maybe<&2, U32> & Maybe<&2, U32>:
553 seek.block.words.copies(ByteSstFile.clone.split(bytes))
554
555def 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
565def 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
575def 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
583def 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
586def 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
590def 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
593def 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
609def 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
633def 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
648def 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
669def 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
686def 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
702def 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
714def 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
723def 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
730def 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
738def 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
757def 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
782def 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
796def 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
810def 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
823def 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
838def 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
861def route.midpoint(search: BlockRoutePolicy.Search) -> Nat:
862 match search:
863 case BlockRoutePolicy.Search{low, high, _}:
864 BlockRoutePolicy.midpoint(low, high)
865
866def 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
889def 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
922def 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
941def 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
956def 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
965def 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
978def 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
991def 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
1001def 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
1017def 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
1034def 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
1050def 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
1059def 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
1073def 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
1083def 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
1093def 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
1102def 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
1108def 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
1131def 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
1156def 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
1178def 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
1213def 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
1240def 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
1259def 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
1272def 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
1291def 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
1304def 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
1315def 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
1328def 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
1339def 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
1359def 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
1374def 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)
1385def 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
1394def 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))
1405def 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
1413def 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
1420def 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
1431def 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
1445def 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
1458def 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
1468def 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
1477def 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
1489def 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
1504def 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
1515def 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
1528def 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
1539def 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
1554def 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
1569def 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
1586def 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
1598def 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
1643def 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
1656def next.trailer.read(
1657 file: File,
1658 state: State
1659) -> IO(Result<&1, &1, U32 & String, Step>):
1660 index.page.start(file, state)
1661
1662def 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
1673def 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
1678def 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
1688def 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
1699def 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
1708def 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
1713def next(reader: Reader) -> IO(Result<&1, &1, U32 & String, Step>):
1714 next.state(reader)
1715
1716
1717def entry.open(reader: Reader) -> EntryCursor:
1718 EntryReady{reader, Nil{}}
1719
1720def 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
1729def 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
1741def 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
1758def 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{}})