src/DurableDb.bend
mylsm-lsm-store@0.5.0.0 · 511 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.5.0.0/src/Db.bend as Db
3import mylsm-lsm-store@0.5.0.0/src/DbLock.bend as DbLock
4import mylsm-lsm-store@0.5.0.0/src/DbIo.bend as DbIo
5import mylsm-lsm-store@0.5.0.0/src/DurableCommit.bend as Commit
6import mylsm-lsm-store@0.5.0.0/src/DurableError.bend as DurableError
7import mylsm-lsm-store@0.5.0.0/src/DurableDbPolicy.bend as Policy
8import mylsm-lsm-store@0.5.0.0/src/Fs.bend as Fs
9import mylsm-lsm-store@0.5.0.0/src/Recover.bend as Recover
10import mylsm-lsm-store@0.5.0.0/src/Flush.bend as Flush
11import mylsm-lsm-store@0.5.0.0/src/Manifest.bend as Manifest
12import mylsm-lsm-store@0.5.0.0/src/CompactIo.bend as Compact
13import mylsm-lsm-store@0.5.0.0/src/Wal.bend as Wal
14import mylsm-lsm-store@0.5.0.0/src/DurableBatchPolicy.bend as BatchPolicy
15import mylsm-lsm-store@0.5.0.0/src/MemTable.bend as MemTable
16import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
17
18# Owns the database, lock, and lifecycle state.
19type Handle is Type:
20 Handle{path: String, lock: DbLock.Lock, db: Db.Db, state: Policy.HandleState, operation_errors: Nat}
21
22# Collects bounded database, cache, and error counters.
23type Stats is Data:
24 Stats{active_bytes: Nat, wal_bytes: Nat, memtable_entries: Nat, memtable_payload_bytes: Nat, cache_entries: Nat, operation_errors: Nat, state: Policy.HandleState}
25
26# Tracks whether post-commit maintenance is safe to continue.
27type Maintenance is Data:
28 Maintained{}
29 Pending{error: DurableError.Error}
30
31# Separates commit confirmation from maintenance status.
32type WriteOutcome is Data:
33 WriteOutcome{maintenance: Maintenance}
34
35def create_error.kind(exists: Bool, +code: U32, message: String, +path: String) -> DurableError.Error:
36 match exists:
37 case True{}:
38 DurableError.Error{DurableError.AlreadyExists{}, "create", path, code, message}
39 case False{}:
40 DurableError.classify_open(DurableError.OpenHostFailure{code, message}, "create", path)
41
42def create_error.code(+code: U32, message: String, +path: String) -> DurableError.Error:
43 create_error.kind(U32.is_eq(code, 17), code, message, path)
44
45def open_error.kind(missing: Bool, code: U32, message: String, operation: String, path: String) -> DurableError.Error:
46 match missing:
47 case True{}:
48 DurableError.Error{DurableError.NotFound{}, operation, path, code, message}
49 case False{}:
50 DurableError.classify_recovery(code, message, operation, path)
51
52# Performs the open error operation and returns its result.
53def open_error(+code: U32, message: String, operation: String, path: String) -> DurableError.Error:
54 open_error.kind(U32.is_eq(code, 2), code, message, operation, path)
55
56# Performs the open result operation and returns its result.
57def open_result(
58 result: Result<&1, &1, U32 & String, Db.Db>,
59 lock: DbLock.Lock,
60 operation: String,
61 +path: String
62) -> IO(Result<&1, &1, DurableError.Error, Handle>):
63 match result:
64 case Fail{(code, message)}:
65 do IO<Result<&1, &1, DurableError.Error, Handle>>:
66 _released : Result<&1, &1, DurableError.Error, Unit> <- DbLock.release(lock, "close", path)
67 return Fail{open_error(code, message, operation, path)}
68 case Done{db}:
69 IO.pure(Result<&1, &1, DurableError.Error, Handle>,
70 Done{Handle{path, lock, db, Policy.Open{}, 0n}})
71
72# Continues existing-open recovery after acquiring the database lock.
73def locked_result(
74 result: Result<&1, &1, DurableError.Error, DbLock.Lock>,
75 operation: String,
76 +path: String
77) -> IO(Result<&1, &1, DurableError.Error, Handle>):
78 match result:
79 case Fail{error}:
80 IO.pure(Result<&1, &1, DurableError.Error, Handle>, Fail{error})
81 case Done{lock}:
82 do IO<Result<&1, &1, DurableError.Error, Handle>>:
83 opened : Result<&1, &1, U32 & String, Db.Db> <- Recover.open_existing_locked(path)
84 open_result(opened, lock, operation, path)
85
86# Acquires ownership only after validating the existing Manifest.
87def preflight_manifest(
88 manifest: Result<&1, &1, U32 & String, Manifest.Manifest>,
89 +path: String
90) -> IO(Result<&1, &1, DurableError.Error, Handle>):
91 match manifest:
92 case Fail{(code, message)}:
93 IO.pure(Result<&1, &1, DurableError.Error, Handle>,
94 Fail{DurableError.classify_recovery(code, message, "open_existing", path)})
95 case Done{_}:
96 do IO<Result<&1, &1, DurableError.Error, Handle>>:
97 acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <- DbLock.acquire(path, "open_existing")
98 locked_result(acquired, "open_existing", path)
99
100# Rejects a missing Manifest or begins validating the existing one.
101def preflight_result(
102 result: Result<&1, &1, U32 & String, Bool>,
103 +path: String
104) -> IO(Result<&1, &1, DurableError.Error, Handle>):
105 match result:
106 case Fail{(code, message)}:
107 IO.pure(Result<&1, &1, DurableError.Error, Handle>,
108 Fail{open_error(code, message, "open_existing", path)})
109 case Done{False{}}:
110 IO.pure(Result<&1, &1, DurableError.Error, Handle>,
111 Fail{DurableError.Error{DurableError.NotFound{}, "open_existing", path, 2, "manifest missing"}})
112 case Done{True{}}:
113 do IO<Result<&1, &1, DurableError.Error, Handle>>:
114 manifest : Result<&1, &1, U32 & String, Manifest.Manifest> <- Flush.load_mfst(path ++ "/MANIFEST")
115 preflight_manifest(manifest, path)
116
117# Opens and recovers an existing database under its lock.
118def open_existing(+path: String) -> IO(Result<&1, &1, DurableError.Error, Handle>):
119 do IO<Result<&1, &1, DurableError.Error, Handle>>:
120 present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path ++ "/MANIFEST")
121 preflight_result(present, path)
122
123# Performs the create locked operation and returns its result.
124def create_locked(
125 result: Result<&1, &1, DurableError.Error, DbLock.Lock>,
126 +path: String
127) -> IO(Result<&1, &1, DurableError.Error, Handle>):
128 match result:
129 case Fail{error}:
130 IO.pure(Result<&1, &1, DurableError.Error, Handle>, Fail{error})
131 case Done{lock}:
132 do IO<Result<&1, &1, DurableError.Error, Handle>>:
133 initialized : Result<&1, &1, U32 & String, Db.Db> <- Recover.create_locked(path)
134 open_result(initialized, lock, "create", path)
135
136# Acquires ownership after reserving the new database directory.
137def created_dir(
138 result: Result<&1, &1, U32 & String, Unit>,
139 +path: String
140) -> IO(Result<&1, &1, DurableError.Error, Handle>):
141 match result:
142 case Fail{(code, message)}:
143 IO.pure(Result<&1, &1, DurableError.Error, Handle>,
144 Fail{create_error.code(code, message, path)})
145 case Done{Unit{}}:
146 do IO<Result<&1, &1, DurableError.Error, Handle>>:
147 acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <- DbLock.acquire(path, "create")
148 create_locked(acquired, path)
149
150# Creates a database and returns its owned durable handle.
151def create(+path: String) -> IO(Result<&1, &1, DurableError.Error, Handle>):
152 do IO<Result<&1, &1, DurableError.Error, Handle>>:
153 reserved : Result<&1, &1, U32 & String, Unit> <- Fs.make_dir(path)
154 created_dir(reserved, path)
155
156# Closes the durable handle and releases its database lock.
157def close(handle: Handle) -> IO(Result<&1, &1, DurableError.Error, Unit>):
158 match handle:
159 case Handle{path, lock, _, _, _}:
160 DbLock.release(lock, "close", path)
161
162# Returns a value for readable handles and rejects reads after uncertain commit.
163def get_result(
164 state: Policy.HandleState,
165 pair: Db.Db & Maybe<&2, String>,
166 lock: DbLock.Lock,
167 +operation_errors: Nat,
168 +path: String
169) -> Handle & Result<&1, &1, DurableError.Error, Maybe<&2, String>>:
170 match state:
171 case Policy.Open{}:
172 match pair:
173 case (db, value):
174 (Handle{path, lock, db, Policy.Open{}, operation_errors}, Done{value})
175 case Policy.Poisoned{}:
176 match pair:
177 case (db, _):
178 (Handle{path, lock, db, Policy.Poisoned{}, Policy.next_operation_errors(operation_errors)},
179 Fail{DurableError.Error{
180 DurableError.CommitUnknown{}, "get", path, 0, "handle requires recovery"
181 }})
182 case Policy.MaintenanceBlocked{}:
183 match pair:
184 case (db, value):
185 (Handle{path, lock, db, Policy.MaintenanceBlocked{}, operation_errors}, Done{value})
186
187# Reads a key from the owned durable database handle.
188def get(handle: Handle, +key: String) -> IO(Handle & Result<&1, &1, DurableError.Error, Maybe<&2, String>>):
189 match handle:
190 case Handle{path, lock, db, state, operation_errors}:
191 IO.pure(Handle & Result<&1, &1, DurableError.Error, Maybe<&2, String>>,
192 get_result(state, Db.db_get_cached(db, key), lock, operation_errors, path))
193
194# Preserves the confirmed commit when post-commit maintenance fails.
195def maintenance_after_commit(
196 result: Result<&1, &1, U32 & String, Db.Db>,
197 +committed: Db.Db,
198 lock: DbLock.Lock,
199 operation_errors: Nat,
200 +path: String
201) -> IO(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>):
202 match result:
203 case Fail{(code, message)}:
204 IO.pure(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>,
205 (Handle{path, lock, committed, Policy.MaintenanceBlocked{}, Policy.next_operation_errors(operation_errors)},
206 Done{WriteOutcome{Pending{DurableError.Error{DurableError.Io{}, "maintain", path, code, message}}}}))
207 case Done{db}:
208 IO.pure(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>,
209 (Handle{path, lock, db, Policy.Open{}, operation_errors}, Done{WriteOutcome{Maintained{}}}))
210
211# Maps WAL encoding limits to public write-validation causes.
212def encoding_rejected_cause(error: Wal.Error) -> DurableError.WriteCause:
213 match error:
214 case Wal.TooLarge{}:
215 DurableError.WriteLimitReached{}
216 case _:
217 DurableError.ValidationFailure{}
218
219# Updates handle state from the WAL commit stage and maintains durable commits.
220def committed_result(
221 committed: Db.Db & Commit.Stage,
222 lock: DbLock.Lock,
223 operation_errors: Nat,
224 +path: String
225) -> IO(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>):
226 match committed:
227 case (db, Commit.EncodingRejected{error}):
228 IO.pure(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>,
229 (Handle{path, lock, db, Policy.Open{}, Policy.next_operation_errors(operation_errors)}, Fail{DurableError.classify_write(
230 DurableError.BeforeAppend{}, encoding_rejected_cause(error), "write", path)}))
231 case (db, Commit.AppendRejected{code, message}):
232 IO.pure(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>,
233 (Handle{path, lock, db, Policy.Open{}, Policy.next_operation_errors(operation_errors)}, Fail{DurableError.classify_write(
234 DurableError.BeforeAppend{}, DurableError.WriteHostFailure{code, message}, "write", path)}))
235 case (db, Commit.AppendUnknown{code, message}):
236 IO.pure(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>,
237 (Handle{path, lock, db, Policy.Poisoned{}, Policy.next_operation_errors(operation_errors)}, Fail{DurableError.Error{
238 DurableError.CommitUnknown{}, "write", path, code, message}}))
239 case (+changed, Commit.DurableAppend{}):
240 do IO<Handle & Result<&1, &1, DurableError.Error, WriteOutcome>>:
241 maintenance : Result<&1, &1, U32 & String, Db.Db> <- Recover.maintain(changed)
242 outcome : Handle & Result<&1, &1, DurableError.Error, WriteOutcome> <- maintenance_after_commit(maintenance, changed, lock, operation_errors, path)
243 return outcome
244
245# Commits a validated mutation batch and returns the updated owned handle.
246def batch_result(
247 +db: Db.Db,
248 batch: Wal.Batch,
249 lock: DbLock.Lock,
250 operation_errors: Nat,
251 +path: String,
252) -> IO(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>):
253 do IO<Handle & Result<&1, &1, DurableError.Error, WriteOutcome>>:
254 committed : Db.Db & Commit.Stage <- DbIo.db_write_durable(db, batch)
255 outcome : Handle & Result<&1, &1, DurableError.Error, WriteOutcome> <- committed_result(committed, lock, operation_errors, path)
256 return outcome
257
258# Rejects empty and oversized batches before WAL access.
259def batch_count_result(
260 +db: Db.Db,
261 muts: List<&2, Wal.Mut>,
262 lock: DbLock.Lock,
263 operation_errors: Nat,
264 +path: String,
265 validation: BatchPolicy.Validation,
266) -> IO(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>):
267 match validation:
268 case BatchPolicy.ValidBatch{}:
269 batch_result(db, Wal.Batch{muts}, lock, operation_errors, path)
270 case BatchPolicy.EmptyBatch{}:
271 IO.pure(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>,
272 (Handle{path, lock, db, Policy.Open{}, Policy.next_operation_errors(operation_errors)},
273 Fail{DurableError.Error{
274 DurableError.InvalidArgument{}, "write", path, 0, "empty batch"
275 }}))
276 case BatchPolicy.BatchCountExceeded{}:
277 IO.pure(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>,
278 (Handle{path, lock, db, Policy.Open{}, Policy.next_operation_errors(operation_errors)},
279 Fail{DurableError.Error{
280 DurableError.ResourceLimit{}, "write", path, 0, "batch mutation count exceeds 256"
281 }}))
282# Validates the mutation count before attempting a commit.
283def batch_count(
284 +db: Db.Db,
285 +muts: List<&2, Wal.Mut>,
286 lock: DbLock.Lock,
287 operation_errors: Nat,
288 +path: String,
289) -> IO(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>):
290 batch_count_result(db, muts, lock, operation_errors, path, BatchPolicy.validate_count(List.length(&2, Wal.Mut, muts)))
291
292# Validates and commits a public mutation batch atomically.
293def write_batch(
294 handle: Handle,
295 muts: List<&2, Wal.Mut>,
296) -> IO(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>):
297 match handle:
298 case Handle{+path, lock, db, state, operation_errors}:
299 match state:
300 case Policy.Open{}:
301 batch_count(db, muts, lock, operation_errors, path)
302 case Policy.Poisoned{}:
303 IO.pure(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>,
304 (Handle{path, lock, db, Policy.Poisoned{}, Policy.next_operation_errors(operation_errors)}, Fail{DurableError.Error{
305 DurableError.CommitUnknown{}, "write", path, 0, "handle requires recovery"}}))
306 case Policy.MaintenanceBlocked{}:
307 IO.pure(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>,
308 (Handle{path, lock, db, Policy.MaintenanceBlocked{}, Policy.next_operation_errors(operation_errors)}, Fail{DurableError.Error{
309 DurableError.Io{}, "write", path, 0, "maintenance requires recovery"}}))
310
311
312# Writes one key and value through the durable handle.
313def put(handle: Handle, +key: String, +value: String) -> IO(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>):
314 write_batch(handle, Con{Wal.Put{key, value}, Nil{}})
315
316# Deletes one key through the durable handle.
317def delete(handle: Handle, +key: String) -> IO(Handle & Result<&1, &1, DurableError.Error, WriteOutcome>):
318 write_batch(handle, Con{Wal.Del{key}, Nil{}})
319
320# Preserves the original database when maintenance fails.
321def maintenance_result(
322 result: Result<&1, &1, U32 & String, Db.Db>,
323 +original: Db.Db,
324 operation: String,
325 lock: DbLock.Lock,
326 operation_errors: Nat,
327 +path: String
328) -> Handle & Result<&1, &1, DurableError.Error, Unit>:
329 match result:
330 case Fail{(code, message)}:
331 (Handle{path, lock, original, Policy.MaintenanceBlocked{}, Policy.next_operation_errors(operation_errors)},
332 Fail{DurableError.Error{DurableError.Io{}, operation, path, code, message}})
333 case Done{db}:
334 (Handle{path, lock, db, Policy.Open{}, operation_errors}, Done{Unit{}})
335
336# Flushes the database and updates handle state from the result.
337def flush_result(
338 +db: Db.Db,
339 lock: DbLock.Lock,
340 operation_errors: Nat,
341 +path: String
342) -> IO(Handle & Result<&1, &1, DurableError.Error, Unit>):
343 do IO<Handle & Result<&1, &1, DurableError.Error, Unit>>:
344 result : Result<&1, &1, U32 & String, Db.Db> <- Flush.flush(db)
345 return maintenance_result(result, db, "flush", lock, operation_errors, path)
346
347# Compacts the database and updates handle state from the result.
348def compact_result(
349 +db: Db.Db,
350 lock: DbLock.Lock,
351 operation_errors: Nat,
352 +path: String,
353) -> IO(Handle & Result<&1, &1, DurableError.Error, Unit>):
354 do IO<Handle & Result<&1, &1, DurableError.Error, Unit>>:
355 result : Result<&1, &1, U32 & String, Db.Db> <- Compact.compact(db)
356 return maintenance_result(result, db, "compact", lock, operation_errors, path)
357
358# Flushes the durable handle and returns maintenance status.
359def flush(handle: Handle) -> IO(Handle & Result<&1, &1, DurableError.Error, Unit>):
360 match handle:
361 case Handle{+path, lock, db, Policy.Open{}, operation_errors}:
362 flush_result(db, lock, operation_errors, path)
363 case Handle{+path, lock, db, Policy.Poisoned{}, operation_errors}:
364 IO.pure(Handle & Result<&1, &1, DurableError.Error, Unit>,
365 (Handle{path, lock, db, Policy.Poisoned{}, Policy.next_operation_errors(operation_errors)}, Fail{DurableError.Error{
366 DurableError.CommitUnknown{}, "flush", path, 0, "handle requires recovery"}}))
367 case Handle{+path, lock, db, Policy.MaintenanceBlocked{}, operation_errors}:
368 IO.pure(Handle & Result<&1, &1, DurableError.Error, Unit>,
369 (Handle{path, lock, db, Policy.MaintenanceBlocked{}, Policy.next_operation_errors(operation_errors)}, Fail{DurableError.Error{
370 DurableError.Io{}, "flush", path, 0, "maintenance requires recovery"}}))
371
372# Compacts the durable handle and returns maintenance status.
373def compact(handle: Handle) -> IO(Handle & Result<&1, &1, DurableError.Error, Unit>):
374 match handle:
375 case Handle{+path, lock, db, Policy.Open{}, operation_errors}:
376 compact_result(db, lock, operation_errors, path)
377 case Handle{+path, lock, db, Policy.Poisoned{}, operation_errors}:
378 IO.pure(Handle & Result<&1, &1, DurableError.Error, Unit>,
379 (Handle{path, lock, db, Policy.Poisoned{}, Policy.next_operation_errors(operation_errors)}, Fail{DurableError.Error{
380 DurableError.CommitUnknown{}, "compact", path, 0, "handle requires recovery"}}))
381 case Handle{+path, lock, db, Policy.MaintenanceBlocked{}, operation_errors}:
382 IO.pure(Handle & Result<&1, &1, DurableError.Error, Unit>,
383 (Handle{path, lock, db, Policy.MaintenanceBlocked{}, Policy.next_operation_errors(operation_errors)}, Fail{DurableError.Error{
384 DurableError.Io{}, "compact", path, 0, "maintenance requires recovery"}}))
385
386# Tracks directory entries while collecting active-file sizes.
387type StatsNamesState is Type:
388 StatsNames{names: List<&2, String>, dir: String, total: Nat}
389 StatsNamesSized{result: Result<&1, &1, U32 & String, Nat>, names: List<&2, String>, dir: String, total: Nat}
390
391# Sums file sizes for the supplied directory entries within a fixed budget.
392def stats_names(fuel: Nat, state: StatsNamesState) -> IO(Result<&1, &1, U32 & String, Nat>):
393 match fuel:
394 case 0n:
395 IO.pure(Result<&1, &1, U32 & String, Nat>, Fail{(7, "active-byte stat limit exceeded")})
396 case 1n+rest:
397 match state:
398 case StatsNames{names, +dir, total}:
399 match names:
400 case Nil{}:
401 IO.pure(Result<&1, &1, U32 & String, Nat>, Done{total})
402 case Con{name, tail}:
403 do IO<Result<&1, &1, U32 & String, Nat>>:
404 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(dir ++ "/" ++ name)
405 stats_names(rest, StatsNamesSized{size, tail, dir, total})
406 case StatsNamesSized{result, names, +dir, total}:
407 match result:
408 case Fail{error}:
409 IO.pure(Result<&1, &1, U32 & String, Nat>, Fail{error})
410 case Done{size}:
411 stats_names(rest, StatsNames{names, dir, Nat.add(total, size)})
412
413# Tracks Manifest levels while collecting active-file sizes.
414type StatsLevelsState is Type:
415 StatsLevels{levels: List<&2, List<&2, String>>, index: Nat, dir: String, total: Nat}
416 StatsLevelsSized{result: Result<&1, &1, U32 & String, Nat>, levels: List<&2, List<&2, String>>, index: Nat, dir: String, total: Nat}
417
418# Sums published SSTable sizes across Manifest levels within a fixed budget.
419def stats_levels(fuel: Nat, state: StatsLevelsState) -> IO(Result<&1, &1, U32 & String, Nat>):
420 match fuel:
421 case 0n:
422 IO.pure(Result<&1, &1, U32 & String, Nat>, Fail{(7, "active-byte stat limit exceeded")})
423 case 1n+rest:
424 match state:
425 case StatsLevels{levels, +index, +dir, total}:
426 match levels:
427 case Nil{}:
428 IO.pure(Result<&1, &1, U32 & String, Nat>, Done{total})
429 case Con{+names, tail}:
430 +level_dir = dir ++ "/l" ++ Nat.show(index)
431 do IO<Result<&1, &1, U32 & String, Nat>>:
432 files : Result<&1, &1, U32 & String, Nat> <- stats_names(
433 Nat.add(Nat.mul(2n, List.length(&2, String, names)), 1n),
434 StatsNames{names, level_dir, 0n})
435 stats_levels(rest, StatsLevelsSized{files, tail, Nat.add(index, 1n), dir, total})
436 case StatsLevelsSized{result, levels, index, +dir, total}:
437 match result:
438 case Fail{error}:
439 IO.pure(Result<&1, &1, U32 & String, Nat>, Fail{error})
440 case Done{size}:
441 stats_levels(rest, StatsLevels{levels, index, dir, Nat.add(total, size)})
442
443# Adds the sizes of all published SSTables to the Manifest size.
444def stats_active_manifest_size(
445 result: Result<&1, &1, U32 & String, Nat>,
446 +levels: List<&2, List<&2, String>>,
447 +path: String
448) -> IO(Result<&1, &1, U32 & String, Nat>):
449 match result:
450 case Fail{error}:
451 IO.pure(Result<&1, &1, U32 & String, Nat>, Fail{error})
452 case Done{size}:
453 stats_levels(Nat.add(Nat.mul(2n, List.length(&2, List<&2, String>, levels)), 1n),
454 StatsLevels{levels, 0n, path, size})
455
456# Reads the Manifest and sums only the files it references.
457def stats_active_manifest(
458 result: Result<&1, &1, U32 & String, Manifest.Manifest>,
459 +path: String
460) -> IO(Result<&1, &1, U32 & String, Nat>):
461 match result:
462 case Fail{error}:
463 IO.pure(Result<&1, &1, U32 & String, Nat>, Fail{error})
464 case Done{Manifest.M{levels}}:
465 do IO<Result<&1, &1, U32 & String, Nat>>:
466 manifest_bytes : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path ++ "/MANIFEST")
467 stats_active_manifest_size(manifest_bytes, levels, path)
468
469# Builds counters from file sizes and the in-memory database state.
470def stats_result(
471 results: Result<&1, &1, U32 & String, Nat> & Result<&1, &1, U32 & String, Nat>,
472 +db: Db.Db,
473 +state: Policy.HandleState,
474 lock: DbLock.Lock,
475 +operation_errors: Nat,
476 +path: String
477) -> Handle & Result<&1, &1, DurableError.Error, Stats>:
478 match results:
479 case (Fail{(code, message)}, _):
480 (Handle{path, lock, db, state, Policy.next_operation_errors(operation_errors)}, Fail{DurableError.classify_recovery(code, message, "stats", path)})
481 case (_, Fail{(code, message)}):
482 (Handle{path, lock, db, state, Policy.next_operation_errors(operation_errors)}, Fail{DurableError.Error{DurableError.Io{}, "stats", path, code, message}})
483 case (Done{active_bytes}, Done{wal_bytes}):
484 match db:
485 case Db.Db{_, mem, frozen, _, bcache, _, _, _, mem_count, frozen_count}:
486 (Handle{path, lock, db, state, operation_errors}, Done{Stats{
487 active_bytes, wal_bytes, Nat.add(mem_count, frozen_count),
488 Nat.add(MemTable.payload_bytes(mem), MemTable.payload_bytes(frozen)),
489 List.length(&2, Db.BEntry, bcache), operation_errors, state
490 }})
491
492# Reads Manifest and WAL sizes before building the stats record.
493def stats_io(
494 db: Db.Db,
495 state: Policy.HandleState,
496 lock: DbLock.Lock,
497 operation_errors: Nat,
498 +path: String,
499) -> IO(Handle & Result<&1, &1, DurableError.Error, Stats>):
500
501 do IO<Handle & Result<&1, &1, DurableError.Error, Stats>>:
502 manifest : Result<&1, &1, U32 & String, Manifest.Manifest> <- Flush.load_mfst(path ++ "/MANIFEST")
503 active : Result<&1, &1, U32 & String, Nat> <- stats_active_manifest(manifest, path)
504 wal : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(Db.wal_path(path))
505 return stats_result((active, wal), db, state, lock, operation_errors, path)
506
507# Reads bounded operational stats from the durable handle.
508def stats(handle: Handle) -> IO(Handle & Result<&1, &1, DurableError.Error, Stats>):
509 match handle:
510 case Handle{+path, lock, db, state, operation_errors}:
511 stats_io(db, state, lock, operation_errors, path)