1	import Base
2	import ./Wal.bend as Wal
3	import ./Keys.bend as Keys
4	import ./StorageBytes.bend as StorageBytes
5	import ./hub_sha/sha256.bend as SHA
6	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7	
8	# Owns one raw byte mutation without a text conversion.
9	type 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.
14	type Batch is Type:
15	  Batch{mutations: List<&1, Mutation>}
16	
17	# Carries a raw WAL record parse without decoding either byte buffer.
18	type 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.
23	type 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.
28	type RecordMeasure is Type:
29	  RecordMeasure{frame: Bytes.Bytes,
30	    result: Result<&1, &1, Wal.Error, RecordMetadata>}
31	
32	# Carries the state for ValueMetadata.
33	type ValueMetadata is Data:
34	  ValueAbsent{}
35	  ValueDeleted{}
36	  ValuePresent{value_bytes: U32}
37	
38	# Carries the state for MeasureProgress.
39	type 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.
44	type 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.
55	type 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.
62	type 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.
66	type ParsedBatch is Type:
67	  ParsedBatch{offset: U32, batch: Batch}
68	
69	def u32.clone.pair(+value: U32) -> U32 & U32:
70	  (value, value)
71	
72	def 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.
76	def 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.
80	def 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.
87	def 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.
95	def 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.
112	def 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.
122	def 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.
132	def 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.
138	def 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.
146	def header() -> Bytes.Bytes:
147	  header.finish(Bytes.from_hex("4d594c534d345700"))
148	
149	# Converts a raw WAL header comparison into a format decision.
150	def 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.
160	def 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.
164	def 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.
169	def 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.
173	def record.malformed(frame: Bytes.Bytes) -> DecodedRecord:
174	  RecordError{frame, Wal.Malformed{}}
175	
176	# Advances from a copied value to a completed put record.
177	def 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.
189	def 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.
199	def 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.
213	def 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.
224	def 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.
236	def 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.
246	def 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.
260	def 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.
272	def 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.
286	def 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.
296	def 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.
312	def 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.
324	def 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.
339	def 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.
352	def 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.
365	def 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.
380	def record.decode(frame: Bytes.Bytes, +offset: U32) -> DecodedRecord:
381	  record.tag(Bytes.get(frame, offset), offset)
382	
383	def record.measure.pack(
384	  frame: Bytes.Bytes,
385	  result: Result<&1, &1, Wal.Error, RecordMetadata>
386	) -> RecordMeasure:
387	  RecordMeasure{frame, result}
388	
389	def record.measure.result(measured: RecordMeasure)
390	  -> Result<&1, &1, Wal.Error, RecordMetadata>:
391	  match measured:
392	    case RecordMeasure{_, result}: result
393	
394	def 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	
408	def 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	
423	def 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	
432	def 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	
446	def 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	
456	def 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	
474	def 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	
489	def 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	
504	def 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	
513	def 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	
527	def 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	
547	def 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	
552	def record.measure(frame: Bytes.Bytes, +offset: U32)
553	  -> Result<&1, &1, Wal.Error, RecordMetadata>:
554	  record.measure.result(record.measure.retained(frame, offset))
555	
556	def 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	
563	def 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	
571	def 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	
577	def 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	
588	def 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	
596	def 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	
602	def 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	
617	def 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	
630	def measure.lookup.progress(
631	  query: Bytes.Bytes,
632	  records_end: U32
633	) -> MeasureProgress:
634	  MeasureProgress{query, records_end, False{}, False{}, 0}
635	
636	def 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	
654	def 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	
671	def 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	
680	def 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	
714	def 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	
725	def 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	
740	def 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	
746	def 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	
760	def 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	
774	def 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	
785	def 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	
791	def 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.
800	def 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.
822	def 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.
836	def 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.
849	def 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.
860	def 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.
871	def 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.
881	def 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.
892	def 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.
901	def 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.
914	def 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.
927	def 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.
939	def 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.
945	def 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	
950	def 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.
955	def 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.
965	def decode(frame: Bytes.Bytes) -> Result<&1, &1, Wal.Error, Batch>:
966	  decode.frame.header.result(decode.frame.length(Bytes.length(frame)))