src/transfer/MigrateIo.bend
mylsm-lsm-store@0.6.0.1 · 963 lines · 81 definitions
1import Base
2import mylsm-lsm-store@0.6.0.1/src/DbLock.bend as DbLock
3import mylsm-lsm-store@0.6.0.1/src/DurableError.bend as DurableError
4import mylsm-lsm-store@0.6.0.1/src/Fs.bend as Fs
5import mylsm-lsm-store@0.6.0.1/src/legacy/ManifestIo.bend as LegacyManifestIo
6import mylsm-lsm-store@0.6.0.1/src/Manifest.bend as Manifest
7import mylsm-lsm-store@0.6.0.1/src/MemTable.bend as MemTable
8import mylsm-lsm-store@0.6.0.1/src/Sstable.bend as Sstable
9import mylsm-lsm-store@0.6.0.1/src/SstStreamIo.bend as SstStreamIo
10import mylsm-lsm-store@0.6.0.1/src/Wal.bend as Wal
11import mylsm-lsm-store@0.6.0.1/src/transfer/LegacyWal.bend as LegacyWal
12import mylsm-lsm-store@0.6.0.1/src/SstReadPolicy.bend as SstReadPolicy
13import mylsm-lsm-store@0.6.0.1/src/coordinator/QueuePolicy.bend as QueuePolicy
14import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
15
16
17type Error is Data:
18 LockError{error: DurableError.Error}
19 HostError{code: U32, message: String}
20 InvalidLegacyWal{}
21 InvalidLegacyManifest{}
22 InvalidLegacySst{}
23 ResourceLimit{}
24
25
26type TableValidation is Type:
27 TableValidation{lock: DbLock.Lock, path: String,
28 result: Result<&1, &1, Error, Nat & Nat>,
29 records: List<&2, MemTable.Entry>, batches: List<&2, Wal.Batch>}
30
31
32type SourceCounts is Data:
33 LegacySourceCounts{sst_entries: Nat, wal_frames: Nat, encoded_bytes: Nat}
34
35
36type SourceSnapshot is Data:
37 LegacySourceSnapshot{sst_records: List<&2, MemTable.Entry>,
38 wal_batches: List<&2, Wal.Batch>, sst_entries: Nat, wal_frames: Nat,
39 encoded_bytes: Nat}
40
41
42type LegacyTableWalk is Type:
43 LegacyLevels{lock: DbLock.Lock, read_path: String, release_path: String,
44 levels: List<&2, List<&2, String>>, index: Nat, total_bytes: Nat,
45 entries: Nat, records: List<&2, MemTable.Entry>}
46 LegacyPath{lock: DbLock.Lock, read_path: String, release_path: String,
47 ldir: Maybe<&2, String>, names: List<&2, String>,
48 levels: List<&2, List<&2, String>>, index: Nat, total_bytes: Nat,
49 entries: Nat, records: List<&2, MemTable.Entry>}
50 LegacyNames{lock: DbLock.Lock, read_path: String, release_path: String,
51 levels: List<&2, List<&2, String>>, index: Nat, dir: String,
52 names: List<&2, String>, total_bytes: Nat, entries: Nat,
53 records: List<&2, MemTable.Entry>}
54 LegacySize{lock: DbLock.Lock, read_path: String, release_path: String,
55 levels: List<&2, List<&2, String>>, index: Nat, dir: String,
56 name: String, names: List<&2, String>, total_bytes: Nat, entries: Nat,
57 result: Result<&1, &1, U32 & String, Nat>, records: List<&2, MemTable.Entry>}
58 LegacyBudget{lock: DbLock.Lock, read_path: String, release_path: String,
59 levels: List<&2, List<&2, String>>, index: Nat, dir: String,
60 name: String, names: List<&2, String>, entries: Nat,
61 decision: SstReadPolicy.TotalDecision, records: List<&2, MemTable.Entry>}
62 LegacyRead{lock: DbLock.Lock, read_path: String, release_path: String,
63 levels: List<&2, List<&2, String>>, index: Nat, dir: String,
64 names: List<&2, String>, total_bytes: Nat, entries: Nat,
65 result: Result<&1, &1, U32 & String, Sstable.Table>,
66 records: List<&2, MemTable.Entry>}
67
68def path.clone.pair.go(
69 path: String,
70 left: String,
71 right: String
72) -> String & String:
73 match path:
74 case SNil{}:
75 (String.reverse(left), String.reverse(right))
76 case SCon{+character, rest}:
77 path.clone.pair.go(rest, SCon{character, left}, SCon{character, right})
78
79def path.clone.pair(path: String) -> String & String:
80 path.clone.pair.go(path, SNil{}, SNil{})
81
82def table.validation(
83 lock: DbLock.Lock,
84 path: String,
85 result: Result<&1, &1, Error, Nat & Nat>
86) -> TableValidation:
87 TableValidation{lock, path, result, Nil{}, Nil{}}
88
89def table.validation.records(
90 lock: DbLock.Lock,
91 path: String,
92 result: Result<&1, &1, Error, Nat & Nat>,
93 records: List<&2, MemTable.Entry>
94) -> TableValidation:
95 TableValidation{lock, path, result, records, Nil{}}
96
97def table.validation.snapshot(
98 lock: DbLock.Lock,
99 path: String,
100 result: Result<&1, &1, Error, Nat & Nat>,
101 records: List<&2, MemTable.Entry>,
102 batches: List<&2, Wal.Batch>
103) -> TableValidation:
104 TableValidation{lock, path, result, records, batches}
105
106def table.release.result(
107 value: Result<&1, &1, Error, Nat & Nat>,
108 result: Result<&1, &1, DurableError.Error, Unit>
109) -> Result<&1, &1, Error, Nat & Nat>:
110 match result:
111 case Fail{error}: Fail{LockError{error}}
112 case Done{Unit{}}: value
113
114def table.release(state: TableValidation) -> IO(Result<&1, &1, Error, Nat & Nat>):
115 match state:
116 case TableValidation{lock, path, result, _, _}:
117 do IO<Result<&1, &1, Error, Nat & Nat>>:
118 released : Result<&1, &1, DurableError.Error, Unit> <-
119 DbLock.release(lock, "migrate-verify", path)
120 return table.release.result(result, released)
121
122def table.finished(
123 lock: DbLock.Lock,
124 path: String,
125 result: Result<&1, &1, Error, Nat & Nat>
126) -> IO(TableValidation):
127 IO.pure(TableValidation, table.validation(lock, path, result))
128
129def table.finished.records(
130 lock: DbLock.Lock,
131 path: String,
132 result: Result<&1, &1, Error, Nat & Nat>,
133 records: List<&2, MemTable.Entry>
134) -> IO(TableValidation):
135 IO.pure(TableValidation, table.validation.records(lock, path, result, records))
136
137def table.entries.count.keep.go(
138 entries: List<&2, MemTable.Entry>,
139 reversed: List<&2, MemTable.Entry>,
140 count: Nat
141) -> List<&2, MemTable.Entry> & Nat:
142 match entries:
143 case Nil{}: (List.reverse(&2, MemTable.Entry, reversed), count)
144 case Con{entry, tail}:
145 table.entries.count.keep.go(tail, Con{entry, reversed}, Nat.add(count, 1n))
146
147def table.entries.count.keep(
148 entries: List<&2, MemTable.Entry>
149) -> List<&2, MemTable.Entry> & Nat:
150 table.entries.count.keep.go(entries, Nil{}, 0n)
151
152def table.read.next(
153 read: List<&2, MemTable.Entry> & Nat,
154 lock: DbLock.Lock,
155 read_path: String,
156 release_path: String,
157 levels: List<&2, List<&2, String>>,
158 index: Nat,
159 dir: String,
160 names: List<&2, String>,
161 total_bytes: Nat,
162 prior_entries: Nat,
163 prior_records: List<&2, MemTable.Entry>
164) -> LegacyTableWalk:
165 match read:
166 case (table_entries, table_count):
167 LegacyNames{lock, read_path, release_path, levels, index, dir,
168 names, total_bytes, Nat.add(prior_entries, table_count),
169 List.append(&2, MemTable.Entry, prior_records, table_entries)}
170
171def table.failure(code: U32) -> Error:
172 Bool.pick(Error, U32.is_eq(code, SstStreamIo.resource_limit_error_code()),
173 ResourceLimit{}, InvalidLegacySst{})
174
175def table.manifest.measure(
176 levels: List<&2, List<&2, String>>,
177 index: Nat,
178 acc: List<&2, List<&2, String>>
179) -> List<&2, List<&2, String>> & Nat:
180 match levels:
181 case Nil{}: (List.reverse(&2, List<&2, String>, acc), index)
182 case Con{level, tail}:
183 table.manifest.measure(tail, Nat.add(index, 1n), Con{level, acc})
184
185def table.walk.size(
186 paths: (String & String) & ((String & String) & (String & String)),
187 lock: DbLock.Lock,
188 release_path: String,
189 levels: List<&2, List<&2, String>>,
190 index: Nat,
191 names: List<&2, String>,
192 total_bytes: Nat,
193 entries: Nat,
194 records: List<&2, MemTable.Entry>
195) -> IO(LegacyTableWalk):
196 match paths:
197 case ((file_root, read_path), ((file_name, name), (file_dir, dir))):
198 do IO<LegacyTableWalk>:
199 result : Result<&1, &1, U32 & String, Nat> <-
200 Fs.file_size(file_root ++ "/" ++ file_dir ++ "/" ++ file_name)
201 return LegacySize{lock, read_path, release_path, levels, index,
202 dir, name, names, total_bytes, entries, result, records}
203
204def table.walk.decode(
205 paths: (String & String) & (String & String),
206 lock: DbLock.Lock,
207 release_path: String,
208 levels: List<&2, List<&2, String>>,
209 index: Nat,
210 name: String,
211 names: List<&2, String>,
212 total_bytes: Nat,
213 entries: Nat,
214 records: List<&2, MemTable.Entry>
215) -> IO(LegacyTableWalk):
216 match paths:
217 case ((file_root, read_path), (file_dir, dir)):
218 do IO<LegacyTableWalk>:
219 result : Result<&1, &1, U32 & String, Sstable.Table> <-
220 SstStreamIo.read_table_limited(
221 file_root ++ "/" ++ file_dir ++ "/" ++ name, 67108864n)
222 return LegacyRead{lock, read_path, release_path, levels, index,
223 dir, names, total_bytes, entries, result, records}
224
225def table.walk.level.path(
226 indices: Nat & Nat,
227 lock: DbLock.Lock,
228 read_path: String,
229 release_path: String,
230 names: List<&2, String>,
231 levels: List<&2, List<&2, String>>,
232 total_bytes: Nat,
233 entries: Nat,
234 records: List<&2, MemTable.Entry>
235) -> LegacyTableWalk:
236 match indices:
237 case (lookup_index, next_index):
238 LegacyPath{lock, read_path, release_path,
239 LegacyManifestIo.ldir_of(lookup_index), names, levels, next_index,
240 total_bytes, entries, records}
241
242def table.walk.go(fuel: Nat, state: LegacyTableWalk) -> IO(TableValidation):
243 match fuel:
244 case 0n:
245 match state:
246 case LegacyLevels{lock, _, release_path, _, _, _, _, records}:
247 table.finished(lock, release_path, Fail{ResourceLimit{}})
248 case LegacyPath{lock, _, release_path, _, _, _, _, _, _, records}:
249 table.finished(lock, release_path, Fail{ResourceLimit{}})
250 case LegacyNames{lock, _, release_path, _, _, _, _, _, _, records}:
251 table.finished(lock, release_path, Fail{ResourceLimit{}})
252 case LegacySize{lock, _, release_path, _, _, _, _, _, _, _, _, records}:
253 table.finished(lock, release_path, Fail{ResourceLimit{}})
254 case LegacyBudget{lock, _, release_path, _, _, _, _, _, _, _, records}:
255 table.finished(lock, release_path, Fail{ResourceLimit{}})
256 case LegacyRead{lock, _, release_path, _, _, _, _, _, _, _, records}:
257 table.finished(lock, release_path, Fail{ResourceLimit{}})
258 case 1n+rest:
259 match state:
260 case LegacyLevels{lock, read_path, release_path, levels, index,
261 total_bytes, entries, records}:
262 match levels:
263 case Nil{}:
264 table.finished.records(lock, release_path,
265 Done{(entries, total_bytes)}, records)
266 case Con{names, tail}:
267 table.walk.go(rest, table.walk.level.path(
268 QueuePolicy.nat.clone.pair(index), lock, read_path,
269 release_path, names, tail, total_bytes, entries, records))
270 case LegacyPath{lock, read_path, release_path, ldir, names, levels,
271 index, total_bytes, entries, records}:
272 match ldir:
273 case None{}:
274 table.finished.records(lock, release_path,
275 Fail{InvalidLegacyManifest{}}, records)
276 case Some{dir}:
277 table.walk.go(rest, LegacyNames{lock, read_path, release_path,
278 levels, index, dir, names, total_bytes, entries, records})
279 case LegacyNames{lock, read_path, release_path, levels, index, dir,
280 names, total_bytes, entries, records}:
281 match names:
282 case Nil{}:
283 table.walk.go(rest, LegacyLevels{lock, read_path, release_path,
284 levels, Nat.add(index, 1n), total_bytes, entries, records})
285 case Con{name, tail}:
286 do IO<TableValidation>:
287 next : LegacyTableWalk <- table.walk.size(
288 (path.clone.pair(read_path),
289 (path.clone.pair(name), path.clone.pair(dir))),
290 lock, release_path, levels, index, tail,
291 total_bytes, entries, records)
292 table.walk.go(rest, next)
293 case LegacySize{lock, read_path, release_path, levels, index, dir,
294 name, names, total_bytes, entries, result, records}:
295 match result:
296 case Fail{(code, message)}:
297 table.finished.records(lock, release_path,
298 Fail{HostError{code, message}}, records)
299 case Done{size}:
300 table.walk.go(rest, LegacyBudget{lock, read_path, release_path,
301 levels, index, dir, name, names, entries,
302 SstReadPolicy.total_decision(total_bytes, size, 67108864n),
303 records})
304 case LegacyBudget{lock, read_path, release_path, levels, index, dir,
305 name, names, entries, decision, records}:
306 match decision:
307 case SstReadPolicy.TotalRejected{}:
308 table.finished.records(lock, release_path,
309 Fail{ResourceLimit{}}, records)
310 case SstReadPolicy.TotalAdmitted{next_total}:
311 do IO<TableValidation>:
312 next : LegacyTableWalk <- table.walk.decode(
313 (path.clone.pair(read_path), path.clone.pair(dir)), lock,
314 release_path, levels, index, name, names, next_total,
315 entries, records)
316 table.walk.go(rest, next)
317 case LegacyRead{lock, read_path, release_path, levels, index, dir,
318 names, total_bytes, entries, result, records}:
319 match result:
320 case Fail{(code, _)}:
321 table.finished.records(lock, release_path,
322 Fail{table.failure(code)}, records)
323 case Done{Sstable.Tbl{table_entries, _, _, _, _, _, _, _}}:
324 table.walk.go(rest, table.read.next(
325 table.entries.count.keep(table_entries), lock, read_path,
326 release_path, levels, index, dir, names, total_bytes,
327 entries, records))
328
329def table.walk.start.paths(
330 paths: String & String,
331 lock: DbLock.Lock,
332 release_path: String,
333 measured: List<&2, List<&2, String>> & Nat
334) -> IO(TableValidation):
335 match paths:
336 case (read_path, _):
337 match measured:
338 case (levels, _):
339 table.walk.go(4194305n, LegacyLevels{lock, read_path,
340 release_path, levels, 0n, 0n, 0n, Nil{}})
341
342def table.walk.start(
343 lock: DbLock.Lock,
344 paths: String & String,
345 measured: List<&2, List<&2, String>> & Nat
346) -> IO(TableValidation):
347 match paths:
348 case (read_path, release_path):
349 table.walk.start.paths(path.clone.pair(read_path), lock, release_path,
350 measured)
351
352def table.manifest.loaded(
353 state: TableValidation,
354 result: Result<&1, &1, U32 & String, Manifest.Manifest>
355) -> IO(TableValidation):
356 match state:
357 case TableValidation{lock, path, _, _, _}:
358 match result:
359 case Fail{_}:
360 table.finished(lock, path, Fail{InvalidLegacyManifest{}})
361 case Done{Manifest.M{levels}}:
362 table.walk.start(lock, path.clone.pair(path),
363 table.manifest.measure(levels, 0n, Nil{}))
364
365def table.manifest.read(
366 lock: DbLock.Lock,
367 paths: String & String
368) -> IO(TableValidation):
369 match paths:
370 case (read_path, retained_path):
371 do IO<TableValidation>:
372 manifest : Result<&1, &1, U32 & String, Manifest.Manifest> <-
373 LegacyManifestIo.load_mfst(read_path ++ "/MANIFEST")
374 table.manifest.loaded(
375 table.validation(lock, retained_path, Done{(0n, 0n)}), manifest)
376
377def table.manifest.present(
378 lock: DbLock.Lock,
379 path: String,
380 result: Result<&1, &1, U32 & String, Bool>
381) -> IO(TableValidation):
382 match result:
383 case Fail{(code, message)}:
384 table.finished(lock, path, Fail{HostError{code, message}})
385 case Done{False{}}:
386 table.finished(lock, path, Fail{InvalidLegacyManifest{}})
387 case Done{True{}}:
388 do IO<TableValidation>:
389 paths : String & String = path.clone.pair(path)
390 table.manifest.read(lock, paths)
391
392def table.locked.manifest(
393 lock: DbLock.Lock,
394 paths: String & String
395) -> IO(Result<&1, &1, Error, Nat & Nat>):
396 match paths:
397 case (exists_path, retained_path):
398 do IO<Result<&1, &1, Error, Nat & Nat>>:
399 present : Result<&1, &1, U32 & String, Bool> <-
400 Fs.exists(exists_path ++ "/MANIFEST")
401 validation : TableValidation <-
402 table.manifest.present(lock, retained_path, present)
403 table.release(validation)
404
405def table.locked(
406 path: String,
407 result: Result<&1, &1, DurableError.Error, DbLock.Lock>
408) -> IO(Result<&1, &1, Error, Nat & Nat>):
409 match result:
410 case Fail{error}:
411 IO.pure(Result<&1, &1, Error, Nat & Nat>, Fail{LockError{error}})
412 case Done{lock}:
413 do IO<Result<&1, &1, Error, Nat & Nat>>:
414 paths : String & String = path.clone.pair(path)
415 table.locked.manifest(lock, paths)
416
417def table.lock.acquired(
418 retained_path: String,
419 result: Result<&1, &1, DurableError.Error, DbLock.Lock>
420) -> IO(Result<&1, &1, Error, Nat & Nat>):
421 table.locked(retained_path, result)
422
423def table.lock.present.start(paths: String & String) -> IO(Result<&1, &1, Error, Nat & Nat>):
424 match paths:
425 case (lock_path, retained_path):
426 do IO<Result<&1, &1, Error, Nat & Nat>>:
427 acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <-
428 DbLock.acquire(lock_path, "migrate-verify")
429 table.lock.acquired(retained_path, acquired)
430
431def table.lock.present(
432 path: String,
433 result: Result<&1, &1, U32 & String, Bool>
434) -> IO(Result<&1, &1, Error, Nat & Nat>):
435 match result:
436 case Fail{(code, message)}:
437 IO.pure(Result<&1, &1, Error, Nat & Nat>, Fail{HostError{code, message}})
438 case Done{False{}}:
439 IO.pure(Result<&1, &1, Error, Nat & Nat>,
440 Fail{HostError{2, "database LOCK file is missing"}})
441 case Done{True{}}:
442 do IO<Result<&1, &1, Error, Nat & Nat>>:
443 paths : String & String = path.clone.pair(path)
444 table.lock.present.start(paths)
445
446def table.canonical.lock(
447 paths: String & String
448) -> IO(Result<&1, &1, Error, Nat & Nat>):
449 match paths:
450 case (exists_path, retained_path):
451 do IO<Result<&1, &1, Error, Nat & Nat>>:
452 present : Result<&1, &1, U32 & String, Bool> <-
453 Fs.exists(exists_path ++ "/LOCK")
454 table.lock.present(retained_path, present)
455
456def table.canonical(
457 result: Result<&1, &1, U32 & String, String>
458) -> IO(Result<&1, &1, Error, Nat & Nat>):
459 match result:
460 case Fail{(code, message)}:
461 IO.pure(Result<&1, &1, Error, Nat & Nat>, Fail{HostError{code, message}})
462 case Done{path}:
463 do IO<Result<&1, &1, Error, Nat & Nat>>:
464 paths : String & String = path.clone.pair(path)
465 table.canonical.lock(paths)
466
467
468def table.count.result(
469 result: Result<&1, &1, Error, Nat & Nat>
470) -> Result<&1, &1, Error, Nat>:
471 match result:
472 case Fail{error}: Fail{error}
473 case Done{(entries, _)}: Done{entries}
474
475
476def verify_legacy_tables(path: String) -> IO(Result<&1, &1, Error, Nat>):
477 do IO<Result<&1, &1, Error, Nat>>:
478 canonical : Result<&1, &1, U32 & String, String> <- Fs.canonical_path(path)
479 verified : Result<&1, &1, Error, Nat & Nat> <- table.canonical(canonical)
480 return table.count.result(verified)
481
482
483def max_wal_bytes() -> Nat:
484 67108864n
485
486
487def release.result(
488 value: Result<&1, &1, Error, Nat>,
489 released: Result<&1, &1, DurableError.Error, Unit>
490) -> Result<&1, &1, Error, Nat>:
491 match released:
492 case Fail{error}: Fail{LockError{error}}
493 case Done{Unit{}}: value
494
495
496def release(
497 lock: DbLock.Lock,
498 value: Result<&1, &1, Error, Nat>,
499 path: String
500) -> IO(Result<&1, &1, Error, Nat>):
501 do IO<Result<&1, &1, Error, Nat>>:
502 released : Result<&1, &1, DurableError.Error, Unit> <-
503 DbLock.release(lock, "migrate-verify", path)
504 return release.result(value, released)
505
506
507def host.failed(
508 lock: DbLock.Lock,
509 error: U32 & String,
510 path: String
511) -> IO(Result<&1, &1, Error, Nat>):
512 match error:
513 case (code, message):
514 release(lock, Fail{HostError{code, message}}, path)
515
516
517def host.without_lock(error: U32 & String) -> IO(Result<&1, &1, Error, Nat>):
518 match error:
519 case (code, message):
520 IO.pure(Result<&1, &1, Error, Nat>, Fail{HostError{code, message}})
521
522def wal.read.failure(
523 error: U32 & String,
524 released: Result<&1, &1, DurableError.Error, Unit>
525) -> Result<&1, &1, Error, Nat>:
526 match error:
527 case (code, message):
528 match released:
529 case Fail{lock_error}: Fail{LockError{lock_error}}
530 case Done{Unit{}}: Fail{HostError{code, message}}
531
532def wal.classify(result: LegacyWal.VerifyResult) -> Result<&1, &1, Error, Nat>:
533 match result:
534 case LegacyWal.WalValid{frames}: Done{frames}
535 case LegacyWal.WalInvalid{}: Fail{InvalidLegacyWal{}}
536
537def wal.read.success(
538 bytes: Bytes.Bytes,
539 released: Result<&1, &1, DurableError.Error, Unit>
540) -> Result<&1, &1, Error, Nat>:
541 match released:
542 case Fail{error}: Fail{LockError{error}}
543 case Done{Unit{}}: wal.classify(LegacyWal.verify(bytes))
544
545
546def wal.read.dispatch(
547 result: Result<&1, &1, U32 & String, Bytes.Bytes>,
548 released: Result<&1, &1, DurableError.Error, Unit>
549) -> Result<&1, &1, Error, Nat>:
550 match result:
551 case Fail{error}: wal.read.failure(error, released)
552 case Done{bytes}: wal.read.success(bytes, released)
553
554def wal.read.closed(
555 file: File,
556 result: Result<&1, &1, U32 & String, Bytes.Bytes>,
557 lock: DbLock.Lock,
558 path: String
559) -> IO(Result<&1, &1, Error, Nat>):
560 do IO<Result<&1, &1, Error, Nat>>:
561 _closed : Unit <- File.close(file)
562 released : Result<&1, &1, DurableError.Error, Unit> <-
563 DbLock.release(lock, "migrate-verify", path)
564 return wal.read.dispatch(result, released)
565
566
567def wal.read.pair(
568 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
569 lock: DbLock.Lock,
570 path: String
571) -> IO(Result<&1, &1, Error, Nat>):
572 match pair:
573 case (file, result): wal.read.closed(file, result, lock, path)
574
575
576def wal.opened(
577 lock: DbLock.Lock,
578 size: Nat,
579 opened: Result<&1, &1, U32 & String, File>,
580 path: String
581) -> IO(Result<&1, &1, Error, Nat>):
582 match opened:
583 case Fail{error}: host.failed(lock, error, path)
584 case Done{file}:
585 do IO<Result<&1, &1, Error, Nat>>:
586 read : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
587 Fs.read_at(file, 0, U32.from_nat(size))
588 wal.read.pair(read, lock, path)
589
590
591def wal.sized(
592 lock: DbLock.Lock,
593 size: Nat,
594 +path: String,
595 within_limit: Bool
596) -> IO(Result<&1, &1, Error, Nat>):
597 match within_limit:
598 case False{}: release(lock, Fail{ResourceLimit{}}, path)
599 case True{}:
600 do IO<Result<&1, &1, Error, Nat>>:
601 opened : Result<&1, &1, U32 & String, File> <-
602 File.open(path ++ "/WAL", "r")
603 wal.opened(lock, size, opened, path)
604
605
606def wal.size.result(
607 lock: DbLock.Lock,
608 result: Result<&1, &1, U32 & String, Nat>,
609 path: String
610) -> IO(Result<&1, &1, Error, Nat>):
611 match result:
612 case Fail{error}: host.failed(lock, error, path)
613 case Done{+size}:
614 wal.sized(lock, size, path, Nat.is_le(size, max_wal_bytes()))
615
616
617def wal.present(
618 lock: DbLock.Lock,
619 present: Bool,
620 +path: String
621) -> IO(Result<&1, &1, Error, Nat>):
622 match present:
623 case False{}: release(lock, Done{0n}, path)
624 case True{}:
625 do IO<Result<&1, &1, Error, Nat>>:
626 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path ++ "/WAL")
627 wal.size.result(lock, size, path)
628
629
630def wal.exists(
631 lock: DbLock.Lock,
632 result: Result<&1, &1, U32 & String, Bool>,
633 path: String
634) -> IO(Result<&1, &1, Error, Nat>):
635 match result:
636 case Fail{error}: host.failed(lock, error, path)
637 case Done{present}: wal.present(lock, present, path)
638
639
640def locked(
641 +path: String,
642 result: Result<&1, &1, DurableError.Error, DbLock.Lock>
643) -> IO(Result<&1, &1, Error, Nat>):
644 match result:
645 case Fail{error}:
646 IO.pure(Result<&1, &1, Error, Nat>, Fail{LockError{error}})
647 case Done{lock}:
648 do IO<Result<&1, &1, Error, Nat>>:
649 present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path ++ "/WAL")
650 wal.exists(lock, present, path)
651
652
653def lock.present(
654 +path: String,
655 result: Result<&1, &1, U32 & String, Bool>
656) -> IO(Result<&1, &1, Error, Nat>):
657 match result:
658 case Fail{error}: host.without_lock(error)
659 case Done{False{}}:
660 IO.pure(Result<&1, &1, Error, Nat>, Fail{HostError{2, "database LOCK file is missing"}})
661 case Done{True{}}:
662 do IO<Result<&1, &1, Error, Nat>>:
663 acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <-
664 DbLock.acquire(path, "migrate-verify")
665 locked(path, acquired)
666
667
668def canonical.result(
669 result: Result<&1, &1, U32 & String, String>
670) -> IO(Result<&1, &1, Error, Nat>):
671 match result:
672 case Fail{(code, message)}:
673 IO.pure(Result<&1, &1, Error, Nat>, Fail{HostError{code, message}})
674 case Done{+path}:
675 do IO<Result<&1, &1, Error, Nat>>:
676 present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path ++ "/LOCK")
677 lock.present(path, present)
678
679def wal.batches.measure.go(
680 batches: List<&2, Wal.Batch>,
681 reversed: List<&2, Wal.Batch>,
682 count: Nat
683) -> List<&2, Wal.Batch> & Nat:
684 match batches:
685 case Nil{}: (List.reverse(&2, Wal.Batch, reversed), count)
686 case Con{batch, tail}:
687 wal.batches.measure.go(tail, Con{batch, reversed}, Nat.add(count, 1n))
688
689def wal.batches.measure(batches: List<&2, Wal.Batch>) -> List<&2, Wal.Batch> & Nat:
690 wal.batches.measure.go(batches, Nil{}, 0n)
691
692def wal.locked.valid.batches(
693 measured: List<&2, Wal.Batch> & Nat,
694 lock: DbLock.Lock,
695 path: String,
696 size: Nat,
697 records: List<&2, MemTable.Entry>
698) -> TableValidation:
699 match measured:
700 case (batches, frames):
701 table.validation.snapshot(lock, path,
702 Done{(frames, size)}, records, batches)
703
704def wal.locked.classified(
705 result: LegacyWal.ReplayResult,
706 lock: DbLock.Lock,
707 path: String,
708 size: Nat,
709 records: List<&2, MemTable.Entry>
710) -> TableValidation:
711 match result:
712 case LegacyWal.ReplayInvalid{}:
713 table.validation.records(lock, path, Fail{InvalidLegacyWal{}}, records)
714 case LegacyWal.ReplayValid{batches}:
715 wal.locked.valid.batches(wal.batches.measure(batches), lock, path,
716 size, records)
717
718def wal.locked.read.classify(
719 result: Result<&1, &1, U32 & String, Bytes.Bytes>,
720 lock: DbLock.Lock,
721 path: String,
722 size: Nat,
723 records: List<&2, MemTable.Entry>
724) -> TableValidation:
725 match result:
726 case Fail{(code, message)}:
727 table.validation.records(lock, path, Fail{HostError{code, message}}, records)
728 case Done{bytes}:
729 wal.locked.classified(LegacyWal.replay(bytes), lock, path, size, records)
730
731def wal.locked.read.closed(
732 file: File,
733 result: Result<&1, &1, U32 & String, Bytes.Bytes>,
734 lock: DbLock.Lock,
735 path: String,
736 size: Nat,
737 records: List<&2, MemTable.Entry>
738) -> IO(TableValidation):
739 do IO<TableValidation>:
740 _closed : Unit <- File.close(file)
741 return wal.locked.read.classify(result, lock, path, size, records)
742
743def wal.locked.read.pair(
744 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
745 lock: DbLock.Lock,
746 path: String,
747 size: Nat,
748 records: List<&2, MemTable.Entry>
749) -> IO(TableValidation):
750 match pair:
751 case (file, result): wal.locked.read.closed(file, result, lock, path, size, records)
752
753def wal.locked.opened(
754 lock: DbLock.Lock,
755 sizes: Nat & Nat,
756 opened: Result<&1, &1, U32 & String, File>,
757 path: String,
758 records: List<&2, MemTable.Entry>
759) -> IO(TableValidation):
760 match sizes opened:
761 case (read_size, retained_size) Fail{(code, message)}:
762 IO.pure(TableValidation,
763 table.validation.records(lock, path, Fail{HostError{code, message}}, records))
764 case (read_size, retained_size) Done{file}:
765 do IO<TableValidation>:
766 read : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
767 Fs.read_at(file, 0, U32.from_nat(read_size))
768 wal.locked.read.pair(read, lock, path, retained_size, records)
769
770def wal.locked.sized(
771 lock: DbLock.Lock,
772 size: Nat,
773 +path: String,
774 within_limit: Bool,
775 records: List<&2, MemTable.Entry>
776) -> IO(TableValidation):
777 match within_limit:
778 case False{}:
779 IO.pure(TableValidation,
780 table.validation.records(lock, path, Fail{ResourceLimit{}}, records))
781 case True{}:
782 do IO<TableValidation>:
783 opened : Result<&1, &1, U32 & String, File> <-
784 File.open(path ++ "/WAL", "r")
785 wal.locked.opened(lock, QueuePolicy.nat.clone.pair(size), opened, path,
786 records)
787
788def wal.locked.size.result(
789 lock: DbLock.Lock,
790 result: Result<&1, &1, U32 & String, Nat>,
791 path: String,
792 records: List<&2, MemTable.Entry>
793) -> IO(TableValidation):
794 match result:
795 case Fail{(code, message)}:
796 IO.pure(TableValidation,
797 table.validation.records(lock, path, Fail{HostError{code, message}}, records))
798 case Done{+size}:
799 wal.locked.sized(lock, size, path, Nat.is_le(size, max_wal_bytes()), records)
800
801def wal.locked.present(
802 lock: DbLock.Lock,
803 present: Bool,
804 +path: String,
805 records: List<&2, MemTable.Entry>
806) -> IO(TableValidation):
807 match present:
808 case False{}:
809 IO.pure(TableValidation,
810 table.validation.snapshot(lock, path, Done{(0n, 0n)}, records, Nil{}))
811 case True{}:
812 do IO<TableValidation>:
813 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path ++ "/WAL")
814 wal.locked.size.result(lock, size, path, records)
815
816def wal.locked.exists(
817 lock: DbLock.Lock,
818 result: Result<&1, &1, U32 & String, Bool>,
819 path: String,
820 records: List<&2, MemTable.Entry>
821) -> IO(TableValidation):
822 match result:
823 case Fail{(code, message)}:
824 IO.pure(TableValidation,
825 table.validation.records(lock, path, Fail{HostError{code, message}}, records))
826 case Done{present}: wal.locked.present(lock, present, path, records)
827
828def source.release.result(
829 value: Result<&1, &1, Error, SourceSnapshot>,
830 released: Result<&1, &1, DurableError.Error, Unit>
831) -> Result<&1, &1, Error, SourceSnapshot>:
832 match released:
833 case Fail{error}: Fail{LockError{error}}
834 case Done{Unit{}}: value
835
836def source.release(
837 lock: DbLock.Lock,
838 path: String,
839 value: Result<&1, &1, Error, SourceSnapshot>
840) -> IO(Result<&1, &1, Error, SourceSnapshot>):
841 do IO<Result<&1, &1, Error, SourceSnapshot>>:
842 released : Result<&1, &1, DurableError.Error, Unit> <-
843 DbLock.release(lock, "migrate-verify", path)
844 return source.release.result(value, released)
845
846def source.wal.validated(
847 sst: Nat & Nat,
848 state: TableValidation
849) -> IO(Result<&1, &1, Error, SourceSnapshot>):
850 match sst state:
851 case (sst_entries, sst_bytes) TableValidation{lock, path, result,
852 sst_records, wal_batches}:
853 match result:
854 case Fail{error}:
855 source.release(lock, path, Fail{error})
856 case Done{(wal_frames, wal_bytes)}:
857 source.release(lock, path, Done{LegacySourceSnapshot{sst_records,
858 wal_batches, sst_entries, wal_frames, Nat.add(sst_bytes, wal_bytes)}})
859
860def source.tables.present(
861 sst: Nat & Nat,
862 sst_records: List<&2, MemTable.Entry>,
863 lock: DbLock.Lock,
864 paths: String & String
865) -> IO(Result<&1, &1, Error, SourceSnapshot>):
866 match paths:
867 case (exists_path, validation_path):
868 do IO<Result<&1, &1, Error, SourceSnapshot>>:
869 present : Result<&1, &1, U32 & String, Bool> <-
870 Fs.exists(exists_path ++ "/WAL")
871 wal : TableValidation <- wal.locked.exists(lock, present,
872 validation_path, sst_records)
873 source.wal.validated(sst, wal)
874
875def source.tables.validated(
876 state: TableValidation
877) -> IO(Result<&1, &1, Error, SourceSnapshot>):
878 match state:
879 case TableValidation{lock, path, result, records, _}:
880 match result:
881 case Fail{error}: source.release(lock, path, Fail{error})
882 case Done{sst}:
883 source.tables.present(sst, records, lock, path.clone.pair(path))
884
885def source.manifest.present(
886 lock: DbLock.Lock,
887 paths: String & String
888) -> IO(Result<&1, &1, Error, SourceSnapshot>):
889 match paths:
890 case (exists_path, validation_path):
891 do IO<Result<&1, &1, Error, SourceSnapshot>>:
892 present : Result<&1, &1, U32 & String, Bool> <-
893 Fs.exists(exists_path ++ "/MANIFEST")
894 tables : TableValidation <-
895 table.manifest.present(lock, validation_path, present)
896 source.tables.validated(tables)
897
898def source.locked.manifest(
899 +path: String,
900 result: Result<&1, &1, DurableError.Error, DbLock.Lock>
901) -> IO(Result<&1, &1, Error, SourceSnapshot>):
902 match result:
903 case Fail{error}:
904 IO.pure(Result<&1, &1, Error, SourceSnapshot>, Fail{LockError{error}})
905 case Done{lock}:
906 source.manifest.present(lock, path.clone.pair(path))
907
908def source.lock.present(
909 +path: String,
910 result: Result<&1, &1, U32 & String, Bool>
911) -> IO(Result<&1, &1, Error, SourceSnapshot>):
912 match result:
913 case Fail{(code, message)}:
914 IO.pure(Result<&1, &1, Error, SourceSnapshot>, Fail{HostError{code, message}})
915 case Done{False{}}:
916 IO.pure(Result<&1, &1, Error, SourceSnapshot>,
917 Fail{HostError{2, "database LOCK file is missing"}})
918 case Done{True{}}:
919 do IO<Result<&1, &1, Error, SourceSnapshot>>:
920 acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <-
921 DbLock.acquire(path, "migrate-verify")
922 source.locked.manifest(path, acquired)
923
924def source.canonical(
925 result: Result<&1, &1, U32 & String, String>
926) -> IO(Result<&1, &1, Error, SourceSnapshot>):
927 match result:
928 case Fail{(code, message)}:
929 IO.pure(Result<&1, &1, Error, SourceSnapshot>, Fail{HostError{code, message}})
930 case Done{+path}:
931 do IO<Result<&1, &1, Error, SourceSnapshot>>:
932 present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path ++ "/LOCK")
933 source.lock.present(path, present)
934
935def source.counts(
936 result: Result<&1, &1, Error, SourceSnapshot>
937) -> Result<&1, &1, Error, SourceCounts>:
938 match result:
939 case Fail{error}: Fail{error}
940 case Done{LegacySourceSnapshot{_, _, sst_entries, wal_frames, encoded_bytes}}:
941 Done{LegacySourceCounts{sst_entries, wal_frames, encoded_bytes}}
942
943
944def read_legacy_source(
945 +path: String
946) -> IO(Result<&1, &1, Error, SourceSnapshot>):
947 do IO<Result<&1, &1, Error, SourceSnapshot>>:
948 canonical : Result<&1, &1, U32 & String, String> <- Fs.canonical_path(path)
949 source.canonical(canonical)
950
951
952def verify_legacy_source(
953 +path: String
954) -> IO(Result<&1, &1, Error, SourceCounts>):
955 do IO<Result<&1, &1, Error, SourceCounts>>:
956 snapshot : Result<&1, &1, Error, SourceSnapshot> <- read_legacy_source(path)
957 return source.counts(snapshot)
958
959
960def verify_legacy_wal(+path: String) -> IO(Result<&1, &1, Error, Nat>):
961 do IO<Result<&1, &1, Error, Nat>>:
962 canonical : Result<&1, &1, U32 & String, String> <- Fs.canonical_path(path)
963 canonical.result(canonical)