1 import Base 2 import ./src/Keys.bend as Keys 3 import ./src/MemTable.bend as MemTable 4 import ./src/SortedRun.bend as SortedRun 5 import ./src/Sstable.bend as Sstable 6 import ./src/SstFile.bend as SstFile 7 import ./src/Wal.bend as Wal 8 import ./src/Manifest.bend as Manifest 9 import ./src/MergeIter.bend as MergeIter 10 import ./src/Db.bend as Db 11 import ./src/Recover.bend as Recover 12 import ./src/DurableDb.bend as DurableDb 13 import ./src/DurableError.bend as DurableError 14 import ./src/DurableDbPolicy.bend as DurableDbPolicy 15 import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes 16 17 type InMemory.Session is Kind(a <&> &1): 18 InMemorySession{run: Db.Db -> (Db.Db & A)} 19 20 # --- Level 2: part control (1:1 delegates) --- 21 def cmp(+s1: String, +s2: String) -> Cmp: 22 Keys.cmp(s1, s2) 23 24 # Key equality: true exactly for identical strings. 25 def eq(+s1: String, +s2: String) -> Bool: 26 Keys.eq(s1, s2) 27 28 # Empty MemTable: no entries, count zero. 29 def mem_empty() -> MemTable.MemTable: 30 MemTable.empty() 31 32 # MemTable with one live version prepended (newest-first log). 33 def mem_put(tab: MemTable.MemTable, key: String, val: String) -> MemTable.MemTable: 34 MemTable.put(tab, key, val) 35 36 # MemTable with one tombstone prepended; hides older versions on read. 37 def mem_del(tab: MemTable.MemTable, key: String) -> MemTable.MemTable: 38 MemTable.del(tab, key) 39 40 # Newest-first read; a tombstone answers None and hides older versions. 41 def mem_get(tab: MemTable.MemTable, +key: String) -> Maybe<&2, String>: 42 MemTable.get(tab, key) 43 44 # Number of log entries, live and tombstoned. 45 def mem_count(tab: MemTable.MemTable) -> Nat: 46 MemTable.count(tab) 47 48 # Canonicalize newest-first entries into a strict ordered run. 49 def sort_newest(entries: List<&2, MemTable.Entry>) -> List<&2, MemTable.Entry>: 50 SortedRun.sort_newest(entries) 51 52 # Range scan over a merged run: lo <= key < hi, tombstones dropped. 53 def range_scan(+merged: List<&2, MemTable.Entry>, lo: String, hi: String) -> List<&2, MemTable.Entry>: 54 MergeIter.scan(merged, lo, hi) 55 56 # Build an SSTable, canonicalizing entries newest-first. 57 def sst_build(+entries: List<&2, MemTable.Entry>, level: Nat, est_keys: Nat) -> Sstable.Table: 58 Sstable.build(entries, level, est_keys) 59 60 # Build an SSTable trusting an already strict, unique run (no sorting). 61 def sst_from_sorted_unique(+entries: List<&2, MemTable.Entry>, level: Nat) -> Sstable.Table: 62 Sstable.from_sorted_unique(entries, level) 63 64 # Build an SSTable from a sorted run with an explicit key estimate. 65 def sst_build_sorted(+entries: List<&2, MemTable.Entry>, level: Nat, est_keys: Nat) -> Sstable.Table: 66 Sstable.build_sorted(entries, level, est_keys) 67 68 # Handle wal encode in the public module exports. 69 def wal_encode(batch: Wal.Batch) -> Result<&1, &1, Wal.Error, Bytes.Bytes>: 70 Wal.encode_frame(batch) 71 72 # Handle wal decode in the public module exports. 73 def wal_decode(encoded: Bytes.Bytes) -> Result<&1, &1, Wal.Error, Wal.Batch>: 74 Wal.decode_frame(encoded) 75 76 # Encode an SST as packed v3 bytes. 77 def sst_encode( 78 +entries: List<&2, MemTable.Entry>, 79 +level: U32 80 ) -> Result<&1, &1, SstFile.Error, Bytes.Bytes>: 81 SstFile.encode_file(entries, level) 82 83 # Parse packed v3 bytes with strict bounds and checksum verification. 84 def sst_parse( 85 encoded: Bytes.Bytes 86 ) -> Result<&1, &1, SstFile.Error, Sstable.Table>: 87 SstFile.parse(encoded) 88 89 # Serialize a Manifest to its exact on-disk bytes. 90 def mfst_serialize(mfst: Manifest.Manifest) -> Result<&1, &1, Manifest.Error, Bytes.Bytes>: 91 Manifest.serialize(mfst) 92 93 # Handle mfst parse in the public module exports. 94 def mfst_parse(encoded: Bytes.Bytes) -> Maybe<&2, Manifest.Manifest>: 95 Manifest.parse(encoded) 96 97 def InMemory.Session.pure(a, -A: Kind(a), value: A) -> InMemory.Session: 98 InMemorySession{database => (database, value)} 99 100 def InMemory.Session.run(a, -A: Kind(a), session: InMemory.Session, database: Db.Db) -> Db.Db & A: 101 match session: 102 case InMemorySession{run}: run(database) 103 104 def InMemory.Session.apply(a, -A: Kind(a), b, -B: Kind(b), pair: Db.Db & A, next: A -> InMemory.Session) -> Db.Db & B: 105 match pair: 106 case (database, value): InMemory.Session.run(b, B, next(value), database) 107 108 def InMemory.Session.bind(a, -A: Kind(a), -B: Kind(a), session: InMemory.Session, next: A -> InMemory.Session) -> InMemory.Session: 109 InMemorySession{database => InMemory.Session.apply(a, A, a, B, InMemory.Session.run(a, A, session, database), next)} 110 111 def InMemory.open(+path: String) -> Db.Db: 112 Db.open_db(path) 113 114 def InMemory.put_database(+database: Db.Db, +key: String, +value: String) -> Db.Db: 115 match database: 116 case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}: 117 Db.apply_done(Db.apply_and_rotate(Con{Wal.Put{key, value}, Nil{}}, mem, frozen, mem_count, frozen_count), dir, batch_cap, bcache, levels, flushed, manifest_token) 118 119 def InMemory.delete_database(+database: Db.Db, +key: String) -> Db.Db: 120 match database: 121 case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}: 122 Db.apply_done(Db.apply_and_rotate(Con{Wal.Del{key}, Nil{}}, mem, frozen, mem_count, frozen_count), dir, batch_cap, bcache, levels, flushed, manifest_token) 123 124 def InMemory.write_database_batch(+database: Db.Db, +mutations: List<&2, Wal.Mut>) -> Db.Db: 125 match database: 126 case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}: 127 Db.apply_done(Db.apply_and_rotate(mutations, mem, frozen, mem_count, frozen_count), dir, batch_cap, bcache, levels, flushed, manifest_token) 128 129 def InMemory.read_database(+database: Db.Db, +key: String) -> Db.Db & Maybe<&2, String>: 130 Db.db_get_cached(database, key) 131 132 def InMemory.Session.put(+key: String, +value: String) -> InMemory.Session<&2, Unit>: 133 InMemorySession{database => (InMemory.put_database(database, key, value), Unit{})} 134 135 def InMemory.Session.delete(+key: String) -> InMemory.Session<&2, Unit>: 136 InMemorySession{database => (InMemory.delete_database(database, key), Unit{})} 137 138 def InMemory.Session.write_batch(+mutations: List<&2, Wal.Mut>) -> InMemory.Session<&2, Unit>: 139 InMemorySession{database => (InMemory.write_database_batch(database, mutations), Unit{})} 140 141 def InMemory.Session.get(+key: String) -> InMemory.Session<&2, Maybe<&2, String>>: 142 InMemorySession{database => InMemory.read_database(database, key)} 143 144 def InMemory.run_session(a, -A: Kind(a), database: Db.Db, session: InMemory.Session) -> Db.Db & A: 145 InMemory.Session.run(a, A, session, database) 146 147 def InMemory.value_of(a, -A: Kind(a), pair: Db.Db & A) -> A: 148 match pair: 149 case (database, value): value 150 151 # Recover a persisted database from its directory. 152 def open_recovering(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>): 153 Recover.open_db(dir) 154 155 156 # Lists the error categories exposed by the durable API. 157 type ErrorKind is Data: 158 NotFoundError{} 159 AlreadyExistsError{} 160 BusyError{} 161 InvalidArgumentError{} 162 UnsupportedFormatError{} 163 CorruptionError{} 164 ResourceLimitError{} 165 IoError{} 166 CommitUnknownError{} 167 168 # Carries structured error details for API callers. 169 type Error is Data: 170 Error{kind: ErrorKind, operation: String, path: String, host_code: U32, host_message: String} 171 172 # Owns the database and its process lock. 173 type Handle is Type: 174 Handle{inner: DurableDb.Handle} 175 176 # Carries failures from scoped session execution. 177 type SessionError is Data: 178 OpenFailure{error: Error} 179 OperationFailure{error: Error} 180 CloseFailure{error: Error} 181 OperationAndCloseFailure{operation_error: Error, close_error: Error} 182 183 # Holds durable operations that thread one owned handle. 184 type Session is Kind(a <&> &1): 185 Session{run: Handle -> IO(Handle & Result<&1, &1, Error, A>)} 186 187 # Reports database size, cache use, errors, and maintenance state. 188 type DurableStats is Data: 189 DurableStats{active_bytes: Nat, wal_bytes: Nat, memtable_entries: Nat, memtable_payload_bytes: Nat, cache_entries: Nat, operation_errors: Nat, maintenance_blocked: Bool} 190 191 # Reports whether post-commit maintenance completed. 192 type MaintenanceStatus is Data: 193 MaintenanceComplete{} 194 MaintenancePending{error: Error} 195 196 # Reports maintenance status for a confirmed write. 197 type WriteOutcome is Data: 198 WriteOutcome{maintenance: MaintenanceStatus} 199 200 # Describes a public key/value update or deletion. 201 type Mutation is Data: 202 SetValue{key: String, value: String} 203 DeleteKey{key: String} 204 205 # Converts a public mutation to the WAL representation. 206 def to_storage_mut(mutation: Mutation) -> Wal.Mut: 207 match mutation: 208 case SetValue{key, value}: Wal.Put{key, value} 209 case DeleteKey{key}: Wal.Del{key} 210 211 # Converts each public mutation to its WAL representation. 212 def to_storage_muts(+mutations: List<&2, Mutation>) -> List<&2, Wal.Mut>: 213 match mutations: 214 case Nil{}: Nil{} 215 case Con{mutation, tail}: Con{to_storage_mut(mutation), to_storage_muts(tail)} 216 217 # Maps internal error categories to the public error kind. 218 def public_error_kind(kind: DurableError.ErrorKind) -> ErrorKind: 219 match kind: 220 case DurableError.NotFound{}: NotFoundError{} 221 case DurableError.AlreadyExists{}: AlreadyExistsError{} 222 case DurableError.Busy{}: BusyError{} 223 case DurableError.InvalidArgument{}: InvalidArgumentError{} 224 case DurableError.UnsupportedFormat{}: UnsupportedFormatError{} 225 case DurableError.Corruption{}: CorruptionError{} 226 case DurableError.ResourceLimit{}: ResourceLimitError{} 227 case DurableError.Io{}: IoError{} 228 case DurableError.CommitUnknown{}: CommitUnknownError{} 229 230 # Builds the public error while preserving its structured details. 231 def public_error(error: DurableError.Error) -> Error: 232 match error: 233 case DurableError.Error{kind, operation, path, host_code, host_message}: 234 Error{public_error_kind(kind), operation, path, host_code, host_message} 235 236 # Converts an internal result to the public error type. 237 def public_error_result(result: Result<&1, &1, DurableError.Error, Unit>) -> Result<&1, &1, Error, Unit>: 238 match result: 239 case Fail{error}: Fail{public_error(error)} 240 case Done{value}: Done{value} 241 242 # Converts a durable open result to the public handle. 243 def public_open_result(result: Result<&1, &1, DurableError.Error, DurableDb.Handle>) -> Result<&1, &1, Error, Handle>: 244 match result: 245 case Fail{error}: Fail{public_error(error)} 246 case Done{handle}: Done{Handle{handle}} 247 248 # Creates a database and returns its owned durable handle. 249 def create(+dir: String) -> IO(Result<&1, &1, Error, Handle>): 250 do IO>: 251 result : Result<&1, &1, DurableError.Error, DurableDb.Handle> <- DurableDb.create(dir) 252 return public_open_result(result) 253 254 # Opens and recovers an existing database under its lock. 255 def open_existing(+dir: String) -> IO(Result<&1, &1, Error, Handle>): 256 do IO>: 257 result : Result<&1, &1, DurableError.Error, DurableDb.Handle> <- DurableDb.open_existing(dir) 258 return public_open_result(result) 259 260 # Converts an optional stored value and its error. 261 def public_maybe_result( 262 result: Result<&1, &1, DurableError.Error, Maybe<&2, String>>, 263 ) -> Result<&1, &1, Error, Maybe<&2, String>>: 264 match result: 265 case Fail{error}: Fail{public_error(error)} 266 case Done{value}: Done{value} 267 268 # Converts a durable read result to the public value type. 269 def durable_get_result( 270 pair: DurableDb.Handle & Result<&1, &1, DurableError.Error, Maybe<&2, String>>, 271 ) -> Handle & Result<&1, &1, Error, Maybe<&2, String>>: 272 match pair: 273 case (handle, result): (Handle{handle}, public_maybe_result(result)) 274 275 # Reads a key from the owned durable database handle. 276 def get(handle: Handle, +key: String) -> IO(Handle & Result<&1, &1, Error, Maybe<&2, String>>): 277 match handle: 278 case Handle{inner}: 279 do IO>>: 280 pair : DurableDb.Handle & Result<&1, &1, DurableError.Error, Maybe<&2, String>> <- DurableDb.get(inner, key) 281 return durable_get_result(pair) 282 283 # Maps internal maintenance state to its public status. 284 def public_maintenance(maintenance: DurableDb.Maintenance) -> MaintenanceStatus: 285 match maintenance: 286 case DurableDb.Maintained{}: MaintenanceComplete{} 287 case DurableDb.Pending{error}: MaintenancePending{public_error(error)} 288 289 # Converts a confirmed write outcome to the public type. 290 def public_write_result( 291 result: Result<&1, &1, DurableError.Error, DurableDb.WriteOutcome>, 292 ) -> Result<&1, &1, Error, WriteOutcome>: 293 match result: 294 case Fail{error}: Fail{public_error(error)} 295 case Done{DurableDb.WriteOutcome{maintenance}}: Done{WriteOutcome{public_maintenance(maintenance)}} 296 297 # Preserves maintenance status while converting write errors. 298 def durable_write_result( 299 pair: DurableDb.Handle & Result<&1, &1, DurableError.Error, Unit>, 300 ) -> Handle & Result<&1, &1, Error, Unit>: 301 match pair: 302 case (handle, result): (Handle{handle}, public_error_result(result)) 303 304 # Converts a batch outcome while preserving handle ownership. 305 def durable_batch_result( 306 pair: DurableDb.Handle & Result<&1, &1, DurableError.Error, DurableDb.WriteOutcome>, 307 ) -> Handle & Result<&1, &1, Error, WriteOutcome>: 308 match pair: 309 case (handle, result): (Handle{handle}, public_write_result(result)) 310 311 # Validates and commits a public mutation batch atomically. 312 def write_batch(handle: Handle, mutations: List<&2, Mutation>) -> IO(Handle & Result<&1, &1, Error, WriteOutcome>): 313 match handle: 314 case Handle{inner}: 315 do IO>: 316 pair : DurableDb.Handle & Result<&1, &1, DurableError.Error, DurableDb.WriteOutcome> <- DurableDb.write_batch(inner, to_storage_muts(mutations)) 317 return durable_batch_result(pair) 318 319 # Writes one key and value through the durable handle. 320 def put(handle: Handle, +key: String, +value: String) -> IO(Handle & Result<&1, &1, Error, WriteOutcome>): 321 match handle: 322 case Handle{inner}: 323 do IO>: 324 pair : DurableDb.Handle & Result<&1, &1, DurableError.Error, DurableDb.WriteOutcome> <- DurableDb.put(inner, key, value) 325 return durable_batch_result(pair) 326 327 # Deletes one key through the durable handle. 328 def delete(handle: Handle, +key: String) -> IO(Handle & Result<&1, &1, Error, WriteOutcome>): 329 match handle: 330 case Handle{inner}: 331 do IO>: 332 pair : DurableDb.Handle & Result<&1, &1, DurableError.Error, DurableDb.WriteOutcome> <- DurableDb.delete(inner, key) 333 return durable_batch_result(pair) 334 335 # Flushes the durable handle and returns maintenance status. 336 def flush(handle: Handle) -> IO(Handle & Result<&1, &1, Error, Unit>): 337 match handle: 338 case Handle{inner}: 339 do IO>: 340 pair : DurableDb.Handle & Result<&1, &1, DurableError.Error, Unit> <- DurableDb.flush(inner) 341 return durable_write_result(pair) 342 343 # Compacts the durable handle and returns maintenance status. 344 def compact(handle: Handle) -> IO(Handle & Result<&1, &1, Error, Unit>): 345 match handle: 346 case Handle{inner}: 347 do IO>: 348 pair : DurableDb.Handle & Result<&1, &1, DurableError.Error, Unit> <- DurableDb.compact(inner) 349 return durable_write_result(pair) 350 351 # Maps internal counters to the public stats record. 352 def public_stats(result: Result<&1, &1, DurableError.Error, DurableDb.Stats>) -> Result<&1, &1, Error, DurableStats>: 353 match result: 354 case Fail{error}: Fail{public_error(error)} 355 case Done{DurableDb.Stats{active_bytes, wal_bytes, memtable_entries, memtable_payload_bytes, cache_entries, operation_errors, state}}: 356 match state: 357 case DurableDbPolicy.MaintenanceBlocked{}: Done{DurableStats{active_bytes, wal_bytes, memtable_entries, memtable_payload_bytes, cache_entries, operation_errors, True{}}} 358 case _: Done{DurableStats{active_bytes, wal_bytes, memtable_entries, memtable_payload_bytes, cache_entries, operation_errors, False{}}} 359 360 # Converts durable stats errors to the public error type. 361 def durable_stats_result( 362 pair: DurableDb.Handle & Result<&1, &1, DurableError.Error, DurableDb.Stats>, 363 ) -> Handle & Result<&1, &1, Error, DurableStats>: 364 match pair: 365 case (handle, result): (Handle{handle}, public_stats(result)) 366 367 # Reads bounded operational stats from the durable handle. 368 def stats(handle: Handle) -> IO(Handle & Result<&1, &1, Error, DurableStats>): 369 match handle: 370 case Handle{inner}: 371 do IO>: 372 pair : DurableDb.Handle & Result<&1, &1, DurableError.Error, DurableDb.Stats> <- DurableDb.stats(inner) 373 return durable_stats_result(pair) 374 375 # Closes the durable handle and releases its database lock. 376 def close(handle: Handle) -> IO(Result<&1, &1, Error, Unit>): 377 match handle: 378 case Handle{inner}: 379 do IO>: 380 result : Result<&1, &1, DurableError.Error, Unit> <- DurableDb.close(inner) 381 return public_error_result(result) 382 383 def Session.pure(a, -A: Kind(a), value: A) -> Session: 384 Session{handle => IO.pure(Handle & Result<&1, &1, Error, A>, (handle, Done{value}))} 385 386 def Session.run(a, -A: Kind(a), session: Session, handle: Handle) -> IO(Handle & Result<&1, &1, Error, A>): 387 match session: 388 case Session{run}: run(handle) 389 390 def Session.bind_result(a, -A: Kind(a), -B: Kind(a), pair: Handle & Result<&1, &1, Error, A>, next: A -> Session) -> IO(Handle & Result<&1, &1, Error, B>): 391 match pair: 392 case (handle, Fail{error}): 393 IO.pure(Handle & Result<&1, &1, Error, B>, (handle, Fail{error})) 394 case (handle, Done{value}): 395 Session.run(a, B, next(value), handle) 396 397 def Session.bind(a, -A: Kind(a), -B: Kind(a), session: Session, next: A -> Session) -> Session: 398 Session{handle => 399 do IO>: 400 pair : Handle & Result<&1, &1, Error, A> <- Session.run(a, A, session, handle) 401 Session.bind_result(a, A, B, pair, next) 402 } 403 404 def Session.put(+key: String, +value: String) -> Session<&2, WriteOutcome>: 405 Session{handle => put(handle, key, value)} 406 407 def Session.delete(+key: String) -> Session<&2, WriteOutcome>: 408 Session{handle => delete(handle, key)} 409 410 def Session.write_batch(mutations: List<&2, Mutation>) -> Session<&2, WriteOutcome>: 411 Session{handle => write_batch(handle, mutations)} 412 413 def Session.get(+key: String) -> Session<&2, Maybe<&2, String>>: 414 Session{handle => get(handle, key)} 415 416 def Session.flush() -> Session<&2, Unit>: 417 Session{handle => flush(handle)} 418 419 def Session.compact() -> Session<&2, Unit>: 420 Session{handle => compact(handle)} 421 422 def Session.stats() -> Session<&2, DurableStats>: 423 Session{handle => stats(handle)} 424 425 def Session.finish_result(a, -A: Kind(a), result: Result<&1, &1, Error, A>, close_result: Result<&1, &1, Error, Unit>) -> Result<&1, &1, SessionError, A>: 426 match result close_result: 427 case Done{value} Done{Unit{}}: Done{value} 428 case Fail{error} Done{Unit{}}: Fail{OperationFailure{error}} 429 case Done{_} Fail{error}: Fail{CloseFailure{error}} 430 case Fail{operation_error} Fail{close_error}: 431 Fail{OperationAndCloseFailure{operation_error, close_error}} 432 433 def Session.finish(a, -A: Kind(a), handle: Handle, result: Result<&1, &1, Error, A>) -> IO(Result<&1, &1, SessionError, A>): 434 do IO>: 435 close_result : Result<&1, &1, Error, Unit> <- close(handle) 436 return Session.finish_result(a, A, result, close_result) 437 438 def Session.finish_pair(a, -A: Kind(a), pair: Handle & Result<&1, &1, Error, A>) -> IO(Result<&1, &1, SessionError, A>): 439 match pair: 440 case (handle, result): Session.finish(a, A, handle, result) 441 442 def Session.run_opened(a, -A: Kind(a), opened: Result<&1, &1, Error, Handle>, session: Session) -> IO(Result<&1, &1, SessionError, A>): 443 match opened: 444 case Fail{error}: 445 IO.pure(Result<&1, &1, SessionError, A>, Fail{OpenFailure{error}}) 446 case Done{handle}: 447 do IO>: 448 pair : Handle & Result<&1, &1, Error, A> <- Session.run(a, A, session, handle) 449 Session.finish_pair(a, A, pair) 450 451 # Runs a program and returns its still-owned handle. 452 def run_session(a, -A: Kind(a), handle: Handle, session: Session) -> IO(Handle & Result<&1, &1, Error, A>): 453 Session.run(a, A, session, handle) 454 455 # Creates, runs, and closes a scoped durable session. 456 def create_database(a, -A: Kind(a), +path: String, session: Session) -> IO(Result<&1, &1, SessionError, A>): 457 do IO>: 458 opened : Result<&1, &1, Error, Handle> <- create(path) 459 Session.run_opened(a, A, opened, session) 460 461 # Opens, runs, and closes a scoped durable session. 462 def open_database(a, -A: Kind(a), +path: String, session: Session) -> IO(Result<&1, &1, SessionError, A>): 463 do IO>: 464 opened : Result<&1, &1, Error, Handle> <- open_existing(path) 465 Session.run_opened(a, A, opened, session) 466 467 # Checks whether the public error means the path exists. 468 def error_is_already_exists(error: Error) -> Bool: 469 match error: 470 case Error{AlreadyExistsError{}, _, _, _, _}: True{} 471 case _: False{} 472 473 # Checks whether the public error means the database is locked. 474 def error_is_busy(error: Error) -> Bool: 475 match error: 476 case Error{BusyError{}, _, _, _, _}: True{} 477 case _: False{}