src/Flush.bend
mylsm-lsm-store@0.4.0.0 · 380 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/SortedRun.bend as SortedRun
6import mylsm-lsm-store@0.4.0.0/src/SstStreamIo.bend as SstStreamIo
7import mylsm-lsm-store@0.4.0.0/src/Wal.bend as Wal
8import mylsm-lsm-store@0.4.0.0/src/Manifest.bend as Manifest
9import mylsm-lsm-store@0.4.0.0/src/StorageBytes.bend as StorageBytes
10import mylsm-lsm-store@0.4.0.0/src/Fs.bend as Fs
11import mylsm-lsm-store@0.4.0.0/src/DbIo.bend as DbIo
12import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
13import mylsm-lsm-store@0.4.0.0/src/Db.bend as Db
14import mylsm-lsm-store@0.4.0.0/src/CrashPoint.bend as CrashPoint
15import mylsm-lsm-store@0.4.0.0/src/FlushPolicy.bend as Policy
16
17# Represent ManifestReadState data used by the memtable flush.
18type ManifestReadState is Type:
19 ManifestNeed{file: File, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
20 ManifestGot{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
21
22# Flush (Task 8): sealed memtable -> L0 SSTable file + manifest + WAL reset.
23#
24# Crash protocol (each step durable before the next, tmp + rename for
25# atomic visibility):
26# 1. ensure <dir>/l0; write <name>.tmp; fsync; rename to <name>; fsync l0.
27# 2. load manifest (absent file = fresh M{Nil}; corrupt = ABORT Fail).
28# 3. write MANIFEST.tmp; fsync; rename; fsync dir.
29# 4. remove wal.log (failure ignored: a stale WAL replays idempotently —
30# same versions, mem-first reads return equal values; merge dedups).
31# A crash before step 3 leaves the table file orphaned (recovery ignores
32# files absent from the manifest; Task 12 sweeps orphans). A crash after
33# step 3 with a live WAL replays duplicates harmlessly (see above).
34# Empty memtable flush is a no-op (no files touched).
35# CrashPoint calls are test-only host effects. Unset or unequal checkpoints are
36# successful no-ops; signal delivery and filesystem ordering are not Bend proofs.
37
38# Handle l0 add in the memtable flush.
39def l0_add(+levels: List<&2, List<&2, Sstable.Table>>, +tbl: Sstable.Table) -> List<&2, List<&2, Sstable.Table>>:
40 match levels:
41 case Nil{}:
42 Con{Con{tbl, Nil{}}, Nil{}}
43 case Con{l0, rest}:
44 Con{Con{tbl, l0}, rest}
45
46# Handle mfst add in the memtable flush.
47def mfst_add(+mfst: Manifest.Manifest, +name: String) -> Manifest.Manifest:
48 match mfst:
49 case Manifest.M{lvs}:
50 match lvs:
51 case Nil{}:
52 Manifest.M{Con{Con{name, Nil{}}, Nil{}}}
53 case Con{l0, rest}:
54 Manifest.M{Con{Con{name, l0}, rest}}
55
56# --- File helpers (pair-splits contained; straight-line callers above) ---
57
58# Write tail for the memtable flush.
59def write_tail(
60 fr: (File & Result<&1, &1, U32 & String, Unit>),
61 +path: String
62) -> IO(Result<&1, &1, U32 & String, Unit>):
63 match fr:
64 case (f2, r):
65 do IO<Result<&1, &1, U32 & String, Unit>>:
66 res : Unit <- IO.try(Unit, IO.pure(Result<&1, &1, U32 & String, Unit>, r))
67 _cls : Unit <- File.close(f2)
68 _syn : Unit <- IO.try(Unit, Fs.fsync(path))
69 _p_rm : Unit <- IO.try(Unit, Fs.chmod(path, U32.from_nat(384n)))
70 return Done{res}
71
72# Write file for the memtable flush.
73def write_file(+path: String, +data: String) -> IO(Result<&1, &1, U32 & String, Unit>):
74 do IO<Result<&1, &1, U32 & String, Unit>>:
75 f : File <- IO.try(File, File.open(path, "w"))
76 fr : (File & Result<&1, &1, U32 & String, Unit>) <- File.write(f, data)
77 write_tail(fr, path)
78
79def write_manifest.bytes(+path: String, bytes: Bytes.Bytes) -> IO(Result<&1, &1, U32 & String, Unit>):
80 do IO<Result<&1, &1, U32 & String, Unit>>:
81 f : File <- IO.try(File, File.open(path, "w"))
82 fr : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(f, bytes)
83 write_tail(fr, path)
84
85def write_manifest.encoded(
86 path: String,
87 encoded: Result<&1, &1, Manifest.Error, Bytes.Bytes>
88) -> IO(Result<&1, &1, U32 & String, Unit>):
89 match encoded:
90 case Fail{_}:
91 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{(U32.from_nat(1n), "cannot encode manifest")})
92 case Done{bytes}:
93 write_manifest.bytes(path, bytes)
94
95# Write manifest for the memtable flush.
96def write_manifest(path: String, manifest: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Unit>):
97 write_manifest.encoded(path, Manifest.serialize(manifest))
98
99# Read tail3 for the memtable flush.
100def read_tail3(fr: (File & Result<&1, &1, U32 & String, String>)) -> IO(Result<&1, &1, U32 & String, String>):
101 match fr:
102 case (f3, r1):
103 do IO<Result<&1, &1, U32 & String, String>>:
104 _cls : Unit <- File.close(f3)
105 IO.pure(Result<&1, &1, U32 & String, String>, r1)
106
107
108# Base File.read decodes each returned chunk independently. A chunk boundary can
109# split a multi-byte UTF-8 character, so table files are read in one bounded call
110# until a byte-oriented streaming effect with decoder carry is available.
111def read_file(+path: String) -> IO(Result<&1, &1, U32 & String, String>):
112 do IO<Result<&1, &1, U32 & String, String>>:
113 f : File <- IO.try(File, File.open(path, "r"))
114 fr : (File & Result<&1, &1, U32 & String, String>) <- File.read(f, U32.from_nat(4294967295n))
115 read_tail3(fr)
116
117# Handle mfst parsed in the memtable flush.
118def mfst_parsed(+opt: Maybe<&2, Manifest.Manifest>) -> Result<&1, &1, U32 & String, Manifest.Manifest>:
119 match opt:
120 case None{}:
121 Fail{(U32.from_nat(1n), "bad manifest")}
122 case Some{mm}:
123 Done{mm}
124
125def manifest.read.result(
126 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
127) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
128 match pair:
129 case (file, Fail{error}):
130 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
131 _closed : Unit <- File.close(file)
132 return Fail{error}
133 case (file, Done{bytes}):
134 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
135 _closed : Unit <- File.close(file)
136 return mfst_parsed(Manifest.parse(bytes))
137
138def manifest.read.loop(fuel: Nat, state: ManifestReadState) -> IO(File & Result<&1, &1, U32 & String, Bytes.Bytes>):
139 match fuel:
140 case 0n:
141 match state:
142 case ManifestNeed{file, +remaining, chunks}:
143 match remaining:
144 case 0n:
145 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Done{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks))}))
146 case 1n+_:
147 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(7n), "bounded manifest read exhausted")}))
148 case ManifestGot{pair, _, _}:
149 match pair:
150 case (file, Fail{error}):
151 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{error}))
152 case (file, Done{_}):
153 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(7n), "bounded manifest read exhausted")}))
154 case 1n+rest:
155 match state:
156 case ManifestNeed{file, +remaining, chunks}:
157 match remaining:
158 case 0n:
159 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Done{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks))}))
160 case 1n+_:
161 do IO<File & Result<&1, &1, U32 & String, Bytes.Bytes>>:
162 pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <- Fs.read_bytes(file, U32.from_nat(remaining))
163 manifest.read.loop(rest, ManifestGot{pair, remaining, chunks})
164 case ManifestGot{pair, +remaining, chunks}:
165 match pair:
166 case (file, Fail{error}):
167 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{error}))
168 case (file, Done{Bytes.Bytes{0, _}}):
169 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(8n), "truncated manifest file")}))
170 case (file, Done{Bytes.Bytes{+len, buf}}):
171 manifest.read.loop(rest, ManifestNeed{file, Nat.sub(remaining, U32.to_nat(len)), Con{Bytes.Bytes{len, buf}, chunks}})
172
173def manifest.read.size.run(file: File, +size: Nat) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
174 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
175 pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <- manifest.read.loop(Nat.add(size, 1n), ManifestNeed{file, size, Nil{}})
176 manifest.read.result(pair)
177
178def manifest.read.size.valid(
179 file: File,
180 size: Nat,
181 within: Bool
182) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
183 match within:
184 case False{}:
185 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
186 _closed : Unit <- File.close(file)
187 return Fail{(U32.from_nat(9n), "manifest too large")}
188 case True{}:
189 manifest.read.size.run(file, size)
190
191def manifest.read.size(
192 file: File,
193 result: Result<&1, &1, U32 & String, Nat>
194) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
195 match result:
196 case Fail{error}:
197 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
198 _closed : Unit <- File.close(file)
199 return Fail{error}
200 case Done{+size}:
201 manifest.read.size.valid(file, size, Nat.is_le(size, U32.to_nat(StorageBytes.MAX_MANIFEST_BYTES())))
202
203# Handle mfst open error in the memtable flush.
204def mfst_open_error(
205 missing: Bool,
206 +code: U32
207) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
208 match missing:
209 case True{}:
210 IO.pure(Result<&1, &1, U32 & String, Manifest.Manifest>, Done{Manifest.M{Nil{}}})
211 case False{}:
212 IO.pure(Result<&1, &1, U32 & String, Manifest.Manifest>, Fail{(code, "cannot open manifest")})
213
214# Handle mfst opened in the memtable flush.
215def mfst_opened(
216 path: String,
217 result: Result<&1, &1, U32 & String, File>
218) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
219 match result:
220 case Fail{(+code, _)}:
221 mfst_open_error(U32.is_eq(code, 2), code)
222 case Done{file}:
223 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
224 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path)
225 manifest.read.size(file, size)
226
227# Handle load mfst in the memtable flush.
228def load_mfst(+path: String) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
229 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
230 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
231 mfst_opened(path, opened)
232
233# Handle dir tail in the memtable flush.
234def dir_tail(res: Result<&1, &1, U32 & String, Nat>, +path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
235 match res:
236 case Done{n}:
237 IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
238 case Fail{e}:
239 Fs.make_dir(path)
240
241# Handle ensure dir in the memtable flush.
242def ensure_dir(+path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
243 do IO<Result<&1, &1, U32 & String, Unit>>:
244 r : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(path)
245 dir_tail(r, path)
246
247# --- Flush chain (defined bottom-up: tails first, entry last) ---
248
249# Flush goB for the memtable flush.
250def flush_goB(
251 +dir: String,
252 +mpath: String,
253 +mtmp: String,
254 +m2: Manifest.Manifest,
255 +tbl: Sstable.Table,
256 +levels: List<&2, List<&2, Sstable.Table>>,
257 +flushed: Nat,
258 +mem: MemTable.MemTable,
259 +batch_cap: Nat,
260) -> IO(Result<&1, &1, U32 & String, Db.Db>):
261 do IO<Result<&1, &1, U32 & String, Db.Db>>:
262 _v1 : Unit <- IO.try(Unit, write_manifest(mtmp, m2))
263 _cp1 : Unit <- IO.try(Unit, CrashPoint.hit("flush.manifest_synced"))
264 _v2 : Unit <- IO.try(Unit, Fs.rename(mtmp, mpath))
265 _v3 : Unit <- IO.try(Unit, Fs.fsync(dir))
266 _cp2 : Unit <- IO.try(Unit, CrashPoint.hit("flush.manifest_published"))
267 _rm : Result<&1, &1, U32 & String, Unit> <- Fs.remove(Db.wal_path(dir))
268 _wal : Unit <- IO.try(Unit, DbIo.wal_initialize(dir))
269 return Done{Db.Db{dir, mem, MemTable.MT{Nil{}}, batch_cap, Nil{}, l0_add(levels, tbl), Nat.add(flushed, 1n), Manifest.token(m2), 0n, 0n}}
270
271# Flush drift for the memtable flush.
272def flush_drift(
273 ok: Bool,
274 +dir: String,
275 +mpath: String,
276 +mtmp: String,
277 +mfst: Manifest.Manifest,
278 +tbl: Sstable.Table,
279 +name: String,
280 +levels: List<&2, List<&2, Sstable.Table>>,
281 +flushed: Nat,
282 +mem: MemTable.MemTable,
283 +batch_cap: Nat,
284) -> IO(Result<&1, &1, U32 & String, Db.Db>):
285 match ok:
286 case False{}:
287 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(2n), "manifest drift")})
288 case True{}:
289 flush_goB(dir, mpath, mtmp, mfst_add(mfst, name), tbl, levels, flushed, mem, batch_cap)
290
291# Flush goA for the memtable flush.
292def flush_goA(
293 +dir: String,
294 +l0dir: String,
295 +tmp: String,
296 +final: String,
297 +mpath: String,
298 +mtmp: String,
299 +entries: List<&2, MemTable.Entry>,
300 +level: U32,
301 +tbl: Sstable.Table,
302 +name: String,
303 +levels: List<&2, List<&2, Sstable.Table>>,
304 +flushed: Nat,
305 +manifest_token: String,
306 +mem: MemTable.MemTable,
307 +batch_cap: Nat,
308) -> IO(Result<&1, &1, U32 & String, Db.Db>):
309 do IO<Result<&1, &1, U32 & String, Db.Db>>:
310 _u1 : Unit <- IO.try(Unit, ensure_dir(l0dir))
311 _u2 : Unit <- IO.try(Unit, SstStreamIo.write_table(tmp, entries, level))
312 _cp1 : Unit <- IO.try(Unit, CrashPoint.hit("flush.table_synced"))
313 _u3 : Unit <- IO.try(Unit, Fs.rename(tmp, final))
314 _u4 : Unit <- IO.try(Unit, Fs.fsync(l0dir))
315 _cp2 : Unit <- IO.try(Unit, CrashPoint.hit("flush.table_published"))
316 +m : Manifest.Manifest <- IO.try(Manifest.Manifest, load_mfst(mpath))
317 flush_drift(String.eq(Manifest.token(m), manifest_token), dir, mpath, mtmp, m, tbl, name, levels, flushed, mem, batch_cap)
318
319# Flush pre for the memtable flush.
320def flush_pre(
321 +dir: String,
322 +entries: List<&2, MemTable.Entry>,
323 +levels: List<&2, List<&2, Sstable.Table>>,
324 +flushed: Nat,
325 +manifest_token: String,
326 +batch_cap: Nat,
327) -> IO(Result<&1, &1, U32 & String, Db.Db>):
328 +tbl = Sstable.build(entries, 0n, List.length(&2, MemTable.Entry, entries))
329 +name = Policy.table_name(flushed)
330 +l0dir = dir ++ "/l0"
331 flush_goA(dir, l0dir, l0dir ++ "/" ++ name ++ ".tmp", l0dir ++ "/" ++ name, dir ++ "/MANIFEST", dir ++ "/MANIFEST.tmp", Db.table_entries(tbl), 0, tbl, name, levels, flushed,
332 manifest_token, MemTable.MT{Nil{}}, batch_cap)
333
334# Frozen-path flush MERGES mem (newer) over frozen (older) into one L0 table
335# and drains both: the WAL is truncated on publish, so anything left in mem
336# would otherwise exist nowhere on disk (loss window caught by 20485 smoke).
337def flush_frozen(
338 +dir: String,
339 +mem: MemTable.MemTable,
340 +entries: List<&2, MemTable.Entry>,
341 +levels: List<&2, List<&2, Sstable.Table>>,
342 +flushed: Nat,
343 +manifest_token: String,
344 +batch_cap: Nat,
345) -> IO(Result<&1, &1, U32 & String, Db.Db>):
346 match mem:
347 case MemTable.MT{live}:
348 +merged = SortedRun.merge_many_newest(Con{live, Con{entries, Nil{}}})
349 +tbl = Sstable.build(merged, 0n, List.length(&2, MemTable.Entry, merged))
350 +name = Policy.table_name(flushed)
351 +l0dir = dir ++ "/l0"
352 flush_goA(dir, l0dir, l0dir ++ "/" ++ name ++ ".tmp", l0dir ++ "/" ++ name, dir ++ "/MANIFEST", dir ++ "/MANIFEST.tmp", Db.table_entries(tbl), 0, tbl, name, levels, flushed,
353 manifest_token, MemTable.MT{Nil{}}, batch_cap)
354
355# Flush pick for the memtable flush.
356def flush_pick(
357 +frozen: MemTable.MemTable,
358 +mem: MemTable.MemTable,
359 +dir: String,
360 +levels: List<&2, List<&2, Sstable.Table>>,
361 +flushed: Nat,
362 +manifest_token: String,
363 +batch_cap: Nat,
364 +bcache: List<&2, Db.BEntry>,
365) -> IO(Result<&1, &1, U32 & String, Db.Db>):
366 match frozen mem:
367 case MemTable.MT{Nil{}} MemTable.MT{Nil{}}:
368 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Done{Db.Db{dir, MemTable.MT{Nil{}}, MemTable.MT{Nil{}}, batch_cap, bcache, levels, flushed, manifest_token, 0n, 0n}})
369 case MemTable.MT{Nil{}} MemTable.MT{Con{e, t}}:
370 flush_pre(dir, Con{e, t}, levels, flushed, manifest_token, batch_cap)
371 case MemTable.MT{Con{fe, ft}} _:
372 flush_frozen(dir, mem, Con{fe, ft}, levels, flushed, manifest_token, batch_cap)
373
374# Handle flush in the memtable flush.
375def flush(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
376 match db:
377 case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
378 flush_pick(frozen, mem, dir, levels, flushed, manifest_token, batch_cap, bcache)
379
380# --- Closed-vector laws live in laws/Flush.bend (spec law 6) ---