src/Flush.bend
mylsm-lsm-store@0.5.0.0 · 621 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/SortedRun.bend as SortedRun
6import mylsm-lsm-store@0.5.0.0/src/SstStreamIo.bend as SstStreamIo
7import mylsm-lsm-store@0.5.0.0/src/Wal.bend as Wal
8import mylsm-lsm-store@0.5.0.0/src/Manifest.bend as Manifest
9import mylsm-lsm-store@0.5.0.0/src/StorageBytes.bend as StorageBytes
10import mylsm-lsm-store@0.5.0.0/src/Fs.bend as Fs
11import mylsm-lsm-store@0.5.0.0/src/DbIo.bend as DbIo
12import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
13import mylsm-lsm-store@0.5.0.0/src/Db.bend as Db
14import mylsm-lsm-store@0.5.0.0/src/CrashPoint.bend as CrashPoint
15import mylsm-lsm-store@0.5.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
58def write_tail.chmod(
59 result: Result<&1, &1, U32 & String, Unit>,
60 +path: String
61) -> IO(Result<&1, &1, U32 & String, Unit>):
62 match result:
63 case Fail{error}:
64 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
65 case Done{Unit{}}:
66 Fs.chmod(path, U32.from_nat(384n))
67
68def write_tail.fsync(
69 result: Result<&1, &1, U32 & String, Unit>,
70 +path: String
71) -> IO(Result<&1, &1, U32 & String, Unit>):
72 match result:
73 case Fail{error}:
74 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
75 case Done{Unit{}}:
76 do IO<Result<&1, &1, U32 & String, Unit>>:
77 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path)
78 write_tail.chmod(synced, path)
79
80# Writes the remaining bytes of a file buffer.
81def write_tail(
82 fr: File & Result<&1, &1, U32 & String, Unit>,
83 +path: String
84) -> IO(Result<&1, &1, U32 & String, Unit>):
85 match fr:
86 case (file, result):
87 do IO<Result<&1, &1, U32 & String, Unit>>:
88 _closed : Unit <- File.close(file)
89 write_tail.fsync(result, path)
90
91def write_file.opened(
92 opened: Result<&1, &1, U32 & String, File>,
93 +path: String,
94 +data: String
95) -> IO(Result<&1, &1, U32 & String, Unit>):
96 match opened:
97 case Fail{error}:
98 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
99 case Done{file}:
100 do IO<Result<&1, &1, U32 & String, Unit>>:
101 written : File & Result<&1, &1, U32 & String, Unit> <- File.write(file, data)
102 write_tail(written, path)
103
104# Creates and syncs one storage file.
105def write_file(+path: String, +data: String) -> IO(Result<&1, &1, U32 & String, Unit>):
106 do IO<Result<&1, &1, U32 & String, Unit>>:
107 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "w")
108 write_file.opened(opened, path, data)
109
110def write_manifest.bytes.opened(
111 opened: Result<&1, &1, U32 & String, File>,
112 +path: String,
113 bytes: Bytes.Bytes
114) -> IO(Result<&1, &1, U32 & String, Unit>):
115 match opened:
116 case Fail{error}:
117 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
118 case Done{file}:
119 do IO<Result<&1, &1, U32 & String, Unit>>:
120 written : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(file, bytes)
121 write_tail(written, path)
122
123def write_manifest.bytes(+path: String, bytes: Bytes.Bytes) -> IO(Result<&1, &1, U32 & String, Unit>):
124 do IO<Result<&1, &1, U32 & String, Unit>>:
125 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "w")
126 write_manifest.bytes.opened(opened, path, bytes)
127
128def write_manifest.encoded(
129 path: String,
130 encoded: Result<&1, &1, Manifest.Error, Bytes.Bytes>
131) -> IO(Result<&1, &1, U32 & String, Unit>):
132 match encoded:
133 case Fail{_}:
134 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{(U32.from_nat(1n), "cannot encode manifest")})
135 case Done{bytes}:
136 write_manifest.bytes(path, bytes)
137
138# Write manifest for the memtable flush.
139def write_manifest(path: String, manifest: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Unit>):
140 write_manifest.encoded(path, Manifest.serialize(manifest))
141
142# Read tail3 for the memtable flush.
143def read_tail3(fr: (File & Result<&1, &1, U32 & String, String>)) -> IO(Result<&1, &1, U32 & String, String>):
144 match fr:
145 case (f3, r1):
146 do IO<Result<&1, &1, U32 & String, String>>:
147 _cls : Unit <- File.close(f3)
148 IO.pure(Result<&1, &1, U32 & String, String>, r1)
149
150
151# Base File.read decodes each returned chunk independently. A chunk boundary can
152# split a multi-byte UTF-8 character, so table files are read in one bounded call
153# Reads a bounded storage file and returns its bytes.
154def read_file.opened(
155 opened: Result<&1, &1, U32 & String, File>
156) -> IO(Result<&1, &1, U32 & String, String>):
157 match opened:
158 case Fail{error}:
159 IO.pure(Result<&1, &1, U32 & String, String>, Fail{error})
160 case Done{file}:
161 do IO<Result<&1, &1, U32 & String, String>>:
162 result : File & Result<&1, &1, U32 & String, String> <- File.read(file, U32.from_nat(4294967295n))
163 read_tail3(result)
164
165# Reads a bounded storage file and returns its bytes.
166def read_file(+path: String) -> IO(Result<&1, &1, U32 & String, String>):
167 do IO<Result<&1, &1, U32 & String, String>>:
168 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
169 read_file.opened(opened)
170
171# Classifies the Manifest parse result before recovery.
172def mfst_parsed(parsed: Manifest.ParseResult) -> Result<&1, &1, U32 & String, Manifest.Manifest>:
173 match parsed:
174 case Manifest.Parsed{manifest}:
175 Done{manifest}
176 case Manifest.UnsupportedVersion{_}:
177 Fail{(Manifest.unsupported_version_error_code(), "unsupported manifest version")}
178 case Manifest.Malformed{}:
179 Fail{(Manifest.malformed_error_code(), "malformed manifest")}
180 case Manifest.ParseLimitExceeded{}:
181 Fail{(Manifest.parse_limit_error_code(), "manifest size limit exceeded")}
182
183def manifest.read.result(
184 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
185) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
186 match pair:
187 case (file, Fail{error}):
188 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
189 _closed : Unit <- File.close(file)
190 return Fail{error}
191 case (file, Done{bytes}):
192 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
193 _closed : Unit <- File.close(file)
194 return mfst_parsed(Manifest.parse_result(bytes))
195
196def manifest.read.loop(fuel: Nat, state: ManifestReadState) -> IO(File & Result<&1, &1, U32 & String, Bytes.Bytes>):
197 match fuel:
198 case 0n:
199 match state:
200 case ManifestNeed{file, +remaining, chunks}:
201 match remaining:
202 case 0n:
203 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Done{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks))}))
204 case 1n+_:
205 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(7n), "bounded manifest read exhausted")}))
206 case ManifestGot{pair, _, _}:
207 match pair:
208 case (file, Fail{error}):
209 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{error}))
210 case (file, Done{_}):
211 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(7n), "bounded manifest read exhausted")}))
212 case 1n+rest:
213 match state:
214 case ManifestNeed{file, +remaining, chunks}:
215 match remaining:
216 case 0n:
217 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Done{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks))}))
218 case 1n+_:
219 do IO<File & Result<&1, &1, U32 & String, Bytes.Bytes>>:
220 pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <- Fs.read_bytes(file, U32.from_nat(remaining))
221 manifest.read.loop(rest, ManifestGot{pair, remaining, chunks})
222 case ManifestGot{pair, +remaining, chunks}:
223 match pair:
224 case (file, Fail{error}):
225 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{error}))
226 case (file, Done{Bytes.Bytes{0, _}}):
227 IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(8n), "truncated manifest file")}))
228 case (file, Done{Bytes.Bytes{+len, buf}}):
229 manifest.read.loop(rest, ManifestNeed{file, Nat.sub(remaining, U32.to_nat(len)), Con{Bytes.Bytes{len, buf}, chunks}})
230
231def manifest.read.size.run(file: File, +size: Nat) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
232 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
233 pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <- manifest.read.loop(Nat.add(size, 1n), ManifestNeed{file, size, Nil{}})
234 manifest.read.result(pair)
235
236def manifest.read.size.valid(
237 file: File,
238 size: Nat,
239 within: Bool
240) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
241 match within:
242 case False{}:
243 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
244 _closed : Unit <- File.close(file)
245 return Fail{(U32.from_nat(9n), "manifest too large")}
246 case True{}:
247 manifest.read.size.run(file, size)
248
249def manifest.read.size(
250 file: File,
251 result: Result<&1, &1, U32 & String, Nat>
252) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
253 match result:
254 case Fail{error}:
255 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
256 _closed : Unit <- File.close(file)
257 return Fail{error}
258 case Done{+size}:
259 manifest.read.size.valid(file, size, Nat.is_le(size, U32.to_nat(StorageBytes.MAX_MANIFEST_BYTES())))
260
261# Handle mfst open error in the memtable flush.
262def mfst_open_error(
263 missing: Bool,
264 +code: U32
265) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
266 match missing:
267 case True{}:
268 IO.pure(Result<&1, &1, U32 & String, Manifest.Manifest>, Done{Manifest.M{Nil{}}})
269 case False{}:
270 IO.pure(Result<&1, &1, U32 & String, Manifest.Manifest>, Fail{(code, "cannot open manifest")})
271
272# Handle mfst opened in the memtable flush.
273def mfst_opened(
274 path: String,
275 result: Result<&1, &1, U32 & String, File>
276) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
277 match result:
278 case Fail{(+code, _)}:
279 mfst_open_error(U32.is_eq(code, 2), code)
280 case Done{file}:
281 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
282 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path)
283 manifest.read.size(file, size)
284
285# Handle load mfst in the memtable flush.
286def load_mfst(+path: String) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
287 do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
288 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
289 mfst_opened(path, opened)
290
291# Handle dir tail in the memtable flush.
292def dir_tail(res: Result<&1, &1, U32 & String, Nat>, +path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
293 match res:
294 case Done{n}:
295 IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
296 case Fail{e}:
297 Fs.make_dir(path)
298
299# Handle ensure dir in the memtable flush.
300def ensure_dir(+path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
301 do IO<Result<&1, &1, U32 & String, Unit>>:
302 r : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(path)
303 dir_tail(r, path)
304
305# --- Flush chain (defined bottom-up: tails first, entry last) ---
306
307def flush_goB.wal(
308 result: Result<&1, &1, U32 & String, Unit>,
309 +dir: String,
310 +m2: Manifest.Manifest,
311 +tbl: Sstable.Table,
312 +levels: List<&2, List<&2, Sstable.Table>>,
313 +flushed: Nat,
314 +mem: MemTable.MemTable,
315 +batch_cap: Nat
316) -> IO(Result<&1, &1, U32 & String, Db.Db>):
317 match result:
318 case Fail{error}:
319 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
320 case Done{Unit{}}:
321 IO.pure(Result<&1, &1, U32 & String, Db.Db>,
322 Done{Db.Db{dir, mem, MemTable.MT{Nil{}}, batch_cap, Nil{},
323 l0_add(levels, tbl), Nat.add(flushed, 1n), Manifest.token(m2), 0n, 0n}})
324
325def flush_goB.synced(
326 result: Result<&1, &1, U32 & String, Unit>,
327 +dir: String,
328 +m2: Manifest.Manifest,
329 +tbl: Sstable.Table,
330 +levels: List<&2, List<&2, Sstable.Table>>,
331 +flushed: Nat,
332 +mem: MemTable.MemTable,
333 +batch_cap: Nat
334) -> IO(Result<&1, &1, U32 & String, Db.Db>):
335 match result:
336 case Fail{error}:
337 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
338 case Done{Unit{}}:
339 do IO<Result<&1, &1, U32 & String, Db.Db>>:
340 _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("flush.manifest_published"))
341 _removed : Result<&1, &1, U32 & String, Unit> <- Fs.remove(Db.wal_path(dir))
342 initialized : Result<&1, &1, U32 & String, Unit> <- DbIo.wal_initialize(dir)
343 flush_goB.wal(initialized, dir, m2, tbl, levels, flushed, mem, batch_cap)
344
345def flush_goB.renamed(
346 result: Result<&1, &1, U32 & String, Unit>,
347 +dir: String,
348 +m2: Manifest.Manifest,
349 +tbl: Sstable.Table,
350 +levels: List<&2, List<&2, Sstable.Table>>,
351 +flushed: Nat,
352 +mem: MemTable.MemTable,
353 +batch_cap: Nat
354) -> IO(Result<&1, &1, U32 & String, Db.Db>):
355 match result:
356 case Fail{error}:
357 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
358 case Done{Unit{}}:
359 do IO<Result<&1, &1, U32 & String, Db.Db>>:
360 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(dir)
361 flush_goB.synced(synced, dir, m2, tbl, levels, flushed, mem, batch_cap)
362
363def flush_goB.written(
364 result: Result<&1, &1, U32 & String, Unit>,
365 +dir: String,
366 +mpath: String,
367 +mtmp: String,
368 +m2: Manifest.Manifest,
369 +tbl: Sstable.Table,
370 +levels: List<&2, List<&2, Sstable.Table>>,
371 +flushed: Nat,
372 +mem: MemTable.MemTable,
373 +batch_cap: Nat
374) -> IO(Result<&1, &1, U32 & String, Db.Db>):
375 match result:
376 case Fail{error}:
377 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
378 case Done{Unit{}}:
379 do IO<Result<&1, &1, U32 & String, Db.Db>>:
380 _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("flush.manifest_synced"))
381 renamed : Result<&1, &1, U32 & String, Unit> <- Fs.rename(mtmp, mpath)
382 flush_goB.renamed(renamed, dir, m2, tbl, levels, flushed, mem, batch_cap)
383
384# Publishes flushed SSTables and updates the Manifest.
385def flush_goB(
386 +dir: String,
387 +mpath: String,
388 +mtmp: String,
389 +m2: Manifest.Manifest,
390 +tbl: Sstable.Table,
391 +levels: List<&2, List<&2, Sstable.Table>>,
392 +flushed: Nat,
393 +mem: MemTable.MemTable,
394 +batch_cap: Nat,
395) -> IO(Result<&1, &1, U32 & String, Db.Db>):
396 do IO<Result<&1, &1, U32 & String, Db.Db>>:
397 written : Result<&1, &1, U32 & String, Unit> <- write_manifest(mtmp, m2)
398 flush_goB.written(written, dir, mpath, mtmp, m2, tbl, levels, flushed, mem, batch_cap)
399
400# Flush drift for the memtable flush.
401def flush_drift(
402 ok: Bool,
403 +dir: String,
404 +mpath: String,
405 +mtmp: String,
406 +mfst: Manifest.Manifest,
407 +tbl: Sstable.Table,
408 +name: String,
409 +levels: List<&2, List<&2, Sstable.Table>>,
410 +flushed: Nat,
411 +mem: MemTable.MemTable,
412 +batch_cap: Nat,
413) -> IO(Result<&1, &1, U32 & String, Db.Db>):
414 match ok:
415 case False{}:
416 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(2n), "manifest drift")})
417 case True{}:
418 flush_goB(dir, mpath, mtmp, mfst_add(mfst, name), tbl, levels, flushed, mem, batch_cap)
419
420def flush_goA.loaded(
421 result: Result<&1, &1, U32 & String, Manifest.Manifest>,
422 +dir: String,
423 +mpath: String,
424 +mtmp: String,
425 +tbl: Sstable.Table,
426 +name: String,
427 +levels: List<&2, List<&2, Sstable.Table>>,
428 +flushed: Nat,
429 +manifest_token: String,
430 +mem: MemTable.MemTable,
431 +batch_cap: Nat
432) -> IO(Result<&1, &1, U32 & String, Db.Db>):
433 match result:
434 case Fail{error}:
435 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
436 case Done{+current}:
437 flush_drift(String.eq(Manifest.token(current), manifest_token), dir,
438 mpath, mtmp, current, tbl, name, levels, flushed, mem, batch_cap)
439
440def flush_goA.synced(
441 result: Result<&1, &1, U32 & String, Unit>,
442 +dir: String,
443 +mpath: String,
444 +mtmp: String,
445 +tbl: Sstable.Table,
446 +name: String,
447 +levels: List<&2, List<&2, Sstable.Table>>,
448 +flushed: Nat,
449 +manifest_token: String,
450 +mem: MemTable.MemTable,
451 +batch_cap: Nat
452) -> IO(Result<&1, &1, U32 & String, Db.Db>):
453 match result:
454 case Fail{error}:
455 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
456 case Done{Unit{}}:
457 do IO<Result<&1, &1, U32 & String, Db.Db>>:
458 _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("flush.table_published"))
459 current : Result<&1, &1, U32 & String, Manifest.Manifest> <- load_mfst(mpath)
460 flush_goA.loaded(current, dir, mpath, mtmp, tbl, name, levels,
461 flushed, manifest_token, mem, batch_cap)
462
463def flush_goA.renamed(
464 result: Result<&1, &1, U32 & String, Unit>,
465 +dir: String,
466 +l0dir: String,
467 +mpath: String,
468 +mtmp: String,
469 +tbl: Sstable.Table,
470 +name: String,
471 +levels: List<&2, List<&2, Sstable.Table>>,
472 +flushed: Nat,
473 +manifest_token: String,
474 +mem: MemTable.MemTable,
475 +batch_cap: Nat
476) -> IO(Result<&1, &1, U32 & String, Db.Db>):
477 match result:
478 case Fail{error}:
479 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
480 case Done{Unit{}}:
481 do IO<Result<&1, &1, U32 & String, Db.Db>>:
482 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(l0dir)
483 flush_goA.synced(synced, dir, mpath, mtmp, tbl, name, levels,
484 flushed, manifest_token, mem, batch_cap)
485
486# Writes the MemTable to an SSTable and completes the flush.
487def flush_goA.written(
488 result: Result<&1, &1, U32 & String, Unit>,
489 +dir: String,
490 +l0dir: String,
491 +tmp: String,
492 +final: String,
493 +mpath: String,
494 +mtmp: String,
495 +tbl: Sstable.Table,
496 +name: String,
497 +levels: List<&2, List<&2, Sstable.Table>>,
498 +flushed: Nat,
499 +manifest_token: String,
500 +mem: MemTable.MemTable,
501 +batch_cap: Nat,
502) -> IO(Result<&1, &1, U32 & String, Db.Db>):
503 match result:
504 case Fail{error}:
505 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
506 case Done{Unit{}}:
507 do IO<Result<&1, &1, U32 & String, Db.Db>>:
508 _cp1 : Unit <- IO.try(Unit, CrashPoint.hit("flush.table_synced"))
509 renamed : Result<&1, &1, U32 & String, Unit> <- Fs.rename(tmp, final)
510 flush_goA.renamed(renamed, dir, l0dir, mpath, mtmp, tbl, name, levels, flushed, manifest_token, mem, batch_cap)
511
512def flush_goA.ensured(
513 result: Result<&1, &1, U32 & String, Unit>,
514 +dir: String,
515 +l0dir: String,
516 +tmp: String,
517 +final: String,
518 +mpath: String,
519 +mtmp: String,
520 +entries: List<&2, MemTable.Entry>,
521 +level: U32,
522 +tbl: Sstable.Table,
523 +name: String,
524 +levels: List<&2, List<&2, Sstable.Table>>,
525 +flushed: Nat,
526 +manifest_token: String,
527 +mem: MemTable.MemTable,
528 +batch_cap: Nat,
529) -> IO(Result<&1, &1, U32 & String, Db.Db>):
530 match result:
531 case Fail{error}:
532 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
533 case Done{Unit{}}:
534 do IO<Result<&1, &1, U32 & String, Db.Db>>:
535 written : Result<&1, &1, U32 & String, Unit> <- SstStreamIo.write_table(tmp, entries, level)
536 flush_goA.written(written, dir, l0dir, tmp, final, mpath, mtmp, tbl, name, levels, flushed, manifest_token, mem, batch_cap)
537
538# Writes the MemTable to an SSTable and completes the flush.
539def flush_goA(
540 +dir: String,
541 +l0dir: String,
542 +tmp: String,
543 +final: String,
544 +mpath: String,
545 +mtmp: String,
546 +entries: List<&2, MemTable.Entry>,
547 +level: U32,
548 +tbl: Sstable.Table,
549 +name: String,
550 +levels: List<&2, List<&2, Sstable.Table>>,
551 +flushed: Nat,
552 +manifest_token: String,
553 +mem: MemTable.MemTable,
554 +batch_cap: Nat,
555) -> IO(Result<&1, &1, U32 & String, Db.Db>):
556 do IO<Result<&1, &1, U32 & String, Db.Db>>:
557 ensured : Result<&1, &1, U32 & String, Unit> <- ensure_dir(l0dir)
558 flush_goA.ensured(ensured, dir, l0dir, tmp, final, mpath, mtmp, entries, level, tbl, name, levels, flushed, manifest_token, mem, batch_cap)
559
560# Flush pre for the memtable flush.
561def flush_pre(
562 +dir: String,
563 +entries: List<&2, MemTable.Entry>,
564 +levels: List<&2, List<&2, Sstable.Table>>,
565 +flushed: Nat,
566 +manifest_token: String,
567 +batch_cap: Nat,
568) -> IO(Result<&1, &1, U32 & String, Db.Db>):
569 +tbl = Sstable.build(entries, 0n, List.length(&2, MemTable.Entry, entries))
570 +name = Policy.table_name(flushed)
571 +l0dir = dir ++ "/l0"
572 flush_goA(dir, l0dir, l0dir ++ "/" ++ name ++ ".tmp", l0dir ++ "/" ++ name, dir ++ "/MANIFEST", dir ++ "/MANIFEST.tmp", Db.table_entries(tbl), 0, tbl, name, levels, flushed,
573 manifest_token, MemTable.MT{Nil{}}, batch_cap)
574
575# Frozen-path flush MERGES mem (newer) over frozen (older) into one L0 table
576# and drains both: the WAL is truncated on publish, so anything left in mem
577# would otherwise exist nowhere on disk (loss window caught by 20485 smoke).
578def flush_frozen(
579 +dir: String,
580 +mem: MemTable.MemTable,
581 +entries: List<&2, MemTable.Entry>,
582 +levels: List<&2, List<&2, Sstable.Table>>,
583 +flushed: Nat,
584 +manifest_token: String,
585 +batch_cap: Nat,
586) -> IO(Result<&1, &1, U32 & String, Db.Db>):
587 match mem:
588 case MemTable.MT{live}:
589 +merged = SortedRun.merge_many_newest(Con{live, Con{entries, Nil{}}})
590 +tbl = Sstable.build(merged, 0n, List.length(&2, MemTable.Entry, merged))
591 +name = Policy.table_name(flushed)
592 +l0dir = dir ++ "/l0"
593 flush_goA(dir, l0dir, l0dir ++ "/" ++ name ++ ".tmp", l0dir ++ "/" ++ name, dir ++ "/MANIFEST", dir ++ "/MANIFEST.tmp", Db.table_entries(tbl), 0, tbl, name, levels, flushed,
594 manifest_token, MemTable.MT{Nil{}}, batch_cap)
595
596# Flush pick for the memtable flush.
597def flush_pick(
598 +frozen: MemTable.MemTable,
599 +mem: MemTable.MemTable,
600 +dir: String,
601 +levels: List<&2, List<&2, Sstable.Table>>,
602 +flushed: Nat,
603 +manifest_token: String,
604 +batch_cap: Nat,
605 +bcache: List<&2, Db.BEntry>,
606) -> IO(Result<&1, &1, U32 & String, Db.Db>):
607 match frozen mem:
608 case MemTable.MT{Nil{}} MemTable.MT{Nil{}}:
609 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}})
610 case MemTable.MT{Nil{}} MemTable.MT{Con{e, t}}:
611 flush_pre(dir, Con{e, t}, levels, flushed, manifest_token, batch_cap)
612 case MemTable.MT{Con{fe, ft}} _:
613 flush_frozen(dir, mem, Con{fe, ft}, levels, flushed, manifest_token, batch_cap)
614
615# Handle flush in the memtable flush.
616def flush(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
617 match db:
618 case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
619 flush_pick(frozen, mem, dir, levels, flushed, manifest_token, batch_cap, bcache)
620
621# --- Closed-vector laws live in laws/Flush.bend (spec law 6) ---