src/transfer/BackupIo.bend

mylsm-lsm-store@0.6.0.1 · 272 lines · 22 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/transfer/Backup.bend as Backup
3import mylsm-lsm-store@0.6.0.1/src/transfer/PublicationPolicy.bend as PublicationPolicy
4import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
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 bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
8
9# Distinguishes pre-publication IO failures from an uncertain durable commit.
10type Error is Data:
11  AlreadyExists{}
12  UnsupportedPlatform{}
13  ResourceLimit{}
14  Io{code: U32, message: String}
15  CommitUnknown{code: U32, message: String}
16
17# Carries the open staged file and one backup stream transition.
18type State is Type:
19  Need{file: File, stream: Backup.Stream}
20  Output{file: File, result: Result<&1, &1, Backup.Error, Backup.Step>}
21  Written{pair: File & Result<&1, &1, U32 & String, Unit>, stream: Backup.Stream}
22
23def path.reverse(path: String, reversed: String) -> String:
24  match path:
25    case SNil{}: reversed
26    case SCon{character, rest}: path.reverse(rest, SCon{character, reversed})
27
28def path.clone(path: String) -> String:
29  path.reverse(path.reverse(path, SNil{}), SNil{})
30
31# Removes known staging contents after a failure or successful rename.
32def cleanup(+directory: String) -> IO(Unit):
33  do IO<Unit>:
34    _file : Result<&1, &1, U32 & String, Unit> <- Fs.remove(directory ++ "/BACKUP")
35    _directory : Result<&1, &1, U32 & String, Unit> <- Fs.remove(directory)
36    return Unit{}
37
38def close.failed.cleaned(
39  +directory: String,
40  error: U32 & String
41) -> IO(Result<&1, &1, Error, Unit>):
42  match error:
43    case (code, message):
44      do IO<Result<&1, &1, Error, Unit>>:
45        cleanup(directory)
46        return Fail{Io{code, message}}
47
48def close.failed(file: File, +directory: String, error: U32 & String) -> IO(Result<&1, &1, Error, Unit>):
49  do IO<Result<&1, &1, Error, Unit>>:
50    _closed : Unit <- File.close(file)
51    close.failed.cleaned(directory, error)
52
53def publish.parent.failed(
54  +temporary: String,
55  disposition: PublicationPolicy.Failure,
56  error: U32 & String
57) -> IO(Result<&1, &1, Error, Unit>):
58  match disposition:
59    case PublicationPolicy.CommitUnknown{}:
60      match error:
61        case (code, message):
62          do IO<Result<&1, &1, Error, Unit>>:
63            cleanup(temporary)
64            return Fail{CommitUnknown{code, "backup published but parent sync failed: " ++ message}}
65    case PublicationPolicy.KnownFailure{}:
66      match error:
67        case (code, message):
68          IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
69
70def publish.parent(
71  +temporary: String,
72  result: Result<&1, &1, U32 & String, Unit>
73) -> IO(Result<&1, &1, Error, Unit>):
74  match result:
75    case Fail{error}:
76      publish.parent.failed(temporary,
77        PublicationPolicy.failure(PublicationPolicy.AfterRename{}), error)
78    case Done{Unit{}}:
79      do IO<Result<&1, &1, Error, Unit>>:
80        _stopped : Result<&1, &1, U32 & String, Unit> <- CrashPoint.hit("backup.parent_synced")
81        cleanup(temporary)
82        return Done{Unit{}}
83
84def publish.rename.failed(
85  +temporary: String,
86  disposition: PublicationPolicy.Failure,
87  error: U32 & String
88) -> IO(Result<&1, &1, Error, Unit>):
89  match disposition:
90    case PublicationPolicy.KnownFailure{}:
91      match error:
92        case (code, message):
93          do IO<Result<&1, &1, Error, Unit>>:
94            cleanup(temporary)
95            return Fail{Io{code, "exclusive backup rename failed: " ++ message}}
96    case PublicationPolicy.CommitUnknown{}:
97      match error:
98        case (code, message):
99          IO.pure(Result<&1, &1, Error, Unit>, Fail{CommitUnknown{code, message}})
100
101def publish.rename(
102  +destination: String,
103  +temporary: String,
104  result: Result<&1, &1, U32 & String, Unit>
105) -> IO(Result<&1, &1, Error, Unit>):
106  match result:
107    case Fail{error}:
108      publish.rename.failed(temporary,
109        PublicationPolicy.failure(PublicationPolicy.BeforeRename{}), error)
110    case Done{Unit{}}:
111      do IO<Result<&1, &1, Error, Unit>>:
112        _stopped : Result<&1, &1, U32 & String, Unit> <- CrashPoint.hit("backup.renamed")
113        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path.clone(destination ++ "/.."))
114        publish.parent(temporary, synced)
115
116def publish.directory.synced(
117  +destination: String,
118  +temporary: String,
119  result: Result<&1, &1, U32 & String, Unit>
120) -> IO(Result<&1, &1, Error, Unit>):
121  match result:
122    case Fail{(code, message)}:
123      do IO<Result<&1, &1, Error, Unit>>:
124        cleanup(temporary)
125        return Fail{Io{code, message}}
126    case Done{Unit{}}:
127      do IO<Result<&1, &1, Error, Unit>>:
128        _stopped : Result<&1, &1, U32 & String, Unit> <- CrashPoint.hit("backup.before_rename")
129        renamed : Result<&1, &1, U32 & String, Unit> <-
130          Fs.rename_noreplace(path.clone(temporary), path.clone(destination))
131        publish.rename(destination, temporary, renamed)
132
133
134def publish.synced(
135  +destination: String,
136  +temporary: String,
137  result: Result<&1, &1, U32 & String, Unit>
138) -> IO(Result<&1, &1, Error, Unit>):
139  match result:
140    case Fail{(code, message)}:
141      do IO<Result<&1, &1, Error, Unit>>:
142        cleanup(temporary)
143        return Fail{Io{code, message}}
144    case Done{Unit{}}:
145      do IO<Result<&1, &1, Error, Unit>>:
146        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path.clone(temporary))
147        publish.directory.synced(destination, temporary, synced)
148
149def publish.closed.result(
150  +destination: String,
151  +temporary: String,
152  result: Result<&1, &1, U32 & String, Unit>
153) -> IO(Result<&1, &1, Error, Unit>):
154  match result:
155    case Fail{(code, message)}:
156      do IO<Result<&1, &1, Error, Unit>>:
157        cleanup(temporary)
158        return Fail{Io{code, message}}
159    case Done{Unit{}}:
160      do IO<Result<&1, &1, Error, Unit>>:
161        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path.clone(temporary ++ "/BACKUP"))
162        publish.synced(destination, temporary, synced)
163
164def publish.closed(
165  +destination: String,
166  +temporary: String,
167  file: File,
168  result: Result<&1, &1, U32 & String, Unit>
169) -> IO(Result<&1, &1, Error, Unit>):
170  do IO<Result<&1, &1, Error, Unit>>:
171    _closed : Unit <- File.close(file)
172    publish.closed.result(destination, temporary, result)
173
174# Writes one bounded encoded chunk per state transition before publication.
175def loop(fuel: Nat, +destination: String, +temporary: String, state: State) -> IO(Result<&1, &1, Error, Unit>):
176  match fuel:
177    case 0n:
178      match state:
179        case Need{file, _}: close.failed(file, temporary, (7, "backup stream exceeded its transition budget"))
180        case Output{file, _}: close.failed(file, temporary, (7, "backup stream exceeded its transition budget"))
181        case Written{(file, _), _}:
182          do IO<Result<&1, &1, Error, Unit>>:
183            _closed : Unit <- File.close(file)
184            cleanup(temporary)
185            return Fail{Io{7, "backup stream exceeded its transition budget"}}
186    case 1n+rest:
187      match state:
188        case Need{file, stream}:
189          loop(rest, destination, temporary, Output{file, Backup.next(stream)})
190        case Output{file, result}:
191          match result:
192            case Fail{_}:
193              close.failed(file, temporary, (22, "backup encoder rejected its snapshot"))
194            case Done{Backup.End{}}:
195              publish.closed(destination, temporary, file, Done{Unit{}})
196            case Done{Backup.Chunk{stream, bytes}}:
197              do IO<Result<&1, &1, Error, Unit>>:
198                written : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(file, bytes)
199                loop(rest, destination, temporary, Written{written, stream})
200        case Written{pair, stream}:
201          match pair:
202            case (file, Fail{error}): close.failed(file, temporary, error)
203            case (file, Done{Unit{}}): loop(rest, destination, temporary, Need{file, stream})
204
205def directory.file.opened(
206  +destination: String,
207  +temporary: String,
208  stream: Backup.Stream,
209  fuel: Nat,
210  opened: Result<&1, &1, U32 & String, File>
211) -> IO(Result<&1, &1, Error, Unit>):
212  match opened:
213    case Fail{(code, message)}:
214      do IO<Result<&1, &1, Error, Unit>>:
215        cleanup(temporary)
216        return Fail{Io{code, message}}
217    case Done{file}: loop(fuel, destination, temporary, Need{file, stream})
218
219def directory.created(
220  +destination: String,
221  +temporary: String,
222  result: Result<&1, &1, U32 & String, Unit>,
223  stream: Backup.Stream,
224  fuel: Nat
225) -> IO(Result<&1, &1, Error, Unit>):
226  match result:
227    case Fail{(code, _)}:
228      match code:
229        case 17: IO.pure(Result<&1, &1, Error, Unit>, Fail{AlreadyExists{}})
230        case _: IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, "backup staging directory creation failed"}})
231    case Done{Unit{}}:
232      do IO<Result<&1, &1, Error, Unit>>:
233        opened : Result<&1, &1, U32 & String, File> <- File.open(path.clone(temporary ++ "/BACKUP"), "w")
234        directory.file.opened(destination, temporary, stream, fuel, opened)
235
236def destination.checked(
237  +destination: String,
238  +temporary: String,
239  exists: Result<&1, &1, U32 & String, Bool>,
240  stream: Backup.Stream,
241  fuel: Nat
242) -> IO(Result<&1, &1, Error, Unit>):
243  match exists:
244    case Fail{(code, message)}:
245      IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
246    case Done{True{}}: IO.pure(Result<&1, &1, Error, Unit>, Fail{AlreadyExists{}})
247    case Done{False{}}:
248      do IO<Result<&1, &1, Error, Unit>>:
249        created : Result<&1, &1, U32 & String, Unit> <- Fs.make_dir(path.clone(temporary))
250        directory.created(destination, temporary, created, stream, fuel)
251
252# Stages a backup bundle and publishes it without replacing an existing path.
253def start(+destination: String, stream: Backup.Stream) -> IO(Result<&1, &1, Error, Unit>):
254  match stream:
255    case Backup.Stream{_, _, _, +block_count, _}:
256      +temporary = destination ++ ".backup-060.tmp"
257      do IO<Result<&1, &1, Error, Unit>>:
258        exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path.clone(destination))
259        destination.checked(destination, temporary, exists, stream,
260          Backup.write_fuel(U32.to_nat(block_count)))
261
262def start.result(
263  +destination: String,
264  result: Result<&1, &1, Backup.Error, Backup.Stream>
265) -> IO(Result<&1, &1, Error, Unit>):
266  match result:
267    case Fail{_}: IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{22, "backup encoder rejected its snapshot"}})
268    case Done{stream}: start(destination, stream)
269
270# Starts the checksummed file export from an immutable logical snapshot.
271def export(+destination: String, entries: List<&1, ByteMemTable.Entry>) -> IO(Result<&1, &1, Error, Unit>):
272  start.result(destination, Backup.start(entries))