src/Recover.bend
mylsm-lsm-store@0.5.0.0 · 1,123 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.5.0.0/src/Keys.bend as Keys
3import mylsm-lsm-store@0.5.0.0/src/MemTable.bend as MemTable
4import mylsm-lsm-store@0.5.0.0/src/Sstable.bend as Sstable
5import mylsm-lsm-store@0.5.0.0/src/SstStreamIo.bend as SstStreamIo
6import mylsm-lsm-store@0.5.0.0/src/Wal.bend as Wal
7import mylsm-lsm-store@0.5.0.0/src/Manifest.bend as Manifest
8import mylsm-lsm-store@0.5.0.0/src/Flush.bend as Flush
9import mylsm-lsm-store@0.5.0.0/src/Compact.bend as Compact
10import mylsm-lsm-store@0.5.0.0/src/CompactIo.bend as CompactIo
11import mylsm-lsm-store@0.5.0.0/src/Fs.bend as Fs
12import mylsm-lsm-store@0.5.0.0/src/FsPolicy.bend as FsPolicy
13import mylsm-lsm-store@0.5.0.0/src/Db.bend as Db
14import mylsm-lsm-store@0.5.0.0/src/DbIo.bend as DbIo
15import mylsm-lsm-store@0.5.0.0/src/RecoverPure.bend as Pure
16import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
17
18# Represent ExactResult data used by the database recovery effects.
19type ExactResult is Type:
20 ExactResult{bytes: Bytes.Bytes, complete: Bool}
21
22# Represent ExactState data used by the database recovery effects.
23type 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.
28type 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.
33type 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.
38def wal_corruption_error_code() -> U32:
39 4294967290
40
41def exact.max(large: Bool, +remaining: Nat) -> U32:
42 match large:
43 case True{}:
44 1048576
45 case False{}:
46 U32.from_nat(remaining)
47
48def 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
56def 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.
116type 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.
121type 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.
127def 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.
149def 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.
169def 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
199def wal.empty() -> Db.RotRes:
200 Db.Rot{MemTable.empty(), MemTable.empty(), 0n, 0n}
201
202def 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
205def 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
215def 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
228def 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
237def wal.u32.pair(pair: Bytes.Cursor & Maybe<&2, U32>) -> Maybe<&2, U32>:
238 match pair:
239 case (_, value):
240 value
241
242def wal.u32(bytes: Bytes.Bytes) -> Maybe<&2, U32>:
243 wal.u32.pair(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))
244
245def 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
250def 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
265def 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
286def 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
306def 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
319def 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
338def 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
358def 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
378def 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
400def 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
430def 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
447def 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
478def 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
493def 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
502def 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
514def 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
532def 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
545def 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
557def 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
569def 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.
580def 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.
588def 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.
596def 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.
604def 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.
612def 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.
624def 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.
628def 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.
638def 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.
644def 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.
652def 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
659def 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.
670def 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.
676def 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
687def 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.
700def 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
714def 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
725def 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.
729def 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.
735def 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.
743def 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.
749def 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.
757def 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
762def 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
771def 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
781def 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.
794def 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
799def 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.
807def 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.
811def 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.
817type 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.
822def 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.
846def 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.
858def 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.
864def 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
873def 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.
885def 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.
892type 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.
897def 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.
922def 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.
927def 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.
939def 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.
952def 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
964def 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.
975def 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.
981def 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.
992def 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.
1005def 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.
1018def 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.
1028def 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.
1036def 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.
1042def 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.
1048def 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.
1058def 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.
1064def 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.
1072def drain_go(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
1073 drain_pick(l0_full(db), db)
1074
1075def 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
1084def 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.
1096def 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.
1102def 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.
1110def 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.
1116def 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.
1120def 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) ---