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 ./Decimal.bend as Decimal
9	import ./CrashPoint.bend as CrashPoint
10	
11	# Db API (Task 7): durable write path + reads.
12	#
13	# Write path (durability ordering, the load-bearing property): encode the
14	# batch -> append the WAL -> fsync -> THEN update mem. A crash before fsync
15	# loses at most unacked writes (recovery replays the WAL prefix, Task 10).
16	# This ordering is a code-review property (no law over IO exists); Task 12
17	# fault-injection crashes between each step to verify it empirically.
18	# CrashPoint calls are test-only host effects. Unset or unequal checkpoints are
19	# successful no-ops; signal delivery and filesystem ordering are not Bend proofs.
20	#
21	# Read path: mem (newest-first) ++ L0 newest-first ++ L1 ... — first match
22	# wins INCLUDING tombstones (MemTable.scan_go freezes), so deletes can never
23	# resurrect older versions. Cross-level newest-first order is CORRECT iff
24	# Task 9 maintains the L0-drain invariant: compaction always drains ALL of
25	# L0 (inputs deleted), so every L0 table is strictly newer than every L1+
26	# table; within a level, tables are stored newest-first. Flush (Task 8)
27	# Cons'es new tables at the head; compaction (Task 9) preserves the order.
28	
29	type BEntry is Data:
30	  BEntry{tab: String, key: String, val: Maybe<&2, String>}
31	
32	def default_batch_cap() -> Nat:
33	  1n
34	
35	def bcache_bound() -> Nat:
36	  256n
37	
38	# manifest_token is the exact serialized Manifest observed by this handle. Flush
39	# and compaction compare it before publication to reject stale-handle drift.
40	type Db is Data:
41	  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}
42	
43	# Rotation result: both tables plus their exact counts, so no pass ever
44	# walks a memtable to learn its size. Counts stay exact by construction
45	# (open_db starts 0/0; every transition does arithmetic, never List.length).
46	type RotRes is Data:
47	  Rot{mem: MemTable.MemTable, frozen: MemTable.MemTable, mem_count: Nat, frozen_count: Nat}
48	
49	def wal_path(+dir: String) -> String:
50	  dir ++ "/wal.log"
51	
52	def open_db(+dir: String) -> Db:
53	  Db{dir, MemTable.empty(), MemTable.empty(), default_batch_cap(), Nil{}, Nil{}, 0n, Manifest.serialize(Manifest.M{Nil{}}), 0n, 0n}
54	
55	# --- Pure write core (laws below pin batch == sequential) ---
56	
57	def apply_mut(+mem: MemTable.MemTable, +mut: Wal.Mut) -> MemTable.MemTable:
58	  match mut:
59	    case Wal.Put{key, val}:
60	      MemTable.put(mem, key, val)
61	    case Wal.Del{key}:
62	      MemTable.del(mem, key)
63	
64	def apply_batch(+muts: List<&2, Wal.Mut>, +mem: MemTable.MemTable) -> MemTable.MemTable:
65	  match muts:
66	    case Nil{}:
67	      mem
68	    case Con{+h, t}:
69	      apply_batch(t, apply_mut(mem, h))
70	
71	# --- Pure read core (concat newest-first, first match wins) ---
72	
73	def table_entries(+tab: Sstable.Table) -> List<&2, MemTable.Entry>:
74	  match tab:
75	    case Sstable.Tbl{entries, filter, nbits, smallest, largest, count, chunks, blocks}:
76	      entries
77	
78	def level_entries(+tabs: List<&2, Sstable.Table>) -> List<&2, MemTable.Entry>:
79	  match tabs:
80	    case Nil{}:
81	      Nil{}
82	    case Con{+h, t}:
83	      List.append(&2, MemTable.Entry, table_entries(h), level_entries(t))
84	
85	def all_level_entries(+lvls: List<&2, List<&2, Sstable.Table>>) -> List<&2, MemTable.Entry>:
86	  match lvls:
87	    case Nil{}:
88	      Nil{}
89	    case Con{+h, t}:
90	      List.append(&2, MemTable.Entry, level_entries(h), all_level_entries(t))
91	
92	def all_entries(+db: Db) -> List<&2, MemTable.Entry>:
93	  match db:
94	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
95	      match mem:
96	        case MemTable.MT{entries}:
97	          List.append(&2, MemTable.Entry, entries, all_level_entries(levels))
98	
99	def table_hit_if(present: Bool, tab: Sstable.Table, +key: String) -> Maybe<&2, Maybe<&2, String>>:
100	  match present:
101	    case False{}:
102	      None{}
103	    case True{}:
104	      Sstable.block_get_hit(tab, key)
105	
106	def table_hit(+tab: Sstable.Table, +key: String) -> Maybe<&2, Maybe<&2, String>>:
107	  table_hit_if(Sstable.maybe_present(tab, key), tab, key)
108	
109	def tables_hits(+tabs: List<&2, Sstable.Table>, +key: String) -> List<&2, Maybe<&2, Maybe<&2, String>>>:
110	  match tabs:
111	    case Nil{}:
112	      Nil{}
113	    case Con{+h, t}:
114	      Con{table_hit(h, key), tables_hits(t, key)}
115	
116	def levels_hits(+lvls: List<&2, List<&2, Sstable.Table>>, +key: String) -> List<&2, Maybe<&2, Maybe<&2, String>>>:
117	  match lvls:
118	    case Nil{}:
119	      Nil{}
120	    case Con{h, t}:
121	      List.append(&2, Maybe<&2, Maybe<&2, String>>, tables_hits(h, key), levels_hits(t, key))
122	
123	def hits_first(+hits: List<&2, Maybe<&2, Maybe<&2, String>>>) -> Maybe<&2, Maybe<&2, String>>:
124	  match hits:
125	    case Nil{}:
126	      None{}
127	    case Con{None{}, t}:
128	      hits_first(t)
129	    case Con{Some{found}, t}:
130	      Some{found}
131	
132	def flatten_opt(+opt: Maybe<&2, Maybe<&2, String>>) -> Maybe<&2, String>:
133	  match opt:
134	    case None{}:
135	      None{}
136	    case Some{inner}:
137	      inner
138	
139	def db_get(+db: Db, +key: String) -> Maybe<&2, String>:
140	  match db:
141	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
142	      flatten_opt(hits_first(Con{MemTable.get_hit(mem, key), Con{MemTable.get_hit(frozen, key), levels_hits(levels, key)}}))
143	
144	# --- Phase A3: table-identity read cache (assoc-list, bounded FIFO) ---
145	# LRU verdict (spike-by-inspection, 2026-09-25): hub lru has no simple public
146	# put, reads hand the cache back, and every op needs a W.U64 clock — adopting
147	# it threads time through the pure read path and changes db_get's shape for a
148	# marginal gain over this list. Custom assoc-list wins; see the Task 3 commit.
149	
150	def pick_hit(hit: Bool, ans: Maybe<&2, String>, best: Maybe<&2, Maybe<&2, String>>) -> Maybe<&2, Maybe<&2, String>>:
151	  match hit:
152	    case True{}:
153	      Some{ans}
154	    case False{}:
155	      best
156	
157	def bcache_go(+cache: List<&2, BEntry>, +tab: String, +key: String, +best: Maybe<&2, Maybe<&2, String>>) -> Maybe<&2, Maybe<&2, String>>:
158	  match cache:
159	    case Nil{}:
160	      best
161	    case Con{BEntry{t, ky, val}, rest}:
162	      bcache_go(rest, tab, key, pick_hit(Bool.and(String.eq(t, tab), String.eq(ky, key)), val, best))
163	
164	def bcache_lookup(+cache: List<&2, BEntry>, +tab: String, +key: String) -> Maybe<&2, Maybe<&2, String>>:
165	  bcache_go(cache, tab, key, None{})
166	
167	def bcache_push_all(+fresh: List<&2, BEntry>, +old: List<&2, BEntry>) -> List<&2, BEntry>:
168	  List.take(&2, BEntry, List.append(&2, BEntry, fresh, old), bcache_bound())
169	
170	def probe_fill(hit: Maybe<&2, Maybe<&2, String>>, +tab: String, +key: String) -> (Maybe<&2, Maybe<&2, String>> & List<&2, BEntry>):
171	  match hit:
172	    case None{}:
173	      (None{}, Nil{})
174	    case Some{+inner}:
175	      (Some{inner}, Con{BEntry{tab, key, inner}, Nil{}})
176	
177	def probe_cached(cached: Maybe<&2, Maybe<&2, String>>, +tab: Sstable.Table, +key: String) -> (Maybe<&2, Maybe<&2, String>> & List<&2, BEntry>):
178	  match cached:
179	    case None{}:
180	      probe_fill(table_hit(tab, key), Sstable.table_id(tab), key)
181	    case Some{ans}:
182	      (Some{ans}, Nil{})
183	
184	def table_probe(+tab: Sstable.Table, +key: String, +cache: List<&2, BEntry>) -> (Maybe<&2, Maybe<&2, String>> & List<&2, BEntry>):
185	  probe_cached(bcache_lookup(cache, Sstable.table_id(tab), key), tab, key)
186	
187	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>):
188	  match cur:
189	    case (hit, delta):
190	      match rest:
191	        case (rhits, rdelta):
192	          (Con{hit, rhits}, List.append(&2, BEntry, delta, rdelta))
193	
194	def tables_walk(+tabs: List<&2, Sstable.Table>, +key: String, +cache: List<&2, BEntry>) -> (List<&2, Maybe<&2, Maybe<&2, String>>> & List<&2, BEntry>):
195	  match tabs:
196	    case Nil{}:
197	      (Nil{}, Nil{})
198	    case Con{+h, t}:
199	      tables_cons(table_probe(h, key, cache), tables_walk(t, key, cache))
200	
201	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>):
202	  match cur:
203	    case (chits, cdelta):
204	      match rest:
205	        case (rhits, rdelta):
206	          (List.append(&2, Maybe<&2, Maybe<&2, String>>, chits, rhits), List.append(&2, BEntry, cdelta, rdelta))
207	
208	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>):
209	  match lvls:
210	    case Nil{}:
211	      (Nil{}, Nil{})
212	    case Con{+h, t}:
213	      levels_cons(tables_walk(h, key, cache), levels_cached(t, key, cache))
214	
215	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>:
216	  match lr:
217	    case (hits, delta):
218	      (Db{dir, mem, frozen, batch_cap, bcache_push_all(delta, bcache), levels, flushed, manifest_token, mem_count, frozen_count}, flatten_opt(hits_first(hits)))
219	
220	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>:
221	  match fhit:
222	    case Some{found}:
223	      (Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}, flatten_opt(Some{found}))
224	    case None{}:
225	      cached_levels(levels_cached(levels, key, bcache), dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count)
226	
227	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>:
228	  match mhit:
229	    case Some{found}:
230	      (Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}, flatten_opt(Some{found}))
231	    case None{}:
232	      cached_frozen(MemTable.get_hit(frozen, key), dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, key, mem_count, frozen_count)
233	
234	def db_get_cached(+db: Db, +key: String) -> Db & Maybe<&2, String>:
235	  match db:
236	    case Db{dir, +mem, +frozen, +batch_cap, +bcache, +levels, +flushed, +manifest_token, +mem_count, +frozen_count}:
237	      cached_mem(MemTable.get_hit(mem, key), dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, key, mem_count, frozen_count)
238	
239	# WAL framing v1 is applied here (see Recover): each batch is stored as
240	# dashes(len) ++ ";" ++ encode(batch) so replay can stream frames and
241	# truncate a torn tail. Files are permissioned 0600 at creation.
242	def wal_frame(+data: String) -> String:
243	  Wal.dashes(String.length(data)) ++ ";" ++ data
244	
245	# Tail of the append: match heads the def body (params always
246	# destructurable), so the write's pair splits with single uses.
247	def wal_tail(fr: (File & Result<&1, &1, U32 & String, Unit>), +dir: String) -> IO(Result<&1, &1, U32 & String, Unit>):
248	  match fr:
249	    case (f2, r):
250	      do IO<Result<&1, &1, U32 & String, Unit>>:
251	        res : Unit <- IO.try(Unit, IO.pure(Result<&1, &1, U32 & String, Unit>, r))
252	        _cp1 : Unit <- IO.try(Unit, CrashPoint.hit("wal.appended"))
253	        _cls : Unit <- File.close(f2)
254	        _syn : Unit <- IO.try(Unit, Fs.fsync(wal_path(dir)))
255	        _cp2 : Unit <- IO.try(Unit, CrashPoint.hit("wal.synced"))
256	        _prm : Unit <- IO.try(Unit, Fs.chmod(wal_path(dir), U32.from_nat(384n)))
257	        return Done{res}
258	
259	def wal_append(+dir: String, +data: String) -> IO(Result<&1, &1, U32 & String, Unit>):
260	  do IO<Result<&1, &1, U32 & String, Unit>>:
261	    f : File <- IO.try(File, File.open(wal_path(dir), "a"))
262	    fr : (File & Result<&1, &1, U32 & String, Unit>) <- File.write(f, wal_frame(data))
263	    wal_tail(fr, dir)
264	
265	def batch_len(+muts: List<&2, Wal.Mut>) -> Nat:
266	  List.length(&2, Wal.Mut, muts)
267	
268	def rotate_cnt(full: Bool, +grown: MemTable.MemTable, +frozen: MemTable.MemTable, +count: Nat, +fcount: Nat) -> RotRes:
269	  match full:
270	    case True{}:
271	      Rot{MemTable.empty(), grown, 0n, count}
272	    case False{}:
273	      Rot{grown, frozen, count, fcount}
274	
275	def frozen_empty(+frozen: MemTable.MemTable) -> Bool:
276	  match frozen:
277	    case MemTable.MT{entries}:
278	      match entries:
279	        case Nil{}:
280	          True{}
281	        case Con{_, _}:
282	          False{}
283	
284	def rotate_after(+grown: MemTable.MemTable, +frozen: MemTable.MemTable, +count: Nat, +fcount: Nat) -> RotRes:
285	  rotate_cnt(Bool.and(Nat.is_le(4096n, count), frozen_empty(frozen)), grown, frozen, count, fcount)
286	
287	def apply_and_rotate(+muts: List<&2, Wal.Mut>, +mem: MemTable.MemTable, +frozen: MemTable.MemTable, +mem_count: Nat, +frozen_count: Nat) -> RotRes:
288	  rotate_after(apply_batch(muts, mem), frozen, Nat.add(mem_count, batch_len(muts)), frozen_count)
289	
290	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:
291	  match res:
292	    case Rot{mem2, frozen2, mc, fc}:
293	      Db{dir, mem2, frozen2, batch_cap, bcache, levels, flushed, manifest_token, mc, fc}
294	
295	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>:
296	  Done{apply_done(res, dir, batch_cap, bcache, levels, flushed, manifest_token)}
297	
298	def db_write(+db: Db, +batch: Wal.Batch) -> IO(Result<&1, &1, U32 & String, Db>):
299	  match db:
300	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
301	      match batch:
302	        case Wal.Batch{muts}:
303	          do IO<Result<&1, &1, U32 & String, Db>>:
304	            _res : Unit <- IO.try(Unit, wal_append(dir, Wal.encode(Wal.Batch{muts})))
305	            return rot_done(apply_and_rotate(muts, mem, frozen, mem_count, frozen_count), dir, batch_cap, bcache, levels, flushed, manifest_token)
306	
307	def db_put(+db: Db, +key: String, +val: String) -> IO(Result<&1, &1, U32 & String, Db>):
308	  db_write(db, Wal.Batch{Con{Wal.Put{key, val}, Nil{}}})
309	
310	def batch_cap_of(+db: Db) -> Nat:
311	  match db:
312	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
313	      batch_cap
314	
315	def with_batch_cap(+db: Db, +cap: Nat) -> Db:
316	  match db:
317	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
318	      Db{dir, mem, frozen, cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}
319	
320	def staged_muts(+staged: List<&2, Wal.Batch>) -> List<&2, Wal.Mut>:
321	  match staged:
322	    case Nil{}:
323	      Nil{}
324	    case Con{Wal.Batch{muts}, t}:
325	      List.append(&2, Wal.Mut, muts, staged_muts(t))
326	
327	def db_write_staged(+db: Db, +staged: List<&2, Wal.Batch>) -> IO(Result<&1, &1, U32 & String, Db>):
328	  match db:
329	    case Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
330	      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))})
331	
332	def db_del(+db: Db, +key: String) -> IO(Result<&1, &1, U32 & String, Db>):
333	  db_write(db, Wal.Batch{Con{Wal.Del{key}, Nil{}}})
334	
335	# --- reopen (public API v0.3.2.0): WAL-only recovery over wal.log ---
336	# Checks wal.log existence; missing dir/file = open_db (empty tables).
337	# Present file is read positionally by chunks, concatenated, split into
338	# wal_frame frames (dashes(len) ++ ";" ++ data), each payload decoded with
339	# Wal.decode (checksum-validated). A torn/corrupt final suffix from a
340	# mid-append crash is discarded; valid batches apply in order over an empty
341	# memtable via apply_batch. Read errors propagate as Fail.
342	
343	def reopen_chunk_chars() -> Nat:
344	  1048576n
345	
346	type ReChunk is Data:
347	  ReChunk{bytes: Nat, text: String}
348	
349	def reopen_chunk_parse(pair: Decimal.Parse) -> Result<&1, &1, U32 & String, ReChunk>:
350	  match pair:
351	    case Decimal.Rejected{error}:
352	      Fail{(U32.from_nat(8n), "bad UTF-8 chunk envelope")}
353	    case Decimal.Accepted{Decimal.Dec{value, rest}}:
354	      Done{ReChunk{value, rest}}
355	
356	def reopen_dash_step(+done: Bool, isdash: Bool, +acc: Nat) -> (Bool & Nat):
357	  match done:
358	    case True{}:
359	      (True{}, acc)
360	    case False{}:
361	      match isdash:
362	        case True{}:
363	          (False{}, Nat.add(1n, acc))
364	        case False{}:
365	          (True{}, acc)
366	
367	def reopen_dash_snd(st: (Bool & Nat)) -> Nat:
368	  match st:
369	    case (d, a):
370	      a
371	
372	def reopen_dash_fst(st: (Bool & Nat), +ch: Char) -> (Bool & Nat):
373	  match st:
374	    case (done, acc):
375	      reopen_dash_step(done, Char.is_eq(ch, Char.from_u32(45)), acc)
376	
377	def reopen_dash_go(fuel: Nat, st: (Bool & Nat), str: String) -> Nat:
378	  match fuel:
379	    case 0n:
380	      reopen_dash_snd(st)
381	    case 1n+f:
382	      match str:
383	        case SNil{}:
384	          reopen_dash_snd(st)
385	        case SCon{h, t}:
386	          reopen_dash_go(f, reopen_dash_fst(st, h), t)
387	
388	def reopen_dashes_len(+str: String) -> Nat:
389	  reopen_dash_go(Nat.add(String.length(str), 1n), (False{}, 0n), str)
390	
391	def reopen_semi_is(eq: Bool, +num: Nat) -> Maybe<&2, Nat>:
392	  match eq:
393	    case True{}:
394	      Some{num}
395	    case False{}:
396	      None{}
397	
398	def reopen_semi(+ch: Char, +num: Nat) -> Maybe<&2, Nat>:
399	  reopen_semi_is(Char.is_eq(ch, Char.from_u32(59)), num)
400	
401	def reopen_semi_go(dropped: String, +num: Nat) -> Maybe<&2, Nat>:
402	  match dropped:
403	    case SNil{}:
404	      None{}
405	    case SCon{h, t}:
406	      reopen_semi(h, num)
407	
408	def reopen_frame_len(+str: String) -> Maybe<&2, Nat>:
409	  +num = reopen_dashes_len(str)
410	  reopen_semi_go(String.drop(str, num), num)
411	
412	def reopen_consumed(+num: Nat) -> Nat:
413	  Nat.add(String.length(Wal.dashes(num)), Nat.add(1n, num))
414	
415	def reopen_clip(+mlen: Maybe<&2, Nat>, +str: String) -> String:
416	  match mlen:
417	    case None{}:
418	      str
419	    case Some{n}:
420	      String.take(String.drop(str, Nat.add(String.length(Wal.dashes(n)), 1n)), n)
421	
422	def reopen_drop(+mlen: Maybe<&2, Nat>, +str: String) -> String:
423	  match mlen:
424	    case None{}:
425	      str
426	    case Some{n}:
427	      String.drop(str, reopen_consumed(n))
428	
429	def reopen_dec_apply(+m2: Maybe<&2, Wal.Batch>, +res: MemTable.MemTable) -> MemTable.MemTable:
430	  match m2:
431	    case None{}:
432	      res
433	    case Some{Wal.Batch{muts}}:
434	      apply_batch(muts, res)
435	
436	def reopen_replay_go(fuel: Nat, +mem: MemTable.MemTable, +str: String) -> MemTable.MemTable:
437	  match fuel:
438	    case 0n:
439	      mem
440	    case 1n+g:
441	      +fl = reopen_frame_len(str)
442	      reopen_replay_go(g, reopen_dec_apply(Wal.decode(reopen_clip(fl, str)), mem), reopen_drop(fl, str))
443	
444	def reopen_replay(+str: String) -> MemTable.MemTable:
445	  reopen_replay_go(Nat.add(String.length(str), 1n), MemTable.empty(), str)
446	
447	def reopen_assemble(+mem: MemTable.MemTable, +dir: String) -> Db:
448	  Db{dir, mem, MemTable.empty(), default_batch_cap(), Nil{}, Nil{}, 0n, Manifest.serialize(Manifest.M{Nil{}}), MemTable.count(mem), 0n}
449	
450	# Chunked WAL reader (SstStream.stream_loop shape): the freshly read chunk
451	# re-enters as the `pending` parameter, so every match scrutinee is a
452	# parameter or field and the loop is a single self-recursive def.
453	def reopen_read_loop(chunks_left: Nat, +path: String, +offset: Nat, +acc: String, pending: Maybe<&2, ReChunk>) -> IO(Result<&1, &1, U32 & String, String>):
454	  match chunks_left pending:
455	    case 0n _:
456	      IO.pure(Result<&1, &1, U32 & String, String>, Fail{(U32.from_nat(7n), "wal exceeds streaming read limit")})
457	    case 1n+rest None{}:
458	      do IO<Result<&1, &1, U32 & String, String>>:
459	        raw : String <- IO.try(String, Fs.read_utf8_chunk(path, offset, reopen_chunk_chars()))
460	        chunk : ReChunk <- IO.try(ReChunk, IO.pure(Result<&1, &1, U32 & String, ReChunk>, reopen_chunk_parse(Decimal.parse_semicolon(raw, reopen_chunk_chars()))))
461	        reopen_read_loop(rest, path, offset, acc, Some{chunk})
462	    case 1n+rest Some{ReChunk{bytes, text}}:
463	      match bytes:
464	        case 0n:
465	          IO.pure(Result<&1, &1, U32 & String, String>, Done{acc})
466	        case 1n+more:
467	          reopen_read_loop(rest, path, Nat.add(offset, 1n+more), acc ++ text, None{})
468	
469	def reopen_read(+dir: String) -> IO(Result<&1, &1, U32 & String, Db>):
470	  do IO<Result<&1, &1, U32 & String, Db>>:
471	    content : String <- IO.try(String, reopen_read_loop(8192n, wal_path(dir), 0n, "", None{}))
472	    return Done{reopen_assemble(reopen_replay(content), dir)}
473	
474	def reopen_exists(ok: Result<&1, &1, U32 & String, Bool>, +dir: String) -> IO(Result<&1, &1, U32 & String, Db>):
475	  match ok:
476	    case Fail{e}:
477	      IO.pure(Result<&1, &1, U32 & String, Db>, Fail{e})
478	    case Done{present}:
479	      match present:
480	        case False{}:
481	          IO.pure(Result<&1, &1, U32 & String, Db>, Done{open_db(dir)})
482	        case True{}:
483	          reopen_read(dir)
484	
485	def reopen(+dir: String) -> IO(Result<&1, &1, U32 & String, Db>):
486	  do IO<Result<&1, &1, U32 & String, Db>>:
487	    r : Result<&1, &1, U32 & String, Bool> <- Fs.exists(wal_path(dir))
488	    reopen_exists(r, dir)
489	
490	# --- Closed-vector laws live in laws/Db.bend (spec laws 1, 2, 3, 7) ---