src/ByteDurableDb.bend

mylsm-lsm-store@0.6.0.1 · 3,202 lines · 253 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.1/src/ByteManifest.bend as ByteManifest
4import mylsm-lsm-store@0.6.0.1/src/ByteManifestIo.bend as ByteManifestIo
5import mylsm-lsm-store@0.6.0.1/src/ByteSstFile.bend as ByteSstFile
6import mylsm-lsm-store@0.6.0.1/src/ByteSstIo.bend as ByteSstIo
7import mylsm-lsm-store@0.6.0.1/src/ByteWal.bend as ByteWal
8import mylsm-lsm-store@0.6.0.1/src/ByteWalIo.bend as ByteWalIo
9import mylsm-lsm-store@0.6.0.1/src/DbLock.bend as DbLock
10import mylsm-lsm-store@0.6.0.1/src/DurableError.bend as DurableError
11import mylsm-lsm-store@0.6.0.1/src/DurableBatchPolicy.bend as BatchPolicy
12import mylsm-lsm-store@0.6.0.1/src/coordinator/QueuePolicy.bend as QueuePolicy
13import mylsm-lsm-store@0.6.0.1/src/coordinator/SnapshotValueLengthPolicy.bend as SnapshotValueLengthPolicy
14import mylsm-lsm-store@0.6.0.1/src/Fs.bend as Fs
15import mylsm-lsm-store@0.6.0.1/src/FilesCompat.bend as Files
16import mylsm-lsm-store@0.6.0.1/src/Keys.bend as Keys
17import mylsm-lsm-store@0.6.0.1/src/scan/Cursor.bend as Cursor
18import mylsm-lsm-store@0.6.0.1/src/scan/Snapshot.bend as SnapshotPolicy
19import mylsm-lsm-store@0.6.0.1/src/scan/ByteBlockReader.bend as ByteBlockReader
20import mylsm-lsm-store@0.6.0.1/src/scan/SstCursor.bend as SstCursor
21import mylsm-lsm-store@0.6.0.1/src/scan/RangeCursor.bend as RangeCursor
22import mylsm-lsm-store@0.6.0.1/src/scan/VisibleCursor.bend as VisibleCursor
23import mylsm-lsm-store@0.6.0.1/src/scan/VisibleStats.bend as VisibleStats
24import mylsm-lsm-store@0.6.0.1/src/CrashPoint.bend as CrashPoint
25import mylsm-lsm-store@0.6.0.1/src/scan/ScanWorkPolicy.bend as ScanWorkPolicy
26import mylsm-lsm-store@0.6.0.1/src/ByteRecoveryPolicy.bend as RecoveryPolicy
27import mylsm-lsm-store@0.6.0.1/src/ValueLengthPolicy.bend as ValueLengthPolicy
28import mylsm-lsm-store@0.6.0.1/src/Wal.bend as Wal
29import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
30
31# Reports byte-database lifecycle, validation, storage, and uncertain-commit failures.
32type 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.
45type HandleState is Data:
46  Open{}
47  Poisoned{}
48
49# Owns one exclusive database lock and the recovered byte mutation table.
50type 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.
55type 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.
60type SnapshotCursorRoot is Type:
61  SnapshotCursorRoot{retained_bytes: Nat, cursor: VisibleCursor.Cursor}
62
63# Holds an incremental point-read transition over a retained snapshot cursor.
64type 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.
69type Stats is Data:
70  Stats{entries: Nat, payload_bytes: Nat}
71
72# Carries normalized mutations and the visible base totals before WAL append.
73type 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.
77type 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.
96type 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.
107type 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.
114def 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.
134def 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.
140def 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.
153def 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.
166def 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
179def 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.
193def 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
205def 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.
218def 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.
232def 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
237def 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.
242def 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.
259def 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
269def 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
280def 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
297def 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
319def 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
327def 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
337def 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
342def 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.
353def 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.
359def 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
371def 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.
384def 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.
398def 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
410def 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.
416def 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
428def 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
442def 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.
459def 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.
469def 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.
475def 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.
485def 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.
496def 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
508def 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.
517def 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.
535def 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.
549def 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.
563def 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
577def 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
599def 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
614def 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.
634def 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.
648def 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
662def 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.
683def 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.
690def path.clone(path: String) -> String:
691  path.reverse(path.reverse(path, SNil{}), SNil{})
692
693# Copies one natural value into two equal affine outputs.
694def 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.
699def 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.
705def 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.
716def 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.
732def 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.
736def 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
748def 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.
755def 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.
763def 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.
769def path.clone.triple(path: String) -> String & String & String:
770  path.clone.triple.first(path.clone.pair(path))
771
772
773def 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
793def 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
812def 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
822def 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
842def 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.
863def 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.
877def 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.
888def 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.
898def 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
915def 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.
946def 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.
957def 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
965def 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
976def 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
993def 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
1024def 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.
1037def 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
1045def 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
1060def 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
1071def 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
1079def 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
1090def 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
1100def 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
1124def 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
1169def 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
1184def 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
1207def 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
1222def 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
1238def 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
1249def 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
1262def 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
1284def 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
1298def 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
1308def 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
1313def 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
1322def 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
1332def 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
1348def 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
1359def 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
1382def 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.
1396def 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
1404def 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
1424def 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
1437def 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
1467def 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
1476def 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
1497def 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
1508def 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
1518def 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.
1537def 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.
1554def 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
1569def 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
1580def 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
1587def 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
1592def batch.keys(key: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
1593  batch.keys.first(ByteMemTable.bytes.clone(key))
1594
1595def 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
1603def 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
1639def 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
1656def 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.
1714def 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
1731def 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
1743def 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
1762def 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
1775def 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
1783def 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
1792def 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
1801def 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
1813def 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
1822def 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
1834def 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
1839def 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
1850def 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
1858def 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
1865def 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
1883def 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
1894def 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.
1907def 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
1913def 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
1920def 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
1927def put.keys(key: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
1928  put.keys.first(ByteMemTable.bytes.clone(key))
1929
1930def 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
1942def 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
1956def 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
1970def 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
1987def 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,
2001            keys, value, Nat.is_eq(entries, 0n))
2002
2003def put.probe.absent(
2004  +path: String,
2005  lock: DbLock.Lock,
2006  table: ByteMemTable.Table,
2007  state: HandleState,
2008  base_stats: VisibleStats.Stats,
2009  keys: Bytes.Bytes & Bytes.Bytes,
2010  value: Bytes.Bytes
2011) -> IO(Handle & Result<&1, &1, Error, Unit>):
2012  put.probe.absent.cloned(path, lock, table, state,
2013    VisibleStats.clone(base_stats), keys, value)
2014
2015def put.probe(
2016  +path: String,
2017  lock: DbLock.Lock,
2018  table: ByteMemTable.Table,
2019  state: HandleState,
2020  base_stats: VisibleStats.Stats,
2021  probe: ByteMemTable.Probe,
2022  keys: Bytes.Bytes & Bytes.Bytes,
2023  value: Bytes.Bytes
2024) -> IO(Handle & Result<&1, &1, Error, Unit>):
2025  match probe:
2026    case ByteMemTable.Absent{}:
2027      put.probe.absent(path, lock, table, state, base_stats, keys, value)
2028    case _:
2029      put.probe.present(path, lock, table, state, base_stats, keys, value)
2030
2031def put.probed(
2032  +path: String,
2033  lock: DbLock.Lock,
2034  state: HandleState,
2035  base_stats: VisibleStats.Stats,
2036  result: ByteMemTable.Table & ByteMemTable.Probe,
2037  keys: Bytes.Bytes & Bytes.Bytes,
2038  value: Bytes.Bytes
2039) -> IO(Handle & Result<&1, &1, Error, Unit>):
2040  match result:
2041    case (table, probe):
2042      put.probe(path, lock, table, state, base_stats, probe, keys, value)
2043
2044def put.keyed(
2045  +path: String,
2046  lock: DbLock.Lock,
2047  table: ByteMemTable.Table,
2048  state: HandleState,
2049  base_stats: VisibleStats.Stats,
2050  keys: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes,
2051  value: Bytes.Bytes
2052) -> IO(Handle & Result<&1, &1, Error, Unit>):
2053  match keys:
2054    case (probe_key, (read_key, write_key)):
2055      put.probed(path, lock, state, base_stats,
2056        ByteMemTable.probe(table, probe_key),
2057        (read_key, write_key), value)
2058
2059# Writes a raw byte key and value after subtracting the SST value it shadows.
2060def put(handle: Handle, key: Bytes.Bytes, value: Bytes.Bytes) -> IO(Handle & Result<&1, &1, Error, Unit>):
2061  match handle:
2062    case Handle{path, lock, table, state, base_stats}:
2063      put.keyed(path, lock, table, state, base_stats, put.keys(key), value)
2064
2065def delete.absent(
2066  +path: String,
2067  lock: DbLock.Lock,
2068  table: ByteMemTable.Table,
2069  state: HandleState,
2070  base_stats: VisibleStats.Stats,
2071  delete_key: Bytes.Bytes
2072) -> IO(Handle & Result<&1, &1, Error, Unit>):
2073  write_batch(Handle{path, lock, table, state, base_stats},
2074    Con{ByteWal.Delete{delete_key}, Nil{}})
2075
2076def delete.probe.absent.nonempty(
2077  +path: String,
2078  lock: DbLock.Lock,
2079  table: ByteMemTable.Table,
2080  state: HandleState,
2081  base_stats: VisibleStats.Stats,
2082  keys: Bytes.Bytes & Bytes.Bytes
2083) -> IO(Handle & Result<&1, &1, Error, Unit>):
2084  match keys:
2085    case (_, delete_key):
2086      delete.absent(path, lock, table, state, base_stats, delete_key)
2087
2088def delete.probe.present(
2089  +path: String,
2090  lock: DbLock.Lock,
2091  table: ByteMemTable.Table,
2092  state: HandleState,
2093  base_stats: VisibleStats.Stats,
2094  keys: Bytes.Bytes & Bytes.Bytes
2095) -> IO(Handle & Result<&1, &1, Error, Unit>):
2096  match keys:
2097    case (_, delete_key):
2098      write_batch(Handle{path, lock, table, state, base_stats},
2099        Con{ByteWal.Delete{delete_key}, Nil{}})
2100
2101def delete.probe.absent.count(
2102  +path: String,
2103  lock: DbLock.Lock,
2104  table: ByteMemTable.Table,
2105  state: HandleState,
2106  base_stats: VisibleStats.Stats,
2107  keys: Bytes.Bytes & Bytes.Bytes,
2108  empty: Bool
2109) -> IO(Handle & Result<&1, &1, Error, Unit>):
2110  match empty:
2111    case True{}:
2112      delete.probe.present(path, lock, table, state, base_stats, keys)
2113    case False{}:
2114      delete.probe.absent.nonempty(path, lock, table, state, base_stats, keys)
2115
2116def delete.probe.absent.cloned(
2117  +path: String,
2118  lock: DbLock.Lock,
2119  table: ByteMemTable.Table,
2120  state: HandleState,
2121  copies: VisibleStats.Stats & VisibleStats.Stats,
2122  keys: Bytes.Bytes & Bytes.Bytes
2123) -> IO(Handle & Result<&1, &1, Error, Unit>):
2124  match copies:
2125    case (measured, retained):
2126      match measured:
2127        case VisibleStats.Stats{entries, _}:
2128          delete.probe.absent.count(path, lock, table, state, retained,
2129            keys, Nat.is_eq(entries, 0n))
2130
2131def delete.probe.absent(
2132  +path: String,
2133  lock: DbLock.Lock,
2134  table: ByteMemTable.Table,
2135  state: HandleState,
2136  base_stats: VisibleStats.Stats,
2137  keys: Bytes.Bytes & Bytes.Bytes
2138) -> IO(Handle & Result<&1, &1, Error, Unit>):
2139  delete.probe.absent.cloned(path, lock, table, state,
2140    VisibleStats.clone(base_stats), keys)
2141
2142def delete.probe(
2143  +path: String,
2144  lock: DbLock.Lock,
2145  table: ByteMemTable.Table,
2146  state: HandleState,
2147  base_stats: VisibleStats.Stats,
2148  probe: ByteMemTable.Probe,
2149  keys: Bytes.Bytes & Bytes.Bytes
2150) -> IO(Handle & Result<&1, &1, Error, Unit>):
2151  match probe:
2152    case ByteMemTable.Absent{}:
2153      delete.probe.absent(path, lock, table, state, base_stats, keys)
2154    case _:
2155      delete.probe.present(path, lock, table, state, base_stats, keys)
2156
2157def delete.probed(
2158  +path: String,
2159  lock: DbLock.Lock,
2160  state: HandleState,
2161  base_stats: VisibleStats.Stats,
2162  result: ByteMemTable.Table & ByteMemTable.Probe,
2163  keys: Bytes.Bytes & Bytes.Bytes
2164) -> IO(Handle & Result<&1, &1, Error, Unit>):
2165  match result:
2166    case (table, probe):
2167      delete.probe(path, lock, table, state, base_stats, probe, keys)
2168
2169# Deletes a raw byte key after removing the SST value it shadows.
2170def delete.keyed(
2171  +path: String,
2172  lock: DbLock.Lock,
2173  table: ByteMemTable.Table,
2174  state: HandleState,
2175  base_stats: VisibleStats.Stats,
2176  keys: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes
2177) -> IO(Handle & Result<&1, &1, Error, Unit>):
2178  match keys:
2179    case (probe_key, (read_key, delete_key)):
2180      delete.probed(path, lock, state, base_stats,
2181        ByteMemTable.probe(table, probe_key),
2182        (read_key, delete_key))
2183
2184# Deletes a raw byte key after removing the SST value it shadows.
2185def delete(handle: Handle, key: Bytes.Bytes) -> IO(Handle & Result<&1, &1, Error, Unit>):
2186  match handle:
2187    case Handle{path, lock, table, state, base_stats}:
2188      delete.keyed(path, lock, table, state, base_stats, put.keys(key))
2189
2190def scan.sst.bounds(bounds: RangeCursor.Bounds) -> SstCursor.Bounds:
2191  match bounds:
2192    case RangeCursor.Bounds{lower, upper, prefix}:
2193      SstCursor.Bounds{lower, upper, prefix}
2194
2195def scan.sources.opened(
2196  handle: Handle,
2197  delta: List<&1, ByteMemTable.Entry>,
2198  bounds: RangeCursor.Bounds,
2199  opened: Result<&1, &1, U32 & String, SstCursor.Cursor>
2200) -> IO(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>):
2201  match opened:
2202    case Fail{(code, message)}:
2203      IO.pure(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>,
2204        (handle, Fail{recovery.error(code, message)}))
2205    case Done{base}:
2206      IO.pure(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>,
2207        (handle, Done{VisibleCursor.open(Some{base}, delta, bounds)}))
2208
2209def scan.manifest.result(
2210  +path: String,
2211  handle: Handle,
2212  delta: List<&1, ByteMemTable.Entry>,
2213  bounds: RangeCursor.Bounds,
2214  sst_bounds: SstCursor.Bounds,
2215  result: Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue>
2216) -> IO(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>):
2217  match result:
2218    case Fail{ByteManifestIo.Invalid{_}}:
2219      IO.pure(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>,
2220        (handle, Fail{Corruption{"byte manifest validation failed"}}))
2221    case Fail{ByteManifestIo.Host{code, message}}:
2222      IO.pure(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>,
2223        (handle, Fail{Io{code, message}}))
2224    case Done{ByteManifest.Empty{}}:
2225      IO.pure(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>,
2226        (handle, Done{VisibleCursor.open(None{}, delta, bounds)}))
2227    case Done{ByteManifest.GenerationEmpty{}}:
2228      IO.pure(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>,
2229        (handle, Done{VisibleCursor.open(None{}, delta, bounds)}))
2230    case Done{ByteManifest.HasSst{}}:
2231      do IO<Handle & Result<&1, &1, Error, VisibleCursor.Cursor>>:
2232        opened : Result<&1, &1, U32 & String, SstCursor.Cursor> <-
2233          SstCursor.open.trusted(path ++ "/SST", sst_bounds)
2234        scan.sources.opened(handle, delta, bounds, opened)
2235    case Done{ByteManifest.GenerationSst{file_id}}:
2236      do IO<Handle & Result<&1, &1, Error, VisibleCursor.Cursor>>:
2237        opened : Result<&1, &1, U32 & String, SstCursor.Cursor> <-
2238          SstCursor.open.trusted(
2239            ByteManifest.sst.generation.path(path, file_id), sst_bounds)
2240        scan.sources.opened(handle, delta, bounds, opened)
2241
2242def scan.path.kept.triple(
2243  copies: String & String & String,
2244  lock: DbLock.Lock,
2245  table: ByteMemTable.Table,
2246  state: HandleState,
2247  base_stats: VisibleStats.Stats
2248) -> String & String & Handle:
2249  match copies:
2250    case (exists_path, (sst_path, retained)):
2251      (exists_path, sst_path, Handle{retained, lock, table, state, base_stats})
2252
2253def scan.path.kept(handle: Handle) -> String & String & Handle:
2254  match handle:
2255    case Handle{path, lock, table, state, base_stats}:
2256      scan.path.kept.triple(path.clone.triple(path), lock, table, state,
2257        base_stats)
2258
2259def scan.exists.start(
2260  paths: String & String & Handle,
2261  delta: List<&1, ByteMemTable.Entry>,
2262  bounds: RangeCursor.Bounds,
2263  sst_bounds: SstCursor.Bounds
2264) -> IO(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>):
2265  match paths:
2266    case (exists_path, (sst_path, handle)):
2267      do IO<Handle & Result<&1, &1, Error, VisibleCursor.Cursor>>:
2268        manifest : Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue> <-
2269          ByteManifestIo.read(exists_path ++ "/MANIFEST")
2270        scan.manifest.result(sst_path, handle, delta, bounds, sst_bounds, manifest)
2271
2272def scan.bounds.cloned(
2273  handle: Handle,
2274  delta: List<&1, ByteMemTable.Entry>,
2275  copies: RangeCursor.Bounds & RangeCursor.Bounds
2276) -> IO(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>):
2277  match copies:
2278    case (retained, for_sst):
2279      scan.exists.start(scan.path.kept(handle), delta, retained,
2280        scan.sst.bounds(for_sst))
2281
2282def scan.delta.copied(
2283  +path: String,
2284  lock: DbLock.Lock,
2285  state: HandleState,
2286  base_stats: VisibleStats.Stats,
2287  entries: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>,
2288  bounds: RangeCursor.Bounds
2289) -> IO(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>):
2290  match entries:
2291    case (retained, copied):
2292      scan.bounds.cloned(
2293        Handle{path, lock, ByteMemTable.Table{retained}, state, base_stats},
2294        ByteMemTable.ordered_entries(ByteMemTable.Table{copied}),
2295        RangeCursor.bounds.clone(bounds))
2296
2297def scan.open(handle: Handle, bounds: RangeCursor.Bounds) -> IO(Handle & Result<&1, &1, Error, VisibleCursor.Cursor>):
2298  match handle:
2299    case Handle{path, lock, ByteMemTable.Table{entries}, state, base_stats}:
2300      scan.delta.copied(path, lock, state, base_stats,
2301        ByteMemTable.entries.clone(entries), bounds)
2302
2303def scan.work_budget.copied(
2304  +path: String,
2305  lock: DbLock.Lock,
2306  state: HandleState,
2307  base_entries: Nat & Nat,
2308  payload_bytes: Nat,
2309  copies: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
2310) -> Handle & Nat:
2311  match base_entries:
2312    case (retained_base_entries, budget_base_entries):
2313      match copies:
2314        case (entries, copied):
2315          +delta_count = List.length(&1, ByteMemTable.Entry, copied)
2316          (Handle{path, lock, ByteMemTable.Table{entries}, state,
2317              VisibleStats.Stats{retained_base_entries, payload_bytes}},
2318            ScanWorkPolicy.writer_transitions(budget_base_entries, delta_count))
2319
2320def scan.work_budget(handle: Handle) -> Handle & Nat:
2321  match handle:
2322    case Handle{path, lock, ByteMemTable.Table{entries}, state,
2323        VisibleStats.Stats{base_entries, payload_bytes}}:
2324      scan.work_budget.copied(path, lock, state,
2325        ScanWorkPolicy.nat.clone.pair(base_entries), payload_bytes,
2326        ByteMemTable.entries.clone(entries))
2327
2328def scan.error(error: U32 & String) -> Error:
2329  match error:
2330    case (code, message): recovery.error(code, message)
2331
2332# Clones canonical visible entries so a cursor can advance independently of writes.
2333def snapshot.entry.value(
2334  key: Bytes.Bytes,
2335  pair: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>
2336) -> ByteMemTable.Entry:
2337  match pair:
2338    case (_, copied_value):
2339      ByteMemTable.Entry{key, copied_value}
2340
2341def snapshot.entry.key(
2342  pair: Bytes.Bytes & Bytes.Bytes,
2343  value: Maybe<&1, Bytes.Bytes>
2344) -> ByteMemTable.Entry:
2345  match pair:
2346    case (_, copied_key):
2347      snapshot.entry.value(copied_key, ByteMemTable.value.clone(value))
2348
2349def snapshot.entries(entries: List<&1, ByteMemTable.Entry>) -> List<&1, ByteMemTable.Entry>:
2350  match entries:
2351    case Nil{}:
2352      Nil{}
2353    case Con{ByteMemTable.Entry{key, value}, tail}:
2354      Con{snapshot.entry.key(ByteMemTable.bytes.clone(key), value), snapshot.entries(tail)}
2355
2356# Returns a stable ordered copy while preserving the live database handle.
2357def snapshot.table(
2358  +path: String,
2359  lock: DbLock.Lock,
2360  state: HandleState,
2361  base_stats: VisibleStats.Stats,
2362  cloned: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
2363) -> Handle & List<&1, ByteMemTable.Entry>:
2364  match cloned:
2365    case (original, copied):
2366      (Handle{path, lock, ByteMemTable.Table{original}, state, base_stats},
2367        ByteMemTable.ordered_entries(ByteMemTable.Table{copied}))
2368
2369# Clones the ordered visible table while preserving the owned handle.
2370def snapshot(handle: Handle) -> Handle & List<&1, ByteMemTable.Entry>:
2371  match handle:
2372    case Handle{path, lock, ByteMemTable.Table{entries}, state, base_stats}:
2373      snapshot.table(path, lock, state, base_stats,
2374        ByteSstFile.entries.clone(entries))
2375
2376def snapshot.limited.saved(
2377  pair: Handle & List<&1, ByteMemTable.Entry>
2378) -> Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>:
2379  match pair:
2380    case (handle, entries): (handle, Done{entries})
2381
2382def snapshot.estimate(base_stats: VisibleStats.Stats, delta_bytes: Nat) -> Nat:
2383  match base_stats:
2384    case VisibleStats.Stats{entries, payload_bytes}:
2385      Nat.add(delta_bytes, Nat.add(payload_bytes, Nat.mul(entries, 32n)))
2386
2387def snapshot.limited.saved.result(
2388  handle: Handle,
2389  result: Result<&1, &1, U32 & String, ByteMemTable.Table>
2390) -> IO(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>):
2391  match handle:
2392    case Handle{path, lock, ByteMemTable.Table{delta}, state, base_stats}:
2393      match result:
2394        case Fail{(code, message)}:
2395          IO.pure(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>,
2396            (Handle{path, lock, ByteMemTable.Table{delta}, state, base_stats},
2397              Fail{recovery.error(code, message)}))
2398        case Done{ByteMemTable.Table{base}}:
2399          IO.pure(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>,
2400            snapshot.limited.saved(snapshot.table(path, lock, state, base_stats,
2401              ByteSstFile.entries.clone(ByteMemTable.compact.live(
2402                ByteMemTable.append(delta, base))))))
2403
2404def snapshot.manifest.result.paths(
2405  paths: String & String,
2406  lock: DbLock.Lock,
2407  table: ByteMemTable.Table,
2408  state: HandleState,
2409  base_stats: VisibleStats.Stats,
2410  result: Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue>
2411) -> IO(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>):
2412  match paths:
2413    case (read_path, retained_path):
2414      match result:
2415        case Fail{ByteManifestIo.Invalid{_}}:
2416          IO.pure(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>,
2417            (Handle{retained_path, lock, table, state, base_stats},
2418              Fail{Corruption{"byte manifest validation failed"}}))
2419        case Fail{ByteManifestIo.Host{code, message}}:
2420          IO.pure(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>,
2421            (Handle{retained_path, lock, table, state, base_stats}, Fail{Io{code, message}}))
2422        case Done{ByteManifest.Empty{}}:
2423          snapshot.limited.saved.result(
2424            Handle{retained_path, lock, table, state, base_stats}, Done{ByteMemTable.empty()})
2425        case Done{ByteManifest.GenerationEmpty{}}:
2426          snapshot.limited.saved.result(
2427            Handle{retained_path, lock, table, state, base_stats}, Done{ByteMemTable.empty()})
2428        case Done{ByteManifest.HasSst{}}:
2429          do IO<Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>>:
2430            base : Result<&1, &1, U32 & String, ByteMemTable.Table> <-
2431              ByteSstIo.read(read_path ++ "/SST")
2432            snapshot.limited.saved.result(
2433              Handle{retained_path, lock, table, state, base_stats}, base)
2434        case Done{ByteManifest.GenerationSst{file_id}}:
2435          do IO<Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>>:
2436            base : Result<&1, &1, U32 & String, ByteMemTable.Table> <-
2437              ByteSstIo.read(ByteManifest.sst.generation.path(read_path, file_id))
2438            snapshot.limited.saved.result(
2439              Handle{retained_path, lock, table, state, base_stats}, base)
2440
2441def snapshot.limited.admitted.paths(
2442  paths: String & String & String,
2443  lock: DbLock.Lock,
2444  table: ByteMemTable.Table,
2445  state: HandleState,
2446  base_stats: VisibleStats.Stats,
2447  decision: SnapshotPolicy.Decision
2448) -> IO(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>):
2449  match paths:
2450    case (manifest_path, (read_path, retained_path)):
2451      match decision:
2452        case SnapshotPolicy.ResourceLimit{}:
2453          IO.pure(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>,
2454            (Handle{retained_path, lock, table, state, base_stats}, Fail{ResourceLimit{}}))
2455        case SnapshotPolicy.Retained{_, _}:
2456          do IO<Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>>:
2457            manifest : Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue> <-
2458              ByteManifestIo.read(manifest_path ++ "/MANIFEST")
2459            snapshot.manifest.result.paths((read_path, retained_path), lock,
2460              table, state, base_stats, manifest)
2461
2462def snapshot.limited.admitted(
2463  handle: Handle,
2464  decision: SnapshotPolicy.Decision
2465) -> IO(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>):
2466  match handle:
2467    case Handle{path, lock, table, state, base_stats}:
2468      snapshot.limited.admitted.paths(path.clone.triple(path), lock, table,
2469        state, base_stats, decision)
2470
2471# Preflights retained bytes before reading the published SST snapshot.
2472def snapshot.limited.retained.stats(
2473  +path: String,
2474  lock: DbLock.Lock,
2475  table: ByteMemTable.Table,
2476  state: HandleState,
2477  stats: VisibleStats.Stats & VisibleStats.Stats,
2478  delta_bytes: Nat,
2479  max_bytes: Nat
2480) -> IO(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>):
2481  match stats:
2482    case (for_estimate, for_handle):
2483          snapshot.limited.admitted(
2484            Handle{path, lock, table, state, for_handle},
2485            SnapshotPolicy.reserve(0n, 0n,
2486              snapshot.estimate(for_estimate, delta_bytes), 1n, max_bytes))
2487
2488def snapshot.limited.retained(
2489  handle: Handle,
2490  retained: ByteMemTable.Table & Nat,
2491  max_bytes: Nat
2492) -> IO(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>):
2493  match handle:
2494    case Handle{+path, lock, _, state, +base_stats}:
2495      match retained:
2496        case (table, delta_bytes):
2497          snapshot.limited.retained.stats(path, lock, table, state,
2498            VisibleStats.clone(base_stats), delta_bytes, max_bytes)
2499
2500# Preflights retained bytes before reading the published SST snapshot.
2501def snapshot_limited(
2502  max_bytes: Nat,
2503  handle: Handle
2504) -> IO(Handle & Result<&1, &1, Error, List<&1, ByteMemTable.Entry>>):
2505  match handle:
2506    case Handle{path, lock, table, state, base_stats}:
2507      snapshot.limited.retained(
2508        Handle{path, lock, ByteMemTable.empty(), state, base_stats},
2509        ByteMemTable.retained_bytes.keep(table), max_bytes)
2510
2511def snapshot.root.reservation.result(
2512  handle: Handle,
2513  path: Maybe<&1, String>,
2514  generation: Maybe<&1, U32>,
2515  delta: List<&1, ByteMemTable.Entry>,
2516  decision: SnapshotPolicy.Decision
2517) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2518  match decision:
2519    case SnapshotPolicy.ResourceLimit{}:
2520      IO.pure(Handle & Result<&1, &1, Error, SnapshotRoot>,
2521        (handle, Fail{ResourceLimit{}}))
2522    case SnapshotPolicy.Retained{_, retained_bytes}:
2523      IO.pure(Handle & Result<&1, &1, Error, SnapshotRoot>,
2524        (handle, Done{SnapshotRoot{path, generation, retained_bytes, delta}}))
2525
2526def snapshot.root.reservation(
2527  handle: Handle,
2528  path: Maybe<&1, String>,
2529  generation: Maybe<&1, U32>,
2530  sst_bytes: Nat,
2531  delta_bytes: Nat,
2532  delta: List<&1, ByteMemTable.Entry>,
2533  max_bytes: Nat
2534) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2535  snapshot.root.reservation.result(handle, path, generation, delta,
2536    SnapshotPolicy.reserve(0n, 0n,
2537      Nat.add(sst_bytes, delta_bytes), 1n, max_bytes))
2538
2539def snapshot.root.file.size.result(
2540  handle: Handle,
2541  path: String,
2542  generation: Maybe<&1, U32>,
2543  delta: List<&1, ByteMemTable.Entry>,
2544  delta_bytes: Nat,
2545  max_bytes: Nat,
2546  result: Result<&1, &1, U32 & String, Nat>
2547) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2548  match result:
2549    case Fail{(code, message)}:
2550      IO.pure(Handle & Result<&1, &1, Error, SnapshotRoot>,
2551        (handle, Fail{Io{code, message}}))
2552    case Done{size}:
2553      snapshot.root.reservation(handle, Some{path}, generation, size,
2554        delta_bytes, delta, max_bytes)
2555
2556def snapshot.root.file.size(
2557  handle: Handle,
2558  paths: String & String,
2559  generation: Maybe<&1, U32>,
2560  delta: List<&1, ByteMemTable.Entry>,
2561  delta_bytes: Nat,
2562  max_bytes: Nat
2563) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2564  match paths:
2565    case (file_path, retained_path):
2566      do IO<Handle & Result<&1, &1, Error, SnapshotRoot>>:
2567        size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(file_path)
2568        snapshot.root.file.size.result(handle, retained_path, generation, delta,
2569          delta_bytes, max_bytes, size)
2570
2571def snapshot.root.generation.file(
2572  handle: Handle,
2573  read_path: String,
2574  delta: List<&1, ByteMemTable.Entry>,
2575  delta_bytes: Nat,
2576  max_bytes: Nat,
2577  file_ids: U32 & U32
2578) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2579  match file_ids:
2580    case (path_id, generation_id):
2581      snapshot.root.file.size(handle,
2582        path.clone.pair(ByteManifest.sst.generation.path(read_path, path_id)),
2583        Some{generation_id}, delta, delta_bytes, max_bytes)
2584
2585def snapshot.root.manifest.result(
2586  read_path: String,
2587  handle: Handle,
2588  delta: List<&1, ByteMemTable.Entry>,
2589  delta_bytes: Nat,
2590  max_bytes: Nat,
2591  result: Result<&1, &1, ByteManifestIo.Error, ByteManifest.ManifestValue>
2592) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2593  match result:
2594    case Fail{ByteManifestIo.Invalid{_}}:
2595      IO.pure(Handle & Result<&1, &1, Error, SnapshotRoot>,
2596        (handle, Fail{Corruption{"byte manifest validation failed"}}))
2597    case Fail{ByteManifestIo.Host{code, message}}:
2598      IO.pure(Handle & Result<&1, &1, Error, SnapshotRoot>,
2599        (handle, Fail{Io{code, message}}))
2600    case Done{ByteManifest.Empty{}}:
2601      snapshot.root.reservation(handle, None{}, None{}, 0n, delta_bytes, delta,
2602        max_bytes)
2603    case Done{ByteManifest.GenerationEmpty{}}:
2604      snapshot.root.reservation(handle, None{}, None{}, 0n, delta_bytes, delta,
2605        max_bytes)
2606    case Done{ByteManifest.HasSst{}}:
2607      snapshot.root.file.size(handle,
2608        path.clone.pair(read_path ++ "/SST"), None{}, delta, delta_bytes,
2609        max_bytes)
2610    case Done{ByteManifest.GenerationSst{file_id}}:
2611      snapshot.root.generation.file(handle, read_path, delta, delta_bytes,
2612        max_bytes, ByteManifest.clone.file.id(file_id))
2613
2614def snapshot.root.manifest(
2615  paths: String & String,
2616  handle: Handle,
2617  delta: List<&1, ByteMemTable.Entry>,
2618  delta_bytes: Nat,
2619  max_bytes: Nat
2620) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2621  match paths:
2622    case (manifest_path, read_path):
2623      do IO<Handle & Result<&1, &1, Error, SnapshotRoot>>:
2624        manifest : Result<&1, &1, ByteManifestIo.Error,
2625          ByteManifest.ManifestValue> <-
2626            ByteManifestIo.read(manifest_path ++ "/MANIFEST")
2627        snapshot.root.manifest.result(read_path, handle, delta, delta_bytes,
2628          max_bytes, manifest)
2629
2630def snapshot.root.snapshot.paths(
2631  paths: String & String & String,
2632  lock: DbLock.Lock,
2633  table: ByteMemTable.Table,
2634  state: HandleState,
2635  base_stats: VisibleStats.Stats,
2636  delta: List<&1, ByteMemTable.Entry>,
2637  delta_bytes: Nat,
2638  max_bytes: Nat
2639) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2640  match paths:
2641    case (manifest_path, (read_path, retained_path)):
2642      snapshot.root.manifest((manifest_path, read_path),
2643        Handle{retained_path, lock, table, state, base_stats}, delta,
2644        delta_bytes, max_bytes)
2645
2646def snapshot.root.snapshot.delta(
2647  handle: Handle,
2648  copied: List<&1, ByteMemTable.Entry>,
2649  delta_bytes: Nat,
2650  max_bytes: Nat
2651) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2652  match handle:
2653    case Handle{path, lock, table, state, base_stats}:
2654      snapshot.root.snapshot.paths(path.clone.triple(path), lock, table,
2655        state, base_stats,
2656        ByteMemTable.ordered_entries(ByteMemTable.Table{copied}), delta_bytes,
2657        max_bytes)
2658
2659def snapshot.root.preflight.estimated(
2660  handle: Handle,
2661  copied: List<&1, ByteMemTable.Entry>,
2662  delta_bytes: Nat,
2663  max_bytes: Nat
2664) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2665  snapshot.root.snapshot.delta(handle, copied, delta_bytes, max_bytes)
2666
2667def snapshot.root.preflight.ready(
2668  +path: String,
2669  lock: DbLock.Lock,
2670  state: HandleState,
2671  stats: VisibleStats.Stats,
2672  entries: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>,
2673  delta_bytes: Nat,
2674  max_bytes: Nat
2675) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2676  match entries:
2677    case (live_entries, copied_entries):
2678      snapshot.root.preflight.estimated(
2679        Handle{path, lock, ByteMemTable.Table{live_entries}, state, stats},
2680        copied_entries, delta_bytes, max_bytes)
2681
2682def snapshot.root.preflight(
2683  handle: Handle,
2684  delta_bytes: Nat,
2685  max_bytes: Nat
2686) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2687  match handle:
2688    case Handle{path, lock, table, state, base_stats}:
2689      match table:
2690        case ByteMemTable.Table{entries}:
2691          snapshot.root.preflight.ready(path, lock, state,
2692            base_stats, ByteMemTable.entries.clone(entries),
2693            delta_bytes, max_bytes)
2694
2695def snapshot.root.retained(
2696  +path: String,
2697  lock: DbLock.Lock,
2698  state: HandleState,
2699  base_stats: VisibleStats.Stats,
2700  retained: ByteMemTable.Table & Nat,
2701  max_bytes: Nat
2702) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2703  match retained:
2704    case (table, delta_bytes):
2705      snapshot.root.preflight(Handle{path, lock, table, state, base_stats},
2706        delta_bytes, max_bytes)
2707
2708# Captures the active delta and immutable SST path within the retained-byte limit.
2709def snapshot_root_limited(
2710  max_bytes: Nat,
2711  handle: Handle
2712) -> IO(Handle & Result<&1, &1, Error, SnapshotRoot>):
2713  match handle:
2714    case Handle{path, lock, table, state, base_stats}:
2715      snapshot.root.retained(path, lock, state, base_stats,
2716        ByteMemTable.retained_bytes.keep(table), max_bytes)
2717
2718def snapshot.root.path.clone.copies(
2719  copies: String & String
2720) -> Maybe<&1, String> & Maybe<&1, String>:
2721  match copies:
2722    case (first, second): (Some{first}, Some{second})
2723
2724def snapshot.root.path.clone(
2725  path: Maybe<&1, String>
2726) -> Maybe<&1, String> & Maybe<&1, String>:
2727  match path:
2728    case None{}: (None{}, None{})
2729    case Some{value}: snapshot.root.path.clone.copies(path.clone.pair(value))
2730
2731def snapshot.root.generation.clone.copies(
2732  values: Nat & Nat
2733) -> Maybe<&1, U32> & Maybe<&1, U32>:
2734  match values:
2735    case (first, second):
2736      (Some{U32.from_nat(first)}, Some{U32.from_nat(second)})
2737
2738def snapshot.root.generation.clone(
2739  generation: Maybe<&1, U32>
2740) -> Maybe<&1, U32> & Maybe<&1, U32>:
2741  match generation:
2742    case None{}: (None{}, None{})
2743    case Some{value}:
2744      snapshot.root.generation.clone.copies(duplicate_nat_pair(
2745        U32.to_nat(value)))
2746
2747def snapshot.root.clone.fields(
2748  paths: Maybe<&1, String> & Maybe<&1, String>,
2749  generations: Maybe<&1, U32> & Maybe<&1, U32>,
2750  retained_bytes: Nat & Nat,
2751  deltas: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
2752) -> SnapshotRoot & SnapshotRoot:
2753  match paths:
2754    case (first_path, second_path):
2755      match generations:
2756        case (first_generation, second_generation):
2757          match retained_bytes:
2758            case (first_bytes, second_bytes):
2759              match deltas:
2760                case (first_delta, second_delta):
2761                  (SnapshotRoot{first_path, first_generation, first_bytes,
2762                    first_delta},
2763                    SnapshotRoot{second_path, second_generation, second_bytes,
2764                      second_delta})
2765
2766def snapshot.root.clone(root: SnapshotRoot) -> SnapshotRoot & SnapshotRoot:
2767  match root:
2768    case SnapshotRoot{sst_path, sst_generation, retained_bytes, delta}:
2769      snapshot.root.clone.fields(snapshot.root.path.clone(sst_path),
2770        snapshot.root.generation.clone(sst_generation),
2771        duplicate_nat_pair(retained_bytes),
2772        ByteMemTable.entries.clone(delta))
2773
2774def snapshot.root.sst.bounds(bounds: RangeCursor.Bounds) -> SstCursor.Bounds:
2775  match bounds:
2776    case RangeCursor.Bounds{lower, upper, prefix}:
2777      SstCursor.Bounds{lower, upper, prefix}
2778
2779def snapshot.root.cursor.opened(
2780  retained_bytes: Nat,
2781  delta: List<&1, ByteMemTable.Entry>,
2782  bounds: RangeCursor.Bounds,
2783  opened: Result<&1, &1, U32 & String, SstCursor.Cursor>
2784) -> IO(Result<&1, &1, Error, SnapshotCursorRoot>):
2785  match opened:
2786    case Fail{(code, message)}:
2787      IO.pure(Result<&1, &1, Error, SnapshotCursorRoot>,
2788        Fail{recovery.error(code, message)})
2789    case Done{base}:
2790      IO.pure(Result<&1, &1, Error, SnapshotCursorRoot>,
2791        Done{SnapshotCursorRoot{retained_bytes,
2792          VisibleCursor.open(Some{base}, delta, bounds)}})
2793
2794def snapshot.root.cursor.open.copied(
2795  bounds: RangeCursor.Bounds & RangeCursor.Bounds,
2796  root: SnapshotRoot
2797) -> IO(Result<&1, &1, Error, SnapshotCursorRoot>):
2798  match bounds:
2799    case (visible_bounds, sst_bounds):
2800      match root:
2801        case SnapshotRoot{sst_path, _, retained_bytes, delta}:
2802          match sst_path:
2803            case None{}:
2804              IO.pure(Result<&1, &1, Error, SnapshotCursorRoot>,
2805                Done{SnapshotCursorRoot{retained_bytes,
2806                  VisibleCursor.open(None{}, delta, visible_bounds)}})
2807            case Some{path}:
2808              do IO<Result<&1, &1, Error, SnapshotCursorRoot>>:
2809                opened : Result<&1, &1, U32 & String, SstCursor.Cursor> <-
2810                  SstCursor.open.trusted(path,
2811                    snapshot.root.sst.bounds(sst_bounds))
2812                snapshot.root.cursor.opened(retained_bytes, delta,
2813                  visible_bounds, opened)
2814
2815def snapshot.root.cursor.open.state(
2816  root: SnapshotRoot,
2817  bounds: RangeCursor.Bounds
2818) -> IO(Result<&1, &1, Error, SnapshotCursorRoot>):
2819  snapshot.root.cursor.open.copied(RangeCursor.bounds.clone(bounds), root)
2820
2821def snapshot.root.cursor.open.result(
2822  result: Result<&1, &1, Error, SnapshotCursorRoot>
2823) -> Result<&1, &1, Error, VisibleCursor.Cursor>:
2824  match result:
2825    case Fail{error}: Fail{error}
2826    case Done{SnapshotCursorRoot{_, cursor}}: Done{cursor}
2827
2828def snapshot.root.cursor.open(
2829  root: SnapshotRoot,
2830  bounds: RangeCursor.Bounds
2831) -> IO(Result<&1, &1, Error, VisibleCursor.Cursor>):
2832  do IO<Result<&1, &1, Error, VisibleCursor.Cursor>>:
2833    opened : Result<&1, &1, Error, SnapshotCursorRoot> <-
2834      snapshot.root.cursor.open.state(root, bounds)
2835    return snapshot.root.cursor.open.result(opened)
2836
2837def snapshot.root.get.closed(
2838  value: Maybe<&1, Bytes.Bytes>,
2839  closed: Result<&1, &1, U32 & String, Unit>
2840) -> Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>:
2841  match closed:
2842    case Fail{(code, message)}: Fail{recovery.error(code, message)}
2843    case Done{Unit{}}: Done{value}
2844
2845def snapshot.root.get.close(
2846  cursor: VisibleCursor.Cursor,
2847  value: Maybe<&1, Bytes.Bytes>
2848) -> IO(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
2849  do IO<Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
2850    closed : Result<&1, &1, U32 & String, Unit> <- VisibleCursor.close(cursor)
2851    return snapshot.root.get.closed(value, closed)
2852
2853def snapshot.root.get.item.checked(
2854  value: Maybe<&1, Bytes.Bytes>,
2855  matches: Bool
2856) -> Maybe<&1, Bytes.Bytes>:
2857  match value:
2858    case None{}: None{}
2859    case Some{bytes}:
2860      match matches:
2861        case True{}: Some{bytes}
2862        case False{}: None{}
2863
2864def snapshot.root.get.item.compared(
2865  value: Maybe<&1, Bytes.Bytes>,
2866  keys: Bytes.Bytes & Bytes.Bytes
2867) -> Maybe<&1, Bytes.Bytes>:
2868  match keys:
2869    case (key, query):
2870      snapshot.root.get.item.checked(value,
2871        SnapshotPolicy.point_read.matches(Keys.Bytes.cmp(key, query)))
2872
2873def snapshot.root.get.item.copies(
2874  value: Maybe<&1, Bytes.Bytes>,
2875  copies: (Bytes.Bytes & Bytes.Bytes) & (Bytes.Bytes & Bytes.Bytes)
2876) -> Maybe<&1, Bytes.Bytes>:
2877  match copies:
2878    case ((_, key), (query, _)):
2879      snapshot.root.get.item.compared(value, (key, query))
2880
2881def snapshot.root.get.item.value(
2882  key: Bytes.Bytes,
2883  value: Maybe<&1, Bytes.Bytes>,
2884  query: Bytes.Bytes
2885) -> Maybe<&1, Bytes.Bytes>:
2886  snapshot.root.get.item.copies(value,
2887    (ByteMemTable.bytes.clone(key), ByteMemTable.bytes.clone(query)))
2888
2889def snapshot.root.get.item(
2890  entry: ByteMemTable.Entry,
2891  cursor: VisibleCursor.Cursor,
2892  query: Bytes.Bytes
2893) -> IO(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
2894  match entry:
2895    case ByteMemTable.Entry{key, value}:
2896      snapshot.root.get.close(cursor,
2897        snapshot.root.get.item.value(key, value, query))
2898
2899def snapshot.root.get.step(
2900  query: Bytes.Bytes,
2901  step: VisibleCursor.Step
2902) -> SnapshotPointRead:
2903  SnapshotPointStep{step, query}
2904
2905def snapshot.root.get.exhausted.closed(
2906  closed: Result<&1, &1, U32 & String, Unit>
2907) -> Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>:
2908  match closed:
2909    case Fail{(code, message)}: Fail{recovery.error(code, message)}
2910    case Done{Unit{}}: Fail{ResourceLimit{}}
2911
2912def snapshot.root.get.exhausted(
2913  cursor: VisibleCursor.Cursor
2914) -> IO(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
2915  do IO<Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
2916    closed : Result<&1, &1, U32 & String, Unit> <- VisibleCursor.close(cursor)
2917    return snapshot.root.get.exhausted.closed(closed)
2918
2919def snapshot.root.get.exhausted.step(
2920  query: Bytes.Bytes,
2921  step: VisibleCursor.Step
2922) -> IO(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
2923  match step:
2924    case VisibleCursor.Item{entry, cursor}:
2925      snapshot.root.get.item(entry, cursor, query)
2926    case VisibleCursor.Skip{cursor}:
2927      snapshot.root.get.exhausted(cursor)
2928    case VisibleCursor.End{}:
2929      IO.pure(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>, Done{None{}})
2930    case VisibleCursor.Error{(code, message)}:
2931      IO.pure(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
2932        Fail{recovery.error(code, message)})
2933
2934def snapshot.root.get.loop(
2935  fuel: Nat,
2936  state: SnapshotPointRead
2937) -> IO(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
2938  match fuel:
2939    case 0n:
2940      match state:
2941        case SnapshotPointNeed{cursor, _}:
2942          snapshot.root.get.exhausted(cursor)
2943        case SnapshotPointStep{step, query}:
2944          snapshot.root.get.exhausted.step(query, step)
2945    case 1n+rest:
2946      match state:
2947        case SnapshotPointNeed{cursor, query}:
2948          do IO<Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
2949            step : VisibleCursor.Step <- VisibleCursor.next(cursor)
2950            snapshot.root.get.loop(rest,
2951              snapshot.root.get.step(query, step))
2952        case SnapshotPointStep{step, query}:
2953          match step:
2954            case VisibleCursor.Item{entry, cursor}:
2955              snapshot.root.get.item(entry, cursor, query)
2956            case VisibleCursor.Skip{cursor}:
2957              snapshot.root.get.loop(rest,
2958                SnapshotPointNeed{cursor, query})
2959            case VisibleCursor.End{}:
2960              IO.pure(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
2961                Done{None{}})
2962            case VisibleCursor.Error{(code, message)}:
2963              IO.pure(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>,
2964                Fail{recovery.error(code, message)})
2965
2966def snapshot.root.get.opened(
2967  query: Bytes.Bytes,
2968  opened: Result<&1, &1, Error, SnapshotCursorRoot>
2969) -> IO(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
2970  match opened:
2971    case Fail{error}:
2972      IO.pure(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>, Fail{error})
2973    case Done{SnapshotCursorRoot{retained_bytes, cursor}}:
2974      snapshot.root.get.loop(Nat.add(retained_bytes, 2n),
2975        SnapshotPointNeed{cursor, query})
2976
2977def snapshot.root.get.copied(
2978  root: SnapshotRoot,
2979  copies: Bytes.Bytes & Bytes.Bytes
2980) -> IO(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
2981  match copies:
2982    case (query, lower):
2983      do IO<Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>>:
2984        opened : Result<&1, &1, Error, SnapshotCursorRoot> <-
2985          snapshot.root.cursor.open.state(root,
2986            RangeCursor.Bounds{Some{lower}, None{}, None{}})
2987        snapshot.root.get.opened(query, opened)
2988
2989def snapshot.root.get(
2990  root: SnapshotRoot,
2991  key: Bytes.Bytes
2992) -> IO(Result<&1, &1, Error, Maybe<&1, Bytes.Bytes>>):
2993  snapshot.root.get.copied(root, ByteMemTable.bytes.clone(key))
2994
2995def snapshot.root.value.length.decision(
2996  decision: SnapshotValueLengthPolicy.Decision
2997) -> Result<&1, &1, Error, Maybe<&1, U32>>:
2998  match decision:
2999    case SnapshotValueLengthPolicy.NeedSst{}: Fail{ResourceLimit{}}
3000    case SnapshotValueLengthPolicy.Missing{}: Done{None{}}
3001    case SnapshotValueLengthPolicy.Found{value_bytes}:
3002      Done{Some{value_bytes}}
3003
3004def snapshot.root.value.length.metadata(
3005  probe: ByteMemTable.LengthProbe,
3006  measured: Result<&1, &1, U32 & String, ByteWal.ValueMetadata>
3007) -> Result<&1, &1, Error, Maybe<&1, U32>>:
3008  match measured:
3009    case Fail{(code, message)}:
3010      Fail{recovery.error(code, message)}
3011    case Done{metadata}:
3012      snapshot.root.value.length.decision(
3013        SnapshotValueLengthPolicy.result(probe, metadata))
3014
3015def snapshot.root.value.length.metadata.read(
3016  sst_path: Maybe<&1, String>,
3017  probe: ByteMemTable.LengthProbe,
3018  key: Bytes.Bytes
3019) -> IO(Result<&1, &1, Error, Maybe<&1, U32>>):
3020  match sst_path:
3021    case None{}:
3022      IO.pure(Result<&1, &1, Error, Maybe<&1, U32>>,
3023        snapshot.root.value.length.metadata(probe,
3024          Done{ByteWal.ValueAbsent{}}))
3025    case Some{path}:
3026      do IO<Result<&1, &1, Error, Maybe<&1, U32>>>:
3027        measured : Result<&1, &1, U32 & String, ByteWal.ValueMetadata> <-
3028          SstCursor.measure_key(path, key)
3029        return snapshot.root.value.length.metadata(probe, measured)
3030
3031def snapshot.root.value.length.delta.output(
3032  sst_path: Maybe<&1, String>,
3033  sst_generation: Maybe<&1, U32>,
3034  retained_bytes: Nat,
3035  lookup_key: Bytes.Bytes,
3036  looked_up: ByteMemTable.Table & ByteMemTable.LengthProbe
3037) -> SnapshotRoot & ByteMemTable.LengthProbe & Bytes.Bytes:
3038  match looked_up:
3039    case (ByteMemTable.Table{entries}, probe):
3040      (SnapshotRoot{sst_path, sst_generation, retained_bytes, entries},
3041        probe, lookup_key)
3042
3043def snapshot.root.value.length.delta.counted(
3044  sst_path: Maybe<&1, String>,
3045  sst_generation: Maybe<&1, U32>,
3046  retained_bytes: Nat,
3047  keys: Bytes.Bytes & Bytes.Bytes,
3048  counted: List<&1, ByteMemTable.Entry> & Nat
3049) -> SnapshotRoot & ByteMemTable.LengthProbe & Bytes.Bytes:
3050  match keys:
3051    case (probe_key, lookup_key):
3052      snapshot.root.value.length.delta.output(sst_path, sst_generation,
3053        retained_bytes, lookup_key,
3054        ByteMemTable.lookup.length.start.counted(probe_key, counted))
3055
3056def snapshot.root.value.length.delta(
3057  root: SnapshotRoot,
3058  keys: Bytes.Bytes & Bytes.Bytes
3059) -> SnapshotRoot & ByteMemTable.LengthProbe & Bytes.Bytes:
3060  match root:
3061    case SnapshotRoot{sst_path, sst_generation, retained_bytes, delta}:
3062      snapshot.root.value.length.delta.counted(sst_path, sst_generation,
3063        retained_bytes, keys, ByteMemTable.count.keep(delta))
3064
3065def snapshot.root.value.length.dispatch(
3066  root: SnapshotRoot,
3067  +probe: ByteMemTable.LengthProbe,
3068  key: Bytes.Bytes,
3069  decision: SnapshotValueLengthPolicy.Decision
3070) -> IO(Result<&1, &1, Error, Maybe<&1, U32>>):
3071  match root:
3072    case SnapshotRoot{sst_path, _, _, _}:
3073      match decision:
3074        case SnapshotValueLengthPolicy.NeedSst{}:
3075          snapshot.root.value.length.metadata.read(sst_path, probe, key)
3076        case SnapshotValueLengthPolicy.Missing{}:
3077          IO.pure(Result<&1, &1, Error, Maybe<&1, U32>>, Done{None{}})
3078        case SnapshotValueLengthPolicy.Found{value_bytes}:
3079          IO.pure(Result<&1, &1, Error, Maybe<&1, U32>>,
3080            Done{Some{value_bytes}})
3081
3082def snapshot.root.value.length.probed(
3083  measured: SnapshotRoot & ByteMemTable.LengthProbe & Bytes.Bytes
3084) -> IO(Result<&1, &1, Error, Maybe<&1, U32>>):
3085  match measured:
3086    case (root, (+probe, key)):
3087      snapshot.root.value.length.dispatch(root, probe, key,
3088        SnapshotValueLengthPolicy.delta(probe))
3089
3090def snapshot.root.value_length(
3091  root: SnapshotRoot,
3092  key: Bytes.Bytes
3093) -> IO(Result<&1, &1, Error, Maybe<&1, U32>>):
3094  snapshot.root.value.length.probed(snapshot.root.value.length.delta(root,
3095    ByteMemTable.bytes.clone(key)))
3096
3097# Restores a read handle after byte lookup.
3098def close.released(
3099  result: Result<&1, &1, DurableError.Error, Unit>
3100) -> Result<&1, &1, Error, Unit>:
3101  match result:
3102    case Fail{error}:
3103      Fail{lock.error(error)}
3104    case Done{Unit{}}:
3105      Done{Unit{}}
3106
3107# Releases database ownership and consumes the handle.
3108def close(handle: Handle) -> IO(Result<&1, &1, Error, Unit>):
3109  match handle:
3110    case Handle{path, lock, _, _, _}:
3111      do IO<Result<&1, &1, Error, Unit>>:
3112        result : Result<&1, &1, DurableError.Error, Unit> <-
3113          DbLock.release(lock, "close", path)
3114        return close.released(result)
3115
3116def stats.aggregate.copies(
3117  +path: String,
3118  lock: DbLock.Lock,
3119  state: HandleState,
3120  original: List<&1, ByteMemTable.Entry>,
3121  copies: VisibleStats.Stats & VisibleStats.Stats,
3122  delta_stats: VisibleStats.Stats
3123) -> Handle & Stats:
3124  match copies:
3125    case (measured_base, retained_base):
3126      match measured_base:
3127        case VisibleStats.Stats{base_entries, base_payload}:
3128          match delta_stats:
3129            case VisibleStats.Stats{delta_entries, delta_payload}:
3130              (Handle{path, lock, ByteMemTable.Table{original}, state, retained_base},
3131                Stats{Nat.add(base_entries, delta_entries),
3132                  Nat.add(base_payload, delta_payload)})
3133
3134
3135
3136def stats.aggregate(
3137  +path: String,
3138  lock: DbLock.Lock,
3139  state: HandleState,
3140  base_stats: VisibleStats.Stats,
3141  original: List<&1, ByteMemTable.Entry>,
3142  delta_stats: VisibleStats.Stats
3143) -> Handle & Stats:
3144  stats.aggregate.copies(path, lock, state, original,
3145    VisibleStats.clone(base_stats), delta_stats)
3146
3147def stats.measured.result(
3148  +path: String,
3149  lock: DbLock.Lock,
3150  state: HandleState,
3151  base_stats: VisibleStats.Stats,
3152  original: List<&1, ByteMemTable.Entry>,
3153  measured: VisibleStats.Stats
3154) -> Handle & Stats:
3155  match measured:
3156    case VisibleStats.Stats{entries, payload_bytes}:
3157      stats.aggregate(path, lock, state, base_stats, original,
3158        VisibleStats.Stats{entries, payload_bytes})
3159
3160def stats.visible(
3161  entries: List<&1, ByteMemTable.Entry>,
3162  current: VisibleStats.Stats
3163) -> VisibleStats.Stats:
3164  match entries:
3165    case Nil{}: current
3166    case Con{ByteMemTable.Entry{Bytes.Bytes{+len, _}, val}, tail}:
3167      match val:
3168        case None{}: stats.visible(tail, current)
3169        case Some{Bytes.Bytes{+value_len, _}}:
3170          stats.visible(tail, VisibleStats.result(VisibleStats.apply(current,
3171            U32.to_nat(len), None{}, Some{U32.to_nat(value_len)})))
3172
3173def stats.measured(
3174  +path: String,
3175  lock: DbLock.Lock,
3176  state: HandleState,
3177  base_stats: VisibleStats.Stats,
3178  original: List<&1, ByteMemTable.Entry>,
3179  visible: List<&1, ByteMemTable.Entry>
3180) -> Handle & Stats:
3181  stats.measured.result(path, lock, state, base_stats, original,
3182    stats.visible(visible, VisibleStats.Stats{0n, 0n}))
3183
3184def stats.canonical(
3185  +path: String,
3186  lock: DbLock.Lock,
3187  state: HandleState,
3188  base_stats: VisibleStats.Stats,
3189  pair: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
3190) -> Handle & Stats:
3191  match pair:
3192    case (original, working):
3193      stats.measured(path, lock, state, base_stats, original,
3194        ByteMemTable.compact.live(
3195          ByteMemTable.ordered_entries(ByteMemTable.Table{working})))
3196
3197# Counts visible entries and exact key/value payload bytes.
3198def stats(handle: Handle) -> Handle & Stats:
3199  match handle:
3200    case Handle{path, lock, ByteMemTable.Table{entries}, state, base_stats}:
3201      stats.canonical(path, lock, state, base_stats,
3202        ByteSstFile.entries.clone(entries))