1 import Base 2 import ./Db.bend as Db 3 import ./DbLock.bend as DbLock 4 import ./DbIo.bend as DbIo 5 import ./DurableCommit.bend as Commit 6 import ./DurableError.bend as DurableError 7 import ./DurableDbPolicy.bend as Policy 8 import ./Fs.bend as Fs 9 import ./Recover.bend as Recover 10 import ./Flush.bend as Flush 11 import ./Manifest.bend as Manifest 12 import ./CompactIo.bend as Compact 13 import ./Wal.bend as Wal 14 import ./DurableBatchPolicy.bend as BatchPolicy 15 import ./MemTable.bend as MemTable 16 import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes 17 18 # Owns the database, lock, and lifecycle state. 19 type 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. 23 type 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. 27 type Maintenance is Data: 28 Maintained{} 29 Pending{error: DurableError.Error} 30 31 # Separates commit confirmation from maintenance status. 32 type WriteOutcome is Data: 33 WriteOutcome{maintenance: Maintenance} 34 35 def 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 42 def create_error.code(+code: U32, message: String, +path: String) -> DurableError.Error: 43 create_error.kind(U32.is_eq(code, 17), code, message, path) 44 45 def 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. 53 def 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. 57 def 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>: 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. 73 def 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>: 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. 87 def 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>: 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. 101 def 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>: 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. 118 def open_existing(+path: String) -> IO(Result<&1, &1, DurableError.Error, Handle>): 119 do IO>: 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. 124 def 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>: 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. 137 def 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>: 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. 151 def create(+path: String) -> IO(Result<&1, &1, DurableError.Error, Handle>): 152 do IO>: 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. 157 def 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. 163 def 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. 188 def 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. 195 def 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. 212 def 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. 220 def 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>: 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. 246 def 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>: 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. 259 def 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. 283 def 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. 293 def 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. 313 def 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. 317 def 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. 321 def 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. 337 def 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>: 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. 348 def 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>: 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. 359 def 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. 373 def 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. 387 type 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. 392 def 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>: 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. 414 type 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. 419 def 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>: 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. 444 def 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. 457 def 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>: 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. 470 def 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. 493 def 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>: 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. 508 def 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)