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	# Compact goB for the compaction filesystem effects.
24	def compact_goB(
25	  +dir: String,
26	  +mpath: String,
27	  +mtmp: String,
28	  +m2: Manifest.Manifest,
29	  +dels: List<&2, String>,
30	  +mem: MemTable.MemTable,
31	  +levels2: List<&2, List<&2, Sstable.Table>>,
32	  +flushed: Nat,
33	  +frozen: MemTable.MemTable,
34	  +batch_cap: Nat,
35	  +mem_count: Nat,
36	  +frozen_count: Nat,
37	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
38	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
39	    _v1 : Unit <- IO.try(Unit, Flush.write_manifest(mtmp, m2))
40	    _cp1 : Unit <- IO.try(Unit, CrashPoint.hit("compact.manifest_synced"))
41	    _v2 : Unit <- IO.try(Unit, Fs.rename(mtmp, mpath))
42	    _v3 : Unit <- IO.try(Unit, Fs.fsync(dir))
43	    _cp2 : Unit <- IO.try(Unit, CrashPoint.hit("compact.manifest_published"))
44	    _ign : Result<&1, &1, U32 & String, Unit> <- remove_list(dels)
45	    return Done{Db.Db{dir, mem, frozen, batch_cap, Nil{}, levels2, Nat.add(flushed, 1n), Manifest.token(m2), mem_count, frozen_count}}
46	
47	# Handle drift go in the compaction filesystem effects.
48	def drift_go(
49	  ok: Bool,
50	  +dir: String,
51	  +mpath: String,
52	  +mtmp: String,
53	  +mfst: Manifest.Manifest,
54	  +oldl1: List<&2, Sstable.Table>,
55	  +remt: List<&2, Sstable.Table>,
56	  +outname: String,
57	  +mem: MemTable.MemTable,
58	  +levels2: List<&2, List<&2, Sstable.Table>>,
59	  +flushed: Nat,
60	  +frozen: MemTable.MemTable,
61	  +batch_cap: Nat,
62	  +mem_count: Nat,
63	  +frozen_count: Nat,
64	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
65	  match ok:
66	    case True{}:
67	      +new_mfst = Compact.mfst_new(mfst, Compact.rem_names(oldl1, Compact.ml1(mfst), remt), outname)
68	      +removed = Compact.abs_names(oldl1, Compact.ml1(mfst), remt)
69	      +dels = List.append(&2, String, Compact.prefix(Compact.ml0(mfst), dir ++ "/l0"), Compact.prefix(removed, dir ++ "/l1"))
70	      compact_goB(dir, mpath, mtmp, new_mfst, dels, mem, levels2, flushed, frozen, batch_cap, mem_count, frozen_count)
71	    case False{}:
72	      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(2n), "manifest drift")})
73	
74	# Compact goA for the compaction filesystem effects.
75	def compact_goA(
76	  +dir: String,
77	  +l1dir: String,
78	  +tmp: String,
79	  +final: String,
80	  +mpath: String,
81	  +mtmp: String,
82	  +entries: List<&2, MemTable.Entry>,
83	  +level: U32,
84	  +outname: String,
85	  +mem: MemTable.MemTable,
86	  +levels2: List<&2, List<&2, Sstable.Table>>,
87	  +oldl1: List<&2, Sstable.Table>,
88	  +remt: List<&2, Sstable.Table>,
89	  +flushed: Nat,
90	  +manifest_token: String,
91	  +frozen: MemTable.MemTable,
92	  +batch_cap: Nat,
93	  +mem_count: Nat,
94	  +frozen_count: Nat,
95	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
96	  do IO<Result<&1, &1, U32 & String, Db.Db>>:
97	    _u1 : Unit <- IO.try(Unit, Flush.ensure_dir(l1dir))
98	    _u2 : Unit <- IO.try(Unit, SstStreamIo.write_table(tmp, entries, level))
99	    _cp1 : Unit <- IO.try(Unit, CrashPoint.hit("compact.output_synced"))
100	    _u3 : Unit <- IO.try(Unit, Fs.rename(tmp, final))
101	    _u4 : Unit <- IO.try(Unit, Fs.fsync(l1dir))
102	    _cp2 : Unit <- IO.try(Unit, CrashPoint.hit("compact.output_published"))
103	    +m : Manifest.Manifest <- IO.try(Manifest.Manifest, Flush.load_mfst(mpath))
104	    drift_go(String.eq(Manifest.token(m), manifest_token), dir, mpath, mtmp, m, oldl1, remt, outname, mem, levels2, flushed, frozen, batch_cap, mem_count, frozen_count)
105	
106	# Compact pre for the compaction filesystem effects.
107	def compact_pre(
108	  +dir: String,
109	  +mem: MemTable.MemTable,
110	  +levels: List<&2, List<&2, Sstable.Table>>,
111	  +flushed: Nat,
112	  +manifest_token: String,
113	  +frozen: MemTable.MemTable,
114	  +batch_cap: Nat,
115	  +mem_count: Nat,
116	  +frozen_count: Nat,
117	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
118	  +levels2 = Compact.compact_levels(levels)
119	  +newl1 = Compact.l1list(levels2)
120	  +out = Compact.hd_tbl(newl1)
121	  +remt = Compact.tail_of(newl1)
122	  +l1dir = dir ++ "/l1"
123	  +outname = "l1-" ++ Nat.show(flushed) ++ ".tbl"
124	  +tmp = l1dir ++ "/" ++ outname ++ ".tmp"
125	  +final = l1dir ++ "/" ++ outname
126	  compact_goA(
127	    dir, l1dir, tmp, final, dir ++ "/MANIFEST", dir ++ "/MANIFEST.tmp",
128	    Db.table_entries(out), 1, outname, mem, levels2, Compact.l1list(levels), remt, flushed,
129	    manifest_token, frozen, batch_cap, mem_count, frozen_count,
130	  )
131	
132	# Compact trig for the compaction filesystem effects.
133	def compact_trig(
134	  gt: Bool,
135	  +dir: String,
136	  +mem: MemTable.MemTable,
137	  +levels: List<&2, List<&2, Sstable.Table>>,
138	  +flushed: Nat,
139	  +manifest_token: String,
140	  +frozen: MemTable.MemTable,
141	  +batch_cap: Nat,
142	  +bcache: List<&2, Db.BEntry>,
143	  +mem_count: Nat,
144	  +frozen_count: Nat,
145	) -> IO(Result<&1, &1, U32 & String, Db.Db>):
146	  match gt:
147	    case True{}:
148	      compact_pre(dir, mem, levels, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)
149	    case False{}:
150	      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}})
151	
152	# Handle compact in the compaction filesystem effects.
153	def compact(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
154	  match db:
155	    case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
156	      match levels:
157	        case Nil{}:
158	          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}})
159	        case Con{+l0, r1}:
160	          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)