src/Recover.bend source
src/Recover.bend on the hub · documented module
import Baseimport ./Keys.bend as Keysimport ./MemTable.bend as MemTableimport ./Sstable.bend as Sstableimport ./SstStreamIo.bend as SstStreamIoimport ./Wal.bend as Walimport ./Manifest.bend as Manifestimport ./Flush.bend as Flushimport ./Compact.bend as Compactimport ./CompactIo.bend as CompactIoimport ./Fs.bend as Fsimport ./FsPolicy.bend as FsPolicyimport ./Db.bend as Dbimport ./DbIo.bend as DbIoimport ./RecoverPure.bend as Pureimport bend-kit-bytes@0.3.2.0/bytes.bend as Bytes# Represent ExactResult data used by the database recovery effects.type ExactResult is Type: ExactResult{bytes: Bytes.Bytes, complete: Bool}# Represent ExactState data used by the database recovery effects.type ExactState is Type: ExactNeed{file: File, remaining: Nat, chunks: List<&1, Bytes.Bytes>} ExactRead{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>, remaining: Nat, chunks: List<&1, Bytes.Bytes>}# Represent WalProgress data used by the database recovery effects.type WalProgress is Type: WalMore{file: File, offset: Nat, state: Db.RotRes} WalStop{result: Result<&1, &1, U32 & String, Db.RotRes>}# Represent WalState data used by the database recovery effects.type WalState is Type: WalAt{file: File, offset: Nat, state: Db.RotRes} WalAfter{progress: WalProgress}# Returns the typed corruption code for malformed WAL data.def wal_corruption_error_code() -> U32: 4294967290def exact.max(large: Bool, +remaining: Nat) -> U32: match large: case True{}: 1048576 case False{}: U32.from_nat(remaining)def exact.result( file: File, chunks: List<&1, Bytes.Bytes>, complete: Bool) -> IO(File & Result<&1, &1, U32 & String, ExactResult>): IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Done{ExactResult{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks)), complete}}))def exact.loop( fuel: Nat, state: ExactState) -> IO(File & Result<&1, &1, U32 & String, ExactResult>): match fuel: case 0n: match state: case ExactNeed{file, remaining, chunks}: match remaining: case 0n: exact.result(file, chunks, True{}) case 1n+_: IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{(U32.from_nat(7n), "bounded read exhausted")})) case ExactRead{pair, _, _}: match pair: case (file, Fail{error}): IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{error})) case (file, Done{_}): IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{(U32.from_nat(7n), "bounded read exhausted")})) case 1n+rest: match state: case ExactNeed{file, +remaining, chunks}: match remaining: case 0n: exact.result(file, chunks, True{}) case 1n+_: do IO<File & Result<&1, &1, U32 & String, ExactResult>>: next : File & Result<&1, &1, U32 & String, Bytes.Bytes> <- Fs.read_bytes(file, exact.max(Nat.is_le(1048577n, remaining), remaining)) exact.loop(rest, ExactRead{next, remaining, chunks}) case ExactRead{pair, remaining, chunks}: match pair: case (file, Fail{error}): IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{error})) case (file, Done{Bytes.Bytes{0, _}}): exact.result(file, chunks, False{}) case (file, Done{Bytes.Bytes{+len, buf}}): exact.loop(rest, ExactNeed{file, (remaining - U32.to_nat(len) : Nat), Con{Bytes.Bytes{len, buf}, chunks}})# Recovery + maintenance (Task 10).## Open: ensure dir; sweep *.tmp (best-effort); load manifest (absent =# fresh; corrupt = fatal); shape-check all names; load tables positionally# (listed-but-missing/corrupt = fatal, never silent); restore the flush# counter from the greatest table-name generation; replay + truncate# the WAL; return the Db.## Maintenance (synchronous, observably identical ordering to background):# after every acked batch: drain while L0 >= 8 (stall = blocking# maintenance), flush when mem >= 4096, compact (self-gated at L0 > 4).# True fork-based background (frozen mem + IO.spawn) is a benchmark-gated# follow-up; ordering and stall semantics are already exact.# --- Loaders (self-recursive IO; fatal via IO.die, never silent) ---# Represents loadtablesstate data in recover.type LoadTablesState is Type: LoadTablesNames{names: List<&2, String>, ldir: String, dir: String, acc: List<&2, Sstable.Table>} LoadTablesRead{result: Result<&1, &1, U32 & String, Sstable.Table>, names: List<&2, String>, ldir: String, dir: String, acc: List<&2, Sstable.Table>}# Represents loadlevelsstate data in recover.type LoadLevelsState is Type: LoadLevelsNames{levels: List<&2, List<&2, String>>, idx: Nat, dir: String, acc: List<&2, List<&2, Sstable.Table>>} 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>>} 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>>}# Loads the SSTables referenced by a Manifest and propagates failures.def load_tables(fuel: Nat, state: LoadTablesState) -> IO(Result<&1, &1, U32 & String, List<&2, Sstable.Table>>): match fuel: case 0n: IO.pure(Result<&1, &1, U32 & String, List<&2, Sstable.Table>>, Fail{(U32.from_nat(7n), "table recovery limit exceeded")}) case 1n+rest: match state: case LoadTablesNames{names, +ldir, +dir, acc}: match names: case Nil{}: IO.pure(Result<&1, &1, U32 & String, List<&2, Sstable.Table>>, Done{List.reverse(&2, Sstable.Table, acc)}) case Con{+name, tail}: do IO<Result<&1, &1, U32 & String, List<&2, Sstable.Table>>>: table : Result<&1, &1, U32 & String, Sstable.Table> <- SstStreamIo.read_table(dir ++ "/" ++ ldir ++ "/" ++ name) load_tables(rest, LoadTablesRead{table, tail, ldir, dir, acc}) case LoadTablesRead{result, names, +ldir, +dir, acc}: match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, List<&2, Sstable.Table>>, Fail{error}) case Done{table}: load_tables(rest, LoadTablesNames{names, ldir, dir, Con{table, acc}})# Handle ldir of in the database recovery effects.def ldir_of(idx: Nat) -> Maybe<&2, String>: match idx: case 0n: Some{"l0"} case 1n+m: match m: case 0n: Some{"l1"} case 1n+p: match p: case 0n: Some{"l2"} case 1n+q: match q: case 0n: Some{"l3"} case 1n+r: None{}# Builds recovered levels from the Manifest table entries.def load_levels( fuel: Nat, state: LoadLevelsState,) -> IO(Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>): match fuel: case 0n: IO.pure(Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>, Fail{(U32.from_nat(7n), "level recovery limit exceeded")}) case 1n+rest: match state: case LoadLevelsNames{levels, +idx, +dir, acc}: match levels: case Nil{}: IO.pure(Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>, Done{List.reverse(&2, List<&2, Sstable.Table>, acc)}) case Con{+names, tail}: load_levels(rest, LoadLevelsPath{ldir_of(idx), names, tail, Nat.add(idx, 1n), dir, acc}) case LoadLevelsPath{ldir, +names, levels, idx, +dir, acc}: match ldir: case None{}: IO.pure(Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>, Fail{(U32.from_nat(5n), "bad manifest names")}) case Some{ldir}: do IO<Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>>: 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{}}) load_levels(rest, LoadLevelsRead{tables, levels, idx, dir, acc}) case LoadLevelsRead{result, levels, idx, +dir, acc}: match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>, Fail{error}) case Done{tables}: load_levels(rest, LoadLevelsNames{levels, idx, dir, Con{tables, acc}})def wal.empty() -> Db.RotRes: Db.Rot{MemTable.empty(), MemTable.empty(), 0n, 0n}def exact.start(file: File, +needed: Nat) -> IO(File & Result<&1, &1, U32 & String, ExactResult>): exact.loop(Nat.add(Nat.mul(2n, needed), 1n), ExactNeed{file, needed, Nil{}})def wal.truncate.synced( result: Result<&1, &1, U32 & String, Unit>, state: Db.RotRes) -> IO(WalProgress): match result: case Fail{error}: IO.pure(WalProgress, WalStop{Fail{error}}) case Done{Unit{}}: IO.pure(WalProgress, WalStop{Done{state}})def wal.truncate.result( result: Result<&1, &1, U32 & String, Unit>, +path: String, state: Db.RotRes) -> IO(WalProgress): match result: case Fail{error}: IO.pure(WalProgress, WalStop{Fail{error}}) case Done{Unit{}}: do IO<WalProgress>: synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path) wal.truncate.synced(synced, state)def wal.truncate( +path: String, +offset: Nat, state: Db.RotRes) -> IO(WalProgress): do IO<WalProgress>: truncated : Result<&1, &1, U32 & String, Unit> <- Fs.truncate(path, offset) wal.truncate.result(truncated, path, state)def wal.u32.pair(pair: Bytes.Cursor & Maybe<&2, U32>) -> Maybe<&2, U32>: match pair: case (_, value): valuedef wal.u32(bytes: Bytes.Bytes) -> Maybe<&2, U32>: wal.u32.pair(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))def wal.apply(batch: Wal.Batch, state: Db.RotRes) -> Db.RotRes: match batch state: case Wal.Batch{muts} Db.Rot{mem, frozen, mem_count, frozen_count}: Db.apply_and_rotate(muts, mem, frozen, mem_count, frozen_count)def wal.frames.decoded( result: Result<&1, &1, Wal.Error, Wal.Batch>, file: File, +next_offset: Nat, state: Db.RotRes) -> IO(WalProgress): match result: case Fail{_}: do IO<WalProgress>: _closed : Unit <- File.close(file) return WalStop{Fail{(wal_corruption_error_code(), "corrupt WAL frame")}} case Done{batch}: IO.pure(WalProgress, WalMore{file, next_offset, wal.apply(batch, state)})def wal.frames.decode( pair: File & Result<&1, &1, U32 & String, ExactResult>, prefix: Bytes.Bytes, +offset: Nat, +next_offset: Nat, state: Db.RotRes, +path: String) -> IO(WalProgress): match pair: case (file, Fail{error}): do IO<WalProgress>: _closed : Unit <- File.close(file) return WalStop{Fail{error}} case (file, Done{ExactResult{_, False{}}}): do IO<WalProgress>: _closed : Unit <- File.close(file) wal.truncate(path, offset, state) case (file, Done{ExactResult{body, True{}}}): wal.frames.decoded(Wal.decode_frame(Bytes.concat([prefix, body])), file, next_offset, state)def wal.frames.body.bound( enough: Bool, file: File, prefix: Bytes.Bytes, +frame_size: Nat, +offset: Nat, +next_offset: Nat, state: Db.RotRes, +path: String) -> IO(WalProgress): match enough: case False{}: do IO<WalProgress>: _closed : Unit <- File.close(file) wal.truncate(path, offset, state) case True{}: do IO<WalProgress>: pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, frame_size) wal.frames.decode(pair, prefix, offset, next_offset, state, path)def wal.frames.body( file: File, prefix: Bytes.Bytes, +frame_size: Nat, +size: Nat, +offset: Nat, state: Db.RotRes, +path: String) -> IO(WalProgress): +next_offset = Nat.add(offset, Nat.add(4n, frame_size)) wal.frames.body.bound(Nat.is_le(next_offset, size), file, prefix, frame_size, offset, next_offset, state, path)def wal.frames.bound( within: Bool, file: File, bytes: Bytes.Bytes, frame_len: U32, +size: Nat, +offset: Nat, state: Db.RotRes, +path: String) -> IO(WalProgress): match within: case False{}: do IO<WalProgress>: _closed : Unit <- File.close(file) return WalStop{Fail{(wal_corruption_error_code(), "WAL frame length out of bounds")}} case True{}: wal.frames.body(file, bytes, U32.to_nat(frame_len), size, offset, state, path)def wal.frames.length.checked( enough: Bool, within: Bool, file: File, prefix: Bytes.Bytes, frame_len: U32, +size: Nat, +offset: Nat, state: Db.RotRes, +path: String) -> IO(WalProgress): match enough: case False{}: do IO<WalProgress>: _closed : Unit <- File.close(file) wal.truncate(path, offset, state) case True{}: wal.frames.bound(within, file, prefix, frame_len, size, offset, state, path)def wal.frames.length.value( file: File, maybe_len: Maybe<&2, U32>, +size: Nat, +offset: Nat, state: Db.RotRes, +path: String) -> IO(WalProgress): match maybe_len: case None{}: do IO<WalProgress>: _closed : Unit <- File.close(file) return WalStop{Fail{(wal_corruption_error_code(), "malformed WAL frame length")}} case Some{+frame_len}: +next_offset = Nat.add(offset, Nat.add(4n, U32.to_nat(frame_len))) wal.frames.length.checked(Nat.is_le(next_offset, size), U32.is_le(36, frame_len) && U32.is_le(frame_len, 1073741860), file, Wal.u32_bytes(frame_len), frame_len, size, offset, state, path)def wal.frames.length( pair: File & Result<&1, &1, U32 & String, ExactResult>, +size: Nat, +offset: Nat, state: Db.RotRes, +path: String) -> IO(WalProgress): match pair: case (file, Fail{error}): do IO<WalProgress>: _closed : Unit <- File.close(file) return WalStop{Fail{error}} case (file, Done{ExactResult{bytes, complete}}): match complete: case False{}: do IO<WalProgress>: _closed : Unit <- File.close(file) wal.truncate(path, offset, state) case True{}: wal.frames.length.value(file, wal.u32(bytes), size, offset, state, path)def wal.frames.remaining( file: File, +size: Nat, +offset: Nat, state: Db.RotRes, +path: String, remaining: Nat) -> IO(WalProgress): match remaining: case 0n: do IO<WalProgress>: _closed : Unit <- File.close(file) return WalStop{Done{state}} case 1n+p: match p: case 0n: do IO<WalProgress>: _closed : Unit <- File.close(file) wal.truncate(path, offset, state) case 1n+q: match q: case 0n: do IO<WalProgress>: _closed : Unit <- File.close(file) wal.truncate(path, offset, state) case 1n+_: do IO<WalProgress>: pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 4n) wal.frames.length(pair, size, offset, state, path)def wal.frames.end( at_end: Bool, file: File, +size: Nat, +offset: Nat, state: Db.RotRes, +path: String) -> IO(WalProgress): match at_end: case True{}: do IO<WalProgress>: _closed : Unit <- File.close(file) return WalStop{Done{state}} case False{}: wal.frames.remaining(file, size, offset, state, path, (size - offset : Nat))def wal.frames( +fuel: Nat, +size: Nat, +path: String, state: WalState) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): match fuel: case 0n: match state: case WalAt{file, _, _}: do IO<Result<&1, &1, U32 & String, Db.RotRes>>: _closed : Unit <- File.close(file) return Fail{(U32.from_nat(7n), "WAL frame limit exceeded")} case WalAfter{WalStop{result}}: IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, result) case WalAfter{WalMore{file, _, _}}: do IO<Result<&1, &1, U32 & String, Db.RotRes>>: _closed : Unit <- File.close(file) return Fail{(U32.from_nat(7n), "WAL frame limit exceeded")} case 1n+rest: match state: case WalAt{file, +offset, res}: do IO<Result<&1, &1, U32 & String, Db.RotRes>>: progress : WalProgress <- wal.frames.end( Nat.is_eq(offset, size), file, size, offset, res, path) wal.frames(rest, size, path, WalAfter{progress}) case WalAfter{WalStop{result}}: IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, result) case WalAfter{WalMore{file, offset, res}}: wal.frames(rest, size, path, WalAt{file, offset, res})def wal.header.decoded( result: Result<&1, &1, Wal.Error, Unit>, file: File, +size: Nat, +path: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): match result: case Fail{_}: do IO<Result<&1, &1, U32 & String, Db.RotRes>>: _closed : Unit <- File.close(file) return Fail{(wal_corruption_error_code(), "unsupported WAL format")} case Done{Unit{}}: wal.frames(Nat.add(Nat.div(size, 18n), 4n), size, path, WalAt{file, 8n, wal.empty()})def wal.header.synced( result: Result<&1, &1, U32 & String, Unit>) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error}) case Done{Unit{}}: IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Done{wal.empty()})def wal.header.truncated( result: Result<&1, &1, U32 & String, Unit>, +path: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error}) case Done{Unit{}}: do IO<Result<&1, &1, U32 & String, Db.RotRes>>: synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path) wal.header.synced(synced)def wal.header( pair: File & Result<&1, &1, U32 & String, ExactResult>, +size: Nat, +path: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): match pair: case (file, Fail{error}): do IO<Result<&1, &1, U32 & String, Db.RotRes>>: _closed : Unit <- File.close(file) return Fail{error} case (file, Done{ExactResult{_, False{}}}): do IO<Result<&1, &1, U32 & String, Db.RotRes>>: _closed : Unit <- File.close(file) truncated : Result<&1, &1, U32 & String, Unit> <- Fs.truncate(path, 0n) wal.header.truncated(truncated, path) case (file, Done{ExactResult{bytes, True{}}}): wal.header.decoded(Wal.decode_header(bytes), file, size, path)def wal.open.file( size: Nat, opened: Result<&1, &1, U32 & String, File>, +path: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): match opened: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error}) case Done{file}: do IO<Result<&1, &1, U32 & String, Db.RotRes>>: pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 8n) wal.header(pair, size, path)def wal.open.size( result: Result<&1, &1, U32 & String, Nat>, +path: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error}) case Done{size}: do IO<Result<&1, &1, U32 & String, Db.RotRes>>: opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r") wal.open.file(size, opened, path)def wal.open.present( present: Bool, +path: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): match present: case False{}: IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Done{wal.empty()}) case True{}: do IO<Result<&1, &1, U32 & String, Db.RotRes>>: size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path) wal.open.size(size, path)def wal_open.exists( result: Result<&1, &1, U32 & String, Bool>, +dir: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error}) case Done{present}: wal.open.present(present, Db.wal_path(dir))# Handle wal open in the database recovery effects.def wal_open(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): do IO<Result<&1, &1, U32 & String, Db.RotRes>>: present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(Db.wal_path(dir)) wal_open.exists(present, dir)# --- Tmp sweep (best-effort) ---# Handle tmp pick in the database recovery effects.def tmp_pick(keep: Bool, +name: String, +acc: List<&2, String>) -> List<&2, String>: match keep: case True{}: Con{name, acc} case False{}: acc# Handle tmp some in the database recovery effects.def tmp_some(+ok: Bool, +name: String) -> Maybe<&2, String>: match ok: case True{}: Some{name} case False{}: None{}# Handle tmp map list in the database recovery effects.def tmp_map_list(xs: List<&1, String>) -> List<&2, Maybe<&2, String>>: match xs: case Nil{}: Nil{} case Con{+h, t}: Con{tmp_some(String.ends_with(h, ".tmp"), h), tmp_map_list(t)}# Handle cat go in the database recovery effects.def cat_go(xs: List<&2, Maybe<&2, String>>) -> List<&2, String>: match xs: case Nil{}: Nil{} case Con{h, t}: match h: case None{}: cat_go(t) case Some{s}: Con{s, cat_go(t)}# Handle tmp names in the database recovery effects.def tmp_names(xs: List<&1, String>) -> List<&2, String>: cat_go(tmp_map_list(xs))# Handle sweep one in the database recovery effects.def sweep_one(res: Result<&1, &1, U32 & String, List<String>>, +dir: String) -> IO(Unit): match res: case Fail{e}: IO.pure(Unit, Unit{}) case Done{ns}: do IO<Unit>: _ign : Result<&1, &1, U32 & String, Unit> <- CompactIo.remove_list(Compact.prefix(tmp_names(ns), dir)) IO.pure(Unit, Unit{})# Handle sweep get in the database recovery effects.def sweep_get(+dir: String, +sub: String) -> IO(Unit): do IO<Unit>: r : Result<&1, &1, U32 & String, List<String>> <- Fs.read_dir(dir ++ "/" ++ sub) sweep_one(r, dir ++ "/" ++ sub)# Handle sweep cnt in the database recovery effects.def sweep_cnt(res: Result<&1, &1, U32 & String, Nat>, +dir: String, +sub: String) -> IO(Unit): match res: case Fail{e}: IO.pure(Unit, Unit{}) case Done{n}: sweep_get(dir, sub)# Handle sweep dir in the database recovery effects.def sweep_dir(+dir: String, +sub: String) -> IO(Unit): do IO<Unit>: c : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(dir ++ "/" ++ sub) sweep_cnt(c, dir, sub)# --- Open assembly ---def wal_part.initialized( result: Result<&1, &1, U32 & String, Unit>, +dir: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.RotRes>, Fail{error}) case Done{Unit{}}: wal_open(dir)# Splits the WAL replay stream at the recovery boundary.def wal_part(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.RotRes>): do IO<Result<&1, &1, U32 & String, Db.RotRes>>: initialized : Result<&1, &1, U32 & String, Unit> <- DbIo.wal_initialize(dir) wal_part.initialized(initialized, dir)# Open assemble for the database recovery effects.def open_assemble( res: Db.RotRes, +dir: String, +mfst: Manifest.Manifest, levels: List<&2, List<&2, Sstable.Table>>) -> IO(Result<&1, &1, U32 & String, Db.Db>): match res: case Db.Rot{mem, frozen, mc, fc}: do IO<Result<&1, &1, U32 & String, Db.Db>>: 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}}def open_levels.wal( result: Result<&1, &1, U32 & String, Db.RotRes>, +dir: String, +mfst: Manifest.Manifest, levels: List<&2, List<&2, Sstable.Table>>) -> IO(Result<&1, &1, U32 & String, Db.Db>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{recovered}: open_assemble(recovered, dir, mfst, levels)# Loads Manifest levels and checks WAL recovery errors.def open_levels_checked( ok: Bool, +dir: String, +mfst: Manifest.Manifest, levels: List<&2, List<&2, Sstable.Table>>) -> IO(Result<&1, &1, U32 & String, Db.Db>): match ok: case False{}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(6n), "overlapping tables in L1+ manifest")}) case True{}: do IO<Result<&1, &1, U32 & String, Db.Db>>: recovered : Result<&1, &1, U32 & String, Db.RotRes> <- wal_part(dir) open_levels.wal(recovered, dir, mfst, levels)def open_levels.loaded( result: Result<&1, &1, U32 & String, List<&2, List<&2, Sstable.Table>>>, +dir: String, +mfst: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Db.Db>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{+levels}: open_levels_checked(Pure.l1_plus_disjoint(levels), dir, mfst, levels)def open_levels.fuel(+mfst: Manifest.Manifest) -> Nat: Nat.add(Nat.mul(3n, List.length(&2, List<&2, String>, Pure.mfst_lists(mfst))), 1n)# Opens levels through the legacy recovery result path.def open_levels(+dir: String, +mfst: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Db.Db>): do IO<Result<&1, &1, U32 & String, Db.Db>>: 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{}}) open_levels.loaded(levels, dir, mfst)# Open names go for the database recovery effects.def open_names_go(ok: Bool, +dir: String, +mfst: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Db.Db>): match ok: case True{}: open_levels(dir, mfst) case False{}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(5n), "bad manifest names")})# Handle sweep root go in the database recovery effects.def sweep_root_go(+dir: String) -> IO(Unit): do IO<Unit>: r : Result<&1, &1, U32 & String, List<String>> <- Fs.read_dir(dir) sweep_one(r, dir)# Handle sweep root cnt in the database recovery effects.def sweep_root_cnt(res: Result<&1, &1, U32 & String, Nat>, +dir: String) -> IO(Unit): match res: case Fail{e}: IO.pure(Unit, Unit{}) case Done{n}: sweep_root_go(dir)# Handle sweep root in the database recovery effects.def sweep_root(+dir: String) -> IO(Unit): do IO<Unit>: c : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(dir) sweep_root_cnt(c, dir)def open_db.swept(+dir: String, +mfst: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Db.Db>): do IO<Result<&1, &1, U32 & String, Db.Db>>: _s0 : Unit <- sweep_dir(dir, "l0") _s1 : Unit <- sweep_dir(dir, "l1") _s2 : Unit <- sweep_dir(dir, "l2") _s3 : Unit <- sweep_dir(dir, "l3") _s4 : Unit <- sweep_root(dir) open_names_go(Pure.manifest_names_ok(Pure.mfst_lists(mfst), 0n), dir, mfst)def open_db.manifest( result: Result<&1, &1, U32 & String, Manifest.Manifest>, +dir: String,) -> IO(Result<&1, &1, U32 & String, Db.Db>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{mfst}: open_db.swept(dir, mfst)def open_db.ensured( ensured: Result<&1, &1, U32 & String, Unit>, +dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>): match ensured: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{Unit{}}: do IO<Result<&1, &1, U32 & String, Db.Db>>: manifest : Result<&1, &1, U32 & String, Manifest.Manifest> <- Flush.load_mfst(dir ++ "/MANIFEST") open_db.manifest(manifest, dir)# Opens the legacy database and applies best-effort recovery cleanup.def open_db(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>): do IO<Result<&1, &1, U32 & String, Db.Db>>: ensured : Result<&1, &1, U32 & String, Unit> <- Flush.ensure_dir(dir) open_db.ensured(ensured, dir)def sweep_checked_path.select(root: Bool, +dir: String, +subdir: String) -> String: match root: case True{}: dir case False{}: dir ++ "/" ++ subdir# Removes one temporary path and propagates removal failures.def sweep_checked_path(+dir: String, +subdir: String) -> String: sweep_checked_path.select(String.eq(subdir, "."), dir, subdir)# Removes every temporary path with checked errors.def sweep_checked_remove_list(+names: List<&2, String>) -> IO(Result<&1, &1, U32 & String, Unit>): CompactIo.remove_list_strict_start(( Nat.add(Nat.mul(2n, List.length(&2, String, names)), 2n), names ))# Represents sweepreadstate data in recover.type SweepReadState is Type: SweepReadNeed{path: String, idx: Nat, remaining: Nat, names: List<String>} SweepReadResult{result: Result<&1, &1, U32 & String, String>, path: String, idx: Nat, remaining: Nat, names: List<String>}# Reads a directory listing through checked indexed IO.def sweep_read_checked(fuel: Nat, state: SweepReadState) -> IO(Result<&1, &1, U32 & String, List<String>>): match fuel: case 0n: IO.pure(Result<&1, &1, U32 & String, List<String>>, Fail{(7, "recovery directory read limit exceeded")}) case 1n+rest: match state: case SweepReadNeed{+path, +idx, remaining, names}: match remaining: case 0n: IO.pure(Result<&1, &1, U32 & String, List<String>>, Done{List.reverse(&1, String, names)}) case 1n+more: do IO<Result<&1, &1, U32 & String, List<String>>>: name : Result<&1, &1, U32 & String, String> <- Fs.read_dir_at(path, idx) sweep_read_checked(rest, SweepReadResult{name, path, Nat.add(idx, 1n), more, names}) case SweepReadResult{result, path, idx, remaining, names}: match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, List<String>>, Fail{error}) case Done{name}: sweep_read_checked(rest, SweepReadNeed{path, idx, remaining, Fs.grab_name(name, names)})# Handles the checked directory-entry count result.def sweep_read_checked_count( result: Result<&1, &1, U32 & String, Nat>, +path: String) -> IO(Result<&1, &1, U32 & String, List<String>>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, List<String>>, Fail{error}) case Done{+count}: sweep_read_checked(Nat.add(Nat.mul(2n, Nat.add(count, 1n)), 1n), SweepReadNeed{path, 0n, Nat.add(count, 1n), Nil{}})# Reads one indexed directory entry and preserves errors.def sweep_read_checked_path(+path: String) -> IO(Result<&1, &1, U32 & String, List<String>>): do IO<Result<&1, &1, U32 & String, List<String>>>: count : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(path) sweep_read_checked_count(count, path)# Converts a directory-read failure to the recovery result.def sweep_checked_read_error( allow_missing: Bool, +code: U32, message: String) -> IO(Result<&1, &1, U32 & String, Unit>): IO.pure(Result<&1, &1, U32 & String, Unit>, FsPolicy.optional_directory_result(allow_missing, U32.is_eq(code, 2) || U32.is_eq(code, 20), code, message))def sweep_checked_dir.result( result: Result<&1, &1, U32 & String, List<String>>, allow_missing: Bool, +path: String) -> IO(Result<&1, &1, U32 & String, Unit>): match result: case Fail{(+code, message)}: sweep_checked_read_error(allow_missing, code, message) case Done{names}: sweep_checked_remove_list(Compact.prefix(tmp_names(names), path))# Sweeps temporary files from one optional level directory.def sweep_checked_dir(+dir: String, +subdir: String) -> IO(Result<&1, &1, U32 & String, Unit>): +path = sweep_checked_path(dir, subdir) do IO<Result<&1, &1, U32 & String, Unit>>: listed : Result<&1, &1, U32 & String, List<String>> <- sweep_read_checked_path(path) sweep_checked_dir.result(listed, Bool.not(String.eq(subdir, ".")), path)# Represents sweepcheckedstate data in recover.type SweepCheckedState is Type: SweepCheckedDirs{dirs: List<&2, String>, dir: String} SweepCheckedResult{result: Result<&1, &1, U32 & String, Unit>, dirs: List<&2, String>, dir: String}# Sweeps temporary files from all optional level directories.def sweep_checked_dirs( fuel: Nat, state: SweepCheckedState) -> IO(Result<&1, &1, U32 & String, Unit>): match fuel: case 0n: IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{(U32.from_nat(7n), "recovery sweep limit exceeded")}) case 1n+rest: match state: case SweepCheckedDirs{dirs, +dir}: match dirs: case Nil{}: IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}}) case Con{subdir, tail}: do IO<Result<&1, &1, U32 & String, Unit>>: swept : Result<&1, &1, U32 & String, Unit> <- sweep_checked_dir(dir, subdir) sweep_checked_dirs(rest, SweepCheckedResult{swept, tail, dir}) case SweepCheckedResult{result, dirs, +dir}: match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error}) case Done{Unit{}}: sweep_checked_dirs(rest, SweepCheckedDirs{dirs, dir})# Runs the checked temporary-file sweep across database levels.def sweep_checked(+dir: String) -> IO(Result<&1, &1, U32 & String, Unit>): sweep_checked_dirs(11n, SweepCheckedDirs{Con{"l0", Con{"l1", Con{"l2", Con{"l3", Con{".", Nil{}}}}}}, dir})# Continues existing-open recovery after the checked sweep.def open_existing_sweep_result( result: Result<&1, &1, U32 & String, Unit>, +dir: String, +manifest: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Db.Db>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{Unit{}}: open_names_go(Pure.manifest_names_ok(Pure.mfst_lists(manifest), 0n), dir, manifest)# Loads the Manifest before opening existing database levels.def open_existing_manifest( result: Result<&1, &1, U32 & String, Manifest.Manifest>, +dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{manifest}: do IO<Result<&1, &1, U32 & String, Db.Db>>: swept : Result<&1, &1, U32 & String, Unit> <- sweep_checked(dir) open_existing_sweep_result(swept, dir, manifest)# Recovers an existing database and propagates filesystem errors.def open_existing_checked( manifest_present: Bool, +dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>): match manifest_present: case False{}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(2n), "manifest missing")}) case True{}: do IO<Result<&1, &1, U32 & String, Db.Db>>: manifest : Result<&1, &1, U32 & String, Manifest.Manifest> <- Flush.load_mfst(dir ++ "/MANIFEST") open_existing_manifest(manifest, dir)def open_existing.present( result: Result<&1, &1, U32 & String, Bool>, +dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{present}: open_existing_checked(present, dir)# Opens an existing database while the caller retains its lock.def open_existing_locked(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>): do IO<Result<&1, &1, U32 & String, Db.Db>>: present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(dir ++ "/MANIFEST") open_existing.present(present, dir)# Initializes the database state after creating its directories.def create_initialized( synced: Result<&1, &1, U32 & String, Unit>, +dir: String,) -> IO(Result<&1, &1, U32 & String, Db.Db>): match synced: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{Unit{}}: open_existing_checked(True{}, dir)# Writes and publishes the initial Manifest.def create_published( renamed: Result<&1, &1, U32 & String, Unit>, +dir: String,) -> IO(Result<&1, &1, U32 & String, Db.Db>): match renamed: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{Unit{}}: do IO<Result<&1, &1, U32 & String, Db.Db>>: synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(dir) create_initialized(synced, dir)# Initializes the WAL after publishing the Manifest.def create_written( written: Result<&1, &1, U32 & String, Unit>, +dir: String,) -> IO(Result<&1, &1, U32 & String, Db.Db>): match written: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{Unit{}}: do IO<Result<&1, &1, U32 & String, Db.Db>>: renamed : Result<&1, &1, U32 & String, Unit> <- Fs.rename(dir ++ "/MANIFEST.tmp", dir ++ "/MANIFEST") create_published(renamed, dir)# Creates a new database without acquiring another lock.def create_locked(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>): do IO<Result<&1, &1, U32 & String, Db.Db>>: written : Result<&1, &1, U32 & String, Unit> <- Flush.write_manifest(dir ++ "/MANIFEST.tmp", Manifest.M{Nil{}}) create_written(written, dir)# --- Maintenance (synchronous auto flush/compact + stall) ---# One drain round always suffices (L0 >= 8 > 4, so the gated compaction# fires and empties L0) — no loop, no cycle.# Flush dec for the database recovery effects.def flush_dec(full: Bool, +db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>): match full: case True{}: Flush.flush(db) case False{}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Done{db})# Flush gate for the database recovery effects.def flush_gate(db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>): match db: case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, +mem_count, frozen_count}: flush_dec(Nat.is_lt(4096n, mem_count), Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count})# Handle l0 full in the database recovery effects.def l0_full(db: Db.Db) -> Bool: match db: case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}: Nat.is_lt(7n, List.length(&2, Sstable.Table, Compact.l0list(levels)))# Flushes one full MemTable and returns the updated database.def drain_once.flushed( result: Result<&1, &1, U32 & String, Db.Db>) -> IO(Result<&1, &1, U32 & String, Db.Db>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{db}: CompactIo.compact(db)# Flushes one full MemTable and returns the updated database.def drain_once(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>): do IO<Result<&1, &1, U32 & String, Db.Db>>: flushed : Result<&1, &1, U32 & String, Db.Db> <- Flush.flush(db) drain_once.flushed(flushed)# Handle drain pick in the database recovery effects.def drain_pick(full: Bool, +db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>): match full: case True{}: drain_once(db) case False{}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Done{db})# Handle drain go in the database recovery effects.def drain_go(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>): drain_pick(l0_full(db), db)def maintain.flushed( result: Result<&1, &1, U32 & String, Db.Db>) -> IO(Result<&1, &1, U32 & String, Db.Db>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{db}: CompactIo.compact(db)def maintain.drained( result: Result<&1, &1, U32 & String, Db.Db>) -> IO(Result<&1, &1, U32 & String, Db.Db>): match result: case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error}) case Done{db}: do IO<Result<&1, &1, U32 & String, Db.Db>>: flushed : Result<&1, &1, U32 & String, Db.Db> <- flush_gate(db) maintain.flushed(flushed)# Runs the required flush and compaction maintenance.def maintain(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>): do IO<Result<&1, &1, U32 & String, Db.Db>>: drained : Result<&1, &1, U32 & String, Db.Db> <- drain_go(db) maintain.drained(drained)# Handle maintain unwrap in the database recovery effects.def maintain_unwrap(res: Result<&1, &1, U32 & String, Db.Db>) -> IO(Result<&1, &1, U32 & String, Db.Db>): match res: case Fail{e}: IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{e}) case Done{db}: maintain(db)# Handle write in the database recovery effects.def write(+db: Db.Db, +batch: Wal.Batch) -> IO(Result<&1, &1, U32 & String, Db.Db>): do IO<Result<&1, &1, U32 & String, Db.Db>>: r : Result<&1, &1, U32 & String, Db.Db> <- DbIo.db_write(db, batch) maintain_unwrap(r)# Handle put in the database recovery effects.def put(+db: Db.Db, +key: String, +val: String) -> IO(Result<&1, &1, U32 & String, Db.Db>): write(db, Wal.Batch{Con{Wal.Put{key, val}, Nil{}}})# Handle del in the database recovery effects.def del(+db: Db.Db, +key: String) -> IO(Result<&1, &1, U32 & String, Db.Db>): write(db, Wal.Batch{Con{Wal.Del{key}, Nil{}}})# --- Closed-vector laws live in laws/Recover.bend (spec law 7) ---