1	import Base
2	import ../coordinator/QueuePolicy.bend as QueuePolicy
3	import ../Wal.bend as Wal
4	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
5	
6	# Reports a validated frame count or a malformed legacy WAL.
7	type VerifyResult is Data:
8	  WalValid{frames: Nat}
9	  WalInvalid{}
10	
11	# Returns all validated legacy WAL batches in their durable order.
12	type ReplayResult is Data:
13	  ReplayValid{batches: List<&2, Wal.Batch>}
14	  ReplayInvalid{}
15	
16	# Carries bounded offsets and decoded batches through read-only replay.
17	type ReplayState is Type:
18	  ReplayNeed{bytes: Bytes.Bytes, offset: Nat, batches: List<&2, Wal.Batch>}
19	  ReplayMeasured{bytes: Bytes.Bytes, offset: Nat, batches: List<&2, Wal.Batch>,
20	    size: U32}
21	  ReplayBound{bytes: Bytes.Bytes, offset: Nat, batches: List<&2, Wal.Batch>,
22	    total: Nat, at_end: Bool, has_length: Bool}
23	  ReplayLength{bytes: Bytes.Bytes, next_offset: Nat, total: Nat,
24	    batches: List<&2, Wal.Batch>, length: Maybe<&2, U32>}
25	  ReplayFrame{bytes: Bytes.Bytes, next_offset: Nat, frame_end: Nat,
26	    batches: List<&2, Wal.Batch>, length: U32, fits: Bool}
27	  ReplayDecoded{bytes: Bytes.Bytes, next_offset: Nat,
28	    batches: List<&2, Wal.Batch>, result: Result<&1, &1, Wal.Error, Wal.Batch>}
29	
30	# Carries the read-only legacy WAL parser across bounded pure transitions.
31	type WalkState is Type:
32	  Header{pair: Bytes.Bytes & Bytes.Bytes}
33	  HeaderDecoded{source: Bytes.Bytes, result: Result<&1, &1, Wal.Error, Unit>}
34	  HeaderMeasured{pair: Bytes.Bytes & U32}
35	  HeaderSeek{pair: Bytes.Cursor & Bool}
36	  Read{cursor: Bytes.Cursor, frames: Nat}
37	  Remaining{pair: Bytes.Cursor & U32, frames: Nat}
38	  Empty{cursor: Bytes.Cursor, remaining: Nat, frames: Nat}
39	  EmptyChecked{cursor: Bytes.Cursor, remaining: Nat, frames: Nat, empty: Bool}
40	  HasLength{cursor: Bytes.Cursor, remaining: Nat, frames: Nat, has_length: Bool}
41	  Length{pair: Bytes.Cursor & Maybe<&2, U32>, remaining: U32, frames: Nat}
42	  Bounds{cursor: Bytes.Cursor, length: U32, frames: Nat, valid: Bool}
43	  Position{cursor: Bytes.Cursor, positions: Nat & Nat,
44	    frame_bytes: U32, frame_length: U32, frames: Nat}
45	  Frame{pair: Bytes.Bytes & Bytes.Bytes, end: U32, frames: Nat}
46	  Decoded{result: Result<&1, &1, Wal.Error, Wal.Batch>, source: Bytes.Bytes, end: U32, frames: Nat}
47	  Seek{pair: Bytes.Cursor & Bool, frames: Nat}
48	  Finish{frames: Nat}
49	  Invalid{}
50	
51	# Separates an empty log suffix from the next frame-length transition.
52	def walk.empty.checked(
53	  cursor: Bytes.Cursor,
54	  frames: Nat,
55	  copies: Nat & Nat
56	) -> WalkState:
57	  match copies:
58	    case (for_check, retained):
59	      EmptyChecked{cursor, retained, frames, Nat.is_eq(for_check, 0n)}
60	
61	# Checks that a nonempty log suffix contains a complete frame-length word.
62	def walk.length.checked(
63	  cursor: Bytes.Cursor,
64	  frames: Nat,
65	  copies: Nat & Nat
66	) -> WalkState:
67	  match copies:
68	    case (for_check, retained):
69	      HasLength{cursor, retained, frames, Nat.is_le(4n, for_check)}
70	
71	def walk.nat.clone.remainder(
72	  first: Nat,
73	  pair: Nat & Nat
74	) -> Nat & Nat & Nat:
75	  match pair:
76	    case (second, third): (first, second, third)
77	
78	def walk.nat.clone.pair(pair: Nat & Nat) -> Nat & Nat & Nat:
79	  match pair:
80	    case (first, remainder):
81	      walk.nat.clone.remainder(first,
82	        QueuePolicy.nat.clone.pair(remainder))
83	
84	def walk.nat.clone.triple(value: Nat) -> Nat & Nat & Nat:
85	  walk.nat.clone.pair(QueuePolicy.nat.clone.pair(value))
86	
87	def walk.length.copies.for.cursor(
88	  cursor: Bytes.Cursor,
89	  remaining: U32,
90	  frames: Nat,
91	  pair: Nat & Nat & Nat
92	) -> WalkState:
93	  match pair:
94	    case (minimum_check, (limit_check, retained_length)):
95	      +valid = Nat.is_le(36n, minimum_check) &&
96	        Nat.is_le(limit_check, U32.to_nat(remaining))
97	      Bounds{cursor, U32.from_nat(retained_length), frames, valid}
98	
99	def walk.length.validated(
100	  cursor: Bytes.Cursor,
101	  remaining: U32,
102	  frames: Nat,
103	  length: U32
104	) -> WalkState:
105	  walk.length.copies.for.cursor(cursor, remaining, frames,
106	    walk.nat.clone.triple(U32.to_nat(length)))
107	
108	# Prepares independent frame-size and endpoint lengths for the cursor state.
109	def walk.position.measured(
110	  pair: Bytes.Cursor & U32,
111	  frames: Nat,
112	  lengths: Nat & Nat
113	) -> WalkState:
114	  match pair:
115	    case (cursor, position):
116	      match lengths:
117	        case (frame_length, endpoint_length):
118	          Position{cursor, QueuePolicy.nat.clone.pair(U32.to_nat(position)),
119	            U32.from_nat(Nat.add(frame_length, 4n)),
120	            U32.from_nat(endpoint_length), frames}
121	
122	def walk.position.state(
123	  cursor: Bytes.Cursor,
124	  frames: Nat,
125	  lengths: Nat & Nat
126	) -> WalkState:
127	  walk.position.measured(Bytes.Cursor.position(cursor), frames, lengths)
128	
129	# Applies one pure parser transition, rejecting malformed or truncated frames.
130	def walk(fuel: Nat, state: WalkState) -> VerifyResult:
131	  match fuel:
132	    case 0n: WalInvalid{}
133	    case 1n+rest:
134	      match state:
135	        case Header{pair}:
136	          match pair:
137	            case (source, header):
138	              walk(rest, HeaderDecoded{source, Wal.decode_header(header)})
139	        case HeaderDecoded{source, result}:
140	          match result:
141	            case Fail{_}: WalInvalid{}
142	            case Done{Unit{}}:
143	              walk(rest, HeaderMeasured{Bytes.length(source)})
144	        case HeaderMeasured{pair}:
145	          match pair:
146	            case (source, size):
147	              walk(rest, HeaderSeek{
148	                Bytes.Cursor.seek(Bytes.Cursor.new(source), 8)})
149	        case HeaderSeek{pair}:
150	          match pair:
151	            case (_, False{}): WalInvalid{}
152	            case (cursor, True{}): walk(rest, Read{cursor, 0n})
153	        case Read{cursor, frames}:
154	          walk(rest, Remaining{Bytes.Cursor.remaining(cursor), frames})
155	        case Remaining{pair, frames}:
156	          match pair:
157	            case (cursor, remaining):
158	              walk(rest, Empty{cursor, U32.to_nat(remaining), frames})
159	        case Empty{cursor, remaining, frames}:
160	          walk(rest, walk.empty.checked(cursor, frames,
161	            QueuePolicy.nat.clone.pair(remaining)))
162	        case EmptyChecked{cursor, remaining, frames, empty}:
163	          match empty:
164	            case True{}: walk(rest, Finish{frames})
165	            case False{}:
166	              walk(rest, walk.length.checked(cursor, frames,
167	                QueuePolicy.nat.clone.pair(remaining)))
168	        case HasLength{cursor, remaining, frames, has_length}:
169	          match has_length:
170	            case False{}: WalInvalid{}
171	            case True{}:
172	              walk(rest, Length{Bytes.Cursor.u32be(cursor),
173	                U32.from_nat(Nat.sub(remaining, 4n)), frames})
174	        case Length{pair, remaining, frames}:
175	          match pair:
176	            case (_, None{}): WalInvalid{}
177	            case (cursor, Some{length}):
178	              walk(rest, walk.length.validated(cursor, remaining, frames, length))
179	        case Bounds{cursor, length, frames, valid}:
180	          match valid:
181	            case False{}: WalInvalid{}
182	            case True{}:
183	              walk(rest, walk.position.state(cursor, frames,
184	                QueuePolicy.nat.clone.pair(U32.to_nat(length))))
185	        case Position{cursor, positions, frame_bytes, frame_length, frames}:
186	          match cursor:
187	            case Bytes.Cursor{bytes, _, _, _}:
188	              match positions:
189	                case (start_nat, end_nat):
190	                  +start = U32.from_nat(Nat.sub(start_nat, 4n))
191	                  +end = U32.add(U32.from_nat(end_nat), frame_length)
192	                  walk(rest, Frame{Bytes.slice(bytes, start, frame_bytes),
193	                    end, frames})
194	        case Frame{pair, end, frames}:
195	          match pair:
196	            case (source, frame):
197	              walk(rest, Decoded{Wal.decode_frame(frame), source, end, frames})
198	        case Decoded{result, source, end, frames}:
199	          match result:
200	            case Fail{_}: WalInvalid{}
201	            case Done{_}:
202	              walk(rest, Seek{Bytes.Cursor.seek(Bytes.Cursor.new(source), end),
203	                Nat.add(frames, 1n)})
204	        case Seek{pair, frames}:
205	          match pair:
206	            case (_, False{}): WalInvalid{}
207	            case (cursor, True{}): walk(rest, Read{cursor, frames})
208	        case Finish{frames}: WalValid{frames}
209	        case Invalid{}: WalInvalid{}
210	
211	# Converts the file byte count into a finite transition bound.
212	def verify.measured(pair: Bytes.Bytes & U32) -> VerifyResult:
213	  match pair:
214	    case (bytes, size):
215	      walk(Nat.add(5n,
216	        Nat.mul(12n, Nat.add(Nat.div(U32.to_nat(size), 40n), 1n))),
217	        Header{Bytes.slice(bytes, 0, 8)})
218	
219	# Validates a complete generation-three WAL without repairing its bytes.
220	def verify(bytes: Bytes.Bytes) -> VerifyResult:
221	  verify.measured(Bytes.length(bytes))
222	
223	def replay.measure(
224	  measured: Bytes.Bytes & U32,
225	  offset: Nat,
226	  batches: List<&2, Wal.Batch>
227	) -> ReplayState:
228	  match measured:
229	    case (bytes, size): ReplayMeasured{bytes, offset, batches, size}
230	
231	def replay.nat.clone.remainder(first: Nat, pair: Nat & Nat) -> Nat & Nat & Nat:
232	  match pair:
233	    case (second, third): (first, second, third)
234	
235	def replay.nat.clone.pair(pair: Nat & Nat) -> Nat & Nat & Nat:
236	  match pair:
237	    case (first, remainder):
238	      replay.nat.clone.remainder(first, QueuePolicy.nat.clone.pair(remainder))
239	
240	def replay.nat.clone.triple(value: Nat) -> Nat & Nat & Nat:
241	  replay.nat.clone.pair(QueuePolicy.nat.clone.pair(value))
242	
243	def replay.bound.measured(
244	  bytes: Bytes.Bytes,
245	  batches: List<&2, Wal.Batch>,
246	  offsets: Nat & Nat & Nat,
247	  totals: Nat & Nat & Nat
248	) -> ReplayState:
249	  match offsets:
250	    case (retained_offset, (end_offset, length_offset)):
251	      match totals:
252	        case (retained_total, (end_total, length_total)):
253	          ReplayBound{bytes, retained_offset, batches, retained_total,
254	            Nat.is_eq(end_offset, end_total),
255	            Nat.is_le(Nat.add(length_offset, 4n), length_total)}
256	
257	def replay.bound.state(
258	  bytes: Bytes.Bytes,
259	  offset: Nat,
260	  batches: List<&2, Wal.Batch>,
261	  total: Nat
262	) -> ReplayState:
263	  replay.bound.measured(bytes, batches,
264	    replay.nat.clone.triple(offset), replay.nat.clone.triple(total))
265	
266	def replay.length.pair(
267	  measured: Bytes.Bytes & Maybe<&2, U32>,
268	  next_offset: Nat,
269	  total: Nat,
270	  batches: List<&2, Wal.Batch>
271	) -> ReplayState:
272	  match measured:
273	    case (bytes, length):
274	      ReplayLength{bytes, next_offset, total, batches, length}
275	
276	def replay.length.indexed(
277	  bytes: Bytes.Bytes,
278	  offsets: Nat & Nat,
279	  total: Nat,
280	  batches: List<&2, Wal.Batch>
281	) -> ReplayState:
282	  match offsets:
283	    case (next_offset, read_offset):
284	      replay.length.pair(Bytes.get.u32be(bytes, U32.from_nat(read_offset)),
285	        Nat.add(next_offset, 4n), total, batches)
286	
287	def replay.length.state(
288	  bytes: Bytes.Bytes,
289	  offset: Nat,
290	  total: Nat,
291	  batches: List<&2, Wal.Batch>
292	) -> ReplayState:
293	  replay.length.indexed(bytes, QueuePolicy.nat.clone.pair(offset),
294	    total, batches)
295	
296	def replay.frame.measured(
297	  bytes: Bytes.Bytes,
298	  batches: List<&2, Wal.Batch>,
299	  total: Nat,
300	  offsets: Nat & Nat,
301	  lengths: Nat & Nat
302	) -> ReplayState:
303	  match offsets:
304	    case (slice_offset, end_offset):
305	      match lengths:
306	        case (end_length, slice_length):
307	          +frame_end = Nat.add(end_offset, end_length)
308	          +frame_start = Nat.sub(slice_offset, 4n)
309	          +serialized_length = Nat.add(slice_length, 4n)
310	          ReplayFrame{bytes, frame_start, frame_end, batches,
311	            U32.from_nat(serialized_length), Nat.is_le(frame_end, total)}
312	
313	def replay.frame.state(
314	  bytes: Bytes.Bytes,
315	  next_offset: Nat,
316	  total: Nat,
317	  batches: List<&2, Wal.Batch>,
318	  length: U32
319	) -> ReplayState:
320	  replay.frame.measured(bytes, batches, total,
321	    QueuePolicy.nat.clone.pair(next_offset),
322	    QueuePolicy.nat.clone.pair(U32.to_nat(length)))
323	
324	def replay.frame.decoded(
325	  pair: Bytes.Bytes & Bytes.Bytes,
326	  frame_end: Nat,
327	  batches: List<&2, Wal.Batch>
328	) -> ReplayState:
329	  match pair:
330	    case (bytes, frame):
331	      ReplayDecoded{bytes, frame_end, batches, Wal.decode_frame(frame)}
332	
333	def replay.loop(fuel: Nat, state: ReplayState) -> ReplayResult:
334	  match fuel:
335	    case 0n: ReplayInvalid{}
336	    case 1n+rest:
337	      match state:
338	        case ReplayNeed{bytes, offset, batches}:
339	          replay.loop(rest, replay.measure(Bytes.length(bytes), offset, batches))
340	        case ReplayMeasured{bytes, offset, batches, size}:
341	          +total = U32.to_nat(size)
342	          replay.loop(rest, replay.bound.state(bytes, offset, batches, total))
343	        case ReplayBound{bytes, offset, batches, total, at_end, has_length}:
344	              match at_end:
345	                case True{}:
346	                  ReplayValid{batches}
347	                case False{}:
348	                  match has_length:
349	                    case True{}:
350	                      replay.loop(rest,
351	                        replay.length.state(bytes, offset, total, batches))
352	                    case False{}: ReplayInvalid{}
353	        case ReplayLength{bytes, next_offset, total, batches, length}:
354	          match length:
355	            case None{}: ReplayInvalid{}
356	            case Some{frame_length}:
357	              replay.loop(rest, replay.frame.state(bytes, next_offset,
358	                total, batches, frame_length))
359	        case ReplayFrame{bytes, next_offset, frame_end, batches, length, fits}:
360	              match fits:
361	                case True{}:
362	                  replay.loop(rest, replay.frame.decoded(Bytes.slice(bytes,
363	                    U32.from_nat(next_offset), length), frame_end, batches))
364	                case False{}: ReplayInvalid{}
365	        case ReplayDecoded{bytes, next_offset, batches, result}:
366	          match result:
367	            case Fail{_}: ReplayInvalid{}
368	            case Done{batch}:
369	              replay.loop(rest, ReplayNeed{bytes, next_offset,
370	                List.append(&2, Wal.Batch, batches, Con{batch, Nil{}})})
371	
372	def replay.header.validated(
373	  source: Bytes.Bytes,
374	  fuel: Nat,
375	  validation: Result<&1, &1, Wal.Error, Unit>
376	) -> ReplayResult:
377	  match validation:
378	    case Fail{_}: ReplayInvalid{}
379	    case Done{Unit{}}: replay.loop(fuel, ReplayNeed{source, 8n, Nil{}})
380	
381	def replay.header.pair(
382	  pair: Bytes.Bytes & Bytes.Bytes,
383	  fuel: Nat
384	) -> ReplayResult:
385	  match pair:
386	    case (source, header):
387	      replay.header.validated(source, fuel, Wal.decode_header(header))
388	
389	# Derives replay fuel from the source size and validates the header before decoding frames.
390	def replay.measured(pair: Bytes.Bytes & U32) -> ReplayResult:
391	  match pair:
392	    case (bytes, size):
393	      replay.header.pair(Bytes.slice(bytes, 0, 8),
394	        Nat.add(3n, Nat.mul(6n, Nat.div(U32.to_nat(size), 40n))))
395	
396	# Validates and replays all frames without repairing or truncating the source.
397	def replay(bytes: Bytes.Bytes) -> ReplayResult:
398	  replay.measured(Bytes.length(bytes))