src/transfer/BackupOwnerCursorIo.bend
mylsm-lsm-store@0.6.0.1 · 378 lines · 25 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/coordinator/Client.bend as Client
4import mylsm-lsm-store@0.6.0.1/src/coordinator/ByteClient.bend as ByteClient
5import mylsm-lsm-store@0.6.0.1/src/coordinator/CursorCodec.bend as CursorCodec
6import mylsm-lsm-store@0.6.0.1/src/coordinator/Io.bend as ClientIo
7import mylsm-lsm-store@0.6.0.1/src/coordinator/UnixSocketIo.bend as ClientSocket
8import mylsm-lsm-store@0.6.0.1/src/coordinator/Protocol.bend as Protocol
9import mylsm-lsm-store@0.6.0.1/src/transfer/BackupCursor.bend as BackupCursor
10import mylsm-lsm-store@0.6.0.1/src/transfer/BackupCursorIo.bend as BackupCursorIo
11import mylsm-lsm-store@0.6.0.1/src/transfer/BackupIo.bend as BackupIo
12import mylsm-lsm-store@0.6.0.1/src/transfer/BackupOwnerPolicy.bend as Policy
13import mylsm-lsm-store@0.6.0.1/src/transfer/Backup.bend as Backup
14
15
16type OwnerSource is Type:
17 OwnerSource{session: Client.Session, snapshot_id: U32, cursor_id: U32,
18 cursor_live: Bool}
19
20
21type State is Type:
22 Need{file: File, source: OwnerSource, stream: BackupCursor.Stream}
23 Output{file: File, source: OwnerSource, step: BackupCursor.Step}
24 Written{pair: File & Result<&1, &1, U32 & String, Unit>,
25 source: OwnerSource, stream: BackupCursor.Stream}
26
27def source.clone.values(
28 sessions: Client.Session & Client.Session,
29 snapshot_ids: U32 & U32 & U32,
30 cursor_ids: U32 & U32 & U32,
31 cursor_live: Bool
32) -> OwnerSource & OwnerSource:
33 match sessions:
34 case (first_session, second_session):
35 match snapshot_ids:
36 case (first_snapshot, (second_snapshot, _)):
37 match cursor_ids:
38 case (first_cursor, (second_cursor, _)):
39 match cursor_live:
40 case True{}:
41 (OwnerSource{first_session, first_snapshot, first_cursor,
42 True{}}, OwnerSource{second_session, second_snapshot,
43 second_cursor, True{}})
44 case False{}:
45 (OwnerSource{first_session, first_snapshot, first_cursor,
46 False{}}, OwnerSource{second_session, second_snapshot,
47 second_cursor, False{}})
48
49def source.clone(source: OwnerSource) -> OwnerSource & OwnerSource:
50 match source:
51 case OwnerSource{session, snapshot_id, cursor_id, cursor_live}:
52 source.clone.values(Client.session.clone(session),
53 Protocol.u32.duplicate.triple(snapshot_id),
54 Protocol.u32.duplicate.triple(cursor_id), cursor_live)
55
56
57def source.error(error: Client.Error) -> Backup.Error:
58 match error:
59 case Client.ConnectFailure{ClientSocket.Host{code, message}}:
60 Policy.source.error(Policy.ConnectHost{code, message})
61 case Client.ConnectFailure{ClientSocket.UnsupportedPlatform{}}:
62 Policy.source.error(Policy.ConnectUnsupported{})
63 case Client.ConnectFailure{ClientSocket.Transient{code}}:
64 Policy.source.error(Policy.ConnectHost{code,
65 "owner socket connection was interrupted"})
66 case Client.RetryRead{ClientIo.Host{code, message}}:
67 Policy.source.error(Policy.ReadHost{code, message})
68 case Client.RetryRead{ClientIo.UnsupportedPlatform{}}:
69 Policy.source.error(Policy.ReadUnsupported{})
70 case Client.RetryRead{ClientIo.Transient{code}}:
71 Policy.source.error(Policy.ReadHost{code,
72 "owner socket read was interrupted"})
73 case Client.RetryRead{_}: Policy.source.error(Policy.ReadFailure{})
74 case Client.CommitUnknown{ClientIo.Host{code, message}}:
75 Policy.source.error(Policy.CommitHost{code, message})
76 case Client.CommitUnknown{ClientIo.Transient{code}}:
77 Policy.source.error(Policy.CommitHost{code,
78 "owner mutation response was interrupted"})
79 case Client.CommitUnknown{_}: Policy.source.error(Policy.CommitFailure{})
80 case Client.FrameTooLarge{}: Policy.source.error(Policy.FrameTooLarge{})
81 case Client.StaleOwner{}: Policy.source.error(Policy.StaleOwner{})
82 case Client.CorrelationMismatch{}:
83 Policy.source.error(Policy.CorrelationMismatch{})
84 case Client.MalformedResponse{}:
85 Policy.source.error(Policy.MalformedResponse{})
86 case Client.RequestIdsExhausted{}:
87 Policy.source.error(Policy.RequestIdsExhausted{})
88 case Client.OwnerRejected{_}: Policy.source.error(Policy.OwnerRejected{})
89
90def source.step.result(
91 source: OwnerSource,
92 stream: BackupCursor.Stream,
93 pair: Client.Session & Result<&1, &1, Client.Error, CursorCodec.Frame>
94) -> OwnerSource & BackupCursor.Step:
95 match source:
96 case OwnerSource{_, snapshot_id, cursor_id, cursor_live}:
97 match pair:
98 case (session, Fail{error}):
99 (OwnerSource{session, snapshot_id, cursor_id, cursor_live},
100 BackupCursor.Failure{None{},
101 source.error(error)})
102 case (session, Done{frame}):
103 match frame:
104 case CursorCodec.CursorEnd{}:
105 (OwnerSource{session, snapshot_id, cursor_id, False{}},
106 BackupCursor.advance.owner(stream, frame))
107 case CursorCodec.CursorItem{_, _}:
108 (OwnerSource{session, snapshot_id, cursor_id, True{}},
109 BackupCursor.advance.owner(stream, frame))
110
111def source.step.state(
112 file: File,
113 advanced: OwnerSource & BackupCursor.Step
114) -> State:
115 match advanced:
116 case (source, step): Output{file, source, step}
117
118def source.step.cloned(
119 file: File,
120 stream: BackupCursor.Stream,
121 sources: OwnerSource & OwnerSource
122) -> IO(State):
123 match sources:
124 case (request_source, retained_source):
125 match request_source:
126 case OwnerSource{session, _, cursor_id, _}:
127 do IO<State>:
128 pair : Client.Session &
129 Result<&1, &1, Client.Error, CursorCodec.Frame> <-
130 ByteClient.snapshot.cursor.next(session, cursor_id)
131 return source.step.state(file,
132 source.step.result(retained_source, stream, pair))
133
134def source.step(
135 source: OwnerSource,
136 stream: BackupCursor.Stream,
137 file: File
138) -> IO(State):
139 match stream:
140 case BackupCursor.Stream{BackupCursor.CursorBlocks{}, _, _, _, _, _, _, _, _}:
141 source.step.cloned(file, stream, source.clone(source))
142 case _:
143 IO.pure(State, Output{file, source,
144 BackupCursor.advance.owner(stream, CursorCodec.CursorEnd{})})
145
146def close.failed.snapshot(
147 file: File,
148 +temporary: String,
149 session: Client.Session,
150 snapshot_id: U32,
151 error: Backup.Error
152) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
153 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
154 _closed : Client.Session & Result<&1, &1, Client.Error, Unit> <-
155 ByteClient.snapshot.close(session, snapshot_id)
156 BackupCursorIo.close.error(file, temporary, None{}, error)
157
158def close.failed.cursor(
159 file: File,
160 +temporary: String,
161 snapshot_id: U32,
162 error: Backup.Error,
163 closed: Client.Session & Result<&1, &1, Client.Error, Unit>
164) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
165 match closed:
166 case (after_cursor, _):
167 close.failed.snapshot(file, temporary, after_cursor, snapshot_id, error)
168
169def close.failed.source(
170 file: File,
171 +temporary: String,
172 source: OwnerSource,
173 error: Backup.Error
174) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
175 match source:
176 case OwnerSource{session, snapshot_id, cursor_id, cursor_live}:
177 match cursor_live:
178 case True{}:
179 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
180 closed : Client.Session &
181 Result<&1, &1, Client.Error, Unit> <-
182 ByteClient.snapshot.cursor.close(session, cursor_id)
183 close.failed.cursor(file, temporary, snapshot_id,
184 error, closed)
185 case False{}:
186 close.failed.snapshot(file, temporary, session, snapshot_id, error)
187
188def close.success.snapshot(
189 +destination: String,
190 +temporary: String,
191 file: File,
192 pair: Client.Session & Result<&1, &1, Client.Error, Unit>
193) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
194 match pair:
195 case (_, Fail{error}):
196 BackupCursorIo.close.error(file, temporary, None{},
197 source.error(error))
198 case (_, Done{Unit{}}):
199 BackupCursorIo.close.file(destination, temporary, file)
200
201def close.success(
202 +destination: String,
203 +temporary: String,
204 file: File,
205 source: OwnerSource
206) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
207 match source:
208 case OwnerSource{session, snapshot_id, _, _}:
209 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
210 closed : Client.Session & Result<&1, &1, Client.Error, Unit> <-
211 ByteClient.snapshot.close(session, snapshot_id)
212 close.success.snapshot(destination, temporary, file, closed)
213
214def loop.abort(
215 +temporary: String,
216 state: State
217) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
218 match state:
219 case Need{file, source, _}:
220 close.failed.source(file, temporary, source, Backup.ResourceLimit{})
221 case Output{file, source, _}:
222 close.failed.source(file, temporary, source, Backup.ResourceLimit{})
223 case Written{(file, _), source, _}:
224 close.failed.source(file, temporary, source, Backup.ResourceLimit{})
225
226
227def loop(
228 fuel: Nat,
229 +destination: String,
230 +temporary: String,
231 state: State
232) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
233 match fuel:
234 case 0n: loop.abort(temporary, state)
235 case 1n+rest:
236 match state:
237 case Need{file, source, stream}:
238 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
239 advanced : State <- source.step(source, stream, file)
240 loop(rest, destination, temporary, advanced)
241 case Output{file, source, step}:
242 match step:
243 case BackupCursor.Failure{_, error}:
244 close.failed.source(file, temporary, source, error)
245 case BackupCursor.Continue{stream}:
246 loop(rest, destination, temporary, Need{file, source, stream})
247 case BackupCursor.Chunk{stream, bytes}:
248 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
249 written : File & Result<&1, &1, U32 & String, Unit> <-
250 Fs.write_bytes(file, bytes)
251 loop(rest, destination, temporary,
252 Written{written, source, stream})
253 case BackupCursor.End{}:
254 close.success(destination, temporary, file, source)
255 case Written{pair, source, stream}:
256 match pair:
257 case (file, Fail{(code, message)}):
258 close.failed.source(file, temporary, source,
259 Backup.SourceFailure{code, message})
260 case (file, Done{Unit{}}):
261 loop(rest, destination, temporary, Need{file, source, stream})
262
263def close.failed.snapshot.no_file(
264 session: Client.Session,
265 snapshot_id: U32
266) -> IO(Unit):
267 do IO<Unit>:
268 _closed : Client.Session & Result<&1, &1, Client.Error, Unit> <-
269 ByteClient.snapshot.close(session, snapshot_id)
270 return Unit{}
271
272def close.failed.source.no_file.cursor(
273 pair: Client.Session & Result<&1, &1, Client.Error, Unit>,
274 snapshot_id: U32
275) -> IO(Unit):
276 match pair:
277 case (session, _): close.failed.snapshot.no_file(session, snapshot_id)
278
279def close.failed.source.no_file(source: OwnerSource) -> IO(Unit):
280 match source:
281 case OwnerSource{session, snapshot_id, cursor_id, cursor_live}:
282 match cursor_live:
283 case True{}:
284 do IO<Unit>:
285 closed : Client.Session &
286 Result<&1, &1, Client.Error, Unit> <-
287 ByteClient.snapshot.cursor.close(session, cursor_id)
288 close.failed.source.no_file.cursor(closed, snapshot_id)
289 case False{}:
290 close.failed.snapshot.no_file(session, snapshot_id)
291
292def error.io(pair: U32 & String) -> BackupIo.Error:
293 match pair:
294 case (17, _): BackupIo.AlreadyExists{}
295 case (code, message): BackupIo.Io{code, message}
296
297def initial.failed.io(
298 initial: BackupCursor.Step,
299 pair: U32 & String
300) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
301 match pair:
302 case (code, message):
303 BackupCursorIo.initial.failed(initial, BackupIo.Io{code, message})
304
305def file.opened(
306 +destination: String,
307 +temporary: String,
308 source: OwnerSource,
309 initial: BackupCursor.Step,
310 fuel: Nat,
311 opened: Result<&1, &1, U32 & String, File>
312) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
313 match opened:
314 case Fail{error}:
315 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
316 _closed : Unit <- close.failed.source.no_file(source)
317 BackupCursorIo.cleanup(temporary)
318 initial.failed.io(initial, error)
319 case Done{file}:
320 loop(fuel, destination, temporary, Output{file, source, initial})
321
322def directory.created(
323 +destination: String,
324 +temporary: String,
325 source: OwnerSource,
326 initial: BackupCursor.Step,
327 fuel: Nat,
328 created: Result<&1, &1, U32 & String, Unit>
329) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
330 match created:
331 case Fail{error}:
332 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
333 _closed : Unit <- close.failed.source.no_file(source)
334 BackupCursorIo.initial.failed(initial, error.io(error))
335 case Done{Unit{}}:
336 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
337 opened : Result<&1, &1, U32 & String, File> <-
338 File.open(BackupCursorIo.path.clone(temporary ++ "/BACKUP"), "w")
339 file.opened(destination, temporary, source, initial, fuel, opened)
340
341def destination.checked(
342 +destination: String,
343 +temporary: String,
344 source: OwnerSource,
345 initial: BackupCursor.Step,
346 fuel: Nat,
347 exists: Result<&1, &1, U32 & String, Bool>
348) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
349 match exists:
350 case Fail{error}:
351 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
352 _closed : Unit <- close.failed.source.no_file(source)
353 BackupCursorIo.initial.failed(initial, error.io(error))
354 case Done{True{}}:
355 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
356 _closed : Unit <- close.failed.source.no_file(source)
357 BackupCursorIo.initial.failed(initial, BackupIo.AlreadyExists{})
358 case Done{False{}}:
359 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
360 created : Result<&1, &1, U32 & String, Unit> <-
361 Fs.make_dir(BackupCursorIo.path.clone(temporary))
362 directory.created(destination, temporary, source, initial, fuel,
363 created)
364
365
366def export(
367 +destination: String,
368 source: OwnerSource,
369 entry_count: Nat,
370 fuel: Nat
371) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
372 +temporary = destination ++ ".backup-060.tmp"
373 do IO<Result<&1, &1, BackupIo.Error, Unit>>:
374 initial : BackupCursor.Step <- IO.pure(BackupCursor.Step,
375 BackupCursor.start.owner(entry_count))
376 exists : Result<&1, &1, U32 & String, Bool> <-
377 Fs.exists(BackupCursorIo.path.clone(destination))
378 destination.checked(destination, temporary, source, initial, fuel, exists)