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