mylsm.bend

mylsm-lsm-store@0.6.0.1 · 2,544 lines · 255 definitions

1# Durable LSM key-value database with binary and UTF-8 APIs, atomic batches, snapshots, and backups.
2import Base
3import mylsm-lsm-store@0.6.0.1/src/DurableError.bend as DurableError
4import bend-kit-bytes@0.3.2.0/bytes.bend as PackedBytes
5import mylsm-lsm-store@0.6.0.1/src/ByteDurableDb.bend as RawByteDb
6import mylsm-lsm-store@0.6.0.1/src/ByteWal.bend as RawByteWal
7import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
8import mylsm-lsm-store@0.6.0.1/src/scan/RangeCursor.bend as ByteRangeCursor
9import mylsm-lsm-store@0.6.0.1/src/scan/VisibleCursor.bend as VisibleByteCursor
10import mylsm-lsm-store@0.6.0.1/src/scan/Snapshot.bend as SnapshotPolicy
11import mylsm-lsm-store@0.6.0.1/src/scan/Utf8CursorPolicy.bend as Utf8CursorPolicy
12import mylsm-lsm-store@0.6.0.1/src/scan/CursorCleanupPolicy.bend as CursorCleanupPolicy
13import mylsm-lsm-store@0.6.0.1/src/transfer/Backup.bend as BackupCodec
14import mylsm-lsm-store@0.6.0.1/src/transfer/BackupIo.bend as BackupIo
15import mylsm-lsm-store@0.6.0.1/src/transfer/BackupCursorIo.bend as BackupCursorIo
16import mylsm-lsm-store@0.6.0.1/src/transfer/BackupOwnerCursorIo.bend as BackupOwnerCursorIo
17import mylsm-lsm-store@0.6.0.1/src/transfer/RestoreIo.bend as RestoreIo
18import mylsm-lsm-store@0.6.0.1/src/transfer/FormatInspect.bend as FormatInspect
19import mylsm-lsm-store@0.6.0.1/src/transfer/FormatInspectIo.bend as FormatInspectIo
20import mylsm-lsm-store@0.6.0.1/src/transfer/MigrateRunIo.bend as MigrateRunIo
21import mylsm-lsm-store@0.6.0.1/src/StorageBytes.bend as Utf8Codec
22import mylsm-lsm-store@0.6.0.1/src/Fs.bend as Fs
23import mylsm-lsm-store@0.6.0.1/src/coordinator/EndpointPath.bend as EndpointPath
24import mylsm-lsm-store@0.6.0.1/src/coordinator/OwnerGenerationIo.bend as OwnerGenerationIo
25import mylsm-lsm-store@0.6.0.1/src/coordinator/Client.bend as OwnerClient
26import mylsm-lsm-store@0.6.0.1/src/coordinator/ByteClient.bend as OwnerByteClient
27import mylsm-lsm-store@0.6.0.1/src/coordinator/CursorCodec.bend as OwnerCursorCodec
28import mylsm-lsm-store@0.6.0.1/src/coordinator/Protocol.bend as CoordinatorProtocol
29import mylsm-lsm-store@0.6.0.1/src/coordinator/StatsCodec.bend as OwnerStatsCodec
30import mylsm-lsm-store@0.6.0.1/src/coordinator/Io.bend as OwnerIo
31import mylsm-lsm-store@0.6.0.1/src/coordinator/UnixSocketIo.bend as OwnerSocket
32import mylsm-lsm-store@0.6.0.1/src/coordinator/CoordinatorPolicy.bend as OwnerPolicy
33import mylsm-lsm-store@0.6.0.1/src/coordinator/OwnerOpenPolicy.bend as OwnerOpenPolicy
34import mylsm-lsm-store@0.6.0.1/src/ResourceOptions.bend as ResourceOptions
35import mylsm-lsm-store@0.6.0.1/src/coordinator/OptionsCodec.bend as OwnerOptionsCodec
36
37# Describes manifest validation independently from package/storage identity.
38type ManifestStatus is Data:
39  ManifestValid{}
40  ManifestVersionUnsupported{generation: U32}
41  ManifestComparatorUnsupported{generation: U32}
42  ManifestFormatUnknown{}
43  ManifestCorrupt{}
44  ManifestResourceLimit{}
45
46# Reports package version, storage generation, and validation status.
47type ManifestInspection is Data:
48  ManifestInspection{package_version: String, storage_version: String,
49    generation: U32, status: ManifestStatus}
50
51# Reports lock or host errors from the read-only inspection operation.
52type InspectionError is Data:
53  InspectionBusy{}
54  InspectionTimeout{}
55  InspectionUnsupportedPlatform{}
56  InspectionLockIo{code: U32, message: String}
57  InspectionIo{code: U32, message: String}
58
59type Bytes.Error is Data:
60  Busy{}
61  Timeout{}
62  NotFound{}
63  AlreadyExists{}
64  InvalidArgument{}
65  UnsupportedPlatform{}
66  UnsupportedOperation{}
67  ResourceLimit{}
68  Corruption{message: String}
69  Io{code: U32, message: String}
70  CommitUnknown{code: U32, message: String}
71  StaleOwner{}
72
73type Bytes.Stats is Data:
74  ByteStats{entries: Nat, payload_bytes: Nat, snapshots: Nat, retained_bytes: Nat, cursors: Nat}
75
76type Bytes.Snapshot is Type:
77  ByteSnapshot{entries: List<&1, ByteMemTable.Entry>}
78  RemoteByteSnapshot{session: OwnerClient.Session, snapshot_id: U32}
79
80type Bytes.BackupStream is Type:
81  ByteBackupStream{inner: BackupCodec.Stream}
82
83type Bytes.BackupStep is Type:
84  BackupChunk{stream: Bytes.BackupStream, bytes: PackedBytes.Bytes}
85  BackupEnd{}
86
87# Converts internal byte-engine errors into the public byte API error type.
88def Bytes.error(error: RawByteDb.Error) -> Bytes.Error:
89  match error:
90    case RawByteDb.Busy{}: Busy{}
91    case RawByteDb.Timeout{}: Timeout{}
92    case RawByteDb.NotFound{}: NotFound{}
93    case RawByteDb.AlreadyExists{}: AlreadyExists{}
94    case RawByteDb.InvalidArgument{}: InvalidArgument{}
95    case RawByteDb.UnsupportedPlatform{}: UnsupportedPlatform{}
96    case RawByteDb.ResourceLimit{}: ResourceLimit{}
97    case RawByteDb.Corruption{message}: Corruption{message}
98    case RawByteDb.Io{code, message}: Io{code, message}
99    case RawByteDb.CommitUnknown{code, message}: CommitUnknown{code, message}
100
101def Bytes.error.result(result: Result<&1, &1, RawByteDb.Error, Unit>) -> Result<&1, &1, Bytes.Error, Unit>:
102  match result:
103    case Fail{error}: Fail{Bytes.error(error)}
104    case Done{Unit{}}: Done{Unit{}}
105
106def Bytes.error.maybe(
107  result: Result<&1, &1, RawByteDb.Error, Maybe<&1, PackedBytes.Bytes>>
108) -> Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>:
109  match result:
110    case Fail{error}: Fail{Bytes.error(error)}
111    case Done{value}: Done{value}
112
113def Bytes.owner.error.status(status: CoordinatorProtocol.ResponseStatus) -> Bytes.Error:
114  match status:
115    case CoordinatorProtocol.Success{}: InvalidArgument{}
116    case CoordinatorProtocol.InvalidArgument{}: InvalidArgument{}
117    case CoordinatorProtocol.NotFound{}: NotFound{}
118    case CoordinatorProtocol.Busy{}: Busy{}
119    case CoordinatorProtocol.Timeout{}: Timeout{}
120    case CoordinatorProtocol.ResponseResourceLimit{}: ResourceLimit{}
121    case CoordinatorProtocol.Corruption{}: Corruption{"owner reported corruption"}
122    case CoordinatorProtocol.IoFailure{}: Io{0, "owner IO failure"}
123    case CoordinatorProtocol.CommitUnknown{}: CommitUnknown{0, "owner commit is uncertain"}
124    case CoordinatorProtocol.AlreadyExists{}: AlreadyExists{}
125    case CoordinatorProtocol.UnsupportedPlatform{}: UnsupportedPlatform{}
126    case CoordinatorProtocol.InvalidUtf8{}: Corruption{"owner rejected invalid UTF-8"}
127    case CoordinatorProtocol.UnsupportedFormat{}: Corruption{"owner storage format is unsupported"}
128
129def Bytes.owner.error.socket(error: OwnerSocket.Error) -> Bytes.Error:
130  match error:
131    case OwnerSocket.UnsupportedPlatform{}: UnsupportedPlatform{}
132    case OwnerSocket.Transient{code}: Io{code, "owner socket operation was interrupted"}
133    case OwnerSocket.Host{code, message}: Io{code, message}
134
135def Bytes.owner.error.io(error: OwnerIo.Error) -> Bytes.Error:
136  match error:
137    case OwnerIo.UnsupportedPlatform{}: UnsupportedPlatform{}
138    case OwnerIo.Transient{code}: Io{code, "owner response IO was interrupted"}
139    case OwnerIo.Host{code, message}: Io{code, message}
140    case _: Corruption{"owner response transport was invalid"}
141
142def Bytes.owner.error.commit(error: OwnerIo.Error) -> Bytes.Error:
143  match error:
144    case OwnerIo.Transient{code}:
145      CommitUnknown{code, "mutation response IO was interrupted"}
146    case OwnerIo.Host{code, message}: CommitUnknown{code, message}
147    case _: CommitUnknown{0, "mutation outcome is uncertain"}
148
149def Bytes.owner.error(error: OwnerClient.Error) -> Bytes.Error:
150  match error:
151    case OwnerClient.ConnectFailure{error}: Bytes.owner.error.socket(error)
152    case OwnerClient.RetryRead{error}: Bytes.owner.error.io(error)
153    case OwnerClient.CommitUnknown{error}: Bytes.owner.error.commit(error)
154    case OwnerClient.FrameTooLarge{}: ResourceLimit{}
155    case OwnerClient.StaleOwner{}: StaleOwner{}
156    case OwnerClient.CorrelationMismatch{}: Corruption{"owner response correlation mismatch"}
157    case OwnerClient.MalformedResponse{}: Corruption{"owner response is malformed"}
158    case OwnerClient.RequestIdsExhausted{}: ResourceLimit{}
159    case OwnerClient.OwnerRejected{status}: Bytes.owner.error.status(status)
160
161def Bytes.owner.unit.result(
162  pair: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit>
163) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>:
164  match pair:
165    case (session, Fail{error}):
166      (RemoteByteHandle{session}, Fail{Bytes.owner.error(error)})
167    case (session, Done{Unit{}}):
168      (RemoteByteHandle{session}, Done{Unit{}})
169
170def Bytes.owner.get.result(
171  pair:
172    OwnerClient.Session &
173      Result<&1, &1, OwnerClient.Error, Maybe<&1, PackedBytes.Bytes>>
174) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>:
175  match pair:
176    case (session, Fail{error}):
177      (RemoteByteHandle{session}, Fail{Bytes.owner.error(error)})
178    case (session, Done{value}):
179      (RemoteByteHandle{session}, Done{value})
180
181type Bytes.Handle is Type:
182  ByteHandle{raw: RawByteDb.Handle}
183  RemoteByteHandle{session: OwnerClient.Session}
184
185type Bytes.Mutation is Type:
186  BytesPut{key: PackedBytes.Bytes, value: PackedBytes.Bytes}
187  BytesDelete{key: PackedBytes.Bytes}
188
189type Bytes.ScanBounds is Type:
190  ScanBounds{lower: Maybe<&1, PackedBytes.Bytes>, upper: Maybe<&1, PackedBytes.Bytes>,
191    prefix: Maybe<&1, PackedBytes.Bytes>}
192
193type Bytes.Cursor is Type:
194  ByteCursor{inner: ByteRangeCursor.Cursor, bounds: ByteRangeCursor.Bounds}
195  ByteStreamCursor{inner: VisibleByteCursor.Cursor}
196  ByteOwnerCursor{session: OwnerClient.Session, snapshot_id: U32, cursor_id: U32}
197  ByteOwnedCursor{snapshot: Bytes.Snapshot, inner: Bytes.Cursor,
198    bounds: ByteRangeCursor.Bounds}
199  ByteFailureCursor{error: Bytes.Error}
200  ByteClosedCursor{}
201
202type Bytes.ScanStep is Type:
203  ScanItem{key: PackedBytes.Bytes, value: PackedBytes.Bytes, cursor: Bytes.Cursor}
204  ScanSkip{cursor: Bytes.Cursor}
205  ScanEnd{cursor: Bytes.Cursor}
206  ScanFailure{error: Bytes.Error, cursor: Bytes.Cursor}
207
208type Bytes.StreamStep is Type:
209  StreamItem{key: PackedBytes.Bytes, value: PackedBytes.Bytes,
210    cursor: Bytes.Cursor}
211  StreamSkip{cursor: Bytes.Cursor}
212  StreamEnd{cursor: Bytes.Cursor}
213  StreamFailure{error: Bytes.Error, cursor: Bytes.Cursor}
214
215# Converts public byte mutations into the durable WAL batch representation.
216def Bytes.mutations(mutations: List<&1, Bytes.Mutation>) -> List<&1, RawByteWal.Mutation>:
217  match mutations:
218    case Nil{}: Nil{}
219    case Con{BytesPut{key, value}, tail}:
220      Con{RawByteWal.Put{key, value}, Bytes.mutations(tail)}
221    case Con{BytesDelete{key}, tail}:
222      Con{RawByteWal.Delete{key}, Bytes.mutations(tail)}
223
224
225def Bytes.handle(
226  result: Result<&1, &1, RawByteDb.Error, RawByteDb.Handle>
227) -> Result<&1, &1, Bytes.Error, Bytes.Handle>:
228  match result:
229    case Fail{error}: Fail{Bytes.error(error)}
230    case Done{raw}: Done{ByteHandle{raw}}
231
232def Bytes.remote.handshake.validated(
233  session: OwnerClient.Session,
234  decision: ResourceOptions.Decision
235) -> Result<&1, &1, Bytes.Error, Bytes.Handle>:
236  match decision:
237    case ResourceOptions.Accepted{_}: Done{RemoteByteHandle{session}}
238    case _: Fail{Corruption{"owner options are invalid"}}
239
240def Bytes.remote.handshake.result(
241  pair: OwnerClient.Session &
242    Result<&1, &1, OwnerClient.Error, OwnerOptionsCodec.Decision>
243) -> Result<&1, &1, Bytes.Error, Bytes.Handle>:
244  match pair:
245    case (_, Fail{error}): Fail{Bytes.owner.error(error)}
246    case (session, Done{OwnerOptionsCodec.Parsed{options}}):
247      Bytes.remote.handshake.validated(session,
248        ResourceOptions.validate(options))
249    case (_, Done{_}): Fail{Corruption{"owner options response is invalid"}}
250
251def Bytes.remote.handshake(
252  pair: OwnerClient.Session &
253    Result<&1, &1, OwnerClient.Error, OwnerOptionsCodec.Decision>
254) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
255  IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
256    Bytes.remote.handshake.result(pair))
257
258def Bytes.remote.role(
259  endpoint: String,
260  generation: U32,
261  role: OwnerPolicy.OpenRole
262) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
263  match role:
264    case OwnerPolicy.ExistingOwnerClient{}:
265      do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
266        described : OwnerClient.Session &
267          Result<&1, &1, OwnerClient.Error, OwnerOptionsCodec.Decision> <-
268            OwnerByteClient.describe_owner(
269              OwnerClient.session.open(endpoint, generation))
270        Bytes.remote.handshake(described)
271    case OwnerPolicy.ReturnBusy{}:
272      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>, Fail{Busy{}})
273    case OwnerPolicy.PropagateOpenFailure{}:
274      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
275        Fail{Io{0, "owner role could not be established"}})
276    case OwnerPolicy.LocalOwner{}:
277      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>, Fail{Busy{}})
278
279def Bytes.endpoint.generation(
280  endpoint: String,
281  result: Result<&1, &1, OwnerGenerationIo.Error, U32>
282) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
283  match result:
284    case Fail{OwnerGenerationIo.Host{code, message}}:
285      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>, Fail{Io{code, message}})
286    case Fail{OwnerGenerationIo.Invalid{}}:
287      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
288        Fail{Corruption{"owner generation is invalid"}})
289    case Fail{OwnerGenerationIo.Exhausted{}}:
290      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
291        Fail{Corruption{"owner generation is exhausted"}})
292    case Done{generation}:
293      Bytes.remote.role(endpoint, generation, OwnerPolicy.open.role(
294        OwnerPolicy.LockBusy{}, OwnerPolicy.EndpointPresent{}))
295
296def Bytes.endpoint.checked(
297  canonical: String,
298  endpoint: String,
299  exists: Result<&1, &1, U32 & String, Bool>
300) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
301  match exists:
302    case Fail{(code, message)}:
303      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
304        Fail{Io{code, message}})
305    case Done{True{}}:
306      do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
307        generation : Result<&1, &1, OwnerGenerationIo.Error, U32> <-
308          OwnerGenerationIo.read(canonical ++ "/OWNER.GEN")
309        Bytes.endpoint.generation(endpoint, generation)
310    case Done{False{}}:
311      Bytes.remote.role(endpoint, 0, OwnerPolicy.open.role(
312        OwnerPolicy.LockBusy{}, OwnerPolicy.EndpointMissing{}))
313
314def Bytes.endpoint.path.pair.go(
315  path: String,
316  left: String,
317  right: String
318) -> String & String:
319  match path:
320    case SNil{}: (String.reverse(left), String.reverse(right))
321    case SCon{+character, rest}:
322      Bytes.endpoint.path.pair.go(rest, SCon{character, left},
323        SCon{character, right})
324
325def Bytes.endpoint.path.pair(path: String) -> String & String:
326  Bytes.endpoint.path.pair.go(path, SNil{}, SNil{})
327
328def Bytes.endpoint.exists(
329  paths: String & String & String
330) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
331  match paths:
332    case (canonical, (exists_path, retained_path)):
333      do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
334        exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(exists_path)
335        Bytes.endpoint.checked(canonical, retained_path, exists)
336
337def Bytes.endpoint.result(
338  canonical: String,
339  decision: EndpointPath.Decision
340) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
341  match decision:
342    case EndpointPath.Endpoint{path}:
343      Bytes.endpoint.exists((canonical, Bytes.endpoint.path.pair(path)))
344    case EndpointPath.PathTooLong{}:
345      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
346        Fail{UnsupportedPlatform{}})
347
348def Bytes.canonical.endpoint.paths(
349  paths: String & String
350) ->
351  IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
352  match paths:
353    case (endpoint_path, retained_path):
354      Bytes.endpoint.result(retained_path,
355        EndpointPath.for_canonical_path(endpoint_path))
356
357def Bytes.canonical.endpoint(
358  result: Result<&1, &1, U32 & String, String>
359) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
360  match result:
361    case Fail{(code, message)}:
362      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
363        Fail{Io{code, message}})
364    case Done{canonical}:
365      Bytes.canonical.endpoint.paths(Bytes.endpoint.path.pair(canonical))
366
367def Bytes.owner.candidate(
368  path: String,
369  result: Result<&1, &1, RawByteDb.Error, RawByteDb.Handle>
370) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
371  match result:
372    case Done{raw}:
373      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
374        Done{ByteHandle{raw}})
375    case Fail{RawByteDb.Busy{}}:
376      do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
377        canonical : Result<&1, &1, U32 & String, String> <-
378          Fs.canonical_path(path)
379        Bytes.canonical.endpoint(canonical)
380    case Fail{error}:
381      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
382        Fail{Bytes.error(error)})
383
384def Bytes.owner.wait.endpoint(
385  canonical: String,
386  timeout_ms: U32,
387  backoff_ms: U32,
388  exists: Result<&1, &1, U32 & String, Bool>
389) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
390  match exists:
391    case Fail{(code, message)}:
392      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
393        Fail{Io{code, message}})
394    case Done{True{}}:
395      Bytes.canonical.endpoint.paths(Bytes.endpoint.path.pair(canonical))
396    case Done{False{}}:
397      do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
398        opened : Result<&1, &1, RawByteDb.Error, RawByteDb.Handle> <-
399          RawByteDb.open_existing_wait(canonical, timeout_ms, backoff_ms)
400        return Bytes.handle(opened)
401
402def Bytes.owner.wait.endpoint.path(
403  paths: String & String,
404  timeout_ms: U32,
405  backoff_ms: U32
406) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
407  match paths:
408    case (canonical, endpoint):
409      do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
410        exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(endpoint)
411        Bytes.owner.wait.endpoint(canonical, timeout_ms, backoff_ms,
412          exists)
413
414def Bytes.owner.wait.decided(
415  canonical: String,
416  decision: EndpointPath.Decision,
417  timeout_ms: U32,
418  backoff_ms: U32
419) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
420  match decision:
421    case EndpointPath.Endpoint{endpoint}:
422      Bytes.owner.wait.endpoint.path((canonical, endpoint), timeout_ms,
423        backoff_ms)
424    case EndpointPath.PathTooLong{}:
425      Bytes.owner.wait.endpoint(canonical, timeout_ms, backoff_ms,
426        Done{False{}})
427
428def Bytes.owner.wait.located(
429  paths: String & String,
430  timeout_ms: U32,
431  backoff_ms: U32
432) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
433  match paths:
434    case (canonical, endpoint_source):
435      Bytes.owner.wait.decided(canonical,
436        EndpointPath.for_canonical_path(endpoint_source), timeout_ms,
437        backoff_ms)
438
439def Bytes.owner.wait.canonical(
440  timeout_ms: U32,
441  backoff_ms: U32,
442  result: Result<&1, &1, U32 & String, String>
443) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
444  match result:
445    case Fail{(code, message)}:
446      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
447        Fail{Io{code, message}})
448    case Done{canonical}:
449      Bytes.owner.wait.located(Bytes.endpoint.path.pair(canonical),
450        timeout_ms, backoff_ms)
451
452def Bytes.owner.wait.candidate(
453  path: String,
454  timeout_ms: U32,
455  backoff_ms: U32,
456  result: Result<&1, &1, RawByteDb.Error, RawByteDb.Handle>
457) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
458  match result:
459    case Done{raw}:
460      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
461        Done{ByteHandle{raw}})
462    case Fail{RawByteDb.Busy{}}:
463      do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
464        canonical : Result<&1, &1, U32 & String, String> <-
465          Fs.canonical_path(path)
466        Bytes.owner.wait.canonical(timeout_ms, backoff_ms,
467          canonical)
468    case Fail{error}:
469      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
470        Fail{Bytes.error(error)})
471
472def Bytes.create(+path: String) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
473  do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
474    opened : Result<&1, &1, RawByteDb.Error, RawByteDb.Handle> <-
475      RawByteDb.create(path)
476    Bytes.owner.candidate(path, opened)
477
478def Bytes.open_existing(+path: String) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
479  do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
480    opened : Result<&1, &1, RawByteDb.Error, RawByteDb.Handle> <-
481      RawByteDb.open_existing(path)
482    Bytes.owner.candidate(path, opened)
483
484def Bytes.options.rejected(
485  handle: Bytes.Handle
486) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
487  match handle:
488    case ByteHandle{raw}:
489      do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
490        _closed : Result<&1, &1, RawByteDb.Error, Unit> <- RawByteDb.close(raw)
491        return Fail{InvalidArgument{}}
492    case RemoteByteHandle{_}:
493      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
494        Fail{InvalidArgument{}})
495
496def Bytes.options.decided(
497  decision: OwnerOpenPolicy.Decision,
498  handle: Bytes.Handle
499) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
500  match decision:
501    case OwnerOpenPolicy.UseLocal{}:
502      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>, Done{handle})
503    case OwnerOpenPolicy.UseRemote{}:
504      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>, Done{handle})
505    case OwnerOpenPolicy.RejectOptions{}: Bytes.options.rejected(handle)
506
507def Bytes.options.local(
508  options: ResourceOptions.Options,
509  handle: Bytes.Handle
510) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
511  Bytes.options.decided(OwnerOpenPolicy.choose(True{},
512    ResourceOptions.same(options, ResourceOptions.defaults())), handle)
513
514def Bytes.options.remote.result(
515  pair: OwnerClient.Session &
516    Result<&1, &1, OwnerClient.Error, OwnerOptionsCodec.Decision>,
517  expected: ResourceOptions.Options
518) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
519  match pair:
520    case (_, Fail{error}):
521      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
522        Fail{Bytes.owner.error(error)})
523    case (session, Done{OwnerOptionsCodec.Parsed{options}}):
524      Bytes.options.decided(OwnerOpenPolicy.choose(False{},
525        ResourceOptions.same(options, expected)), RemoteByteHandle{session})
526    case (_, Done{_}):
527      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
528        Fail{Corruption{"owner options response is invalid"}})
529
530def Bytes.options.remote(
531  session: OwnerClient.Session,
532  expected: ResourceOptions.Options
533) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
534  do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
535    described : OwnerClient.Session &
536      Result<&1, &1, OwnerClient.Error, OwnerOptionsCodec.Decision> <-
537        OwnerByteClient.describe_owner(session)
538    Bytes.options.remote.result(described, expected)
539
540def Bytes.options.opened(
541  handle: Bytes.Handle,
542  options: ResourceOptions.Options
543) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
544  match handle:
545    case ByteHandle{_}: Bytes.options.local(options, handle)
546    case RemoteByteHandle{session}: Bytes.options.remote(session, options)
547
548def Bytes.options.opened.result(
549  result: Result<&1, &1, Bytes.Error, Bytes.Handle>,
550  options: ResourceOptions.Options
551) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
552  match result:
553    case Fail{error}:
554      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>, Fail{error})
555    case Done{handle}: Bytes.options.opened(handle, options)
556
557def Bytes.open_existing.options.decision(
558  +path: String,
559  decision: ResourceOptions.Decision
560) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
561  match decision:
562    case ResourceOptions.Accepted{options}:
563      do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
564        opened : Result<&1, &1, Bytes.Error, Bytes.Handle> <-
565          Bytes.open_existing(path)
566        Bytes.options.opened.result(opened, options)
567    case _: IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
568      Fail{InvalidArgument{}})
569
570def Bytes.open_existing_with_options(
571  +path: String,
572  options: ResourceOptions.Options
573) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
574  Bytes.open_existing.options.decision(path,
575    ResourceOptions.validate(options))
576
577def Bytes.create.options.probe.result(
578  result: Result<&1, &1, Bytes.Error, Bytes.Handle>
579) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
580  match result:
581    case Fail{NotFound{}}:
582      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
583        Fail{InvalidArgument{}})
584    case Fail{error}:
585      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>, Fail{error})
586    case Done{RemoteByteHandle{session}}:
587      IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
588        Done{RemoteByteHandle{session}})
589    case Done{ByteHandle{raw}}:
590      do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
591        _closed : Result<&1, &1, RawByteDb.Error, Unit> <- RawByteDb.close(raw)
592        return Fail{AlreadyExists{}}
593
594def Bytes.create.options.probe(
595  +path: String,
596  options: ResourceOptions.Options
597) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
598  do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
599    opened : Result<&1, &1, Bytes.Error, Bytes.Handle> <-
600      Bytes.open_existing_with_options(path, options)
601    Bytes.create.options.probe.result(opened)
602
603def Bytes.create.options.decision(
604  +path: String,
605  options: ResourceOptions.Options,
606  is_default: Bool
607) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
608  match is_default:
609    case True{}: Bytes.create(path)
610    case False{}: Bytes.create.options.probe(path, options)
611
612def Bytes.create.options.validated(
613  +path: String,
614  decision: ResourceOptions.Decision
615) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
616  match decision:
617    case ResourceOptions.Accepted{+options}:
618      Bytes.create.options.decision(path, options,
619        ResourceOptions.same(options, ResourceOptions.defaults()))
620    case _: IO.pure(Result<&1, &1, Bytes.Error, Bytes.Handle>,
621      Fail{InvalidArgument{}})
622
623def Bytes.create_with_options(
624  +path: String,
625  options: ResourceOptions.Options
626) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
627  Bytes.create.options.validated(path, ResourceOptions.validate(options))
628
629def Bytes.open_existing_wait(
630  +path: String,
631  timeout_ms: U32,
632  backoff_ms: U32
633) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>):
634  do IO<Result<&1, &1, Bytes.Error, Bytes.Handle>>:
635    opened : Result<&1, &1, RawByteDb.Error, RawByteDb.Handle> <-
636      RawByteDb.open_existing(path)
637    Bytes.owner.wait.candidate(path, timeout_ms, backoff_ms, opened)
638
639def Bytes.put.result(
640  pair: RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, Unit>
641) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>:
642  match pair:
643    case (raw, result): (ByteHandle{raw}, Bytes.error.result(result))
644
645# Wraps the raw handle returned after committing one public byte batch.
646def Bytes.write_batch.result(
647  pair: RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, Unit>
648) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>:
649  match pair:
650    case (raw, result): (ByteHandle{raw}, Bytes.error.result(result))
651
652# Commits a public binary batch as one WAL frame.
653def Bytes.write_batch(
654  handle: Bytes.Handle,
655  mutations: List<&1, Bytes.Mutation>
656) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
657  match handle:
658    case ByteHandle{raw}:
659      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
660        pair : RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, Unit> <-
661          RawByteDb.write_batch(raw, Bytes.mutations(mutations))
662        return Bytes.write_batch.result(pair)
663    case RemoteByteHandle{session}:
664      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
665        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit> <-
666          OwnerByteClient.write_batch(session, Bytes.mutations(mutations))
667        return Bytes.owner.unit.result(pair)
668
669def Bytes.put(
670  handle: Bytes.Handle,
671  key: PackedBytes.Bytes,
672  value: PackedBytes.Bytes
673) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
674  match handle:
675    case ByteHandle{raw}:
676      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
677        pair : RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, Unit> <- RawByteDb.put(raw, key, value)
678        return Bytes.put.result(pair)
679    case RemoteByteHandle{session}:
680      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
681        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit> <-
682          OwnerByteClient.put(session, key, value)
683        return Bytes.owner.unit.result(pair)
684
685
686def Bytes.get.result(
687  pair: RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, Maybe<&1, PackedBytes.Bytes>>
688) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>:
689  match pair:
690    case (raw, result): (ByteHandle{raw}, Bytes.error.maybe(result))
691
692def Bytes.delete.result(
693  pair: RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, Unit>
694) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>:
695  match pair:
696    case (raw, result): (ByteHandle{raw}, Bytes.error.result(result))
697
698def Bytes.get(
699  handle: Bytes.Handle,
700  key: PackedBytes.Bytes
701) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>):
702  match handle:
703    case ByteHandle{raw}:
704      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>>:
705        pair : RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, Maybe<&1, PackedBytes.Bytes>> <- RawByteDb.get(raw, key)
706        return Bytes.get.result(pair)
707    case RemoteByteHandle{session}:
708      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error,
709        Maybe<&1, PackedBytes.Bytes>>>:
710        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error,
711          Maybe<&1, PackedBytes.Bytes>> <- OwnerByteClient.get(session, key)
712        return Bytes.owner.get.result(pair)
713
714
715def Bytes.delete(
716  handle: Bytes.Handle,
717  key: PackedBytes.Bytes
718) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
719  match handle:
720    case ByteHandle{raw}:
721      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
722        pair : RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, Unit> <- RawByteDb.delete(raw, key)
723        return Bytes.delete.result(pair)
724    case RemoteByteHandle{session}:
725      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
726        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit> <-
727          OwnerByteClient.delete(session, key)
728        return Bytes.owner.unit.result(pair)
729
730# Wraps the handle after publishing the durable byte SST snapshot.
731def Bytes.flush.result(
732  pair: RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, Unit>
733) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>:
734  match pair:
735    case (raw, result): (ByteHandle{raw}, Bytes.error.result(result))
736
737# Publishes the current byte state to an authenticated SST file.
738def Bytes.flush(handle: Bytes.Handle) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
739  match handle:
740    case ByteHandle{raw}:
741      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
742        pair : RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, Unit> <- RawByteDb.flush(raw)
743        return Bytes.flush.result(pair)
744    case RemoteByteHandle{session}:
745      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
746        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit> <-
747          OwnerByteClient.flush(session)
748        return Bytes.owner.unit.result(pair)
749
750# Publishes a canonical byte snapshot with tombstones removed.
751def Bytes.compact(handle: Bytes.Handle) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
752  match handle:
753    case ByteHandle{raw}:
754      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
755        pair : RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, Unit> <- RawByteDb.compact(raw)
756        return Bytes.flush.result(pair)
757    case RemoteByteHandle{session}:
758      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
759        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit> <-
760          OwnerByteClient.compact(session)
761        return Bytes.owner.unit.result(pair)
762
763def Bytes.stats.local(
764  pair: RawByteDb.Handle & RawByteDb.Stats
765) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Stats>:
766  match pair:
767    case (raw, RawByteDb.Stats{entries, payload_bytes}):
768      (ByteHandle{raw}, Done{ByteStats{entries, payload_bytes, 0n, 0n, 0n}})
769
770def Bytes.stats.remote(
771  pair:
772    OwnerClient.Session & Result<&1, &1, OwnerClient.Error,
773      OwnerStatsCodec.Stats>
774) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Stats>:
775  match pair:
776    case (session, Fail{error}):
777      (RemoteByteHandle{session}, Fail{Bytes.owner.error(error)})
778    case (session, Done{OwnerStatsCodec.ByteStats{entries, payload_bytes, snapshots, retained_bytes, cursors}}):
779      (RemoteByteHandle{session}, Done{ByteStats{entries, payload_bytes, snapshots, retained_bytes, cursors}})
780
781def Bytes.stats(handle: Bytes.Handle) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Stats>):
782  match handle:
783    case ByteHandle{raw}:
784      IO.pure(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Stats>,
785        Bytes.stats.local(RawByteDb.stats(raw)))
786    case RemoteByteHandle{session}:
787      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Stats>>:
788        pair : OwnerClient.Session &
789          Result<&1, &1, OwnerClient.Error, OwnerStatsCodec.Stats> <-
790            OwnerByteClient.stats(session)
791        return Bytes.stats.remote(pair)
792
793def Bytes.scan.from.snapshot(
794  lower_pair: Maybe<&1, PackedBytes.Bytes> & Maybe<&1, PackedBytes.Bytes>,
795  upper: Maybe<&1, PackedBytes.Bytes>,
796  prefix: Maybe<&1, PackedBytes.Bytes>,
797  snapshot: RawByteDb.Handle & List<&1, ByteMemTable.Entry>
798) -> Bytes.Handle & Bytes.Cursor:
799  match lower_pair:
800    case (lower, seek_lower):
801      match snapshot:
802        case (next_raw, entries):
803          (ByteHandle{next_raw}, ByteCursor{
804            ByteRangeCursor.open(ByteRangeCursor.seek(entries, seek_lower)),
805            ByteRangeCursor.Bounds{lower, upper, prefix}})
806
807def Bytes.scan.raw(raw: RawByteDb.Handle, bounds: Bytes.ScanBounds) -> Bytes.Handle & Bytes.Cursor:
808  match bounds:
809    case ScanBounds{lower, upper, prefix}:
810      Bytes.scan.from.snapshot(ByteMemTable.value.clone(lower), upper, prefix, RawByteDb.snapshot(raw))
811
812def Bytes.snapshot.limit() -> Nat:
813  Nat.mul(64n, 1048576n)
814
815def Bytes.snapshot.raw(
816  pair: RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, List<&1, ByteMemTable.Entry>>
817) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot>:
818  match pair:
819    case (raw, Fail{error}): (ByteHandle{raw}, Fail{Bytes.error(error)})
820    case (raw, Done{entries}): (ByteHandle{raw}, Done{ByteSnapshot{entries}})
821
822def Bytes.snapshot.remote.opened.sessions(
823  sessions: OwnerClient.Session & OwnerClient.Session,
824  snapshot_id: U32
825) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot>:
826  match sessions:
827    case (handle_session, snapshot_session):
828      (RemoteByteHandle{handle_session},
829        Done{RemoteByteSnapshot{snapshot_session, snapshot_id}})
830
831def Bytes.snapshot.remote.opened(
832  pair: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, U32>
833) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot>:
834  match pair:
835    case (session, Fail{error}):
836      (RemoteByteHandle{session}, Fail{Bytes.owner.error(error)})
837    case (session, Done{snapshot_id}):
838      Bytes.snapshot.remote.opened.sessions(OwnerClient.session.clone(session),
839        snapshot_id)
840
841def Bytes.snapshot(handle: Bytes.Handle) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot>):
842  match handle:
843    case ByteHandle{raw}:
844      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot>>:
845        pair : RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, List<&1, ByteMemTable.Entry>> <-
846          RawByteDb.snapshot_limited(Bytes.snapshot.limit(), raw)
847        return Bytes.snapshot.raw(pair)
848    case RemoteByteHandle{session}:
849      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot>>:
850        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, U32> <-
851          OwnerByteClient.snapshot.open(session)
852        return Bytes.snapshot.remote.opened(pair)
853
854def Bytes.snapshot.clone.local(
855  entries: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
856) -> Bytes.Snapshot & Bytes.Snapshot:
857  match entries:
858    case (first, second): (ByteSnapshot{first}, ByteSnapshot{second})
859
860def Bytes.snapshot.clone.remote(
861  +snapshot_id: U32,
862  sessions: OwnerClient.Session & OwnerClient.Session
863) -> Bytes.Snapshot & Bytes.Snapshot:
864  match sessions:
865    case (first, second):
866      (RemoteByteSnapshot{first, snapshot_id},
867        RemoteByteSnapshot{second, snapshot_id})
868
869def Bytes.snapshot.clone(snapshot: Bytes.Snapshot) -> Bytes.Snapshot & Bytes.Snapshot:
870  match snapshot:
871    case ByteSnapshot{entries}:
872      Bytes.snapshot.clone.local(ByteMemTable.entries.clone(entries))
873    case RemoteByteSnapshot{session, +snapshot_id}:
874      Bytes.snapshot.clone.remote(snapshot_id,
875        OwnerClient.session.clone(session))
876
877def Bytes.snapshot.cursor.ready(
878  snapshot_entries: List<&1, ByteMemTable.Entry>,
879  cursor_entries: List<&1, ByteMemTable.Entry>,
880  lower: Maybe<&1, PackedBytes.Bytes> & Maybe<&1, PackedBytes.Bytes>,
881  bounds: ByteRangeCursor.Bounds
882) -> Bytes.Snapshot & Bytes.Cursor:
883  match lower:
884    case (seek_lower, retained_lower):
885      match bounds:
886        case ByteRangeCursor.Bounds{_, upper, prefix}:
887          (ByteSnapshot{snapshot_entries}, ByteCursor{
888            ByteRangeCursor.open(ByteRangeCursor.seek(cursor_entries, seek_lower)),
889            ByteRangeCursor.Bounds{retained_lower, upper, prefix}})
890
891def Bytes.snapshot.cursor.opened(
892  entries: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>,
893  bounds: ByteRangeCursor.Bounds & ByteRangeCursor.Bounds
894) -> Bytes.Snapshot & Bytes.Cursor:
895  match entries:
896    case (snapshot_entries, cursor_entries):
897      match bounds:
898        case (seek_bounds, retained_bounds):
899          match seek_bounds:
900            case ByteRangeCursor.Bounds{lower, _, _}:
901              Bytes.snapshot.cursor.ready(snapshot_entries, cursor_entries,
902                ByteMemTable.value.clone(lower), retained_bounds)
903
904def Bytes.snapshot.cursor(
905  entries: List<&1, ByteMemTable.Entry>,
906  bounds: Bytes.ScanBounds
907) -> Bytes.Snapshot & Bytes.Cursor:
908  match bounds:
909    case ScanBounds{lower, upper, prefix}:
910      Bytes.snapshot.cursor.opened(
911        ByteMemTable.entries.clone(entries),
912        ByteRangeCursor.bounds.clone(ByteRangeCursor.Bounds{lower, upper, prefix}))
913
914def Bytes.snapshot.cursor.result(
915  pair: Bytes.Snapshot & Bytes.Cursor
916) -> Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>:
917  match pair:
918    case (snapshot, cursor): (snapshot, Done{cursor})
919
920def Bytes.snapshot.cursor.remote.sessions(
921  +snapshot_id: U32,
922  +cursor_id: U32,
923  sessions: OwnerClient.Session & OwnerClient.Session
924) -> Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>:
925  match sessions:
926    case (snapshot_session, cursor_session):
927      (RemoteByteSnapshot{snapshot_session, snapshot_id},
928        Done{ByteOwnerCursor{cursor_session, snapshot_id, cursor_id}})
929
930def Bytes.snapshot.cursor.remote(
931  snapshot_id: U32,
932  pair: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, U32>
933) -> Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>:
934  match pair:
935    case (session, Fail{error}):
936      (RemoteByteSnapshot{session, snapshot_id}, Fail{Bytes.owner.error(error)})
937    case (session, Done{cursor_id}):
938      Bytes.snapshot.cursor.remote.sessions(snapshot_id, cursor_id,
939        OwnerClient.session.clone(session))
940
941def Bytes.snapshot.scan_range.local(
942  snapshot: Bytes.Snapshot,
943  lower: Maybe<&1, PackedBytes.Bytes>,
944  upper: Maybe<&1, PackedBytes.Bytes>
945) -> Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>:
946  match snapshot:
947    case ByteSnapshot{entries}:
948      Bytes.snapshot.cursor.result(
949        Bytes.snapshot.cursor(entries, ScanBounds{lower, upper, None{}}))
950    case RemoteByteSnapshot{_, _}:
951      (snapshot, Fail{UnsupportedOperation{}})
952
953def Bytes.snapshot.scan_prefix.local(
954  snapshot: Bytes.Snapshot,
955  prefix: PackedBytes.Bytes
956) -> Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>:
957  match snapshot:
958    case ByteSnapshot{entries}:
959      Bytes.snapshot.cursor.result(Bytes.snapshot.cursor(entries,
960        ScanBounds{None{}, None{}, Some{prefix}}))
961    case RemoteByteSnapshot{_, _}:
962      (snapshot, Fail{UnsupportedOperation{}})
963
964def Bytes.snapshot.scan_range(
965  snapshot: Bytes.Snapshot,
966  lower: Maybe<&1, PackedBytes.Bytes>,
967  upper: Maybe<&1, PackedBytes.Bytes>
968) -> IO(Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>):
969  match snapshot:
970    case ByteSnapshot{entries}:
971      IO.pure(Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>,
972        Bytes.snapshot.scan_range.local(snapshot, lower, upper))
973    case RemoteByteSnapshot{session, +snapshot_id}:
974      do IO<Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>>:
975        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, U32> <-
976          OwnerByteClient.snapshot.cursor.open.range(session, snapshot_id,
977            lower, upper)
978        return Bytes.snapshot.cursor.remote(snapshot_id, pair)
979
980def Bytes.snapshot.scan_prefix(
981  snapshot: Bytes.Snapshot,
982  prefix: PackedBytes.Bytes
983) -> IO(Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>):
984  match snapshot:
985    case ByteSnapshot{entries}:
986      IO.pure(Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>,
987        Bytes.snapshot.scan_prefix.local(snapshot, prefix))
988    case RemoteByteSnapshot{session, +snapshot_id}:
989      do IO<Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>>:
990        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, U32> <-
991          OwnerByteClient.snapshot.cursor.open.prefix(session, snapshot_id, prefix)
992        return Bytes.snapshot.cursor.remote(snapshot_id, pair)
993
994def Bytes.snapshot.get.local.result(
995  pair: ByteMemTable.Table & Maybe<&1, PackedBytes.Bytes>
996) -> Bytes.Snapshot & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>:
997  match pair:
998    case (ByteMemTable.Table{entries}, value):
999      (ByteSnapshot{entries}, Done{value})
1000
1001def Bytes.snapshot.get.remote(
1002  snapshot_id: U32,
1003  pair: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Maybe<&1, PackedBytes.Bytes>>
1004) -> Bytes.Snapshot & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>:
1005  match pair:
1006    case (session, Fail{error}):
1007      (RemoteByteSnapshot{session, snapshot_id}, Fail{Bytes.owner.error(error)})
1008    case (session, Done{value}):
1009      (RemoteByteSnapshot{session, snapshot_id}, Done{value})
1010
1011def Bytes.snapshot.get(
1012  snapshot: Bytes.Snapshot,
1013  key: PackedBytes.Bytes
1014) -> IO(Bytes.Snapshot & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>):
1015  match snapshot:
1016    case ByteSnapshot{entries}:
1017      IO.pure(Bytes.Snapshot & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>,
1018        Bytes.snapshot.get.local.result(ByteMemTable.get(
1019          ByteMemTable.Table{entries}, key)))
1020    case RemoteByteSnapshot{session, +snapshot_id}:
1021      do IO<Bytes.Snapshot & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>>:
1022        pair : OwnerClient.Session &
1023          Result<&1, &1, OwnerClient.Error, Maybe<&1, PackedBytes.Bytes>> <-
1024            OwnerByteClient.snapshot.get(session, snapshot_id, key)
1025        return Bytes.snapshot.get.remote(snapshot_id, pair)
1026
1027
1028def Bytes.snapshot.close.remote(
1029  pair: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit>
1030) -> Result<&1, &1, Bytes.Error, Unit>:
1031  match pair:
1032    case (_, Fail{error}): Fail{Bytes.owner.error(error)}
1033    case (_, Done{Unit{}}): Done{Unit{}}
1034
1035def Bytes.snapshot.close(snapshot: Bytes.Snapshot) -> IO(Result<&1, &1, Bytes.Error, Unit>):
1036  match snapshot:
1037    case ByteSnapshot{_}:
1038      IO.pure(Result<&1, &1, Bytes.Error, Unit>, Done{Unit{}})
1039    case RemoteByteSnapshot{session, snapshot_id}:
1040      do IO<Result<&1, &1, Bytes.Error, Unit>>:
1041        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit> <-
1042          OwnerByteClient.snapshot.close(session, snapshot_id)
1043        return Bytes.snapshot.close.remote(pair)
1044
1045def Bytes.backup.error(error: BackupCodec.Error) -> Bytes.Error:
1046  match error:
1047    case BackupCodec.UnsupportedPlatform{}: UnsupportedPlatform{}
1048    case BackupCodec.InvalidOrder{}: InvalidArgument{}
1049    case BackupCodec.ResourceLimit{}: ResourceLimit{}
1050    case BackupCodec.CodecFailure{}: Corruption{"backup encoder rejected the snapshot"}
1051    case BackupCodec.SourceFailure{code, message}: Io{code, message}
1052
1053def Bytes.backup.started(
1054  handle: Bytes.Handle,
1055  result: Result<&1, &1, BackupCodec.Error, BackupCodec.Stream>
1056) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.BackupStream>:
1057  match result:
1058    case Fail{error}: (handle, Fail{Bytes.backup.error(error)})
1059    case Done{inner}: (handle, Done{ByteBackupStream{inner}})
1060
1061def Bytes.backup.snapshot(
1062  pair: Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot>
1063) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.BackupStream>:
1064  match pair:
1065    case (handle, Fail{error}): (handle, Fail{error})
1066    case (handle, Done{ByteSnapshot{entries}}):
1067      Bytes.backup.started(handle, BackupCodec.start(entries))
1068    case (handle, Done{RemoteByteSnapshot{_, _}}):
1069      (handle, Fail{UnsupportedOperation{}})
1070
1071# Opens a stable logical export from one retained database snapshot.
1072def Bytes.backup(handle: Bytes.Handle) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.BackupStream>):
1073  match handle:
1074    case ByteHandle{_}:
1075      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.BackupStream>>:
1076        snapshot : Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot> <-
1077          Bytes.snapshot(handle)
1078        return Bytes.backup.snapshot(snapshot)
1079    case RemoteByteHandle{session}:
1080      IO.pure(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.BackupStream>,
1081        (RemoteByteHandle{session}, Fail{UnsupportedOperation{}}))
1082
1083def Bytes.backup.next.result(
1084  result: Result<&1, &1, BackupCodec.Error, BackupCodec.Step>
1085) -> Result<&1, &1, Bytes.Error, Bytes.BackupStep>:
1086  match result:
1087    case Fail{error}: Fail{Bytes.backup.error(error)}
1088    case Done{BackupCodec.End{}}: Done{BackupEnd{}}
1089    case Done{BackupCodec.Chunk{stream, bytes}}:
1090      Done{BackupChunk{ByteBackupStream{stream}, bytes}}
1091
1092# Produces one bounded checksummed backup chunk per call.
1093def Bytes.backup_next(stream: Bytes.BackupStream) -> Result<&1, &1, Bytes.Error, Bytes.BackupStep>:
1094  match stream:
1095    case ByteBackupStream{inner}:
1096      Bytes.backup.next.result(BackupCodec.next(inner))
1097
1098def Bytes.backup_to.error(error: BackupIo.Error) -> Bytes.Error:
1099  match error:
1100    case BackupIo.AlreadyExists{}: AlreadyExists{}
1101    case BackupIo.UnsupportedPlatform{}: UnsupportedPlatform{}
1102    case BackupIo.ResourceLimit{}: ResourceLimit{}
1103    case BackupIo.Io{code, message}: Io{code, message}
1104    case BackupIo.CommitUnknown{code, message}: CommitUnknown{code, message}
1105
1106def Bytes.backup_to.result(
1107  handle: Bytes.Handle,
1108  result: Result<&1, &1, BackupIo.Error, Unit>
1109) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>:
1110  match result:
1111    case Fail{error}: (handle, Fail{Bytes.backup_to.error(error)})
1112    case Done{Unit{}}: (handle, Done{Unit{}})
1113
1114def Bytes.backup_to.opened(
1115  path: String,
1116  entries: Nat,
1117  fuel: Nat,
1118  pair: RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, VisibleByteCursor.Cursor>
1119) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1120  match pair:
1121    case (raw, Fail{error}):
1122      IO.pure(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>,
1123        (ByteHandle{raw}, Fail{Bytes.error(error)}))
1124    case (raw, Done{cursor}):
1125      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
1126        result : Result<&1, &1, BackupIo.Error, Unit> <-
1127          BackupCursorIo.export(path, cursor, entries, fuel)
1128        return Bytes.backup_to.result(ByteHandle{raw}, result)
1129
1130def Bytes.backup_to.rooted(
1131  path: String,
1132  entries: Nat,
1133  fuel: Nat,
1134  pair:
1135    RawByteDb.Handle &
1136    Result<&1, &1, RawByteDb.Error, RawByteDb.SnapshotRoot>
1137) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1138  match pair:
1139    case (raw, Fail{error}):
1140      IO.pure(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>,
1141        (ByteHandle{raw}, Fail{Bytes.error(error)}))
1142    case (raw, Done{root}):
1143      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
1144        opened : Result<&1, &1, RawByteDb.Error, VisibleByteCursor.Cursor> <-
1145          RawByteDb.snapshot.root.cursor.open(root,
1146            ByteRangeCursor.Bounds{None{}, None{}, None{}})
1147        exported : Bytes.Handle & Result<&1, &1, Bytes.Error, Unit> <-
1148          Bytes.backup_to.opened(path, entries, fuel, (raw, opened))
1149        return exported
1150
1151def Bytes.backup_to.budgeted(
1152  path: String,
1153  entries: Nat,
1154  pair: RawByteDb.Handle & Nat
1155) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1156  match pair:
1157    case (raw, fuel):
1158      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
1159        rooted : RawByteDb.Handle &
1160          Result<&1, &1, RawByteDb.Error, RawByteDb.SnapshotRoot> <-
1161            RawByteDb.snapshot_root_limited(67108864n, raw)
1162        Bytes.backup_to.rooted(path, entries, fuel, rooted)
1163
1164def Bytes.backup_to.statsed(
1165  path: String,
1166  pair: RawByteDb.Handle & RawByteDb.Stats
1167) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1168  match pair:
1169    case (raw, RawByteDb.Stats{entries, _}):
1170      Bytes.backup_to.budgeted(path, entries,
1171        RawByteDb.scan.work_budget(raw))
1172
1173def Bytes.backup_to.remote.failed(
1174  handle_session: OwnerClient.Session,
1175  snapshot_id: U32,
1176  session: OwnerClient.Session,
1177  error: OwnerClient.Error
1178) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1179  do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
1180    _closed : OwnerClient.Session &
1181      Result<&1, &1, OwnerClient.Error, Unit> <-
1182        OwnerByteClient.snapshot.close(session, snapshot_id)
1183    return (RemoteByteHandle{handle_session}, Fail{Bytes.owner.error(error)})
1184
1185def Bytes.backup_to.remote.exported(
1186  handle_session: OwnerClient.Session,
1187  result: Result<&1, &1, BackupIo.Error, Unit>
1188) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1189  IO.pure(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>,
1190    Bytes.backup_to.result(RemoteByteHandle{handle_session}, result))
1191
1192def Bytes.backup_to.remote.export.ready(
1193  path: String,
1194  handle_session: OwnerClient.Session,
1195  source: BackupOwnerCursorIo.OwnerSource,
1196  counts: U32 & U32
1197) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1198  match counts:
1199    case (entries, fuel_entries):
1200      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
1201        result : Result<&1, &1, BackupIo.Error, Unit> <-
1202          BackupOwnerCursorIo.export(path, source, U32.to_nat(entries),
1203            Nat.add(32n, Nat.mul(8n, U32.to_nat(fuel_entries))))
1204        Bytes.backup_to.remote.exported(handle_session, result)
1205
1206def Bytes.backup_to.remote.export.ready.copies(
1207  path: String,
1208  handle_session: OwnerClient.Session,
1209  source: BackupOwnerCursorIo.OwnerSource,
1210  counts: U32 & U32 & U32
1211) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1212  match counts:
1213    case (entries, (fuel_entries, _)):
1214      Bytes.backup_to.remote.export.ready(path, handle_session, source,
1215        (entries, fuel_entries))
1216
1217def Bytes.backup_to.remote.export.opened(
1218  path: String,
1219  handle_session: OwnerClient.Session,
1220  snapshot_id: U32,
1221  entry_count: U32,
1222  opened: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, U32>
1223) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1224  match opened:
1225    case (session, Fail{error}):
1226      Bytes.backup_to.remote.failed(handle_session, snapshot_id,
1227        session, error)
1228    case (session, Done{cursor_id}):
1229      Bytes.backup_to.remote.export.ready.copies(path, handle_session,
1230        BackupOwnerCursorIo.OwnerSource{session, snapshot_id,
1231          cursor_id, True{}},
1232        CoordinatorProtocol.u32.duplicate.triple(entry_count))
1233
1234def Bytes.backup_to.remote.cursor.counted.result.copied(
1235  path: String,
1236  handle_session: OwnerClient.Session,
1237  snapshot_ids: U32 & U32 & U32,
1238  pair: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, U32>
1239) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1240  match snapshot_ids:
1241    case (failure_id, (open_id, export_id)):
1242      match pair:
1243        case (session, Fail{error}):
1244          Bytes.backup_to.remote.failed(handle_session, failure_id,
1245            session, error)
1246        case (session, Done{entry_count}):
1247          do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
1248            opened : OwnerClient.Session &
1249              Result<&1, &1, OwnerClient.Error, U32> <-
1250                OwnerByteClient.snapshot.cursor.open(session, open_id)
1251            Bytes.backup_to.remote.export.opened(path, handle_session,
1252              export_id, entry_count, opened)
1253
1254def Bytes.backup_to.remote.cursor.counted.result(
1255  path: String,
1256  handle_session: OwnerClient.Session,
1257  snapshot_id: U32,
1258  pair: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, U32>
1259) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1260  Bytes.backup_to.remote.cursor.counted.result.copied(path,
1261    handle_session,
1262    CoordinatorProtocol.u32.duplicate.triple(snapshot_id), pair)
1263
1264def Bytes.backup_to.remote.cursor.counted(
1265  path: String,
1266  handle_session: OwnerClient.Session,
1267  snapshot_id: U32,
1268  opened: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, U32>
1269) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1270  match opened:
1271    case (session, Fail{error}):
1272      Bytes.backup_to.remote.failed(handle_session, snapshot_id,
1273        session, error)
1274    case (session, Done{cursor_id}):
1275      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
1276        counted : OwnerClient.Session &
1277          Result<&1, &1, OwnerClient.Error, U32> <-
1278            OwnerByteClient.snapshot.cursor.count(session, cursor_id)
1279        Bytes.backup_to.remote.cursor.counted.result(path, handle_session,
1280          snapshot_id, counted)
1281
1282def Bytes.backup_to.remote.snapshot.cloned(
1283  path: String,
1284  sessions: OwnerClient.Session & OwnerClient.Session,
1285  snapshot_ids: U32 & U32 & U32
1286) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1287  match sessions:
1288    case (handle_session, owner_session):
1289      match snapshot_ids:
1290        case (count_id, (export_id, _)):
1291          do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
1292            opened_cursor : OwnerClient.Session &
1293              Result<&1, &1, OwnerClient.Error, U32> <-
1294                OwnerByteClient.snapshot.cursor.open(owner_session, count_id)
1295            Bytes.backup_to.remote.cursor.counted(path, handle_session,
1296              export_id, opened_cursor)
1297
1298def Bytes.backup_to.remote.snapshot.opened(
1299  path: String,
1300  opened: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, U32>
1301) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1302  match opened:
1303    case (session, Fail{error}):
1304      IO.pure(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>,
1305        (RemoteByteHandle{session}, Fail{Bytes.owner.error(error)}))
1306    case (session, Done{snapshot_id}):
1307      Bytes.backup_to.remote.snapshot.cloned(path,
1308        OwnerClient.session.clone(session),
1309        CoordinatorProtocol.u32.duplicate.triple(snapshot_id))
1310
1311def Bytes.backup_to.remote(
1312  path: String,
1313  session: OwnerClient.Session
1314) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1315  do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>>:
1316    opened : OwnerClient.Session &
1317      Result<&1, &1, OwnerClient.Error, U32> <-
1318        OwnerByteClient.snapshot.open(session)
1319    Bytes.backup_to.remote.snapshot.opened(path, opened)
1320
1321# Publishes one consistent byte snapshot as a checksummed backup bundle.
1322def Bytes.backup_to(
1323  path: String,
1324  handle: Bytes.Handle
1325) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>):
1326  match handle:
1327    case ByteHandle{raw}:
1328      Bytes.backup_to.statsed(path, RawByteDb.stats(raw))
1329    case RemoteByteHandle{session}:
1330      Bytes.backup_to.remote(path, session)
1331
1332def Bytes.restore.error(error: RestoreIo.Error) -> Bytes.Error:
1333  match error:
1334    case RestoreIo.InvalidFormat{}: Corruption{"invalid backup format"}
1335    case RestoreIo.ResourceLimit{}: ResourceLimit{}
1336    case RestoreIo.InvalidOrder{}: InvalidArgument{}
1337    case RestoreIo.ChecksumMismatch{}: Corruption{"backup checksum mismatch"}
1338    case RestoreIo.CountMismatch{}: Corruption{"backup entry count mismatch"}
1339    case RestoreIo.Storage{error}: Bytes.error(error)
1340    case RestoreIo.Io{code, message}: Io{code, message}
1341    case RestoreIo.DestinationExists{}: AlreadyExists{}
1342    case RestoreIo.TemporaryExists{}: AlreadyExists{}
1343    case RestoreIo.VerificationFailed{}: Corruption{"restored database verification failed"}
1344    case RestoreIo.CommitUnknown{code, message}: CommitUnknown{code, message}
1345
1346def Bytes.restore.result(
1347  restored: Result<&1, &1, RestoreIo.Error, Unit>
1348) -> IO(Result<&1, &1, Bytes.Error, Unit>):
1349  match restored:
1350    case Fail{error}: IO.pure(Result<&1, &1, Bytes.Error, Unit>,
1351      Fail{Bytes.restore.error(error)})
1352    case Done{Unit{}}:
1353      IO.pure(Result<&1, &1, Bytes.Error, Unit>, Done{Unit{}})
1354
1355# Validates a complete exchange before building and atomically publishing a database.
1356def Bytes.restore(
1357  destination: String,
1358  input: PackedBytes.Bytes
1359) -> IO(Result<&1, &1, Bytes.Error, Unit>):
1360  do IO<Result<&1, &1, Bytes.Error, Unit>>:
1361    restored : Result<&1, &1, RestoreIo.Error, Unit> <-
1362      RestoreIo.restore(destination, input)
1363    converted : Result<&1, &1, Bytes.Error, Unit> <- Bytes.restore.result(restored)
1364    return converted
1365
1366def Bytes.restore_from.result(
1367  restored: Result<&1, &1, RestoreIo.Error, Unit>
1368) -> IO(Result<&1, &1, Bytes.Error, Unit>):
1369  Bytes.restore.result(restored)
1370
1371# Restores a published backup bundle containing its `BACKUP` exchange file.
1372def Bytes.restore_from(
1373  destination: String,
1374  bundle: String
1375) -> IO(Result<&1, &1, Bytes.Error, Unit>):
1376  do IO<Result<&1, &1, Bytes.Error, Unit>>:
1377    restored : Result<&1, &1, RestoreIo.Error, Unit> <-
1378      RestoreIo.restore_from(destination, bundle)
1379    converted : Result<&1, &1, Bytes.Error, Unit> <-
1380      Bytes.restore_from.result(restored)
1381    return converted
1382
1383# Creates a stable ordered scan snapshot from the current visible byte keys.
1384def Bytes.snapshot.live_scan(handle: Bytes.Handle, bounds: Bytes.ScanBounds) -> Bytes.Handle & Bytes.Cursor:
1385  match handle:
1386    case ByteHandle{raw}:
1387      Bytes.scan.raw(raw, bounds)
1388    case RemoteByteHandle{session}:
1389      (RemoteByteHandle{session}, ByteFailureCursor{UnsupportedOperation{}})
1390
1391def Bytes.snapshot.live_scan_range(
1392  handle: Bytes.Handle,
1393  lower: Maybe<&1, PackedBytes.Bytes>,
1394  upper: Maybe<&1, PackedBytes.Bytes>
1395) -> Bytes.Handle & Bytes.Cursor:
1396  Bytes.snapshot.live_scan(handle, ScanBounds{lower, upper, None{}})
1397
1398def Bytes.scan_open.result(
1399  pair: RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, VisibleByteCursor.Cursor>
1400) -> Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>:
1401  match pair:
1402    case (raw, Fail{error}):
1403      (ByteHandle{raw}, Fail{Bytes.error(error)})
1404    case (raw, Done{cursor}):
1405      (ByteHandle{raw}, Done{ByteStreamCursor{cursor}})
1406
1407def Bytes.scan_open.owned.failed(
1408  handle: Bytes.Handle,
1409  snapshot: Bytes.Snapshot,
1410  error: Bytes.Error
1411) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>):
1412  do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>>:
1413    _closed : Result<&1, &1, Bytes.Error, Unit> <- Bytes.snapshot.close(snapshot)
1414    return (handle, Fail{error})
1415
1416def Bytes.scan_open.owned.cursor(
1417  handle: Bytes.Handle,
1418  bounds: ByteRangeCursor.Bounds,
1419  opened: Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>
1420) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>):
1421  match opened:
1422    case (snapshot, Fail{error}):
1423      Bytes.scan_open.owned.failed(handle, snapshot, error)
1424    case (snapshot, Done{cursor}):
1425      IO.pure(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>,
1426        (handle, Done{ByteOwnedCursor{snapshot, cursor, bounds}}))
1427
1428def Bytes.scan_open.owned.bounds(
1429  handle: Bytes.Handle,
1430  snapshot: Bytes.Snapshot,
1431  copies: ByteRangeCursor.Bounds & ByteRangeCursor.Bounds
1432) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>):
1433  match copies:
1434    case (ByteRangeCursor.Bounds{None{}, None{}, Some{prefix}}, retained):
1435      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>>:
1436        opened : Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor> <-
1437          Bytes.snapshot.scan_prefix(snapshot, prefix)
1438        Bytes.scan_open.owned.cursor(handle, retained, opened)
1439    case (ByteRangeCursor.Bounds{lower, upper, _}, retained):
1440      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>>:
1441        opened : Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor> <-
1442          Bytes.snapshot.scan_range(snapshot, lower, upper)
1443        Bytes.scan_open.owned.cursor(handle, retained, opened)
1444
1445def Bytes.scan_open.owned.snapshot(
1446  opened: Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot>,
1447  bounds: Bytes.ScanBounds
1448) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>):
1449  match opened:
1450    case (handle, Fail{error}):
1451      IO.pure(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>,
1452        (handle, Fail{error}))
1453    case (handle, Done{snapshot}):
1454      match bounds:
1455        case ScanBounds{lower, upper, prefix}:
1456          Bytes.scan_open.owned.bounds(handle, snapshot,
1457            ByteRangeCursor.bounds.clone(ByteRangeCursor.Bounds{lower, upper, prefix}))
1458
1459def Bytes.scan_open(
1460  handle: Bytes.Handle,
1461  bounds: Bytes.ScanBounds
1462) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>):
1463  match handle:
1464    case ByteHandle{raw}:
1465      match bounds:
1466        case ScanBounds{lower, upper, prefix}:
1467      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>>:
1468            pair : RawByteDb.Handle & Result<&1, &1, RawByteDb.Error, VisibleByteCursor.Cursor> <-
1469              RawByteDb.scan.open(raw,
1470                ByteRangeCursor.Bounds{lower, upper, prefix})
1471            return Bytes.scan_open.result(pair)
1472    case RemoteByteHandle{session}:
1473      do IO<Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>>:
1474        opened : Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot> <-
1475          Bytes.snapshot(RemoteByteHandle{session})
1476        Bytes.scan_open.owned.snapshot(opened, bounds)
1477
1478
1479def Bytes.scan(
1480  handle: Bytes.Handle,
1481  bounds: Bytes.ScanBounds
1482) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>):
1483  Bytes.scan_open(handle, bounds)
1484
1485def Bytes.scan_range(
1486  handle: Bytes.Handle,
1487  lower: Maybe<&1, PackedBytes.Bytes>,
1488  upper: Maybe<&1, PackedBytes.Bytes>
1489) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>):
1490  Bytes.scan(handle, ScanBounds{lower, upper, None{}})
1491
1492def Bytes.scan_prefix(
1493  handle: Bytes.Handle,
1494  prefix: PackedBytes.Bytes
1495) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>):
1496  Bytes.scan(handle, ScanBounds{None{}, None{}, Some{prefix}})
1497
1498def Bytes.stream.step.result(step: VisibleByteCursor.Step) -> Bytes.StreamStep:
1499  match step:
1500    case VisibleByteCursor.Item{ByteMemTable.Entry{key, Some{value}}, cursor}:
1501      StreamItem{key, value, ByteStreamCursor{cursor}}
1502    case VisibleByteCursor.Item{ByteMemTable.Entry{_, None{}}, cursor}:
1503      StreamSkip{ByteStreamCursor{cursor}}
1504    case VisibleByteCursor.Skip{cursor}:
1505      StreamSkip{ByteStreamCursor{cursor}}
1506    case VisibleByteCursor.End{}:
1507      StreamEnd{ByteStreamCursor{VisibleByteCursor.Ended{}}}
1508    case VisibleByteCursor.Error{error}:
1509      StreamFailure{Bytes.error(RawByteDb.scan.error(error)),
1510        ByteStreamCursor{VisibleByteCursor.Ended{}}}
1511
1512def Bytes.snapshot.stream.step(step: Bytes.ScanStep) -> Bytes.StreamStep:
1513  match step:
1514    case ScanItem{key, value, cursor}: StreamItem{key, value, cursor}
1515    case ScanSkip{cursor}: StreamSkip{cursor}
1516    case ScanEnd{cursor}: StreamEnd{cursor}
1517    case ScanFailure{error, cursor}: StreamFailure{error, cursor}
1518
1519def Bytes.owner.cursor.frame(
1520  +snapshot_id: U32,
1521  +cursor_id: U32,
1522  pair: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, OwnerCursorCodec.Frame>
1523) -> Bytes.StreamStep:
1524  match pair:
1525    case (session, Fail{error}):
1526      StreamFailure{Bytes.owner.error(error),
1527        ByteOwnerCursor{session, snapshot_id, cursor_id}}
1528    case (session, Done{OwnerCursorCodec.CursorEnd{}}):
1529      StreamEnd{ByteClosedCursor{}}
1530    case (session, Done{OwnerCursorCodec.CursorItem{key, value}}):
1531      StreamItem{key, value, ByteOwnerCursor{session, snapshot_id, cursor_id}}
1532
1533def Bytes.scan_next.remote(
1534  session: OwnerClient.Session,
1535  +snapshot_id: U32,
1536  +cursor_id: U32
1537) -> IO(Bytes.StreamStep):
1538  do IO<Bytes.StreamStep>:
1539    pair : OwnerClient.Session &
1540      Result<&1, &1, OwnerClient.Error, OwnerCursorCodec.Frame> <-
1541        OwnerByteClient.snapshot.cursor.next(session, cursor_id)
1542    return Bytes.owner.cursor.frame(snapshot_id, cursor_id, pair)
1543
1544def Bytes.scan_next_io.stream(cursor: VisibleByteCursor.Cursor) -> IO(Bytes.StreamStep):
1545  do IO<Bytes.StreamStep>:
1546    step : VisibleByteCursor.Step <- VisibleByteCursor.next(cursor)
1547    return Bytes.stream.step.result(step)
1548
1549def Bytes.scan_close_io.result(
1550  result: Result<&1, &1, U32 & String, Unit>
1551) -> Result<&1, &1, Bytes.Error, Unit>:
1552  match result:
1553    case Fail{error}: Fail{Bytes.error(RawByteDb.scan.error(error))}
1554    case Done{Unit{}}: Done{Unit{}}
1555
1556def Bytes.scan_close_io.stream(
1557  inner: VisibleByteCursor.Cursor
1558) -> IO(Result<&1, &1, Bytes.Error, Unit>):
1559  do IO<Result<&1, &1, Bytes.Error, Unit>>:
1560    closed : Result<&1, &1, U32 & String, Unit> <- VisibleByteCursor.close(inner)
1561    return Bytes.scan_close_io.result(closed)
1562
1563def Bytes.scan_close(cursor: Bytes.Cursor) -> IO(Result<&1, &1, Bytes.Error, Unit>):
1564  match cursor:
1565    case ByteOwnedCursor{snapshot, inner, _}:
1566      do IO<Result<&1, &1, Bytes.Error, Unit>>:
1567        first : Result<&1, &1, Bytes.Error, Unit> <- Bytes.scan_close(inner)
1568        second : Result<&1, &1, Bytes.Error, Unit> <- Bytes.snapshot.close(snapshot)
1569        return CursorCleanupPolicy.combine(&1, Bytes.Error, first, second)
1570    case ByteStreamCursor{inner}:
1571      Bytes.scan_close_io.stream(inner)
1572    case ByteOwnerCursor{session, _, cursor_id}:
1573      do IO<Result<&1, &1, Bytes.Error, Unit>>:
1574        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit> <-
1575          OwnerByteClient.snapshot.cursor.close(session, cursor_id)
1576        return Bytes.snapshot.close.remote(pair)
1577    case ByteCursor{inner, _}:
1578      IO.pure(Result<&1, &1, Bytes.Error, Unit>, Done{Unit{}})
1579    case ByteClosedCursor{}:
1580      IO.pure(Result<&1, &1, Bytes.Error, Unit>, Done{Unit{}})
1581    case ByteFailureCursor{_}:
1582      IO.pure(Result<&1, &1, Bytes.Error, Unit>, Done{Unit{}})
1583
1584def Bytes.scan_close_io(cursor: Bytes.Cursor) -> IO(Result<&1, &1, Bytes.Error, Unit>):
1585  Bytes.scan_close(cursor)
1586
1587# Advances a byte scan by one matching item, filtered skip, or end marker.
1588def Bytes.scan_next.result(result: ByteRangeCursor.Step) -> Bytes.ScanStep:
1589  match result:
1590    case ByteRangeCursor.RangeItem{ByteMemTable.Entry{key, Some{value}}, next_inner, next_bounds}:
1591      ScanItem{key, value, ByteCursor{next_inner, next_bounds}}
1592    case ByteRangeCursor.RangeItem{ByteMemTable.Entry{_, None{}}, next_inner, next_bounds}:
1593      ScanSkip{ByteCursor{next_inner, next_bounds}}
1594    case ByteRangeCursor.RangeSkip{next_inner, next_bounds}:
1595      ScanSkip{ByteCursor{next_inner, next_bounds}}
1596    case ByteRangeCursor.RangeEnd{next_inner, next_bounds}:
1597      ScanEnd{ByteCursor{next_inner, next_bounds}}
1598
1599def Bytes.scan_next.cloned(
1600  inner: ByteRangeCursor.Cursor,
1601  pair: ByteRangeCursor.Bounds & ByteRangeCursor.Bounds
1602) -> Bytes.ScanStep:
1603  match pair:
1604    case (read_bounds, retained_bounds):
1605      Bytes.scan_next.result(ByteRangeCursor.next(inner, (read_bounds, retained_bounds)))
1606
1607def Bytes.snapshot.scan_next(cursor: Bytes.Cursor) -> Bytes.ScanStep:
1608  match cursor:
1609    case ByteCursor{inner, bounds}:
1610      Bytes.scan_next.cloned(inner, ByteRangeCursor.bounds.clone(bounds))
1611    case ByteOwnerCursor{_, _, _}:
1612      ScanFailure{UnsupportedOperation{}, ByteClosedCursor{}}
1613    case ByteFailureCursor{error}:
1614      ScanFailure{error, ByteClosedCursor{}}
1615    case _: ScanEnd{cursor}
1616
1617# Advances a live byte cursor to a later inclusive lower bound.
1618def Bytes.snapshot.scan_seek(
1619  cursor: Bytes.Cursor,
1620  lower: Maybe<&1, PackedBytes.Bytes>
1621) -> Bytes.Cursor:
1622  match cursor:
1623    case ByteCursor{inner, bounds}:
1624      ByteCursor{ByteRangeCursor.seek.cursor(inner, lower), bounds}
1625    case ByteFailureCursor{_}: ByteClosedCursor{}
1626    case _: cursor
1627
1628# Releases a scan cursor and its retained bounds explicitly.
1629def Bytes.snapshot.scan_close(cursor: Bytes.Cursor) -> Bytes.Cursor:
1630  match cursor:
1631    case ByteCursor{inner, bounds}:
1632      ByteCursor{ByteRangeCursor.close(inner), bounds}
1633    case ByteFailureCursor{_}: ByteClosedCursor{}
1634    case _: cursor
1635
1636def Bytes.scan_next_io.snapshot(
1637  inner: ByteRangeCursor.Cursor,
1638  bounds: ByteRangeCursor.Bounds
1639) -> IO(Bytes.StreamStep):
1640  IO.pure(Bytes.StreamStep,
1641    Bytes.snapshot.stream.step(Bytes.snapshot.scan_next(ByteCursor{inner, bounds})))
1642
1643def Bytes.scan_next.owned.item(
1644  snapshot: Bytes.Snapshot,
1645  inner: Bytes.Cursor,
1646  key: PackedBytes.Bytes,
1647  value: PackedBytes.Bytes,
1648  bounds: ByteRangeCursor.Bounds,
1649  matches: Bool
1650) -> Bytes.StreamStep:
1651  match matches:
1652    case True{}: StreamItem{key, value, ByteOwnedCursor{snapshot, inner, bounds}}
1653    case False{}: StreamSkip{ByteOwnedCursor{snapshot, inner, bounds}}
1654
1655def Bytes.scan_next.owned.checked(
1656  snapshot: Bytes.Snapshot,
1657  inner: Bytes.Cursor,
1658  keys: PackedBytes.Bytes & PackedBytes.Bytes & PackedBytes.Bytes,
1659  value: PackedBytes.Bytes,
1660  copies: ByteRangeCursor.Bounds & ByteRangeCursor.Bounds
1661) -> Bytes.StreamStep:
1662  match keys:
1663    case (key, (range_key, prefix_key)):
1664      match copies:
1665        case (checked, retained):
1666          Bytes.scan_next.owned.item(snapshot, inner, key, value, retained,
1667            ByteRangeCursor.matches.key((range_key, prefix_key), checked))
1668
1669def Bytes.scan_next.owned.ended(
1670  result: Result<&1, &1, Bytes.Error, Unit>
1671) -> Bytes.StreamStep:
1672  match result:
1673    case Done{Unit{}}: StreamEnd{ByteClosedCursor{}}
1674    case Fail{error}: StreamFailure{error, ByteClosedCursor{}}
1675
1676def Bytes.scan_next.owned(
1677  snapshot: Bytes.Snapshot,
1678  bounds: ByteRangeCursor.Bounds,
1679  step: Bytes.StreamStep
1680) -> IO(Bytes.StreamStep):
1681  match step:
1682    case StreamItem{key, value, inner}:
1683      IO.pure(Bytes.StreamStep, Bytes.scan_next.owned.checked(snapshot, inner,
1684        ByteRangeCursor.key.copies.first(ByteMemTable.bytes.clone(key)), value,
1685        ByteRangeCursor.bounds.clone(bounds)))
1686    case StreamSkip{inner}:
1687      IO.pure(Bytes.StreamStep, StreamSkip{ByteOwnedCursor{snapshot, inner, bounds}})
1688    case StreamFailure{error, inner}:
1689      IO.pure(Bytes.StreamStep, StreamFailure{error, ByteOwnedCursor{snapshot, inner, bounds}})
1690    case StreamEnd{inner}:
1691      do IO<Bytes.StreamStep>:
1692        closed : Result<&1, &1, Bytes.Error, Unit> <-
1693          Bytes.scan_close(ByteOwnedCursor{snapshot, inner, bounds})
1694        return Bytes.scan_next.owned.ended(closed)
1695
1696def Bytes.scan_next(cursor: Bytes.Cursor) -> IO(Bytes.StreamStep):
1697  match cursor:
1698    case ByteOwnedCursor{snapshot, inner, bounds}:
1699      do IO<Bytes.StreamStep>:
1700        step : Bytes.StreamStep <- Bytes.scan_next(inner)
1701        Bytes.scan_next.owned(snapshot, bounds, step)
1702    case ByteStreamCursor{inner}:
1703      Bytes.scan_next_io.stream(inner)
1704    case ByteOwnerCursor{session, snapshot_id, cursor_id}:
1705      Bytes.scan_next.remote(session, snapshot_id, cursor_id)
1706    case ByteCursor{inner, bounds}:
1707      Bytes.scan_next_io.snapshot(inner, bounds)
1708    case ByteClosedCursor{}:
1709      IO.pure(Bytes.StreamStep, StreamEnd{ByteClosedCursor{}})
1710    case ByteFailureCursor{error}:
1711      IO.pure(Bytes.StreamStep, StreamFailure{error, ByteClosedCursor{}})
1712
1713def Bytes.scan_next_io(cursor: Bytes.Cursor) -> IO(Bytes.StreamStep):
1714  Bytes.scan_next(cursor)
1715
1716
1717def Bytes.scan_seek.remote(
1718  +snapshot_id: U32,
1719  +cursor_id: U32,
1720  pair: OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit>
1721) -> Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>:
1722  match pair:
1723    case (session, Fail{error}):
1724      (ByteOwnerCursor{session, snapshot_id, cursor_id},
1725        Fail{Bytes.owner.error(error)})
1726    case (session, Done{Unit{}}):
1727      (ByteOwnerCursor{session, snapshot_id, cursor_id}, Done{Unit{}})
1728
1729def Bytes.scan_seek.ids(
1730  session: OwnerClient.Session,
1731  snapshot_id: U32,
1732  ids: U32 & U32,
1733  lower: Maybe<&1, PackedBytes.Bytes>
1734) -> IO(Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>):
1735  match ids:
1736    case (request_id, cursor_id):
1737      do IO<Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>>:
1738        pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit> <-
1739          OwnerByteClient.snapshot.cursor.seek(session, request_id, lower)
1740        return Bytes.scan_seek.remote(snapshot_id, cursor_id, pair)
1741
1742def Bytes.scan_seek.owned(
1743  snapshot: Bytes.Snapshot,
1744  bounds: ByteRangeCursor.Bounds,
1745  pair: Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>
1746) -> Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>:
1747  match pair:
1748    case (inner, result): (ByteOwnedCursor{snapshot, inner, bounds}, result)
1749
1750
1751def Bytes.scan_seek(
1752  cursor: Bytes.Cursor,
1753  lower: Maybe<&1, PackedBytes.Bytes>
1754) -> IO(Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>):
1755  match cursor:
1756    case ByteOwnedCursor{snapshot, inner, bounds}:
1757      do IO<Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>>:
1758        pair : Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit> <-
1759          Bytes.scan_seek(inner, lower)
1760        return Bytes.scan_seek.owned(snapshot, bounds, pair)
1761    case ByteStreamCursor{inner}:
1762      IO.pure(Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>,
1763        (ByteStreamCursor{VisibleByteCursor.seek(inner, lower)}, Done{Unit{}}))
1764    case ByteOwnerCursor{session, snapshot_id, cursor_id}:
1765      Bytes.scan_seek.ids(session, snapshot_id,
1766        OwnerCursorCodec.u32.duplicate(cursor_id), lower)
1767    case ByteCursor{_, _}:
1768      IO.pure(Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>,
1769        (Bytes.snapshot.scan_seek(cursor, lower), Done{Unit{}}))
1770    case ByteClosedCursor{}:
1771      IO.pure(Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>,
1772        (ByteClosedCursor{}, Done{Unit{}}))
1773    case ByteFailureCursor{error}:
1774      IO.pure(Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>,
1775        (ByteClosedCursor{}, Fail{error}))
1776
1777
1778def Bytes.close.result(result: Result<&1, &1, RawByteDb.Error, Unit>) -> Result<&1, &1, Bytes.Error, Unit>:
1779  Bytes.error.result(result)
1780
1781def Bytes.close(handle: Bytes.Handle) -> IO(Result<&1, &1, Bytes.Error, Unit>):
1782  match handle:
1783    case ByteHandle{raw}:
1784      do IO<Result<&1, &1, Bytes.Error, Unit>>:
1785        result : Result<&1, &1, RawByteDb.Error, Unit> <- RawByteDb.close(raw)
1786        return Bytes.close.result(result)
1787    case RemoteByteHandle{_}:
1788      IO.pure(Result<&1, &1, Bytes.Error, Unit>, Done{Unit{}})
1789
1790type Utf8.Error is Type:
1791  Utf8Failure{inner: Bytes.Error}
1792  InvalidUtf8{}
1793  Utf8ResourceLimit{}
1794  Utf8InvalidArgument{}
1795
1796def Utf8.error.from.failure(failure: Utf8CursorPolicy.Failure) -> Utf8.Error:
1797  match failure:
1798    case Utf8CursorPolicy.TextDecodeFailure{}: InvalidUtf8{}
1799    case Utf8CursorPolicy.ReaderFailure{}: Utf8Failure{Io{7, "scan reader previously failed"}}
1800    case Utf8CursorPolicy.ResourceFailure{}: Utf8ResourceLimit{}
1801    case Utf8CursorPolicy.ArgumentFailure{}: Utf8InvalidArgument{}
1802
1803type Utf8.Handle is Type:
1804  Utf8Handle{inner: Bytes.Handle}
1805
1806type Utf8.Mutation is Data:
1807  Utf8SetValue{key: String, value: String}
1808  Utf8DeleteKey{key: String}
1809
1810type Utf8.Stats is Type:
1811  Utf8Stats{inner: Bytes.Stats}
1812
1813type Utf8.CursorState is Type:
1814  Utf8CursorActive{}
1815  Utf8CursorEnded{}
1816  Utf8CursorFailed{failure: Utf8CursorPolicy.Failure}
1817  Utf8CursorClosed{}
1818
1819type Utf8.Cursor is Type:
1820  Utf8Cursor{inner: Bytes.Cursor, state: Utf8.CursorState}
1821
1822type Utf8.Snapshot is Type:
1823  Utf8Snapshot{inner: Bytes.Snapshot}
1824
1825type Utf8.BackupStream is Type:
1826  Utf8BackupStream{inner: Bytes.BackupStream}
1827
1828type Utf8.BackupStep is Type:
1829  Utf8BackupChunk{stream: Utf8.BackupStream, bytes: PackedBytes.Bytes}
1830  Utf8BackupEnd{}
1831
1832type Utf8.ScanStep is Type:
1833  Utf8ScanItem{key: String, value: String, cursor: Utf8.Cursor}
1834  Utf8ScanSkip{cursor: Utf8.Cursor}
1835  Utf8ScanEnd{cursor: Utf8.Cursor}
1836  Utf8ScanFailure{error: Utf8.Error, cursor: Utf8.Cursor}
1837
1838# Converts the internal manifest validation result to the public status.
1839def public_manifest_status(status: FormatInspect.Integrity) -> ManifestStatus:
1840  match status:
1841    case FormatInspect.Valid{}: ManifestValid{}
1842    case FormatInspect.UnsupportedVersion{generation}:
1843      ManifestVersionUnsupported{generation}
1844    case FormatInspect.UnsupportedComparator{generation}:
1845      ManifestComparatorUnsupported{generation}
1846    case FormatInspect.UnrecognizedFormat{}: ManifestFormatUnknown{}
1847    case FormatInspect.Corrupt{}: ManifestCorrupt{}
1848    case FormatInspect.ResourceLimit{}: ManifestResourceLimit{}
1849
1850# Converts the internal format identity and validation into the public record.
1851def public_manifest_inspection(
1852  inspection: FormatInspect.Inspection
1853) -> ManifestInspection:
1854  match inspection:
1855    case FormatInspect.Inspection{identity, integrity}:
1856      match identity:
1857        case FormatInspect.Identity{package_version, storage_version, generation}:
1858          ManifestInspection{package_version, storage_version, generation,
1859            public_manifest_status(integrity)}
1860
1861# Maps database lock failures to inspection-specific public errors.
1862def public_inspection_lock_error(error: DurableError.Error) -> InspectionError:
1863  match error:
1864    case DurableError.Error{DurableError.Busy{}, _, _, _, _}: InspectionBusy{}
1865    case DurableError.Error{DurableError.Timeout{}, _, _, _, _}: InspectionTimeout{}
1866    case DurableError.Error{DurableError.UnsupportedPlatform{}, _, _, _, _}:
1867      InspectionUnsupportedPlatform{}
1868    case DurableError.Error{_, _, _, code, message}: InspectionLockIo{code, message}
1869
1870# Preserves inspection IO and lock errors while mapping valid inspection data.
1871def public_inspection_result(
1872  result: Result<&1, &1, FormatInspectIo.Error, FormatInspect.Inspection>
1873) -> Result<&1, &1, InspectionError, ManifestInspection>:
1874  match result:
1875    case Fail{FormatInspectIo.Lock{error}}:
1876      Fail{public_inspection_lock_error(error)}
1877    case Fail{FormatInspectIo.Host{code, message}}:
1878      Fail{InspectionIo{code, message}}
1879    case Fail{FormatInspectIo.ResourceLimit{}}:
1880      Fail{InspectionIo{7, "manifest exceeds inspection limit"}}
1881    case Done{inspection}: Done{public_manifest_inspection(inspection)}
1882
1883# Reads format identity and integrity under the existing database lock.
1884def inspect(
1885  +dir: String
1886) ->
1887  IO(Result<&1, &1, InspectionError, ManifestInspection>):
1888  do IO<Result<&1, &1, InspectionError, ManifestInspection>>:
1889    result : Result<&1, &1, FormatInspectIo.Error, FormatInspect.Inspection> <-
1890      FormatInspectIo.inspect(dir)
1891    return public_inspection_result(result)
1892
1893# Converts a verified 0.5.0.0 database into a new generation-four destination.
1894def migrate_legacy(
1895  source: String,
1896  destination: String
1897) -> IO(Result<&1, &1, MigrateRunIo.Error, Unit>):
1898  MigrateRunIo.migrate_legacy_source(source, destination)
1899
1900def Utf8.encode.result(
1901  result: Result<&1, &1, U32 & String, PackedBytes.Bytes>
1902) -> Result<&1, &1, Utf8.Error, PackedBytes.Bytes>:
1903  match result:
1904    case Fail{_}: Fail{Utf8ResourceLimit{}}
1905    case Done{bytes}: Done{bytes}
1906
1907def Utf8.encode(+text: String) -> Result<&1, &1, Utf8.Error, PackedBytes.Bytes>:
1908  Utf8.encode.result(Utf8Codec.from_string(text))
1909
1910def Utf8.set_mutation_result(
1911  key: Result<&1, &1, Utf8.Error, PackedBytes.Bytes>,
1912  value: Maybe<&1, Result<&1, &1, Utf8.Error, PackedBytes.Bytes>>
1913) -> Result<&1, &1, Utf8.Error, Bytes.Mutation>:
1914  match key value:
1915    case Fail{error} _:
1916      Fail{error}
1917    case Done{_} Some{Fail{error}}:
1918      Fail{error}
1919    case Done{key_bytes} Some{Done{value_bytes}}:
1920      Done{BytesPut{key_bytes, value_bytes}}
1921    case Done{_} None{}:
1922      Fail{Utf8InvalidArgument{}}
1923
1924def Utf8.delete_mutation_result(
1925  key: Result<&1, &1, Utf8.Error, PackedBytes.Bytes>,
1926  value: Maybe<&1, Result<&1, &1, Utf8.Error, PackedBytes.Bytes>>
1927) -> Result<&1, &1, Utf8.Error, Bytes.Mutation>:
1928  match key value:
1929    case Fail{error} _:
1930      Fail{error}
1931    case Done{_} Some{Fail{error}}:
1932      Fail{error}
1933    case Done{_} Some{Done{_}}:
1934      Fail{Utf8InvalidArgument{}}
1935    case Done{key_bytes} None{}:
1936      Done{BytesDelete{key_bytes}}
1937
1938def Utf8.set_mutations_result(
1939  head: Result<&1, &1, Utf8.Error, Bytes.Mutation>,
1940  tail: Result<&1, &1, Utf8.Error, List<&1, Bytes.Mutation>>
1941) -> Result<&1, &1, Utf8.Error, List<&1, Bytes.Mutation>>:
1942  match head tail:
1943    case Fail{error} _:
1944      Fail{error}
1945    case Done{_} Fail{error}:
1946      Fail{error}
1947    case Done{mutation} Done{mutations}:
1948      Done{Con{mutation, mutations}}
1949
1950def Utf8.delete_mutations_result(
1951  head: Result<&1, &1, Utf8.Error, Bytes.Mutation>,
1952  tail: Result<&1, &1, Utf8.Error, List<&1, Bytes.Mutation>>
1953) -> Result<&1, &1, Utf8.Error, List<&1, Bytes.Mutation>>:
1954  Utf8.set_mutations_result(head, tail)
1955
1956def Utf8.encode_set_mutation(
1957  +key: String,
1958  +value: String,
1959  tail: Result<&1, &1, Utf8.Error, List<&1, Bytes.Mutation>>
1960) -> Result<&1, &1, Utf8.Error, List<&1, Bytes.Mutation>>:
1961  Utf8.set_mutations_result(
1962    Utf8.set_mutation_result(Utf8.encode(key), Some{Utf8.encode(value)}), tail)
1963
1964def Utf8.encode_delete_mutation(
1965  +key: String,
1966  tail: Result<&1, &1, Utf8.Error, List<&1, Bytes.Mutation>>
1967) -> Result<&1, &1, Utf8.Error, List<&1, Bytes.Mutation>>:
1968  Utf8.delete_mutations_result(
1969    Utf8.delete_mutation_result(Utf8.encode(key), None{}), tail)
1970
1971def Utf8.mutations(
1972  mutations: List<&1, Utf8.Mutation>
1973) -> Result<&1, &1, Utf8.Error, List<&1, Bytes.Mutation>>:
1974  match mutations:
1975    case Nil{}: Done{Nil{}}
1976    case Con{mutation, tail}:
1977      match mutation:
1978        case Utf8SetValue{key, value}:
1979          Utf8.encode_set_mutation(key, value, Utf8.mutations(tail))
1980        case Utf8DeleteKey{key}:
1981          Utf8.encode_delete_mutation(key, Utf8.mutations(tail))
1982
1983def Utf8.open.result(
1984  result: Result<&1, &1, Bytes.Error, Bytes.Handle>
1985) -> Result<&1, &1, Utf8.Error, Utf8.Handle>:
1986  match result:
1987    case Fail{error}: Fail{Utf8Failure{error}}
1988    case Done{handle}: Done{Utf8Handle{handle}}
1989
1990def Utf8.create(+path: String) -> IO(Result<&1, &1, Utf8.Error, Utf8.Handle>):
1991  do IO<Result<&1, &1, Utf8.Error, Utf8.Handle>>:
1992    result : Result<&1, &1, Bytes.Error, Bytes.Handle> <- Bytes.create(path)
1993    return Utf8.open.result(result)
1994
1995def Utf8.create_with_options(
1996  +path: String,
1997  options: ResourceOptions.Options
1998) -> IO(Result<&1, &1, Utf8.Error, Utf8.Handle>):
1999  do IO<Result<&1, &1, Utf8.Error, Utf8.Handle>>:
2000    result : Result<&1, &1, Bytes.Error, Bytes.Handle> <-
2001      Bytes.create_with_options(path, options)
2002    return Utf8.open.result(result)
2003
2004def Utf8.open_existing(+path: String) -> IO(Result<&1, &1, Utf8.Error, Utf8.Handle>):
2005  do IO<Result<&1, &1, Utf8.Error, Utf8.Handle>>:
2006    result : Result<&1, &1, Bytes.Error, Bytes.Handle> <- Bytes.open_existing(path)
2007    return Utf8.open.result(result)
2008
2009def Utf8.open_existing_with_options(
2010  +path: String,
2011  options: ResourceOptions.Options
2012) -> IO(Result<&1, &1, Utf8.Error, Utf8.Handle>):
2013  do IO<Result<&1, &1, Utf8.Error, Utf8.Handle>>:
2014    result : Result<&1, &1, Bytes.Error, Bytes.Handle> <-
2015      Bytes.open_existing_with_options(path, options)
2016    return Utf8.open.result(result)
2017
2018def Utf8.open_existing_wait(
2019  +path: String,
2020  timeout_ms: U32,
2021  backoff_ms: U32
2022) -> IO(Result<&1, &1, Utf8.Error, Utf8.Handle>):
2023  do IO<Result<&1, &1, Utf8.Error, Utf8.Handle>>:
2024    result : Result<&1, &1, Bytes.Error, Bytes.Handle> <- Bytes.open_existing_wait(path, timeout_ms, backoff_ms)
2025    return Utf8.open.result(result)
2026
2027def Utf8.get.decode.result(
2028  handle: Bytes.Handle,
2029  result: Result<&1, &1, U32 & String, String>
2030) -> Utf8.Handle & Result<&1, &1, Utf8.Error, Maybe<&2, String>>:
2031  match result:
2032    case Fail{_}: (Utf8Handle{handle}, Fail{InvalidUtf8{}})
2033    case Done{text}: (Utf8Handle{handle}, Done{Some{text}})
2034
2035def Utf8.get.result(
2036  pair: Bytes.Handle & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>
2037) -> Utf8.Handle & Result<&1, &1, Utf8.Error, Maybe<&2, String>>:
2038  match pair:
2039    case (handle, result):
2040      match result:
2041        case Fail{error}: (Utf8Handle{handle}, Fail{Utf8Failure{error}})
2042        case Done{None{}}: (Utf8Handle{handle}, Done{None{}})
2043        case Done{Some{bytes}}:
2044          Utf8.get.decode.result(handle, Utf8Codec.to_string_strict(bytes))
2045
2046def Utf8.get.encoded(
2047  inner: Bytes.Handle,
2048  encoded: Result<&1, &1, Utf8.Error, PackedBytes.Bytes>
2049) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Maybe<&2, String>>):
2050      match encoded:
2051        case Fail{error}:
2052          IO.pure(Utf8.Handle & Result<&1, &1, Utf8.Error, Maybe<&2, String>>,
2053            (Utf8Handle{inner}, Fail{error}))
2054        case Done{key_bytes}:
2055          do IO<Utf8.Handle & Result<&1, &1, Utf8.Error, Maybe<&2, String>>>:
2056            pair : Bytes.Handle & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>> <- Bytes.get(inner, key_bytes)
2057            return Utf8.get.result(pair)
2058
2059def Utf8.get(
2060  handle: Utf8.Handle,
2061  +key: String
2062) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Maybe<&2, String>>):
2063  match handle:
2064    case Utf8Handle{inner}:
2065      Utf8.get.encoded(inner, Utf8.encode(key))
2066
2067def Utf8.write.result(
2068  pair: Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>
2069) -> Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>:
2070  match pair:
2071    case (handle, Fail{error}): (Utf8Handle{handle}, Fail{Utf8Failure{error}})
2072    case (handle, Done{Unit{}}): (Utf8Handle{handle}, Done{Unit{}})
2073
2074def Utf8.write.encoded(
2075  inner: Bytes.Handle,
2076  encoded: Result<&1, &1, Utf8.Error, List<&1, Bytes.Mutation>>
2077) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>):
2078      match encoded:
2079        case Fail{error}:
2080          IO.pure(Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>,
2081            (Utf8Handle{inner}, Fail{error}))
2082        case Done{encoded}:
2083          do IO<Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>>:
2084            pair : Bytes.Handle & Result<&1, &1, Bytes.Error, Unit> <- Bytes.write_batch(inner, encoded)
2085            return Utf8.write.result(pair)
2086
2087def Utf8.write_batch(
2088  handle: Utf8.Handle,
2089  mutations: List<&1, Utf8.Mutation>
2090) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>):
2091  match handle:
2092    case Utf8Handle{inner}:
2093      Utf8.write.encoded(inner, Utf8.mutations(mutations))
2094
2095def Utf8.put(
2096  handle: Utf8.Handle,
2097  +key: String,
2098  +value: String
2099) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>):
2100  Utf8.write_batch(handle, Con{Utf8SetValue{key, value}, Nil{}})
2101
2102def Utf8.delete(
2103  handle: Utf8.Handle,
2104  +key: String
2105) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>):
2106  Utf8.write_batch(handle, Con{Utf8DeleteKey{key}, Nil{}})
2107
2108def Utf8.unit.result(
2109  pair: Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>
2110) -> Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>:
2111  match pair:
2112    case (handle, Fail{error}): (Utf8Handle{handle}, Fail{Utf8Failure{error}})
2113    case (handle, Done{Unit{}}): (Utf8Handle{handle}, Done{Unit{}})
2114
2115def Utf8.flush(handle: Utf8.Handle) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>):
2116  match handle:
2117    case Utf8Handle{inner}:
2118      do IO<Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>>:
2119        pair : Bytes.Handle & Result<&1, &1, Bytes.Error, Unit> <- Bytes.flush(inner)
2120        return Utf8.unit.result(pair)
2121
2122def Utf8.compact(handle: Utf8.Handle) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>):
2123  match handle:
2124    case Utf8Handle{inner}:
2125      do IO<Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>>:
2126        pair : Bytes.Handle & Result<&1, &1, Bytes.Error, Unit> <- Bytes.compact(inner)
2127        return Utf8.unit.result(pair)
2128
2129def Utf8.scan.bound.encoded(
2130  result: Result<&1, &1, Utf8.Error, PackedBytes.Bytes>
2131) -> Result<&1, &1, Utf8.Error, Maybe<&1, PackedBytes.Bytes>>:
2132  match result:
2133    case Fail{error}: Fail{error}
2134    case Done{bytes}: Done{Some{bytes}}
2135
2136def Utf8.scan.bound(
2137  bound: Maybe<&2, String>
2138) -> Result<&1, &1, Utf8.Error, Maybe<&1, PackedBytes.Bytes>>:
2139  match bound:
2140    case None{}: Done{None{}}
2141    case Some{text}:
2142      Utf8.scan.bound.encoded(Utf8.encode(text))
2143
2144def Utf8.scan_seek.byte.result(
2145  pair: Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit>
2146) -> Bytes.Cursor & Result<&1, &1, Utf8.Error, Unit>:
2147  match pair:
2148    case (cursor, Fail{error}): (cursor, Fail{Utf8Failure{error}})
2149    case (cursor, Done{Unit{}}): (cursor, Done{Unit{}})
2150
2151# Applies an already validated text seek bound to an active byte cursor.
2152def Utf8.scan_seek.active(
2153  inner: Bytes.Cursor,
2154  result: Result<&1, &1, Utf8.Error, Maybe<&1, PackedBytes.Bytes>>
2155) -> IO(Bytes.Cursor & Result<&1, &1, Utf8.Error, Unit>):
2156  match result:
2157    case Fail{error}:
2158      IO.pure(Bytes.Cursor & Result<&1, &1, Utf8.Error, Unit>,
2159        (inner, Fail{error}))
2160    case Done{bound}:
2161      do IO<Bytes.Cursor & Result<&1, &1, Utf8.Error, Unit>>:
2162        pair : Bytes.Cursor & Result<&1, &1, Bytes.Error, Unit> <-
2163          Bytes.scan_seek(inner, bound)
2164        return Utf8.scan_seek.byte.result(pair)
2165
2166def Utf8.scan_seek.failed(
2167  byte_cursor: Bytes.Cursor,
2168  error: Utf8.Error
2169) -> Utf8.Cursor & Result<&1, &1, Utf8.Error, Unit>:
2170  match error:
2171    case Utf8Failure{inner}:
2172      (Utf8Cursor{byte_cursor, Utf8CursorFailed{Utf8CursorPolicy.ReaderFailure{}}}, Fail{Utf8Failure{inner}})
2173    case InvalidUtf8{}:
2174      (Utf8Cursor{byte_cursor, Utf8CursorFailed{Utf8CursorPolicy.TextDecodeFailure{}}}, Fail{InvalidUtf8{}})
2175    case Utf8ResourceLimit{}:
2176      (Utf8Cursor{byte_cursor, Utf8CursorFailed{Utf8CursorPolicy.ResourceFailure{}}}, Fail{Utf8ResourceLimit{}})
2177    case Utf8InvalidArgument{}:
2178      (Utf8Cursor{byte_cursor, Utf8CursorFailed{Utf8CursorPolicy.ArgumentFailure{}}}, Fail{Utf8InvalidArgument{}})
2179
2180def Utf8.scan_seek.result(
2181  pair: Bytes.Cursor & Result<&1, &1, Utf8.Error, Unit>
2182) -> Utf8.Cursor & Result<&1, &1, Utf8.Error, Unit>:
2183  match pair:
2184    case (inner, Fail{error}):
2185      Utf8.scan_seek.failed(inner, error)
2186    case (inner, Done{Unit{}}):
2187      (Utf8Cursor{inner, Utf8CursorActive{}}, Done{Unit{}})
2188
2189# Seeks an active text cursor by UTF-8 key, preserving terminal cursors.
2190def Utf8.scan_seek(
2191  cursor: Utf8.Cursor,
2192  lower: Maybe<&2, String>
2193) -> IO(Utf8.Cursor & Result<&1, &1, Utf8.Error, Unit>):
2194  match cursor:
2195    case Utf8Cursor{inner, Utf8CursorActive{}}:
2196      do IO<Utf8.Cursor & Result<&1, &1, Utf8.Error, Unit>>:
2197        pair : Bytes.Cursor & Result<&1, &1, Utf8.Error, Unit> <-
2198          Utf8.scan_seek.active(inner, Utf8.scan.bound(lower))
2199        return Utf8.scan_seek.result(pair)
2200    case Utf8Cursor{inner, state}:
2201      IO.pure(Utf8.Cursor & Result<&1, &1, Utf8.Error, Unit>,
2202        (Utf8Cursor{inner, state}, Done{Unit{}}))
2203
2204def Utf8.scan.range.opened(
2205  pair: Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>
2206) -> Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Cursor>:
2207  match pair:
2208    case (handle, Fail{error}):
2209      (Utf8Handle{handle}, Fail{Utf8Failure{error}})
2210    case (handle, Done{cursor}):
2211      (Utf8Handle{handle}, Done{Utf8Cursor{cursor, Utf8CursorActive{}}})
2212
2213def Utf8.scan.range.result(
2214  handle: Bytes.Handle,
2215  lower: Result<&1, &1, Utf8.Error, Maybe<&1, PackedBytes.Bytes>>,
2216  upper: Result<&1, &1, Utf8.Error, Maybe<&1, PackedBytes.Bytes>>
2217) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Cursor>):
2218  match lower upper:
2219    case Fail{error} _:
2220      IO.pure(Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Cursor>,
2221        (Utf8Handle{handle}, Fail{error}))
2222    case Done{_} Fail{error}:
2223      IO.pure(Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Cursor>,
2224        (Utf8Handle{handle}, Fail{error}))
2225    case Done{lower_bound} Done{upper_bound}:
2226      do IO<Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Cursor>>:
2227        pair : Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor> <-
2228          Bytes.scan_range(handle, lower_bound, upper_bound)
2229        return Utf8.scan.range.opened(pair)
2230
2231def Utf8.scan_range(
2232  handle: Utf8.Handle,
2233  lower: Maybe<&2, String>,
2234  upper: Maybe<&2, String>
2235) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Cursor>):
2236  match handle:
2237    case Utf8Handle{inner}:
2238      Utf8.scan.range.result(inner,
2239        Utf8.scan.bound(lower), Utf8.scan.bound(upper))
2240
2241def Utf8.scan.prefix.opened(
2242  pair: Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor>
2243) -> Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Cursor>:
2244  Utf8.scan.range.opened(pair)
2245
2246def Utf8.scan.prefix.encoded(
2247  handle: Bytes.Handle,
2248  encoded: Result<&1, &1, Utf8.Error, PackedBytes.Bytes>
2249) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Cursor>):
2250  match encoded:
2251    case Fail{error}:
2252      IO.pure(Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Cursor>,
2253        (Utf8Handle{handle}, Fail{error}))
2254    case Done{prefix}:
2255      do IO<Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Cursor>>:
2256        pair : Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Cursor> <-
2257          Bytes.scan_prefix(handle, prefix)
2258        return Utf8.scan.prefix.opened(pair)
2259
2260def Utf8.scan_prefix(
2261  handle: Utf8.Handle,
2262  +prefix: String
2263) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Cursor>):
2264  match handle:
2265    case Utf8Handle{inner}:
2266      Utf8.scan.prefix.encoded(inner, Utf8.encode(prefix))
2267
2268def Utf8.scan.item.failed(
2269  cursor: Bytes.Cursor,
2270  error: Utf8.Error,
2271  failure: Utf8CursorPolicy.Failure
2272) -> Utf8.ScanStep:
2273  Utf8ScanFailure{error, Utf8Cursor{cursor, Utf8CursorFailed{failure}}}
2274
2275def Utf8.scan.item.result(
2276  cursor: Bytes.Cursor,
2277  key: Result<&1, &1, U32 & String, String>,
2278  value: Result<&1, &1, U32 & String, String>
2279) -> Utf8.ScanStep:
2280  match key value:
2281    case Fail{_} _:
2282      Utf8.scan.item.failed(cursor, InvalidUtf8{}, Utf8CursorPolicy.TextDecodeFailure{})
2283    case Done{_} Fail{_}:
2284      Utf8.scan.item.failed(cursor, InvalidUtf8{}, Utf8CursorPolicy.TextDecodeFailure{})
2285    case Done{key_text} Done{value_text}:
2286      Utf8ScanItem{key_text, value_text, Utf8Cursor{cursor, Utf8CursorActive{}}}
2287
2288def Utf8.scan.byte.step(step: Bytes.StreamStep) -> Utf8.ScanStep:
2289  match step:
2290    case StreamItem{key, value, cursor}:
2291      Utf8.scan.item.result(cursor,
2292        Utf8Codec.to_string_strict(key), Utf8Codec.to_string_strict(value))
2293    case StreamSkip{cursor}:
2294      Utf8ScanSkip{Utf8Cursor{cursor, Utf8CursorActive{}}}
2295    case StreamEnd{cursor}:
2296      Utf8ScanEnd{Utf8Cursor{cursor, Utf8CursorEnded{}}}
2297    case StreamFailure{error, cursor}:
2298      Utf8.scan.item.failed(cursor, Utf8Failure{error}, Utf8CursorPolicy.ReaderFailure{})
2299
2300def Utf8.scan_next.failed.ready(
2301  inner: Bytes.Cursor,
2302  failures: Utf8CursorPolicy.Failure & Utf8CursorPolicy.Failure
2303) -> IO(Utf8.ScanStep):
2304  match failures:
2305    case (stored, returned):
2306      IO.pure(Utf8.ScanStep, Utf8ScanFailure{Utf8.error.from.failure(returned),
2307        Utf8Cursor{inner, Utf8CursorFailed{stored}}})
2308
2309def Utf8.scan_next(cursor: Utf8.Cursor) -> IO(Utf8.ScanStep):
2310  match cursor:
2311    case Utf8Cursor{inner, Utf8CursorActive{}}:
2312      do IO<Utf8.ScanStep>:
2313        step : Bytes.StreamStep <- Bytes.scan_next(inner)
2314        return Utf8.scan.byte.step(step)
2315    case Utf8Cursor{_, Utf8CursorEnded{}}:
2316      IO.pure(Utf8.ScanStep, Utf8ScanEnd{cursor})
2317    case Utf8Cursor{inner, Utf8CursorFailed{failure}}:
2318      Utf8.scan_next.failed.ready(inner, Utf8CursorPolicy.duplicate(failure))
2319    case Utf8Cursor{_, Utf8CursorClosed{}}:
2320      IO.pure(Utf8.ScanStep, Utf8ScanEnd{cursor})
2321
2322def Utf8.scan_close.result(
2323  result: Result<&1, &1, Bytes.Error, Unit>
2324) -> Utf8.Cursor & Result<&1, &1, Utf8.Error, Unit>:
2325  match result:
2326    case Fail{error}:
2327      (Utf8Cursor{ByteClosedCursor{}, Utf8CursorClosed{}}, Fail{Utf8Failure{error}})
2328    case Done{Unit{}}:
2329      (Utf8Cursor{ByteClosedCursor{}, Utf8CursorClosed{}}, Done{Unit{}})
2330
2331def Utf8.scan_close(cursor: Utf8.Cursor) -> IO(Utf8.Cursor & Result<&1, &1, Utf8.Error, Unit>):
2332  match cursor:
2333    case Utf8Cursor{inner, _}:
2334      do IO<Utf8.Cursor & Result<&1, &1, Utf8.Error, Unit>>:
2335        closed : Result<&1, &1, Bytes.Error, Unit> <- Bytes.scan_close(inner)
2336        return Utf8.scan_close.result(closed)
2337
2338def Utf8.stats.result(
2339  pair: Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Stats>
2340) -> Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Stats>:
2341  match pair:
2342    case (next, Fail{error}):
2343      (Utf8Handle{next}, Fail{Utf8Failure{error}})
2344    case (next, Done{byte_stats}):
2345      (Utf8Handle{next}, Done{Utf8Stats{byte_stats}})
2346
2347def Utf8.stats(handle: Utf8.Handle) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Stats>):
2348  match handle:
2349    case Utf8Handle{inner}:
2350      do IO<Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Stats>>:
2351        pair : Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Stats> <-
2352          Bytes.stats(inner)
2353        return Utf8.stats.result(pair)
2354
2355def Utf8.close.result(
2356  result: Result<&1, &1, Bytes.Error, Unit>
2357) -> Result<&1, &1, Utf8.Error, Unit>:
2358  match result:
2359    case Fail{error}: Fail{Utf8Failure{error}}
2360    case Done{Unit{}}: Done{Unit{}}
2361
2362def Utf8.close(handle: Utf8.Handle) -> IO(Result<&1, &1, Utf8.Error, Unit>):
2363  match handle:
2364    case Utf8Handle{inner}:
2365      do IO<Result<&1, &1, Utf8.Error, Unit>>:
2366        result : Result<&1, &1, Bytes.Error, Unit> <- Bytes.close(inner)
2367        return Utf8.close.result(result)
2368
2369def Utf8.snapshot.created(
2370  pair: Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot>
2371) -> Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Snapshot>:
2372  match pair:
2373    case (handle, Fail{error}):
2374      (Utf8Handle{handle}, Fail{Utf8Failure{error}})
2375    case (handle, Done{snapshot}):
2376      (Utf8Handle{handle}, Done{Utf8Snapshot{snapshot}})
2377
2378def Utf8.snapshot(handle: Utf8.Handle) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Snapshot>):
2379  match handle:
2380    case Utf8Handle{inner}:
2381      do IO<Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.Snapshot>>:
2382        pair : Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.Snapshot> <-
2383          Bytes.snapshot(inner)
2384        return Utf8.snapshot.created(pair)
2385
2386def Utf8.snapshot.get.decoded(
2387  snapshot: Bytes.Snapshot,
2388  result: Result<&1, &1, U32 & String, String>
2389) -> Utf8.Snapshot & Result<&1, &1, Utf8.Error, Maybe<&1, String>>:
2390  match result:
2391    case Fail{_}: (Utf8Snapshot{snapshot}, Fail{InvalidUtf8{}})
2392    case Done{text}: (Utf8Snapshot{snapshot}, Done{Some{text}})
2393
2394def Utf8.snapshot.get.result(
2395  pair: Bytes.Snapshot & Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>>
2396) -> Utf8.Snapshot & Result<&1, &1, Utf8.Error, Maybe<&1, String>>:
2397  match pair:
2398    case (snapshot, Fail{error}): (Utf8Snapshot{snapshot}, Fail{Utf8Failure{error}})
2399    case (snapshot, Done{None{}}): (Utf8Snapshot{snapshot}, Done{None{}})
2400    case (snapshot, Done{Some{value}}):
2401      Utf8.snapshot.get.decoded(snapshot, Utf8Codec.to_string_strict(value))
2402
2403def Utf8.snapshot.get.encoded(
2404  snapshot: Bytes.Snapshot,
2405  encoded: Result<&1, &1, Utf8.Error, PackedBytes.Bytes>
2406) -> IO(Utf8.Snapshot & Result<&1, &1, Utf8.Error, Maybe<&1, String>>):
2407  match encoded:
2408    case Fail{error}:
2409      IO.pure(Utf8.Snapshot & Result<&1, &1, Utf8.Error, Maybe<&1, String>>,
2410        (Utf8Snapshot{snapshot}, Fail{error}))
2411    case Done{key}:
2412      do IO<Utf8.Snapshot & Result<&1, &1, Utf8.Error, Maybe<&1, String>>>:
2413        pair : Bytes.Snapshot &
2414          Result<&1, &1, Bytes.Error, Maybe<&1, PackedBytes.Bytes>> <-
2415            Bytes.snapshot.get(snapshot, key)
2416        return Utf8.snapshot.get.result(pair)
2417
2418def Utf8.snapshot.get(
2419  snapshot: Utf8.Snapshot,
2420  +key: String
2421) -> IO(Utf8.Snapshot & Result<&1, &1, Utf8.Error, Maybe<&1, String>>):
2422  match snapshot:
2423    case Utf8Snapshot{inner}:
2424      Utf8.snapshot.get.encoded(inner, Utf8.encode(key))
2425
2426def Utf8.snapshot.range.opened(
2427  pair: Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>
2428) -> Utf8.Snapshot & Result<&1, &1, Utf8.Error, Utf8.Cursor>:
2429  match pair:
2430    case (snapshot, Fail{error}): (Utf8Snapshot{snapshot}, Fail{Utf8Failure{error}})
2431    case (snapshot, Done{cursor}):
2432      (Utf8Snapshot{snapshot}, Done{Utf8Cursor{cursor, Utf8CursorActive{}}})
2433
2434def Utf8.snapshot.range.result(
2435  snapshot: Bytes.Snapshot,
2436  lower: Result<&1, &1, Utf8.Error, Maybe<&1, PackedBytes.Bytes>>,
2437  upper: Result<&1, &1, Utf8.Error, Maybe<&1, PackedBytes.Bytes>>
2438) -> IO(Utf8.Snapshot & Result<&1, &1, Utf8.Error, Utf8.Cursor>):
2439  match lower upper:
2440    case Fail{error} _:
2441      IO.pure(Utf8.Snapshot & Result<&1, &1, Utf8.Error, Utf8.Cursor>,
2442        (Utf8Snapshot{snapshot}, Fail{error}))
2443    case Done{_} Fail{error}:
2444      IO.pure(Utf8.Snapshot & Result<&1, &1, Utf8.Error, Utf8.Cursor>,
2445        (Utf8Snapshot{snapshot}, Fail{error}))
2446    case Done{lower_bound} Done{upper_bound}:
2447      do IO<Utf8.Snapshot & Result<&1, &1, Utf8.Error, Utf8.Cursor>>:
2448        pair : Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor> <-
2449          Bytes.snapshot.scan_range(snapshot, lower_bound, upper_bound)
2450        return Utf8.snapshot.range.opened(pair)
2451
2452def Utf8.snapshot.scan_range(
2453  snapshot: Utf8.Snapshot,
2454  lower: Maybe<&2, String>,
2455  upper: Maybe<&2, String>
2456) -> IO(Utf8.Snapshot & Result<&1, &1, Utf8.Error, Utf8.Cursor>):
2457  match snapshot:
2458    case Utf8Snapshot{inner}:
2459      Utf8.snapshot.range.result(inner, Utf8.scan.bound(lower), Utf8.scan.bound(upper))
2460
2461def Utf8.snapshot.prefix.opened(
2462  pair: Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor>
2463) -> Utf8.Snapshot & Result<&1, &1, Utf8.Error, Utf8.Cursor>:
2464  Utf8.snapshot.range.opened(pair)
2465
2466def Utf8.snapshot.scan_prefix.encoded(
2467  snapshot: Bytes.Snapshot,
2468  encoded: Result<&1, &1, Utf8.Error, PackedBytes.Bytes>
2469) -> IO(Utf8.Snapshot & Result<&1, &1, Utf8.Error, Utf8.Cursor>):
2470  match encoded:
2471    case Fail{error}:
2472      IO.pure(Utf8.Snapshot & Result<&1, &1, Utf8.Error, Utf8.Cursor>,
2473        (Utf8Snapshot{snapshot}, Fail{error}))
2474    case Done{bytes}:
2475      do IO<Utf8.Snapshot & Result<&1, &1, Utf8.Error, Utf8.Cursor>>:
2476        pair : Bytes.Snapshot & Result<&1, &1, Bytes.Error, Bytes.Cursor> <-
2477          Bytes.snapshot.scan_prefix(snapshot, bytes)
2478        return Utf8.snapshot.prefix.opened(pair)
2479
2480def Utf8.snapshot.scan_prefix(
2481  snapshot: Utf8.Snapshot,
2482  +prefix: String
2483) -> IO(Utf8.Snapshot & Result<&1, &1, Utf8.Error, Utf8.Cursor>):
2484  match snapshot:
2485    case Utf8Snapshot{inner}:
2486      Utf8.snapshot.scan_prefix.encoded(inner, Utf8.encode(prefix))
2487
2488def Utf8.snapshot.close(snapshot: Utf8.Snapshot) -> IO(Result<&1, &1, Utf8.Error, Unit>):
2489  match snapshot:
2490    case Utf8Snapshot{inner}:
2491      do IO<Result<&1, &1, Utf8.Error, Unit>>:
2492        result : Result<&1, &1, Bytes.Error, Unit> <- Bytes.snapshot.close(inner)
2493        return Utf8.close.result(result)
2494
2495# Opens the same logical byte backup through the text facade.
2496def Utf8.backup.started(
2497  pair: Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.BackupStream>
2498) -> Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.BackupStream>:
2499  match pair:
2500    case (handle, Fail{error}): (Utf8Handle{handle}, Fail{Utf8Failure{error}})
2501    case (handle, Done{inner}):
2502      (Utf8Handle{handle}, Done{Utf8BackupStream{inner}})
2503
2504def Utf8.backup(handle: Utf8.Handle) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.BackupStream>):
2505  match handle:
2506    case Utf8Handle{inner}:
2507      do IO<Utf8.Handle & Result<&1, &1, Utf8.Error, Utf8.BackupStream>>:
2508        pair : Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.BackupStream> <-
2509          Bytes.backup(inner)
2510        return Utf8.backup.started(pair)
2511
2512def Utf8.backup.next.result(
2513  result: Result<&1, &1, Bytes.Error, Bytes.BackupStep>
2514) -> Result<&1, &1, Utf8.Error, Utf8.BackupStep>:
2515  match result:
2516    case Fail{error}: Fail{Utf8Failure{error}}
2517    case Done{BackupEnd{}}: Done{Utf8BackupEnd{}}
2518    case Done{BackupChunk{stream, bytes}}:
2519      Done{Utf8BackupChunk{Utf8BackupStream{stream}, bytes}}
2520
2521# Emits raw backup chunks without attempting text decoding.
2522def Utf8.backup_next(stream: Utf8.BackupStream) -> Result<&1, &1, Utf8.Error, Utf8.BackupStep>:
2523  match stream:
2524    case Utf8BackupStream{inner}:
2525      Utf8.backup.next.result(Bytes.backup_next(inner))
2526
2527def Utf8.backup_to.started(
2528  pair: Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>
2529) -> Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>:
2530  match pair:
2531    case (updated, Fail{error}):
2532      (Utf8Handle{updated}, Fail{Utf8Failure{error}})
2533    case (updated, Done{Unit{}}): (Utf8Handle{updated}, Done{Unit{}})
2534
2535def Utf8.backup_to(
2536  path: String,
2537  handle: Utf8.Handle
2538) -> IO(Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>):
2539  match handle:
2540    case Utf8Handle{inner}:
2541      do IO<Utf8.Handle & Result<&1, &1, Utf8.Error, Unit>>:
2542        pair : Bytes.Handle & Result<&1, &1, Bytes.Error, Unit> <-
2543          Bytes.backup_to(path, inner)
2544        return Utf8.backup_to.started(pair)