src/ByteWal.bend

mylsm-lsm-store@0.6.0.1 · 966 lines · 96 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/Wal.bend as Wal
3import mylsm-lsm-store@0.6.0.1/src/Keys.bend as Keys
4import mylsm-lsm-store@0.6.0.1/src/StorageBytes.bend as StorageBytes
5import mylsm-lsm-store@0.6.0.1/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)))