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 ./Db.bend as Db
13	import ./DbIo.bend as DbIo
14	import ./RecoverPure.bend as Pure
15	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
16	
17	# Represent ExactResult data used by the database recovery effects.
18	type ExactResult is Type:
19	  ExactResult{bytes: Bytes.Bytes, complete: Bool}
20	
21	# Represent ExactState data used by the database recovery effects.
22	type ExactState is Type:
23	  ExactNeed{file: File, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
24	  ExactRead{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
25	
26	# Represent WalProgress data used by the database recovery effects.
27	type WalProgress is Type:
28	  WalMore{file: File, offset: Nat, state: Db.RotRes}
29	  WalStop{result: Result<&1, &1, U32 & String, Db.RotRes>}
30	
31	# Represent WalState data used by the database recovery effects.
32	type WalState is Type:
33	  WalAt{file: File, offset: Nat, state: Db.RotRes}
34	  WalAfter{progress: WalProgress}
35	
36	def exact.max(large: Bool, +remaining: Nat) -> U32:
37	  match large:
38	    case True{}:
39	      1048576
40	    case False{}:
41	      U32.from_nat(remaining)
42	
43	def exact.result(
44	  file: File,
45	  chunks: List<&1, Bytes.Bytes>,
46	  complete: Bool
47	) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
48	  IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
49	    (file, Done{ExactResult{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks)), complete}}))
50	
51	def exact.loop(
52	  fuel: Nat,
53	  state: ExactState
54	) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
55	  match fuel:
56	    case 0n:
57	      match state:
58	        case ExactNeed{file, remaining, chunks}:
59	          match remaining:
60	            case 0n:
61	              exact.result(file, chunks, True{})
62	            case 1n+_:
63	              IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
64	                (file, Fail{(U32.from_nat(7n), "bounded read exhausted")}))
65	        case ExactRead{pair, _, _}:
66	          match pair:
67	            case (file, Fail{error}):
68	              IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{error}))
69	            case (file, Done{_}):
70	              IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
71	                (file, Fail{(U32.from_nat(7n), "bounded read exhausted")}))
72	    case 1n+rest:
73	      match state:
74	        case ExactNeed{file, +remaining, chunks}:
75	          match remaining:
76	            case 0n:
77	              exact.result(file, chunks, True{})
78	            case 1n+_:
79	              do IO<File & Result<&1, &1, U32 & String, ExactResult>>:
80	                next : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
81	                  Fs.read_bytes(file, exact.max(Nat.is_le(1048577n, remaining), remaining))
82	                exact.loop(rest, ExactRead{next, remaining, chunks})
83	        case ExactRead{pair, remaining, chunks}:
84	          match pair:
85	            case (file, Fail{error}):
86	              IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{error}))
87	            case (file, Done{Bytes.Bytes{0, _}}):
88	              exact.result(file, chunks, False{})
89	            case (file, Done{Bytes.Bytes{+len, buf}}):
90	              exact.loop(rest, ExactNeed{file,
91	                (remaining - U32.to_nat(len) : Nat),
92	                Con{Bytes.Bytes{len, buf}, chunks}})
93	
94	# Recovery + maintenance (Task 10).
95	#
96	# Open: ensure dir; sweep *.tmp (best-effort); load manifest (absent =
97	# fresh; corrupt = fatal); shape-check all names; load tables positionally
98	# (listed-but-missing/corrupt = fatal, never silent); restore the flush
99	# counter from the greatest table-name generation; replay + truncate
100	# the WAL; return the Db.
101	#
102	# Maintenance (synchronous, observably identical ordering to background):
103	# after every acked batch: drain while L0 >= 8 (stall = blocking
104	# maintenance), flush when mem >= 4096, compact (self-gated at L0 > 4).
105	# True fork-based background (frozen mem + IO.spawn) is a benchmark-gated
106	# follow-up; ordering and stall semantics are already exact.
107	
108	# --- Loaders (self-recursive IO; fatal via IO.die, never silent) ---
109	
110	
111	# Handle load tables in the database recovery effects.
112	def load_tables(
113	  names: List<&2, String>,
114	  +ldir: String,
115	  +dir: String,
116	  +acc: List<&2, Sstable.Table>
117	) -> IO(List<&2, Sstable.Table>):
118	  match names:
119	    case Nil{}:
120	      IO.pure(List<&2, Sstable.Table>, List.reverse(&2, Sstable.Table, acc))
121	    case Con{+n, t}:
122	      do IO<List<&2, Sstable.Table>>:
123	        tbl : Sstable.Table <- IO.try(Sstable.Table, SstStreamIo.read_table(dir ++ "/" ++ ldir ++ "/" ++ n))
124	        load_tables(t, ldir, dir, Con{tbl, acc})
125	
126	# Handle ldir io in the database recovery effects.
127	def ldir_io(+opt: Maybe<&2, String>) -> IO(String):
128	  match opt:
129	    case None{}:
130	      IO.die(String, U32.from_nat(4n), "too many levels")
131	    case Some{d}:
132	      IO.pure(String, d)
133	
134	# Handle ldir of in the database recovery effects.
135	def ldir_of(idx: Nat) -> Maybe<&2, String>:
136	  match idx:
137	    case 0n:
138	      Some{"l0"}
139	    case 1n+m:
140	      match m:
141	        case 0n:
142	          Some{"l1"}
143	        case 1n+p:
144	          match p:
145	            case 0n:
146	              Some{"l2"}
147	            case 1n+q:
148	              match q:
149	                case 0n:
150	                  Some{"l3"}
151	                case 1n+r:
152	                  None{}
153	
154	# Handle load levels in the database recovery effects.
155	def load_levels(
156	  lvls: List<&2, List<&2, String>>,
157	  +idx: Nat,
158	  +dir: String,
159	  +acc: List<&2, List<&2, Sstable.Table>>
160	) -> IO(List<&2, List<&2, Sstable.Table>>):
161	  match lvls:
162	    case Nil{}:
163	      IO.pure(List<&2, List<&2, Sstable.Table>>, List.reverse(&2, List<&2, Sstable.Table>, acc))
164	    case Con{ns, t}:
165	      do IO<List<&2, List<&2, Sstable.Table>>>:
166	        ld : String <- ldir_io(ldir_of(idx))
167	        tabs : List<&2, Sstable.Table> <- load_tables(ns, ld, dir, Nil{})
168	        load_levels(t, Nat.add(idx, 1n), dir, Con{tabs, acc})
169	
170	def wal.empty() -> Db.RotRes:
171	  Db.Rot{MemTable.empty(), MemTable.empty(), 0n, 0n}
172	
173	def exact.start(file: File, +needed: Nat) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
174	  exact.loop(Nat.add(Nat.mul(2n, needed), 1n), ExactNeed{file, needed, Nil{}})
175	
176	def wal.truncate(
177	  +path: String,
178	  +offset: Nat,
179	  state: Db.RotRes
180	) -> IO(WalProgress):
181	  do IO<WalProgress>:
182	    _truncated : Unit <- IO.try(Unit, Fs.truncate(path, offset))
183	    _synced : Unit <- IO.try(Unit, Fs.fsync(path))
184	    return WalStop{Done{state}}
185	
186	def wal.u32.pair(pair: Bytes.Cursor & Maybe<&2, U32>) -> Maybe<&2, U32>:
187	  match pair:
188	    case (_, value):
189	      value
190	
191	def wal.u32(bytes: Bytes.Bytes) -> Maybe<&2, U32>:
192	  wal.u32.pair(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))
193	
194	def wal.apply(batch: Wal.Batch, state: Db.RotRes) -> Db.RotRes:
195	  match batch state:
196	    case Wal.Batch{muts} Db.Rot{mem, frozen, mem_count, frozen_count}:
197	      Db.apply_and_rotate(muts, mem, frozen, mem_count, frozen_count)
198	
199	def wal.frames.decoded(
200	  result: Result<&1, &1, Wal.Error, Wal.Batch>,
201	  file: File,
202	  +next_offset: Nat,
203	  state: Db.RotRes
204	) -> IO(WalProgress):
205	  match result:
206	    case Fail{_}:
207	      do IO<WalProgress>:
208	        _closed : Unit <- File.close(file)
209	        return WalStop{Fail{(U32.from_nat(3n), "corrupt WAL frame")}}
210	    case Done{batch}:
211	      IO.pure(WalProgress,
212	        WalMore{file, next_offset, wal.apply(batch, state)})
213	
214	def wal.frames.decode(
215	  pair: File & Result<&1, &1, U32 & String, ExactResult>,
216	  prefix: Bytes.Bytes,
217	  +offset: Nat,
218	  +next_offset: Nat,
219	  state: Db.RotRes,
220	  +path: String
221	) -> IO(WalProgress):
222	  match pair:
223	    case (file, Fail{error}):
224	      do IO<WalProgress>:
225	        _closed : Unit <- File.close(file)
226	        return WalStop{Fail{error}}
227	    case (file, Done{ExactResult{_, False{}}}):
228	      do IO<WalProgress>:
229	        _closed : Unit <- File.close(file)
230	        wal.truncate(path, offset, state)
231	    case (file, Done{ExactResult{body, True{}}}):
232	      wal.frames.decoded(Wal.decode_frame(Bytes.concat([prefix, body])),
233	        file, next_offset, state)
234	
235	def wal.frames.body.bound(
236	  enough: Bool,
237	  file: File,
238	  prefix: Bytes.Bytes,
239	  +frame_size: Nat,
240	  +offset: Nat,
241	  +next_offset: Nat,
242	  state: Db.RotRes,
243	  +path: String
244	) -> IO(WalProgress):
245	  match enough:
246	    case False{}:
247	      do IO<WalProgress>:
248	        _closed : Unit <- File.close(file)
249	        wal.truncate(path, offset, state)
250	    case True{}:
251	      do IO<WalProgress>:
252	        pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, frame_size)
253	        wal.frames.decode(pair, prefix, offset, next_offset, state, path)
254	
255	def wal.frames.body(
256	  file: File,
257	  prefix: Bytes.Bytes,
258	  +frame_size: Nat,
259	  +size: Nat,
260	  +offset: Nat,
261	  state: Db.RotRes,
262	  +path: String
263	) -> IO(WalProgress):
264	  +next_offset = Nat.add(offset, Nat.add(4n, frame_size))
265	  wal.frames.body.bound(Nat.is_le(next_offset, size), file, prefix,
266	    frame_size, offset, next_offset, state, path)
267	
268	def wal.frames.bound(
269	  within: Bool,
270	  file: File,
271	  bytes: Bytes.Bytes,
272	  frame_len: U32,
273	  +size: Nat,
274	  +offset: Nat,
275	  state: Db.RotRes,
276	  +path: String
277	) -> IO(WalProgress):
278	  match within:
279	    case False{}:
280	      do IO<WalProgress>:
281	        _closed : Unit <- File.close(file)
282	        return WalStop{Fail{(U32.from_nat(3n), "WAL frame length out of bounds")}}
283	    case True{}:
284	      wal.frames.body(file, bytes, U32.to_nat(frame_len), size,
285	        offset, state, path)
286	
287	def wal.frames.length.checked(
288	  enough: Bool,
289	  within: Bool,
290	  file: File,
291	  prefix: Bytes.Bytes,
292	  frame_len: U32,
293	  +size: Nat,
294	  +offset: Nat,
295	  state: Db.RotRes,
296	  +path: String
297	) -> IO(WalProgress):
298	  match enough:
299	    case False{}:
300	      do IO<WalProgress>:
301	        _closed : Unit <- File.close(file)
302	        wal.truncate(path, offset, state)
303	    case True{}:
304	      wal.frames.bound(within, file, prefix, frame_len, size,
305	        offset, state, path)
306	
307	def wal.frames.length.value(
308	  file: File,
309	  maybe_len: Maybe<&2, U32>,
310	  +size: Nat,
311	  +offset: Nat,
312	  state: Db.RotRes,
313	  +path: String
314	) -> IO(WalProgress):
315	  match maybe_len:
316	    case None{}:
317	      do IO<WalProgress>:
318	        _closed : Unit <- File.close(file)
319	        return WalStop{Fail{(U32.from_nat(3n), "malformed WAL frame length")}}
320	    case Some{+frame_len}:
321	      +next_offset = Nat.add(offset, Nat.add(4n, U32.to_nat(frame_len)))
322	      wal.frames.length.checked(Nat.is_le(next_offset, size),
323	        U32.is_le(36, frame_len) && U32.is_le(frame_len, 1073741860),
324	        file, Wal.u32_bytes(frame_len), frame_len, size, offset,
325	        state, path)
326	
327	def wal.frames.length(
328	  pair: File & Result<&1, &1, U32 & String, ExactResult>,
329	  +size: Nat,
330	  +offset: Nat,
331	  state: Db.RotRes,
332	  +path: String
333	) -> IO(WalProgress):
334	  match pair:
335	    case (file, Fail{error}):
336	      do IO<WalProgress>:
337	        _closed : Unit <- File.close(file)
338	        return WalStop{Fail{error}}
339	    case (file, Done{ExactResult{bytes, complete}}):
340	      match complete:
341	        case False{}:
342	          do IO<WalProgress>:
343	            _closed : Unit <- File.close(file)
344	            wal.truncate(path, offset, state)
345	        case True{}:
346	          wal.frames.length.value(file, wal.u32(bytes),
347	            size, offset, state, path)
348	
349	def wal.frames.remaining(
350	  file: File,
351	  +size: Nat,
352	  +offset: Nat,
353	  state: Db.RotRes,
354	  +path: String,
355	  remaining: Nat
356	) -> IO(WalProgress):
357	  match remaining:
358	    case 0n:
359	      do IO<WalProgress>:
360	        _closed : Unit <- File.close(file)
361	        return WalStop{Done{state}}
362	    case 1n+p:
363	      match p:
364	        case 0n:
365	          do IO<WalProgress>:
366	            _closed : Unit <- File.close(file)
367	            wal.truncate(path, offset, state)
368	        case 1n+q:
369	          match q:
370	            case 0n:
371	              do IO<WalProgress>:
372	                _closed : Unit <- File.close(file)
373	                wal.truncate(path, offset, state)
374	            case 1n+_:
375	              do IO<WalProgress>:
376	                pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 4n)
377	                wal.frames.length(pair, size, offset, state, path)
378	
379	def wal.frames.end(
380	  at_end: Bool,
381	  file: File,
382	  +size: Nat,
383	  +offset: Nat,
384	  state: Db.RotRes,
385	  +path: String
386	) -> IO(WalProgress):
387	  match at_end:
388	    case True{}:
389	      do IO<WalProgress>:
390	        _closed : Unit <- File.close(file)
391	        return WalStop{Done{state}}
392	    case False{}:
393	      wal.frames.remaining(file, size, offset, state, path,
394	        (size - offset : Nat))
395	
396	def wal.frames(
397	  +fuel: Nat,
398	  +size: Nat,
399	  +path: String,
400	  state: WalState
401	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
402	  match fuel:
403	    case 0n:
404	      match state:
405	        case WalAt{file, _, _}:
406	          do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
407	            _closed : Unit <- File.close(file)
408	            return Fail{(U32.from_nat(7n), "WAL frame limit exceeded")}
409	        case WalAfter{WalStop{result}}:
410	          IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, result)
411	        case WalAfter{WalMore{file, _, _}}:
412	          do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
413	            _closed : Unit <- File.close(file)
414	            return Fail{(U32.from_nat(7n), "WAL frame limit exceeded")}
415	    case 1n+rest:
416	      match state:
417	        case WalAt{file, +offset, res}:
418	          do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
419	            progress : WalProgress <- wal.frames.end(
420	              Nat.is_eq(offset, size), file, size, offset, res, path)
421	            wal.frames(rest, size, path, WalAfter{progress})
422	        case WalAfter{WalStop{result}}:
423	          IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, result)
424	        case WalAfter{WalMore{file, offset, res}}:
425	          wal.frames(rest, size, path, WalAt{file, offset, res})
426	
427	def wal.header.decoded(
428	  result: Result<&1, &1, Wal.Error, Unit>,
429	  file: File,
430	  +size: Nat,
431	  +path: String
432	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
433	  match result:
434	    case Fail{_}:
435	      do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
436	        _closed : Unit <- File.close(file)
437	        return Fail{(U32.from_nat(3n), "unsupported WAL format")}
438	    case Done{Unit{}}:
439	      wal.frames(Nat.add(Nat.div(size, 18n), 4n), size, path,
440	        WalAt{file, 8n, wal.empty()})
441	
442	def wal.header(
443	  pair: File & Result<&1, &1, U32 & String, ExactResult>,
444	  +size: Nat,
445	  +path: String
446	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
447	  match pair:
448	    case (file, Fail{error}):
449	      do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
450	        _closed : Unit <- File.close(file)
451	        return Fail{error}
452	    case (file, Done{ExactResult{_, False{}}}):
453	      do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
454	        _closed : Unit <- File.close(file)
455	        _truncated : Unit <- IO.try(Unit, Fs.truncate(path, 0n))
456	        _synced : Unit <- IO.try(Unit, Fs.fsync(path))
457	        return Done{wal.empty()}
458	    case (file, Done{ExactResult{bytes, True{}}}):
459	      wal.header.decoded(Wal.decode_header(bytes), file, size, path)
460	
461	def wal.open.present(
462	  present: Bool,
463	  +path: String
464	) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
465	  match present:
466	    case False{}:
467	      IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Done{wal.empty()})
468	    case True{}:
469	      do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
470	        size : Nat <- IO.try(Nat, Fs.file_size(path))
471	        file : File <- IO.try(File, File.open(path, "r"))
472	        pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 8n)
473	        wal.header(pair, size, path)
474	
475	# Handle wal open in the database recovery effects.
476	def wal_open(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
477	  do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
478	    present : Bool <- IO.try(Bool, Fs.exists(Db.wal_path(dir)))
479	    wal.open.present(present, Db.wal_path(dir))
480	
481	# --- Tmp sweep (best-effort) ---
482	
483	# Handle tmp pick in the database recovery effects.
484	def tmp_pick(keep: Bool, +name: String, +acc: List<&2, String>) -> List<&2, String>:
485	  match keep:
486	    case True{}:
487	      Con{name, acc}
488	    case False{}:
489	      acc
490	
491	# Handle tmp some in the database recovery effects.
492	def tmp_some(+ok: Bool, +name: String) -> Maybe<&2, String>:
493	  match ok:
494	    case True{}:
495	      Some{name}
496	    case False{}:
497	      None{}
498	
499	# Handle tmp map list in the database recovery effects.
500	def tmp_map_list(xs: List<&1, String>) -> List<&2, Maybe<&2, String>>:
501	  match xs:
502	    case Nil{}:
503	      Nil{}
504	    case Con{+h, t}:
505	      Con{tmp_some(String.ends_with(h, ".tmp"), h), tmp_map_list(t)}
506	
507	# Handle cat go in the database recovery effects.
508	def cat_go(xs: List<&2, Maybe<&2, String>>) -> List<&2, String>:
509	  match xs:
510	    case Nil{}:
511	      Nil{}
512	    case Con{h, t}:
513	      match h:
514	        case None{}:
515	          cat_go(t)
516	        case Some{s}:
517	          Con{s, cat_go(t)}
518	
519	# Handle tmp names in the database recovery effects.
520	def tmp_names(xs: List<&1, String>) -> List<&2, String>:
521	  cat_go(tmp_map_list(xs))
522	
523	# Handle sweep one in the database recovery effects.
524	def sweep_one(res: Result<&1, &1, U32 & String, List<String>>, +dir: String) -> IO(Unit):
525	  match res:
526	    case Fail{e}:
527	      IO.pure(Unit, Unit{})
528	    case Done{ns}:
529	      do IO<Unit>:
530	        _ign : Result<&1, &1, U32 & String, Unit> <- CompactIo.remove_list(Compact.prefix(tmp_names(ns), dir))
531	        IO.pure(Unit, Unit{})
532	
533	# Handle sweep get in the database recovery effects.
534	def sweep_get(+dir: String, +sub: String) -> IO(Unit):
535	  do IO<Unit>:
536	    r : Result<&1, &1, U32 & String, List<String>> <- Fs.read_dir(dir ++ "/" ++ sub)
537	    sweep_one(r, dir ++ "/" ++ sub)
538	
539	# Handle sweep cnt in the database recovery effects.
540	def sweep_cnt(res: Result<&1, &1, U32 & String, Nat>, +dir: String, +sub: String) -> IO(Unit):
541	  match res:
542	    case Fail{e}:
543	      IO.pure(Unit, Unit{})
544	    case Done{n}:
545	      sweep_get(dir, sub)
546	
547	# Handle sweep dir in the database recovery effects.
548	def sweep_dir(+dir: String, +sub: String) -> IO(Unit):
549	  do IO<Unit>:
550	    c : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(dir ++ "/" ++ sub)
551	    sweep_cnt(c, dir, sub)
552	
553	# --- Open assembly ---
554	
555	# Handle wal part in the database recovery effects.
556	def wal_part(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>):
557	  do IO<Result<&1, &1, U32 & String, Db.RotRes>>:
558	    _initialized : Unit <- IO.try(Unit, DbIo.wal_initialize(dir))
559	    wal_open(dir)
560	
561	# Open assemble for the database recovery effects.
562	def open_assemble(
563	  res: Db.RotRes,
564	  +dir: String,
565	  +mfst: Manifest.Manifest,
566	  levels: List<&2, List<&2, Sstable.Table>>
567	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
568	  match res:
569	    case Db.Rot{mem, frozen, mc, fc}:
570	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
571	        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}}
572	
573	# Open levels checked for the database recovery effects.
574	def open_levels_checked(
575	  ok: Bool,
576	  +dir: String,
577	  +mfst: Manifest.Manifest,
578	  levels: List<&2, List<&2, Sstable.Table>>
579	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
580	  match ok:
581	    case False{}:
582	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(6n), "overlapping tables in L1+ manifest")})
583	    case True{}:
584	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
585	        res : Db.RotRes <- IO.try(Db.RotRes, wal_part(dir))
586	        open_assemble(res, dir, mfst, levels)
587	
588	# Open levels for the database recovery effects.
589	def open_levels(+dir: String, +mfst: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Db.Db>):
590	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
591	    +levels : List<&2, List<&2, Sstable.Table>> <- load_levels(Pure.mfst_lists(mfst), 0n, dir, Nil{})
592	    open_levels_checked(Pure.l1_plus_disjoint(levels), dir, mfst, levels)
593	
594	# Open names go for the database recovery effects.
595	def open_names_go(ok: Bool, +dir: String, +mfst: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Db.Db>):
596	  match ok:
597	    case True{}:
598	      open_levels(dir, mfst)
599	    case False{}:
600	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(5n), "bad manifest names")})
601	
602	# Handle sweep root go in the database recovery effects.
603	def sweep_root_go(+dir: String) -> IO(Unit):
604	  do IO<Unit>:
605	    r : Result<&1, &1, U32 & String, List<String>> <- Fs.read_dir(dir)
606	    sweep_one(r, dir)
607	
608	# Handle sweep root cnt in the database recovery effects.
609	def sweep_root_cnt(res: Result<&1, &1, U32 & String, Nat>, +dir: String) -> IO(Unit):
610	  match res:
611	    case Fail{e}:
612	      IO.pure(Unit, Unit{})
613	    case Done{n}:
614	      sweep_root_go(dir)
615	
616	# Handle sweep root in the database recovery effects.
617	def sweep_root(+dir: String) -> IO(Unit):
618	  do IO<Unit>:
619	    c : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(dir)
620	    sweep_root_cnt(c, dir)
621	
622	# Open db for the database recovery effects.
623	def open_db(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
624	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
625	    _u0 : Unit <- IO.try(Unit, Flush.ensure_dir(dir))
626	    _s0 : Unit <- sweep_dir(dir, "l0")
627	    _s1 : Unit <- sweep_dir(dir, "l1")
628	    _s2 : Unit <- sweep_dir(dir, "l2")
629	    _s3 : Unit <- sweep_dir(dir, "l3")
630	    _s4 : Unit <- sweep_root(dir)
631	    +m : Manifest.Manifest <- IO.try(Manifest.Manifest, Flush.load_mfst(dir ++ "/MANIFEST"))
632	    open_names_go(Pure.manifest_names_ok(Pure.mfst_lists(m), 0n), dir, m)
633	
634	# --- Maintenance (synchronous auto flush/compact + stall) ---
635	# One drain round always suffices (L0 >= 8 > 4, so the gated compaction
636	# fires and empties L0) — no loop, no cycle.
637	
638	# Flush dec for the database recovery effects.
639	def flush_dec(full: Bool, +db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
640	  match full:
641	    case True{}:
642	      Flush.flush(db)
643	    case False{}:
644	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Done{db})
645	
646	# Flush gate for the database recovery effects.
647	def flush_gate(db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
648	  match db:
649	    case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, +mem_count, frozen_count}:
650	      flush_dec(Nat.is_lt(4096n, mem_count), Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count})
651	
652	# Handle l0 full in the database recovery effects.
653	def l0_full(db: Db.Db) -> Bool:
654	  match db:
655	    case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
656	      Nat.is_lt(7n, List.length(&2, Sstable.Table, Compact.l0list(levels)))
657	
658	# Handle drain once in the database recovery effects.
659	def drain_once(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
660	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
661	    f : Db.Db <- IO.try(Db.Db, Flush.flush(db))
662	    c : Db.Db <- IO.try(Db.Db, CompactIo.compact(f))
663	    return Done{c}
664	
665	# Handle drain pick in the database recovery effects.
666	def drain_pick(full: Bool, +db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
667	  match full:
668	    case True{}:
669	      drain_once(db)
670	    case False{}:
671	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Done{db})
672	
673	# Handle drain go in the database recovery effects.
674	def drain_go(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
675	  drain_pick(l0_full(db), db)
676	
677	# Handle maintain in the database recovery effects.
678	def maintain(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
679	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
680	    d : Db.Db <- IO.try(Db.Db, drain_go(db))
681	    f : Db.Db <- IO.try(Db.Db, flush_gate(d))
682	    CompactIo.compact(f)
683	
684	# Handle maintain unwrap in the database recovery effects.
685	def maintain_unwrap(res: Result<&1, &1, U32 & String, Db.Db>) -> IO(Result<&1, &1, U32 & String, Db.Db>):
686	  match res:
687	    case Fail{e}:
688	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{e})
689	    case Done{db}:
690	      maintain(db)
691	
692	# Handle write in the database recovery effects.
693	def write(+db: Db.Db, +batch: Wal.Batch) -> IO(Result<&1, &1, U32 & String, Db.Db>):
694	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
695	    r : Result<&1, &1, U32 & String, Db.Db> <- DbIo.db_write(db, batch)
696	    maintain_unwrap(r)
697	
698	# Handle put in the database recovery effects.
699	def put(+db: Db.Db, +key: String, +val: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
700	  write(db, Wal.Batch{Con{Wal.Put{key, val}, Nil{}}})
701	
702	# Handle del in the database recovery effects.
703	def del(+db: Db.Db, +key: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
704	  write(db, Wal.Batch{Con{Wal.Del{key}, Nil{}}})
705	
706	# --- Closed-vector laws live in laws/Recover.bend (spec law 7) ---