1	import Base
2	import ../ByteMemTable.bend as ByteMemTable
3	import ../ByteWal.bend as ByteWal
4	import ./RestorePolicy.bend as RestorePolicy
5	import ../StorageBytes.bend as StorageBytes
6	import ../Wal.bend as Wal
7	import ../coordinator/QueuePolicy.bend as QueuePolicy
8	import ../hub_sha/sha256.bend as SHA
9	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
10	
11	# Distinguishes malformed exchange data from integrity and resource failures.
12	type Error is Data:
13	  InvalidFormat{}
14	  ResourceLimit{}
15	  InvalidOrder{}
16	  ChecksumMismatch{}
17	  CountMismatch{}
18	
19	# Carries the verified logical records produced by restore validation.
20	type Decoded is Type:
21	  Decoded{entries: List<&1, ByteMemTable.Entry>}
22	
23	# Carries the fixed header metadata shared by all restore paths.
24	type Header is Type:
25	  Header{input: Bytes.Bytes, entry_count: U32, block_count: U32}
26	
27	# Tracks block traversal and the reverse-ordered decoded entries.
28	type State is Type:
29	  State{input: Bytes.Bytes & Bytes.Bytes, offset: Nat, entry_count: U32,
30	    block_count: U32, remaining_blocks: Nat, block_index: Nat,
31	    decoded_count: Nat, reversed: List<&1, ByteMemTable.Entry>}
32	
33	# Retains the source slice and decoded entries for one validated block.
34	type BlockMeta is Type:
35	  BlockMeta{input: Bytes.Bytes, offset: Nat, frame_offset: Nat,
36	    frame_length: Nat, record_count: Nat, block_length: Nat,
37	    entries: List<&1, ByteMemTable.Entry>}
38	
39	# Holds validated block-header fields while checksum and frame decoding run.
40	type BlockHeader is Type:
41	  BlockHeader{input: Bytes.Bytes, offset: Nat, index: U32,
42	    record_count: U32, frame_length: U32, block_length: Nat}
43	
44	# Pairs decoded entries with their observed count.
45	type EntryBatch is Type:
46	  EntryBatch{entries: List<&1, ByteMemTable.Entry>, count: Nat}
47	
48	# Projects byte equality from the packed-byte comparison result.
49	def equal(pair: Bytes.Bytes & Bytes.Bytes & Bool) -> Bool:
50	  match pair:
51	    case (_, _, same): same
52	
53	def slice.result(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
54	  match pair:
55	    case (_, bytes): bytes
56	
57	def u32.field.is(result: Maybe<&2, U32>, expected: U32) -> Bool:
58	  match result:
59	    case None{}: False{}
60	    case Some{field}: U32.is_eq(field, expected)
61	
62	def clone.pair(+len: U32, copied: Array<U32> & Array<U32>) -> Bytes.Bytes & Bytes.Bytes:
63	  match copied:
64	    case (left, right):
65	      (Bytes.Bytes{len, left}, Bytes.Bytes{len, right})
66	
67	def clone.split(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes:
68	  match bytes:
69	    case Bytes.Bytes{+len, buf}:
70	      clone.pair(len, Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
71	
72	def clone.triple.left(
73	  second: Bytes.Bytes,
74	  copies: Bytes.Bytes & Bytes.Bytes
75	) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
76	  match copies:
77	    case (left, right): (left, right, second)
78	
79	def clone.triple.step(copies: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
80	  match copies:
81	    case (first, second):
82	      clone.triple.left(second, clone.split(first))
83	
84	def clone.triple(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
85	  clone.triple.step(clone.split(bytes))
86	
87	def bytes.size(pair: Bytes.Bytes & U32) -> U32:
88	  match pair:
89	    case (_, size): size
90	
91	# Unwraps a fixed exchange-format marker.
92	def magic(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
93	  match value:
94	    case Some{bytes}: bytes
95	    case None{}: Bytes.new(0)
96	
97	def magic.header() -> Bytes.Bytes:
98	  magic(Bytes.from_hex("4d594c534d424b31"))
99	
100	def footer.magic() -> Bytes.Bytes:
101	  magic(Bytes.from_hex("454e4431"))
102	
103	def block.magic() -> Bytes.Bytes:
104	  magic(Bytes.from_hex("424c4b31"))
105	
106	def digest.matches(
107	  claimed: Bytes.Bytes,
108	  input: Bytes.Bytes
109	) -> Bool:
110	  equal(Bytes.eq(claimed, SHA.sha256_packed_bytes(input)))
111	
112	def header.ready.block(pair: Bytes.Bytes & Maybe<&2, U32>, entry_count: U32) -> Result<&1, &1, Error, Header>:
113	  match pair:
114	    case (input, value):
115	      match value:
116	        case None{}: Fail{InvalidFormat{}}
117	        case Some{block_count}: Done{Header{input, entry_count, block_count}}
118	
119	def header.entry_count.read(pair: Bytes.Bytes & Maybe<&2, U32>) -> Result<&1, &1, Error, Header>:
120	  match pair:
121	    case (input, value):
122	      match value:
123	        case None{}: Fail{InvalidFormat{}}
124	        case Some{entry_count}: header.ready.block(Bytes.get.u32be(input, 16), entry_count)
125	
126	def header.block_count.checked(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, Header>:
127	  match valid:
128	    case False{}: Fail{ResourceLimit{}}
129	    case True{}: header.entry_count.read(Bytes.get.u32be(input, 12))
130	
131	def header.block_count(pair: Bytes.Bytes & Maybe<&2, U32>) -> Result<&1, &1, Error, Header>:
132	  match pair:
133	    case (input, value):
134	      match value:
135	        case None{}: Fail{InvalidFormat{}}
136	        case Some{block_count}:
137	          header.block_count.checked(input, U32.is_le(block_count, 65536))
138	
139	def header.entry_count.checked(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, Header>:
140	  match valid:
141	    case False{}: Fail{ResourceLimit{}}
142	    case True{}: header.block_count(Bytes.get.u32be(input, 16))
143	
144	def header.entry_count(pair: Bytes.Bytes & Maybe<&2, U32>) -> Result<&1, &1, Error, Header>:
145	  match pair:
146	    case (input, value):
147	      match value:
148	        case None{}: Fail{InvalidFormat{}}
149	        case Some{entry_count}:
150	          header.entry_count.checked(input, U32.is_le(entry_count, 1048576))
151	
152	def header.version.checked(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, Header>:
153	  match valid:
154	    case False{}: Fail{InvalidFormat{}}
155	    case True{}: header.entry_count(Bytes.get.u32be(input, 12))
156	
157	def header.version(pair: Bytes.Bytes & Maybe<&2, U32>) -> Result<&1, &1, Error, Header>:
158	  match pair:
159	    case (input, value):
160	      match value:
161	        case None{}: Fail{InvalidFormat{}}
162	        case Some{version}: header.version.checked(input, U32.is_eq(version, 1))
163	
164	def header.magic.checked(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, Header>:
165	  match valid:
166	    case False{}: Fail{InvalidFormat{}}
167	    case True{}: header.version(Bytes.get.u32be(input, 8))
168	
169	def header.marker(pair: Bytes.Bytes & Bytes.Bytes) -> Result<&1, &1, Error, Header>:
170	  match pair:
171	    case (input, marker):
172	      header.magic.checked(input, equal(Bytes.eq(marker, magic.header())))
173	
174	def header.parse(input: Bytes.Bytes) -> Result<&1, &1, Error, Header>:
175	  header.marker(Bytes.slice(input, 0, 8))
176	
177	def block.checksum.lengths(
178	  lengths: U32 & U32,
179	  copies: Bytes.Bytes & Bytes.Bytes
180	) -> Bool:
181	  match lengths:
182	    case (hash_length, claim_offset):
183	      match copies:
184	        case (for_hash, for_claim):
185	          digest.matches(slice.result(Bytes.slice(for_claim, claim_offset, 32)),
186	            slice.result(Bytes.slice(for_hash, 0, hash_length)))
187	
188	def block.checksum.copies.lengths(
189	  lengths: Nat & Nat,
190	  copies: Bytes.Bytes & Bytes.Bytes
191	) -> Bool:
192	  match lengths:
193	    case (hash_length, claim_offset):
194	      block.checksum.lengths((U32.from_nat(hash_length),
195	        U32.from_nat(claim_offset)), copies)
196	
197	def block.checksum.copies(
198	  block_length: U32,
199	  copies: Bytes.Bytes & Bytes.Bytes
200	) -> Bool:
201	  block.checksum.copies.lengths(
202	    QueuePolicy.nat.clone.pair(U32.to_nat(U32.sub(block_length, 32))), copies)
203	
204	def block.checksum.sized.validated(
205	  block: Bytes.Bytes,
206	  hash_length: U32,
207	  valid: Bool
208	) -> Bool:
209	  match valid:
210	    case False{}: False{}
211	    case True{}:
212	      block.checksum.copies(hash_length, clone.split(block))
213	
214	def block.checksum.sized.measure(
215	  block: Bytes.Bytes,
216	  compare_length: U32,
217	  hash_length: U32,
218	  measured: Bytes.Bytes & U32
219	) -> Bool:
220	  match measured:
221	    case (_, actual):
222	      block.checksum.sized.validated(block, hash_length,
223	        U32.is_eq(compare_length, actual))
224	
225	def block.checksum.sized.lengths(
226	  block: Bytes.Bytes,
227	  lengths: Nat & Nat,
228	  measured: Bytes.Bytes & U32
229	) -> Bool:
230	  match lengths:
231	    case (for_compare, for_hash):
232	      block.checksum.sized.measure(block, U32.from_nat(for_compare),
233	        U32.from_nat(for_hash), measured)
234	
235	def block.checksum.sized(
236	  block: Bytes.Bytes,
237	  block_length: U32,
238	  measured: Bytes.Bytes & U32
239	) -> Bool:
240	  block.checksum.sized.lengths(block,
241	    QueuePolicy.nat.clone.pair(U32.to_nat(block_length)), measured)
242	
243	def block.checksum.input.cloned(
244	  block_length: U32,
245	  copies: Bytes.Bytes & Bytes.Bytes
246	) -> Bool:
247	  match copies:
248	    case (for_length, for_hash):
249	      block.checksum.sized(for_hash, block_length, Bytes.length(for_length))
250	
251	def block.checksum.input(
252	  block_length: U32,
253	  block: Bytes.Bytes & Bytes.Bytes
254	) -> Bool:
255	  match block:
256	    case (_, bytes):
257	      block.checksum.input.cloned(block_length, clone.split(bytes))
258	
259	def mutations.entries.loop(
260	  fuel: Nat,
261	  mutations: List<&1, ByteWal.Mutation>,
262	  reversed: List<&1, ByteMemTable.Entry>,
263	  count: Nat
264	) -> Result<&1, &1, Error, EntryBatch>:
265	  match fuel:
266	    case 0n: Fail{ResourceLimit{}}
267	    case 1n+rest:
268	      match mutations:
269	        case Nil{}:
270	          Done{EntryBatch{List.reverse(&1, ByteMemTable.Entry, reversed), count}}
271	        case Con{mutation, tail}:
272	          match mutation:
273	            case ByteWal.Put{key, value}:
274	              mutations.entries.loop(rest, tail,
275	                Con{ByteMemTable.Entry{key, Some{value}}, reversed},
276	                Nat.add(count, 1n))
277	            case ByteWal.Delete{_}: Fail{InvalidFormat{}}
278	
279	def mutations.entries(
280	  result: Result<&1, &1, Wal.Error, ByteWal.Batch>,
281	  count: U32
282	) -> Result<&1, &1, Error, EntryBatch>:
283	  match result:
284	    case Fail{_}: Fail{InvalidFormat{}}
285	    case Done{ByteWal.Batch{mutations}}:
286	      mutations.entries.loop(Nat.add(U32.to_nat(count), 1n), mutations, Nil{}, 0n)
287	
288	def footer.expected_prefix(entry_count: Nat, block_count: Nat) -> Bytes.Bytes:
289	  Bytes.concat([footer.magic(),
290	    Bytes.set.u32be(Bytes.new(4), 0, U32.from_nat(entry_count)),
291	    Bytes.set.u32be(Bytes.new(4), 0, U32.from_nat(block_count))])
292	
293	def footer.valid.digest(copies: Bytes.Bytes & Bytes.Bytes) -> Bool:
294	  match copies:
295	    case (for_claim, for_hash):
296	      digest.matches(slice.result(Bytes.slice(for_claim, 12, 32)),
297	        slice.result(Bytes.slice(for_hash, 0, 12)))
298	
299	def footer.valid.copies(
300	  copies: Bytes.Bytes & Bytes.Bytes,
301	  entry_count: Nat,
302	  block_count: Nat
303	) -> Bool:
304	  match copies:
305	    case (for_fields, for_digest):
306	      equal(Bytes.eq(slice.result(Bytes.slice(for_fields, 0, 12)),
307	        footer.expected_prefix(entry_count, block_count))) &&
308	        footer.valid.digest(clone.split(for_digest))
309	
310	def footer.valid.pair(
311	  pair: Bytes.Bytes & Bytes.Bytes,
312	  entry_count: Nat,
313	  block_count: Nat
314	) -> Bool:
315	  match pair:
316	    case (_, footer): footer.valid.copies(clone.split(footer), entry_count, block_count)
317	
318	def footer.valid(
319	  input: Bytes.Bytes,
320	  offset: U32,
321	  entry_count: Nat,
322	  block_count: Nat
323	) -> Bool:
324	  footer.valid.pair(Bytes.slice(input, offset, 44), entry_count, block_count)
325	
326	def footer.checked.checksum(
327	  entries: List<&1, ByteMemTable.Entry>,
328	  valid: Bool
329	) -> Result<&1, &1, Error, Decoded>:
330	  match valid:
331	    case False{}: Fail{ChecksumMismatch{}}
332	    case True{}: Done{Decoded{entries}}
333	
334	def footer.checked(
335	  input: Bytes.Bytes,
336	  offset: U32,
337	  count: Nat,
338	  block_count: Nat,
339	  entries: List<&1, ByteMemTable.Entry>,
340	  size_valid: Bool
341	) -> Result<&1, &1, Error, Decoded>:
342	  match size_valid:
343	    case False{}: Fail{InvalidFormat{}}
344	    case True{}:
345	      footer.checked.checksum(entries, footer.valid(input, offset, count, block_count))
346	
347	def footer.result(
348	  input: Bytes.Bytes,
349	  offset: U32,
350	  count: Nat,
351	  block_count: Nat,
352	  entries: List<&1, ByteMemTable.Entry>,
353	  size_valid: Bool
354	) -> Result<&1, &1, Error, Decoded>:
355	  footer.checked(input, offset, count, block_count, entries, size_valid)
356	
357	def empty.footer.valid.copies(copies: Bytes.Bytes & Bytes.Bytes) -> Bool:
358	  match copies:
359	    case (for_footer, for_digest):
360	      digest.matches(slice.result(Bytes.slice(for_footer, 32, 32)),
361	        slice.result(Bytes.slice(for_digest, 20, 12)))
362	
363	def empty.footer.valid(input: Bytes.Bytes) -> Bool:
364	  empty.footer.valid.copies(clone.split(input))
365	
366	def empty.header.blocks(pair: Bytes.Bytes & Maybe<&2, U32>) -> Bool:
367	  match pair:
368	    case (_, count): u32.field.is(count, 0)
369	
370	def empty.header.entries.checked(input: Bytes.Bytes, valid: Bool) -> Bool:
371	  match valid:
372	    case False{}: False{}
373	    case True{}: empty.header.blocks(Bytes.get.u32be(input, 16))
374	
375	def empty.header.entries(pair: Bytes.Bytes & Maybe<&2, U32>) -> Bool:
376	  match pair:
377	    case (input, count):
378	      empty.header.entries.checked(input, u32.field.is(count, 0))
379	
380	def empty.header.version.checked(input: Bytes.Bytes, valid: Bool) -> Bool:
381	  match valid:
382	    case False{}: False{}
383	    case True{}: empty.header.entries(Bytes.get.u32be(input, 12))
384	
385	def empty.header.version(pair: Bytes.Bytes & Maybe<&2, U32>) -> Bool:
386	  match pair:
387	    case (input, version):
388	      empty.header.version.checked(input, u32.field.is(version, 1))
389	
390	def empty.header.magic.checked(input: Bytes.Bytes, valid: Bool) -> Bool:
391	  match valid:
392	    case False{}: False{}
393	    case True{}: empty.header.version(Bytes.get.u32be(input, 8))
394	
395	def empty.header.magic(pair: Bytes.Bytes & Bytes.Bytes) -> Bool:
396	  match pair:
397	    case (input, marker):
398	      empty.header.magic.checked(input, equal(Bytes.eq(marker, magic.header())))
399	
400	def empty.header.valid(input: Bytes.Bytes) -> Bool:
401	  empty.header.magic(Bytes.slice(input, 0, 8))
402	
403	def decode.empty(valid: Bool) -> Result<&1, &1, Error, Decoded>:
404	  match valid:
405	    case False{}: Fail{InvalidFormat{}}
406	    case True{}: Done{Decoded{Nil{}}}
407	
408	def empty.footer.fields(input: Bytes.Bytes) -> Bool:
409	  equal(Bytes.eq(slice.result(Bytes.slice(input, 20, 12)),
410	    Bytes.concat([footer.magic(), Bytes.new(8)])))
411	
412	def decode.empty.footer.checked(
413	  input: Bytes.Bytes,
414	  checksum_valid: Bool
415	) -> Result<&1, &1, Error, Decoded>:
416	  match checksum_valid:
417	    case False{}: Fail{ChecksumMismatch{}}
418	    case True{}: decode.empty(empty.footer.fields(input))
419	
420	def decode.empty.footer.copies(
421	  copies: Bytes.Bytes & Bytes.Bytes
422	) -> Result<&1, &1, Error, Decoded>:
423	  match copies:
424	    case (for_fields, for_digest):
425	      decode.empty.footer.checked(for_fields, empty.footer.valid(for_digest))
426	
427	def decode.empty.header.checked(
428	  input: Bytes.Bytes,
429	  header_valid: Bool
430	) -> Result<&1, &1, Error, Decoded>:
431	  match header_valid:
432	    case False{}: Fail{InvalidFormat{}}
433	    case True{}: decode.empty.footer.copies(clone.split(input))
434	
435	def decode.empty.header.copies(
436	  copies: Bytes.Bytes & Bytes.Bytes
437	) -> Result<&1, &1, Error, Decoded>:
438	  match copies:
439	    case (for_footer, for_header):
440	      decode.empty.header.checked(for_footer, empty.header.valid(for_header))
441	
442	def decode.empty.size.validated(
443	  input: Bytes.Bytes,
444	  size_valid: Bool
445	) -> Result<&1, &1, Error, Decoded>:
446	  match size_valid:
447	    case False{}: Fail{InvalidFormat{}}
448	    case True{}: decode.empty.header.copies(clone.split(input))
449	
450	def decode.empty.size(input: Bytes.Bytes, size: U32) -> Result<&1, &1, Error, Decoded>:
451	  decode.empty.size.validated(input, U32.is_eq(size, 64))
452	
453	def decode.length(pair: Bytes.Bytes & U32) -> Result<&1, &1, Error, Decoded>:
454	  match pair:
455	    case (input, size): decode.empty.size(input, size)
456	
457	def restore.block.frame.bounded(
458	  input: Bytes.Bytes,
459	  +offset: Nat,
460	  index: U32,
461	  +record_count: U32,
462	  +frame_length: U32,
463	  +block_length: Nat,
464	  valid: Bool
465	) -> Result<&1, &1, Error, BlockHeader>:
466	  match valid:
467	    case False{}: Fail{InvalidFormat{}}
468	    case True{}:
469	      Done{BlockHeader{input, offset, index, record_count,
470	        frame_length, block_length}}
471	def restore.block.frame.bounds(
472	  input: Bytes.Bytes,
473	  +offset: Nat,
474	  index: U32,
475	  record_count: U32,
476	  +frame_length: U32,
477	  +block_length: Nat,
478	  +input_size: U32
479	) -> Result<&1, &1, Error, BlockHeader>:
480	  restore.block.frame.bounded(input, offset, index, record_count,
481	    frame_length, block_length,
482	    Nat.is_le(Nat.add(offset, block_length), U32.to_nat(input_size)))
483	def restore.block.frame.limit(
484	  input: Bytes.Bytes,
485	  +offset: Nat,
486	  index: U32,
487	  record_count: U32,
488	  +frame_length: U32,
489	  +input_size: U32,
490	  valid: Bool
491	) -> Result<&1, &1, Error, BlockHeader>:
492	  match valid:
493	    case False{}: Fail{ResourceLimit{}}
494	    case True{}:
495	      +block_length = Nat.add(U32.to_nat(frame_length), 48n)
496	      restore.block.frame.bounds(input, offset, index, record_count,
497	        frame_length, block_length, input_size)
498	def restore.block.frame.length.checked(
499	  measured: Bytes.Bytes & U32,
500	  +offset: Nat,
501	  index: U32,
502	  record_count: U32,
503	  +frame_length: U32,
504	  valid: Bool
505	) -> Result<&1, &1, Error, BlockHeader>:
506	  match measured:
507	    case (input, input_size):
508	      restore.block.frame.limit(input, offset, index, record_count,
509	        frame_length, input_size, valid)
510	
511	def restore.block.frame.field(
512	  pair: Bytes.Bytes & Maybe<&2, U32>,
513	  +offset: Nat,
514	  index: U32,
515	  record_count: U32
516	) -> Result<&1, &1, Error, BlockHeader>:
517	  match pair:
518	    case (input, result):
519	      match result:
520	        case None{}: Fail{InvalidFormat{}}
521	        case Some{+frame_length}:
522	          restore.block.frame.length.checked(Bytes.length(input), offset,
523	            index, record_count, frame_length,
524	            U32.is_le(frame_length, StorageBytes.MAX_BLOCK_BYTES()))
525	
526	def restore.block.count.valid(
527	  input: Bytes.Bytes,
528	  +offset: Nat,
529	  index: U32,
530	  record_count: U32,
531	  within_limit: Bool,
532	  empty: Bool
533	) -> Result<&1, &1, Error, BlockHeader>:
534	  match within_limit:
535	    case False{}: Fail{ResourceLimit{}}
536	    case True{}:
537	      match empty:
538	        case True{}: Fail{InvalidFormat{}}
539	        case False{}:
540	          restore.block.frame.field(Bytes.get.u32be(input,
541	            U32.from_nat(Nat.add(offset, 12n))), offset,
542	            index, record_count)
543	def restore.block.count.field(
544	  pair: Bytes.Bytes & Maybe<&2, U32>,
545	  +offset: Nat,
546	  index: U32
547	) -> Result<&1, &1, Error, BlockHeader>:
548	  match pair:
549	    case (input, result):
550	      match result:
551	        case None{}: Fail{InvalidFormat{}}
552	        case Some{+record_count}:
553	          restore.block.count.valid(input, offset, index, record_count,
554	            U32.is_le(record_count, 16), U32.is_eq(record_count, 0))
555	def restore.block.index.checked(
556	  input: Bytes.Bytes,
557	  +offset: Nat,
558	  index: U32,
559	  valid: Bool
560	) -> Result<&1, &1, Error, BlockHeader>:
561	  match valid:
562	    case False{}: Fail{InvalidFormat{}}
563	    case True{}:
564	      restore.block.count.field(Bytes.get.u32be(input,
565	        U32.from_nat(Nat.add(offset, 8n))), offset, index)
566	def restore.block.index.field(
567	  pair: Bytes.Bytes & Maybe<&2, U32>,
568	  +offset: Nat,
569	  expected_index: U32
570	) -> Result<&1, &1, Error, BlockHeader>:
571	  match pair:
572	    case (input, result):
573	      match result:
574	        case None{}: Fail{InvalidFormat{}}
575	        case Some{+index}:
576	          restore.block.index.checked(input, offset, index,
577	            U32.is_eq(index, expected_index))
578	def restore.block.index.valid(
579	  pair: Bytes.Bytes & Maybe<&2, U32>,
580	  +offset: Nat,
581	  expected_index: U32
582	) -> Result<&1, &1, Error, BlockHeader>:
583	  restore.block.index.field(pair, offset, expected_index)
584	def restore.block.magic.valid(
585	  input: Bytes.Bytes,
586	  +offset: Nat,
587	  expected_index: U32,
588	  valid: Bool
589	) -> Result<&1, &1, Error, BlockHeader>:
590	  match valid:
591	    case False{}: Fail{InvalidFormat{}}
592	    case True{}:
593	      restore.block.index.valid(Bytes.get.u32be(input,
594	        U32.from_nat(Nat.add(offset, 4n))), offset, expected_index)
595	
596	
597	def restore.block.magic(
598	  pair: Bytes.Bytes & Bytes.Bytes,
599	  +offset: Nat,
600	  expected_index: U32
601	) -> Result<&1, &1, Error, BlockHeader>:
602	  match pair:
603	    case (input, marker):
604	      restore.block.magic.valid(input, offset, expected_index,
605	        equal(Bytes.eq(marker, block.magic())))
606	
607	def restore.block.header(
608	  input: Bytes.Bytes,
609	  +offset: Nat,
610	  expected_index: U32
611	) -> Result<&1, &1, Error, BlockHeader>:
612	  restore.block.magic(Bytes.slice(input, U32.from_nat(offset), 4),
613	    offset, expected_index)
614	
615	def restore.block.batch.checked(
616	  input: Bytes.Bytes,
617	  +offset: Nat,
618	  index: U32,
619	  record_count: U32,
620	  frame_length: U32,
621	  +block_length: Nat,
622	  batch: EntryBatch,
623	  valid: Bool
624	) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
625	  match batch:
626	    case EntryBatch{entries, _}:
627	      match valid:
628	        case False{}: Fail{CountMismatch{}}
629	        case True{}:
630	          Done{(BlockHeader{input, offset, index, record_count,
631	            frame_length, block_length}, entries)}
632	
633	
634	def restore.block.batch.measure(
635	  input: Bytes.Bytes,
636	  +offset: Nat,
637	  index: U32,
638	  +record_count: U32,
639	  frame_length: U32,
640	  +block_length: Nat,
641	  batch: EntryBatch
642	) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
643	  match batch:
644	    case EntryBatch{_, +count}:
645	      restore.block.batch.checked(input, offset, index, record_count,
646	        frame_length, block_length, batch,
647	        Nat.is_eq(count, U32.to_nat(record_count)))
648	
649	
650	
651	def restore.block.frame.entries(
652	  header: BlockHeader,
653	  result: Result<&1, &1, Error, EntryBatch>
654	) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
655	  match header:
656	    case BlockHeader{input, offset, index, record_count, frame_length,
657	        block_length}:
658	      match result:
659	        case Fail{error}: Fail{error}
660	        case Done{batch}:
661	          restore.block.batch.measure(input, offset, index, record_count,
662	            frame_length, block_length, batch)
663	
664	def restore.block.frame.decoded(
665	  header: BlockHeader,
666	  result: Result<&1, &1, Wal.Error, ByteWal.Batch>
667	) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
668	  match header:
669	    case BlockHeader{_, _, _, +record_count, _, _}:
670	      restore.block.frame.entries(header, mutations.entries(result, record_count))
671	
672	
673	def restore.block.checksum.validated(
674	  header: BlockHeader,
675	  decoded: Result<&1, &1, Wal.Error, ByteWal.Batch>,
676	  valid: Bool
677	) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
678	  match valid:
679	    case False{}: Fail{ChecksumMismatch{}}
680	    case True{}: restore.block.frame.decoded(header, decoded)
681	
682	
683	def restore.block.checksum.copies(
684	  +offset: Nat,
685	  index: U32,
686	  record_count: U32,
687	  +frame_length: U32,
688	  +block_length: Nat,
689	  copies: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes
690	) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
691	  match copies:
692	    case (for_hash, for_frame, for_input):
693	      restore.block.checksum.validated(
694	        BlockHeader{for_input, offset, index, record_count,
695	          frame_length, block_length},
696	        ByteWal.decode(slice.result(Bytes.slice(for_frame,
697	          U32.from_nat(Nat.add(offset, 16n)), frame_length))),
698	        block.checksum.input(U32.from_nat(block_length),
699	          Bytes.slice(for_hash, U32.from_nat(offset),
700	            U32.from_nat(block_length))))
701	
702	
703	def restore.block.checksum.valid(
704	  header: BlockHeader
705	) -> Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>:
706	  match header:
707	    case BlockHeader{input, +offset, index, record_count, +frame_length,
708	        +block_length}:
709	      restore.block.checksum.copies(offset, index, record_count,
710	        frame_length, block_length, clone.triple(input))
711	
712	def restore.block.checked(
713	  result: Result<&1, &1, Error, BlockHeader & List<&1, ByteMemTable.Entry>>
714	) -> Result<&1, &1, Error, BlockMeta>:
715	  match result:
716	    case Fail{error}: Fail{error}
717	    case Done{(verified, entries)}:
718	      match verified:
719	        case BlockHeader{input, +offset, _, +record_count,
720	            +frame_length, +block_length}:
721	          Done{BlockMeta{input, offset, Nat.add(offset, 16n),
722	            U32.to_nat(frame_length), U32.to_nat(record_count), block_length,
723	            entries}}
724	
725	def restore.block.open.result(
726	  result: Result<&1, &1, Error, BlockHeader>
727	) -> Result<&1, &1, Error, BlockMeta>:
728	  match result:
729	    case Fail{error}: Fail{error}
730	    case Done{header}:
731	      restore.block.checked(restore.block.checksum.valid(header))
732	
733	
734	def restore.block.open(
735	  input: Bytes.Bytes,
736	  +offset: Nat,
737	  expected_index: U32
738	) -> Result<&1, &1, Error, BlockMeta>:
739	  restore.block.open.result(restore.block.header(input, offset, expected_index))
740	
741	
742	def restore.prepend.entries(
743	  entries: List<&1, ByteMemTable.Entry>,
744	  reversed: List<&1, ByteMemTable.Entry>
745	) -> List<&1, ByteMemTable.Entry>:
746	  match entries:
747	    case Nil{}: reversed
748	    case Con{entry, tail}:
749	      restore.prepend.entries(tail, Con{entry, reversed})
750	
751	def restore.finish.order(
752	  input: Bytes.Bytes,
753	  +offset: Nat,
754	  +entry_count: U32,
755	  +block_count: U32,
756	  entries: List<&1, ByteMemTable.Entry>,
757	  ordered: Bool
758	) -> Result<&1, &1, Error, Decoded>:
759	  match ordered:
760	    case False{}: Fail{InvalidOrder{}}
761	    case True{}:
762	      footer.checked.checksum(entries,
763	        footer.valid(input, U32.from_nat(offset), U32.to_nat(entry_count),
764	          U32.to_nat(block_count)))
765	
766	
767	def restore.finish.footer.copies(
768	  input: Bytes.Bytes,
769	  +offset: Nat,
770	  entry_count: U32,
771	  block_count: U32,
772	  copies: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
773	) -> Result<&1, &1, Error, Decoded>:
774	  match copies:
775	    case (for_order, entries):
776	      restore.finish.order(input, offset, entry_count, block_count, entries,
777	        ByteMemTable.ordered.strict(for_order))
778	
779	
780	
781	def restore.finish.footer(
782	  input: Bytes.Bytes,
783	  +offset: Nat,
784	  entry_count: U32,
785	  block_count: U32,
786	  entries: List<&1, ByteMemTable.Entry>
787	) -> Result<&1, &1, Error, Decoded>:
788	  restore.finish.footer.copies(input, offset, entry_count, block_count,
789	    ByteMemTable.entries.clone(entries))
790	
791	def restore.finish.size.valid(
792	  input: Bytes.Bytes,
793	  +offset: Nat,
794	  entry_count: U32,
795	  block_count: U32,
796	  reversed: List<&1, ByteMemTable.Entry>,
797	  valid: Bool
798	) -> Result<&1, &1, Error, Decoded>:
799	  match valid:
800	    case False{}: Fail{InvalidFormat{}}
801	    case True{}:
802	      restore.finish.footer(input, offset, entry_count, block_count,
803	        List.reverse(&1, ByteMemTable.Entry, reversed))
804	
805	
806	def restore.finish.count.valid(
807	  input: Bytes.Bytes,
808	  +offset: Nat,
809	  entry_count: U32,
810	  block_count: U32,
811	  reversed: List<&1, ByteMemTable.Entry>,
812	  +size: U32,
813	  valid: Bool
814	) -> Result<&1, &1, Error, Decoded>:
815	  match valid:
816	    case False{}: Fail{CountMismatch{}}
817	    case True{}:
818	      restore.finish.size.valid(input, offset, entry_count, block_count,
819	        reversed, Nat.is_eq(Nat.add(offset, 44n), U32.to_nat(size)))
820	
821	
822	def restore.finish(
823	  measured: Bytes.Bytes & U32,
824	  +offset: Nat,
825	  +entry_count: U32,
826	  +block_count: U32,
827	  +decoded_count: Nat,
828	  reversed: List<&1, ByteMemTable.Entry>
829	) -> Result<&1, &1, Error, Decoded>:
830	  match measured:
831	    case (input, size):
832	      restore.finish.count.valid(input, offset, entry_count, block_count,
833	        reversed, size,
834	        U32.is_eq(U32.from_nat(decoded_count), entry_count))
835	
836	def restore.loop(
837	  fuel: Nat,
838	  state: State,
839	  current: Result<&1, &1, Error, BlockMeta>
840	) -> Result<&1, &1, Error, Decoded>:
841	  match fuel:
842	    case 0n: Fail{InvalidFormat{}}
843	    case 1n+fuel_rest:
844	    match state:
845	      case State{input, +offset, entry_count, block_count, +remaining_blocks,
846	          +block_index, +decoded_count, reversed}:
847	        match input:
848	          case (for_size, retained):
849	            match remaining_blocks:
850	              case 0n: Fail{InvalidFormat{}}
851	              case 1n+rest:
852	                match rest:
853	                  case 0n:
854	                    match current:
855	                      case Fail{error}: Fail{error}
856	                      case Done{block}:
857	                        match block:
858	                          case BlockMeta{_, _, _, _, +record_count, +block_length, entries}:
859	                            +next_offset = Nat.add(offset, block_length)
860	                            +next_count = Nat.add(decoded_count, record_count)
861	                            next_reversed = restore.prepend.entries(entries, reversed)
862	                            restore.finish(Bytes.length(for_size), next_offset,
863	                              entry_count, block_count, next_count,
864	                              next_reversed)
865	                  case 1n+more:
866	                    match current:
867	                      case Fail{error}: Fail{error}
868	                      case Done{block}:
869	                        match block:
870	                          case BlockMeta{_, _, _, _, +record_count, +block_length, entries}:
871	                            +next_offset = Nat.add(offset, block_length)
872	                            +next_index = Nat.add(block_index, 1n)
873	                            +next_count = Nat.add(decoded_count, record_count)
874	                            next_reversed = restore.prepend.entries(entries, reversed)
875	                            restore.loop(fuel_rest,
876	                            State{clone.split(retained), next_offset, entry_count,
877	                              block_count,
878	                              RestorePolicy.remaining(
879	                                RestorePolicy.after_block(remaining_blocks)),
880	                              next_index, next_count,
881	                                next_reversed},
882	                              restore.block.open(for_size, next_offset,
883	                                U32.from_nat(next_index)))
884	
885	def restore.loop.start(
886	  copies: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes,
887	  entry_count: U32,
888	  +block_count: U32
889	) -> Result<&1, &1, Error, Decoded>:
890	  match copies:
891	    case (for_state, for_retained, for_open):
892	      restore.loop(U32.to_nat(block_count),
893	        State{(for_state, for_retained), 20n, entry_count, block_count,
894	          U32.to_nat(block_count), 0n, 0n, Nil{}},
895	        restore.block.open(for_open, 20n, 0))
896	
897	
898	def decode.header.blocks(input: Bytes.Bytes, header: Header) -> Result<&1, &1, Error, Decoded>:
899	  match header:
900	    case Header{_, entry_count, block_count}:
901	      restore.loop.start(clone.triple(input), entry_count, block_count)
902	
903	def decode.header.route(input: Bytes.Bytes, header: Header, empty: Bool) -> Result<&1, &1, Error, Decoded>:
904	  match empty:
905	    case True{}: decode.length(Bytes.length(input))
906	    case False{}: decode.header.blocks(input, header)
907	
908	def decode.header.ready.count(
909	  input: Bytes.Bytes,
910	  header_input: Bytes.Bytes,
911	  entry_count: U32,
912	  counts: Nat & Nat
913	) -> Result<&1, &1, Error, Decoded>:
914	  match counts:
915	    case (for_check, for_header):
916	      decode.header.route(input,
917	        Header{header_input, entry_count, U32.from_nat(for_header)},
918	        U32.is_eq(U32.from_nat(for_check), 0))
919	
920	def decode.header.ready(input: Bytes.Bytes, header: Header) -> Result<&1, &1, Error, Decoded>:
921	  match header:
922	    case Header{header_input, entry_count, block_count}:
923	      decode.header.ready.count(input, header_input, entry_count,
924	        QueuePolicy.nat.clone.pair(U32.to_nat(block_count)))
925	
926	def decode.header(input: Bytes.Bytes, result: Result<&1, &1, Error, Header>) -> Result<&1, &1, Error, Decoded>:
927	  match result:
928	    case Fail{error}: Fail{error}
929	    case Done{header}: decode.header.ready(input, header)
930	
931	def decode.copies(copies: Bytes.Bytes & Bytes.Bytes) -> Result<&1, &1, Error, Decoded>:
932	  match copies:
933	    case (for_restore, for_header):
934	      decode.header(for_restore, header.parse(for_header))
935	
936	# Validates a bounded exchange and decodes the currently supported block layout.
937	def decode(input: Bytes.Bytes) -> Result<&1, &1, Error, Decoded>:
938	  decode.copies(clone.split(input))