1	import Base
2	import ./Keys.bend as Keys
3	import ./MemTable.bend as MemTable
4	import ./Sstable.bend as Sstable
5	import ./SstStreamIo.bend as SstStreamIo
6	import ./Wal.bend as Wal
7	import ./Manifest.bend as Manifest
8	import ./Flush.bend as Flush
9	import ./Compact.bend as Compact
10	import ./CompactIo.bend as CompactIo
11	import ./Fs.bend as Fs
12	import ./FsPolicy.bend as FsPolicy
13	import ./Db.bend as Db
14	import ./DbIo.bend as DbIo
15	import ./RecoverPure.bend as Pure
16	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
17	
18	# Represent ExactResult data used by the database recovery effects.
19	type ExactResult is Type:
20	  ExactResult{bytes: Bytes.Bytes, complete: Bool}
21	
22	# Represent ExactState data used by the database recovery effects.
23	type ExactState is Type:
24	  ExactNeed{file: File, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
25	  ExactRead{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
26	
27	# Represent WalProgress data used by the database recovery effects.
28	type WalProgress is Type:
29	  WalMore{file: File, offset: Nat, state: Db.RotRes}
30	  WalStop{result: Result<&1, &1, U32 & String, Db.RotRes>}
31	
32	# Represent WalState data used by the database recovery effects.
33	type WalState is Type:
34	  WalAt{file: File, offset: Nat, state: Db.RotRes}
35	  WalAfter{progress: WalProgress}
36	
37	# Returns the typed corruption code for malformed WAL data.
38	def wal_corruption_error_code() -> U32:
39	  4294967290
40	
41	def exact.max(large: Bool, +remaining: Nat) -> U32:
42	  match large:
43	    case True{}:
44	      1048576
45	    case False{}:
46	      U32.from_nat(remaining)
47	
48	def exact.result(
49	  file: File,
50	  chunks: List<&1, Bytes.Bytes>,
51	  complete: Bool
52	) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
53	  IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
54	    (file, Done{ExactResult{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks)), complete}}))
55	
56	def exact.loop(
57	  fuel: Nat,
58	  state: ExactState
59	) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
60	  match fuel:
61	    case 0n:
62	      match state:
63	        case ExactNeed{file, remaining, chunks}:
64	          match remaining:
65	            case 0n:
66	              exact.result(file, chunks, True{})
67	            case 1n+_:
68	              IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
69	                (file, Fail{(U32.from_nat(7n), "bounded read exhausted")}))
70	        case ExactRead{pair, _, _}:
71	          match pair:
72	            case (file, Fail{error}):
73	              IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{error}))
74	            case (file, Done{_}):
75	              IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
76	                (file, Fail{(U32.from_nat(7n), "bounded read exhausted")}))
77	    case 1n+rest:
78	      match state:
79	        case ExactNeed{file, +remaining, chunks}:
80	          match remaining:
81	            case 0n:
82	              exact.result(file, chunks, True{})
83	            case 1n+_:
84	              do IO<File & Result<&1, &1, U32 & String, ExactResult>>:
85	                next : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
86	                  Fs.read_bytes(file, exact.max(Nat.is_le(1048577n, remaining), remaining))
87	                exact.loop(rest, ExactRead{next, remaining, chunks})
88	        case ExactRead{pair, remaining, chunks}:
89	          match pair:
90	            case (file, Fail{error}):
91	              IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{error}))
92	            case (file, Done{Bytes.Bytes{0, _}}):
93	              exact.result(file, chunks, False{})
94	            case (file, Done{Bytes.Bytes{+len, buf}}):
95	              exact.loop(rest, ExactNeed{file,
96	                (remaining - U32.to_nat(len) : Nat),
97	                Con{Bytes.Bytes{len, buf}, chunks}})
98	
99	# Recovery + maintenance (Task 10).
100	#
101	# Open: ensure dir; sweep *.tmp (best-effort); load manifest (absent =
102	# fresh; corrupt = fatal); shape-check all names; load tables positionally
103	# (listed-but-missing/corrupt = fatal, never silent); restore the flush
104	# counter from the greatest table-name generation; replay + truncate
105	# the WAL; return the Db.
106	#
107	# Maintenance (synchronous, observably identical ordering to background):
108	# after every acked batch: drain while L0 >= 8 (stall = blocking
109	# maintenance), flush when mem >= 4096, compact (self-gated at L0 > 4).
110	# True fork-based background (frozen mem + IO.spawn) is a benchmark-gated
111	# follow-up; ordering and stall semantics are already exact.
112	
113	# --- Loaders (self-recursive IO; fatal via IO.die, never silent) ---
114	
115	# Represents loadtablesstate data in recover.
116	type LoadTablesState is Type:
117	  LoadTablesNames{names: List<&2, String>, ldir: String, dir: String, acc: List<&2, Sstable.Table>}
118	  LoadTablesRead{result: Result<&1, &1, U32 & String, Sstable.Table>, names: List<&2, String>, ldir: String, dir: String, acc: List<&2, Sstable.Table>}
119	
120	# Represents loadlevelsstate data in recover.
121	type LoadLevelsState is Type:
122	  LoadLevelsNames{levels: List<&2, List<&2, String>>, idx: Nat, dir: String, acc: List<&2, List<&2, Sstable.Table>>}
123	  LoadLevelsPath{ldir: Maybe<&2, String>, names: List<&2, String>, levels: List<&2, List<&2, String>>, idx: Nat, dir: String, acc: List<&2, List<&2, Sstable.Table>>}
124	  LoadLevelsRead{result: Result<&1, &1, U32 & String, List<&2, Sstable.Table>>, levels: List<&2, List<&2, String>>, idx: Nat, dir: String, acc: List<&2, List<&2, Sstable.Table>>}
125	
126	# Loads the SSTables referenced by a Manifest and propagates failures.
127	def load_tables(fuel: Nat, state: LoadTablesState) -> IO(Result<&1, &1, U32 & String, List<&2, Sstable.Table>>):
128	  match fuel:
129	    case 0n:
130	      IO.pure(Result<&1, &1, U32 & String, List<&2, Sstable.Table>>, Fail{(U32.from_nat(7n), "table recovery limit exceeded")})
131	    case 1n+rest:
132	      match state:
133	        case LoadTablesNames{names, +ldir, +dir, acc}:
134	          match names:
135	            case Nil{}:
136	              IO.pure(Result<&1, &1, U32 & String, List<&2, Sstable.Table>>, Done{List.reverse(&2, Sstable.Table, acc)})
137	            case Con{+name, tail}:
138	              do IO<Result<&1, &1, U32 & String, List<&2, Sstable.Table>>>:
139	                table : Result<&1, &1, U32 & String, Sstable.Table> <- SstStreamIo.read_table(dir ++ "/" ++ ldir ++ "/" ++ name)
140	                load_tables(rest, LoadTablesRead{table, tail, ldir, dir, acc})
141	        case LoadTablesRead{result, names, +ldir, +dir, acc}:
142	          match result:
143	            case Fail{error}:
144	              IO.pure(Result<&1, &1, U32 & String, List<&2, Sstable.Table>>, Fail{error})
145	            case Done{table}:
146	              load_tables(rest, LoadTablesNames{names, ldir, dir, Con{table, acc}})
147	
148	# Handle ldir of in the database recovery effects.
149	def ldir_of(idx: Nat) -> Maybe<&2, String>:
150	  match idx:
151	    case 0n:
152	      Some{"l0"}
153	    case 1n+m:
154	      match m:
155	        case 0n:
156	          Some{"l1"}
157	        case 1n+p:
158	          match p:
159	            case 0n:
160	              Some{"l2"}
161	            case 1n+q:
162	              match q:
163	                case 0n:
164	                  Some{"l3"}
165	                case 1n+r:
166	                  None{}
167	
168	# Builds recovered levels from the Manifest table entries.
169	def load_levels(
170	  fuel: Nat,
171	  state: LoadLevelsState,
172	) -> IO(Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>):
173	  match fuel:
174	    case 0n:
175	      IO.pure(Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>, Fail{(U32.from_nat(7n), "level recovery limit exceeded")})
176	    case 1n+rest:
177	      match state:
178	        case LoadLevelsNames{levels, +idx, +dir, acc}:
179	          match levels:
180	            case Nil{}:
181	              IO.pure(Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>, Done{List.reverse(&2, List<&2, Sstable.Table>, acc)})
182	            case Con{+names, tail}:
183	              load_levels(rest, LoadLevelsPath{ldir_of(idx), names, tail, Nat.add(idx, 1n), dir, acc})
184	        case LoadLevelsPath{ldir, +names, levels, idx, +dir, acc}:
185	          match ldir:
186	            case None{}:
187	              IO.pure(Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>, Fail{(U32.from_nat(5n), "bad manifest names")})
188	            case Some{ldir}:
189	              do IO<Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>>:
190	                tables : Result<&1, &1, U32 & String, List<&2, Sstable.Table>> <- load_tables(Nat.add(Nat.mul(2n, List.length(&2, String, names)), 1n), LoadTablesNames{names, ldir, dir, Nil{}})
191	                load_levels(rest, LoadLevelsRead{tables, levels, idx, dir, acc})
192	        case LoadLevelsRead{result, levels, idx, +dir, acc}:
193	          match result:
194	            case Fail{error}:
195	              IO.pure(Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>, Fail{error})
196	            case Done{tables}:
197	              load_levels(rest, LoadLevelsNames{levels, idx, dir, Con{tables, acc}})
198	
199	def wal.empty() -> Db.RotRes:
200	  Db.Rot{MemTable.empty(), MemTable.empty(), 0n, 0n}
201	
202	def exact.start(file: File, +needed: Nat) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
203	  exact.loop(Nat.add(Nat.mul(2n, needed), 1n), ExactNeed{file, needed, Nil{}})
204	
205	def wal.truncate.synced(
206	  result: Result<&1, &1, U32 & String, Unit>,
207	  state: Db.RotRes
208	) -> IO(WalProgress):
209	  match result:
210	    case Fail{error}:
211	      IO.pure(WalProgress, WalStop{Fail{error}})
212	    case Done{Unit{}}:
213	      IO.pure(WalProgress, WalStop{Done{state}})
214	
215	def wal.truncate.result(
216	  result: Result<&1, &1, U32 & String, Unit>,
217	  +path: String,
218	  state: Db.RotRes
219	) -> IO(WalProgress):
220	  match result:
221	    case Fail{error}:
222	      IO.pure(WalProgress, WalStop{Fail{error}})
223	    case Done{Unit{}}:
224	      do IO<WalProgress>:
225	        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path)
226	        wal.truncate.synced(synced, state)
227	
228	def wal.truncate(
229	  +path: String,
230	  +offset: Nat,
231	  state: Db.RotRes
232	) -> IO(WalProgress):
233	  do IO<WalProgress>:
234	    truncated : Result<&1, &1, U32 & String, Unit> <- Fs.truncate(path, offset)
235	    wal.truncate.result(truncated, path, state)
236	
237	def wal.u32.pair(pair: Bytes.Cursor & Maybe<&2, U32>) -> Maybe<&2, U32>:
238	  match pair:
239	    case (_, value):
240	      value
241	
242	def wal.u32(bytes: Bytes.Bytes) -> Maybe<&2, U32>:
243	  wal.u32.pair(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))
244	
245	def wal.apply(batch: Wal.Batch, state: Db.RotRes) -> Db.RotRes:
246	  match batch state:
247	    case Wal.Batch{muts} Db.Rot{mem, frozen, mem_count, frozen_count}:
248	      Db.apply_and_rotate(muts, mem, frozen, mem_count, frozen_count)
249	
250	def wal.frames.decoded(
251	  result: Result<&1, &1, Wal.Error, Wal.Batch>,
252	  file: File,
253	  +next_offset: Nat,
254	  state: Db.RotRes
255	) -> IO(WalProgress):
256	  match result:
257	    case Fail{_}:
258	      do IO<WalProgress>:
259	        _closed : Unit <- File.close(file)
260	        return WalStop{Fail{(wal_corruption_error_code(), "corrupt WAL frame")}}
261	    case Done{batch}:
262	      IO.pure(WalProgress,
263	        WalMore{file, next_offset, wal.apply(batch, state)})
264	
265	def wal.frames.decode(
266	  pair: File & Result<&1, &1, U32 & String, ExactResult>,
267	  prefix: Bytes.Bytes,
268	  +offset: Nat,
269	  +next_offset: Nat,
270	  state: Db.RotRes,
271	  +path: String
272	) -> IO(WalProgress):
273	  match pair:
274	    case (file, Fail{error}):
275	      do IO<WalProgress>:
276	        _closed : Unit <- File.close(file)
277	        return WalStop{Fail{error}}
278	    case (file, Done{ExactResult{_, False{}}}):
279	      do IO<WalProgress>:
280	        _closed : Unit <- File.close(file)
281	        wal.truncate(path, offset, state)
282	    case (file, Done{ExactResult{body, True{}}}):
283	      wal.frames.decoded(Wal.decode_frame(Bytes.concat([prefix, body])),
284	        file, next_offset, state)
285	
286	def wal.frames.body.bound(
287	  enough: Bool,
288	  file: File,
289	  prefix: Bytes.Bytes,
290	  +frame_size: Nat,
291	  +offset: Nat,
292	  +next_offset: Nat,
293	  state: Db.RotRes,
294	  +path: String
295	) -> IO(WalProgress):
296	  match enough:
297	    case False{}:
298	      do IO<WalProgress>:
299	        _closed : Unit <- File.close(file)
300	        wal.truncate(path, offset, state)
301	    case True{}:
302	      do IO<WalProgress>:
303	        pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, frame_size)
304	        wal.frames.decode(pair, prefix, offset, next_offset, state, path)
305	
306	def wal.frames.body(
307	  file: File,
308	  prefix: Bytes.Bytes,
309	  +frame_size: Nat,
310	  +size: Nat,
311	  +offset: Nat,
312	  state: Db.RotRes,
313	  +path: String
314	) -> IO(WalProgress):
315	  +next_offset = Nat.add(offset, Nat.add(4n, frame_size))
316	  wal.frames.body.bound(Nat.is_le(next_offset, size), file, prefix,
317	    frame_size, offset, next_offset, state, path)
318	
319	def wal.frames.bound(
320	  within: Bool,
321	  file: File,
322	  bytes: Bytes.Bytes,
323	  frame_len: U32,
324	  +size: Nat,
325	  +offset: Nat,
326	  state: Db.RotRes,
327	  +path: String
328	) -> IO(WalProgress):
329	  match within:
330	    case False{}:
331	      do IO<WalProgress>:
332	        _closed : Unit <- File.close(file)
333	        return WalStop{Fail{(wal_corruption_error_code(), "WAL frame length out of bounds")}}
334	    case True{}:
335	      wal.frames.body(file, bytes, U32.to_nat(frame_len), size,
336	        offset, state, path)
337	
338	def wal.frames.length.checked(
339	  enough: Bool,
340	  within: Bool,
341	  file: File,
342	  prefix: Bytes.Bytes,
343	  frame_len: U32,
344	  +size: Nat,
345	  +offset: Nat,
346	  state: Db.RotRes,
347	  +path: String
348	) -> IO(WalProgress):
349	  match enough:
350	    case False{}:
351	      do IO<WalProgress>:
352	        _closed : Unit <- File.close(file)
353	        wal.truncate(path, offset, state)
354	    case True{}:
355	      wal.frames.bound(within, file, prefix, frame_len, size,
356	        offset, state, path)
357	
358	def wal.frames.length.value(
359	  file: File,
360	  maybe_len: Maybe<&2, U32>,
361	  +size: Nat,
362	  +offset: Nat,
363	  state: Db.RotRes,
364	  +path: String
365	) -> IO(WalProgress):
366	  match maybe_len:
367	    case None{}:
368	      do IO<WalProgress>:
369	        _closed : Unit <- File.close(file)
370	        return WalStop{Fail{(wal_corruption_error_code(), "malformed WAL frame length")}}
371	    case Some{+frame_len}:
372	      +next_offset = Nat.add(offset, Nat.add(4n, U32.to_nat(frame_len)))
373	      wal.frames.length.checked(Nat.is_le(next_offset, size),
374	        U32.is_le(36, frame_len) && U32.is_le(frame_len, 1073741860),
375	        file, Wal.u32_bytes(frame_len), frame_len, size, offset,
376	        state, path)
377	
378	def wal.frames.length(
379	  pair: File & Result<&1, &1, U32 & String, ExactResult>,
380	  +size: Nat,
381	  +offset: Nat,
382	  state: Db.RotRes,
383	  +path: String
384	) -> IO(WalProgress):
385	  match pair:
386	    case (file, Fail{error}):
387	      do IO<WalProgress>:
388	        _closed : Unit <- File.close(file)
389	        return WalStop{Fail{error}}
390	    case (file, Done{ExactResult{bytes, complete}}):
391	      match complete:
392	        case False{}:
393	          do IO<WalProgress>:
394	            _closed : Unit <- File.close(file)
395	            wal.truncate(path, offset, state)
396	        case True{}:
397	          wal.frames.length.value(file, wal.u32(bytes),
398	            size, offset, state, path)
399	
400	def wal.frames.remaining(
401	  file: File,
402	  +size: Nat,
403	  +offset: Nat,
404	  state: Db.RotRes,
405	  +path: String,
406	  remaining: Nat
407	) -> IO(WalProgress):
408	  match remaining:
409	    case 0n:
410	      do IO<WalProgress>:
411	        _closed : Unit <- File.close(file)
412	        return WalStop{Done{state}}
413	    case 1n+p:
414	      match p:
415	        case 0n:
416	          do IO<WalProgress>:
417	            _closed : Unit <- File.close(file)
418	            wal.truncate(path, offset, state)
419	        case 1n+q:
420	          match q:
421	            case 0n:
422	              do IO<WalProgress>:
423	                _closed : Unit <- File.close(file)
424	                wal.truncate(path, offset, state)
425	            case 1n+_:
426	              do IO<WalProgress>:
427	                pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 4n)
428	                wal.frames.length(pair, size, offset, state, path)
429	
430	def wal.frames.end(
431	  at_end: Bool,
432	  file: File,
433	  +size: Nat,
434	  +offset: Nat,
435	  state: Db.RotRes,
436	  +path: String
437	) -> IO(WalProgress):
438	  match at_end:
439	    case True{}:
440	      do IO<WalProgress>:
441	        _closed : Unit <- File.close(file)
442	        return WalStop{Done{state}}
443	    case False{}:
444	      wal.frames.remaining(file, size, offset, state, path,
445	        (size - offset : Nat))
446	
447	def wal.frames(
448	  +fuel: Nat,
449	  +size: Nat,
450	  +path: String,
451	  state: WalState
452	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
453	  match fuel:
454	    case 0n:
455	      match state:
456	        case WalAt{file, _, _}:
457	          do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
458	            _closed : Unit <- File.close(file)
459	            return Fail{(U32.from_nat(7n), "WAL frame limit exceeded")}
460	        case WalAfter{WalStop{result}}:
461	          IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, result)
462	        case WalAfter{WalMore{file, _, _}}:
463	          do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
464	            _closed : Unit <- File.close(file)
465	            return Fail{(U32.from_nat(7n), "WAL frame limit exceeded")}
466	    case 1n+rest:
467	      match state:
468	        case WalAt{file, +offset, res}:
469	          do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
470	            progress : WalProgress <- wal.frames.end(
471	              Nat.is_eq(offset, size), file, size, offset, res, path)
472	            wal.frames(rest, size, path, WalAfter{progress})
473	        case WalAfter{WalStop{result}}:
474	          IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, result)
475	        case WalAfter{WalMore{file, offset, res}}:
476	          wal.frames(rest, size, path, WalAt{file, offset, res})
477	
478	def wal.header.decoded(
479	  result: Result<&1, &1, Wal.Error, Unit>,
480	  file: File,
481	  +size: Nat,
482	  +path: String
483	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
484	  match result:
485	    case Fail{_}:
486	      do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
487	        _closed : Unit <- File.close(file)
488	        return Fail{(wal_corruption_error_code(), "unsupported WAL format")}
489	    case Done{Unit{}}:
490	      wal.frames(Nat.add(Nat.div(size, 18n), 4n), size, path,
491	        WalAt{file, 8n, wal.empty()})
492	
493	def wal.header.synced(
494	  result: Result<&1, &1, U32 & String, Unit>
495	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
496	  match result:
497	    case Fail{error}:
498	      IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error})
499	    case Done{Unit{}}:
500	      IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Done{wal.empty()})
501	
502	def wal.header.truncated(
503	  result: Result<&1, &1, U32 & String, Unit>,
504	  +path: String
505	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
506	  match result:
507	    case Fail{error}:
508	      IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error})
509	    case Done{Unit{}}:
510	      do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
511	        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path)
512	        wal.header.synced(synced)
513	
514	def wal.header(
515	  pair: File & Result<&1, &1, U32 & String, ExactResult>,
516	  +size: Nat,
517	  +path: String
518	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
519	  match pair:
520	    case (file, Fail{error}):
521	      do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
522	        _closed : Unit <- File.close(file)
523	        return Fail{error}
524	    case (file, Done{ExactResult{_, False{}}}):
525	      do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
526	        _closed : Unit <- File.close(file)
527	        truncated : Result<&1, &1, U32 & String, Unit> <- Fs.truncate(path, 0n)
528	        wal.header.truncated(truncated, path)
529	    case (file, Done{ExactResult{bytes, True{}}}):
530	      wal.header.decoded(Wal.decode_header(bytes), file, size, path)
531	
532	def wal.open.file(
533	  size: Nat,
534	  opened: Result<&1, &1, U32 & String, File>,
535	  +path: String
536	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
537	  match opened:
538	    case Fail{error}:
539	      IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error})
540	    case Done{file}:
541	      do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
542	        pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 8n)
543	        wal.header(pair, size, path)
544	
545	def wal.open.size(
546	  result: Result<&1, &1, U32 & String, Nat>,
547	  +path: String
548	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
549	  match result:
550	    case Fail{error}:
551	      IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error})
552	    case Done{size}:
553	      do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
554	        opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
555	        wal.open.file(size, opened, path)
556	
557	def wal.open.present(
558	  present: Bool,
559	  +path: String
560	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
561	  match present:
562	    case False{}:
563	      IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Done{wal.empty()})
564	    case True{}:
565	      do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
566	        size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path)
567	        wal.open.size(size, path)
568	
569	def wal_open.exists(
570	  result: Result<&1, &1, U32 & String, Bool>,
571	  +dir: String
572	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
573	  match result:
574	    case Fail{error}:
575	      IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error})
576	    case Done{present}:
577	      wal.open.present(present, Db.wal_path(dir))
578	
579	# Handle wal open in the database recovery effects.
580	def wal_open(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
581	  do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
582	    present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(Db.wal_path(dir))
583	    wal_open.exists(present, dir)
584	
585	# --- Tmp sweep (best-effort) ---
586	
587	# Handle tmp pick in the database recovery effects.
588	def tmp_pick(keep: Bool, +name: String, +acc: List<&2, String>) -> List<&2, String>:
589	  match keep:
590	    case True{}:
591	      Con{name, acc}
592	    case False{}:
593	      acc
594	
595	# Handle tmp some in the database recovery effects.
596	def tmp_some(+ok: Bool, +name: String) -> Maybe<&2, String>:
597	  match ok:
598	    case True{}:
599	      Some{name}
600	    case False{}:
601	      None{}
602	
603	# Handle tmp map list in the database recovery effects.
604	def tmp_map_list(xs: List<&1, String>) -> List<&2, Maybe<&2, String>>:
605	  match xs:
606	    case Nil{}:
607	      Nil{}
608	    case Con{+h, t}:
609	      Con{tmp_some(String.ends_with(h, ".tmp"), h), tmp_map_list(t)}
610	
611	# Handle cat go in the database recovery effects.
612	def cat_go(xs: List<&2, Maybe<&2, String>>) -> List<&2, String>:
613	  match xs:
614	    case Nil{}:
615	      Nil{}
616	    case Con{h, t}:
617	      match h:
618	        case None{}:
619	          cat_go(t)
620	        case Some{s}:
621	          Con{s, cat_go(t)}
622	
623	# Handle tmp names in the database recovery effects.
624	def tmp_names(xs: List<&1, String>) -> List<&2, String>:
625	  cat_go(tmp_map_list(xs))
626	
627	# Handle sweep one in the database recovery effects.
628	def sweep_one(res: Result<&1, &1, U32 & String, List<String>>, +dir: String) -> IO(Unit):
629	  match res:
630	    case Fail{e}:
631	      IO.pure(Unit, Unit{})
632	    case Done{ns}:
633	      do IO<Unit>:
634	        _ign : Result<&1, &1, U32 & String, Unit> <- CompactIo.remove_list(Compact.prefix(tmp_names(ns), dir))
635	        IO.pure(Unit, Unit{})
636	
637	# Handle sweep get in the database recovery effects.
638	def sweep_get(+dir: String, +sub: String) -> IO(Unit):
639	  do IO<Unit>:
640	    r : Result<&1, &1, U32 & String, List<String>> <- Fs.read_dir(dir ++ "/" ++ sub)
641	    sweep_one(r, dir ++ "/" ++ sub)
642	
643	# Handle sweep cnt in the database recovery effects.
644	def sweep_cnt(res: Result<&1, &1, U32 & String, Nat>, +dir: String, +sub: String) -> IO(Unit):
645	  match res:
646	    case Fail{e}:
647	      IO.pure(Unit, Unit{})
648	    case Done{n}:
649	      sweep_get(dir, sub)
650	
651	# Handle sweep dir in the database recovery effects.
652	def sweep_dir(+dir: String, +sub: String) -> IO(Unit):
653	  do IO<Unit>:
654	    c : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(dir ++ "/" ++ sub)
655	    sweep_cnt(c, dir, sub)
656	
657	# --- Open assembly ---
658	
659	def wal_part.initialized(
660	  result: Result<&1, &1, U32 & String, Unit>,
661	  +dir: String
662	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
663	  match result:
664	    case Fail{error}:
665	      IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error})
666	    case Done{Unit{}}:
667	      wal_open(dir)
668	
669	# Splits the WAL replay stream at the recovery boundary.
670	def wal_part(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
671	  do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
672	    initialized : Result<&1, &1, U32 & String, Unit> <- DbIo.wal_initialize(dir)
673	    wal_part.initialized(initialized, dir)
674	
675	# Open assemble for the database recovery effects.
676	def open_assemble(
677	  res: Db.RotRes,
678	  +dir: String,
679	  +mfst: Manifest.Manifest,
680	  levels: List<&2, List<&2, Sstable.Table>>
681	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
682	  match res:
683	    case Db.Rot{mem, frozen, mc, fc}:
684	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
685	        return Done{Db.Db{dir, mem, frozen, Db.default_batch_cap(), Nil{}, levels, Pure.count_gen(Pure.mfst_lists(mfst)), Manifest.token(mfst), mc, fc}}
686	
687	def open_levels.wal(
688	  result: Result<&1, &1, U32 & String, Db.RotRes>,
689	  +dir: String,
690	  +mfst: Manifest.Manifest,
691	  levels: List<&2, List<&2, Sstable.Table>>
692	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
693	  match result:
694	    case Fail{error}:
695	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
696	    case Done{recovered}:
697	      open_assemble(recovered, dir, mfst, levels)
698	
699	# Loads Manifest levels and checks WAL recovery errors.
700	def open_levels_checked(
701	  ok: Bool,
702	  +dir: String,
703	  +mfst: Manifest.Manifest,
704	  levels: List<&2, List<&2, Sstable.Table>>
705	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
706	  match ok:
707	    case False{}:
708	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(6n), "overlapping tables in L1+ manifest")})
709	    case True{}:
710	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
711	        recovered : Result<&1, &1, U32 & String, Db.RotRes> <- wal_part(dir)
712	        open_levels.wal(recovered, dir, mfst, levels)
713	
714	def open_levels.loaded(
715	  result: Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>,
716	  +dir: String,
717	  +mfst: Manifest.Manifest
718	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
719	  match result:
720	    case Fail{error}:
721	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
722	    case Done{+levels}:
723	      open_levels_checked(Pure.l1_plus_disjoint(levels), dir, mfst, levels)
724	
725	def open_levels.fuel(+mfst: Manifest.Manifest) -> Nat:
726	  Nat.add(Nat.mul(3n, List.length(&2, List<&2, String>, Pure.mfst_lists(mfst))), 1n)
727	
728	# Opens levels through the legacy recovery result path.
729	def open_levels(+dir: String, +mfst: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Db.Db>):
730	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
731	    levels : Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>> <- load_levels(open_levels.fuel(mfst), LoadLevelsNames{Pure.mfst_lists(mfst), 0n, dir, Nil{}})
732	    open_levels.loaded(levels, dir, mfst)
733	
734	# Open names go for the database recovery effects.
735	def open_names_go(ok: Bool, +dir: String, +mfst: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Db.Db>):
736	  match ok:
737	    case True{}:
738	      open_levels(dir, mfst)
739	    case False{}:
740	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(5n), "bad manifest names")})
741	
742	# Handle sweep root go in the database recovery effects.
743	def sweep_root_go(+dir: String) -> IO(Unit):
744	  do IO<Unit>:
745	    r : Result<&1, &1, U32 & String, List<String>> <- Fs.read_dir(dir)
746	    sweep_one(r, dir)
747	
748	# Handle sweep root cnt in the database recovery effects.
749	def sweep_root_cnt(res: Result<&1, &1, U32 & String, Nat>, +dir: String) -> IO(Unit):
750	  match res:
751	    case Fail{e}:
752	      IO.pure(Unit, Unit{})
753	    case Done{n}:
754	      sweep_root_go(dir)
755	
756	# Handle sweep root in the database recovery effects.
757	def sweep_root(+dir: String) -> IO(Unit):
758	  do IO<Unit>:
759	    c : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(dir)
760	    sweep_root_cnt(c, dir)
761	
762	def open_db.swept(+dir: String, +mfst: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Db.Db>):
763	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
764	    _s0 : Unit <- sweep_dir(dir, "l0")
765	    _s1 : Unit <- sweep_dir(dir, "l1")
766	    _s2 : Unit <- sweep_dir(dir, "l2")
767	    _s3 : Unit <- sweep_dir(dir, "l3")
768	    _s4 : Unit <- sweep_root(dir)
769	    open_names_go(Pure.manifest_names_ok(Pure.mfst_lists(mfst), 0n), dir, mfst)
770	
771	def open_db.manifest(
772	  result: Result<&1, &1, U32 & String, Manifest.Manifest>,
773	  +dir: String,
774	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
775	  match result:
776	    case Fail{error}:
777	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
778	    case Done{mfst}:
779	      open_db.swept(dir, mfst)
780	
781	def open_db.ensured(
782	  ensured: Result<&1, &1, U32 & String, Unit>,
783	  +dir: String
784	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
785	  match ensured:
786	    case Fail{error}:
787	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
788	    case Done{Unit{}}:
789	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
790	        manifest : Result<&1, &1, U32 & String, Manifest.Manifest> <- Flush.load_mfst(dir ++ "/MANIFEST")
791	        open_db.manifest(manifest, dir)
792	
793	# Opens the legacy database and applies best-effort recovery cleanup.
794	def open_db(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
795	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
796	    ensured : Result<&1, &1, U32 & String, Unit> <- Flush.ensure_dir(dir)
797	    open_db.ensured(ensured, dir)
798	
799	def sweep_checked_path.select(root: Bool, +dir: String, +subdir: String) -> String:
800	  match root:
801	    case True{}:
802	      dir
803	    case False{}:
804	      dir ++ "/" ++ subdir
805	
806	# Removes one temporary path and propagates removal failures.
807	def sweep_checked_path(+dir: String, +subdir: String) -> String:
808	  sweep_checked_path.select(String.eq(subdir, "."), dir, subdir)
809	
810	# Removes every temporary path with checked errors.
811	def sweep_checked_remove_list(+names: List<&2, String>) -> IO(Result<&1, &1, U32 & String, Unit>):
812	  CompactIo.remove_list_strict_start((
813	    Nat.add(Nat.mul(2n, List.length(&2, String, names)), 2n), names
814	  ))
815	
816	# Represents sweepreadstate data in recover.
817	type SweepReadState is Type:
818	  SweepReadNeed{path: String, idx: Nat, remaining: Nat, names: List<String>}
819	  SweepReadResult{result: Result<&1, &1, U32 & String, String>, path: String, idx: Nat, remaining: Nat, names: List<String>}
820	
821	# Reads a directory listing through checked indexed IO.
822	def sweep_read_checked(fuel: Nat, state: SweepReadState) -> IO(Result<&1, &1, U32 & String, List<String>>):
823	  match fuel:
824	    case 0n:
825	      IO.pure(Result<&1, &1, U32 & String, List<String>>, Fail{(7, "recovery directory read limit exceeded")})
826	    case 1n+rest:
827	      match state:
828	        case SweepReadNeed{+path, +idx, remaining, names}:
829	          match remaining:
830	            case 0n:
831	              IO.pure(Result<&1, &1, U32 & String, List<String>>,
832	                Done{List.reverse(&1, String, names)})
833	            case 1n+more:
834	              do IO<Result<&1, &1, U32 & String, List<String>>>:
835	                name : Result<&1, &1, U32 & String, String> <- Fs.read_dir_at(path, idx)
836	                sweep_read_checked(rest, SweepReadResult{name, path, Nat.add(idx, 1n), more, names})
837	        case SweepReadResult{result, path, idx, remaining, names}:
838	          match result:
839	            case Fail{error}:
840	              IO.pure(Result<&1, &1, U32 & String, List<String>>, Fail{error})
841	            case Done{name}:
842	              sweep_read_checked(rest, SweepReadNeed{path, idx, remaining,
843	                Fs.grab_name(name, names)})
844	
845	# Handles the checked directory-entry count result.
846	def sweep_read_checked_count(
847	  result: Result<&1, &1, U32 & String, Nat>,
848	  +path: String
849	) -> IO(Result<&1, &1, U32 & String, List<String>>):
850	  match result:
851	    case Fail{error}:
852	      IO.pure(Result<&1, &1, U32 & String, List<String>>, Fail{error})
853	    case Done{+count}:
854	      sweep_read_checked(Nat.add(Nat.mul(2n, Nat.add(count, 1n)), 1n),
855	        SweepReadNeed{path, 0n, Nat.add(count, 1n), Nil{}})
856	
857	# Reads one indexed directory entry and preserves errors.
858	def sweep_read_checked_path(+path: String) -> IO(Result<&1, &1, U32 & String, List<String>>):
859	  do IO<Result<&1, &1, U32 & String, List<String>>>:
860	    count : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(path)
861	    sweep_read_checked_count(count, path)
862	
863	# Converts a directory-read failure to the recovery result.
864	def sweep_checked_read_error(
865	  allow_missing: Bool,
866	  +code: U32,
867	  message: String
868	) -> IO(Result<&1, &1, U32 & String, Unit>):
869	  IO.pure(Result<&1, &1, U32 & String, Unit>,
870	    FsPolicy.optional_directory_result(allow_missing,
871	      U32.is_eq(code, 2) || U32.is_eq(code, 20), code, message))
872	
873	def sweep_checked_dir.result(
874	  result: Result<&1, &1, U32 & String, List<String>>,
875	  allow_missing: Bool,
876	  +path: String
877	) -> IO(Result<&1, &1, U32 & String, Unit>):
878	  match result:
879	    case Fail{(+code, message)}:
880	      sweep_checked_read_error(allow_missing, code, message)
881	    case Done{names}:
882	      sweep_checked_remove_list(Compact.prefix(tmp_names(names), path))
883	
884	# Sweeps temporary files from one optional level directory.
885	def sweep_checked_dir(+dir: String, +subdir: String) -> IO(Result<&1, &1, U32 & String, Unit>):
886	  +path = sweep_checked_path(dir, subdir)
887	  do IO<Result<&1, &1, U32 & String, Unit>>:
888	    listed : Result<&1, &1, U32 & String, List<String>> <- sweep_read_checked_path(path)
889	    sweep_checked_dir.result(listed, Bool.not(String.eq(subdir, ".")), path)
890	
891	# Represents sweepcheckedstate data in recover.
892	type SweepCheckedState is Type:
893	  SweepCheckedDirs{dirs: List<&2, String>, dir: String}
894	  SweepCheckedResult{result: Result<&1, &1, U32 & String, Unit>, dirs: List<&2, String>, dir: String}
895	
896	# Sweeps temporary files from all optional level directories.
897	def sweep_checked_dirs(
898	  fuel: Nat,
899	  state: SweepCheckedState
900	) -> IO(Result<&1, &1, U32 & String, Unit>):
901	  match fuel:
902	    case 0n:
903	      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{(U32.from_nat(7n), "recovery sweep limit exceeded")})
904	    case 1n+rest:
905	      match state:
906	        case SweepCheckedDirs{dirs, +dir}:
907	          match dirs:
908	            case Nil{}:
909	              IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
910	            case Con{subdir, tail}:
911	              do IO<Result<&1, &1, U32 & String, Unit>>:
912	                swept : Result<&1, &1, U32 & String, Unit> <- sweep_checked_dir(dir, subdir)
913	                sweep_checked_dirs(rest, SweepCheckedResult{swept, tail, dir})
914	        case SweepCheckedResult{result, dirs, +dir}:
915	          match result:
916	            case Fail{error}:
917	              IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
918	            case Done{Unit{}}:
919	              sweep_checked_dirs(rest, SweepCheckedDirs{dirs, dir})
920	
921	# Runs the checked temporary-file sweep across database levels.
922	def sweep_checked(+dir: String) -> IO(Result<&1, &1, U32 & String, Unit>):
923	  sweep_checked_dirs(11n,
924	    SweepCheckedDirs{Con{"l0", Con{"l1", Con{"l2", Con{"l3", Con{".", Nil{}}}}}}, dir})
925	
926	# Continues existing-open recovery after the checked sweep.
927	def open_existing_sweep_result(
928	  result: Result<&1, &1, U32 & String, Unit>,
929	  +dir: String,
930	  +manifest: Manifest.Manifest
931	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
932	  match result:
933	    case Fail{error}:
934	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
935	    case Done{Unit{}}:
936	      open_names_go(Pure.manifest_names_ok(Pure.mfst_lists(manifest), 0n), dir, manifest)
937	
938	# Loads the Manifest before opening existing database levels.
939	def open_existing_manifest(
940	  result: Result<&1, &1, U32 & String, Manifest.Manifest>,
941	  +dir: String
942	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
943	  match result:
944	    case Fail{error}:
945	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
946	    case Done{manifest}:
947	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
948	        swept : Result<&1, &1, U32 & String, Unit> <- sweep_checked(dir)
949	        open_existing_sweep_result(swept, dir, manifest)
950	
951	# Recovers an existing database and propagates filesystem errors.
952	def open_existing_checked(
953	  manifest_present: Bool,
954	  +dir: String
955	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
956	  match manifest_present:
957	    case False{}:
958	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(2n), "manifest missing")})
959	    case True{}:
960	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
961	        manifest : Result<&1, &1, U32 & String, Manifest.Manifest> <- Flush.load_mfst(dir ++ "/MANIFEST")
962	        open_existing_manifest(manifest, dir)
963	
964	def open_existing.present(
965	  result: Result<&1, &1, U32 & String, Bool>,
966	  +dir: String
967	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
968	  match result:
969	    case Fail{error}:
970	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
971	    case Done{present}:
972	      open_existing_checked(present, dir)
973	
974	# Opens an existing database while the caller retains its lock.
975	def open_existing_locked(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
976	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
977	    present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(dir ++ "/MANIFEST")
978	    open_existing.present(present, dir)
979	
980	# Initializes the database state after creating its directories.
981	def create_initialized(
982	  synced: Result<&1, &1, U32 & String, Unit>,
983	  +dir: String,
984	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
985	  match synced:
986	    case Fail{error}:
987	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
988	    case Done{Unit{}}:
989	      open_existing_checked(True{}, dir)
990	
991	# Writes and publishes the initial Manifest.
992	def create_published(
993	  renamed: Result<&1, &1, U32 & String, Unit>,
994	  +dir: String,
995	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
996	  match renamed:
997	    case Fail{error}:
998	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
999	    case Done{Unit{}}:
1000	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
1001	        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(dir)
1002	        create_initialized(synced, dir)
1003	
1004	# Initializes the WAL after publishing the Manifest.
1005	def create_written(
1006	  written: Result<&1, &1, U32 & String, Unit>,
1007	  +dir: String,
1008	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1009	  match written:
1010	    case Fail{error}:
1011	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
1012	    case Done{Unit{}}:
1013	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
1014	        renamed : Result<&1, &1, U32 & String, Unit> <- Fs.rename(dir ++ "/MANIFEST.tmp", dir ++ "/MANIFEST")
1015	        create_published(renamed, dir)
1016	
1017	# Creates a new database without acquiring another lock.
1018	def create_locked(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1019	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
1020	    written : Result<&1, &1, U32 & String, Unit> <- Flush.write_manifest(dir ++ "/MANIFEST.tmp", Manifest.M{Nil{}})
1021	    create_written(written, dir)
1022	
1023	# --- Maintenance (synchronous auto flush/compact + stall) ---
1024	# One drain round always suffices (L0 >= 8 > 4, so the gated compaction
1025	# fires and empties L0) — no loop, no cycle.
1026	
1027	# Flush dec for the database recovery effects.
1028	def flush_dec(full: Bool, +db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1029	  match full:
1030	    case True{}:
1031	      Flush.flush(db)
1032	    case False{}:
1033	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Done{db})
1034	
1035	# Flush gate for the database recovery effects.
1036	def flush_gate(db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1037	  match db:
1038	    case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, +mem_count, frozen_count}:
1039	      flush_dec(Nat.is_lt(4096n, mem_count), Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count})
1040	
1041	# Handle l0 full in the database recovery effects.
1042	def l0_full(db: Db.Db) -> Bool:
1043	  match db:
1044	    case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
1045	      Nat.is_lt(7n, List.length(&2, Sstable.Table, Compact.l0list(levels)))
1046	
1047	# Flushes one full MemTable and returns the updated database.
1048	def drain_once.flushed(
1049	  result: Result<&1, &1, U32 & String, Db.Db>
1050	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1051	  match result:
1052	    case Fail{error}:
1053	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
1054	    case Done{db}:
1055	      CompactIo.compact(db)
1056	
1057	# Flushes one full MemTable and returns the updated database.
1058	def drain_once(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1059	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
1060	    flushed : Result<&1, &1, U32 & String, Db.Db> <- Flush.flush(db)
1061	    drain_once.flushed(flushed)
1062	
1063	# Handle drain pick in the database recovery effects.
1064	def drain_pick(full: Bool, +db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1065	  match full:
1066	    case True{}:
1067	      drain_once(db)
1068	    case False{}:
1069	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Done{db})
1070	
1071	# Handle drain go in the database recovery effects.
1072	def drain_go(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1073	  drain_pick(l0_full(db), db)
1074	
1075	def maintain.flushed(
1076	  result: Result<&1, &1, U32 & String, Db.Db>
1077	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1078	  match result:
1079	    case Fail{error}:
1080	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
1081	    case Done{db}:
1082	      CompactIo.compact(db)
1083	
1084	def maintain.drained(
1085	  result: Result<&1, &1, U32 & String, Db.Db>
1086	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1087	  match result:
1088	    case Fail{error}:
1089	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
1090	    case Done{db}:
1091	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
1092	        flushed : Result<&1, &1, U32 & String, Db.Db> <- flush_gate(db)
1093	        maintain.flushed(flushed)
1094	
1095	# Runs the required flush and compaction maintenance.
1096	def maintain(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1097	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
1098	    drained : Result<&1, &1, U32 & String, Db.Db> <- drain_go(db)
1099	    maintain.drained(drained)
1100	
1101	# Handle maintain unwrap in the database recovery effects.
1102	def maintain_unwrap(res: Result<&1, &1, U32 & String, Db.Db>) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1103	  match res:
1104	    case Fail{e}:
1105	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{e})
1106	    case Done{db}:
1107	      maintain(db)
1108	
1109	# Handle write in the database recovery effects.
1110	def write(+db: Db.Db, +batch: Wal.Batch) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1111	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
1112	    r : Result<&1, &1, U32 & String, Db.Db> <- DbIo.db_write(db, batch)
1113	    maintain_unwrap(r)
1114	
1115	# Handle put in the database recovery effects.
1116	def put(+db: Db.Db, +key: String, +val: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1117	  write(db, Wal.Batch{Con{Wal.Put{key, val}, Nil{}}})
1118	
1119	# Handle del in the database recovery effects.
1120	def del(+db: Db.Db, +key: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1121	  write(db, Wal.Batch{Con{Wal.Del{key}, Nil{}}})
1122	
1123	# --- Closed-vector laws live in laws/Recover.bend (spec law 7) ---