src/ByteWal.bend
mylsm-lsm-store@0.6.0.0 · 966 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/Wal.bend as Wal
3import mylsm-lsm-store@0.6.0.0/src/Keys.bend as Keys
4import mylsm-lsm-store@0.6.0.0/src/StorageBytes.bend as StorageBytes
5import mylsm-lsm-store@0.6.0.0/src/hub_sha/sha256.bend as SHA
6import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7
8# Owns one raw byte mutation without a text conversion.
9type Mutation is Type:
10 Put{key: Bytes.Bytes, value: Bytes.Bytes}
11 Delete{key: Bytes.Bytes}
12
13# Owns the mutations that will be encoded into one WAL frame.
14type Batch is Type:
15 Batch{mutations: List<&1, Mutation>}
16
17# Carries a raw WAL record parse without decoding either byte buffer.
18type DecodedRecord is Type:
19 RecordDone{frame: Bytes.Bytes, offset: U32, mutation: Mutation}
20 RecordError{frame: Bytes.Bytes, error: Wal.Error}
21
22# Carries the state for RecordMetadata.
23type RecordMetadata is Type:
24 RecordMetadata{next_offset: U32, key: Bytes.Bytes,
25 has_value: Bool, value_bytes: U32}
26
27# Carries the state for RecordMeasure.
28type RecordMeasure is Type:
29 RecordMeasure{frame: Bytes.Bytes,
30 result: Result<&1, &1, Wal.Error, RecordMetadata>}
31
32# Carries the state for ValueMetadata.
33type ValueMetadata is Data:
34 ValueAbsent{}
35 ValueDeleted{}
36 ValuePresent{value_bytes: U32}
37
38# Carries the state for MeasureProgress.
39type MeasureProgress is Type:
40 MeasureProgress{query: Bytes.Bytes, records_end: U32,
41 found: Bool, has_value: Bool, value_bytes: U32}
42
43# Carries the state for MeasureState.
44type MeasureState is Type:
45 MeasureNext{frame: Bytes.Bytes, offset: U32, remaining: Nat,
46 progress: MeasureProgress}
47 MeasureRead{remaining: Nat, progress: MeasureProgress,
48 measured: RecordMeasure}
49 MeasureCompared{frame: Bytes.Bytes, next_offset: U32,
50 remaining: Nat, progress: MeasureProgress, has_value: Bool,
51 value_bytes: U32, order: Cmp}
52 MeasureFailed{error: Wal.Error}
53
54# Carries bounded frame parsing and newest-first decoded mutations.
55type DecodeState is Type:
56 DecodeNext{frame: Bytes.Bytes, offset: U32, remaining: Nat,
57 reversed: List<&1, Mutation>}
58 DecodeRead{record: DecodedRecord, remaining: Nat,
59 reversed: List<&1, Mutation>}
60
61# Keeps a parsed frame header with its exact source length.
62type FrameHeader is Type:
63 FrameHeader{frame: Bytes.Bytes, frame_length: U32, count: U32, total: U32}
64
65# Carries the final parser offset with its affine mutations.
66type ParsedBatch is Type:
67 ParsedBatch{offset: U32, batch: Batch}
68
69def u32.clone.pair(+value: U32) -> U32 & U32:
70 (value, value)
71
72def u32.clone.triple(+value: U32) -> U32 & U32 & U32:
73 (value, (value, value))
74
75# Encodes a raw key deletion using the shared bounded record layout.
76def record.delete(key: Bytes.Bytes) -> Result<&1, &1, Wal.Error, Bytes.Bytes>:
77 Wal.record.delete(Bytes.length(key))
78
79# Encodes a raw key/value pair using exact byte lengths.
80def record.put.value(
81 key: Bytes.Bytes,
82 value: Bytes.Bytes
83) -> Result<&1, &1, Wal.Error, Bytes.Bytes>:
84 Wal.record.put(Bytes.length(key), Bytes.length(value))
85
86# Converts one byte mutation into its binary WAL record.
87def record.encode(mutation: Mutation) -> Result<&1, &1, Wal.Error, Bytes.Bytes>:
88 match mutation:
89 case Put{key, value}:
90 record.put.value(key, value)
91 case Delete{key}:
92 record.delete(key)
93
94# Continues record encoding while preserving the first failure.
95def records.step(
96 rest: List<&1, Mutation>,
97 current: Result<&1, &1, Wal.Error, Bytes.Bytes>,
98 reversed: List<&1, Bytes.Bytes>,
99 count: U32
100) -> Result<&1, &1, Wal.Error, List<&1, Bytes.Bytes> & U32>:
101 match rest current:
102 case Nil{} Fail{error}:
103 Fail{error}
104 case Nil{} Done{bytes}:
105 Done{(List.reverse(&1, Bytes.Bytes, Con{bytes, reversed}), (count + 1 : U32))}
106 case Con{_, _} Fail{error}:
107 Fail{error}
108 case Con{mutation, tail} Done{bytes}:
109 records.step(tail, record.encode(mutation), Con{bytes, reversed}, (count + 1 : U32))
110
111# Encodes all mutations in their input order.
112def records(
113 mutations: List<&1, Mutation>
114) -> Result<&1, &1, Wal.Error, List<&1, Bytes.Bytes> & U32>:
115 match mutations:
116 case Nil{}:
117 Done{(Nil{}, 0)}
118 case Con{mutation, tail}:
119 records.step(tail, record.encode(mutation), Nil{}, 0)
120
121# Builds the checksummed frame from encoded records and their shared count.
122def encode.records(
123 result: Result<&1, &1, Wal.Error, List<&1, Bytes.Bytes> & U32>
124) -> Result<&1, &1, Wal.Error, Bytes.Bytes>:
125 match result:
126 case Fail{error}:
127 Fail{error}
128 case Done{(parts, count)}:
129 Wal.frame.payload(count, Done{parts})
130
131# Encodes a byte batch with the existing bounded frame checksum policy.
132def encode(batch: Batch) -> Result<&1, &1, Wal.Error, Bytes.Bytes>:
133 match batch:
134 case Batch{mutations}:
135 encode.records(records(mutations))
136
137# Returns the generation-four WAL file identity.
138def header.finish(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
139 match value:
140 case Some{bytes}:
141 bytes
142 case None{}:
143 Bytes.new(0)
144
145# Returns the generation-four WAL file identity.
146def header() -> Bytes.Bytes:
147 header.finish(Bytes.from_hex("4d594c534d345700"))
148
149# Converts a raw WAL header comparison into a format decision.
150def decode_header.comparison(
151 comparison: Bytes.Bytes & Bytes.Bytes & Bool
152) -> Result<&1, &1, Wal.Error, Unit>:
153 match comparison:
154 case (_, _, True{}):
155 Done{Unit{}}
156 case (_, _, False{}):
157 Fail{Wal.Malformed{}}
158
159# Accepts only the generation-four byte WAL signature.
160def decode_header(bytes: Bytes.Bytes) -> Result<&1, &1, Wal.Error, Unit>:
161 decode_header.comparison(Bytes.eq(header(), bytes))
162
163# Returns a bounded byte slice while preserving the source frame.
164def bytes.slice(frame: Bytes.Bytes, start: U32, length: U32) -> Bytes.Bytes & Bytes.Bytes:
165 Bytes.slice(frame, start, length)
166
167# Checks that a byte interval lies within its source frame.
168# Checks a byte interval against the source length.
169def bytes.interval.length(total: U32, start: U32, length: U32) -> Bool:
170 Nat.is_le(Nat.add(U32.to_nat(start), U32.to_nat(length)), U32.to_nat(total))
171
172# Returns a malformed-record result while retaining the source frame.
173def record.malformed(frame: Bytes.Bytes) -> DecodedRecord:
174 RecordError{frame, Wal.Malformed{}}
175
176# Advances from a copied value to a completed put record.
177def record.value.finished(
178 frame: Bytes.Bytes,
179 value: Bytes.Bytes,
180 value_offset: U32,
181 +value_length: U32
182) -> DecodedRecord:
183 match frame:
184 case Bytes.Bytes{len, buf}:
185 RecordDone{Bytes.Bytes{len, buf}, (value_offset + value_length : U32),
186 Put{Bytes.new(0), value}}
187
188# Finishes a value mutation after checking its complete record bound.
189def record.value.sliced(
190 pair: Bytes.Bytes & Bytes.Bytes,
191 value_offset: U32,
192 value_length: U32
193) -> DecodedRecord:
194 match pair:
195 case (frame, value):
196 record.value.finished(frame, value, value_offset, value_length)
197
198# Finishes a value mutation after checking its complete record bound.
199def record.value.checked(
200 frame: Bytes.Bytes,
201 +value_length: U32,
202 +value_offset: U32,
203 valid: Bool
204) -> DecodedRecord:
205 match valid:
206 case False{}:
207 record.malformed(frame)
208 case True{}:
209 record.value.sliced(bytes.slice(frame, value_offset, value_length),
210 value_offset, value_length)
211
212# Parses the value length and validates its exact byte range.
213def record.value.length.checked(
214 pair: Bytes.Bytes & U32,
215 +length: U32,
216 +value_offset: U32
217) -> DecodedRecord:
218 match pair:
219 case (frame, total):
220 record.value.checked(frame, length, value_offset,
221 bytes.interval.length(total, value_offset, length))
222
223# Parses the value length and validates its exact byte range.
224def record.value.length(
225 frame: Bytes.Bytes,
226 value_length: Maybe<&2, U32>,
227 +value_offset: U32
228) -> DecodedRecord:
229 match value_length:
230 case None{}:
231 record.malformed(frame)
232 case Some{+length}:
233 record.value.length.checked(Bytes.length(frame), length, value_offset)
234
235# Replaces the placeholder key in a completed put record.
236def record.put.key(record: DecodedRecord, key: Bytes.Bytes) -> DecodedRecord:
237 match record:
238 case RecordDone{frame, offset, Put{_, value}}:
239 RecordDone{frame, offset, Put{key, value}}
240 case RecordDone{frame, offset, Delete{old_key}}:
241 RecordDone{frame, offset, Delete{old_key}}
242 case RecordError{frame, error}:
243 RecordError{frame, error}
244
245# Parses the raw put value and restores its key in the record result.
246def record.put.value.checked(
247 frame: Bytes.Bytes,
248 key: Bytes.Bytes,
249 +value_length: U32,
250 value_offset: U32,
251 valid: Bool
252) -> DecodedRecord:
253 match valid:
254 case False{}:
255 record.malformed(frame)
256 case True{}:
257 record.put.key(record.value.checked(frame, value_length, value_offset, True{}), key)
258
259# Parses a put value length after the raw key bytes.
260def record.put.value.length.checked(
261 pair: Bytes.Bytes & U32,
262 key: Bytes.Bytes,
263 +length: U32,
264 +value_offset: U32
265) -> DecodedRecord:
266 match pair:
267 case (frame, total):
268 record.put.value.checked(frame, key, length, value_offset,
269 bytes.interval.length(total, value_offset, length))
270
271# Parses a put value length after the raw key bytes.
272def record.put.value.length(
273 frame: Bytes.Bytes,
274 key: Bytes.Bytes,
275 value_length: Maybe<&2, U32>,
276 value_length_offset: U32
277) -> DecodedRecord:
278 match value_length:
279 case None{}:
280 record.malformed(frame)
281 case Some{+length}:
282 record.put.value.length.checked(Bytes.length(frame), key, length,
283 (value_length_offset + 4 : U32))
284
285# Reads and parses a put's value length from its source frame.
286def record.put.value.length.read(
287 pair: Bytes.Bytes & Maybe<&2, U32>,
288 key: Bytes.Bytes,
289 value_length_offset: U32
290) -> DecodedRecord:
291 match pair:
292 case (frame, value_length):
293 record.put.value.length(frame, key, value_length, value_length_offset)
294
295# Finishes a raw delete record or reads the following put value length.
296def record.key.finished(
297 frame: Bytes.Bytes,
298 tag: U32,
299 key: Bytes.Bytes,
300 +value_length_offset: U32
301) -> DecodedRecord:
302 match tag:
303 case 1:
304 RecordDone{frame, (value_length_offset : U32), Delete{key}}
305 case 0:
306 record.put.value.length.read(
307 Bytes.get.u32be(frame, value_length_offset), key, value_length_offset)
308 case _:
309 record.malformed(frame)
310
311# Parses a raw key and selects delete or put record completion.
312def record.key.sliced(
313 pair: Bytes.Bytes & Bytes.Bytes,
314 tag: U32,
315 +key_length: U32,
316 +key_offset: U32
317) -> DecodedRecord:
318 match pair:
319 case (frame, key):
320 record.key.finished(frame, tag, key,
321 (key_offset + key_length : U32))
322
323# Parses a raw key and selects delete or put record completion.
324def record.key.checked(
325 frame: Bytes.Bytes,
326 tag: U32,
327 +key_length: U32,
328 +key_offset: U32,
329 valid: Bool
330) -> DecodedRecord:
331 match valid:
332 case False{}:
333 record.malformed(frame)
334 case True{}:
335 record.key.sliced(bytes.slice(frame, key_offset, key_length), tag,
336 key_length, key_offset)
337
338# Parses a record key length and verifies its bounded range.
339def record.key.length.checked(
340 pair: Bytes.Bytes & U32,
341 tag: U32,
342 +length: U32,
343 +key_offset: U32
344) -> DecodedRecord:
345 match pair:
346 case (frame, total):
347 record.key.checked(frame, tag, length, key_offset,
348 U32.is_le(length, StorageBytes.MAX_RECORD_BYTES()) &&
349 bytes.interval.length(total, key_offset, length))
350
351# Parses a record key length and verifies its bounded range.
352def record.key.length(
353 pair: Bytes.Bytes & Maybe<&2, U32>,
354 tag: U32,
355 offset: U32
356) -> DecodedRecord:
357 match pair:
358 case (frame, None{}):
359 record.malformed(frame)
360 case (frame, Some{+length}):
361 record.key.length.checked(Bytes.length(frame), tag, length,
362 (offset + 5 : U32))
363
364# Validates the tag and reads the corresponding raw key length.
365def record.tag(
366 pair: Bytes.Bytes & Maybe<&2, U32>,
367 +offset: U32
368) -> DecodedRecord:
369 match pair:
370 case (frame, None{}):
371 record.malformed(frame)
372 case (frame, Some{0}):
373 record.key.length(Bytes.get.u32be(frame, (offset + 1 : U32)), 0, offset)
374 case (frame, Some{1}):
375 record.key.length(Bytes.get.u32be(frame, (offset + 1 : U32)), 1, offset)
376 case (frame, Some{_}):
377 record.malformed(frame)
378
379# Decodes one raw WAL record from its length-delimited frame position.
380def record.decode(frame: Bytes.Bytes, +offset: U32) -> DecodedRecord:
381 record.tag(Bytes.get(frame, offset), offset)
382
383def record.measure.pack(
384 frame: Bytes.Bytes,
385 result: Result<&1, &1, Wal.Error, RecordMetadata>
386) -> RecordMeasure:
387 RecordMeasure{frame, result}
388
389def record.measure.result(measured: RecordMeasure)
390 -> Result<&1, &1, Wal.Error, RecordMetadata>:
391 match measured:
392 case RecordMeasure{_, result}: result
393
394def record.measure.value.result(
395 frame: Bytes.Bytes,
396 key: Bytes.Bytes,
397 +value_length: U32,
398 value_offset: U32,
399 valid: Bool
400) -> RecordMeasure:
401 match valid:
402 case True{}:
403 value_end : U32 = U32.add(value_offset, 4)
404 record.measure.pack(frame, Done{RecordMetadata{
405 U32.add(value_end, value_length), key, True{}, value_length}})
406 case False{}: record.measure.pack(frame, Fail{Wal.Malformed{}})
407
408def record.measure.value.checked(
409 measured: Bytes.Bytes & U32,
410 key: Bytes.Bytes,
411 +value_length: U32,
412 offsets: U32 & U32
413) -> RecordMeasure:
414 match measured:
415 case (frame, total):
416 match offsets:
417 case (checked_offset, result_offset):
418 record.measure.value.result(frame, key, value_length, result_offset,
419 U32.is_le(value_length, StorageBytes.MAX_RECORD_BYTES()) &&
420 bytes.interval.length(total, U32.add(checked_offset, 4),
421 value_length))
422
423def record.measure.value(
424 measured: Bytes.Bytes & U32,
425 key: Bytes.Bytes,
426 +value_length: U32,
427 value_offset: U32
428) -> RecordMeasure:
429 record.measure.value.checked(measured, key, value_length,
430 u32.clone.pair(value_offset))
431
432def record.measure.value.length(
433 pair: Bytes.Bytes & Maybe<&2, U32>,
434 key: Bytes.Bytes,
435 value_length_offset: U32
436) -> RecordMeasure:
437 match pair:
438 case (frame, length):
439 match length:
440 case Some{value_length}:
441 record.measure.value(Bytes.length(frame), key, value_length,
442 value_length_offset)
443 case None{}:
444 record.measure.pack(frame, Fail{Wal.Malformed{}})
445
446def record.measure.value.length.read(
447 frame: Bytes.Bytes,
448 key: Bytes.Bytes,
449 offsets: U32 & U32
450) -> RecordMeasure:
451 match offsets:
452 case (read_offset, value_length_offset):
453 record.measure.value.length(
454 Bytes.get.u32be(frame, read_offset), key, value_length_offset)
455
456def record.measure.key.finished(
457 pair: Bytes.Bytes & Bytes.Bytes,
458 tag: U32,
459 +_key_length: U32,
460 value_offset: U32
461) -> RecordMeasure:
462 match pair:
463 case (frame, key):
464 match tag:
465 case 0:
466 record.measure.value.length.read(frame, key,
467 u32.clone.pair(value_offset))
468 case 1:
469 record.measure.pack(frame,
470 Done{RecordMetadata{value_offset, key, False{}, 0}})
471 case _:
472 record.measure.pack(frame, Fail{Wal.Malformed{}})
473
474def record.measure.key.result(
475 frame: Bytes.Bytes,
476 tag: U32,
477 +key_length: U32,
478 valid: Bool,
479 offsets: U32 & U32
480) -> RecordMeasure:
481 match valid:
482 case True{}:
483 match offsets:
484 case (slice_offset, key_offset):
485 record.measure.key.finished(Bytes.slice(frame, slice_offset,
486 key_length), tag, key_length, key_offset)
487 case False{}: record.measure.pack(frame, Fail{Wal.Malformed{}})
488
489def record.measure.key.checked(
490 measured: Bytes.Bytes & U32,
491 tag: U32,
492 +key_length: U32,
493 offsets: U32 & U32 & U32
494) -> RecordMeasure:
495 match measured:
496 case (frame, total):
497 match offsets:
498 case (checked_offset, (slice_offset, value_offset)):
499 record.measure.key.result(frame, tag, key_length,
500 U32.is_le(key_length, StorageBytes.MAX_RECORD_BYTES()) &&
501 bytes.interval.length(total, checked_offset, key_length),
502 (slice_offset, U32.add(value_offset, key_length)))
503
504def record.measure.key.length(
505 measured: Bytes.Bytes & U32,
506 tag: U32,
507 +key_length: U32,
508 key_offset: U32
509) -> RecordMeasure:
510 record.measure.key.checked(measured, tag, key_length,
511 u32.clone.triple(key_offset))
512
513def record.measure.key.length.read(
514 pair: Bytes.Bytes & Maybe<&2, U32>,
515 tag: U32,
516 key_offset: U32
517) -> RecordMeasure:
518 match pair:
519 case (frame, length):
520 match length:
521 case Some{key_length}:
522 record.measure.key.length(Bytes.length(frame), tag, key_length,
523 key_offset)
524 case None{}:
525 record.measure.pack(frame, Fail{Wal.Malformed{}})
526
527def record.measure.tag(
528 pair: Bytes.Bytes & Maybe<&2, U32>,
529 offsets: U32 & U32
530) -> RecordMeasure:
531 match pair:
532 case (frame, tag):
533 match tag:
534 case Some{0}:
535 match offsets:
536 case (length_offset, key_offset):
537 record.measure.key.length.read(
538 Bytes.get.u32be(frame, length_offset), 0, key_offset)
539 case Some{1}:
540 match offsets:
541 case (length_offset, key_offset):
542 record.measure.key.length.read(
543 Bytes.get.u32be(frame, length_offset), 1, key_offset)
544 case _:
545 record.measure.pack(frame, Fail{Wal.Malformed{}})
546
547def record.measure.retained(frame: Bytes.Bytes, +offset: U32)
548 -> RecordMeasure:
549 record.measure.tag(Bytes.get(frame, offset),
550 (U32.add(offset, 1), U32.add(offset, 5)))
551
552def record.measure(frame: Bytes.Bytes, +offset: U32)
553 -> Result<&1, &1, Wal.Error, RecordMetadata>:
554 record.measure.result(record.measure.retained(frame, offset))
555
556def measure.checksum.comparison(
557 frame: Bytes.Bytes,
558 comparison: Bytes.Bytes & Bytes.Bytes & Bool
559) -> Bytes.Bytes & Bool:
560 match comparison:
561 case (_, _, valid): (frame, valid)
562
563def measure.checksum.claimed(
564 pair: Bytes.Bytes & Bytes.Bytes,
565 digest: Bytes.Bytes
566) -> Bytes.Bytes & Bool:
567 match pair:
568 case (frame, claimed):
569 measure.checksum.comparison(frame, Bytes.eq(claimed, digest))
570
571def measure.checksum.range.claimed(
572 pair: Bytes.Bytes & Bytes.Bytes,
573 digest: Bytes.Bytes
574) -> Bytes.Bytes & Bool:
575 measure.checksum.claimed(pair, digest)
576
577def measure.checksum.range.prefix(
578 pair: Bytes.Bytes & Bytes.Bytes,
579 +base: U32,
580 +total: U32
581) -> Bytes.Bytes & Bool:
582 match pair:
583 case (frame, prefix):
584 measure.checksum.range.claimed(
585 Bytes.slice(frame, U32.add(base, U32.sub(total, 32)), 32),
586 SHA.sha256_packed_bytes(prefix))
587
588def measure.checksum.range(
589 frame: Bytes.Bytes,
590 +base: U32,
591 +total: U32
592) -> Bytes.Bytes & Bool:
593 measure.checksum.range.prefix(
594 Bytes.slice(frame, base, U32.sub(total, 32)), base, total)
595
596def measure.lookup.finish.absent(exact: Bool)
597 -> Result<&1, &1, Wal.Error, ValueMetadata>:
598 match exact:
599 case False{}: Fail{Wal.Malformed{}}
600 case True{}: Done{ValueAbsent{}}
601
602def measure.lookup.finish.present(
603 has_value: Bool,
604 +value_bytes: U32,
605 exact: Bool
606) -> Result<&1, &1, Wal.Error, ValueMetadata>:
607 match has_value:
608 case True{}:
609 match exact:
610 case False{}: Fail{Wal.Malformed{}}
611 case True{}: Done{ValuePresent{value_bytes}}
612 case False{}:
613 match exact:
614 case False{}: Fail{Wal.Malformed{}}
615 case True{}: Done{ValueDeleted{}}
616
617def measure.lookup.finish(
618 progress: MeasureProgress,
619 offset: U32
620) -> Result<&1, &1, Wal.Error, ValueMetadata>:
621 match progress:
622 case MeasureProgress{_, records_end, found, has_value, value_bytes}:
623 match found:
624 case False{}: measure.lookup.finish.absent(
625 U32.is_eq(offset, records_end))
626 case True{}:
627 measure.lookup.finish.present(has_value, value_bytes,
628 U32.is_eq(offset, records_end))
629
630def measure.lookup.progress(
631 query: Bytes.Bytes,
632 records_end: U32
633) -> MeasureProgress:
634 MeasureProgress{query, records_end, False{}, False{}, 0}
635
636def measure.lookup.compared(
637 frame: Bytes.Bytes,
638 next_offset: U32,
639 remaining: Nat,
640 records_end: U32,
641 found: Bool,
642 prior_has_value: Bool,
643 prior_value_bytes: U32,
644 record_has_value: Bool,
645 record_value_bytes: U32,
646 comparison: Bytes.Bytes & Bytes.Bytes & Cmp
647) -> MeasureState:
648 match comparison:
649 case (_, query, order):
650 MeasureCompared{frame, next_offset, remaining,
651 MeasureProgress{query, records_end, found, prior_has_value,
652 prior_value_bytes}, record_has_value, record_value_bytes, order}
653
654def measure.lookup.read.result(
655 frame: Bytes.Bytes,
656 remaining: Nat,
657 progress: MeasureProgress,
658 result: Result<&1, &1, Wal.Error, RecordMetadata>
659) -> MeasureState:
660 match progress:
661 case MeasureProgress{query, records_end, found, prior_has_value,
662 prior_value_bytes}:
663 match result:
664 case Fail{error}: MeasureFailed{error}
665 case Done{RecordMetadata{next_offset, key, has_value, value_bytes}}:
666 measure.lookup.compared(frame, next_offset, remaining,
667 records_end, found, prior_has_value, prior_value_bytes,
668 has_value, value_bytes,
669 Keys.Bytes.cmp.with_inputs(key, query))
670
671def measure.lookup.read(
672 remaining: Nat,
673 progress: MeasureProgress,
674 measured: RecordMeasure
675) -> MeasureState:
676 match measured:
677 case RecordMeasure{frame, result}:
678 measure.lookup.read.result(frame, remaining, progress, result)
679
680def measure.lookup.loop(fuel: Nat, state: MeasureState)
681 -> Result<&1, &1, Wal.Error, ValueMetadata>:
682 match fuel:
683 case 0n: Fail{Wal.Malformed{}}
684 case 1n+rest:
685 match state:
686 case MeasureNext{frame, offset, remaining, progress}:
687 match remaining:
688 case 0n: measure.lookup.finish(progress, offset)
689 case 1n+more:
690 measure.lookup.loop(rest,
691 MeasureRead{more, progress,
692 record.measure.retained(frame, offset)})
693 case MeasureRead{remaining, progress, measured}:
694 measure.lookup.loop(rest,
695 measure.lookup.read(remaining, progress, measured))
696 case MeasureCompared{frame, next_offset, remaining, progress,
697 record_has_value, record_value_bytes, order}:
698 match progress:
699 case MeasureProgress{query, records_end, found, has_value,
700 value_bytes}:
701 match order:
702 case EQ{}:
703 measure.lookup.loop(rest,
704 MeasureNext{frame, next_offset, remaining,
705 MeasureProgress{query, records_end, True{},
706 record_has_value, record_value_bytes}})
707 case _:
708 measure.lookup.loop(rest,
709 MeasureNext{frame, next_offset, remaining,
710 MeasureProgress{query, records_end, found, has_value,
711 value_bytes}})
712 case MeasureFailed{error}: Fail{error}
713
714def measure.lookup.records.bound(
715 frame: Bytes.Bytes,
716 query: Bytes.Bytes,
717 +count: U32,
718 +offset: U32,
719 +records_end: U32
720) -> Result<&1, &1, Wal.Error, ValueMetadata>:
721 measure.lookup.loop(Nat.add(Nat.mul(3n, U32.to_nat(count)), 1n),
722 MeasureNext{frame, offset, U32.to_nat(count),
723 measure.lookup.progress(query, records_end)})
724
725def measure.lookup.checksum.result(
726 pair: Bytes.Bytes & Bool,
727 query: Bytes.Bytes,
728 +base: U32,
729 +count: U32,
730 +total: U32
731) -> Result<&1, &1, Wal.Error, ValueMetadata>:
732 match pair:
733 case (frame, valid):
734 match valid:
735 case True{}:
736 measure.lookup.records.bound(frame, query, count,
737 U32.add(base, 8), U32.add(base, U32.sub(total, 32)))
738 case False{}: Fail{Wal.ChecksumMismatch{}}
739
740def frame.bounds.valid(+frame_length: U32, +count: U32, +total: U32) -> Bool:
741 U32.is_le(36, frame_length) &&
742 Nat.is_eq(U32.to_nat(total), Nat.add(U32.to_nat(frame_length), 4n)) &&
743 Nat.is_le(U32.to_nat(count), Nat.div(U32.to_nat(frame_length), 5n)) &&
744 Wal.frame_length_allowed(count, frame_length)
745
746def measure.lookup.bounded.valid(
747 frame: Bytes.Bytes,
748 query: Bytes.Bytes,
749 +base: U32,
750 +count: U32,
751 +total: U32,
752 valid: Bool
753) -> Result<&1, &1, Wal.Error, ValueMetadata>:
754 match valid:
755 case False{}: Fail{Wal.Malformed{}}
756 case True{}:
757 measure.lookup.checksum.result(
758 measure.checksum.range(frame, base, total), query, base, count, total)
759
760def measure.lookup.bounded.measured(
761 measured: Bytes.Bytes & U32,
762 query: Bytes.Bytes,
763 +base: U32,
764 +frame_length: U32,
765 +count: U32,
766 +total: U32
767) -> Result<&1, &1, Wal.Error, ValueMetadata>:
768 match measured:
769 case (frame, buffer_length):
770 measure.lookup.bounded.valid(frame, query, base, count, total,
771 frame.bounds.valid(frame_length, count, total) &&
772 bytes.interval.length(buffer_length, base, total))
773
774def measure.lookup.bounded(
775 frame: Bytes.Bytes,
776 query: Bytes.Bytes,
777 +base: U32,
778 +frame_length: U32,
779 +count: U32,
780 +total: U32
781) -> Result<&1, &1, Wal.Error, ValueMetadata>:
782 measure.lookup.bounded.measured(Bytes.length(frame), query, base,
783 frame_length, count, total)
784
785def measure.lookup.header(header: FrameHeader, query: Bytes.Bytes)
786 -> Result<&1, &1, Wal.Error, ValueMetadata>:
787 match header:
788 case FrameHeader{frame, +frame_length, +count, +total}:
789 measure.lookup.bounded(frame, query, 0, frame_length, count, total)
790
791def measure.lookup.header.result(
792 result: Result<&1, &1, Wal.Error, FrameHeader>,
793 query: Bytes.Bytes
794) -> Result<&1, &1, Wal.Error, ValueMetadata>:
795 match result:
796 case Fail{error}: Fail{error}
797 case Done{header}: measure.lookup.header(header, query)
798
799# Decodes the exact number of records declared by a WAL frame.
800def decode.records.loop(fuel: Nat, state: DecodeState) -> Result<&1, &1, Wal.Error, ParsedBatch>:
801 match fuel:
802 case 0n:
803 Fail{Wal.Malformed{}}
804 case 1n+rest:
805 match state:
806 case DecodeNext{frame, offset, remaining, reversed}:
807 match remaining:
808 case 0n:
809 Done{ParsedBatch{offset, Batch{List.reverse(&1, Mutation, reversed)}}}
810 case 1n+more:
811 decode.records.loop(rest,
812 DecodeRead{record.decode(frame, offset), more, reversed})
813 case DecodeRead{record, remaining, reversed}:
814 match record:
815 case RecordError{_, error}:
816 Fail{error}
817 case RecordDone{frame, offset, mutation}:
818 decode.records.loop(rest,
819 DecodeNext{frame, offset, remaining, Con{mutation, reversed}})
820
821# Converts the pure frame bound decision into a typed outcome.
822def decode.header.validated(
823 frame: Bytes.Bytes,
824 frame_length: U32,
825 count: U32,
826 total: U32,
827 valid: Bool
828) -> Result<&1, &1, Wal.Error, FrameHeader>:
829 match valid:
830 case False{}:
831 Fail{Wal.Malformed{}}
832 case True{}:
833 Done{FrameHeader{frame, frame_length, count, total}}
834
835# Validates the declared count and full frame length.
836def decode.header.count(
837 pair: Bytes.Bytes & Maybe<&2, U32>,
838 +frame_length: U32,
839 +total: U32
840) -> Result<&1, &1, Wal.Error, FrameHeader>:
841 match pair:
842 case (_, None{}):
843 Fail{Wal.Malformed{}}
844 case (frame, Some{+count}):
845 decode.header.validated(frame, frame_length, count, total,
846 frame.bounds.valid(frame_length, count, total))
847
848# Reads frame length and count into a validated header.
849def decode.header.length(
850 pair: Bytes.Bytes & Maybe<&2, U32>,
851 total: U32
852) -> Result<&1, &1, Wal.Error, FrameHeader>:
853 match pair:
854 case (frame, None{}):
855 Fail{Wal.Malformed{}}
856 case (frame, Some{+length}):
857 decode.header.count(Bytes.get.u32be(frame, 4), length, total)
858
859# Returns the decoded batch only when the checksum agrees.
860def decode.checksum.comparison(
861 comparison: Bytes.Bytes & Bytes.Bytes & Bool,
862 batch: Batch
863) -> Result<&1, &1, Wal.Error, Batch>:
864 match comparison:
865 case (_, _, True{}):
866 Done{batch}
867 case (_, _, False{}):
868 Fail{Wal.ChecksumMismatch{}}
869
870# Compares the stored digest with the computed digest.
871def decode.checksum.claimed(
872 pair: Bytes.Bytes & Bytes.Bytes,
873 batch: Batch,
874 digest: Bytes.Bytes
875) -> Result<&1, &1, Wal.Error, Batch>:
876 match pair:
877 case (_, claimed):
878 decode.checksum.comparison(Bytes.eq(claimed, digest), batch)
879
880# Parses the retained digest after hashing its exact frame prefix.
881def decode.checksum.prefix(
882 pair: Bytes.Bytes & Bytes.Bytes,
883 batch: Batch,
884 digest_offset: U32
885) -> Result<&1, &1, Wal.Error, Batch>:
886 match pair:
887 case (frame, prefix):
888 decode.checksum.claimed(bytes.slice(frame, digest_offset, 32), batch,
889 SHA.sha256_packed_bytes(prefix))
890
891# Compares a frame's stored digest with the digest of its exact prefix.
892def decode.checksum.bytes(
893 frame: Bytes.Bytes,
894 batch: Batch,
895 +digest_offset: U32
896) -> Result<&1, &1, Wal.Error, Batch>:
897 decode.checksum.prefix(bytes.slice(frame, 0, digest_offset), batch,
898 digest_offset)
899
900# Converts the record-end decision into a typed frame result.
901def decode.checksum.offset(
902 frame: Bytes.Bytes,
903 batch: Batch,
904 +total: U32,
905 valid: Bool
906) -> Result<&1, &1, Wal.Error, Batch>:
907 match valid:
908 case False{}:
909 Fail{Wal.Malformed{}}
910 case True{}:
911 decode.checksum.bytes(frame, batch, (total - 32 : U32))
912
913# Verifies the checksum and exact record payload boundary.
914def decode.checksum(
915 frame: Bytes.Bytes,
916 +total: U32,
917 result: Result<&1, &1, Wal.Error, ParsedBatch>
918) -> Result<&1, &1, Wal.Error, Batch>:
919 match result:
920 case Fail{error}:
921 Fail{error}
922 case Done{ParsedBatch{+offset, batch}}:
923 decode.checksum.offset(frame, batch, total,
924 U32.is_eq(offset, (total - 32 : U32)))
925
926# Validates record payload and checksum after parsing a bounded frame header.
927def decode.frame.copies(
928 pair: Bytes.Bytes & Bytes.Bytes,
929 +count: U32,
930 +total: U32
931) -> Result<&1, &1, Wal.Error, Batch>:
932 match pair:
933 case (checksum_frame, parse_frame):
934 decode.checksum(checksum_frame, total,
935 decode.records.loop(Nat.add(Nat.mul(2n, U32.to_nat(count)), 1n),
936 DecodeNext{parse_frame, 8, U32.to_nat(count), Nil{}}))
937
938# Validates record payload and checksum after parsing a bounded frame header.
939def decode.frame.header(header: FrameHeader) -> Result<&1, &1, Wal.Error, Batch>:
940 match header:
941 case FrameHeader{frame, _, +count, +total}:
942 decode.frame.copies(Wal.bytes.clone.split(frame), count, total)
943
944# Validates a packed frame length before reading its count and records.
945def decode.frame.length(pair: Bytes.Bytes & U32) -> Result<&1, &1, Wal.Error, FrameHeader>:
946 match pair:
947 case (frame, +total):
948 decode.header.length(Bytes.get.u32be(frame, 0), total)
949
950def measure.lookup(frame: Bytes.Bytes, query: Bytes.Bytes)
951 -> Result<&1, &1, Wal.Error, ValueMetadata>:
952 measure.lookup.header.result(decode.frame.length(Bytes.length(frame)), query)
953
954# Dispatches a parsed header to the full bounded frame decoder.
955def decode.frame.header.result(
956 result: Result<&1, &1, Wal.Error, FrameHeader>
957) -> Result<&1, &1, Wal.Error, Batch>:
958 match result:
959 case Fail{error}:
960 Fail{error}
961 case Done{header}:
962 decode.frame.header(header)
963
964# Decodes a generation-four WAL frame, rejecting malformed lengths and checksums.
965def decode(frame: Bytes.Bytes) -> Result<&1, &1, Wal.Error, Batch>:
966 decode.frame.header.result(decode.frame.length(Bytes.length(frame)))