def walk

mylsm-lsm-store@0.6.0.1 · src/transfer/LegacyWal.bend:129

import Base
import mylsm-lsm-store@0.6.0.1/src/coordinator/QueuePolicy.bend as QueuePolicy
import mylsm-lsm-store@0.6.0.1/src/Wal.bend as Wal
import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes

129# Applies one pure parser transition, rejecting malformed or truncated frames.
130def walk(fuel: Nat, state: WalkState) -> VerifyResult:
131  match fuel:
132    case 0n: WalInvalid{}
133    case 1n+rest:
134      match state:
135        case Header{pair}:
136          match pair:
137            case (source, header):
138              walk(rest, HeaderDecoded{source, Wal.decode_header(header)})
139        case HeaderDecoded{source, result}:
140          match result:
141            case Fail{_}: WalInvalid{}
142            case Done{Unit{}}:
143              walk(rest, HeaderMeasured{Bytes.length(source)})
144        case HeaderMeasured{pair}:
145          match pair:
146            case (source, size):
147              walk(rest, HeaderSeek{
148                Bytes.Cursor.seek(Bytes.Cursor.new(source), 8)})
149        case HeaderSeek{pair}:
150          match pair:
151            case (_, False{}): WalInvalid{}
152            case (cursor, True{}): walk(rest, Read{cursor, 0n})
153        case Read{cursor, frames}:
154          walk(rest, Remaining{Bytes.Cursor.remaining(cursor), frames})
155        case Remaining{pair, frames}:
156          match pair:
157            case (cursor, remaining):
158              walk(rest, Empty{cursor, U32.to_nat(remaining), frames})
159        case Empty{cursor, remaining, frames}:
160          walk(rest, walk.empty.checked(cursor, frames,
161            QueuePolicy.nat.clone.pair(remaining)))
162        case EmptyChecked{cursor, remaining, frames, empty}:
163          match empty:
164            case True{}: walk(rest, Finish{frames})
165            case False{}:
166              walk(rest, walk.length.checked(cursor, frames,
167                QueuePolicy.nat.clone.pair(remaining)))
168        case HasLength{cursor, remaining, frames, has_length}:
169          match has_length:
170            case False{}: WalInvalid{}
171            case True{}:
172              walk(rest, Length{Bytes.Cursor.u32be(cursor),
173                U32.from_nat(Nat.sub(remaining, 4n)), frames})
174        case Length{pair, remaining, frames}:
175          match pair:
176            case (_, None{}): WalInvalid{}
177            case (cursor, Some{length}):
178              walk(rest, walk.length.validated(cursor, remaining, frames, length))
179        case Bounds{cursor, length, frames, valid}:
180          match valid:
181            case False{}: WalInvalid{}
182            case True{}:
183              walk(rest, walk.position.state(cursor, frames,
184                QueuePolicy.nat.clone.pair(U32.to_nat(length))))
185        case Position{cursor, positions, frame_bytes, frame_length, frames}:
186          match cursor:
187            case Bytes.Cursor{bytes, _, _, _}:
188              match positions:
189                case (start_nat, end_nat):
190                  +start = U32.from_nat(Nat.sub(start_nat, 4n))
191                  +end = U32.add(U32.from_nat(end_nat), frame_length)
192                  walk(rest, Frame{Bytes.slice(bytes, start, frame_bytes),
193                    end, frames})
194        case Frame{pair, end, frames}:
195          match pair:
196            case (source, frame):
197              walk(rest, Decoded{Wal.decode_frame(frame), source, end, frames})
198        case Decoded{result, source, end, frames}:
199          match result:
200            case Fail{_}: WalInvalid{}
201            case Done{_}:
202              walk(rest, Seek{Bytes.Cursor.seek(Bytes.Cursor.new(source), end),
203                Nat.add(frames, 1n)})
204        case Seek{pair, frames}:
205          match pair:
206            case (_, False{}): WalInvalid{}
207            case (cursor, True{}): walk(rest, Read{cursor, frames})
208        case Finish{frames}: WalValid{frames}
209        case Invalid{}: WalInvalid{}