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 ./SstFile.bend as SstFile
7	import ./Wal.bend as Wal
8	import ./Manifest.bend as Manifest
9	import ./Db.bend as Db
10	
11	# Tiered compaction (Task 9): L0-drain + L1 overlap-closure.
12	#
13	# Invariants (Task 7/8 read path depends on them):
14	# - L0-DRAIN: every compaction consumes ALL of L0 (post L0 = []), so every
15	#   L0 table is strictly newer than every L1+ table.
16	# - L1-DISJOINT: output is disjoint from every remaining L1 table, via
17	#   overlap-closure over the final hull (rounds = len(l1)+1 unconditionally;
18	#   rounds past fixpoint are idempotent, so no done-flag is needed).
19	# - READ-ORDER MERGE: inputs merge in exact read order (L0 stored order ++
20	#   absorbed-L1 stored order), so first-wins == read first-match.
21	# - TOMBSTONE-SAFE: all winning tombstones are retained at this checkpoint.
22	#   This conservative policy prevents resurrection without requiring zero-output
23	#   publication support.
24	# Trigger: L0 count > 4 (spec: backpressure stall at 8 = 2T, Task 10).
25	# CrashPoint calls are test-only host effects. Unset or unequal checkpoints are
26	# successful no-ops; signal delivery and filesystem ordering are not Bend proofs.
27	# Soundness sketch (spec law 11): keys resolving outside the closure are
28	# untouched (a key in both closure-inputs and a remainder table would put
29	# it in the hull and in a hull-disjoint range — contradiction); keys inside
30	# resolve to the same newest version (subset order == read order).
31	
32	# Handle ov and in the level compaction.
33	def ov_and(lhs: Bool, rhs: Bool) -> Bool:
34	  match lhs:
35	    case True{}:
36	      rhs
37	    case False{}:
38	      False{}
39	
40	# Handle le cmp in the level compaction.
41	def le_cmp(ord: Cmp) -> Bool:
42	  match ord:
43	    case LT{}:
44	      True{}
45	    case EQ{}:
46	      True{}
47	    case GT{}:
48	      False{}
49	
50	# Handle ov12 in the level compaction.
51	def ov12(fa: Maybe<&2, String>, lb: Maybe<&2, String>) -> Bool:
52	  match fa lb:
53	    case None{} _:
54	      False{}
55	    case _ None{}:
56	      False{}
57	    case Some{x} Some{y}:
58	      le_cmp(Keys.cmp(x, y))
59	
60	
61	# Return the table overlap for the level compaction.
62	def table_overlap(+ta: Sstable.Table, +tb: Sstable.Table) -> Bool:
63	  ov_and(ov12(Sstable.table_smallest(ta), Sstable.table_largest(tb)), ov12(Sstable.table_smallest(tb), Sstable.table_largest(ta)))
64	
65	# Hull bound folds (pump+fuel+leaf; single-Maybe state, no pairs).
66	def pick_lo(ord: Cmp, k1: String, k2: String) -> Maybe<&2, String>:
67	  match ord:
68	    case LT{}:
69	      Some{k1}
70	    case EQ{}:
71	      Some{k1}
72	    case GT{}:
73	      Some{k2}
74	
75	# Handle min step in the level compaction.
76	def min_step(fm: Maybe<&2, String>, cur: Maybe<&2, String>) -> Maybe<&2, String>:
77	  match fm cur:
78	    case None{} _:
79	      cur
80	    case _ None{}:
81	      fm
82	    case Some{+a} Some{+b}:
83	      pick_lo(Keys.cmp(a, b), a, b)
84	
85	# Handle min go in the level compaction.
86	def min_go(fuel: Nat, +cur: Maybe<&2, String>, tabs: List<&2, Sstable.Table>) -> Maybe<&2, String>:
87	  match fuel:
88	    case 0n:
89	      cur
90	    case 1n+f:
91	      match tabs:
92	        case Nil{}:
93	          cur
94	        case Con{h, t}:
95	          min_go(f, min_step(Sstable.table_smallest(h), cur), t)
96	
97	# Handle min first in the level compaction.
98	def min_first(+tabs: List<&2, Sstable.Table>) -> Maybe<&2, String>:
99	  min_go(List.length(&2, Sstable.Table, tabs), None{}, tabs)
100	
101	# Select hi for the level compaction.
102	def pick_hi(ord: Cmp, k1: String, k2: String) -> Maybe<&2, String>:
103	  match ord:
104	    case LT{}:
105	      Some{k2}
106	    case EQ{}:
107	      Some{k2}
108	    case GT{}:
109	      Some{k1}
110	
111	# Handle max step in the level compaction.
112	def max_step(lm: Maybe<&2, String>, cur: Maybe<&2, String>) -> Maybe<&2, String>:
113	  match lm cur:
114	    case None{} _:
115	      cur
116	    case _ None{}:
117	      lm
118	    case Some{+a} Some{+b}:
119	      pick_hi(Keys.cmp(a, b), a, b)
120	
121	# Handle max go in the level compaction.
122	def max_go(fuel: Nat, +cur: Maybe<&2, String>, tabs: List<&2, Sstable.Table>) -> Maybe<&2, String>:
123	  match fuel:
124	    case 0n:
125	      cur
126	    case 1n+f:
127	      match tabs:
128	        case Nil{}:
129	          cur
130	        case Con{h, t}:
131	          max_go(f, max_step(Sstable.table_largest(h), cur), t)
132	
133	# Handle max last in the level compaction.
134	def max_last(+tabs: List<&2, Sstable.Table>) -> Maybe<&2, String>:
135	  max_go(List.length(&2, Sstable.Table, tabs), None{}, tabs)
136	
137	# Overlap of one table against a hull (bounds as params).
138	def hull_step(keep: Bool, +tab: Sstable.Table, +acc: List<&2, Sstable.Table>) -> List<&2, Sstable.Table>:
139	  match keep:
140	    case True{}:
141	      Con{tab, acc}
142	    case False{}:
143	      acc
144	
145	# Handle hull neg in the level compaction.
146	def hull_neg(keep: Bool, +tab: Sstable.Table, +acc: List<&2, Sstable.Table>) -> List<&2, Sstable.Table>:
147	  match keep:
148	    case True{}:
149	      acc
150	    case False{}:
151	      Con{tab, acc}
152	
153	# Handle ov hull in the level compaction.
154	def ov_hull(+tab: Sstable.Table, +lo: Maybe<&2, String>, +hi: Maybe<&2, String>) -> Bool:
155	  ov_and(ov12(Sstable.table_smallest(tab), hi), ov12(lo, Sstable.table_largest(tab)))
156	
157	# Handle filt go in the level compaction.
158	def filt_go(
159	  fuel: Nat,
160	  +lo: Maybe<&2, String>,
161	  +hi: Maybe<&2, String>,
162	  +acc: List<&2, Sstable.Table>,
163	  rest: List<&2, Sstable.Table>
164	) -> List<&2, Sstable.Table>:
165	  match fuel:
166	    case 0n:
167	      List.reverse(&2, Sstable.Table, acc)
168	    case 1n+f:
169	      match rest:
170	        case Nil{}:
171	          List.reverse(&2, Sstable.Table, acc)
172	        case Con{+h, t}:
173	          filt_go(f, lo, hi, hull_step(ov_hull(h, lo, hi), h, acc), t)
174	
175	# Handle filter pos in the level compaction.
176	def filter_pos(
177	  +rest: List<&2, Sstable.Table>,
178	  +lo: Maybe<&2, String>,
179	  +hi: Maybe<&2, String>
180	) -> List<&2, Sstable.Table>:
181	  filt_go(List.length(&2, Sstable.Table, rest), lo, hi, Nil{}, rest)
182	
183	# Handle filt neg go in the level compaction.
184	def filt_neg_go(
185	  fuel: Nat,
186	  +lo: Maybe<&2, String>,
187	  +hi: Maybe<&2, String>,
188	  +acc: List<&2, Sstable.Table>,
189	  rest: List<&2, Sstable.Table>
190	) -> List<&2, Sstable.Table>:
191	  match fuel:
192	    case 0n:
193	      List.reverse(&2, Sstable.Table, acc)
194	    case 1n+f:
195	      match rest:
196	        case Nil{}:
197	          List.reverse(&2, Sstable.Table, acc)
198	        case Con{+h, t}:
199	          filt_neg_go(f, lo, hi, hull_neg(ov_hull(h, lo, hi), h, acc), t)
200	
201	# Handle filter neg in the level compaction.
202	def filter_neg(
203	  +rest: List<&2, Sstable.Table>,
204	  +lo: Maybe<&2, String>,
205	  +hi: Maybe<&2, String>
206	) -> List<&2, Sstable.Table>:
207	  filt_neg_go(List.length(&2, Sstable.Table, rest), lo, hi, Nil{}, rest)
208	
209	# The production policy is conservative: every winning tombstone is retained
210	# regardless of the lower-level shadow list.
211	def drop_none(
212	  +key: String,
213	  +_shadow: List<&2, MemTable.Entry>,
214	  +acc: List<&2, MemTable.Entry>
215	) -> List<&2, MemTable.Entry>:
216	  Con{MemTable.Entry{key, None{}}, acc}
217	
218	# Handle drop go in the level compaction.
219	def drop_go(
220	  fuel: Nat,
221	  +shadow: List<&2, MemTable.Entry>,
222	  +acc: List<&2, MemTable.Entry>,
223	  xs: List<&2, MemTable.Entry>
224	) -> List<&2, MemTable.Entry>:
225	  match fuel:
226	    case 0n:
227	      List.reverse(&2, MemTable.Entry, acc)
228	    case 1n+f:
229	      match xs:
230	        case Nil{}:
231	          List.reverse(&2, MemTable.Entry, acc)
232	        case Con{MemTable.Entry{k, v}, t}:
233	          match v:
234	            case None{}:
235	              drop_go(f, shadow, drop_none(k, shadow, acc), t)
236	            case Some{s}:
237	              drop_go(f, shadow, Con{MemTable.Entry{k, Some{s}}, acc}, t)
238	
239	# Handle shadow drop in the level compaction.
240	def shadow_drop(+xs: List<&2, MemTable.Entry>, +shadow: List<&2, MemTable.Entry>) -> List<&2, MemTable.Entry>:
241	  drop_go(List.length(&2, MemTable.Entry, xs), shadow, Nil{}, xs)
242	
243	
244	# Preserve each table's strict sorted-run boundary and stored read order.
245	def table_runs(tabs: List<&2, Sstable.Table>) -> List<&2, List<&2, MemTable.Entry>>:
246	  match tabs:
247	    case Nil{}:
248	      Nil{}
249	    case Con{h, rest}:
250	      Con{Db.table_entries(h), table_runs(rest)}
251	
252	
253	# Production output: merge L0 in chronological stored order, merge absorbed L1
254	# in its original stored order, then merge L0 as strictly newer than L1.
255	# Tombstones are retained, so a triggered compaction publishes exactly one table.
256	def finish_runs(
257	  +l0_runs: List<&2, List<&2, MemTable.Entry>>,
258	  +l1_runs: List<&2, List<&2, MemTable.Entry>>,
259	  +rest: List<&2, Sstable.Table>
260	) -> List<&2, Sstable.Table>:
261	  merged_l0 merged_l1 = SortedRun.merge_many_newest(l0_runs) SortedRun.merge_many_newest(l1_runs)
262	  +merged = SortedRun.merge_many_newest(Con{merged_l0, Con{merged_l1, Nil{}}})
263	  Con{Sstable.from_sorted_unique(shadow_drop(merged, Nil{}), 1n), rest}
264	
265	# Closure rounds retain the original L1 list so the final absorbed runs are
266	# selected in exact stored read order, independent of the round that found them.
267	def closed_runs_go(
268	  major: Nat,
269	  +lo: Maybe<&2, String>,
270	  +hi: Maybe<&2, String>,
271	  +l0_runs: List<&2, List<&2, MemTable.Entry>>,
272	  +all_l1: List<&2, Sstable.Table>,
273	  +rest: List<&2, Sstable.Table>,
274	) -> List<&2, Sstable.Table>:
275	  match major:
276	    case 0n:
277	      finish_runs(l0_runs, table_runs(filter_pos(all_l1, lo, hi)), rest)
278	    case 1n+m:
279	      +abs_new +rem_new = filter_pos(rest, lo, hi) filter_neg(rest, lo, hi)
280	      closed_runs_go(m, min_step(min_first(abs_new), lo), max_step(max_last(abs_new), hi), l0_runs, all_l1, rem_new)
281	
282	
283	# Top dispatch (all matches on params; count gate inside).
284	def compact_snd(
285	  +l0: List<&2, Sstable.Table>,
286	  rest1: List<&2, List<&2, Sstable.Table>>
287	) -> List<&2, List<&2, Sstable.Table>>:
288	  match rest1:
289	    case Nil{}:
290	      Con{Nil{}, Con{finish_runs(table_runs(l0), Nil{}, Nil{}), Nil{}}}
291	    case Con{+l1, +below}:
292	      Con{Nil{}, Con{closed_runs_go(Nat.add(List.length(&2, Sstable.Table, l1), 1n), min_first(l0), max_last(l0), table_runs(l0), l1, l1), below}}
293	
294	# Compact dec for the level compaction.
295	def compact_dec(
296	  gt: Bool,
297	  +l0: List<&2, Sstable.Table>,
298	  +rest1: List<&2, List<&2, Sstable.Table>>
299	) -> List<&2, List<&2, Sstable.Table>>:
300	  match gt:
301	    case True{}:
302	      compact_snd(l0, rest1)
303	    case False{}:
304	      Con{l0, rest1}
305	
306	# Compact levels for the level compaction.
307	def compact_levels(levels: List<&2, List<&2, Sstable.Table>>) -> List<&2, List<&2, Sstable.Table>>:
308	  match levels:
309	    case Nil{}:
310	      Nil{}
311	    case Con{+l0, rest1}:
312	      compact_dec(Nat.is_lt(4n, List.length(&2, Sstable.Table, l0)), l0, rest1)
313	
314	# --- IO helpers (Task 9b): remainder-membership, name splits, level surgery.
315	# Remainder test via ranges (sound by old-L1 disjointness, spec law 12):
316	# in a disjoint family, overlap with a remainder table means identity.
317	
318	# Combine leaf for the level compaction.
319	def or_leaf(+hit: Bool, +acc: Bool) -> Bool:
320	  match hit:
321	    case True{}:
322	      True{}
323	    case False{}:
324	      acc
325	
326	# Handle in go in the level compaction.
327	def in_go(fuel: Nat, +tab: Sstable.Table, +acc: Bool, rem: List<&2, Sstable.Table>) -> Bool:
328	  match fuel:
329	    case 0n:
330	      acc
331	    case 1n+f:
332	      match rem:
333	        case Nil{}:
334	          acc
335	        case Con{h, tt}:
336	          in_go(f, tab, or_leaf(table_overlap(tab, h), acc), tt)
337	
338	# Handle in rem in the level compaction.
339	def in_rem(+tab: Sstable.Table, +rem: List<&2, Sstable.Table>) -> Bool:
340	  in_go(List.length(&2, Sstable.Table, rem), tab, False{}, rem)
341	
342	# Handle abs pick in the level compaction.
343	def abs_pick(+inr: Bool, +name: String, +acc: List<&2, String>) -> List<&2, String>:
344	  match inr:
345	    case True{}:
346	      acc
347	    case False{}:
348	      Con{name, acc}
349	
350	# Handle abs go in the level compaction.
351	def abs_go(
352	  fuel: Nat,
353	  +rem: List<&2, Sstable.Table>,
354	  +acc: List<&2, String>,
355	  tabs: List<&2, Sstable.Table>,
356	  names: List<&2, String>
357	) -> List<&2, String>:
358	  match fuel:
359	    case 0n:
360	      List.reverse(&2, String, acc)
361	    case 1n+f:
362	      match tabs names:
363	        case Nil{} Nil{}:
364	          List.reverse(&2, String, acc)
365	        case Nil{} Con{nh, nt}:
366	          List.reverse(&2, String, acc)
367	        case Con{th, tt} Nil{}:
368	          List.reverse(&2, String, acc)
369	        case Con{th, tt} Con{nh, nt}:
370	          abs_go(f, rem, abs_pick(in_rem(th, rem), nh, acc), tt, nt)
371	
372	# Handle abs names in the level compaction.
373	def abs_names(
374	  +tabs: List<&2, Sstable.Table>,
375	  +names: List<&2, String>,
376	  +rem: List<&2, Sstable.Table>
377	) -> List<&2, String>:
378	  abs_go(List.length(&2, Sstable.Table, tabs), rem, Nil{}, tabs, names)
379	
380	# Handle rem pick in the level compaction.
381	def rem_pick(+inr: Bool, +name: String, +acc: List<&2, String>) -> List<&2, String>:
382	  match inr:
383	    case True{}:
384	      Con{name, acc}
385	    case False{}:
386	      acc
387	
388	# Handle rem go in the level compaction.
389	def rem_go(
390	  fuel: Nat,
391	  +rem: List<&2, Sstable.Table>,
392	  +acc: List<&2, String>,
393	  tabs: List<&2, Sstable.Table>,
394	  names: List<&2, String>
395	) -> List<&2, String>:
396	  match fuel:
397	    case 0n:
398	      List.reverse(&2, String, acc)
399	    case 1n+f:
400	      match tabs names:
401	        case Nil{} Nil{}:
402	          List.reverse(&2, String, acc)
403	        case Nil{} Con{nh, nt}:
404	          List.reverse(&2, String, acc)
405	        case Con{th, tt} Nil{}:
406	          List.reverse(&2, String, acc)
407	        case Con{th, tt} Con{nh, nt}:
408	          rem_go(f, rem, rem_pick(in_rem(th, rem), nh, acc), tt, nt)
409	
410	# Handle rem names in the level compaction.
411	def rem_names(
412	  +tabs: List<&2, Sstable.Table>,
413	  +names: List<&2, String>,
414	  +rem: List<&2, Sstable.Table>
415	) -> List<&2, String>:
416	  rem_go(List.length(&2, Sstable.Table, tabs), rem, Nil{}, tabs, names)
417	
418	# --- Closed-vector laws live in laws/Compact.bend (spec laws 6, 11, 12) ---
419	
420	# --- IO chain (Task 9b): output file + manifest rewrite + input removal.
421	# Crash discipline mirrors flush: output durable, then manifest (tmp +
422	# rename + dir fsync), then input files become orphans (recovery ignores
423	# files absent from the manifest; Task 12 sweeps them). Manifest/table
424	# drift (exact serialized Manifest identity mismatch) aborts fail-closed;
425	# names stay positionally aligned with tables (output head, remainder tail).
426	
427	# Handle l1hd in the level compaction.
428	def l1hd(r1: List<&2, List<&2, Sstable.Table>>) -> List<&2, Sstable.Table>:
429	  match r1:
430	    case Nil{}:
431	      Nil{}
432	    case Con{l1, below}:
433	      l1
434	
435	# Handle l1list in the level compaction.
436	def l1list(lvls: List<&2, List<&2, Sstable.Table>>) -> List<&2, Sstable.Table>:
437	  match lvls:
438	    case Nil{}:
439	      Nil{}
440	    case Con{l0, r1}:
441	      l1hd(r1)
442	
443	# Handle l0list in the level compaction.
444	def l0list(lvls: List<&2, List<&2, Sstable.Table>>) -> List<&2, Sstable.Table>:
445	  match lvls:
446	    case Nil{}:
447	      Nil{}
448	    case Con{l0, r1}:
449	      l0
450	
451	# Handle hd tbl in the level compaction.
452	def hd_tbl(tabs: List<&2, Sstable.Table>) -> Sstable.Table:
453	  match tabs:
454	    case Nil{}:
455	      Sstable.from_sorted_unique(Nil{}, 1n)
456	    case Con{h, t}:
457	      h
458	
459	# Process the remaining of for the level compaction.
460	def tail_of(tabs: List<&2, Sstable.Table>) -> List<&2, Sstable.Table>:
461	  match tabs:
462	    case Nil{}:
463	      Nil{}
464	    case Con{h, t}:
465	      t
466	
467	# Handle ml0 in the level compaction.
468	def ml0(mfst: Manifest.Manifest) -> List<&2, String>:
469	  match mfst:
470	    case Manifest.M{lvs}:
471	      match lvs:
472	        case Nil{}:
473	          Nil{}
474	        case Con{l0n, r1}:
475	          l0n
476	
477	# Handle ml1b in the level compaction.
478	def ml1b(r1: List<&2, List<&2, String>>) -> List<&2, String>:
479	  match r1:
480	    case Nil{}:
481	      Nil{}
482	    case Con{l1n, below}:
483	      l1n
484	
485	# Handle ml1 in the level compaction.
486	def ml1(mfst: Manifest.Manifest) -> List<&2, String>:
487	  match mfst:
488	    case Manifest.M{lvs}:
489	      match lvs:
490	        case Nil{}:
491	          Nil{}
492	        case Con{l0n, r1}:
493	          ml1b(r1)
494	
495	# Handle mbelb in the level compaction.
496	def mbelb(r1: List<&2, List<&2, String>>) -> List<&2, List<&2, String>>:
497	  match r1:
498	    case Nil{}:
499	      Nil{}
500	    case Con{l1n, below}:
501	      below
502	
503	# Handle mbelow in the level compaction.
504	def mbelow(mfst: Manifest.Manifest) -> List<&2, List<&2, String>>:
505	  match mfst:
506	    case Manifest.M{lvs}:
507	      match lvs:
508	        case Nil{}:
509	          Nil{}
510	        case Con{l0n, r1}:
511	          mbelb(r1)
512	
513	# Handle mfst new in the level compaction.
514	def mfst_new(+mfst: Manifest.Manifest, +remnames: List<&2, String>, +outname: String) -> Manifest.Manifest:
515	  Manifest.M{Con{Nil{}, Con{Con{outname, remnames}, mbelow(mfst)}}}
516	
517	# Handle prefix go in the level compaction.
518	def prefix_go(names: List<&2, String>, +dir: String, +acc: List<&2, String>) -> List<&2, String>:
519	  match names:
520	    case Nil{}:
521	      List.reverse(&2, String, acc)
522	    case Con{h, t}:
523	      prefix_go(t, dir, Con{dir ++ "/" ++ h, acc})
524	
525	# Handle prefix in the level compaction.
526	def prefix(+names: List<&2, String>, +dir: String) -> List<&2, String>:
527	  prefix_go(names, dir, Nil{})