1	import Base
2	import ./Compact.bend as Compact
3	import ./MemTable.bend as MemTable
4	import ./Sstable.bend as Sstable
5	import ./SstStreamIo.bend as SstStreamIo
6	import ./Wal.bend as Wal
7	import ./Manifest.bend as Manifest
8	import ./Flush.bend as Flush
9	import ./Fs.bend as Fs
10	import ./Db.bend as Db
11	import ./CrashPoint.bend as CrashPoint
12	
13	# Handle remove list in the compaction filesystem effects.
14	def remove_list(names: List<&2, String>) -> IO(Result<&1, &1, U32 & String, Unit>):
15	  match names:
16	    case Nil{}:
17	      IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
18	    case Con{n, t}:
19	      do IO<Result<&1, &1, U32 & String, Unit>>:
20	        _rm : Result<&1, &1, U32 & String, Unit> <- Fs.remove(n)
21	        remove_list(t)
22	
23	# Tracks progress while removing published files.
24	type RemoveState is Type:
25	  RemoveStart{names: List<&2, String>}
26	  RemoveNames{names: List<&2, String>}
27	  RemoveResult{result: Result<&1, &1, U32 & String, Unit>, names: List<&2, String>}
28	
29	# Removes every path and propagates the first failure.
30	def remove_list_strict(fuel: Nat, state: RemoveState) -> IO(Result<&1, &1, U32 & String, Unit>):
31	  match fuel:
32	    case 0n:
33	      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{(U32.from_nat(7n), "remove list exhausted")})
34	    case 1n+rest:
35	      match state:
36	        case RemoveStart{names}:
37	          remove_list_strict(rest, RemoveNames{names})
38	        case RemoveNames{names}:
39	          match names:
40	            case Nil{}:
41	              IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
42	            case Con{name, tail}:
43	              do IO<Result<&1, &1, U32 & String, Unit>>:
44	                removed : Result<&1, &1, U32 & String, Unit> <- Fs.remove(name)
45	                remove_list_strict(rest, RemoveResult{removed, tail})
46	        case RemoveResult{result, names}:
47	          match result:
48	            case Fail{error}:
49	              IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
50	            case Done{Unit{}}:
51	              remove_list_strict(rest, RemoveNames{names})
52	
53	# Starts checked removal of the published file list.
54	def remove_list_strict_start(pair: Nat & List<&2, String>) -> IO(Result<&1, &1, U32 & String, Unit>):
55	  match pair:
56	    case (fuel, names):
57	      remove_list_strict(fuel, RemoveStart{names})
58	
59	def compact_goB.removed(
60	  result: Result<&1, &1, U32 & String, Unit>,
61	  +dir: String,
62	  +mem: MemTable.MemTable,
63	  +levels2: List<&2, List<&2, Sstable.Table>>,
64	  +flushed: Nat,
65	  +manifest: Manifest.Manifest,
66	  +frozen: MemTable.MemTable,
67	  +batch_cap: Nat,
68	  +mem_count: Nat,
69	  +frozen_count: Nat
70	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
71	  match result:
72	    case Fail{error}:
73	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
74	    case Done{Unit{}}:
75	      IO.pure(Result<&1, &1, U32 & String, Db.Db>,
76	        Done{Db.Db{dir, mem, frozen, batch_cap, Nil{}, levels2,
77	          Nat.add(flushed, 1n), Manifest.token(manifest), mem_count, frozen_count}})
78	
79	def compact_goB.directory_synced(
80	  result: Result<&1, &1, U32 & String, Unit>,
81	  +dir: String,
82	  +dels: List<&2, String>,
83	  +m2: Manifest.Manifest,
84	  +mem: MemTable.MemTable,
85	  +levels2: List<&2, List<&2, Sstable.Table>>,
86	  +flushed: Nat,
87	  +frozen: MemTable.MemTable,
88	  +batch_cap: Nat,
89	  +mem_count: Nat,
90	  +frozen_count: Nat
91	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
92	  match result:
93	    case Fail{error}:
94	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
95	    case Done{Unit{}}:
96	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
97	        _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.manifest_published"))
98	        removed : Result<&1, &1, U32 & String, Unit> <- remove_list_strict_start((Nat.add(Nat.mul(2n, List.length(&2, String, dels)), 2n), dels))
99	        compact_goB.removed(removed, dir, mem, levels2, flushed, m2,
100	          frozen, batch_cap, mem_count, frozen_count)
101	
102	def compact_goB.renamed(
103	  result: Result<&1, &1, U32 & String, Unit>,
104	  +dir: String,
105	  +dels: List<&2, String>,
106	  +m2: Manifest.Manifest,
107	  +mem: MemTable.MemTable,
108	  +levels2: List<&2, List<&2, Sstable.Table>>,
109	  +flushed: Nat,
110	  +frozen: MemTable.MemTable,
111	  +batch_cap: Nat,
112	  +mem_count: Nat,
113	  +frozen_count: Nat
114	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
115	  match result:
116	    case Fail{error}:
117	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
118	    case Done{Unit{}}:
119	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
120	        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(dir)
121	        compact_goB.directory_synced(synced, dir, dels, m2, mem, levels2,
122	          flushed, frozen, batch_cap, mem_count, frozen_count)
123	
124	def compact_goB.written(
125	  result: Result<&1, &1, U32 & String, Unit>,
126	  +dir: String,
127	  +mpath: String,
128	  +mtmp: String,
129	  +dels: List<&2, String>,
130	  +m2: Manifest.Manifest,
131	  +mem: MemTable.MemTable,
132	  +levels2: List<&2, List<&2, Sstable.Table>>,
133	  +flushed: Nat,
134	  +frozen: MemTable.MemTable,
135	  +batch_cap: Nat,
136	  +mem_count: Nat,
137	  +frozen_count: Nat
138	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
139	  match result:
140	    case Fail{error}:
141	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
142	    case Done{Unit{}}:
143	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
144	        _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.manifest_synced"))
145	        renamed : Result<&1, &1, U32 & String, Unit> <- Fs.rename(mtmp, mpath)
146	        compact_goB.renamed(renamed, dir, dels, m2, mem,
147	          levels2, flushed, frozen, batch_cap, mem_count, frozen_count)
148	
149	# Compact goB for the compaction filesystem effects.
150	def compact_goB(
151	  +dir: String,
152	  +mpath: String,
153	  +mtmp: String,
154	  +m2: Manifest.Manifest,
155	  +dels: List<&2, String>,
156	  +mem: MemTable.MemTable,
157	  +levels2: List<&2, List<&2, Sstable.Table>>,
158	  +flushed: Nat,
159	  +frozen: MemTable.MemTable,
160	  +batch_cap: Nat,
161	  +mem_count: Nat,
162	  +frozen_count: Nat,
163	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
164	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
165	    written : Result<&1, &1, U32 & String, Unit> <- Flush.write_manifest(mtmp, m2)
166	    compact_goB.written(written, dir, mpath, mtmp, dels, m2, mem, levels2,
167	      flushed, frozen, batch_cap, mem_count, frozen_count)
168	
169	# Handle drift go in the compaction filesystem effects.
170	def drift_go(
171	  ok: Bool,
172	  +dir: String,
173	  +mpath: String,
174	  +mtmp: String,
175	  +mfst: Manifest.Manifest,
176	  +oldl1: List<&2, Sstable.Table>,
177	  +remt: List<&2, Sstable.Table>,
178	  +outname: String,
179	  +mem: MemTable.MemTable,
180	  +levels2: List<&2, List<&2, Sstable.Table>>,
181	  +flushed: Nat,
182	  +frozen: MemTable.MemTable,
183	  +batch_cap: Nat,
184	  +mem_count: Nat,
185	  +frozen_count: Nat,
186	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
187	  match ok:
188	    case True{}:
189	      +new_mfst = Compact.mfst_new(mfst, Compact.rem_names(oldl1, Compact.ml1(mfst), remt), outname)
190	      +removed = Compact.abs_names(oldl1, Compact.ml1(mfst), remt)
191	      +dels = List.append(&2, String, Compact.prefix(Compact.ml0(mfst), dir ++ "/l0"), Compact.prefix(removed, dir ++ "/l1"))
192	      compact_goB(dir, mpath, mtmp, new_mfst, dels, mem, levels2, flushed, frozen, batch_cap, mem_count, frozen_count)
193	    case False{}:
194	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(2n), "manifest drift")})
195	
196	def compact_goA.loaded(
197	  result: Result<&1, &1, U32 & String, Manifest.Manifest>,
198	  +dir: String,
199	  +mpath: String,
200	  +mtmp: String,
201	  +oldl1: List<&2, Sstable.Table>,
202	  +remt: List<&2, Sstable.Table>,
203	  +outname: String,
204	  +mem: MemTable.MemTable,
205	  +levels2: List<&2, List<&2, Sstable.Table>>,
206	  +flushed: Nat,
207	  +manifest_token: String,
208	  +frozen: MemTable.MemTable,
209	  +batch_cap: Nat,
210	  +mem_count: Nat,
211	  +frozen_count: Nat
212	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
213	  match result:
214	    case Fail{error}:
215	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
216	    case Done{+manifest}:
217	      drift_go(String.eq(Manifest.token(manifest), manifest_token), dir, mpath,
218	        mtmp, manifest, oldl1, remt, outname, mem, levels2, flushed,
219	        frozen, batch_cap, mem_count, frozen_count)
220	
221	def compact_goA.synced(
222	  result: Result<&1, &1, U32 & String, Unit>,
223	  +dir: String,
224	  +mpath: String,
225	  +mtmp: String,
226	  +oldl1: List<&2, Sstable.Table>,
227	  +remt: List<&2, Sstable.Table>,
228	  +outname: String,
229	  +mem: MemTable.MemTable,
230	  +levels2: List<&2, List<&2, Sstable.Table>>,
231	  +flushed: Nat,
232	  +manifest_token: String,
233	  +frozen: MemTable.MemTable,
234	  +batch_cap: Nat,
235	  +mem_count: Nat,
236	  +frozen_count: Nat
237	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
238	  match result:
239	    case Fail{error}:
240	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
241	    case Done{Unit{}}:
242	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
243	        _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.output_published"))
244	        manifest : Result<&1, &1, U32 & String, Manifest.Manifest> <- Flush.load_mfst(mpath)
245	        compact_goA.loaded(manifest, dir, mpath, mtmp, oldl1, remt, outname,
246	          mem, levels2, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)
247	
248	def compact_goA.renamed(
249	  result: Result<&1, &1, U32 & String, Unit>,
250	  +dir: String,
251	  +l1dir: String,
252	  +mpath: String,
253	  +mtmp: String,
254	  +oldl1: List<&2, Sstable.Table>,
255	  +remt: List<&2, Sstable.Table>,
256	  +outname: String,
257	  +mem: MemTable.MemTable,
258	  +levels2: List<&2, List<&2, Sstable.Table>>,
259	  +flushed: Nat,
260	  +manifest_token: String,
261	  +frozen: MemTable.MemTable,
262	  +batch_cap: Nat,
263	  +mem_count: Nat,
264	  +frozen_count: Nat
265	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
266	  match result:
267	    case Fail{error}:
268	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
269	    case Done{Unit{}}:
270	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
271	        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(l1dir)
272	        compact_goA.synced(synced, dir, mpath, mtmp, oldl1, remt, outname,
273	          mem, levels2, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)
274	
275	def compact_goA.written(
276	  result: Result<&1, &1, U32 & String, Unit>,
277	  +dir: String,
278	  +l1dir: String,
279	  +tmp: String,
280	  +final: String,
281	  +mpath: String,
282	  +mtmp: String,
283	  +oldl1: List<&2, Sstable.Table>,
284	  +remt: List<&2, Sstable.Table>,
285	  +outname: String,
286	  +mem: MemTable.MemTable,
287	  +levels2: List<&2, List<&2, Sstable.Table>>,
288	  +flushed: Nat,
289	  +manifest_token: String,
290	  +frozen: MemTable.MemTable,
291	  +batch_cap: Nat,
292	  +mem_count: Nat,
293	  +frozen_count: Nat
294	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
295	  match result:
296	    case Fail{error}:
297	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
298	    case Done{Unit{}}:
299	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
300	        _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.output_synced"))
301	        renamed : Result<&1, &1, U32 & String, Unit> <- Fs.rename(tmp, final)
302	        compact_goA.renamed(renamed, dir, l1dir, mpath, mtmp, oldl1, remt,
303	          outname, mem, levels2, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)
304	
305	def compact_goA.ensured(
306	  result: Result<&1, &1, U32 & String, Unit>,
307	  +dir: String,
308	  +l1dir: String,
309	  +tmp: String,
310	  +final: String,
311	  +mpath: String,
312	  +mtmp: String,
313	  +entries: List<&2, MemTable.Entry>,
314	  +level: U32,
315	  +outname: String,
316	  +mem: MemTable.MemTable,
317	  +levels2: List<&2, List<&2, Sstable.Table>>,
318	  +oldl1: List<&2, Sstable.Table>,
319	  +remt: List<&2, Sstable.Table>,
320	  +flushed: Nat,
321	  +manifest_token: String,
322	  +frozen: MemTable.MemTable,
323	  +batch_cap: Nat,
324	  +mem_count: Nat,
325	  +frozen_count: Nat
326	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
327	  match result:
328	    case Fail{error}:
329	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
330	    case Done{Unit{}}:
331	      do IO<Result<&1, &1, U32 & String, Db.Db>>:
332	        written : Result<&1, &1, U32 & String, Unit> <- SstStreamIo.write_table(tmp, entries, level)
333	        compact_goA.written(written, dir, l1dir, tmp, final, mpath, mtmp,
334	          oldl1, remt, outname, mem, levels2, flushed, manifest_token, frozen,
335	          batch_cap, mem_count, frozen_count)
336	
337	# Publishes the compacted run and removes obsolete files.
338	def compact_goA(
339	  +dir: String,
340	  +l1dir: String,
341	  +tmp: String,
342	  +final: String,
343	  +mpath: String,
344	  +mtmp: String,
345	  +entries: List<&2, MemTable.Entry>,
346	  +level: U32,
347	  +outname: String,
348	  +mem: MemTable.MemTable,
349	  +levels2: List<&2, List<&2, Sstable.Table>>,
350	  +oldl1: List<&2, Sstable.Table>,
351	  +remt: List<&2, Sstable.Table>,
352	  +flushed: Nat,
353	  +manifest_token: String,
354	  +frozen: MemTable.MemTable,
355	  +batch_cap: Nat,
356	  +mem_count: Nat,
357	  +frozen_count: Nat,
358	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
359	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
360	    ensured : Result<&1, &1, U32 & String, Unit> <- Flush.ensure_dir(l1dir)
361	    compact_goA.ensured(ensured, dir, l1dir, tmp, final, mpath, mtmp,
362	      entries, level, outname, mem, levels2, oldl1, remt, flushed,
363	      manifest_token, frozen, batch_cap, mem_count, frozen_count)
364	
365	# Compact pre for the compaction filesystem effects.
366	def compact_pre(
367	  +dir: String,
368	  +mem: MemTable.MemTable,
369	  +levels: List<&2, List<&2, Sstable.Table>>,
370	  +flushed: Nat,
371	  +manifest_token: String,
372	  +frozen: MemTable.MemTable,
373	  +batch_cap: Nat,
374	  +mem_count: Nat,
375	  +frozen_count: Nat,
376	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
377	  +levels2 = Compact.compact_levels(levels)
378	  +newl1 = Compact.l1list(levels2)
379	  +out = Compact.hd_tbl(newl1)
380	  +remt = Compact.tail_of(newl1)
381	  +l1dir = dir ++ "/l1"
382	  +outname = "l1-" ++ Nat.show(flushed) ++ ".tbl"
383	  +tmp = l1dir ++ "/" ++ outname ++ ".tmp"
384	  +final = l1dir ++ "/" ++ outname
385	  compact_goA(
386	    dir, l1dir, tmp, final, dir ++ "/MANIFEST", dir ++ "/MANIFEST.tmp",
387	    Db.table_entries(out), 1, outname, mem, levels2, Compact.l1list(levels), remt, flushed,
388	    manifest_token, frozen, batch_cap, mem_count, frozen_count,
389	  )
390	
391	# Compact trig for the compaction filesystem effects.
392	def compact_trig(
393	  gt: Bool,
394	  +dir: String,
395	  +mem: MemTable.MemTable,
396	  +levels: List<&2, List<&2, Sstable.Table>>,
397	  +flushed: Nat,
398	  +manifest_token: String,
399	  +frozen: MemTable.MemTable,
400	  +batch_cap: Nat,
401	  +bcache: List<&2, Db.BEntry>,
402	  +mem_count: Nat,
403	  +frozen_count: Nat,
404	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
405	  match gt:
406	    case True{}:
407	      compact_pre(dir, mem, levels, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)
408	    case False{}:
409	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Done{Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}})
410	
411	# Handle compact in the compaction filesystem effects.
412	def compact(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
413	  match db:
414	    case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
415	      match levels:
416	        case Nil{}:
417	          IO.pure(Result<&1, &1, U32 & String, Db.Db>, Done{Db.Db{dir, mem, frozen, batch_cap, bcache, Nil{}, flushed, manifest_token, mem_count, frozen_count}})
418	        case Con{+l0, r1}:
419	          compact_trig(Nat.is_lt(4n, List.length(&2, Sstable.Table, l0)), dir, mem, Con{l0, r1}, flushed, manifest_token, frozen, batch_cap, bcache, mem_count, frozen_count)