src/transfer/BackupCursorIo.bend
mylsm-lsm-store@0.6.0.1 · 331 lines · 28 definitions
1import Base
2import mylsm-lsm-store@0.6.0.1/src/Fs.bend as Fs
3import mylsm-lsm-store@0.6.0.1/src/scan/VisibleCursor.bend as VisibleCursor
4import mylsm-lsm-store@0.6.0.1/src/transfer/BackupCursor.bend as BackupCursor
5import mylsm-lsm-store@0.6.0.1/src/transfer/BackupIo.bend as BackupIo
6import mylsm-lsm-store@0.6.0.1/src/transfer/Backup.bend as Backup
7
8type State is Type:
9 Need{file: File, stream: BackupCursor.Stream}
10 Output{file: File, step: BackupCursor.Step}
11 Written{pair: File & Result<&1, &1, U32 & String, Unit>, stream: BackupCursor.Stream}
12
13def path.reverse(path: String, reversed: String) -> String:
14 match path:
15 case SNil{}: reversed
16 case SCon{character, rest}: path.reverse(rest, SCon{character, reversed})
17
18def path.clone(path: String) -> String:
19 path.reverse(path.reverse(path, SNil{}), SNil{})
20
21def error.host(error: Backup.Error) -> U32 & String:
22 match error:
23 case Backup.UnsupportedPlatform{}:
24 (95, "backup source transport is unsupported")
25 case Backup.SourceFailure{code, message}: (code, message)
26 case Backup.ResourceLimit{}: (7, "backup cursor exceeded its bounded policy")
27 case Backup.InvalidOrder{}: (22, "backup cursor rejected noncanonical order")
28 case Backup.CodecFailure{}: (22, "backup cursor encoder rejected its source")
29
30def error.io.copied(pair: U32 & String) -> BackupIo.Error:
31 match pair:
32 case (code, message): BackupIo.Io{code, message}
33
34def error.io(error: Backup.Error) -> BackupIo.Error:
35 match error:
36 case Backup.UnsupportedPlatform{}: BackupIo.UnsupportedPlatform{}
37 case _: error.io.copied(error.host(error))
38
39
40
41def cleanup(+directory: String) -> IO(Unit):
42 do IO<Unit>:
43 _file : Result<&1, &1, U32 & String, Unit> <- Fs.remove(directory ++ "/BACKUP")
44 _directory : Result<&1, &1, U32 & String, Unit> <- Fs.remove(directory)
45 return Unit{}
46
47def close.source(
48 file: File,
49 +directory: String,
50 cursor: VisibleCursor.Cursor,
51 error: U32 & String
52) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
53 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
54 _closed_cursor : Result<&1, &1, U32 & String, Unit> <- VisibleCursor.close(cursor)
55 BackupIo.close.failed(file, directory, error)
56
57def close.stream(
58 file: File,
59 +directory: String,
60 stream: BackupCursor.Stream,
61 error: U32 & String
62) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
63 match stream:
64 case BackupCursor.Stream{_, Some{cursor}, _, _, _, _, _, _, _}:
65 close.source(file, directory, cursor, error)
66 case BackupCursor.Stream{_, None{}, _, _, _, _, _, _, _}:
67 BackupIo.close.failed(file, directory, error)
68
69def close.optional(
70 file: File,
71 +directory: String,
72 cursor: Maybe<&1, VisibleCursor.Cursor>,
73 error: U32 & String
74) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
75 match cursor:
76 case Some{reader}: close.source(file, directory, reader, error)
77 case None{}: BackupIo.close.failed(file, directory, error)
78
79def close.resource(
80 file: File,
81 +directory: String,
82 cursor: Maybe<&1, VisibleCursor.Cursor>
83) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
84 match cursor:
85 case Some{reader}:
86 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
87 _closed_cursor : Result<&1, &1, U32 & String, Unit> <- VisibleCursor.close(reader)
88 _closed_file : Unit <- File.close(file)
89 cleanup(directory)
90 return Fail{BackupIo.ResourceLimit{}}
91 case None{}:
92 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
93 _closed_file : Unit <- File.close(file)
94 cleanup(directory)
95 return Fail{BackupIo.ResourceLimit{}}
96
97def close.resource.stream(
98 file: File,
99 +directory: String,
100 stream: BackupCursor.Stream
101) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
102 match stream:
103 case BackupCursor.Stream{_, cursor, _, _, _, _, _, _, _}:
104 close.resource(file, directory, cursor)
105
106
107def close.unsupported(
108 file: File,
109 +directory: String,
110 cursor: Maybe<&1, VisibleCursor.Cursor>
111) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
112 match cursor:
113 case Some{reader}:
114 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
115 _closed_cursor : Result<&1, &1, U32 & String, Unit> <-
116 VisibleCursor.close(reader)
117 _closed_file : Unit <- File.close(file)
118 cleanup(directory)
119 return Fail{BackupIo.UnsupportedPlatform{}}
120 case None{}:
121 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
122 _closed_file : Unit <- File.close(file)
123 cleanup(directory)
124 return Fail{BackupIo.UnsupportedPlatform{}}
125
126def close.error.host(
127 file: File,
128 +directory: String,
129 cursor: Maybe<&1, VisibleCursor.Cursor>,
130 pair: U32 & String
131) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
132 match pair:
133 case (code, message): close.optional(file, directory, cursor, (code, message))
134
135def close.error(
136 file: File,
137 +directory: String,
138 cursor: Maybe<&1, VisibleCursor.Cursor>,
139 error: Backup.Error
140) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
141 match error:
142 case Backup.ResourceLimit{}: close.resource(file, directory, cursor)
143 case Backup.UnsupportedPlatform{}:
144 close.unsupported(file, directory, cursor)
145 case _: close.error.host(file, directory, cursor, error.host(error))
146
147def close.cursor(cursor: Maybe<&1, VisibleCursor.Cursor>) -> IO(Unit):
148 match cursor:
149 case Some{reader}:
150 do IO<Unit>:
151 _closed : Result<&1, &1, U32 & String, Unit> <- VisibleCursor.close(reader)
152 return Unit{}
153 case None{}: IO.pure(Unit, Unit{})
154
155def close.initial.stream(stream: BackupCursor.Stream) -> IO(Unit):
156 match stream:
157 case BackupCursor.Stream{_, cursor, _, _, _, _, _, _, _}:
158 close.cursor(cursor)
159
160def close.initial(initial: BackupCursor.Step) -> IO(Unit):
161 match initial:
162 case BackupCursor.Chunk{stream, _}: close.initial.stream(stream)
163 case BackupCursor.Continue{stream}: close.initial.stream(stream)
164 case BackupCursor.Failure{cursor, _}: close.cursor(cursor)
165 case BackupCursor.End{}: IO.pure(Unit, Unit{})
166
167def initial.failed(
168 initial: BackupCursor.Step,
169 error: BackupIo.Error
170) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
171 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
172 _closed : Unit <- close.initial(initial)
173 return Fail{error}
174
175def close.file(
176 +destination: String,
177 +temporary: String,
178 file: File
179) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
180 BackupIo.publish.closed(destination, temporary, file, Done{Unit{}})
181
182def source.step(stream: BackupCursor.Stream) -> IO(BackupCursor.Step):
183 match stream:
184 case BackupCursor.Stream{BackupCursor.CursorBlocks{}, Some{cursor}, pending,
185 pending_count, seen, expected, entry_count, block_count, block_index}:
186 metadata : BackupCursor.Stream = BackupCursor.Stream{BackupCursor.CursorBlocks{}, None{},
187 pending, pending_count, seen, expected, entry_count, block_count,
188 block_index}
189 do IO<BackupCursor.Step>:
190 next : VisibleCursor.Step <- VisibleCursor.next(cursor)
191 return BackupCursor.advance(metadata, next)
192 case _:
193 IO.pure(BackupCursor.Step, BackupCursor.advance(stream, VisibleCursor.End{}))
194
195def loop.abort(
196 +temporary: String,
197 state: State
198) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
199 match state:
200 case Need{file, stream}:
201 close.resource.stream(file, temporary, stream)
202 case Output{file, step}:
203 match step:
204 case BackupCursor.Failure{cursor, error}:
205 close.error(file, temporary, cursor, error)
206 case BackupCursor.Chunk{stream, _}:
207 close.resource.stream(file, temporary, stream)
208 case BackupCursor.Continue{stream}:
209 close.resource.stream(file, temporary, stream)
210 case BackupCursor.End{}:
211 close.resource(file, temporary, None{})
212 case Written{pair, stream}:
213 match pair:
214 case (file, Fail{error}): close.stream(file, temporary, stream, error)
215 case (file, Done{Unit{}}):
216 close.resource.stream(file, temporary, stream)
217
218
219def loop(
220 fuel: Nat,
221 +destination: String,
222 +temporary: String,
223 state: State
224) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
225 match fuel:
226 case 0n: loop.abort(temporary, state)
227 case 1n+rest:
228 match state:
229 case Need{file, stream}:
230 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
231 step : BackupCursor.Step <- source.step(stream)
232 loop(rest, destination, temporary, Output{file, step})
233 case Output{file, step}:
234 match step:
235 case BackupCursor.Failure{cursor, error}:
236 close.error(file, temporary, cursor, error)
237 case BackupCursor.Continue{stream}:
238 loop(rest, destination, temporary, Need{file, stream})
239 case BackupCursor.Chunk{stream, bytes}:
240 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
241 written : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(file, bytes)
242 loop(rest, destination, temporary, Written{written, stream})
243 case BackupCursor.End{}:
244 close.file(destination, temporary, file)
245 case Written{pair, stream}:
246 match pair:
247 case (file, Fail{error}):
248 close.stream(file, temporary, stream, error)
249 case (file, Done{Unit{}}):
250 loop(rest, destination, temporary, Need{file, stream})
251
252def file.opened(
253 +destination: String,
254 +temporary: String,
255 initial: BackupCursor.Step,
256 fuel: Nat,
257 opened: Result<&1, &1, U32 & String, File>
258) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
259 match opened:
260 case Fail{(code, message)}:
261 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
262 _closed : Unit <- close.initial(initial)
263 cleanup(temporary)
264 return Fail{BackupIo.Io{code, message}}
265 case Done{file}:
266 loop(fuel, destination, temporary, Output{file, initial})
267
268def directory.created(
269 +destination: String,
270 +temporary: String,
271 initial: BackupCursor.Step,
272 fuel: Nat,
273 created: Result<&1, &1, U32 & String, Unit>
274) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
275 match created:
276 case Fail{(code, message)}:
277 match code:
278 case 17: initial.failed(initial, BackupIo.AlreadyExists{})
279 case _: initial.failed(initial, BackupIo.Io{code, message})
280 case Done{Unit{}}:
281 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
282 opened : Result<&1, &1, U32 & String, File> <- File.open(path.clone(temporary ++ "/BACKUP"), "w")
283 file.opened(destination, temporary, initial, fuel, opened)
284
285def destination.checked(
286 +destination: String,
287 +temporary: String,
288 initial: BackupCursor.Step,
289 fuel: Nat,
290 exists: Result<&1, &1, U32 & String, Bool>
291) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
292 match exists:
293 case Fail{(code, message)}:
294 initial.failed(initial, BackupIo.Io{code, message})
295 case Done{True{}}:
296 initial.failed(initial, BackupIo.AlreadyExists{})
297 case Done{False{}}:
298 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
299 created : Result<&1, &1, U32 & String, Unit> <- Fs.make_dir(path.clone(temporary))
300 directory.created(destination, temporary, initial, fuel, created)
301
302def initial.checked(
303 +destination: String,
304 +temporary: String,
305 initial: BackupCursor.Step,
306 fuel: Nat,
307 exists: Result<&1, &1, U32 & String, Bool>
308) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
309 match initial:
310 case BackupCursor.Failure{cursor, error}:
311 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
312 _closed : Unit <- close.cursor(cursor)
313 return Fail{error.io(error)}
314 case BackupCursor.Chunk{_, _}:
315 destination.checked(destination, temporary, initial, fuel, exists)
316 case _:
317 initial.failed(initial, BackupIo.Io{22, "backup cursor failed to encode its header"})
318
319
320def export(
321 +destination: String,
322 cursor: VisibleCursor.Cursor,
323 entry_count: Nat,
324 fuel: Nat
325) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
326 +temporary = destination ++ ".backup-060.tmp"
327 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
328 initial : BackupCursor.Step <- IO.pure(BackupCursor.Step,
329 BackupCursor.start(cursor, entry_count))
330 exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path.clone(destination))
331 initial.checked(destination, temporary, initial, fuel, exists)