1	import Base
2	import ./Keys.bend as Keys
3	import ./MemTable.bend as MemTable
4	import ./Sstable.bend as Sstable
5	import ./Wal.bend as Wal
6	import ./Fs.bend as Fs
7	import ./Manifest.bend as Manifest
8	import ./CrashPoint.bend as CrashPoint
9	
10	# Db API (Task 7): durable write path + reads.
11	#
12	# Write path (durability ordering, the load-bearing property): encode the
13	# batch -> append the WAL -> fsync -> THEN update mem. A crash before fsync
14	# loses at most unacked writes (recovery replays the WAL prefix, Task 10).
15	# This ordering is a code-review property (no law over IO exists); Task 12
16	# fault-injection crashes between each step to verify it empirically.
17	# CrashPoint calls are test-only host effects. Unset or unequal checkpoints are
18	# successful no-ops; signal delivery and filesystem ordering are not Bend proofs.
19	#
20	# Read path: mem (newest-first) ++ L0 newest-first ++ L1 ... — first match
21	# wins INCLUDING tombstones (MemTable.scan_go freezes), so deletes can never
22	# resurrect older versions. Cross-level newest-first order is CORRECT iff
23	# Task 9 maintains the L0-drain invariant: compaction always drains ALL of
24	# L0 (inputs deleted), so every L0 table is strictly newer than every L1+
25	# table; within a level, tables are stored newest-first. Flush (Task 8)
26	# Cons'es new tables at the head; compaction (Task 9) preserves the order.
27	
28	type BEntry is Data:
29	  BEntry{tab: String, key: String, val: Maybe<&2, String>}
30	
31	def default_batch_cap() -> Nat:
32	  1n
33	
34	def bcache_bound() -> Nat:
35	  256n
36	
37	# manifest_token is the exact serialized Manifest observed by this handle. Flush
38	# and compaction compare it before publication to reject stale-handle drift.
39	type Db is Data:
40	  Db{dir: String, mem: MemTable.MemTable, frozen: MemTable.MemTable, batch_cap: Nat, bcache: List<&2, BEntry>, levels: List<&2, List<&2, Sstable.Table>>, flushed: Nat, manifest_token: String, mem_count: Nat, frozen_count: Nat}
41	
42	# Rotation result: both tables plus their exact counts, so no pass ever
43	# walks a memtable to learn its size. Counts stay exact by construction
44	# (open_db starts 0/0; every transition does arithmetic, never List.length).
45	type RotRes is Data:
46	  Rot{mem: MemTable.MemTable, frozen: MemTable.MemTable, mem_count: Nat, frozen_count: Nat}
47	
48	def wal_path(+dir: String) -> String:
49	  dir ++ "/wal.log"
50	
51	def open_db(+dir: String) -> Db:
52	  Db{dir, MemTable.empty(), MemTable.empty(), default_batch_cap(), Nil{}, Nil{}, 0n, Manifest.serialize(Manifest.M{Nil{}}), 0n, 0n}
53	
54	# --- Pure write core (laws below pin batch == sequential) ---
55	
56	def apply_mut(+mem: MemTable.MemTable, +mut: Wal.Mut) -> MemTable.MemTable:
57	  match mut:
58	    case Wal.Put{key, val}:
59	      MemTable.put(mem, key, val)
60	    case Wal.Del{key}:
61	      MemTable.del(mem, key)
62	
63	def apply_batch(+muts: List<&2, Wal.Mut>, +mem: MemTable.MemTable) -> MemTable.MemTable:
64	  match muts:
65	    case Nil{}:
66	      mem
67	    case Con{+h, t}:
68	      apply_batch(t, apply_mut(mem, h))
69	
70	# --- Pure read core (concat newest-first, first match wins) ---
71	
72	def table_entries(+tab: Sstable.Table) -> List<&2, MemTable.Entry>:
73	  match tab:
74	    case Sstable.Tbl{entries, filter, nbits, smallest, largest, count, chunks, blocks}:
75	      entries
76	
77	def level_entries(+tabs: List<&2, Sstable.Table>) -> List<&2, MemTable.Entry>:
78	  match tabs:
79	    case Nil{}:
80	      Nil{}
81	    case Con{+h, t}:
82	      List.append(&2, MemTable.Entry, table_entries(h), level_entries(t))
83	
84	def all_level_entries(+lvls: List<&2, List<&2, Sstable.Table>>) -> List<&2, MemTable.Entry>:
85	  match lvls:
86	    case Nil{}:
87	      Nil{}
88	    case Con{+h, t}:
89	      List.append(&2, MemTable.Entry, level_entries(h), all_level_entries(t))
90	
91	def all_entries(+db: Db) -> List<&2, MemTable.Entry>:
92	  match db:
93	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
94	      match mem:
95	        case MemTable.MT{entries}:
96	          List.append(&2, MemTable.Entry, entries, all_level_entries(levels))
97	
98	def table_hit_if(present: Bool, tab: Sstable.Table, +key: String) -> Maybe<&2, Maybe<&2, String>>:
99	  match present:
100	    case False{}:
101	      None{}
102	    case True{}:
103	      Sstable.block_get_hit(tab, key)
104	
105	def table_hit(+tab: Sstable.Table, +key: String) -> Maybe<&2, Maybe<&2, String>>:
106	  table_hit_if(Sstable.maybe_present(tab, key), tab, key)
107	
108	def tables_hits(+tabs: List<&2, Sstable.Table>, +key: String) -> List<&2, Maybe<&2, Maybe<&2, String>>>:
109	  match tabs:
110	    case Nil{}:
111	      Nil{}
112	    case Con{+h, t}:
113	      Con{table_hit(h, key), tables_hits(t, key)}
114	
115	def levels_hits(+lvls: List<&2, List<&2, Sstable.Table>>, +key: String) -> List<&2, Maybe<&2, Maybe<&2, String>>>:
116	  match lvls:
117	    case Nil{}:
118	      Nil{}
119	    case Con{h, t}:
120	      List.append(&2, Maybe<&2, Maybe<&2, String>>, tables_hits(h, key), levels_hits(t, key))
121	
122	def hits_first(+hits: List<&2, Maybe<&2, Maybe<&2, String>>>) -> Maybe<&2, Maybe<&2, String>>:
123	  match hits:
124	    case Nil{}:
125	      None{}
126	    case Con{None{}, t}:
127	      hits_first(t)
128	    case Con{Some{found}, t}:
129	      Some{found}
130	
131	def flatten_opt(+opt: Maybe<&2, Maybe<&2, String>>) -> Maybe<&2, String>:
132	  match opt:
133	    case None{}:
134	      None{}
135	    case Some{inner}:
136	      inner
137	
138	def db_get(+db: Db, +key: String) -> Maybe<&2, String>:
139	  match db:
140	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
141	      flatten_opt(hits_first(Con{MemTable.get_hit(mem, key), Con{MemTable.get_hit(frozen, key), levels_hits(levels, key)}}))
142	
143	# --- Phase A3: table-identity read cache (assoc-list, bounded FIFO) ---
144	# LRU verdict (spike-by-inspection, 2026-09-25): hub lru has no simple public
145	# put, reads hand the cache back, and every op needs a W.U64 clock — adopting
146	# it threads time through the pure read path and changes db_get's shape for a
147	# marginal gain over this list. Custom assoc-list wins; see the Task 3 commit.
148	
149	def pick_hit(hit: Bool, ans: Maybe<&2, String>, best: Maybe<&2, Maybe<&2, String>>) -> Maybe<&2, Maybe<&2, String>>:
150	  match hit:
151	    case True{}:
152	      Some{ans}
153	    case False{}:
154	      best
155	
156	def bcache_go(+cache: List<&2, BEntry>, +tab: String, +key: String, +best: Maybe<&2, Maybe<&2, String>>) -> Maybe<&2, Maybe<&2, String>>:
157	  match cache:
158	    case Nil{}:
159	      best
160	    case Con{BEntry{t, ky, val}, rest}:
161	      bcache_go(rest, tab, key, pick_hit(Bool.and(String.eq(t, tab), String.eq(ky, key)), val, best))
162	
163	def bcache_lookup(+cache: List<&2, BEntry>, +tab: String, +key: String) -> Maybe<&2, Maybe<&2, String>>:
164	  bcache_go(cache, tab, key, None{})
165	
166	def bcache_push_all(+fresh: List<&2, BEntry>, +old: List<&2, BEntry>) -> List<&2, BEntry>:
167	  List.take(&2, BEntry, List.append(&2, BEntry, fresh, old), bcache_bound())
168	
169	def probe_fill(hit: Maybe<&2, Maybe<&2, String>>, +tab: String, +key: String) -> (Maybe<&2, Maybe<&2, String>> & List<&2, BEntry>):
170	  match hit:
171	    case None{}:
172	      (None{}, Nil{})
173	    case Some{+inner}:
174	      (Some{inner}, Con{BEntry{tab, key, inner}, Nil{}})
175	
176	def probe_cached(cached: Maybe<&2, Maybe<&2, String>>, +tab: Sstable.Table, +key: String) -> (Maybe<&2, Maybe<&2, String>> & List<&2, BEntry>):
177	  match cached:
178	    case None{}:
179	      probe_fill(table_hit(tab, key), Sstable.table_id(tab), key)
180	    case Some{ans}:
181	      (Some{ans}, Nil{})
182	
183	def table_probe(+tab: Sstable.Table, +key: String, +cache: List<&2, BEntry>) -> (Maybe<&2, Maybe<&2, String>> & List<&2, BEntry>):
184	  probe_cached(bcache_lookup(cache, Sstable.table_id(tab), key), tab, key)
185	
186	def tables_cons(cur: (Maybe<&2, Maybe<&2, String>> & List<&2, BEntry>), rest: (List<&2, Maybe<&2, Maybe<&2, String>>> & List<&2, BEntry>)) -> (List<&2, Maybe<&2, Maybe<&2, String>>> & List<&2, BEntry>):
187	  match cur:
188	    case (hit, delta):
189	      match rest:
190	        case (rhits, rdelta):
191	          (Con{hit, rhits}, List.append(&2, BEntry, delta, rdelta))
192	
193	def tables_walk(+tabs: List<&2, Sstable.Table>, +key: String, +cache: List<&2, BEntry>) -> (List<&2, Maybe<&2, Maybe<&2, String>>> & List<&2, BEntry>):
194	  match tabs:
195	    case Nil{}:
196	      (Nil{}, Nil{})
197	    case Con{+h, t}:
198	      tables_cons(table_probe(h, key, cache), tables_walk(t, key, cache))
199	
200	def levels_cons(cur: (List<&2, Maybe<&2, Maybe<&2, String>>> & List<&2, BEntry>), rest: (List<&2, Maybe<&2, Maybe<&2, String>>> & List<&2, BEntry>)) -> (List<&2, Maybe<&2, Maybe<&2, String>>> & List<&2, BEntry>):
201	  match cur:
202	    case (chits, cdelta):
203	      match rest:
204	        case (rhits, rdelta):
205	          (List.append(&2, Maybe<&2, Maybe<&2, String>>, chits, rhits), List.append(&2, BEntry, cdelta, rdelta))
206	
207	def levels_cached(+lvls: List<&2, List<&2, Sstable.Table>>, +key: String, +cache: List<&2, BEntry>) -> (List<&2, Maybe<&2, Maybe<&2, String>>> & List<&2, BEntry>):
208	  match lvls:
209	    case Nil{}:
210	      (Nil{}, Nil{})
211	    case Con{+h, t}:
212	      levels_cons(tables_walk(h, key, cache), levels_cached(t, key, cache))
213	
214	def cached_levels(lr: (List<&2, Maybe<&2, Maybe<&2, String>>> & List<&2, BEntry>), +dir: String, +mem: MemTable.MemTable, +frozen: MemTable.MemTable, +batch_cap: Nat, +bcache: List<&2, BEntry>, +levels: List<&2, List<&2, Sstable.Table>>, +flushed: Nat, +manifest_token: String, +mem_count: Nat, +frozen_count: Nat) -> Db & Maybe<&2, String>:
215	  match lr:
216	    case (hits, delta):
217	      (Db{dir, mem, frozen, batch_cap, bcache_push_all(delta, bcache), levels, flushed, manifest_token, mem_count, frozen_count}, flatten_opt(hits_first(hits)))
218	
219	def cached_frozen(fhit: Maybe<&2, Maybe<&2, String>>, +dir: String, +mem: MemTable.MemTable, +frozen: MemTable.MemTable, +batch_cap: Nat, +bcache: List<&2, BEntry>, +levels: List<&2, List<&2, Sstable.Table>>, +flushed: Nat, +manifest_token: String, +key: String, +mem_count: Nat, +frozen_count: Nat) -> Db & Maybe<&2, String>:
220	  match fhit:
221	    case Some{found}:
222	      (Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}, flatten_opt(Some{found}))
223	    case None{}:
224	      cached_levels(levels_cached(levels, key, bcache), dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count)
225	
226	def cached_mem(mhit: Maybe<&2, Maybe<&2, String>>, +dir: String, +mem: MemTable.MemTable, +frozen: MemTable.MemTable, +batch_cap: Nat, +bcache: List<&2, BEntry>, +levels: List<&2, List<&2, Sstable.Table>>, +flushed: Nat, +manifest_token: String, +key: String, +mem_count: Nat, +frozen_count: Nat) -> Db & Maybe<&2, String>:
227	  match mhit:
228	    case Some{found}:
229	      (Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}, flatten_opt(Some{found}))
230	    case None{}:
231	      cached_frozen(MemTable.get_hit(frozen, key), dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, key, mem_count, frozen_count)
232	
233	def db_get_cached(+db: Db, +key: String) -> Db & Maybe<&2, String>:
234	  match db:
235	    case Db{dir, +mem, +frozen, +batch_cap, +bcache, +levels, +flushed, +manifest_token, +mem_count, +frozen_count}:
236	      cached_mem(MemTable.get_hit(mem, key), dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, key, mem_count, frozen_count)
237	
238	# WAL framing v1 is applied here (see Recover): each batch is stored as
239	# dashes(len) ++ ";" ++ encode(batch) so replay can stream frames and
240	# truncate a torn tail. Files are permissioned 0600 at creation.
241	def wal_frame(+data: String) -> String:
242	  Wal.dashes(String.length(data)) ++ ";" ++ data
243	
244	# Tail of the append: match heads the def body (params always
245	# destructurable), so the write's pair splits with single uses.
246	def wal_tail(fr: (File & Result<&1, &1, U32 & String, Unit>), +dir: String) -> IO(Result<&1, &1, U32 & String, Unit>):
247	  match fr:
248	    case (f2, r):
249	      do IO<Result<&1, &1, U32 & String, Unit>>:
250	        res : Unit <- IO.try(Unit, IO.pure(Result<&1, &1, U32 & String, Unit>, r))
251	        _cp1 : Unit <- IO.try(Unit, CrashPoint.hit("wal.appended"))
252	        _cls : Unit <- File.close(f2)
253	        _syn : Unit <- IO.try(Unit, Fs.fsync(wal_path(dir)))
254	        _cp2 : Unit <- IO.try(Unit, CrashPoint.hit("wal.synced"))
255	        _prm : Unit <- IO.try(Unit, Fs.chmod(wal_path(dir), U32.from_nat(384n)))
256	        return Done{res}
257	
258	def wal_append(+dir: String, +data: String) -> IO(Result<&1, &1, U32 & String, Unit>):
259	  do IO<Result<&1, &1, U32 & String, Unit>>:
260	    f : File <- IO.try(File, File.open(wal_path(dir), "a"))
261	    fr : (File & Result<&1, &1, U32 & String, Unit>) <- File.write(f, wal_frame(data))
262	    wal_tail(fr, dir)
263	
264	def batch_len(+muts: List<&2, Wal.Mut>) -> Nat:
265	  List.length(&2, Wal.Mut, muts)
266	
267	def rotate_cnt(full: Bool, +grown: MemTable.MemTable, +frozen: MemTable.MemTable, +count: Nat, +fcount: Nat) -> RotRes:
268	  match full:
269	    case True{}:
270	      Rot{MemTable.empty(), grown, 0n, count}
271	    case False{}:
272	      Rot{grown, frozen, count, fcount}
273	
274	def frozen_empty(+frozen: MemTable.MemTable) -> Bool:
275	  match frozen:
276	    case MemTable.MT{entries}:
277	      match entries:
278	        case Nil{}:
279	          True{}
280	        case Con{_, _}:
281	          False{}
282	
283	def rotate_after(+grown: MemTable.MemTable, +frozen: MemTable.MemTable, +count: Nat, +fcount: Nat) -> RotRes:
284	  rotate_cnt(Bool.and(Nat.is_le(4096n, count), frozen_empty(frozen)), grown, frozen, count, fcount)
285	
286	def apply_and_rotate(+muts: List<&2, Wal.Mut>, +mem: MemTable.MemTable, +frozen: MemTable.MemTable, +mem_count: Nat, +frozen_count: Nat) -> RotRes:
287	  rotate_after(apply_batch(muts, mem), frozen, Nat.add(mem_count, batch_len(muts)), frozen_count)
288	
289	def apply_done(res: RotRes, +dir: String, +batch_cap: Nat, +bcache: List<&2, BEntry>, +levels: List<&2, List<&2, Sstable.Table>>, +flushed: Nat, +manifest_token: String) -> Db:
290	  match res:
291	    case Rot{mem2, frozen2, mc, fc}:
292	      Db{dir, mem2, frozen2, batch_cap, bcache, levels, flushed, manifest_token, mc, fc}
293	
294	def rot_done(res: RotRes, +dir: String, +batch_cap: Nat, +bcache: List<&2, BEntry>, +levels: List<&2, List<&2, Sstable.Table>>, +flushed: Nat, +manifest_token: String) -> Result<&1, &1, U32 & String, Db>:
295	  Done{apply_done(res, dir, batch_cap, bcache, levels, flushed, manifest_token)}
296	
297	def db_write(+db: Db, +batch: Wal.Batch) -> IO(Result<&1, &1, U32 & String, Db>):
298	  match db:
299	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
300	      match batch:
301	        case Wal.Batch{muts}:
302	          do IO<Result<&1, &1, U32 & String, Db>>:
303	            _res : Unit <- IO.try(Unit, wal_append(dir, Wal.encode(Wal.Batch{muts})))
304	            return rot_done(apply_and_rotate(muts, mem, frozen, mem_count, frozen_count), dir, batch_cap, bcache, levels, flushed, manifest_token)
305	
306	def db_put(+db: Db, +key: String, +val: String) -> IO(Result<&1, &1, U32 & String, Db>):
307	  db_write(db, Wal.Batch{Con{Wal.Put{key, val}, Nil{}}})
308	
309	def batch_cap_of(+db: Db) -> Nat:
310	  match db:
311	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
312	      batch_cap
313	
314	def with_batch_cap(+db: Db, +cap: Nat) -> Db:
315	  match db:
316	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
317	      Db{dir, mem, frozen, cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}
318	
319	def staged_muts(+staged: List<&2, Wal.Batch>) -> List<&2, Wal.Mut>:
320	  match staged:
321	    case Nil{}:
322	      Nil{}
323	    case Con{Wal.Batch{muts}, t}:
324	      List.append(&2, Wal.Mut, muts, staged_muts(t))
325	
326	def db_write_staged(+db: Db, +staged: List<&2, Wal.Batch>) -> IO(Result<&1, &1, U32 & String, Db>):
327	  match db:
328	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
329	      db_write(Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}, Wal.Batch{staged_muts(List.reverse(&2, Wal.Batch, staged))})
330	
331	def db_del(+db: Db, +key: String) -> IO(Result<&1, &1, U32 & String, Db>):
332	  db_write(db, Wal.Batch{Con{Wal.Del{key}, Nil{}}})
333	
334	# --- Closed-vector laws live in laws/Db.bend (spec laws 1, 2, 3, 7) ---