1 import Base 2 import ../DbLock.bend as DbLock 3 import ../DurableError.bend as DurableError 4 import ../Fs.bend as Fs 5 import ../legacy/ManifestIo.bend as LegacyManifestIo 6 import ../Manifest.bend as Manifest 7 import ../MemTable.bend as MemTable 8 import ../Sstable.bend as Sstable 9 import ../SstStreamIo.bend as SstStreamIo 10 import ../Wal.bend as Wal 11 import ./LegacyWal.bend as LegacyWal 12 import ../SstReadPolicy.bend as SstReadPolicy 13 import ../coordinator/QueuePolicy.bend as QueuePolicy 14 import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes 15 16 # Separates lock, host, integrity, and resource failures for read-only validation. 17 type Error is Data: 18 LockError{error: DurableError.Error} 19 HostError{code: U32, message: String} 20 InvalidLegacyWal{} 21 InvalidLegacyManifest{} 22 InvalidLegacySst{} 23 ResourceLimit{} 24 25 # Holds source ownership until read-only validation finishes. 26 type TableValidation is Type: 27 TableValidation{lock: DbLock.Lock, path: String, 28 result: Result<&1, &1, Error, Nat & Nat>, 29 records: List<&2, MemTable.Entry>, batches: List<&2, Wal.Batch>} 30 31 # Reports the SST record and legacy WAL frame counts from one locked view. 32 type SourceCounts is Data: 33 LegacySourceCounts{sst_entries: Nat, wal_frames: Nat, encoded_bytes: Nat} 34 35 # Carries the verified SST and replayed WAL records from one locked view. 36 type SourceSnapshot is Data: 37 LegacySourceSnapshot{sst_records: List<&2, MemTable.Entry>, 38 wal_batches: List<&2, Wal.Batch>, sst_entries: Nat, wal_frames: Nat, 39 encoded_bytes: Nat} 40 41 # Carries one legacy SST through size admission, decoding, and entry counting. 42 type LegacyTableWalk is Type: 43 LegacyLevels{lock: DbLock.Lock, read_path: String, release_path: String, 44 levels: List<&2, List<&2, String>>, index: Nat, total_bytes: Nat, 45 entries: Nat, records: List<&2, MemTable.Entry>} 46 LegacyPath{lock: DbLock.Lock, read_path: String, release_path: String, 47 ldir: Maybe<&2, String>, names: List<&2, String>, 48 levels: List<&2, List<&2, String>>, index: Nat, total_bytes: Nat, 49 entries: Nat, records: List<&2, MemTable.Entry>} 50 LegacyNames{lock: DbLock.Lock, read_path: String, release_path: String, 51 levels: List<&2, List<&2, String>>, index: Nat, dir: String, 52 names: List<&2, String>, total_bytes: Nat, entries: Nat, 53 records: List<&2, MemTable.Entry>} 54 LegacySize{lock: DbLock.Lock, read_path: String, release_path: String, 55 levels: List<&2, List<&2, String>>, index: Nat, dir: String, 56 name: String, names: List<&2, String>, total_bytes: Nat, entries: Nat, 57 result: Result<&1, &1, U32 & String, Nat>, records: List<&2, MemTable.Entry>} 58 LegacyBudget{lock: DbLock.Lock, read_path: String, release_path: String, 59 levels: List<&2, List<&2, String>>, index: Nat, dir: String, 60 name: String, names: List<&2, String>, entries: Nat, 61 decision: SstReadPolicy.TotalDecision, records: List<&2, MemTable.Entry>} 62 LegacyRead{lock: DbLock.Lock, read_path: String, release_path: String, 63 levels: List<&2, List<&2, String>>, index: Nat, dir: String, 64 names: List<&2, String>, total_bytes: Nat, entries: Nat, 65 result: Result<&1, &1, U32 & String, Sstable.Table>, 66 records: List<&2, MemTable.Entry>} 67 68 def path.clone.pair.go( 69 path: String, 70 left: String, 71 right: String 72 ) -> String & String: 73 match path: 74 case SNil{}: 75 (String.reverse(left), String.reverse(right)) 76 case SCon{+character, rest}: 77 path.clone.pair.go(rest, SCon{character, left}, SCon{character, right}) 78 79 def path.clone.pair(path: String) -> String & String: 80 path.clone.pair.go(path, SNil{}, SNil{}) 81 82 def table.validation( 83 lock: DbLock.Lock, 84 path: String, 85 result: Result<&1, &1, Error, Nat & Nat> 86 ) -> TableValidation: 87 TableValidation{lock, path, result, Nil{}, Nil{}} 88 89 def table.validation.records( 90 lock: DbLock.Lock, 91 path: String, 92 result: Result<&1, &1, Error, Nat & Nat>, 93 records: List<&2, MemTable.Entry> 94 ) -> TableValidation: 95 TableValidation{lock, path, result, records, Nil{}} 96 97 def table.validation.snapshot( 98 lock: DbLock.Lock, 99 path: String, 100 result: Result<&1, &1, Error, Nat & Nat>, 101 records: List<&2, MemTable.Entry>, 102 batches: List<&2, Wal.Batch> 103 ) -> TableValidation: 104 TableValidation{lock, path, result, records, batches} 105 106 def table.release.result( 107 value: Result<&1, &1, Error, Nat & Nat>, 108 result: Result<&1, &1, DurableError.Error, Unit> 109 ) -> Result<&1, &1, Error, Nat & Nat>: 110 match result: 111 case Fail{error}: Fail{LockError{error}} 112 case Done{Unit{}}: value 113 114 def table.release(state: TableValidation) -> IO(Result<&1, &1, Error, Nat & Nat>): 115 match state: 116 case TableValidation{lock, path, result, _, _}: 117 do IO>: 118 released : Result<&1, &1, DurableError.Error, Unit> <- 119 DbLock.release(lock, "migrate-verify", path) 120 return table.release.result(result, released) 121 122 def table.finished( 123 lock: DbLock.Lock, 124 path: String, 125 result: Result<&1, &1, Error, Nat & Nat> 126 ) -> IO(TableValidation): 127 IO.pure(TableValidation, table.validation(lock, path, result)) 128 129 def table.finished.records( 130 lock: DbLock.Lock, 131 path: String, 132 result: Result<&1, &1, Error, Nat & Nat>, 133 records: List<&2, MemTable.Entry> 134 ) -> IO(TableValidation): 135 IO.pure(TableValidation, table.validation.records(lock, path, result, records)) 136 137 def table.entries.count.keep.go( 138 entries: List<&2, MemTable.Entry>, 139 reversed: List<&2, MemTable.Entry>, 140 count: Nat 141 ) -> List<&2, MemTable.Entry> & Nat: 142 match entries: 143 case Nil{}: (List.reverse(&2, MemTable.Entry, reversed), count) 144 case Con{entry, tail}: 145 table.entries.count.keep.go(tail, Con{entry, reversed}, Nat.add(count, 1n)) 146 147 def table.entries.count.keep( 148 entries: List<&2, MemTable.Entry> 149 ) -> List<&2, MemTable.Entry> & Nat: 150 table.entries.count.keep.go(entries, Nil{}, 0n) 151 152 def table.read.next( 153 read: List<&2, MemTable.Entry> & Nat, 154 lock: DbLock.Lock, 155 read_path: String, 156 release_path: String, 157 levels: List<&2, List<&2, String>>, 158 index: Nat, 159 dir: String, 160 names: List<&2, String>, 161 total_bytes: Nat, 162 prior_entries: Nat, 163 prior_records: List<&2, MemTable.Entry> 164 ) -> LegacyTableWalk: 165 match read: 166 case (table_entries, table_count): 167 LegacyNames{lock, read_path, release_path, levels, index, dir, 168 names, total_bytes, Nat.add(prior_entries, table_count), 169 List.append(&2, MemTable.Entry, prior_records, table_entries)} 170 171 def table.failure(code: U32) -> Error: 172 Bool.pick(Error, U32.is_eq(code, SstStreamIo.resource_limit_error_code()), 173 ResourceLimit{}, InvalidLegacySst{}) 174 175 def table.manifest.measure( 176 levels: List<&2, List<&2, String>>, 177 index: Nat, 178 acc: List<&2, List<&2, String>> 179 ) -> List<&2, List<&2, String>> & Nat: 180 match levels: 181 case Nil{}: (List.reverse(&2, List<&2, String>, acc), index) 182 case Con{level, tail}: 183 table.manifest.measure(tail, Nat.add(index, 1n), Con{level, acc}) 184 185 def table.walk.size( 186 paths: (String & String) & ((String & String) & (String & String)), 187 lock: DbLock.Lock, 188 release_path: String, 189 levels: List<&2, List<&2, String>>, 190 index: Nat, 191 names: List<&2, String>, 192 total_bytes: Nat, 193 entries: Nat, 194 records: List<&2, MemTable.Entry> 195 ) -> IO(LegacyTableWalk): 196 match paths: 197 case ((file_root, read_path), ((file_name, name), (file_dir, dir))): 198 do IO: 199 result : Result<&1, &1, U32 & String, Nat> <- 200 Fs.file_size(file_root ++ "/" ++ file_dir ++ "/" ++ file_name) 201 return LegacySize{lock, read_path, release_path, levels, index, 202 dir, name, names, total_bytes, entries, result, records} 203 204 def table.walk.decode( 205 paths: (String & String) & (String & String), 206 lock: DbLock.Lock, 207 release_path: String, 208 levels: List<&2, List<&2, String>>, 209 index: Nat, 210 name: String, 211 names: List<&2, String>, 212 total_bytes: Nat, 213 entries: Nat, 214 records: List<&2, MemTable.Entry> 215 ) -> IO(LegacyTableWalk): 216 match paths: 217 case ((file_root, read_path), (file_dir, dir)): 218 do IO: 219 result : Result<&1, &1, U32 & String, Sstable.Table> <- 220 SstStreamIo.read_table_limited( 221 file_root ++ "/" ++ file_dir ++ "/" ++ name, 67108864n) 222 return LegacyRead{lock, read_path, release_path, levels, index, 223 dir, names, total_bytes, entries, result, records} 224 225 def table.walk.level.path( 226 indices: Nat & Nat, 227 lock: DbLock.Lock, 228 read_path: String, 229 release_path: String, 230 names: List<&2, String>, 231 levels: List<&2, List<&2, String>>, 232 total_bytes: Nat, 233 entries: Nat, 234 records: List<&2, MemTable.Entry> 235 ) -> LegacyTableWalk: 236 match indices: 237 case (lookup_index, next_index): 238 LegacyPath{lock, read_path, release_path, 239 LegacyManifestIo.ldir_of(lookup_index), names, levels, next_index, 240 total_bytes, entries, records} 241 242 def table.walk.go(fuel: Nat, state: LegacyTableWalk) -> IO(TableValidation): 243 match fuel: 244 case 0n: 245 match state: 246 case LegacyLevels{lock, _, release_path, _, _, _, _, records}: 247 table.finished(lock, release_path, Fail{ResourceLimit{}}) 248 case LegacyPath{lock, _, release_path, _, _, _, _, _, _, records}: 249 table.finished(lock, release_path, Fail{ResourceLimit{}}) 250 case LegacyNames{lock, _, release_path, _, _, _, _, _, _, records}: 251 table.finished(lock, release_path, Fail{ResourceLimit{}}) 252 case LegacySize{lock, _, release_path, _, _, _, _, _, _, _, _, records}: 253 table.finished(lock, release_path, Fail{ResourceLimit{}}) 254 case LegacyBudget{lock, _, release_path, _, _, _, _, _, _, _, records}: 255 table.finished(lock, release_path, Fail{ResourceLimit{}}) 256 case LegacyRead{lock, _, release_path, _, _, _, _, _, _, _, records}: 257 table.finished(lock, release_path, Fail{ResourceLimit{}}) 258 case 1n+rest: 259 match state: 260 case LegacyLevels{lock, read_path, release_path, levels, index, 261 total_bytes, entries, records}: 262 match levels: 263 case Nil{}: 264 table.finished.records(lock, release_path, 265 Done{(entries, total_bytes)}, records) 266 case Con{names, tail}: 267 table.walk.go(rest, table.walk.level.path( 268 QueuePolicy.nat.clone.pair(index), lock, read_path, 269 release_path, names, tail, total_bytes, entries, records)) 270 case LegacyPath{lock, read_path, release_path, ldir, names, levels, 271 index, total_bytes, entries, records}: 272 match ldir: 273 case None{}: 274 table.finished.records(lock, release_path, 275 Fail{InvalidLegacyManifest{}}, records) 276 case Some{dir}: 277 table.walk.go(rest, LegacyNames{lock, read_path, release_path, 278 levels, index, dir, names, total_bytes, entries, records}) 279 case LegacyNames{lock, read_path, release_path, levels, index, dir, 280 names, total_bytes, entries, records}: 281 match names: 282 case Nil{}: 283 table.walk.go(rest, LegacyLevels{lock, read_path, release_path, 284 levels, Nat.add(index, 1n), total_bytes, entries, records}) 285 case Con{name, tail}: 286 do IO: 287 next : LegacyTableWalk <- table.walk.size( 288 (path.clone.pair(read_path), 289 (path.clone.pair(name), path.clone.pair(dir))), 290 lock, release_path, levels, index, tail, 291 total_bytes, entries, records) 292 table.walk.go(rest, next) 293 case LegacySize{lock, read_path, release_path, levels, index, dir, 294 name, names, total_bytes, entries, result, records}: 295 match result: 296 case Fail{(code, message)}: 297 table.finished.records(lock, release_path, 298 Fail{HostError{code, message}}, records) 299 case Done{size}: 300 table.walk.go(rest, LegacyBudget{lock, read_path, release_path, 301 levels, index, dir, name, names, entries, 302 SstReadPolicy.total_decision(total_bytes, size, 67108864n), 303 records}) 304 case LegacyBudget{lock, read_path, release_path, levels, index, dir, 305 name, names, entries, decision, records}: 306 match decision: 307 case SstReadPolicy.TotalRejected{}: 308 table.finished.records(lock, release_path, 309 Fail{ResourceLimit{}}, records) 310 case SstReadPolicy.TotalAdmitted{next_total}: 311 do IO: 312 next : LegacyTableWalk <- table.walk.decode( 313 (path.clone.pair(read_path), path.clone.pair(dir)), lock, 314 release_path, levels, index, name, names, next_total, 315 entries, records) 316 table.walk.go(rest, next) 317 case LegacyRead{lock, read_path, release_path, levels, index, dir, 318 names, total_bytes, entries, result, records}: 319 match result: 320 case Fail{(code, _)}: 321 table.finished.records(lock, release_path, 322 Fail{table.failure(code)}, records) 323 case Done{Sstable.Tbl{table_entries, _, _, _, _, _, _, _}}: 324 table.walk.go(rest, table.read.next( 325 table.entries.count.keep(table_entries), lock, read_path, 326 release_path, levels, index, dir, names, total_bytes, 327 entries, records)) 328 329 def table.walk.start.paths( 330 paths: String & String, 331 lock: DbLock.Lock, 332 release_path: String, 333 measured: List<&2, List<&2, String>> & Nat 334 ) -> IO(TableValidation): 335 match paths: 336 case (read_path, _): 337 match measured: 338 case (levels, _): 339 table.walk.go(4194305n, LegacyLevels{lock, read_path, 340 release_path, levels, 0n, 0n, 0n, Nil{}}) 341 342 def table.walk.start( 343 lock: DbLock.Lock, 344 paths: String & String, 345 measured: List<&2, List<&2, String>> & Nat 346 ) -> IO(TableValidation): 347 match paths: 348 case (read_path, release_path): 349 table.walk.start.paths(path.clone.pair(read_path), lock, release_path, 350 measured) 351 352 def table.manifest.loaded( 353 state: TableValidation, 354 result: Result<&1, &1, U32 & String, Manifest.Manifest> 355 ) -> IO(TableValidation): 356 match state: 357 case TableValidation{lock, path, _, _, _}: 358 match result: 359 case Fail{_}: 360 table.finished(lock, path, Fail{InvalidLegacyManifest{}}) 361 case Done{Manifest.M{levels}}: 362 table.walk.start(lock, path.clone.pair(path), 363 table.manifest.measure(levels, 0n, Nil{})) 364 365 def table.manifest.read( 366 lock: DbLock.Lock, 367 paths: String & String 368 ) -> IO(TableValidation): 369 match paths: 370 case (read_path, retained_path): 371 do IO: 372 manifest : Result<&1, &1, U32 & String, Manifest.Manifest> <- 373 LegacyManifestIo.load_mfst(read_path ++ "/MANIFEST") 374 table.manifest.loaded( 375 table.validation(lock, retained_path, Done{(0n, 0n)}), manifest) 376 377 def table.manifest.present( 378 lock: DbLock.Lock, 379 path: String, 380 result: Result<&1, &1, U32 & String, Bool> 381 ) -> IO(TableValidation): 382 match result: 383 case Fail{(code, message)}: 384 table.finished(lock, path, Fail{HostError{code, message}}) 385 case Done{False{}}: 386 table.finished(lock, path, Fail{InvalidLegacyManifest{}}) 387 case Done{True{}}: 388 do IO: 389 paths : String & String = path.clone.pair(path) 390 table.manifest.read(lock, paths) 391 392 def table.locked.manifest( 393 lock: DbLock.Lock, 394 paths: String & String 395 ) -> IO(Result<&1, &1, Error, Nat & Nat>): 396 match paths: 397 case (exists_path, retained_path): 398 do IO>: 399 present : Result<&1, &1, U32 & String, Bool> <- 400 Fs.exists(exists_path ++ "/MANIFEST") 401 validation : TableValidation <- 402 table.manifest.present(lock, retained_path, present) 403 table.release(validation) 404 405 def table.locked( 406 path: String, 407 result: Result<&1, &1, DurableError.Error, DbLock.Lock> 408 ) -> IO(Result<&1, &1, Error, Nat & Nat>): 409 match result: 410 case Fail{error}: 411 IO.pure(Result<&1, &1, Error, Nat & Nat>, Fail{LockError{error}}) 412 case Done{lock}: 413 do IO>: 414 paths : String & String = path.clone.pair(path) 415 table.locked.manifest(lock, paths) 416 417 def table.lock.acquired( 418 retained_path: String, 419 result: Result<&1, &1, DurableError.Error, DbLock.Lock> 420 ) -> IO(Result<&1, &1, Error, Nat & Nat>): 421 table.locked(retained_path, result) 422 423 def table.lock.present.start(paths: String & String) -> IO(Result<&1, &1, Error, Nat & Nat>): 424 match paths: 425 case (lock_path, retained_path): 426 do IO>: 427 acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <- 428 DbLock.acquire(lock_path, "migrate-verify") 429 table.lock.acquired(retained_path, acquired) 430 431 def table.lock.present( 432 path: String, 433 result: Result<&1, &1, U32 & String, Bool> 434 ) -> IO(Result<&1, &1, Error, Nat & Nat>): 435 match result: 436 case Fail{(code, message)}: 437 IO.pure(Result<&1, &1, Error, Nat & Nat>, Fail{HostError{code, message}}) 438 case Done{False{}}: 439 IO.pure(Result<&1, &1, Error, Nat & Nat>, 440 Fail{HostError{2, "database LOCK file is missing"}}) 441 case Done{True{}}: 442 do IO>: 443 paths : String & String = path.clone.pair(path) 444 table.lock.present.start(paths) 445 446 def table.canonical.lock( 447 paths: String & String 448 ) -> IO(Result<&1, &1, Error, Nat & Nat>): 449 match paths: 450 case (exists_path, retained_path): 451 do IO>: 452 present : Result<&1, &1, U32 & String, Bool> <- 453 Fs.exists(exists_path ++ "/LOCK") 454 table.lock.present(retained_path, present) 455 456 def table.canonical( 457 result: Result<&1, &1, U32 & String, String> 458 ) -> IO(Result<&1, &1, Error, Nat & Nat>): 459 match result: 460 case Fail{(code, message)}: 461 IO.pure(Result<&1, &1, Error, Nat & Nat>, Fail{HostError{code, message}}) 462 case Done{path}: 463 do IO>: 464 paths : String & String = path.clone.pair(path) 465 table.canonical.lock(paths) 466 467 # Validates legacy manifest references and SST contents while holding the source lock. 468 def table.count.result( 469 result: Result<&1, &1, Error, Nat & Nat> 470 ) -> Result<&1, &1, Error, Nat>: 471 match result: 472 case Fail{error}: Fail{error} 473 case Done{(entries, _)}: Done{entries} 474 475 # Validates legacy SST files and reports their record count under one lock. 476 def verify_legacy_tables(path: String) -> IO(Result<&1, &1, Error, Nat>): 477 do IO>: 478 canonical : Result<&1, &1, U32 & String, String> <- Fs.canonical_path(path) 479 verified : Result<&1, &1, Error, Nat & Nat> <- table.canonical(canonical) 480 return table.count.result(verified) 481 482 # Bounds memory used while validating a legacy WAL file. 483 def max_wal_bytes() -> Nat: 484 67108864n 485 486 # Preserves the primary validation result unless lock release fails. 487 def release.result( 488 value: Result<&1, &1, Error, Nat>, 489 released: Result<&1, &1, DurableError.Error, Unit> 490 ) -> Result<&1, &1, Error, Nat>: 491 match released: 492 case Fail{error}: Fail{LockError{error}} 493 case Done{Unit{}}: value 494 495 # Releases database ownership before returning validation status. 496 def release( 497 lock: DbLock.Lock, 498 value: Result<&1, &1, Error, Nat>, 499 path: String 500 ) -> IO(Result<&1, &1, Error, Nat>): 501 do IO>: 502 released : Result<&1, &1, DurableError.Error, Unit> <- 503 DbLock.release(lock, "migrate-verify", path) 504 return release.result(value, released) 505 506 # Converts a raw host failure after releasing the source lock. 507 def host.failed( 508 lock: DbLock.Lock, 509 error: U32 & String, 510 path: String 511 ) -> IO(Result<&1, &1, Error, Nat>): 512 match error: 513 case (code, message): 514 release(lock, Fail{HostError{code, message}}, path) 515 516 # Converts a host failure that occurred before acquiring the source lock. 517 def host.without_lock(error: U32 & String) -> IO(Result<&1, &1, Error, Nat>): 518 match error: 519 case (code, message): 520 IO.pure(Result<&1, &1, Error, Nat>, Fail{HostError{code, message}}) 521 522 def wal.read.failure( 523 error: U32 & String, 524 released: Result<&1, &1, DurableError.Error, Unit> 525 ) -> Result<&1, &1, Error, Nat>: 526 match error: 527 case (code, message): 528 match released: 529 case Fail{lock_error}: Fail{LockError{lock_error}} 530 case Done{Unit{}}: Fail{HostError{code, message}} 531 532 def wal.classify(result: LegacyWal.VerifyResult) -> Result<&1, &1, Error, Nat>: 533 match result: 534 case LegacyWal.WalValid{frames}: Done{frames} 535 case LegacyWal.WalInvalid{}: Fail{InvalidLegacyWal{}} 536 537 def wal.read.success( 538 bytes: Bytes.Bytes, 539 released: Result<&1, &1, DurableError.Error, Unit> 540 ) -> Result<&1, &1, Error, Nat>: 541 match released: 542 case Fail{error}: Fail{LockError{error}} 543 case Done{Unit{}}: wal.classify(LegacyWal.verify(bytes)) 544 545 # Closes the WAL descriptor before classifying its bytes. 546 def wal.read.dispatch( 547 result: Result<&1, &1, U32 & String, Bytes.Bytes>, 548 released: Result<&1, &1, DurableError.Error, Unit> 549 ) -> Result<&1, &1, Error, Nat>: 550 match result: 551 case Fail{error}: wal.read.failure(error, released) 552 case Done{bytes}: wal.read.success(bytes, released) 553 554 def wal.read.closed( 555 file: File, 556 result: Result<&1, &1, U32 & String, Bytes.Bytes>, 557 lock: DbLock.Lock, 558 path: String 559 ) -> IO(Result<&1, &1, Error, Nat>): 560 do IO>: 561 _closed : Unit <- File.close(file) 562 released : Result<&1, &1, DurableError.Error, Unit> <- 563 DbLock.release(lock, "migrate-verify", path) 564 return wal.read.dispatch(result, released) 565 566 # Routes the owned WAL file into its read result handler. 567 def wal.read.pair( 568 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>, 569 lock: DbLock.Lock, 570 path: String 571 ) -> IO(Result<&1, &1, Error, Nat>): 572 match pair: 573 case (file, result): wal.read.closed(file, result, lock, path) 574 575 # Reads a bounded WAL file while retaining the exclusive source lock. 576 def wal.opened( 577 lock: DbLock.Lock, 578 size: Nat, 579 opened: Result<&1, &1, U32 & String, File>, 580 path: String 581 ) -> IO(Result<&1, &1, Error, Nat>): 582 match opened: 583 case Fail{error}: host.failed(lock, error, path) 584 case Done{file}: 585 do IO>: 586 read : File & Result<&1, &1, U32 & String, Bytes.Bytes> <- 587 Fs.read_at(file, 0, U32.from_nat(size)) 588 wal.read.pair(read, lock, path) 589 590 # Rejects oversized WAL input before opening its data file. 591 def wal.sized( 592 lock: DbLock.Lock, 593 size: Nat, 594 +path: String, 595 within_limit: Bool 596 ) -> IO(Result<&1, &1, Error, Nat>): 597 match within_limit: 598 case False{}: release(lock, Fail{ResourceLimit{}}, path) 599 case True{}: 600 do IO>: 601 opened : Result<&1, &1, U32 & String, File> <- 602 File.open(path ++ "/WAL", "r") 603 wal.opened(lock, size, opened, path) 604 605 # Checks file size and rejects inputs beyond the migration memory limit. 606 def wal.size.result( 607 lock: DbLock.Lock, 608 result: Result<&1, &1, U32 & String, Nat>, 609 path: String 610 ) -> IO(Result<&1, &1, Error, Nat>): 611 match result: 612 case Fail{error}: host.failed(lock, error, path) 613 case Done{+size}: 614 wal.sized(lock, size, path, Nat.is_le(size, max_wal_bytes())) 615 616 # Treats a missing WAL as an empty legacy mutation stream. 617 def wal.present( 618 lock: DbLock.Lock, 619 present: Bool, 620 +path: String 621 ) -> IO(Result<&1, &1, Error, Nat>): 622 match present: 623 case False{}: release(lock, Done{0n}, path) 624 case True{}: 625 do IO>: 626 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path ++ "/WAL") 627 wal.size.result(lock, size, path) 628 629 # Requires the legacy WAL path to be readable without changing it. 630 def wal.exists( 631 lock: DbLock.Lock, 632 result: Result<&1, &1, U32 & String, Bool>, 633 path: String 634 ) -> IO(Result<&1, &1, Error, Nat>): 635 match result: 636 case Fail{error}: host.failed(lock, error, path) 637 case Done{present}: wal.present(lock, present, path) 638 639 # Continues source verification after exclusive lock acquisition. 640 def locked( 641 +path: String, 642 result: Result<&1, &1, DurableError.Error, DbLock.Lock> 643 ) -> IO(Result<&1, &1, Error, Nat>): 644 match result: 645 case Fail{error}: 646 IO.pure(Result<&1, &1, Error, Nat>, Fail{LockError{error}}) 647 case Done{lock}: 648 do IO>: 649 present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path ++ "/WAL") 650 wal.exists(lock, present, path) 651 652 # Requires an existing stable lock file and acquires it before reading the WAL. 653 def lock.present( 654 +path: String, 655 result: Result<&1, &1, U32 & String, Bool> 656 ) -> IO(Result<&1, &1, Error, Nat>): 657 match result: 658 case Fail{error}: host.without_lock(error) 659 case Done{False{}}: 660 IO.pure(Result<&1, &1, Error, Nat>, Fail{HostError{2, "database LOCK file is missing"}}) 661 case Done{True{}}: 662 do IO>: 663 acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <- 664 DbLock.acquire(path, "migrate-verify") 665 locked(path, acquired) 666 667 # Converts a canonicalization error into the migration IO error domain. 668 def canonical.result( 669 result: Result<&1, &1, U32 & String, String> 670 ) -> IO(Result<&1, &1, Error, Nat>): 671 match result: 672 case Fail{(code, message)}: 673 IO.pure(Result<&1, &1, Error, Nat>, Fail{HostError{code, message}}) 674 case Done{+path}: 675 do IO>: 676 present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path ++ "/LOCK") 677 lock.present(path, present) 678 679 def wal.batches.measure.go( 680 batches: List<&2, Wal.Batch>, 681 reversed: List<&2, Wal.Batch>, 682 count: Nat 683 ) -> List<&2, Wal.Batch> & Nat: 684 match batches: 685 case Nil{}: (List.reverse(&2, Wal.Batch, reversed), count) 686 case Con{batch, tail}: 687 wal.batches.measure.go(tail, Con{batch, reversed}, Nat.add(count, 1n)) 688 689 def wal.batches.measure(batches: List<&2, Wal.Batch>) -> List<&2, Wal.Batch> & Nat: 690 wal.batches.measure.go(batches, Nil{}, 0n) 691 692 def wal.locked.valid.batches( 693 measured: List<&2, Wal.Batch> & Nat, 694 lock: DbLock.Lock, 695 path: String, 696 size: Nat, 697 records: List<&2, MemTable.Entry> 698 ) -> TableValidation: 699 match measured: 700 case (batches, frames): 701 table.validation.snapshot(lock, path, 702 Done{(frames, size)}, records, batches) 703 704 def wal.locked.classified( 705 result: LegacyWal.ReplayResult, 706 lock: DbLock.Lock, 707 path: String, 708 size: Nat, 709 records: List<&2, MemTable.Entry> 710 ) -> TableValidation: 711 match result: 712 case LegacyWal.ReplayInvalid{}: 713 table.validation.records(lock, path, Fail{InvalidLegacyWal{}}, records) 714 case LegacyWal.ReplayValid{batches}: 715 wal.locked.valid.batches(wal.batches.measure(batches), lock, path, 716 size, records) 717 718 def wal.locked.read.classify( 719 result: Result<&1, &1, U32 & String, Bytes.Bytes>, 720 lock: DbLock.Lock, 721 path: String, 722 size: Nat, 723 records: List<&2, MemTable.Entry> 724 ) -> TableValidation: 725 match result: 726 case Fail{(code, message)}: 727 table.validation.records(lock, path, Fail{HostError{code, message}}, records) 728 case Done{bytes}: 729 wal.locked.classified(LegacyWal.replay(bytes), lock, path, size, records) 730 731 def wal.locked.read.closed( 732 file: File, 733 result: Result<&1, &1, U32 & String, Bytes.Bytes>, 734 lock: DbLock.Lock, 735 path: String, 736 size: Nat, 737 records: List<&2, MemTable.Entry> 738 ) -> IO(TableValidation): 739 do IO: 740 _closed : Unit <- File.close(file) 741 return wal.locked.read.classify(result, lock, path, size, records) 742 743 def wal.locked.read.pair( 744 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>, 745 lock: DbLock.Lock, 746 path: String, 747 size: Nat, 748 records: List<&2, MemTable.Entry> 749 ) -> IO(TableValidation): 750 match pair: 751 case (file, result): wal.locked.read.closed(file, result, lock, path, size, records) 752 753 def wal.locked.opened( 754 lock: DbLock.Lock, 755 sizes: Nat & Nat, 756 opened: Result<&1, &1, U32 & String, File>, 757 path: String, 758 records: List<&2, MemTable.Entry> 759 ) -> IO(TableValidation): 760 match sizes opened: 761 case (read_size, retained_size) Fail{(code, message)}: 762 IO.pure(TableValidation, 763 table.validation.records(lock, path, Fail{HostError{code, message}}, records)) 764 case (read_size, retained_size) Done{file}: 765 do IO: 766 read : File & Result<&1, &1, U32 & String, Bytes.Bytes> <- 767 Fs.read_at(file, 0, U32.from_nat(read_size)) 768 wal.locked.read.pair(read, lock, path, retained_size, records) 769 770 def wal.locked.sized( 771 lock: DbLock.Lock, 772 size: Nat, 773 +path: String, 774 within_limit: Bool, 775 records: List<&2, MemTable.Entry> 776 ) -> IO(TableValidation): 777 match within_limit: 778 case False{}: 779 IO.pure(TableValidation, 780 table.validation.records(lock, path, Fail{ResourceLimit{}}, records)) 781 case True{}: 782 do IO: 783 opened : Result<&1, &1, U32 & String, File> <- 784 File.open(path ++ "/WAL", "r") 785 wal.locked.opened(lock, QueuePolicy.nat.clone.pair(size), opened, path, 786 records) 787 788 def wal.locked.size.result( 789 lock: DbLock.Lock, 790 result: Result<&1, &1, U32 & String, Nat>, 791 path: String, 792 records: List<&2, MemTable.Entry> 793 ) -> IO(TableValidation): 794 match result: 795 case Fail{(code, message)}: 796 IO.pure(TableValidation, 797 table.validation.records(lock, path, Fail{HostError{code, message}}, records)) 798 case Done{+size}: 799 wal.locked.sized(lock, size, path, Nat.is_le(size, max_wal_bytes()), records) 800 801 def wal.locked.present( 802 lock: DbLock.Lock, 803 present: Bool, 804 +path: String, 805 records: List<&2, MemTable.Entry> 806 ) -> IO(TableValidation): 807 match present: 808 case False{}: 809 IO.pure(TableValidation, 810 table.validation.snapshot(lock, path, Done{(0n, 0n)}, records, Nil{})) 811 case True{}: 812 do IO: 813 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path ++ "/WAL") 814 wal.locked.size.result(lock, size, path, records) 815 816 def wal.locked.exists( 817 lock: DbLock.Lock, 818 result: Result<&1, &1, U32 & String, Bool>, 819 path: String, 820 records: List<&2, MemTable.Entry> 821 ) -> IO(TableValidation): 822 match result: 823 case Fail{(code, message)}: 824 IO.pure(TableValidation, 825 table.validation.records(lock, path, Fail{HostError{code, message}}, records)) 826 case Done{present}: wal.locked.present(lock, present, path, records) 827 828 def source.release.result( 829 value: Result<&1, &1, Error, SourceSnapshot>, 830 released: Result<&1, &1, DurableError.Error, Unit> 831 ) -> Result<&1, &1, Error, SourceSnapshot>: 832 match released: 833 case Fail{error}: Fail{LockError{error}} 834 case Done{Unit{}}: value 835 836 def source.release( 837 lock: DbLock.Lock, 838 path: String, 839 value: Result<&1, &1, Error, SourceSnapshot> 840 ) -> IO(Result<&1, &1, Error, SourceSnapshot>): 841 do IO>: 842 released : Result<&1, &1, DurableError.Error, Unit> <- 843 DbLock.release(lock, "migrate-verify", path) 844 return source.release.result(value, released) 845 846 def source.wal.validated( 847 sst: Nat & Nat, 848 state: TableValidation 849 ) -> IO(Result<&1, &1, Error, SourceSnapshot>): 850 match sst state: 851 case (sst_entries, sst_bytes) TableValidation{lock, path, result, 852 sst_records, wal_batches}: 853 match result: 854 case Fail{error}: 855 source.release(lock, path, Fail{error}) 856 case Done{(wal_frames, wal_bytes)}: 857 source.release(lock, path, Done{LegacySourceSnapshot{sst_records, 858 wal_batches, sst_entries, wal_frames, Nat.add(sst_bytes, wal_bytes)}}) 859 860 def source.tables.present( 861 sst: Nat & Nat, 862 sst_records: List<&2, MemTable.Entry>, 863 lock: DbLock.Lock, 864 paths: String & String 865 ) -> IO(Result<&1, &1, Error, SourceSnapshot>): 866 match paths: 867 case (exists_path, validation_path): 868 do IO>: 869 present : Result<&1, &1, U32 & String, Bool> <- 870 Fs.exists(exists_path ++ "/WAL") 871 wal : TableValidation <- wal.locked.exists(lock, present, 872 validation_path, sst_records) 873 source.wal.validated(sst, wal) 874 875 def source.tables.validated( 876 state: TableValidation 877 ) -> IO(Result<&1, &1, Error, SourceSnapshot>): 878 match state: 879 case TableValidation{lock, path, result, records, _}: 880 match result: 881 case Fail{error}: source.release(lock, path, Fail{error}) 882 case Done{sst}: 883 source.tables.present(sst, records, lock, path.clone.pair(path)) 884 885 def source.manifest.present( 886 lock: DbLock.Lock, 887 paths: String & String 888 ) -> IO(Result<&1, &1, Error, SourceSnapshot>): 889 match paths: 890 case (exists_path, validation_path): 891 do IO>: 892 present : Result<&1, &1, U32 & String, Bool> <- 893 Fs.exists(exists_path ++ "/MANIFEST") 894 tables : TableValidation <- 895 table.manifest.present(lock, validation_path, present) 896 source.tables.validated(tables) 897 898 def source.locked.manifest( 899 +path: String, 900 result: Result<&1, &1, DurableError.Error, DbLock.Lock> 901 ) -> IO(Result<&1, &1, Error, SourceSnapshot>): 902 match result: 903 case Fail{error}: 904 IO.pure(Result<&1, &1, Error, SourceSnapshot>, Fail{LockError{error}}) 905 case Done{lock}: 906 source.manifest.present(lock, path.clone.pair(path)) 907 908 def source.lock.present( 909 +path: String, 910 result: Result<&1, &1, U32 & String, Bool> 911 ) -> IO(Result<&1, &1, Error, SourceSnapshot>): 912 match result: 913 case Fail{(code, message)}: 914 IO.pure(Result<&1, &1, Error, SourceSnapshot>, Fail{HostError{code, message}}) 915 case Done{False{}}: 916 IO.pure(Result<&1, &1, Error, SourceSnapshot>, 917 Fail{HostError{2, "database LOCK file is missing"}}) 918 case Done{True{}}: 919 do IO>: 920 acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <- 921 DbLock.acquire(path, "migrate-verify") 922 source.locked.manifest(path, acquired) 923 924 def source.canonical( 925 result: Result<&1, &1, U32 & String, String> 926 ) -> IO(Result<&1, &1, Error, SourceSnapshot>): 927 match result: 928 case Fail{(code, message)}: 929 IO.pure(Result<&1, &1, Error, SourceSnapshot>, Fail{HostError{code, message}}) 930 case Done{+path}: 931 do IO>: 932 present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path ++ "/LOCK") 933 source.lock.present(path, present) 934 935 def source.counts( 936 result: Result<&1, &1, Error, SourceSnapshot> 937 ) -> Result<&1, &1, Error, SourceCounts>: 938 match result: 939 case Fail{error}: Fail{error} 940 case Done{LegacySourceSnapshot{_, _, sst_entries, wal_frames, encoded_bytes}}: 941 Done{LegacySourceCounts{sst_entries, wal_frames, encoded_bytes}} 942 943 # Reads legacy tables and WAL without changing source files. 944 def read_legacy_source( 945 +path: String 946 ) -> IO(Result<&1, &1, Error, SourceSnapshot>): 947 do IO>: 948 canonical : Result<&1, &1, U32 & String, String> <- Fs.canonical_path(path) 949 source.canonical(canonical) 950 951 # Validates the locked legacy view and returns its aggregate measurements. 952 def verify_legacy_source( 953 +path: String 954 ) -> IO(Result<&1, &1, Error, SourceCounts>): 955 do IO>: 956 snapshot : Result<&1, &1, Error, SourceSnapshot> <- read_legacy_source(path) 957 return source.counts(snapshot) 958 959 # Verifies a legacy WAL while holding the existing database lock read-only. 960 def verify_legacy_wal(+path: String) -> IO(Result<&1, &1, Error, Nat>): 961 do IO>: 962 canonical : Result<&1, &1, U32 & String, String> <- Fs.canonical_path(path) 963 canonical.result(canonical)