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	# Write tail for the memtable flush.
59	def 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.
73	def 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	
79	def 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	
85	def 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.
96	def 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.
100	def 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.
111	def 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.
118	def 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	
125	def 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	
138	def 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	
173	def 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	
178	def 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	
191	def 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.
204	def 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.
215	def 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.
228	def 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.
234	def 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.
242	def 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.
250	def 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.
272	def 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.
292	def 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.
320	def 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).
337	def 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.
356	def 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.
375	def 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) ---