src/Recover.bend
mylsm-lsm-store@0.4.0.0 · 706 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.4.0.0/src/Keys.bend as Keys
3import mylsm-lsm-store@0.4.0.0/src/MemTable.bend as MemTable
4import mylsm-lsm-store@0.4.0.0/src/Sstable.bend as Sstable
5import mylsm-lsm-store@0.4.0.0/src/SstStreamIo.bend as SstStreamIo
6import mylsm-lsm-store@0.4.0.0/src/Wal.bend as Wal
7import mylsm-lsm-store@0.4.0.0/src/Manifest.bend as Manifest
8import mylsm-lsm-store@0.4.0.0/src/Flush.bend as Flush
9import mylsm-lsm-store@0.4.0.0/src/Compact.bend as Compact
10import mylsm-lsm-store@0.4.0.0/src/CompactIo.bend as CompactIo
11import mylsm-lsm-store@0.4.0.0/src/Fs.bend as Fs
12import mylsm-lsm-store@0.4.0.0/src/Db.bend as Db
13import mylsm-lsm-store@0.4.0.0/src/DbIo.bend as DbIo
14import mylsm-lsm-store@0.4.0.0/src/RecoverPure.bend as Pure
15import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
16
17# Represent ExactResult data used by the database recovery effects.
18type ExactResult is Type:
19 ExactResult{bytes: Bytes.Bytes, complete: Bool}
20
21# Represent ExactState data used by the database recovery effects.
22type 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.
27type 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.
32type WalState is Type:
33 WalAt{file: File, offset: Nat, state: Db.RotRes}
34 WalAfter{progress: WalProgress}
35
36def exact.max(large: Bool, +remaining: Nat) -> U32:
37 match large:
38 case True{}:
39 1048576
40 case False{}:
41 U32.from_nat(remaining)
42
43def 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
51def 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.
112def 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.
127def 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.
135def 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.
155def 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
170def wal.empty() -> Db.RotRes:
171 Db.Rot{MemTable.empty(), MemTable.empty(), 0n, 0n}
172
173def 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
176def 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
186def wal.u32.pair(pair: Bytes.Cursor & Maybe<&2, U32>) -> Maybe<&2, U32>:
187 match pair:
188 case (_, value):
189 value
190
191def wal.u32(bytes: Bytes.Bytes) -> Maybe<&2, U32>:
192 wal.u32.pair(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))
193
194def 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
199def 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
214def 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
235def 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
255def 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
268def 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
287def 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
307def 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
327def 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
349def 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
379def 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
396def 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
427def 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
442def 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
461def 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.
476def 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.
484def 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.
492def 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.
500def 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.
508def 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.
520def 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.
524def 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.
534def 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.
540def 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.
548def 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.
556def 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.
562def 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.
574def 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.
589def 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.
595def 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.
603def 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.
609def 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.
617def 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.
623def 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.
639def 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.
647def 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.
653def 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.
659def 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.
666def 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.
674def 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.
678def 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.
685def 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.
693def 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.
699def 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.
703def 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) ---