1 # Durable LSM key-value database with binary and UTF-8 APIs, atomic batches, snapshots, and backups. 2 import Base 3 import ./src/DurableError.bend as DurableError 4 import bend-kit-bytes@0.3.2.0/bytes.bend as PackedBytes 5 import ./src/ByteDurableDb.bend as RawByteDb 6 import ./src/ByteWal.bend as RawByteWal 7 import ./src/ByteMemTable.bend as ByteMemTable 8 import ./src/scan/RangeCursor.bend as ByteRangeCursor 9 import ./src/scan/VisibleCursor.bend as VisibleByteCursor 10 import ./src/scan/Snapshot.bend as SnapshotPolicy 11 import ./src/scan/Utf8CursorPolicy.bend as Utf8CursorPolicy 12 import ./src/scan/CursorCleanupPolicy.bend as CursorCleanupPolicy 13 import ./src/transfer/Backup.bend as BackupCodec 14 import ./src/transfer/BackupIo.bend as BackupIo 15 import ./src/transfer/BackupCursorIo.bend as BackupCursorIo 16 import ./src/transfer/BackupOwnerCursorIo.bend as BackupOwnerCursorIo 17 import ./src/transfer/RestoreIo.bend as RestoreIo 18 import ./src/transfer/FormatInspect.bend as FormatInspect 19 import ./src/transfer/FormatInspectIo.bend as FormatInspectIo 20 import ./src/transfer/MigrateRunIo.bend as MigrateRunIo 21 import ./src/StorageBytes.bend as Utf8Codec 22 import ./src/Fs.bend as Fs 23 import ./src/coordinator/EndpointPath.bend as EndpointPath 24 import ./src/coordinator/OwnerGenerationIo.bend as OwnerGenerationIo 25 import ./src/coordinator/Client.bend as OwnerClient 26 import ./src/coordinator/ByteClient.bend as OwnerByteClient 27 import ./src/coordinator/CursorCodec.bend as OwnerCursorCodec 28 import ./src/coordinator/Protocol.bend as CoordinatorProtocol 29 import ./src/coordinator/StatsCodec.bend as OwnerStatsCodec 30 import ./src/coordinator/Io.bend as OwnerIo 31 import ./src/coordinator/UnixSocketIo.bend as OwnerSocket 32 import ./src/coordinator/CoordinatorPolicy.bend as OwnerPolicy 33 import ./src/coordinator/OwnerOpenPolicy.bend as OwnerOpenPolicy 34 import ./src/ResourceOptions.bend as ResourceOptions 35 import ./src/coordinator/OptionsCodec.bend as OwnerOptionsCodec 36 37 # Describes manifest validation independently from package/storage identity. 38 type 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. 47 type 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. 52 type InspectionError is Data: 53 InspectionBusy{} 54 InspectionTimeout{} 55 InspectionUnsupportedPlatform{} 56 InspectionLockIo{code: U32, message: String} 57 InspectionIo{code: U32, message: String} 58 59 type 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 73 type Bytes.Stats is Data: 74 ByteStats{entries: Nat, payload_bytes: Nat, snapshots: Nat, retained_bytes: Nat, cursors: Nat} 75 76 type Bytes.Snapshot is Type: 77 ByteSnapshot{entries: List<&1, ByteMemTable.Entry>} 78 RemoteByteSnapshot{session: OwnerClient.Session, snapshot_id: U32} 79 80 type Bytes.BackupStream is Type: 81 ByteBackupStream{inner: BackupCodec.Stream} 82 83 type 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. 88 def 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 101 def 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 106 def 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 113 def 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 129 def 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 135 def 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 142 def 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 149 def 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 161 def 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 170 def 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 181 type Bytes.Handle is Type: 182 ByteHandle{raw: RawByteDb.Handle} 183 RemoteByteHandle{session: OwnerClient.Session} 184 185 type Bytes.Mutation is Type: 186 BytesPut{key: PackedBytes.Bytes, value: PackedBytes.Bytes} 187 BytesDelete{key: PackedBytes.Bytes} 188 189 type Bytes.ScanBounds is Type: 190 ScanBounds{lower: Maybe<&1, PackedBytes.Bytes>, upper: Maybe<&1, PackedBytes.Bytes>, 191 prefix: Maybe<&1, PackedBytes.Bytes>} 192 193 type 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 202 type 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 208 type 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. 216 def 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 225 def 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 232 def 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 240 def 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 251 def 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 258 def 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>: 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 279 def 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 296 def 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>: 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 314 def 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 325 def Bytes.endpoint.path.pair(path: String) -> String & String: 326 Bytes.endpoint.path.pair.go(path, SNil{}, SNil{}) 327 328 def 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>: 334 exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(exists_path) 335 Bytes.endpoint.checked(canonical, retained_path, exists) 336 337 def 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 348 def 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 357 def 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 367 def 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>: 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 384 def 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>: 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 402 def 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>: 410 exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(endpoint) 411 Bytes.owner.wait.endpoint(canonical, timeout_ms, backoff_ms, 412 exists) 413 414 def 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 428 def 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 439 def 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 452 def 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>: 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 472 def Bytes.create(+path: String) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>): 473 do IO>: 474 opened : Result<&1, &1, RawByteDb.Error, RawByteDb.Handle> <- 475 RawByteDb.create(path) 476 Bytes.owner.candidate(path, opened) 477 478 def Bytes.open_existing(+path: String) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>): 479 do IO>: 480 opened : Result<&1, &1, RawByteDb.Error, RawByteDb.Handle> <- 481 RawByteDb.open_existing(path) 482 Bytes.owner.candidate(path, opened) 483 484 def 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>: 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 496 def 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 507 def 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 514 def 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 530 def Bytes.options.remote( 531 session: OwnerClient.Session, 532 expected: ResourceOptions.Options 533 ) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>): 534 do IO>: 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 540 def 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 548 def 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 557 def 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>: 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 570 def 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 577 def 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>: 591 _closed : Result<&1, &1, RawByteDb.Error, Unit> <- RawByteDb.close(raw) 592 return Fail{AlreadyExists{}} 593 594 def Bytes.create.options.probe( 595 +path: String, 596 options: ResourceOptions.Options 597 ) -> IO(Result<&1, &1, Bytes.Error, Bytes.Handle>): 598 do IO>: 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 603 def 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 612 def 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 623 def 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 629 def 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>: 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 639 def 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. 646 def 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. 653 def 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>: 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>: 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 669 def 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>: 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>: 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 686 def 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 692 def 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 698 def 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>>: 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>>: 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 715 def 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>: 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>: 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. 731 def 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. 738 def Bytes.flush(handle: Bytes.Handle) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>): 739 match handle: 740 case ByteHandle{raw}: 741 do IO>: 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>: 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. 751 def Bytes.compact(handle: Bytes.Handle) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>): 752 match handle: 753 case ByteHandle{raw}: 754 do IO>: 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>: 759 pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, Unit> <- 760 OwnerByteClient.compact(session) 761 return Bytes.owner.unit.result(pair) 762 763 def 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 770 def 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 781 def 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>: 788 pair : OwnerClient.Session & 789 Result<&1, &1, OwnerClient.Error, OwnerStatsCodec.Stats> <- 790 OwnerByteClient.stats(session) 791 return Bytes.stats.remote(pair) 792 793 def 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 807 def 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 812 def Bytes.snapshot.limit() -> Nat: 813 Nat.mul(64n, 1048576n) 814 815 def 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 822 def 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 831 def 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 841 def 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>: 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>: 850 pair : OwnerClient.Session & Result<&1, &1, OwnerClient.Error, U32> <- 851 OwnerByteClient.snapshot.open(session) 852 return Bytes.snapshot.remote.opened(pair) 853 854 def 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 860 def 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 869 def 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 877 def 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 891 def 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 904 def 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 914 def 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 920 def 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 930 def 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 941 def 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 953 def 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 964 def 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>: 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 980 def 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>: 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 994 def 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 1001 def 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 1011 def 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>>: 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 1028 def 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 1035 def 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>: 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 1045 def 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 1053 def 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 1061 def 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. 1072 def Bytes.backup(handle: Bytes.Handle) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Bytes.BackupStream>): 1073 match handle: 1074 case ByteHandle{_}: 1075 do IO>: 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 1083 def 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. 1093 def 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 1098 def 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 1106 def 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 1114 def 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>: 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 1130 def 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>: 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 1151 def 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>: 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 1164 def 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 1173 def 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>: 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 1185 def 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 1192 def 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>: 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 1206 def 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 1217 def 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 1234 def 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>: 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 1254 def 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 1264 def 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>: 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 1282 def 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>: 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 1298 def 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 1311 def Bytes.backup_to.remote( 1312 path: String, 1313 session: OwnerClient.Session 1314 ) -> IO(Bytes.Handle & Result<&1, &1, Bytes.Error, Unit>): 1315 do IO>: 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. 1322 def 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 1332 def 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 1346 def 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. 1356 def Bytes.restore( 1357 destination: String, 1358 input: PackedBytes.Bytes 1359 ) -> IO(Result<&1, &1, Bytes.Error, Unit>): 1360 do IO>: 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 1366 def 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. 1372 def Bytes.restore_from( 1373 destination: String, 1374 bundle: String 1375 ) -> IO(Result<&1, &1, Bytes.Error, Unit>): 1376 do IO>: 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. 1384 def 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 1391 def 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 1398 def 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 1407 def 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>: 1413 _closed : Result<&1, &1, Bytes.Error, Unit> <- Bytes.snapshot.close(snapshot) 1414 return (handle, Fail{error}) 1415 1416 def 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 1428 def 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>: 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>: 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 1445 def 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 1459 def 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>: 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>: 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 1479 def 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 1485 def 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 1492 def 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 1498 def 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 1512 def 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 1519 def 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 1533 def Bytes.scan_next.remote( 1534 session: OwnerClient.Session, 1535 +snapshot_id: U32, 1536 +cursor_id: U32 1537 ) -> IO(Bytes.StreamStep): 1538 do IO: 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 1544 def Bytes.scan_next_io.stream(cursor: VisibleByteCursor.Cursor) -> IO(Bytes.StreamStep): 1545 do IO: 1546 step : VisibleByteCursor.Step <- VisibleByteCursor.next(cursor) 1547 return Bytes.stream.step.result(step) 1548 1549 def 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 1556 def Bytes.scan_close_io.stream( 1557 inner: VisibleByteCursor.Cursor 1558 ) -> IO(Result<&1, &1, Bytes.Error, Unit>): 1559 do IO>: 1560 closed : Result<&1, &1, U32 & String, Unit> <- VisibleByteCursor.close(inner) 1561 return Bytes.scan_close_io.result(closed) 1562 1563 def Bytes.scan_close(cursor: Bytes.Cursor) -> IO(Result<&1, &1, Bytes.Error, Unit>): 1564 match cursor: 1565 case ByteOwnedCursor{snapshot, inner, _}: 1566 do IO>: 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>: 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 1584 def 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. 1588 def 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 1599 def 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 1607 def 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. 1618 def 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. 1629 def 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 1636 def 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 1643 def 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 1655 def 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 1669 def 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 1676 def 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: 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 1696 def Bytes.scan_next(cursor: Bytes.Cursor) -> IO(Bytes.StreamStep): 1697 match cursor: 1698 case ByteOwnedCursor{snapshot, inner, bounds}: 1699 do IO: 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 1713 def Bytes.scan_next_io(cursor: Bytes.Cursor) -> IO(Bytes.StreamStep): 1714 Bytes.scan_next(cursor) 1715 1716 1717 def 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 1729 def 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>: 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 1742 def 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 1751 def 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>: 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 1778 def Bytes.close.result(result: Result<&1, &1, RawByteDb.Error, Unit>) -> Result<&1, &1, Bytes.Error, Unit>: 1779 Bytes.error.result(result) 1780 1781 def Bytes.close(handle: Bytes.Handle) -> IO(Result<&1, &1, Bytes.Error, Unit>): 1782 match handle: 1783 case ByteHandle{raw}: 1784 do IO>: 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 1790 type Utf8.Error is Type: 1791 Utf8Failure{inner: Bytes.Error} 1792 InvalidUtf8{} 1793 Utf8ResourceLimit{} 1794 Utf8InvalidArgument{} 1795 1796 def 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 1803 type Utf8.Handle is Type: 1804 Utf8Handle{inner: Bytes.Handle} 1805 1806 type Utf8.Mutation is Data: 1807 Utf8SetValue{key: String, value: String} 1808 Utf8DeleteKey{key: String} 1809 1810 type Utf8.Stats is Type: 1811 Utf8Stats{inner: Bytes.Stats} 1812 1813 type Utf8.CursorState is Type: 1814 Utf8CursorActive{} 1815 Utf8CursorEnded{} 1816 Utf8CursorFailed{failure: Utf8CursorPolicy.Failure} 1817 Utf8CursorClosed{} 1818 1819 type Utf8.Cursor is Type: 1820 Utf8Cursor{inner: Bytes.Cursor, state: Utf8.CursorState} 1821 1822 type Utf8.Snapshot is Type: 1823 Utf8Snapshot{inner: Bytes.Snapshot} 1824 1825 type Utf8.BackupStream is Type: 1826 Utf8BackupStream{inner: Bytes.BackupStream} 1827 1828 type Utf8.BackupStep is Type: 1829 Utf8BackupChunk{stream: Utf8.BackupStream, bytes: PackedBytes.Bytes} 1830 Utf8BackupEnd{} 1831 1832 type 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. 1839 def 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. 1851 def 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. 1862 def 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. 1871 def 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. 1884 def inspect( 1885 +dir: String 1886 ) -> 1887 IO(Result<&1, &1, InspectionError, ManifestInspection>): 1888 do IO>: 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. 1894 def migrate_legacy( 1895 source: String, 1896 destination: String 1897 ) -> IO(Result<&1, &1, MigrateRunIo.Error, Unit>): 1898 MigrateRunIo.migrate_legacy_source(source, destination) 1899 1900 def 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 1907 def Utf8.encode(+text: String) -> Result<&1, &1, Utf8.Error, PackedBytes.Bytes>: 1908 Utf8.encode.result(Utf8Codec.from_string(text)) 1909 1910 def 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 1924 def 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 1938 def 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 1950 def 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 1956 def 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 1964 def 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 1971 def 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 1983 def 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 1990 def Utf8.create(+path: String) -> IO(Result<&1, &1, Utf8.Error, Utf8.Handle>): 1991 do IO>: 1992 result : Result<&1, &1, Bytes.Error, Bytes.Handle> <- Bytes.create(path) 1993 return Utf8.open.result(result) 1994 1995 def Utf8.create_with_options( 1996 +path: String, 1997 options: ResourceOptions.Options 1998 ) -> IO(Result<&1, &1, Utf8.Error, Utf8.Handle>): 1999 do IO>: 2000 result : Result<&1, &1, Bytes.Error, Bytes.Handle> <- Lines 1-2000 of 2544. More: https://jend.leeeet.dev/read.txt?path=mylsm-lsm-store@0.6.0.1/mylsm.bend&start_line=2001