src/transfer/RestoreIo.bend

mylsm-lsm-store@0.6.0.1 · 528 lines · 45 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/ByteDurableDb.bend as Db
3import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
4import mylsm-lsm-store@0.6.0.1/src/ByteWal.bend as ByteWal
5import mylsm-lsm-store@0.6.0.1/src/CrashPoint.bend as CrashPoint
6import mylsm-lsm-store@0.6.0.1/src/Fs.bend as Fs
7import mylsm-lsm-store@0.6.0.1/src/transfer/Restore.bend as Restore
8import mylsm-lsm-store@0.6.0.1/src/transfer/PublicationPolicy.bend as PublicationPolicy
9import mylsm-lsm-store@0.6.0.1/src/transfer/RestorePolicy.bend as RestorePolicy
10import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
11
12# Classifies exchange validation, staging, publication, and durability failures.
13type Error is Data:
14  InvalidFormat{}
15  ResourceLimit{}
16  InvalidOrder{}
17  ChecksumMismatch{}
18  CountMismatch{}
19  Storage{error: Db.Error}
20  Io{code: U32, message: String}
21  DestinationExists{}
22  TemporaryExists{}
23  VerificationFailed{}
24  CommitUnknown{code: U32, message: String}
25
26# Retains a single staged record and database handle between WAL writes.
27type WriteState is Type:
28  WriteNeed{handle: Db.Handle, entries: List<&1, ByteMemTable.Entry>}
29  WriteBatch{handle: Db.Handle, page: WritePage}
30  WriteResult{pair: Db.Handle & Result<&1, &1, Db.Error, Unit>,
31    rest: List<&1, ByteMemTable.Entry>}
32
33# Carries one bounded mutation page and the remaining ordered entries.
34type WritePage is Type:
35  WritePage{batch: List<&1, ByteMemTable.Entry>, rest: List<&1, ByteMemTable.Entry>}
36
37def decode.error(error: Restore.Error) -> Error:
38  match error:
39    case Restore.InvalidFormat{}: InvalidFormat{}
40    case Restore.ResourceLimit{}: ResourceLimit{}
41    case Restore.InvalidOrder{}: InvalidOrder{}
42    case Restore.ChecksumMismatch{}: ChecksumMismatch{}
43    case Restore.CountMismatch{}: CountMismatch{}
44
45def write.finish(pair: Db.Handle & Result<&1, &1, Db.Error, Unit>) -> WriteState:
46  WriteResult{pair, Nil{}}
47
48def write.result(
49  rest: List<&1, ByteMemTable.Entry>,
50  pair: Db.Handle & Result<&1, &1, Db.Error, Unit>
51) -> WriteState:
52  WriteResult{pair, rest}
53
54def write.page.go(
55  fuel: Nat,
56  entries: List<&1, ByteMemTable.Entry>,
57  reversed: List<&1, ByteMemTable.Entry>
58) -> WritePage:
59  match fuel:
60    case 0n: WritePage{List.reverse(&1, ByteMemTable.Entry, reversed), entries}
61    case 1n+rest:
62      match entries:
63        case Nil{}: WritePage{List.reverse(&1, ByteMemTable.Entry, reversed), Nil{}}
64        case Con{entry, tail}: write.page.go(rest, tail, Con{entry, reversed})
65
66def write.page.ready(
67  measured: List<&1, ByteMemTable.Entry> & Nat
68) -> WritePage:
69  match measured:
70    case (entries, count):
71      write.page.go(Nat.min(RestorePolicy.write.page.size(), count), entries, Nil{})
72
73def write.page(entries: List<&1, ByteMemTable.Entry>) -> WritePage:
74  write.page.ready(ByteMemTable.count.keep(entries))
75
76def write.mutations(entries: List<&1, ByteMemTable.Entry>) -> List<&1, ByteWal.Mutation>:
77  match entries:
78    case Nil{}: Nil{}
79    case Con{ByteMemTable.Entry{key, value}, tail}:
80      match value:
81        case None{}: Con{ByteWal.Delete{key}, write.mutations(tail)}
82        case Some{value}: Con{ByteWal.Put{key, value}, write.mutations(tail)}
83
84def write.loop(fuel: Nat, state: WriteState) -> IO(Db.Handle & Result<&1, &1, Db.Error, Unit>):
85  match fuel:
86    case 0n:
87      match state:
88        case WriteNeed{handle, _}:
89          IO.pure(Db.Handle & Result<&1, &1, Db.Error, Unit>,
90            (handle, Fail{Db.ResourceLimit{}}))
91        case WriteBatch{handle, _}:
92          IO.pure(Db.Handle & Result<&1, &1, Db.Error, Unit>,
93            (handle, Fail{Db.ResourceLimit{}}))
94        case WriteResult{pair, _}: IO.pure(Db.Handle & Result<&1, &1, Db.Error, Unit>, pair)
95    case 1n+rest_fuel:
96      match state:
97        case WriteNeed{handle, entries}:
98          match entries:
99            case Nil{}:
100              IO.pure(Db.Handle & Result<&1, &1, Db.Error, Unit>,
101                (handle, Done{Unit{}}))
102            case Con{_, _}:
103              write.loop(rest_fuel, WriteBatch{handle, write.page(entries)})
104        case WriteBatch{handle, WritePage{batch, rest}}:
105          do IO<Db.Handle & Result<&1, &1, Db.Error, Unit>>:
106            pair : Db.Handle & Result<&1, &1, Db.Error, Unit> <-
107              Db.write_batch(handle, write.mutations(batch))
108            write.loop(rest_fuel, WriteResult{pair, rest})
109        case WriteResult{pair, rest}:
110          match pair:
111            case (handle, Fail{error}):
112              IO.pure(Db.Handle & Result<&1, &1, Db.Error, Unit>,
113                (handle, Fail{error}))
114            case (handle, Done{Unit{}}):
115              write.loop(rest_fuel, WriteNeed{handle, rest})
116
117def remove.ignored(path: String) -> IO(Unit):
118  do IO<Unit>:
119    _removed : Result<&1, &1, U32 & String, Unit> <- Fs.remove(path)
120    return Unit{}
121
122def cleanup.directory(+path: String) -> IO(Unit):
123  do IO<Unit>:
124    remove.ignored(path ++ "/WAL")
125    remove.ignored(path ++ "/WAL.tmp")
126    remove.ignored(path ++ "/SST")
127    remove.ignored(path ++ "/SST.tmp")
128    remove.ignored(path ++ "/LOCK")
129    remove.ignored(path)
130
131def cleanup.result(
132  path: String,
133  result: Result<&1, &1, Db.Error, Unit>
134) -> IO(Result<&1, &1, Error, Unit>):
135  match result:
136    case Fail{error}:
137      do IO<Result<&1, &1, Error, Unit>>:
138        cleanup.directory(path)
139        return Fail{Storage{error}}
140    case Done{Unit{}}:
141      do IO<Result<&1, &1, Error, Unit>>:
142        cleanup.directory(path)
143        return Done{Unit{}}
144
145def cleanup.handle(
146  path: String,
147  handle: Db.Handle,
148  error: Db.Error
149) -> IO(Result<&1, &1, Error, Unit>):
150  do IO<Result<&1, &1, Error, Unit>>:
151    _closed : Result<&1, &1, Db.Error, Unit> <- Db.close(handle)
152    cleanup.result(path, Fail{error})
153
154def flush.closed(
155  path: String,
156  result: Result<&1, &1, Db.Error, Unit>
157) -> IO(Result<&1, &1, Error, Unit>):
158  match result:
159    case Fail{error}: cleanup.result(path, Fail{error})
160    case Done{Unit{}}: IO.pure(Result<&1, &1, Error, Unit>, Done{Unit{}})
161
162def flush.result(
163  path: String,
164  pair: Db.Handle & Result<&1, &1, Db.Error, Unit>
165) -> IO(Result<&1, &1, Error, Unit>):
166  match pair:
167    case (handle, Fail{error}): cleanup.handle(path, handle, error)
168    case (handle, Done{Unit{}}):
169      do IO<Result<&1, &1, Error, Unit>>:
170        closed : Result<&1, &1, Db.Error, Unit> <- Db.close(handle)
171        result : Result<&1, &1, Error, Unit> <- flush.closed(path, closed)
172        return result
173
174
175def write.result.path(
176  path: String,
177  pair: Db.Handle & Result<&1, &1, Db.Error, Unit>
178) -> IO(Result<&1, &1, Error, Unit>):
179  match pair:
180    case (handle, Fail{error}): cleanup.handle(path, handle, error)
181    case (handle, Done{Unit{}}):
182      do IO<Result<&1, &1, Error, Unit>>:
183        flushed : Db.Handle & Result<&1, &1, Db.Error, Unit> <- Db.flush(handle)
184        result : Result<&1, &1, Error, Unit> <- flush.result(path, flushed)
185        return result
186
187# Rejects a staged database whose recovered entry count differs from the exchange.
188def verify_count(path: String, equal: Bool) -> IO(Result<&1, &1, Error, Unit>):
189  match equal:
190    case False{}:
191      do IO<Result<&1, &1, Error, Unit>>:
192        cleanup.directory(path)
193        return Fail{VerificationFailed{}}
194    case True{}: IO.pure(Result<&1, &1, Error, Unit>, Done{Unit{}})
195
196def verified.closed(
197  path: String,
198  expected: Nat,
199  count: Nat,
200  closed: Result<&1, &1, Db.Error, Unit>
201) -> IO(Result<&1, &1, Error, Unit>):
202  match closed:
203    case Fail{error}:
204      do IO<Result<&1, &1, Error, Unit>>:
205        cleanup.directory(path)
206        return Fail{Storage{error}}
207    case Done{Unit{}}:
208      verify_count(path, Nat.is_eq(count, expected))
209
210def path.reverse(path: String, reversed: String) -> String:
211  match path:
212    case SNil{}: reversed
213    case SCon{character, rest}:
214      path.reverse(rest, SCon{character, reversed})
215
216def path.clone(path: String) -> String:
217  path.reverse(path.reverse(path, SNil{}), SNil{})
218
219
220def verified.stats(
221  path: String,
222  expected: Nat,
223  pair: Db.Handle & Db.Stats
224) -> IO(Result<&1, &1, Error, Unit>):
225  match pair:
226    case (handle, Db.Stats{entries, _}):
227      do IO<Result<&1, &1, Error, Unit>>:
228        closed : Result<&1, &1, Db.Error, Unit> <- Db.close(handle)
229        verified.closed(path, expected, entries, closed)
230
231# Opens and checks the staged database before publication.
232def verified(
233  path: String,
234  expected: Nat,
235  opened: Result<&1, &1, Db.Error, Db.Handle>
236) -> IO(Result<&1, &1, Error, Unit>):
237  match opened:
238    case Fail{error}:
239      do IO<Result<&1, &1, Error, Unit>>:
240        cleanup.directory(path)
241        return Fail{Storage{error}}
242    case Done{handle}: verified.stats(path, expected, Db.stats(handle))
243
244# Validates an exchange, builds a sibling database, verifies it, then publishes it.
245def publish.parent.sync.failed(
246  disposition: PublicationPolicy.Failure,
247  error: U32 & String
248) -> IO(Result<&1, &1, Error, Unit>):
249  match disposition:
250    case PublicationPolicy.CommitUnknown{}:
251      match error:
252        case (code, message):
253          IO.pure(Result<&1, &1, Error, Unit>,
254            Fail{CommitUnknown{code, "parent directory sync failed: " ++ message}})
255    case PublicationPolicy.KnownFailure{}:
256      match error:
257        case (code, message):
258          IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
259
260def publish.parent.synced(
261  result: Result<&1, &1, U32 & String, Unit>
262) -> IO(Result<&1, &1, Error, Unit>):
263  match result:
264    case Fail{error}:
265      publish.parent.sync.failed(
266        PublicationPolicy.failure(PublicationPolicy.AfterRename{}), error)
267    case Done{Unit{}}:
268      do IO<Result<&1, &1, Error, Unit>>:
269        _checkpoint : Result<&1, &1, U32 & String, Unit> <-
270          CrashPoint.hit("restore.parent_synced")
271        return Done{Unit{}}
272
273def publish.rename.failed(
274  +temporary: String,
275  disposition: PublicationPolicy.Failure,
276  error: U32 & String
277) -> IO(Result<&1, &1, Error, Unit>):
278  match disposition:
279    case PublicationPolicy.KnownFailure{}:
280      match error:
281        case (code, message):
282          do IO<Result<&1, &1, Error, Unit>>:
283            cleanup.directory(temporary)
284            return Fail{Io{code, "exclusive rename failed: " ++ message}}
285    case PublicationPolicy.CommitUnknown{}:
286      match error:
287        case (code, message):
288          IO.pure(Result<&1, &1, Error, Unit>, Fail{CommitUnknown{code, message}})
289
290def publish.renamed(
291  +destination: String,
292  +temporary: String,
293  result: Result<&1, &1, U32 & String, Unit>
294) -> IO(Result<&1, &1, Error, Unit>):
295  match result:
296    case Fail{error}:
297      publish.rename.failed(temporary,
298        PublicationPolicy.failure(PublicationPolicy.BeforeRename{}), error)
299    case Done{Unit{}}:
300      do IO<Result<&1, &1, Error, Unit>>:
301        _checkpoint : Result<&1, &1, U32 & String, Unit> <-
302          CrashPoint.hit("restore.renamed")
303        synced : Result<&1, &1, U32 & String, Unit> <-
304          Fs.fsync(path.clone(destination ++ "/.."))
305        published : Result<&1, &1, Error, Unit> <- publish.parent.synced(synced)
306        return published
307
308def publish.synced(
309  +destination: String,
310  +temporary: String,
311  result: Result<&1, &1, U32 & String, Unit>
312) -> IO(Result<&1, &1, Error, Unit>):
313  match result:
314    case Fail{(code, message)}:
315      do IO<Result<&1, &1, Error, Unit>>:
316        cleanup.directory(temporary)
317        return Fail{Io{code, "staged directory sync failed: " ++ message}}
318    case Done{Unit{}}:
319      do IO<Result<&1, &1, Error, Unit>>:
320        _checkpoint : Result<&1, &1, U32 & String, Unit> <-
321          CrashPoint.hit("restore.before_rename")
322        renamed : Result<&1, &1, U32 & String, Unit> <-
323          Fs.rename_noreplace(path.clone(temporary), path.clone(destination))
324        published : Result<&1, &1, Error, Unit> <- publish.renamed(destination, temporary, renamed)
325        return published
326
327def publish.verified(
328  +destination: String,
329  +temporary: String,
330  result: Result<&1, &1, Error, Unit>
331) -> IO(Result<&1, &1, Error, Unit>):
332  match result:
333    case Fail{error}: IO.pure(Result<&1, &1, Error, Unit>, Fail{error})
334    case Done{Unit{}}:
335      do IO<Result<&1, &1, Error, Unit>>:
336        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path.clone(temporary))
337        published : Result<&1, &1, Error, Unit> <- publish.synced(destination, temporary, synced)
338        return published
339
340# Reopens staged data and checks that recovery sees the expected record count.
341def publish(
342  +destination: String,
343  +temporary: String,
344  expected: Nat
345) -> IO(Result<&1, &1, Error, Unit>):
346  do IO<Result<&1, &1, Error, Unit>>:
347    opened : Result<&1, &1, Db.Error, Db.Handle> <- Db.open_existing(path.clone(temporary))
348    verified : Result<&1, &1, Error, Unit> <- verified(temporary, expected, opened)
349    published : Result<&1, &1, Error, Unit> <- publish.verified(destination, temporary, verified)
350    return published
351
352def publish.result(
353  destination: String,
354  temporary: String,
355  expected: Nat,
356  populated: Result<&1, &1, Error, Unit>
357) -> IO(Result<&1, &1, Error, Unit>):
358  match populated:
359    case Fail{error}: IO.pure(Result<&1, &1, Error, Unit>, Fail{error})
360    case Done{Unit{}}: publish(destination, temporary, expected)
361
362
363def temporary.create(
364  +destination: String,
365  +temporary: String,
366  created: Result<&1, &1, Db.Error, Db.Handle>,
367  measured: List<&1, ByteMemTable.Entry> & Nat
368) -> IO(Result<&1, &1, Error, Unit>):
369  match created:
370    case Fail{error}: IO.pure(Result<&1, &1, Error, Unit>, Fail{Storage{error}})
371    case Done{handle}:
372      match measured:
373        case (entries, +count):
374          do IO<Result<&1, &1, Error, Unit>>:
375            written : Db.Handle & Result<&1, &1, Db.Error, Unit> <-
376              write.loop(RestorePolicy.write.fuel(count),
377                WriteNeed{handle, entries})
378            populated : Result<&1, &1, Error, Unit> <- write.result.path(temporary, written)
379            published : Result<&1, &1, Error, Unit> <-
380              publish.result(destination, temporary, count, populated)
381            return published
382
383
384def temporary.decoded(
385  +destination: String,
386  +temporary: String,
387  decoded: Result<&1, &1, Restore.Error, Restore.Decoded>
388) -> IO(Result<&1, &1, Error, Unit>):
389  match decoded:
390    case Fail{error}:
391      IO.pure(Result<&1, &1, Error, Unit>, Fail{decode.error(error)})
392    case Done{Restore.Decoded{entries}}:
393      do IO<Result<&1, &1, Error, Unit>>:
394        created : Result<&1, &1, Db.Error, Db.Handle> <- Db.create(path.clone(temporary))
395        result : Result<&1, &1, Error, Unit> <- temporary.create(destination,
396          temporary, created, ByteMemTable.count.keep(entries))
397        return result
398
399
400def temporary.checked(
401  +destination: String,
402  +temporary: String,
403  input: Bytes.Bytes,
404  result: Result<&1, &1, U32 & String, Bool>
405) -> IO(Result<&1, &1, Error, Unit>):
406  match result:
407    case Fail{(code, message)}:
408      IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
409    case Done{True{}}:
410      IO.pure(Result<&1, &1, Error, Unit>, Fail{TemporaryExists{}})
411    case Done{False{}}:
412      temporary.decoded(destination, temporary, Restore.decode(input))
413
414def destination.checked(
415  +destination: String,
416  +temporary: String,
417  input: Bytes.Bytes,
418  result: Result<&1, &1, U32 & String, Bool>
419) -> IO(Result<&1, &1, Error, Unit>):
420  match result:
421    case Fail{(code, message)}:
422      IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
423    case Done{True{}}:
424      IO.pure(Result<&1, &1, Error, Unit>, Fail{DestinationExists{}})
425    case Done{False{}}:
426      do IO<Result<&1, &1, Error, Unit>>:
427        exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path.clone(temporary))
428        temporary.checked(destination, temporary, input, exists)
429
430
431# Validates an exchange before staging and atomically publishing a database.
432def restore(
433  +destination: String,
434  input: Bytes.Bytes
435) -> IO(Result<&1, &1, Error, Unit>):
436  +temporary = path.clone(destination ++ ".restore-060.tmp")
437  do IO<Result<&1, &1, Error, Unit>>:
438    exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path.clone(destination))
439    destination.checked(destination, temporary, input, exists)
440
441def source.read.length(
442  +destination: String,
443  input: Bytes.Bytes,
444  equal: Bool
445) -> IO(Result<&1, &1, Error, Unit>):
446  match equal:
447    case False{}: IO.pure(Result<&1, &1, Error, Unit>, Fail{InvalidFormat{}})
448    case True{}: restore(destination, input)
449
450def source.read.checked(
451  +destination: String,
452  +expected_size: Nat,
453  input: Bytes.Bytes
454) -> IO(Result<&1, &1, Error, Unit>):
455  match input:
456    case Bytes.Bytes{+len, _}:
457      source.read.length(destination, input,
458        Nat.is_eq(U32.to_nat(len), expected_size))
459
460def source.read.result(
461  +destination: String,
462  +expected_size: Nat,
463  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
464) -> IO(Result<&1, &1, Error, Unit>):
465  match pair:
466    case (file, Fail{(code, message)}):
467      do IO<Result<&1, &1, Error, Unit>>:
468        _closed : Unit <- File.close(file)
469        return Fail{Io{code, message}}
470    case (file, Done{input}):
471      do IO<Result<&1, &1, Error, Unit>>:
472        _closed : Unit <- File.close(file)
473        source.read.checked(destination, expected_size, input)
474
475def source.opened(
476  +destination: String,
477  +size: Nat,
478  opened: Result<&1, &1, U32 & String, File>
479) -> IO(Result<&1, &1, Error, Unit>):
480  match opened:
481    case Fail{(code, message)}:
482      IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
483    case Done{file}:
484      do IO<Result<&1, &1, Error, Unit>>:
485        pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
486          Fs.read_at(file, 0, U32.from_nat(size))
487        source.read.result(destination, size, pair)
488
489def source.sized.checked(
490  +destination: String,
491  +source: String,
492  size: Nat,
493  valid: Bool
494) -> IO(Result<&1, &1, Error, Unit>):
495  match valid:
496    case False{}:
497      IO.pure(Result<&1, &1, Error, Unit>, Fail{ResourceLimit{}})
498    case True{}:
499      do IO<Result<&1, &1, Error, Unit>>:
500        opened : Result<&1, &1, U32 & String, File> <- File.open(source, "r")
501        source.opened(destination, size, opened)
502
503def source.sized(
504  +destination: String,
505  +source: String,
506  +size: Nat
507) -> IO(Result<&1, &1, Error, Unit>):
508  source.sized.checked(destination, source, size,
509    RestorePolicy.exchange.fits(RestorePolicy.max.exchange.bytes(), size))
510
511def source.size(
512  +destination: String,
513  +source: String,
514  measured: Result<&1, &1, U32 & String, Nat>
515) -> IO(Result<&1, &1, Error, Unit>):
516  match measured:
517    case Fail{(code, message)}:
518      IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
519    case Done{size}: source.sized(destination, source, size)
520
521# Reads a bounded backup bundle file and restores its validated exchange.
522def restore_from(
523  +destination: String,
524  +bundle: String
525) -> IO(Result<&1, &1, Error, Unit>):
526  do IO<Result<&1, &1, Error, Unit>>:
527    measured : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(bundle ++ "/BACKUP")
528    source.size(destination, bundle ++ "/BACKUP", measured)