1	import Base
2	import ./ByteMemTable.bend as ByteMemTable
3	import ./ByteManifest.bend as ByteManifest
4	import ./ByteManifestIo.bend as ByteManifestIo
5	import ./ByteSstFile.bend as ByteSstFile
6	import ./ByteSstIo.bend as ByteSstIo
7	import ./ByteWal.bend as ByteWal
8	import ./ByteWalIo.bend as ByteWalIo
9	import ./DbLock.bend as DbLock
10	import ./DurableError.bend as DurableError
11	import ./DurableBatchPolicy.bend as BatchPolicy
12	import ./coordinator/QueuePolicy.bend as QueuePolicy
13	import ./coordinator/SnapshotValueLengthPolicy.bend as SnapshotValueLengthPolicy
14	import ./Fs.bend as Fs
15	import ./FilesCompat.bend as Files
16	import ./Keys.bend as Keys
17	import ./scan/Cursor.bend as Cursor
18	import ./scan/Snapshot.bend as SnapshotPolicy
19	import ./scan/ByteBlockReader.bend as ByteBlockReader
20	import ./scan/SstCursor.bend as SstCursor
21	import ./scan/RangeCursor.bend as RangeCursor
22	import ./scan/VisibleCursor.bend as VisibleCursor
23	import ./scan/VisibleStats.bend as VisibleStats
24	import ./CrashPoint.bend as CrashPoint
25	import ./scan/ScanWorkPolicy.bend as ScanWorkPolicy
26	import ./ByteRecoveryPolicy.bend as RecoveryPolicy
27	import ./ValueLengthPolicy.bend as ValueLengthPolicy
28	import ./Wal.bend as Wal
29	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
30	
31	# Reports byte-database lifecycle, validation, storage, and uncertain-commit failures.
32	type Error is Data:
33	  Busy{}
34	  Timeout{}
35	  NotFound{}
36	  AlreadyExists{}
37	  InvalidArgument{}
38	  UnsupportedPlatform{}
39	  ResourceLimit{}
40	  Corruption{message: String}
41	  Io{code: U32, message: String}
42	  CommitUnknown{code: U32, message: String}
43	
44	# Prevents writes after a sync failure leaves commit status unknown.
45	type HandleState is Data:
46	  Open{}
47	  Poisoned{}
48	
49	# Owns one exclusive database lock and the recovered byte mutation table.
50	type Handle is Type:
51	  Handle{path: String, lock: DbLock.Lock, table: ByteMemTable.Table,
52	    state: HandleState, base_stats: VisibleStats.Stats}
53	
54	# Retains one immutable SST generation and an owned copy of the active delta.
55	type SnapshotRoot is Type:
56	  SnapshotRoot{sst_path: Maybe<&1, String>, sst_generation: Maybe<&1, U32>,
57	    retained_bytes: Nat, delta: List<&1, ByteMemTable.Entry>}
58	
59	# Carries a block cursor and its retained-byte budget from one immutable root.
60	type SnapshotCursorRoot is Type:
61	  SnapshotCursorRoot{retained_bytes: Nat, cursor: VisibleCursor.Cursor}
62	
63	# Holds an incremental point-read transition over a retained snapshot cursor.
64	type SnapshotPointRead is Type:
65	  SnapshotPointNeed{cursor: VisibleCursor.Cursor, query: Bytes.Bytes}
66	  SnapshotPointStep{step: VisibleCursor.Step, query: Bytes.Bytes}
67	
68	# Reports canonical byte entry count and payload size.
69	type Stats is Data:
70	  Stats{entries: Nat, payload_bytes: Nat}
71	
72	# Carries normalized mutations and the visible base totals before WAL append.
73	type BatchPrepared is Type:
74	  Prepared{mutations: List<&1, ByteWal.Mutation>, base_stats: VisibleStats.Stats}
75	
76	# Tracks unique-key reconciliation against SST-backed visible statistics.
77	type BatchStatsState is Type:
78	  BatchNeed{handle: Handle, mutations: List<&1, ByteWal.Mutation>,
79	    entries: List<&1, ByteMemTable.Entry>, original_stats: VisibleStats.Stats,
80	    base_stats: VisibleStats.Stats}
81	  BatchRead{pair: Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
82	    mutations: List<&1, ByteWal.Mutation>, entries: List<&1, ByteMemTable.Entry>,
83	    original_stats: VisibleStats.Stats, base_stats: VisibleStats.Stats,
84	    key_bytes: Nat}
85	  BatchProbe{handle: Handle, mutations: List<&1, ByteWal.Mutation>,
86	    entries: List<&1, ByteMemTable.Entry>, original_stats: VisibleStats.Stats,
87	    base_stats: VisibleStats.Stats, key_bytes: Nat,
88	    key: Bytes.Bytes,
89	    result: ByteMemTable.Table & ByteMemTable.Probe}
90	  BatchKey{handle: Handle, mutations: List<&1, ByteWal.Mutation>,
91	    entries: List<&1, ByteMemTable.Entry>, original_stats: VisibleStats.Stats,
92	    base_stats: VisibleStats.Stats,
93	    keys: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes}
94	
95	# Replays normalized mutations into a replacement byte table.
96	type BatchBuildState is Type:
97	  BatchBuild{mutations: List<&1, ByteWal.Mutation>,
98	    table: ByteMemTable.Table, reversed: List<&1, ByteWal.Mutation>}
99	  BatchBuildPut{keys: Bytes.Bytes & Bytes.Bytes,
100	    values: Bytes.Bytes & Bytes.Bytes, tail: List<&1, ByteWal.Mutation>,
101	    table: ByteMemTable.Table, reversed: List<&1, ByteWal.Mutation>}
102	  BatchBuildDelete{keys: Bytes.Bytes & Bytes.Bytes,
103	    tail: List<&1, ByteWal.Mutation>, table: ByteMemTable.Table,
104	    reversed: List<&1, ByteWal.Mutation>}
105	
106	# Holds the next block read or key comparison in a bounded SST point lookup.
107	type PointRead is Type:
108	  Need{cursor: SstCursor.Cursor, query: Bytes.Bytes}
109	  Read{step: SstCursor.Step, query: Bytes.Bytes}
110	  Compare{value: Bytes.Bytes, cursor: SstCursor.Cursor,
111	    query: Bytes.Bytes, order: Cmp}
112	
113	# Converts database lock errors into the byte database error domain.
114	def lock.error(error: DurableError.Error) -> Error:
115	  match error:
116	    case DurableError.Error{DurableError.Busy{}, _, _, _, _}:
117	      Busy{}
118	    case DurableError.Error{DurableError.Timeout{}, _, _, code, message}:
119	      Timeout{}
120	    case DurableError.Error{DurableError.NotFound{}, _, _, code, message}:
121	      NotFound{}
122	    case DurableError.Error{DurableError.AlreadyExists{}, _, _, code, message}:
123	      AlreadyExists{}
124	    case DurableError.Error{DurableError.InvalidArgument{}, _, _, code, message}:
125	      InvalidArgument{}
126	    case DurableError.Error{DurableError.UnsupportedPlatform{}, _, _, _, _}:
127	      UnsupportedPlatform{}
128	    case DurableError.Error{DurableError.Corruption{}, _, _, code, message}:
129	      Corruption{message}
130	    case DurableError.Error{_, _, _, code, message}:
131	      Io{code, message}
132	
133	# Maps a raw filesystem failure into the byte database error domain.
134	def io.error(error: U32 & String) -> Error:
135	  match error:
136	    case (code, message):
137	      Io{code, message}
138	
139	# Converts lock acquisition into an owned database handle.
140	def initialization.failed(
141	  +path: String,
142	  lock: DbLock.Lock,
143	  error: U32 & String
144	) -> IO(Result<&1, &1, Error, Handle>):
145	  match error:
146	    case (code, message):
147	      do IO<Result<&1, &1, Error, Handle>>:
148	        _released : Result<&1, &1, DurableError.Error, Unit> <-
149	          DbLock.release(lock, "create", path)
150	        return Fail{Io{code, message}}
151	
152	# Syncs the new WAL directory before returning its initial handle.
153	def initialized.synced(
154	  +path: String,
155	  lock: DbLock.Lock,
156	  result: Result<&1, &1, U32 & String, Unit>
157	) -> IO(Result<&1, &1, Error, Handle>):
158	  match result:
159	    case Fail{error}:
160	      initialization.failed(path, lock, error)
161	    case Done{Unit{}}:
162	      IO.pure(Result<&1, &1, Error, Handle>,
163	        Done{Handle{path, lock, ByteMemTable.empty(), Open{}, VisibleStats.Stats{0n, 0n}}})
164	
165	# Creates the generation-four WAL after acquiring the directory lock.
166	def initialized(
167	  +path: String,
168	  lock: DbLock.Lock,
169	  result: Result<&1, &1, U32 & String, Unit>
170	) -> IO(Result<&1, &1, Error, Handle>):
171	  match result:
172	    case Fail{error}:
173	      initialization.failed(path, lock, error)
174	    case Done{Unit{}}:
175	      do IO<Result<&1, &1, Error, Handle>>:
176	        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path)
177	        initialized.synced(path, lock, synced)
178	
179	def initialized.manifest(
180	  +path: String,
181	  lock: DbLock.Lock,
182	  result: Result<&1, &1, U32 & String, Unit>
183	) -> IO(Result<&1, &1, Error, Handle>):
184	  match result:
185	    case Fail{error}: initialization.failed(path, lock, error)
186	    case Done{Unit{}}:
187	      do IO<Result<&1, &1, Error, Handle>>:
188	        published : Result<&1, &1, U32 & String, Unit> <-
189	          ByteManifestIo.write_atomic(path, ByteManifest.GenerationEmpty{})
190	        initialized(path, lock, published)
191	
192	# Releases ownership when syncing the created database directory fails.
193	def create.locked(
194	  +path: String,
195	  result: Result<&1, &1, DurableError.Error, DbLock.Lock>
196	) -> IO(Result<&1, &1, Error, Handle>):
197	  match result:
198	    case Fail{error}:
199	      IO.pure(Result<&1, &1, Error, Handle>, Fail{lock.error(error)})
200	    case Done{lock}:
201	      do IO<Result<&1, &1, Error, Handle>>:
202	        created : Result<&1, &1, U32 & String, Unit> <- ByteWalIo.create(path ++ "/WAL")
203	        initialized.manifest(path, lock, created)
204	
205	def create.canonical(
206	  result: Result<&1, &1, U32 & String, String>
207	) -> IO(Result<&1, &1, Error, Handle>):
208	  match result:
209	    case Fail{error}:
210	      IO.pure(Result<&1, &1, Error, Handle>, Fail{io.error(error)})
211	    case Done{+path}:
212	      do IO<Result<&1, &1, Error, Handle>>:
213	        acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <-
214	          DbLock.acquire(path, "create")
215	        create.locked(path, acquired)
216	
217	# Converts successful directory reservation into lock acquisition.
218	def create.reserved(
219	  +path: String,
220	  result: Result<&1, &1, U32 & String, Unit>
221	) -> IO(Result<&1, &1, Error, Handle>):
222	  match result:
223	    case Fail{error}:
224	      IO.pure(Result<&1, &1, Error, Handle>, Fail{io.error(error)})
225	    case Done{Unit{}}:
226	      do IO<Result<&1, &1, Error, Handle>>:
227	        canonical : Result<&1, &1, U32 & String, String> <-
228	          Fs.canonical_path(path)
229	        create.canonical(canonical)
230	
231	# Creates a byte database directory, stable lock, and generation-four WAL.
232	def create(+path: String) -> IO(Result<&1, &1, Error, Handle>):
233	  do IO<Result<&1, &1, Error, Handle>>:
234	    reserved : Result<&1, &1, U32 & String, Unit> <- Fs.make_dir(path)
235	    create.reserved(path, reserved)
236	
237	def recovery.error(code: U32, message: String) -> Error:
238	  Bool.pick(Error, RecoveryPolicy.is_resource_limit(code),
239	    ResourceLimit{}, Corruption{message})
240	
241	# Maps WAL recovery errors into corruption or host failures.
242	def recovered(
243	  +path: String,
244	  lock: DbLock.Lock,
245	  base_stats: VisibleStats.Stats,
246	  result: Result<&1, &1, U32 & String, ByteMemTable.Table>
247	) -> IO(Result<&1, &1, Error, Handle>):
248	  match result:
249	    case Fail{(code, message)}:
250	      do IO<Result<&1, &1, Error, Handle>>:
251	        _released : Result<&1, &1, DurableError.Error, Unit> <-
252	          DbLock.release(lock, "open_existing", path)
253	        return Fail{recovery.error(code, message)}
254	    case Done{table}:
255	      IO.pure(Result<&1, &1, Error, Handle>,
256	        Done{Handle{path, lock, table, Open{}, base_stats}})
257	
258	# Replays the byte WAL into an empty in-memory delta after SST validation.
259	def open.wal(
260	  +path: String,
261	  lock: DbLock.Lock,
262	  base_stats: VisibleStats.Stats
263	) -> IO(Result<&1, &1, Error, Handle>):
264	  do IO<Result<&1, &1, Error, Handle>>:
265	    replayed : Result<&1, &1, U32 & String, ByteMemTable.Table> <-
266	      ByteWalIo.recover.from(path ++ "/WAL", ByteMemTable.empty())
267	    recovered(path, lock, base_stats, replayed)
268	
269	def open.sst.validated(
270	  +path: String,
271	  lock: DbLock.Lock,
272	  result: Result<&1, &1, U32 & String, ByteBlockReader.Metadata>
273	) -> IO(Result<&1, &1, Error, Handle>):
274	  match result:
275	    case Fail{error}:
276	      recovered(path, lock, VisibleStats.Stats{0n, 0n}, Fail{error})
277	    case Done{ByteBlockReader.Metadata{_, _, _, _, live_entries, payload_bytes}}:
278	      open.wal(path, lock, VisibleStats.Stats{live_entries, payload_bytes})
279	
280	def open.manifest.failed(
281	  +path: String,
282	  lock: DbLock.Lock,
283	  error: ByteManifestIo.Error
284	) -> IO(Result<&1, &1, Error, Handle>):
285	  match error:
286	    case ByteManifestIo.Invalid{_}:
287	      do IO<Result<&1, &1, Error, Handle>>:
288	        _released : Result<&1, &1, DurableError.Error, Unit> <-
289	          DbLock.release(lock, "open_existing", path)
290	        return Fail{Corruption{"byte manifest validation failed"}}
291	    case ByteManifestIo.Host{code, message}:
292	      do IO<Result<&1, &1, Error, Handle>>:
293	        _released : Result<&1, &1, DurableError.Error, Unit> <-
294	          DbLock.release(lock, "open_existing", path)
295	        return Fail{Io{code, message}}
296	
297	def open.manifest.result(
298	  +path: String,
299	  lock: DbLock.Lock,
300	  result: Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue>
301	) -> IO(Result<&1, &1, Error, Handle>):
302	  match result:
303	    case Fail{error}: open.manifest.failed(path, lock, error)
304	    case Done{ByteManifest.Empty{}}:
305	      open.wal(path, lock, VisibleStats.Stats{0n, 0n})
306	    case Done{ByteManifest.GenerationEmpty{}}:
307	      open.wal(path, lock, VisibleStats.Stats{0n, 0n})
308	    case Done{ByteManifest.HasSst{}}:
309	      do IO<Result<&1, &1, Error, Handle>>:
310	        metadata : Result<&1, &1, U32 & String, ByteBlockReader.Metadata> <-
311	          ByteSstIo.validate(path ++ "/SST")
312	        open.sst.validated(path, lock, metadata)
313	    case Done{ByteManifest.GenerationSst{file_id}}:
314	      do IO<Result<&1, &1, Error, Handle>>:
315	        metadata : Result<&1, &1, U32 & String, ByteBlockReader.Metadata> <-
316	          ByteSstIo.validate(ByteManifest.sst.generation.path(path, file_id))
317	        open.sst.validated(path, lock, metadata)
318	
319	def reclaim.file(+target: String, info: Result<&1, &1, U32 & String, Files.Info>) -> IO(Unit):
320	  match info:
321	    case Done{Files.Info{Files.Regular{}, _, _}}:
322	      do IO<Unit>:
323	        _removed : Result<&1, &1, U32 & String, Unit> <- Fs.remove(target)
324	        return Unit{}
325	    case _: IO.pure(Unit, Unit{})
326	
327	def reclaim.remove(+path: String, names: List<&2, String>) -> IO(Unit):
328	  match names:
329	    case Nil{}: IO.pure(Unit, Unit{})
330	    case Con{name, rest}:
331	      do IO<Unit>:
332	        +target : String <- IO.pure(String, path ++ "/" ++ name)
333	        info : Result<&1, &1, U32 & String, Files.Info> <- Files.stat(target)
334	        _reclaimed : Unit <- reclaim.file(target, info)
335	        reclaim.remove(path, rest)
336	
337	def reclaim.names(names: List<&1, String>) -> List<&2, String>:
338	  match names:
339	    case Nil{}: Nil{}
340	    case Con{name, rest}: Con{name, reclaim.names(rest)}
341	
342	def reclaim.listed(
343	  +path: String,
344	  value: ByteManifest.ManifestValue,
345	  listed: Result<&1, &1, U32 & String, List<&1, String>>
346	) -> IO(Unit):
347	  match listed:
348	    case Fail{_}: IO.pure(Unit, Unit{})
349	    case Done{names}:
350	      reclaim.remove(path, ByteManifest.retired(value, reclaim.names(names)))
351	
352	# Removes SST files the manifest no longer references, best effort, while holding the lock.
353	def reclaim(+path: String, value: ByteManifest.ManifestValue) -> IO(Unit):
354	  do IO<Unit>:
355	    listed : Result<&1, &1, U32 & String, List<&1, String>> <- Fs.read_dir(path)
356	    reclaim.listed(path, value, listed)
357	
358	# Reclaims retired SST files only after the database opened successfully.
359	def open.reclaim.opened(
360	  +path: String,
361	  value: ByteManifest.ManifestValue,
362	  opened: Result<&1, &1, Error, Handle>
363	) -> IO(Result<&1, &1, Error, Handle>):
364	  match opened:
365	    case Fail{error}: IO.pure(Result<&1, &1, Error, Handle>, Fail{error})
366	    case Done{handle}:
367	      do IO<Result<&1, &1, Error, Handle>>:
368	        _reclaimed : Unit <- reclaim(path, value)
369	        return Done{handle}
370	
371	def open.manifest.reclaimed(
372	  +path: String,
373	  lock: DbLock.Lock,
374	  result: Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue>
375	) -> IO(Result<&1, &1, Error, Handle>):
376	  match result:
377	    case Fail{error}: open.manifest.result(path, lock, Fail{error})
378	    case Done{+value}:
379	      do IO<Result<&1, &1, Error, Handle>>:
380	        opened : Result<&1, &1, Error, Handle> <- open.manifest.result(path, lock, Done{value})
381	        open.reclaim.opened(path, value, opened)
382	
383	# Checks for a published byte SST while retaining database ownership.
384	def open.locked(
385	  +path: String,
386	  result: Result<&1, &1, DurableError.Error, DbLock.Lock>
387	) -> IO(Result<&1, &1, Error, Handle>):
388	  match result:
389	    case Fail{error}:
390	      IO.pure(Result<&1, &1, Error, Handle>, Fail{lock.error(error)})
391	    case Done{lock}:
392	      do IO<Result<&1, &1, Error, Handle>>:
393	        manifest : Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue> <-
394	          ByteManifestIo.read(path ++ "/MANIFEST")
395	        open.manifest.reclaimed(path, lock, manifest)
396	
397	# Acquires exclusive ownership before replaying the byte WAL.
398	def open.canonical(
399	  result: Result<&1, &1, U32 & String, String>
400	) -> IO(Result<&1, &1, Error, Handle>):
401	  match result:
402	    case Fail{error}:
403	      IO.pure(Result<&1, &1, Error, Handle>, Fail{io.error(error)})
404	    case Done{+path}:
405	      do IO<Result<&1, &1, Error, Handle>>:
406	        acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <-
407	          DbLock.acquire(path, "open_existing")
408	        open.locked(path, acquired)
409	
410	def open.existing(+path: String) -> IO(Result<&1, &1, Error, Handle>):
411	  do IO<Result<&1, &1, Error, Handle>>:
412	    canonical : Result<&1, &1, U32 & String, String> <- Fs.canonical_path(path)
413	    open.canonical(canonical)
414	
415	# Validates the WAL path before acquiring ownership.
416	def open.existing.present(
417	  +path: String,
418	  result: Result<&1, &1, U32 & String, Bool>
419	) -> IO(Result<&1, &1, Error, Handle>):
420	  match result:
421	    case Fail{error}:
422	      IO.pure(Result<&1, &1, Error, Handle>, Fail{io.error(error)})
423	    case Done{False{}}:
424	      IO.pure(Result<&1, &1, Error, Handle>, Fail{NotFound{}})
425	    case Done{True{}}:
426	      open.existing(path)
427	
428	def open.existing.wait.canonical(
429	  result: Result<&1, &1, U32 & String, String>,
430	  +timeout_ms: U32,
431	  +backoff_ms: U32
432	) -> IO(Result<&1, &1, Error, Handle>):
433	  match result:
434	    case Fail{error}:
435	      IO.pure(Result<&1, &1, Error, Handle>, Fail{io.error(error)})
436	    case Done{+path}:
437	      do IO<Result<&1, &1, Error, Handle>>:
438	        acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <-
439	          DbLock.acquire_wait(path, "open_existing", timeout_ms, backoff_ms)
440	        open.locked(path, acquired)
441	
442	def open.existing.wait.present(
443	  +path: String,
444	  +timeout_ms: U32,
445	  +backoff_ms: U32,
446	  result: Result<&1, &1, U32 & String, Bool>
447	) -> IO(Result<&1, &1, Error, Handle>):
448	  match result:
449	    case Fail{error}:
450	      IO.pure(Result<&1, &1, Error, Handle>, Fail{io.error(error)})
451	    case Done{False{}}:
452	      IO.pure(Result<&1, &1, Error, Handle>, Fail{NotFound{}})
453	    case Done{True{}}:
454	      do IO<Result<&1, &1, Error, Handle>>:
455	        canonical : Result<&1, &1, U32 & String, String> <- Fs.canonical_path(path)
456	        open.existing.wait.canonical(canonical, timeout_ms, backoff_ms)
457	
458	# Opens an existing byte database after immediate or bounded lock acquisition.
459	def open_existing_wait(
460	  +path: String,
461	  timeout_ms: U32,
462	  backoff_ms: U32
463	) -> IO(Result<&1, &1, Error, Handle>):
464	  do IO<Result<&1, &1, Error, Handle>>:
465	    present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path ++ "/WAL")
466	    open.existing.wait.present(path, timeout_ms, backoff_ms, present)
467	
468	# Opens an existing byte database and replays its binary WAL.
469	def open_existing(+path: String) -> IO(Result<&1, &1, Error, Handle>):
470	  do IO<Result<&1, &1, Error, Handle>>:
471	    present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path ++ "/WAL")
472	    open.existing.present(path, present)
473	
474	# Returns the handle and preserves the decoded batch result.
475	def append.decoded(
476	  result: Result<&1, &1, Wal.Error, ByteWal.Batch>
477	) -> Result<&1, &1, Error, ByteWal.Batch>:
478	  match result:
479	    case Fail{_}:
480	      Fail{Corruption{"encoded byte WAL frame failed self-validation"}}
481	    case Done{batch}:
482	      Done{batch}
483	
484	# Poisons a handle when append or sync leaves commit status uncertain.
485	def append.unknown(
486	  +path: String,
487	  lock: DbLock.Lock,
488	  table: ByteMemTable.Table,
489	  error: U32 & String
490	) -> Handle & Result<&1, &1, Error, Unit>:
491	  match error:
492	    case (code, message):
493	      (Handle{path, lock, table, Poisoned{}, VisibleStats.Stats{0n, 0n}}, Fail{CommitUnknown{code, message}})
494	
495	# Appends the encoded frame and preserves a copy for post-sync replay.
496	def append.applied.decoded(
497	  +path: String,
498	  lock: DbLock.Lock,
499	  table: ByteMemTable.Table,
500	  result: Result<&1, &1, Error, ByteWal.Batch>
501	) -> Handle & Result<&1, &1, Error, Unit>:
502	  match result:
503	    case Fail{error}:
504	      (Handle{path, lock, table, Poisoned{}, VisibleStats.Stats{0n, 0n}}, Fail{CommitUnknown{0, "invalid durable frame"}})
505	    case Done{batch}:
506	      (Handle{path, lock, ByteWalIo.apply.batch(table, batch), Open{}, VisibleStats.Stats{0n, 0n}}, Done{Unit{}})
507	
508	def append.applied(
509	  +path: String,
510	  lock: DbLock.Lock,
511	  table: ByteMemTable.Table,
512	  result: Result<&1, &1, Wal.Error, ByteWal.Batch>
513	) -> Handle & Result<&1, &1, Error, Unit>:
514	  append.applied.decoded(path, lock, table, append.decoded(result))
515	
516	# Converts decoded frame validation into a committed handle state.
517	def append.synced(
518	  +path: String,
519	  lock: DbLock.Lock,
520	  table: ByteMemTable.Table,
521	  apply_frame: Bytes.Bytes,
522	  result: Result<&1, &1, U32 & String, Unit>
523	) -> IO(Handle & Result<&1, &1, Error, Unit>):
524	  match result:
525	    case Fail{error}:
526	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
527	        append.unknown(path, lock, table, error))
528	    case Done{Unit{}}:
529	      do IO<Handle & Result<&1, &1, Error, Unit>>:
530	        _checkpoint : Result<&1, &1, U32 & String, Unit> <-
531	          CrashPoint.hit("wal.after_sync_before_apply")
532	        return append.applied(path, lock, table, ByteWal.decode(apply_frame))
533	
534	# Replaces the in-memory table only after the WAL frame is durable.
535	def append.frames(
536	  +path: String,
537	  lock: DbLock.Lock,
538	  table: ByteMemTable.Table,
539	  frames: Bytes.Bytes & Bytes.Bytes
540	) -> IO(Handle & Result<&1, &1, Error, Unit>):
541	  match frames:
542	    case (append_frame, apply_frame):
543	      do IO<Handle & Result<&1, &1, Error, Unit>>:
544	        appended : Result<&1, &1, U32 & String, Unit> <-
545	          ByteWalIo.append.encoded(Done{append_frame}, path ++ "/WAL")
546	        append.synced(path, lock, table, apply_frame, appended)
547	
548	# Distinguishes pre-append encoding rejection from uncertain host writes.
549	def append.encoded(
550	  +path: String,
551	  lock: DbLock.Lock,
552	  table: ByteMemTable.Table,
553	  result: Result<&1, &1, Wal.Error, Bytes.Bytes>
554	) -> IO(Handle & Result<&1, &1, Error, Unit>):
555	  match result:
556	    case Fail{_}:
557	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
558	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{InvalidArgument{}}))
559	    case Done{frame}:
560	      append.frames(path, lock, table, ByteMemTable.bytes.clone(frame))
561	
562	# Validates open state before encoding or touching the WAL.
563	def write.encoded(
564	  +path: String,
565	  lock: DbLock.Lock,
566	  table: ByteMemTable.Table,
567	  state: HandleState,
568	  batch: ByteWal.Batch
569	) -> IO(Handle & Result<&1, &1, Error, Unit>):
570	  match state:
571	    case Open{}:
572	      append.encoded(path, lock, table, ByteWal.encode(batch))
573	    case Poisoned{}:
574	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
575	        (Handle{path, lock, table, Poisoned{}, VisibleStats.Stats{0n, 0n}}, Fail{CommitUnknown{0, "database handle requires reopen"}}))
576	
577	def write.count.validated(
578	  +path: String,
579	  lock: DbLock.Lock,
580	  table: ByteMemTable.Table,
581	  ordered: List<&1, ByteWal.Mutation>,
582	  validation: BatchPolicy.Validation
583	) -> IO(Handle & Result<&1, &1, Error, Unit>):
584	  match validation:
585	    case BatchPolicy.ValidBatch{}:
586	      write.encoded(path, lock, table, Open{}, ByteWal.Batch{ordered})
587	    case BatchPolicy.BatchBytesExceeded{}:
588	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
589	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
590	          Fail{ResourceLimit{}}))
591	    case BatchPolicy.BatchCountExceeded{}:
592	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
593	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
594	          Fail{ResourceLimit{}}))
595	    case _:
596	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
597	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{InvalidArgument{}}))
598	
599	def write.count.measured(
600	  +path: String,
601	  lock: DbLock.Lock,
602	  table: ByteMemTable.Table,
603	  measured: BatchPolicy.ByteBatchMeasurement
604	) -> IO(Handle & Result<&1, &1, Error, Unit>):
605	  match measured:
606	    case BatchPolicy.BatchMeasured{mutations, count, encoded_bytes}:
607	      write.count.validated(path, lock, table, mutations,
608	        BatchPolicy.validate_batch(count, encoded_bytes))
609	    case BatchPolicy.BatchTooMany{}:
610	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
611	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
612	          Fail{ResourceLimit{}}))
613	
614	def write.counted(
615	  +path: String,
616	  lock: DbLock.Lock,
617	  table: ByteMemTable.Table,
618	  state: HandleState,
619	  batch: ByteWal.Batch
620	) -> IO(Handle & Result<&1, &1, Error, Unit>):
621	  match state:
622	    case Poisoned{}:
623	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
624	        (Handle{path, lock, table, Poisoned{}, VisibleStats.Stats{0n, 0n}},
625	          Fail{CommitUnknown{0, "database handle requires reopen"}}))
626	    case Open{}:
627	      match batch:
628	        case ByteWal.Batch{mutations}:
629	          write.count.measured(path, lock, table,
630	            BatchPolicy.byte_batch.measure(mutations))
631	
632	# Keeps the original state when SST or WAL checkpoint publication fails.
633	# Counts the visible entries and payload bytes in the newly published SST.
634	def flush.stats.entries(
635	  entries: List<&1, ByteMemTable.Entry>,
636	  current: VisibleStats.Stats
637	) -> VisibleStats.Stats:
638	  match entries:
639	    case Nil{}: current
640	    case Con{ByteMemTable.Entry{Bytes.Bytes{+key_len, _}, value}, tail}:
641	      match value:
642	        case None{}: flush.stats.entries(tail, current)
643	        case Some{Bytes.Bytes{+value_len, _}}:
644	          flush.stats.entries(tail, VisibleStats.result(VisibleStats.apply(
645	            current, U32.to_nat(key_len), None{}, Some{U32.to_nat(value_len)})))
646	
647	# Applies the flush transition after the WAL checkpoint is durable.
648	def flush.wal.reset(
649	  +path: String,
650	  lock: DbLock.Lock,
651	  table: ByteMemTable.Table,
652	  entries: List<&1, ByteMemTable.Entry>,
653	  result: Result<&1, &1, U32 & String, Unit>
654	) -> Handle & Result<&1, &1, Error, Unit>:
655	  match result:
656	    case Fail{(code, message)}:
657	      (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{Io{code, message}})
658	    case Done{Unit{}}:
659	      (Handle{path, lock, ByteMemTable.empty(), Open{},
660	        flush.stats.entries(entries, VisibleStats.Stats{0n, 0n})}, Done{Unit{}})
661	
662	def flush.manifest(
663	  +path: String,
664	  lock: DbLock.Lock,
665	  table: ByteMemTable.Table,
666	  entries: List<&1, ByteMemTable.Entry>,
667	  result: Result<&1, &1, U32 & String, Unit>
668	) -> IO(Handle & Result<&1, &1, Error, Unit>):
669	  match result:
670	    case Fail{(code, message)}:
671	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
672	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{Io{code, message}}))
673	    case Done{Unit{}}:
674	      do IO<Handle & Result<&1, &1, Error, Unit>>:
675	        _checkpoint : Result<&1, &1, U32 & String, Unit> <-
676	          CrashPoint.hit("flush.manifest_published")
677	        reset : Result<&1, &1, U32 & String, Unit> <- ByteWalIo.reset_atomic(path)
678	        return flush.wal.reset(path, lock, table, entries, reset)
679	
680	# Keeps the original table when SST publication fails; after publication,
681	# checkpoints the WAL through atomic replacement and retains canonical entries.
682	# Reverses a string while preserving all path characters.
683	def path.reverse(path: String, reversed: String) -> String:
684	  match path:
685	    case SNil{}: reversed
686	    case SCon{character, rest}:
687	      path.reverse(rest, SCon{character, reversed})
688	
689	# Creates an independent string from its characters.
690	def path.clone(path: String) -> String:
691	  path.reverse(path.reverse(path, SNil{}), SNil{})
692	
693	# Copies one natural value into two equal affine outputs.
694	def duplicate_nat_pair_step(pair: Nat & Nat) -> Nat & Nat:
695	  match pair:
696	    case (left, right): (Nat.add(left, 1n), Nat.add(right, 1n))
697	
698	# Reifies one natural number as two independently owned values.
699	def duplicate_nat_pair(value: Nat) -> Nat & Nat:
700	  match value:
701	    case 0n: (0n, 0n)
702	    case 1n+rest: duplicate_nat_pair_step(duplicate_nat_pair(rest))
703	
704	# Prepends one scalar to each independently rebuilt path.
705	def path.clone.pair.scalar(
706	  first: String,
707	  second: String,
708	  pair: Nat & Nat
709	) -> String & String:
710	  match pair:
711	    case (left, right):
712	      (SCon{Chr{U32.from_nat(left)}, first},
713	        SCon{Chr{U32.from_nat(right)}, second})
714	
715	# Rebuilds two identical paths in one traversal.
716	def path.clone.pair.go(
717	  path: String,
718	  copies: String & String
719	) -> String & String:
720	  match path:
721	    case SNil{}:
722	      match copies:
723	        case (first, second):
724	          (path.reverse(first, SNil{}), path.reverse(second, SNil{}))
725	    case SCon{Chr{+scalar}, rest}:
726	      match copies:
727	        case (first, second):
728	          path.clone.pair.go(rest, path.clone.pair.scalar(first, second,
729	            duplicate_nat_pair(U32.to_nat(scalar))))
730	
731	# Produces two owned copies of a canonical filesystem path.
732	def path.clone.pair(path: String) -> String & String:
733	  path.clone.pair.go(path, (SNil{}, SNil{}))
734	
735	# Pairs a retained handle with an independently owned database path.
736	def handle_path_copy_pair(
737	  paths: String & String,
738	  lock: DbLock.Lock,
739	  table: ByteMemTable.Table,
740	  state: HandleState,
741	  base_stats: VisibleStats.Stats
742	) -> Handle & String:
743	  match paths:
744	    case (copy, retained):
745	      (Handle{retained, lock, table, state, base_stats}, copy)
746	
747	
748	def handle.path.copy(handle: Handle) -> Handle & String:
749	  match handle:
750	    case Handle{root_path, lock, table, state, base_stats}:
751	      handle_path_copy_pair(path.clone.pair(root_path), lock, table, state,
752	        base_stats)
753	
754	# Combines an initial path copy with two independently rebuilt copies.
755	def path.clone.triple.second(
756	  first: String,
757	  pair: String & String
758	) -> String & String & String:
759	  match pair:
760	    case (second, third): (first, second, third)
761	
762	# Splits the duplicated suffix into the remaining two path copies.
763	def path.clone.triple.first(pair: String & String) -> String & String & String:
764	  match pair:
765	    case (first, suffix):
766	      path.clone.triple.second(first, path.clone.pair(suffix))
767	
768	# Produces three independently owned copies of a canonical path.
769	def path.clone.triple(path: String) -> String & String & String:
770	  path.clone.triple.first(path.clone.pair(path))
771	
772	
773	def flush.written(
774	  +path: String,
775	  lock: DbLock.Lock,
776	  table: ByteMemTable.Table,
777	  entries: List<&1, ByteMemTable.Entry>,
778	  file_id: U32,
779	  result: Result<&1, &1, U32 & String, Unit>
780	) -> IO(Handle & Result<&1, &1, Error, Unit>):
781	  match result:
782	    case Fail{(code, message)}:
783	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
784	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{Io{code, message}}))
785	    case Done{Unit{}}:
786	      do IO<Handle & Result<&1, &1, Error, Unit>>:
787	        _checkpoint : Result<&1, &1, U32 & String, Unit> <-
788	          CrashPoint.hit("flush.table_synced")
789	        published : Result<&1, &1, U32 & String, Unit> <-
790	          ByteManifestIo.write_atomic(path, ByteManifest.GenerationSst{file_id})
791	        flush.manifest(path, lock, table, entries, published)
792	
793	def flush.snapshot.publish.file_ids(
794	  paths: String & (String & String),
795	  lock: DbLock.Lock,
796	  table: ByteMemTable.Table,
797	  pair: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>,
798	  file_ids: U32 & U32
799	) -> IO(Handle & Result<&1, &1, Error, Unit>):
800	  match paths:
801	    case (directory, (file_directory, retained_path)):
802	      match pair:
803	        case (encoded, retained):
804	          match file_ids:
805	            case (for_path, for_manifest):
806	              do IO<Handle & Result<&1, &1, Error, Unit>>:
807	                written : Result<&1, &1, U32 & String, Unit> <-
808	                  ByteSstIo.write_atomic_at(directory,
809	                    ByteManifest.sst.generation.path(file_directory, for_path), encoded)
810	                flush.written(retained_path, lock, table, retained, for_manifest, written)
811	
812	def flush.snapshot.publish(
813	  paths: String & (String & String),
814	  lock: DbLock.Lock,
815	  table: ByteMemTable.Table,
816	  pair: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>,
817	  file_id: U32
818	) -> IO(Handle & Result<&1, &1, Error, Unit>):
819	  flush.snapshot.publish.file_ids(paths, lock, table, pair,
820	    ByteManifest.clone.file.id(file_id))
821	
822	def flush.snapshot.identifier(
823	  paths: String & (String & String),
824	  lock: DbLock.Lock,
825	  table: ByteMemTable.Table,
826	  entries: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>,
827	  decision: ByteManifest.FileIdDecision
828	) -> IO(Handle & Result<&1, &1, Error, Unit>):
829	  match paths:
830	    case (directory, (file_directory, retained_path)):
831	      match entries:
832	        case (encoded, retained):
833	          match decision:
834	            case ByteManifest.FileIdAvailable{file_id}:
835	              flush.snapshot.publish((directory, (file_directory, retained_path)),
836	                lock, table, (encoded, retained), file_id)
837	            case ByteManifest.FileIdExhausted{}:
838	              IO.pure(Handle & Result<&1, &1, Error, Unit>,
839	                (Handle{retained_path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
840	                  Fail{ResourceLimit{}}))
841	
842	def flush.snapshot.manifest(
843	  path: String,
844	  lock: DbLock.Lock,
845	  table: ByteMemTable.Table,
846	  entries: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>,
847	  result: Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue>
848	) -> IO(Handle & Result<&1, &1, Error, Unit>):
849	  match result:
850	    case Fail{ByteManifestIo.Invalid{_}}:
851	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
852	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
853	          Fail{Corruption{"byte manifest validation failed"}}))
854	    case Fail{ByteManifestIo.Host{code, message}}:
855	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
856	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
857	          Fail{Io{code, message}}))
858	    case Done{value}:
859	      flush.snapshot.identifier(path.clone.triple(path), lock, table, entries,
860	        ByteManifest.next.file.id(value))
861	
862	# Retains the live table while an independently owned ordered copy is encoded.
863	def flush.snapshot(
864	  +path: String,
865	  lock: DbLock.Lock,
866	  table: ByteMemTable.Table,
867	  pair: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
868	) -> IO(Handle & Result<&1, &1, Error, Unit>):
869	  match pair:
870	    case (encoded, retained):
871	      do IO<Handle & Result<&1, &1, Error, Unit>>:
872	        manifest : Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue> <-
873	          ByteManifestIo.read(path ++ "/MANIFEST")
874	        flush.snapshot.manifest(path, lock, table, (encoded, retained), manifest)
875	
876	# Builds a canonical snapshot without consuming the live table.
877	def flush.table(
878	  +path: String,
879	  lock: DbLock.Lock,
880	  pair: ByteMemTable.Table & List<&1, ByteMemTable.Entry>
881	) -> IO(Handle & Result<&1, &1, Error, Unit>):
882	  match pair:
883	    case (table, entries):
884	      flush.snapshot(path, lock, table,
885	        ByteSstFile.entries.clone(ByteMemTable.ordered_entries(ByteMemTable.Table{entries})))
886	
887	# Clones the table entries so flush can retain the live in-memory state.
888	def flush.cloned(
889	  +path: String,
890	  lock: DbLock.Lock,
891	  pair: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
892	) -> IO(Handle & Result<&1, &1, Error, Unit>):
893	  match pair:
894	    case (live, snapshot):
895	      flush.table(path, lock, (ByteMemTable.Table{live}, snapshot))
896	
897	# Merges the persisted SST beneath the newest in-memory WAL delta.
898	def flush.sst.loaded(
899	  +path: String,
900	  lock: DbLock.Lock,
901	  table: ByteMemTable.Table,
902	  result: Result<&1, &1, U32 & String, ByteMemTable.Table>
903	) -> IO(Handle & Result<&1, &1, Error, Unit>):
904	  match table:
905	    case ByteMemTable.Table{delta}:
906	      match result:
907	        case Fail{(code, message)}:
908	          IO.pure(Handle & Result<&1, &1, Error, Unit>,
909	            (Handle{path, lock, ByteMemTable.Table{delta}, Open{}, VisibleStats.Stats{0n, 0n}},
910	              Fail{recovery.error(code, message)}))
911	        case Done{ByteMemTable.Table{base}}:
912	          flush.cloned(path, lock,
913	            ByteSstFile.entries.clone(ByteMemTable.append(delta, base)))
914	
915	def flush.manifest.loaded(
916	  +path: String,
917	  lock: DbLock.Lock,
918	  table: ByteMemTable.Table,
919	  result: Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue>
920	) -> IO(Handle & Result<&1, &1, Error, Unit>):
921	  match table:
922	    case ByteMemTable.Table{entries}:
923	      match result:
924	        case Fail{ByteManifestIo.Invalid{_}}:
925	          IO.pure(Handle & Result<&1, &1, Error, Unit>,
926	            (Handle{path, lock, ByteMemTable.Table{entries}, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{Corruption{"byte manifest validation failed"}}))
927	        case Fail{ByteManifestIo.Host{code, message}}:
928	          IO.pure(Handle & Result<&1, &1, Error, Unit>,
929	            (Handle{path, lock, ByteMemTable.Table{entries}, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{Io{code, message}}))
930	        case Done{ByteManifest.Empty{}}:
931	          flush.cloned(path, lock, ByteSstFile.entries.clone(entries))
932	        case Done{ByteManifest.GenerationEmpty{}}:
933	          flush.cloned(path, lock, ByteSstFile.entries.clone(entries))
934	        case Done{ByteManifest.HasSst{}}:
935	          do IO<Handle & Result<&1, &1, Error, Unit>>:
936	            base : Result<&1, &1, U32 & String, ByteMemTable.Table> <-
937	              ByteSstIo.read(path ++ "/SST")
938	            flush.sst.loaded(path, lock, ByteMemTable.Table{entries}, base)
939	        case Done{ByteManifest.GenerationSst{file_id}}:
940	          do IO<Handle & Result<&1, &1, Error, Unit>>:
941	            base : Result<&1, &1, U32 & String, ByteMemTable.Table> <-
942	              ByteSstIo.read(ByteManifest.sst.generation.path(path, file_id))
943	            flush.sst.loaded(path, lock, ByteMemTable.Table{entries}, base)
944	
945	# Clones table entries before creating a canonical snapshot.
946	def flush.open(
947	  +path: String,
948	  lock: DbLock.Lock,
949	  table: ByteMemTable.Table
950	) -> IO(Handle & Result<&1, &1, Error, Unit>):
951	  do IO<Handle & Result<&1, &1, Error, Unit>>:
952	    manifest : Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue> <-
953	      ByteManifestIo.read(path ++ "/MANIFEST")
954	    flush.manifest.loaded(path, lock, table, manifest)
955	
956	# Publishes the current byte table as an authenticated SST snapshot.
957	def flush(handle: Handle) -> IO(Handle & Result<&1, &1, Error, Unit>):
958	  match handle:
959	    case Handle{path, lock, table, Open{}, base_stats}:
960	      flush.open(path, lock, table)
961	    case Handle{path, lock, table, Poisoned{}, base_stats}:
962	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
963	        (Handle{path, lock, table, Poisoned{}, VisibleStats.Stats{0n, 0n}}, Fail{CommitUnknown{0, "database handle requires reopen"}}))
964	
965	def compact.entries(
966	  +path: String,
967	  lock: DbLock.Lock,
968	  copies: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
969	) -> IO(Handle & Result<&1, &1, Error, Unit>):
970	  match copies:
971	    case (retained, working):
972	      flush.snapshot(path, lock, ByteMemTable.Table{retained},
973	        ByteSstFile.entries.clone(ByteMemTable.compact.live(
974	          ByteMemTable.ordered_entries(ByteMemTable.Table{working}))))
975	
976	def compact.sst.loaded(
977	  +path: String,
978	  lock: DbLock.Lock,
979	  table: ByteMemTable.Table,
980	  result: Result<&1, &1, U32 & String, ByteMemTable.Table>
981	) -> IO(Handle & Result<&1, &1, Error, Unit>):
982	  match table:
983	    case ByteMemTable.Table{delta}:
984	      match result:
985	        case Fail{(code, message)}:
986	          IO.pure(Handle & Result<&1, &1, Error, Unit>,
987	            (Handle{path, lock, ByteMemTable.Table{delta}, Open{}, VisibleStats.Stats{0n, 0n}},
988	              Fail{recovery.error(code, message)}))
989	        case Done{ByteMemTable.Table{base}}:
990	          compact.entries(path, lock,
991	            ByteSstFile.entries.clone(ByteMemTable.append(delta, base)))
992	
993	def compact.manifest.loaded(
994	  +path: String,
995	  lock: DbLock.Lock,
996	  table: ByteMemTable.Table,
997	  result: Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue>
998	) -> IO(Handle & Result<&1, &1, Error, Unit>):
999	  match table:
1000	    case ByteMemTable.Table{entries}:
1001	      match result:
1002	        case Fail{ByteManifestIo.Invalid{_}}:
1003	          IO.pure(Handle & Result<&1, &1, Error, Unit>,
1004	            (Handle{path, lock, ByteMemTable.Table{entries}, Open{}, VisibleStats.Stats{0n, 0n}},
1005	              Fail{Corruption{"byte manifest validation failed"}}))
1006	        case Fail{ByteManifestIo.Host{code, message}}:
1007	          IO.pure(Handle & Result<&1, &1, Error, Unit>,
1008	            (Handle{path, lock, ByteMemTable.Table{entries}, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{Io{code, message}}))
1009	        case Done{ByteManifest.Empty{}}:
1010	          compact.entries(path, lock, ByteSstFile.entries.clone(entries))
1011	        case Done{ByteManifest.GenerationEmpty{}}:
1012	          compact.entries(path, lock, ByteSstFile.entries.clone(entries))
1013	        case Done{ByteManifest.HasSst{}}:
1014	          do IO<Handle & Result<&1, &1, Error, Unit>>:
1015	            base : Result<&1, &1, U32 & String, ByteMemTable.Table> <-
1016	              ByteSstIo.read(path ++ "/SST")
1017	            compact.sst.loaded(path, lock, ByteMemTable.Table{entries}, base)
1018	        case Done{ByteManifest.GenerationSst{file_id}}:
1019	          do IO<Handle & Result<&1, &1, Error, Unit>>:
1020	            base : Result<&1, &1, U32 & String, ByteMemTable.Table> <-
1021	              ByteSstIo.read(ByteManifest.sst.generation.path(path, file_id))
1022	            compact.sst.loaded(path, lock, ByteMemTable.Table{entries}, base)
1023	
1024	def compact.pathcopies(
1025	  copies: String & String,
1026	  lock: DbLock.Lock,
1027	  table: ByteMemTable.Table
1028	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1029	  match copies:
1030	    case (read_path, retained_path):
1031	      do IO<Handle & Result<&1, &1, Error, Unit>>:
1032	        manifest : Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue> <-
1033	          ByteManifestIo.read(read_path ++ "/MANIFEST")
1034	        compact.manifest.loaded(retained_path, lock, table, manifest)
1035	
1036	# Publishes a canonical byte table with tombstones removed.
1037	def compact(handle: Handle) -> IO(Handle & Result<&1, &1, Error, Unit>):
1038	  match handle:
1039	    case Handle{path, lock, table, Open{}, base_stats}:
1040	      compact.pathcopies(path.clone.pair(path), lock, table)
1041	    case Handle{path, lock, table, Poisoned{}, base_stats}:
1042	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
1043	        (Handle{path, lock, table, Poisoned{}, VisibleStats.Stats{0n, 0n}}, Fail{CommitUnknown{0, "database handle requires reopen"}}))
1044	
1045	def get.sst.close.result(
1046	  +path: String,
1047	  lock: DbLock.Lock,
1048	  table: ByteMemTable.Table,
1049	  value: Maybe<&1, Bytes.Bytes>,
1050	  closed: Result<&1, &1, U32 & String, Unit>
1051	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1052	  match closed:
1053	    case Fail{(code, message)}:
1054	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1055	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{Io{code, message}}))
1056	    case Done{Unit{}}:
1057	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1058	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Done{value}))
1059	
1060	def get.sst.closed(
1061	  +path: String,
1062	  lock: DbLock.Lock,
1063	  table: ByteMemTable.Table,
1064	  value: Maybe<&1, Bytes.Bytes>,
1065	  cursor: SstCursor.Cursor
1066	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1067	  do IO<Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
1068	    closed : Result<&1, &1, U32 & String, Unit> <- SstCursor.close(cursor)
1069	    get.sst.close.result(path, lock, table, value, closed)
1070	
1071	def get.sst.compare.decided(
1072	  value: Bytes.Bytes,
1073	  cursor: SstCursor.Cursor,
1074	  query: Bytes.Bytes,
1075	  order: Cmp
1076	) -> PointRead:
1077	  Compare{value, cursor, query, order}
1078	
1079	def get.sst.compare.order(
1080	  value: Bytes.Bytes,
1081	  cursor: SstCursor.Cursor,
1082	  key: Bytes.Bytes,
1083	  queries: Bytes.Bytes & Bytes.Bytes
1084	) -> PointRead:
1085	  match queries:
1086	    case (for_comparison, for_continue):
1087	      get.sst.compare.decided(value, cursor, for_continue,
1088	        Keys.Bytes.cmp(key, for_comparison))
1089	
1090	def get.sst.exhausted.cursor(
1091	  +path: String,
1092	  lock: DbLock.Lock,
1093	  table: ByteMemTable.Table,
1094	  cursor: SstCursor.Cursor
1095	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1096	  do IO<Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
1097	    _closed : Result<&1, &1, U32 & String, Unit> <- SstCursor.close(cursor)
1098	    return (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{ResourceLimit{}})
1099	
1100	def get.sst.exhausted(
1101	  +path: String,
1102	  lock: DbLock.Lock,
1103	  table: ByteMemTable.Table,
1104	  state: PointRead
1105	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1106	  match state:
1107	    case Need{cursor, _}:
1108	      get.sst.exhausted.cursor(path, lock, table, cursor)
1109	    case Compare{_, cursor, _, _}:
1110	      get.sst.exhausted.cursor(path, lock, table, cursor)
1111	    case Read{step, _}:
1112	      match step:
1113	        case SstCursor.Item{_, cursor}:
1114	          get.sst.exhausted.cursor(path, lock, table, cursor)
1115	        case SstCursor.Skip{cursor}:
1116	          get.sst.exhausted.cursor(path, lock, table, cursor)
1117	        case SstCursor.End{}:
1118	          IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1119	            (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Done{None{}}))
1120	        case SstCursor.Error{(code, message)}:
1121	          IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1122	            (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{recovery.error(code, message)}))
1123	
1124	def get.sst.loop(
1125	  fuel: Nat,
1126	  +path: String,
1127	  lock: DbLock.Lock,
1128	  table: ByteMemTable.Table,
1129	  state: PointRead
1130	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1131	  match fuel:
1132	    case 0n:
1133	      get.sst.exhausted(path, lock, table, state)
1134	    case 1n+rest:
1135	      match state:
1136	        case Need{cursor, query}:
1137	          do IO<Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
1138	            step : SstCursor.Step <- SstCursor.next(cursor)
1139	            get.sst.loop(rest, path, lock, table, Read{step, query})
1140	        case Read{step, query}:
1141	          match step:
1142	            case SstCursor.Item{entry, cursor}:
1143	              match entry:
1144	                case ByteMemTable.Entry{key, val}:
1145	                  match val:
1146	                    case None{}:
1147	                      get.sst.loop(rest, path, lock, table, Need{cursor, query})
1148	                    case Some{value}:
1149	                      get.sst.loop(rest, path, lock, table,
1150	                        get.sst.compare.order(value, cursor, key,
1151	                          ByteMemTable.bytes.clone(query)))
1152	            case SstCursor.Skip{cursor}:
1153	              get.sst.loop(rest, path, lock, table, Need{cursor, query})
1154	            case SstCursor.End{}:
1155	              IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1156	                (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Done{None{}}))
1157	            case SstCursor.Error{(code, message)}:
1158	              IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1159	                (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{recovery.error(code, message)}))
1160	        case Compare{value, cursor, query, order}:
1161	          match order:
1162	            case EQ{}:
1163	              get.sst.closed(path, lock, table, Some{value}, cursor)
1164	            case GT{}:
1165	              get.sst.closed(path, lock, table, None{}, cursor)
1166	            case LT{}:
1167	              get.sst.loop(rest, path, lock, table, Need{cursor, query})
1168	
1169	def get.sst.opened(
1170	  +path: String,
1171	  lock: DbLock.Lock,
1172	  table: ByteMemTable.Table,
1173	  query: Bytes.Bytes,
1174	  size: Nat,
1175	  opened: Result<&1, &1, U32 & String, SstCursor.Cursor>
1176	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1177	  match opened:
1178	    case Fail{(code, message)}:
1179	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1180	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{recovery.error(code, message)}))
1181	    case Done{cursor}:
1182	      get.sst.loop(Nat.add(size, 2n), path, lock, table, Need{cursor, query})
1183	
1184	def get.sst.sized(
1185	  paths: String & String,
1186	  lock: DbLock.Lock,
1187	  table: ByteMemTable.Table,
1188	  keys: Bytes.Bytes & Bytes.Bytes,
1189	  size: Result<&1, &1, U32 & String, Nat>
1190	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1191	  match paths:
1192	    case (cursor_path, retained_path):
1193	      match keys:
1194	        case (lower, query):
1195	          match size:
1196	            case Fail{(code, message)}:
1197	              IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1198	                (Handle{retained_path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1199	                  Fail{io.error((code, message))}))
1200	            case Done{size}:
1201	              do IO<Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
1202	                opened : Result<&1, &1, U32 & String, SstCursor.Cursor> <-
1203	                  SstCursor.open.trusted(cursor_path,
1204	                    SstCursor.Bounds{Some{lower}, None{}, None{}})
1205	                get.sst.opened(retained_path, lock, table, query, size, opened)
1206	
1207	def get.sst.start(
1208	  paths: String & (String & String),
1209	  lock: DbLock.Lock,
1210	  table: ByteMemTable.Table,
1211	  keys: Bytes.Bytes & Bytes.Bytes
1212	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1213	  match paths:
1214	    case (size_path, (cursor_path, retained_path)):
1215	      match keys:
1216	        case (lower, query):
1217	          do IO<Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
1218	            size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(size_path)
1219	            get.sst.sized((cursor_path, retained_path), lock, table,
1220	              (lower, query), size)
1221	
1222	def get.sst.generation.file.paths(
1223	  files: String & String,
1224	  retained_path: String,
1225	  lock: DbLock.Lock,
1226	  table: ByteMemTable.Table,
1227	  keys: Bytes.Bytes & Bytes.Bytes
1228	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1229	  match files:
1230	    case (size_path, cursor_path):
1231	      match keys:
1232	        case (lower, query):
1233	          do IO<Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
1234	            size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(size_path)
1235	            get.sst.sized((cursor_path, retained_path), lock, table,
1236	              (lower, query), size)
1237	
1238	def get.sst.generation.file(
1239	  values: String & String,
1240	  lock: DbLock.Lock,
1241	  table: ByteMemTable.Table,
1242	  keys: Bytes.Bytes & Bytes.Bytes
1243	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1244	  match values:
1245	    case (generation_path, retained_path):
1246	      get.sst.generation.file.paths(path.clone.pair(generation_path),
1247	        retained_path, lock, table, keys)
1248	
1249	def get.sst.generation.read(
1250	  paths: String & String,
1251	  file_id: U32,
1252	  lock: DbLock.Lock,
1253	  table: ByteMemTable.Table,
1254	  key: Bytes.Bytes
1255	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1256	  match paths:
1257	    case (generation_dir, retained_path):
1258	      get.sst.generation.file(
1259	        (ByteManifest.sst.generation.path(generation_dir, file_id), retained_path),
1260	        lock, table, ByteMemTable.bytes.clone(key))
1261	
1262	def get.sst.generation.probe(
1263	  paths: String & String,
1264	  file_id: U32,
1265	  lock: DbLock.Lock,
1266	  key: Bytes.Bytes,
1267	  pair: ByteMemTable.Table & ByteMemTable.Probe
1268	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1269	  match paths:
1270	    case (generation_dir, retained_path):
1271	      match pair:
1272	        case (table, ByteMemTable.Absent{}):
1273	          get.sst.generation.read((generation_dir, retained_path), file_id,
1274	            lock, table, key)
1275	        case (table, ByteMemTable.Deleted{}):
1276	          IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1277	            (Handle{retained_path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1278	              Done{None{}}))
1279	        case (table, ByteMemTable.Present{value}):
1280	          IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1281	            (Handle{retained_path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1282	              Done{Some{value}}))
1283	
1284	def get.sst.generation.probe.start(
1285	  paths: String & String,
1286	  file_id: U32,
1287	  lock: DbLock.Lock,
1288	  keys: Bytes.Bytes & Bytes.Bytes,
1289	  table: ByteMemTable.Table
1290	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1291	  match paths:
1292	    case (generation_dir, retained_path):
1293	      match keys:
1294	        case (probe_key, fallback_key):
1295	          get.sst.generation.probe((generation_dir, retained_path), file_id,
1296	            lock, fallback_key, ByteMemTable.probe(table, probe_key))
1297	
1298	def get.sst.generation.path(
1299	  paths: String & String,
1300	  file_id: U32,
1301	  lock: DbLock.Lock,
1302	  table: ByteMemTable.Table,
1303	  key: Bytes.Bytes
1304	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1305	  get.sst.generation.probe.start(paths, file_id, lock,
1306	    ByteMemTable.bytes.clone(key), table)
1307	
1308	def path.append.sst.triple(paths: String & (String & String)) -> String & (String & String):
1309	  match paths:
1310	    case (size_dir, (cursor_dir, retained_path)):
1311	      (size_dir ++ "/SST", (cursor_dir ++ "/SST", retained_path))
1312	
1313	def get.sst(
1314	  +path: String,
1315	  lock: DbLock.Lock,
1316	  key: Bytes.Bytes,
1317	  table: ByteMemTable.Table
1318	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1319	  get.sst.start(path.append.sst.triple(path.clone.triple(path)), lock, table,
1320	    ByteMemTable.bytes.clone(key))
1321	
1322	def get.delta(
1323	  +path: String,
1324	  lock: DbLock.Lock,
1325	  pair: ByteMemTable.Table & Maybe<&1, Bytes.Bytes>
1326	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1327	  match pair:
1328	    case (updated, value):
1329	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1330	        (Handle{path, lock, updated, Open{}, VisibleStats.Stats{0n, 0n}}, Done{value}))
1331	
1332	def get.sst.probe(
1333	  +path: String,
1334	  lock: DbLock.Lock,
1335	  key: Bytes.Bytes,
1336	  pair: ByteMemTable.Table & ByteMemTable.Probe
1337	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1338	  match pair:
1339	    case (table, ByteMemTable.Absent{}):
1340	      get.sst(path, lock, key, table)
1341	    case (table, ByteMemTable.Deleted{}):
1342	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1343	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Done{None{}}))
1344	    case (table, ByteMemTable.Present{value}):
1345	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1346	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Done{Some{value}}))
1347	
1348	def get.sst.probe.start(
1349	  +path: String,
1350	  lock: DbLock.Lock,
1351	  keys: Bytes.Bytes & Bytes.Bytes,
1352	  table: ByteMemTable.Table
1353	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1354	  match keys:
1355	    case (probe_key, fallback_key):
1356	      get.sst.probe(path, lock, fallback_key,
1357	        ByteMemTable.probe(table, probe_key))
1358	
1359	def get.manifest(
1360	  +path: String,
1361	  lock: DbLock.Lock,
1362	  key: Bytes.Bytes,
1363	  table: ByteMemTable.Table,
1364	  result: Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue>
1365	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1366	  match result:
1367	    case Fail{ByteManifestIo.Invalid{_}}:
1368	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1369	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{Corruption{"byte manifest validation failed"}}))
1370	    case Fail{ByteManifestIo.Host{code, message}}:
1371	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1372	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}}, Fail{Io{code, message}}))
1373	    case Done{ByteManifest.Empty{}}:
1374	      get.delta(path, lock, ByteMemTable.get(table, key))
1375	    case Done{ByteManifest.GenerationEmpty{}}:
1376	      get.delta(path, lock, ByteMemTable.get(table, key))
1377	    case Done{ByteManifest.HasSst{}}:
1378	      get.sst.probe.start(path, lock, ByteMemTable.bytes.clone(key), table)
1379	    case Done{ByteManifest.GenerationSst{file_id}}:
1380	      get.sst.generation.path(path.clone.pair(path), file_id, lock, table, key)
1381	
1382	def get.pathcopies(
1383	  copies: String & String,
1384	  lock: DbLock.Lock,
1385	  key: Bytes.Bytes,
1386	  table: ByteMemTable.Table
1387	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1388	  match copies:
1389	    case (read_path, retained_path):
1390	      do IO<Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
1391	        manifest : Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue> <-
1392	          ByteManifestIo.read(read_path ++ "/MANIFEST")
1393	        get.manifest(retained_path, lock, key, table, manifest)
1394	
1395	# Reads one raw byte key while retaining the database handle.
1396	def get.retained(
1397	  base_stats: VisibleStats.Stats,
1398	  pair: Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>
1399	) -> Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>:
1400	  match pair:
1401	    case (Handle{path, lock, table, state, _}, result):
1402	      (Handle{path, lock, table, state, base_stats}, result)
1403	
1404	def get.length.sst.result(
1405	  +path: String,
1406	  lock: DbLock.Lock,
1407	  table: ByteMemTable.Table,
1408	  result: Result<&1, &1, U32 & String, ByteWal.ValueMetadata>
1409	) -> Handle & Result<&1, &1, Error, Maybe<&1, U32>>:
1410	  match result:
1411	    case Fail{(code, message)}:
1412	      (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1413	        Fail{recovery.error(code, message)})
1414	    case Done{ByteWal.ValueAbsent{}}:
1415	      (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1416	        Done{None{}})
1417	    case Done{ByteWal.ValueDeleted{}}:
1418	      (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1419	        Done{None{}})
1420	    case Done{ByteWal.ValuePresent{value_bytes}}:
1421	      (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1422	        Done{Some{value_bytes}})
1423	
1424	def get.length.sst(
1425	  paths: String & String,
1426	  lock: DbLock.Lock,
1427	  table: ByteMemTable.Table,
1428	  key: Bytes.Bytes
1429	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, U32>>):
1430	  match paths:
1431	    case (sst_path, db_path):
1432	      do IO<Handle & Result<&1, &1, Error, Maybe<&1, U32>>>:
1433	        measured : Result<&1, &1, U32 & String, ByteWal.ValueMetadata> <-
1434	          SstCursor.measure_key(sst_path, key)
1435	        return get.length.sst.result(db_path, lock, table, measured)
1436	
1437	def get.length.manifest.fallback(
1438	  +path: String,
1439	  lock: DbLock.Lock,
1440	  key: Bytes.Bytes,
1441	  table: ByteMemTable.Table,
1442	  result: Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue>
1443	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, U32>>):
1444	  match result:
1445	    case Fail{ByteManifestIo.Invalid{_}}:
1446	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, U32>>,
1447	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1448	          Fail{Corruption{"byte manifest validation failed"}}))
1449	    case Fail{ByteManifestIo.Host{code, message}}:
1450	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, U32>>,
1451	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1452	          Fail{Io{code, message}}))
1453	    case Done{ByteManifest.Empty{}}:
1454	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, U32>>,
1455	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1456	          Done{None{}}))
1457	    case Done{ByteManifest.GenerationEmpty{}}:
1458	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, U32>>,
1459	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1460	          Done{None{}}))
1461	    case Done{ByteManifest.HasSst{}}:
1462	      get.length.sst((path.clone.pair(path ++ "/SST")), lock, table, key)
1463	    case Done{ByteManifest.GenerationSst{file_id}}:
1464	      get.length.sst((ByteManifest.sst.generation.path(path, file_id), path),
1465	        lock, table, key)
1466	
1467	def get.length.manifest(
1468	  +path: String,
1469	  lock: DbLock.Lock,
1470	  key: Bytes.Bytes,
1471	  table: ByteMemTable.Table,
1472	  result: Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue>
1473	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, U32>>):
1474	  get.length.manifest.fallback(path, lock, key, table, result)
1475	
1476	def get.length.probed(
1477	  +path: String,
1478	  lock: DbLock.Lock,
1479	  key: Bytes.Bytes,
1480	  pair: ByteMemTable.Table & ByteMemTable.LengthProbe
1481	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, U32>>):
1482	  match pair:
1483	    case (table, ByteMemTable.LengthDeleted{}):
1484	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, U32>>,
1485	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1486	          Done{None{}}))
1487	    case (table, ByteMemTable.LengthPresent{value_bytes}):
1488	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, U32>>,
1489	        (Handle{path, lock, table, Open{}, VisibleStats.Stats{0n, 0n}},
1490	          Done{Some{value_bytes}}))
1491	    case (table, ByteMemTable.LengthAbsent{}):
1492	      do IO<Handle & Result<&1, &1, Error, Maybe<&1, U32>>>:
1493	        manifest : Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue> <-
1494	          ByteManifestIo.read(path ++ "/MANIFEST")
1495	        get.length.manifest(path, lock, key, table, manifest)
1496	
1497	def get.length.delta(
1498	  +path: String,
1499	  lock: DbLock.Lock,
1500	  keys: Bytes.Bytes & Bytes.Bytes,
1501	  table: ByteMemTable.Table
1502	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, U32>>):
1503	  match keys:
1504	    case (probe_key, fallback_key):
1505	      get.length.probed(path, lock, fallback_key,
1506	        ByteMemTable.probe.length(table, probe_key))
1507	
1508	def value_length.preserve.handle(
1509	  +path: String,
1510	  lock: DbLock.Lock,
1511	  table: ByteMemTable.Table,
1512	  state: HandleState,
1513	  base_stats: VisibleStats.Stats,
1514	  result: Result<&1, &1, Error, Maybe<&1, U32>>
1515	) -> Handle & Result<&1, &1, Error, Maybe<&1, U32>>:
1516	  (Handle{path, lock, table, state, base_stats}, result)
1517	
1518	def value_length.preserve(
1519	  base_stats: VisibleStats.Stats,
1520	  pair: Handle & Result<&1, &1, Error, Maybe<&1, U32>>
1521	) -> Handle & Result<&1, &1, Error, Maybe<&1, U32>>:
1522	  match pair:
1523	    case (Handle{path, lock, table, state, _}, Done{Some{value_bytes}}):
1524	      value_length.preserve.handle(path, lock, table, state,
1525	        ValueLengthPolicy.after_lookup(base_stats,
1526	          ValueLengthPolicy.ValueFound{}), Done{Some{value_bytes}})
1527	    case (Handle{path, lock, table, state, _}, Done{None{}}):
1528	      value_length.preserve.handle(path, lock, table, state,
1529	        ValueLengthPolicy.after_lookup(base_stats,
1530	          ValueLengthPolicy.ValueMissing{}), Done{None{}})
1531	    case (Handle{path, lock, table, state, _}, Fail{error}):
1532	      value_length.preserve.handle(path, lock, table, state,
1533	        ValueLengthPolicy.after_lookup(base_stats,
1534	          ValueLengthPolicy.LookupFailed{}), Fail{error})
1535	
1536	# Handles value length.
1537	def value_length(
1538	  handle: Handle,
1539	  key: Bytes.Bytes
1540	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, U32>>):
1541	  match handle:
1542	    case Handle{path, lock, table, Open{}, base_stats}:
1543	      do IO<Handle & Result<&1, &1, Error, Maybe<&1, U32>>>:
1544	        measured : Handle & Result<&1, &1, Error, Maybe<&1, U32>> <-
1545	          get.length.delta(path, lock, ByteMemTable.bytes.clone(key), table)
1546	        return value_length.preserve(base_stats, measured)
1547	    case Handle{path, lock, table, Poisoned{}, base_stats}:
1548	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, U32>>,
1549	        (Handle{path, lock, table, Poisoned{}, base_stats},
1550	          Fail{CommitUnknown{0, "database handle requires reopen"}}))
1551	
1552	
1553	# Reads one raw byte key while retaining the database handle.
1554	def get(
1555	  handle: Handle,
1556	  key: Bytes.Bytes
1557	) -> IO(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
1558	  match handle:
1559	    case Handle{path, lock, table, Open{}, base_stats}:
1560	      do IO<Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
1561	        pair : Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>> <-
1562	          get.pathcopies(path.clone.pair(path), lock, key, table)
1563	        return get.retained(base_stats, pair)
1564	    case Handle{path, lock, table, Poisoned{}, base_stats}:
1565	      IO.pure(Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
1566	        (Handle{path, lock, table, Poisoned{}, base_stats},
1567	          Fail{CommitUnknown{0, "database handle requires reopen"}}))
1568	
1569	def stats.remove.base.length(
1570	  base_stats: VisibleStats.Stats,
1571	  key_bytes: Nat,
1572	  old_value: Maybe<&1, Bytes.Bytes>
1573	) -> VisibleStats.Stats:
1574	  match old_value:
1575	    case None{}: base_stats
1576	    case Some{Bytes.Bytes{+value_len, _}}:
1577	      VisibleStats.result(VisibleStats.removed(base_stats, key_bytes,
1578	        U32.to_nat(value_len)))
1579	
1580	def batch.keys.retained(
1581	  probe_key: Bytes.Bytes,
1582	  pair: Bytes.Bytes & Bytes.Bytes
1583	) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
1584	  match pair:
1585	    case (read_key, stats_key): (probe_key, read_key, stats_key)
1586	
1587	def batch.keys.first(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
1588	  match pair:
1589	    case (probe_key, retained):
1590	      batch.keys.retained(probe_key, ByteMemTable.bytes.clone(retained))
1591	
1592	def batch.keys(key: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
1593	  batch.keys.first(ByteMemTable.bytes.clone(key))
1594	
1595	def batch.restore.handle(
1596	  original_stats: VisibleStats.Stats,
1597	  handle: Handle
1598	) -> Handle:
1599	  match handle:
1600	    case Handle{path, lock, table, state, _}:
1601	      Handle{path, lock, table, state, original_stats}
1602	
1603	def batch.build(fuel: Nat, state: BatchBuildState) -> ByteMemTable.Table & List<&1, ByteWal.Mutation>:
1604	  match fuel:
1605	    case 0n:
1606	      match state:
1607	        case BatchBuild{_, table, reversed}:
1608	          (table, List.reverse(&1, ByteWal.Mutation, reversed))
1609	        case BatchBuildPut{_, _, _, table, reversed}:
1610	          (table, List.reverse(&1, ByteWal.Mutation, reversed))
1611	        case BatchBuildDelete{_, _, table, reversed}:
1612	          (table, List.reverse(&1, ByteWal.Mutation, reversed))
1613	    case 1n+rest:
1614	      match state:
1615	        case BatchBuild{mutations, table, reversed}:
1616	          match mutations:
1617	            case Nil{}:
1618	              (table, List.reverse(&1, ByteWal.Mutation, reversed))
1619	            case Con{ByteWal.Put{key, value}, tail}:
1620	              batch.build(rest, BatchBuildPut{ByteMemTable.bytes.clone(key),
1621	                ByteMemTable.bytes.clone(value), tail, table, reversed})
1622	            case Con{ByteWal.Delete{key}, tail}:
1623	              batch.build(rest, BatchBuildDelete{ByteMemTable.bytes.clone(key),
1624	                tail, table, reversed})
1625	        case BatchBuildPut{keys, values, tail, table, reversed}:
1626	          match keys:
1627	            case (candidate_key, wal_key):
1628	              match values:
1629	                case (candidate_value, wal_value):
1630	                  batch.build(rest, BatchBuild{tail,
1631	                    ByteMemTable.put(table, candidate_key, candidate_value),
1632	                    Con{ByteWal.Put{wal_key, wal_value}, reversed}})
1633	        case BatchBuildDelete{keys, tail, table, reversed}:
1634	          match keys:
1635	            case (candidate_key, wal_key):
1636	              batch.build(rest, BatchBuild{tail, ByteMemTable.del(table, candidate_key),
1637	                Con{ByteWal.Delete{wal_key}, reversed}})
1638	
1639	def batch.stats.exhausted(state: BatchStatsState) -> IO(Handle & Result<&1, &1, Error, BatchPrepared>):
1640	  match state:
1641	    case BatchNeed{handle, _, _, original_stats, _}:
1642	      IO.pure(Handle & Result<&1, &1, Error, BatchPrepared>,
1643	        (batch.restore.handle(original_stats, handle), Fail{ResourceLimit{}}))
1644	    case BatchKey{handle, _, _, original_stats, _, _}:
1645	      IO.pure(Handle & Result<&1, &1, Error, BatchPrepared>,
1646	        (batch.restore.handle(original_stats, handle), Fail{ResourceLimit{}}))
1647	    case BatchProbe{handle, _, _, original_stats, _, _, _, _}:
1648	      IO.pure(Handle & Result<&1, &1, Error, BatchPrepared>,
1649	        (batch.restore.handle(original_stats, handle), Fail{ResourceLimit{}}))
1650	    case BatchRead{pair, _, _, original_stats, _, _}:
1651	      match pair:
1652	        case (handle, _):
1653	          IO.pure(Handle & Result<&1, &1, Error, BatchPrepared>,
1654	            (batch.restore.handle(original_stats, handle), Fail{ResourceLimit{}}))
1655	
1656	def batch.stats.loop(fuel: Nat, state: BatchStatsState) -> IO(Handle & Result<&1, &1, Error, BatchPrepared>):
1657	  match fuel:
1658	    case 0n: batch.stats.exhausted(state)
1659	    case 1n+rest:
1660	      match state:
1661	        case BatchNeed{handle, mutations, entries, original_stats, base_stats}:
1662	          match handle:
1663	            case Handle{path, lock, table, state, _}:
1664	              match entries:
1665	                case Nil{}:
1666	                  IO.pure(Handle & Result<&1, &1, Error, BatchPrepared>,
1667	                    (Handle{path, lock, table, state, original_stats},
1668	                      Done{Prepared{mutations, base_stats}}))
1669	                case Con{ByteMemTable.Entry{key, _}, tail}:
1670	                  batch.stats.loop(rest, BatchKey{
1671	                    Handle{path, lock, table, state, VisibleStats.Stats{0n, 0n}},
1672	                    mutations, tail, original_stats, base_stats, batch.keys(key)})
1673	        case BatchKey{handle, mutations, entries, original_stats, base_stats, keys}:
1674	          match handle:
1675	            case Handle{path, lock, table, state, _}:
1676	              match keys:
1677	                case (probe_key, (read_key, stats_key)):
1678	                  match stats_key:
1679	                    case Bytes.Bytes{+key_len, _}:
1680	                      batch.stats.loop(rest, BatchProbe{
1681	                        Handle{path, lock, ByteMemTable.empty(), state,
1682	                          VisibleStats.Stats{0n, 0n}},
1683	                        mutations, entries, original_stats, base_stats,
1684	                        U32.to_nat(key_len), read_key,
1685	                        ByteMemTable.probe(table, probe_key)})
1686	        case BatchProbe{handle, mutations, entries, original_stats,
1687	            base_stats, key_bytes, key, result}:
1688	          match handle:
1689	            case Handle{path, lock, _, state, _}:
1690	              match result:
1691	                case (table, ByteMemTable.Absent{}):
1692	                  do IO<Handle & Result<&1, &1, Error, BatchPrepared>>:
1693	                    pair : Handle & Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>> <-
1694	                      get(Handle{path, lock, table, state,
1695	                        VisibleStats.Stats{0n, 0n}}, key)
1696	                    batch.stats.loop(rest, BatchRead{pair, mutations, entries,
1697	                      original_stats, base_stats, key_bytes})
1698	                case (table, _):
1699	                  batch.stats.loop(rest, BatchNeed{
1700	                    Handle{path, lock, table, state, VisibleStats.Stats{0n, 0n}},
1701	                    mutations, entries, original_stats, base_stats})
1702	        case BatchRead{pair, mutations, entries, original_stats, base_stats, key_bytes}:
1703	          match pair:
1704	            case (handle, Fail{error}):
1705	              IO.pure(Handle & Result<&1, &1, Error, BatchPrepared>,
1706	                (batch.restore.handle(original_stats, handle), Fail{error}))
1707	            case (Handle{path, lock, table, state, _}, Done{old_value}):
1708	              batch.stats.loop(rest, BatchNeed{
1709	                Handle{path, lock, table, state, VisibleStats.Stats{0n, 0n}},
1710	                mutations, entries, original_stats,
1711	                stats.remove.base.length(base_stats, key_bytes, old_value)})
1712	
1713	# Counts records before preparing a state that owns both the original and adjusted aggregates.
1714	def batch.prepare.cloned(
1715	  handle: Handle,
1716	  mutations: List<&1, ByteWal.Mutation>,
1717	  copies: VisibleStats.Stats & VisibleStats.Stats,
1718	  counted: List<&1, ByteMemTable.Entry> & Nat
1719	) -> IO(Handle & Result<&1, &1, Error, BatchPrepared>):
1720	  match handle:
1721	    case Handle{path, lock, table, state, _}:
1722	      match copies:
1723	        case (original_stats, base_stats):
1724	          match counted:
1725	            case (retained_entries, count):
1726	              batch.stats.loop(Nat.add(Nat.mul(Nat.add(count, 1n), 4n), 1n),
1727	                BatchNeed{Handle{path, lock, table, state,
1728	                  VisibleStats.Stats{0n, 0n}}, mutations, retained_entries,
1729	                  original_stats, base_stats})
1730	
1731	def batch.prepare(
1732	  handle: Handle,
1733	  mutations: List<&1, ByteWal.Mutation>,
1734	  entries: List<&1, ByteMemTable.Entry>
1735	) -> IO(Handle & Result<&1, &1, Error, BatchPrepared>):
1736	  match handle:
1737	    case Handle{path, lock, table, state, base_stats}:
1738	      batch.prepare.cloned(Handle{path, lock, table, state,
1739	        VisibleStats.Stats{0n, 0n}},
1740	        mutations, VisibleStats.clone(base_stats),
1741	        ByteMemTable.count.keep(entries))
1742	
1743	def batch.validation.decided(
1744	  handle: Handle,
1745	  entries: List<&1, ByteMemTable.Entry>,
1746	  mutations: List<&1, ByteWal.Mutation>,
1747	  validation: BatchPolicy.Validation
1748	) -> IO(Handle & Result<&1, &1, Error, BatchPrepared>):
1749	  match validation:
1750	    case BatchPolicy.ValidBatch{}:
1751	      batch.prepare(handle, mutations, entries)
1752	    case BatchPolicy.BatchBytesExceeded{}:
1753	      IO.pure(Handle & Result<&1, &1, Error, BatchPrepared>,
1754	        (handle, Fail{ResourceLimit{}}))
1755	    case BatchPolicy.BatchCountExceeded{}:
1756	      IO.pure(Handle & Result<&1, &1, Error, BatchPrepared>,
1757	        (handle, Fail{ResourceLimit{}}))
1758	    case _:
1759	      IO.pure(Handle & Result<&1, &1, Error, BatchPrepared>,
1760	        (handle, Fail{InvalidArgument{}}))
1761	
1762	def batch.validation.result(
1763	  handle: Handle,
1764	  entries: List<&1, ByteMemTable.Entry>,
1765	  measured: BatchPolicy.ByteBatchMeasurement
1766	) -> IO(Handle & Result<&1, &1, Error, BatchPrepared>):
1767	  match measured:
1768	    case BatchPolicy.BatchMeasured{mutations, count, encoded_bytes}:
1769	      batch.validation.decided(handle, entries, mutations,
1770	        BatchPolicy.validate_batch(count, encoded_bytes))
1771	    case BatchPolicy.BatchTooMany{}:
1772	      IO.pure(Handle & Result<&1, &1, Error, BatchPrepared>,
1773	        (handle, Fail{ResourceLimit{}}))
1774	
1775	def batch.validate(
1776	  handle: Handle,
1777	  mutations: List<&1, ByteWal.Mutation>,
1778	  entries: List<&1, ByteMemTable.Entry>
1779	) -> IO(Handle & Result<&1, &1, Error, BatchPrepared>):
1780	  batch.validation.result(handle, entries,
1781	    BatchPolicy.byte_batch.measure(mutations))
1782	
1783	def batch.build.result(
1784	  handle: Handle,
1785	  pair: ByteMemTable.Table & List<&1, ByteWal.Mutation>
1786	) -> IO(Handle & Result<&1, &1, Error, BatchPrepared>):
1787	  match pair:
1788	    case (candidate_table, mutations):
1789	      batch.validate(handle, mutations,
1790	        ByteMemTable.ordered_entries(candidate_table))
1791	
1792	def batch.commit.handle(
1793	  stats: VisibleStats.Stats,
1794	  handle: Handle,
1795	  result: Result<&1, &1, Error, Unit>
1796	) -> Handle & Result<&1, &1, Error, Unit>:
1797	  match handle:
1798	    case Handle{path, lock, table, state, _}:
1799	      (Handle{path, lock, table, state, stats}, result)
1800	
1801	def batch.commit.choose(
1802	  original_stats: VisibleStats.Stats,
1803	  adjusted_stats: VisibleStats.Stats,
1804	  handle: Handle,
1805	  result: Result<&1, &1, Error, Unit>
1806	) -> Handle & Result<&1, &1, Error, Unit>:
1807	  match result:
1808	    case Done{Unit{}}:
1809	      batch.commit.handle(adjusted_stats, handle, Done{Unit{}})
1810	    case Fail{error}:
1811	      batch.commit.handle(original_stats, handle, Fail{error})
1812	
1813	def batch.commit.result(
1814	  original_stats: VisibleStats.Stats,
1815	  adjusted_stats: VisibleStats.Stats,
1816	  pair: Handle & Result<&1, &1, Error, Unit>
1817	) -> Handle & Result<&1, &1, Error, Unit>:
1818	  match pair:
1819	    case (handle, result):
1820	      batch.commit.choose(original_stats, adjusted_stats, handle, result)
1821	
1822	def batch.prepare.validated(
1823	  prepared: Handle & Result<&1, &1, Error, BatchPrepared>
1824	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1825	  match prepared:
1826	    case (next_handle, Fail{error}):
1827	      IO.pure(Handle & Result<&1, &1, Error, Unit>, (next_handle, Fail{error}))
1828	    case (Handle{path, lock, table, state, original_stats}, Done{Prepared{mutations, base_stats}}):
1829	      do IO<Handle & Result<&1, &1, Error, Unit>>:
1830	        pair : Handle & Result<&1, &1, Error, Unit> <-
1831	          write.counted(path, lock, table, state, ByteWal.Batch{mutations})
1832	        return batch.commit.result(original_stats, base_stats, pair)
1833	
1834	def batch.prepare.result(
1835	  prepared: Handle & Result<&1, &1, Error, BatchPrepared>
1836	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1837	  batch.prepare.validated(prepared)
1838	
1839	def batch.prepare.built(
1840	  handle: Handle,
1841	  built: ByteMemTable.Table & List<&1, ByteWal.Mutation>
1842	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1843	  do IO<Handle & Result<&1, &1, Error, Unit>>:
1844	    prepared : Handle & Result<&1, &1, Error, BatchPrepared> <-
1845	      batch.build.result(handle, built)
1846	    result : Handle & Result<&1, &1, Error, Unit> <-
1847	      batch.prepare.result(prepared)
1848	    return result
1849	
1850	def batch.prepare.start(
1851	  handle: Handle,
1852	  fuel: Nat,
1853	  mutations: List<&1, ByteWal.Mutation>
1854	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1855	  batch.prepare.built(handle,
1856	    batch.build(fuel, BatchBuild{mutations, ByteMemTable.empty(), Nil{}}))
1857	
1858	def write_batch.validated(
1859	  handle: Handle,
1860	  mutations: List<&1, ByteWal.Mutation>,
1861	  count: Nat
1862	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1863	  batch.prepare.start(handle, Nat.add(Nat.mul(count, 2n), 1n), mutations)
1864	
1865	def write_batch.validation.result(
1866	  handle: Handle,
1867	  mutations: List<&1, ByteWal.Mutation>,
1868	  count: Nat,
1869	  validation: BatchPolicy.Validation
1870	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1871	  match validation:
1872	    case BatchPolicy.ValidBatch{}: write_batch.validated(handle, mutations, count)
1873	    case BatchPolicy.BatchBytesExceeded{}:
1874	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
1875	        (handle, Fail{ResourceLimit{}}))
1876	    case BatchPolicy.BatchCountExceeded{}:
1877	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
1878	        (handle, Fail{ResourceLimit{}}))
1879	    case _:
1880	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
1881	        (handle, Fail{InvalidArgument{}}))
1882	
1883	def write_batch.validation.counted(
1884	  handle: Handle,
1885	  mutations: List<&1, ByteWal.Mutation>,
1886	  counts: Nat & Nat,
1887	  encoded_bytes: Nat
1888	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1889	  match counts:
1890	    case (count, validation_count):
1891	      write_batch.validation.result(handle, mutations, count,
1892	        BatchPolicy.validate_batch(validation_count, encoded_bytes))
1893	
1894	def write_batch.measured(
1895	  handle: Handle,
1896	  measured: BatchPolicy.ByteBatchMeasurement
1897	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1898	  match measured:
1899	    case BatchPolicy.BatchMeasured{mutations, count, encoded_bytes}:
1900	      write_batch.validation.counted(handle, mutations,
1901	        QueuePolicy.nat.clone.pair(count), encoded_bytes)
1902	    case BatchPolicy.BatchTooMany{}:
1903	      IO.pure(Handle & Result<&1, &1, Error, Unit>,
1904	        (handle, Fail{ResourceLimit{}}))
1905	
1906	# Reconciles each unique batch key against the SST before one atomic WAL append.
1907	def write_batch(
1908	  handle: Handle,
1909	  mutations: List<&1, ByteWal.Mutation>
1910	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1911	  write_batch.measured(handle, BatchPolicy.byte_batch.measure(mutations))
1912	
1913	def put.keys.retained(
1914	  probe_key: Bytes.Bytes,
1915	  pair: Bytes.Bytes & Bytes.Bytes
1916	) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
1917	  match pair:
1918	    case (read_key, write_key): (probe_key, read_key, write_key)
1919	
1920	def put.keys.first(
1921	  pair: Bytes.Bytes & Bytes.Bytes
1922	) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
1923	  match pair:
1924	    case (probe_key, retained):
1925	      put.keys.retained(probe_key, ByteMemTable.bytes.clone(retained))
1926	
1927	def put.keys(key: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
1928	  put.keys.first(ByteMemTable.bytes.clone(key))
1929	
1930	def put.absent(
1931	  +path: String,
1932	  lock: DbLock.Lock,
1933	  table: ByteMemTable.Table,
1934	  state: HandleState,
1935	  base_stats: VisibleStats.Stats,
1936	  write_key: Bytes.Bytes,
1937	  value: Bytes.Bytes
1938	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1939	  write_batch(Handle{path, lock, table, state, base_stats},
1940	    Con{ByteWal.Put{write_key, value}, Nil{}})
1941	
1942	def put.probe.absent.nonempty(
1943	  +path: String,
1944	  lock: DbLock.Lock,
1945	  table: ByteMemTable.Table,
1946	  state: HandleState,
1947	  base_stats: VisibleStats.Stats,
1948	  keys: Bytes.Bytes & Bytes.Bytes,
1949	  value: Bytes.Bytes
1950	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1951	  match keys:
1952	    case (_, write_key):
1953	      put.absent(path, lock, table, state, base_stats,
1954	        write_key, value)
1955	
1956	def put.probe.present(
1957	  +path: String,
1958	  lock: DbLock.Lock,
1959	  table: ByteMemTable.Table,
1960	  state: HandleState,
1961	  base_stats: VisibleStats.Stats,
1962	  keys: Bytes.Bytes & Bytes.Bytes,
1963	  value: Bytes.Bytes
1964	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1965	  match keys:
1966	    case (_, write_key):
1967	      write_batch(Handle{path, lock, table, state, base_stats},
1968	        Con{ByteWal.Put{write_key, value}, Nil{}})
1969	
1970	def put.probe.absent.count(
1971	  +path: String,
1972	  lock: DbLock.Lock,
1973	  table: ByteMemTable.Table,
1974	  state: HandleState,
1975	  base_stats: VisibleStats.Stats,
1976	  keys: Bytes.Bytes & Bytes.Bytes,
1977	  value: Bytes.Bytes,
1978	  empty: Bool
1979	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1980	  match empty:
1981	    case True{}:
1982	      put.probe.present(path, lock, table, state, base_stats, keys, value)
1983	    case False{}:
1984	      put.probe.absent.nonempty(path, lock, table, state, base_stats,
1985	        keys, value)
1986	
1987	def put.probe.absent.cloned(
1988	  +path: String,
1989	  lock: DbLock.Lock,
1990	  table: ByteMemTable.Table,
1991	  state: HandleState,
1992	  copies: VisibleStats.Stats & VisibleStats.Stats,
1993	  keys: Bytes.Bytes & Bytes.Bytes,
1994	  value: Bytes.Bytes
1995	) -> IO(Handle & Result<&1, &1, Error, Unit>):
1996	  match copies:
1997	    case (measured, retained):
1998	      match measured:
1999	        case VisibleStats.Stats{entries, _}:
2000	          put.probe.absent.count(path, lock, table, state, retained,
Lines 1-2000 of 3202. More: https://jend.leeeet.dev/read.txt?path=mylsm-lsm-store@0.6.0.1/src/ByteDurableDb.bend&start_line=2001