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# Carries one pending read, bounded chunk write, or completed file write.
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# Removes the known staging file and directory after failed publication.
40# Removes the known staging file and directory after failed publication.
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# Closes backup resources and preserves an unsupported source platform error.
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# Pulls at most one visible cursor result and writes at most one encoded chunk.
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# Exports a stable visible cursor through bounded checksummed chunks.
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)