def recover.result

mylsm-lsm-store@0.6.0.1 · src/ByteWalIo.bend:265

import Base
import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
import mylsm-lsm-store@0.6.0.1/src/ByteWal.bend as ByteWal
import mylsm-lsm-store@0.6.0.1/src/SstStreamIo.bend as SstStreamIo
import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes

265def 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<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
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<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
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<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
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<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
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)