1	import Base
2	import ./Keys.bend as Keys
3	import ./MemTable.bend as MemTable
4	import ./Sstable.bend as Sstable
5	import ./SortedRun.bend as SortedRun
6	import ./SstStreamIo.bend as SstStreamIo
7	import ./Wal.bend as Wal
8	import ./Manifest.bend as Manifest
9	import ./StorageBytes.bend as StorageBytes
10	import ./Fs.bend as Fs
11	import ./DbIo.bend as DbIo
12	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
13	import ./Db.bend as Db
14	import ./CrashPoint.bend as CrashPoint
15	import ./FlushPolicy.bend as Policy
16	
17	# Represent ManifestReadState data used by the memtable flush.
18	type 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.
39	def 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.
47	def 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	def 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	
68	def 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.
81	def 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	
91	def 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.
105	def 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	
110	def 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	
123	def 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	
128	def 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.
139	def 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.
143	def 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.
154	def 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.
166	def 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.
172	def 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	
183	def 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	
196	def 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	
231	def 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	
236	def 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	
249	def 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.
262	def 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.
273	def 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.
286	def 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.
292	def 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.
300	def 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	
307	def 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	
325	def 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	
345	def 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	
363	def 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.
385	def 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.
401	def 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	
420	def 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	
440	def 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	
463	def 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.
487	def 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	
512	def 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.
539	def 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.
561	def 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).
578	def 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.
597	def 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.
616	def 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) ---