src/ByteDurableDb.bend
mylsm-lsm-store@0.6.0.0 · 3,202 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.0/src/ByteManifest.bend as ByteManifest
4import mylsm-lsm-store@0.6.0.0/src/ByteManifestIo.bend as ByteManifestIo
5import mylsm-lsm-store@0.6.0.0/src/ByteSstFile.bend as ByteSstFile
6import mylsm-lsm-store@0.6.0.0/src/ByteSstIo.bend as ByteSstIo
7import mylsm-lsm-store@0.6.0.0/src/ByteWal.bend as ByteWal
8import mylsm-lsm-store@0.6.0.0/src/ByteWalIo.bend as ByteWalIo
9import mylsm-lsm-store@0.6.0.0/src/DbLock.bend as DbLock
10import mylsm-lsm-store@0.6.0.0/src/DurableError.bend as DurableError
11import mylsm-lsm-store@0.6.0.0/src/DurableBatchPolicy.bend as BatchPolicy
12import mylsm-lsm-store@0.6.0.0/src/coordinator/QueuePolicy.bend as QueuePolicy
13import mylsm-lsm-store@0.6.0.0/src/coordinator/SnapshotValueLengthPolicy.bend as SnapshotValueLengthPolicy
14import mylsm-lsm-store@0.6.0.0/src/Fs.bend as Fs
15import mylsm-lsm-store@0.6.0.0/src/FilesCompat.bend as Files
16import mylsm-lsm-store@0.6.0.0/src/Keys.bend as Keys
17import mylsm-lsm-store@0.6.0.0/src/scan/Cursor.bend as Cursor
18import mylsm-lsm-store@0.6.0.0/src/scan/Snapshot.bend as SnapshotPolicy
19import mylsm-lsm-store@0.6.0.0/src/scan/ByteBlockReader.bend as ByteBlockReader
20import mylsm-lsm-store@0.6.0.0/src/scan/SstCursor.bend as SstCursor
21import mylsm-lsm-store@0.6.0.0/src/scan/RangeCursor.bend as RangeCursor
22import mylsm-lsm-store@0.6.0.0/src/scan/VisibleCursor.bend as VisibleCursor
23import mylsm-lsm-store@0.6.0.0/src/scan/VisibleStats.bend as VisibleStats
24import mylsm-lsm-store@0.6.0.0/src/CrashPoint.bend as CrashPoint
25import mylsm-lsm-store@0.6.0.0/src/scan/ScanWorkPolicy.bend as ScanWorkPolicy
26import mylsm-lsm-store@0.6.0.0/src/ByteRecoveryPolicy.bend as RecoveryPolicy
27import mylsm-lsm-store@0.6.0.0/src/ValueLengthPolicy.bend as ValueLengthPolicy
28import mylsm-lsm-store@0.6.0.0/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))