1 import Base 2 import ./ByteMemTable.bend as ByteMemTable 3 import ./ByteWal.bend as ByteWal 4 import ./Fs.bend as Fs 5 import ./SstStreamIo.bend as SstStreamIo 6 import ./Wal.bend as Wal 7 import ./CrashPoint.bend as CrashPoint 8 import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes 9 10 # Tracks the owned file and bounded phase state during byte WAL replay. 11 type RecoveryState is Type: 12 At{file: File, size: Nat, offset: Nat, table: ByteMemTable.Table, 13 at_end: Bool, can_read_length: Bool} 14 Length{pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>, 15 size: Nat, offset: Nat, table: ByteMemTable.Table} 16 Prefix{file: File, size: Nat, offset: Nat, table: ByteMemTable.Table, 17 prefix: Bytes.Bytes, frame_length: U32, within_file: Bool} 18 Count{pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>, 19 size: Nat, offset: Nat, table: ByteMemTable.Table, 20 prefix: Bytes.Bytes, frame_length: U32} 21 Counted{file: File, size: Nat, offset: Nat, table: ByteMemTable.Table, 22 prefix: Bytes.Bytes, count_bytes: Bytes.Bytes, frame_length: U32, 23 within_limit: Bool} 24 Body{pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>, 25 size: Nat, offset: Nat, table: ByteMemTable.Table, 26 prefix: Bytes.Bytes, count_bytes: Bytes.Bytes, frame_length: U32} 27 Decoded{file: File, size: Nat, next_offset: Nat, 28 table: ByteMemTable.Table, 29 result: Result<&1, &1, Wal.Error, ByteWal.Batch>} 30 Failed{file: File, error: U32 & String} 31 Tail{file: File, path: String, offset: Nat, table: ByteMemTable.Table} 32 33 # Assigns the stable file-corruption category used by this reader. 34 def corruption_error_code() -> U32: 35 4294967292 36 37 # Pairs a corruption category with its diagnostic text. 38 def error(message: String) -> U32 & String: 39 (corruption_error_code(), message) 40 41 def u32.read(pair: Bytes.Cursor & Maybe<&2, U32>) -> Maybe<&2, U32>: 42 match pair: 43 case (_, value): 44 value 45 46 def frame.length(bytes: Bytes.Bytes) -> Maybe<&2, U32>: 47 u32.read(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes))) 48 49 def frame.count(bytes: Bytes.Bytes) -> Maybe<&2, U32>: 50 frame.length(bytes) 51 52 def recover.at( 53 file: File, 54 +size: Nat, 55 +offset: Nat, 56 table: ByteMemTable.Table 57 ) -> RecoveryState: 58 At{file, size, offset, table, Nat.is_eq(offset, size), 59 Nat.is_le(Nat.add(offset, 4n), size)} 60 61 def recover.prefix.next( 62 file: File, 63 +size: Nat, 64 +offset: Nat, 65 table: ByteMemTable.Table, 66 prefix: Bytes.Bytes, 67 parsed: Maybe<&2, U32> 68 ) -> RecoveryState: 69 match parsed: 70 case None{}: 71 Failed{file, error("malformed byte WAL frame length")} 72 case Some{+frame_length}: 73 Prefix{file, size, offset, table, prefix, frame_length, 74 Nat.is_le(Nat.add(offset, Nat.add(4n, U32.to_nat(frame_length))), size) && 75 U32.is_le(36, frame_length)} 76 77 def recover.prefix.cloned( 78 file: File, 79 pair: Bytes.Bytes & Bytes.Bytes, 80 +size: Nat, 81 +offset: Nat, 82 table: ByteMemTable.Table 83 ) -> RecoveryState: 84 match pair: 85 case (prefix, inspected): 86 recover.prefix.next(file, size, offset, table, 87 prefix, frame.length(inspected)) 88 89 def recover.counted( 90 file: File, 91 +size: Nat, 92 +offset: Nat, 93 table: ByteMemTable.Table, 94 prefix: Bytes.Bytes, 95 count_bytes: Bytes.Bytes, 96 +frame_length: U32, 97 parsed: Maybe<&2, U32> 98 ) -> RecoveryState: 99 match parsed: 100 case None{}: 101 Failed{file, error("malformed byte WAL frame count")} 102 case Some{+count}: 103 Counted{file, size, offset, table, prefix, count_bytes, 104 frame_length, Wal.frame_length_allowed(count, frame_length)} 105 106 def recover.count.cloned( 107 file: File, 108 pair: Bytes.Bytes & Bytes.Bytes, 109 +size: Nat, 110 +offset: Nat, 111 table: ByteMemTable.Table, 112 prefix: Bytes.Bytes, 113 +frame_length: U32 114 ) -> RecoveryState: 115 match pair: 116 case (count_bytes, inspected): 117 recover.counted(file, size, offset, table, prefix, 118 count_bytes, frame_length, frame.count(inspected)) 119 120 def apply.batch.loop( 121 fuel: Nat, 122 table: ByteMemTable.Table, 123 mutations: List<&1, ByteWal.Mutation> 124 ) -> ByteMemTable.Table: 125 match fuel: 126 case 0n: 127 table 128 case 1n+rest: 129 match mutations: 130 case Nil{}: 131 table 132 case Con{ByteWal.Put{key, value}, tail}: 133 apply.batch.loop(rest, ByteMemTable.put(table, key, value), tail) 134 case Con{ByteWal.Delete{key}, tail}: 135 apply.batch.loop(rest, ByteMemTable.del(table, key), tail) 136 137 def mutations.count.go( 138 mutations: List<&1, ByteWal.Mutation>, 139 reversed: List<&1, ByteWal.Mutation>, 140 count: Nat 141 ) -> List<&1, ByteWal.Mutation> & Nat: 142 match mutations: 143 case Nil{}: 144 (List.reverse(&1, ByteWal.Mutation, reversed), count) 145 case Con{mutation, tail}: 146 mutations.count.go(tail, Con{mutation, reversed}, Nat.add(count, 1n)) 147 148 def mutations.count(mutations: List<&1, ByteWal.Mutation>) -> List<&1, ByteWal.Mutation> & Nat: 149 mutations.count.go(mutations, Nil{}, 0n) 150 151 def apply.batch.counted( 152 table: ByteMemTable.Table, 153 pair: List<&1, ByteWal.Mutation> & Nat 154 ) -> ByteMemTable.Table: 155 match pair: 156 case (mutations, count): 157 apply.batch.loop(Nat.add(count, 1n), table, mutations) 158 159 def apply.batch(table: ByteMemTable.Table, batch: ByteWal.Batch) -> ByteMemTable.Table: 160 match batch: 161 case ByteWal.Batch{mutations}: 162 apply.batch.counted(table, mutations.count(mutations)) 163 164 def close.failed(file: File, failure: U32 & String) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 165 do IO>: 166 _closed : Unit <- File.close(file) 167 return Fail{failure} 168 169 def close.done(file: File, table: ByteMemTable.Table) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 170 do IO>: 171 _closed : Unit <- File.close(file) 172 return Done{table} 173 174 def recover.exhausted(state: RecoveryState) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 175 match state: 176 case At{file, _, _, _, _, _}: 177 close.failed(file, error("byte WAL recovery limit exceeded")) 178 case Length{pair, _, _, _}: 179 match pair: 180 case (file, _): 181 close.failed(file, error("byte WAL recovery limit exceeded")) 182 case Prefix{file, _, _, _, _, _, _}: 183 close.failed(file, error("byte WAL recovery limit exceeded")) 184 case Count{pair, _, _, _, _, _}: 185 match pair: 186 case (file, _): 187 close.failed(file, error("byte WAL recovery limit exceeded")) 188 case Counted{file, _, _, _, _, _, _, _}: 189 close.failed(file, error("byte WAL recovery limit exceeded")) 190 case Body{pair, _, _, _, _, _, _}: 191 match pair: 192 case (file, _): 193 close.failed(file, error("byte WAL recovery limit exceeded")) 194 case Decoded{file, _, _, _, _}: 195 close.failed(file, error("byte WAL recovery limit exceeded")) 196 case Failed{file, failure}: 197 close.failed(file, failure) 198 case Tail{file, _, _, _}: 199 close.failed(file, error("byte WAL recovery limit exceeded")) 200 201 def recover.length.completed( 202 +path: String, 203 file: File, 204 +size: Nat, 205 +offset: Nat, 206 table: ByteMemTable.Table, 207 bytes: Bytes.Bytes, 208 complete: Bool 209 ) -> RecoveryState: 210 match complete: 211 case False{}: 212 Tail{file, path, offset, table} 213 case True{}: 214 recover.prefix.cloned(file, Wal.bytes.clone.split(bytes), size, offset, table) 215 216 def recover.count.completed( 217 +path: String, 218 file: File, 219 +size: Nat, 220 +offset: Nat, 221 table: ByteMemTable.Table, 222 prefix: Bytes.Bytes, 223 bytes: Bytes.Bytes, 224 complete: Bool, 225 +frame_length: U32 226 ) -> RecoveryState: 227 match complete: 228 case False{}: 229 Tail{file, path, offset, table} 230 case True{}: 231 recover.count.cloned(file, Wal.bytes.clone.split(bytes), size, offset, table, prefix, frame_length) 232 233 def tail.synced( 234 result: Result<&1, &1, U32 & String, Unit>, 235 table: ByteMemTable.Table 236 ) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 237 match result: 238 case Fail{failure}: 239 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{failure}) 240 case Done{Unit{}}: 241 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Done{table}) 242 243 def tail.truncated( 244 result: Result<&1, &1, U32 & String, Unit>, 245 +path: String, 246 table: ByteMemTable.Table 247 ) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 248 match result: 249 case Fail{failure}: 250 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{failure}) 251 case Done{Unit{}}: 252 do IO>: 253 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path) 254 tail.synced(synced, table) 255 256 def tail.closed( 257 +path: String, 258 +offset: Nat, 259 table: ByteMemTable.Table 260 ) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 261 do IO>: 262 truncated : Result<&1, &1, U32 & String, Unit> <- Fs.truncate(path, offset) 263 tail.truncated(truncated, path, table) 264 265 def recover.result( 266 +path: String, 267 fuel: Nat, 268 state: RecoveryState 269 ) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 270 match fuel: 271 case 0n: 272 recover.exhausted(state) 273 case 1n+rest: 274 match state: 275 case Tail{file, path, +offset, table}: 276 do IO>: 277 _closed : Unit <- File.close(file) 278 tail.closed(path, offset, table) 279 case At{file, +size, +offset, table, at_end, can_read_length}: 280 match at_end: 281 case True{}: 282 close.done(file, table) 283 case False{}: 284 match can_read_length: 285 case False{}: 286 recover.result(path, rest, Tail{file, path, offset, table}) 287 case True{}: 288 do IO>: 289 pair : File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult> <- 290 SstStreamIo.exact.start(file, 4n) 291 recover.result(path, rest, Length{pair, size, offset, table}) 292 case Length{pair, +size, +offset, table}: 293 match pair: 294 case (file, Fail{failure}): 295 close.failed(file, failure) 296 case (file, Done{SstStreamIo.ExactResult{bytes, complete}}): 297 recover.result(path, rest, recover.length.completed(path, file, size, offset, table, bytes, complete)) 298 case Prefix{file, +size, +offset, table, prefix, +frame_length, within_file}: 299 match within_file: 300 case False{}: 301 recover.result(path, rest, Tail{file, path, offset, table}) 302 case True{}: 303 do IO>: 304 pair : File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult> <- 305 SstStreamIo.exact.start(file, 4n) 306 recover.result(path, rest, Count{pair, size, offset, table, 307 prefix, frame_length}) 308 case Count{pair, +size, +offset, table, prefix, +frame_length}: 309 match pair: 310 case (file, Fail{failure}): 311 close.failed(file, failure) 312 case (file, Done{SstStreamIo.ExactResult{bytes, complete}}): 313 recover.result(path, rest, recover.count.completed(path, file, size, offset, table, prefix, bytes, complete, frame_length)) 314 case Counted{file, +size, +offset, table, prefix, count_bytes, 315 +frame_length, within_limit}: 316 match within_limit: 317 case False{}: 318 close.failed(file, error("byte WAL frame exceeds configured limit")) 319 case True{}: 320 do IO>: 321 pair : File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult> <- 322 SstStreamIo.exact.start(file, U32.to_nat((frame_length - 4 : U32))) 323 recover.result(path, rest, Body{pair, size, offset, table, 324 prefix, count_bytes, frame_length}) 325 case Body{pair, +size, +offset, table, prefix, count_bytes, +frame_length}: 326 match pair: 327 case (file, Fail{failure}): 328 close.failed(file, failure) 329 case (file, Done{SstStreamIo.ExactResult{bytes, complete}}): 330 match complete: 331 case False{}: 332 recover.result(path, rest, Tail{file, path, offset, table}) 333 case True{}: 334 recover.result(path, rest, Decoded{file, size, 335 Nat.add(offset, Nat.add(4n, U32.to_nat(frame_length))), table, 336 ByteWal.decode(Bytes.concat([prefix, count_bytes, bytes]))}) 337 case Decoded{file, +size, +next_offset, table, result}: 338 match result: 339 case Fail{_}: 340 close.failed(file, error("corrupt byte WAL frame")) 341 case Done{batch}: 342 recover.result(path, rest, recover.at(file, size, next_offset, 343 apply.batch(table, batch))) 344 case Failed{file, failure}: 345 close.failed(file, failure) 346 347 def header.checked( 348 +path: String, 349 file: File, 350 +size: Nat, 351 result: Result<&1, &1, Wal.Error, Unit>, 352 initial: ByteMemTable.Table 353 ) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 354 match result: 355 case Fail{_}: 356 close.failed(file, error("unsupported byte WAL format")) 357 case Done{Unit{}}: 358 recover.result(path, Nat.add(Nat.mul(size, 8n), 8n), 359 recover.at(file, size, 8n, initial)) 360 361 def header.valid( 362 +path: String, 363 file: File, 364 +size: Nat, 365 result: Result<&1, &1, U32 & String, Bytes.Bytes>, 366 initial: ByteMemTable.Table 367 ) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 368 match result: 369 case Fail{failure}: 370 close.failed(file, failure) 371 case Done{bytes}: 372 header.checked(path, file, size, ByteWal.decode_header(bytes), initial) 373 374 def header.pair( 375 +path: String, 376 +size: Nat, 377 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>, 378 initial: ByteMemTable.Table 379 ) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 380 match pair: 381 case (file, result): 382 header.valid(path, file, size, result, initial) 383 384 def recover.opened( 385 +path: String, 386 +size: Nat, 387 opened: Result<&1, &1, U32 & String, File>, 388 initial: ByteMemTable.Table 389 ) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 390 match opened: 391 case Fail{failure}: 392 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{failure}) 393 case Done{file}: 394 do IO>: 395 pair : File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult> <- 396 SstStreamIo.exact.start(file, 8n) 397 header.pair(path, size, SstStreamIo.read.exact.state(pair), initial) 398 399 def recover.size.checked( 400 +path: String, 401 +size: Nat, 402 valid: Bool, 403 initial: ByteMemTable.Table 404 ) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 405 match valid: 406 case False{}: 407 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, 408 Fail{error("malformed or oversized byte WAL")}) 409 case True{}: 410 do IO>: 411 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r") 412 recover.opened(path, size, opened, initial) 413 414 def recover.sized( 415 +path: String, 416 result: Result<&1, &1, U32 & String, Nat>, 417 initial: ByteMemTable.Table 418 ) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 419 match result: 420 case Fail{failure}: 421 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{failure}) 422 case Done{+size}: 423 recover.size.checked(path, size, 424 Nat.is_le(8n, size) && Nat.is_le(size, 4294967295n), initial) 425 426 427 def recover.from(+path: String, initial: ByteMemTable.Table) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 428 do IO>: 429 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path) 430 recover.sized(path, size, initial) 431 432 # Replays the byte WAL over an empty initial mutation table. 433 def recover(+path: String) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>): 434 recover.from(path, ByteMemTable.empty()) 435 436 # Publishes a fresh WAL header without exposing a partially truncated active file. 437 def create.closed( 438 result: Result<&1, &1, U32 & String, Unit>, 439 +path: String 440 ) -> IO(Result<&1, &1, U32 & String, Unit>): 441 match result: 442 case Fail{failure}: 443 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{failure}) 444 case Done{Unit{}}: 445 Fs.fsync(path) 446 447 def create.written( 448 pair: File & Result<&1, &1, U32 & String, Unit>, 449 +path: String 450 ) -> IO(Result<&1, &1, U32 & String, Unit>): 451 match pair: 452 case (file, result): 453 do IO>: 454 _closed : Unit <- File.close(file) 455 create.closed(result, path) 456 457 def create.opened( 458 opened: Result<&1, &1, U32 & String, File>, 459 +path: String 460 ) -> IO(Result<&1, &1, U32 & String, Unit>): 461 match opened: 462 case Fail{failure}: 463 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{failure}) 464 case Done{file}: 465 do IO>: 466 pair : File & Result<&1, &1, U32 & String, Unit> <- 467 Fs.write_bytes(file, ByteWal.header()) 468 create.written(pair, path) 469 470 # Creates and durably publishes an empty generation-four byte WAL. 471 def create(+path: String) -> IO(Result<&1, &1, U32 & String, Unit>): 472 do IO>: 473 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "w") 474 create.opened(opened, path) 475 476 def reset.directory.synced( 477 +directory: String, 478 result: Result<&1, &1, U32 & String, Unit> 479 ) -> IO(Result<&1, &1, U32 & String, Unit>): 480 match result: 481 case Fail{failure}: 482 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{failure}) 483 case Done{Unit{}}: 484 Fs.fsync(directory) 485 486 # Creates and syncs a replacement WAL before atomic publication. 487 488 def reset.published( 489 +directory: String, 490 result: Result<&1, &1, U32 & String, Unit> 491 ) -> IO(Result<&1, &1, U32 & String, Unit>): 492 match result: 493 case Fail{failure}: 494 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{failure}) 495 case Done{Unit{}}: 496 do IO>: 497 renamed : Result<&1, &1, U32 & String, Unit> <- 498 Fs.rename(directory ++ "/WAL.tmp", directory ++ "/WAL") 499 reset.directory.synced(directory, renamed) 500 501 # Syncs the directory entry after atomically replacing the active WAL. 502 503 # Replaces the active WAL only after its new header is synced. 504 def reset_atomic(+directory: String) -> IO(Result<&1, &1, U32 & String, Unit>): 505 do IO>: 506 created : Result<&1, &1, U32 & String, Unit> <- create(directory ++ "/WAL.tmp") 507 reset.published(directory, created) 508 509 510 def append.synced( 511 result: Result<&1, &1, U32 & String, Unit>, 512 file: File, 513 +path: String 514 ) -> IO(Result<&1, &1, U32 & String, Unit>): 515 match result: 516 case Fail{failure}: 517 do IO>: 518 _closed : Unit <- File.close(file) 519 return Fail{failure} 520 case Done{Unit{}}: 521 do IO>: 522 _closed : Unit <- File.close(file) 523 Fs.fsync(path) 524 525 def append.written( 526 pair: File & Result<&1, &1, U32 & String, Unit>, 527 +path: String 528 ) -> IO(Result<&1, &1, U32 & String, Unit>): 529 match pair: 530 case (file, result): 531 append.synced(result, file, path) 532 533 # Retains the requested portion of a duplicated WAL frame. 534 def append.slice(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes: 535 match pair: 536 case (_, bytes): bytes 537 538 # Stops between WAL chunks or writes the remaining frame before sync. 539 def append.first.written( 540 pair: File & Result<&1, &1, U32 & String, Unit>, 541 remaining: Bytes.Bytes, 542 +path: String 543 ) -> IO(Result<&1, &1, U32 & String, Unit>): 544 match pair: 545 case (file, Fail{failure}): 546 do IO>: 547 _closed : Unit <- File.close(file) 548 return Fail{failure} 549 case (file, Done{Unit{}}): 550 do IO>: 551 _checkpoint : Result<&1, &1, U32 & String, Unit> <- 552 CrashPoint.hit("wal.during_append") 553 written : File & Result<&1, &1, U32 & String, Unit> <- 554 Fs.write_bytes(file, remaining) 555 append.written(written, path) 556 557 # Writes the first bounded WAL chunk and continues through the split point. 558 def append.chunked.rest( 559 file: File, 560 first: Bytes.Bytes, 561 remaining: Bytes.Bytes, 562 +path: String 563 ) -> IO(Result<&1, &1, U32 & String, Unit>): 564 do IO>: 565 written : File & Result<&1, &1, U32 & String, Unit> <- 566 Fs.write_bytes(file, first) 567 append.first.written(written, remaining, path) 568 569 # Splits duplicated frame buffers around one bounded WAL write chunk. 570 def append.chunked.cloned( 571 file: File, 572 sources: Bytes.Bytes & Bytes.Bytes, 573 total: U32, 574 +path: String 575 ) -> IO(Result<&1, &1, U32 & String, Unit>): 576 match sources: 577 case (first_source, rest_source): 578 append.chunked.rest(file, 579 append.slice(Bytes.slice(first_source, 0, 65536)), 580 append.slice(Bytes.slice(rest_source, 65536, 581 (total - 65536 : U32))), path) 582 583 # Clones a large WAL frame before forming its bounded append chunks. 584 def append.chunked.slices( 585 file: File, 586 frame: Bytes.Bytes, 587 total: U32, 588 +path: String 589 ) -> IO(Result<&1, &1, U32 & String, Unit>): 590 do IO>: 591 sources : Bytes.Bytes & Bytes.Bytes <- 592 IO.pure(Bytes.Bytes & Bytes.Bytes, ByteMemTable.bytes.clone(frame)) 593 append.chunked.cloned(file, sources, total, path) 594 595 # Selects full-frame or chunked writing after opening the WAL. 596 def append.opened.checked( 597 file: File, 598 frame: Bytes.Bytes, 599 total: U32, 600 +path: String, 601 large: Bool 602 ) -> IO(Result<&1, &1, U32 & String, Unit>): 603 match large: 604 case True{}: 605 append.chunked.slices(file, frame, total, path) 606 case False{}: 607 do IO>: 608 written : File & Result<&1, &1, U32 & String, Unit> <- 609 Fs.write_bytes(file, frame) 610 append.written(written, path) 611 612 # Routes an opened WAL frame according to its encoded byte length. 613 def append.opened.frame( 614 file: File, 615 frame: Bytes.Bytes, 616 +path: String 617 ) -> IO(Result<&1, &1, U32 & String, Unit>): 618 match frame: 619 case Bytes.Bytes{+total, _}: 620 append.opened.checked(file, frame, total, path, 621 U32.is_le(65537, total)) 622 623 def append.opened( 624 opened: Result<&1, &1, U32 & String, File>, 625 frame: Bytes.Bytes, 626 +path: String 627 ) -> IO(Result<&1, &1, U32 & String, Unit>): 628 match opened: 629 case Fail{failure}: 630 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{failure}) 631 case Done{file}: 632 do IO>: 633 _checkpoint : Result<&1, &1, U32 & String, Unit> <- 634 CrashPoint.hit("wal.before_append") 635 append.opened.frame(file, frame, path) 636 637 def append.encoded( 638 result: Result<&1, &1, Wal.Error, Bytes.Bytes>, 639 +path: String 640 ) -> IO(Result<&1, &1, U32 & String, Unit>): 641 match result: 642 case Fail{_}: 643 IO.pure(Result<&1, &1, U32 & String, Unit>, 644 Fail{(3, "byte WAL batch cannot be encoded")}) 645 case Done{frame}: 646 do IO>: 647 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "a") 648 append.opened(opened, frame, path) 649 650 # Appends one encoded batch and syncs the resulting WAL file. 651 def append(+path: String, batch: ByteWal.Batch) -> IO(Result<&1, &1, U32 & String, Unit>): 652 append.encoded(ByteWal.encode(batch), path)