1	import Base
2	import ../Fs.bend as Fs
3	import ../coordinator/Client.bend as Client
4	import ../coordinator/ByteClient.bend as ByteClient
5	import ../coordinator/CursorCodec.bend as CursorCodec
6	import ../coordinator/Io.bend as ClientIo
7	import ../coordinator/UnixSocketIo.bend as ClientSocket
8	import ../coordinator/Protocol.bend as Protocol
9	import ./BackupCursor.bend as BackupCursor
10	import ./BackupCursorIo.bend as BackupCursorIo
11	import ./BackupIo.bend as BackupIo
12	import ./BackupOwnerPolicy.bend as Policy
13	import ./Backup.bend as Backup
14	
15	# Tracks the retained owner snapshot and its export cursor lifecycle.
16	type OwnerSource is Type:
17	  OwnerSource{session: Client.Session, snapshot_id: U32, cursor_id: U32,
18	    cursor_live: Bool}
19	
20	# Carries the bounded backup writer through remote cursor transitions.
21	type 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	
27	def 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	
49	def 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	# Converts socket and protocol errors into the pure owner-error policy.
57	def 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	
90	def 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	
111	def 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	
118	def 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	
134	def 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	
146	def 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	
158	def 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	
169	def 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	
188	def 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	
201	def 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	
214	def 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	# Advances backup IO state without reading beyond its declared work budget.
227	def 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	
263	def 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	
272	def 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	
279	def 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	
292	def error.io(pair: U32 & String) -> BackupIo.Error:
293	  match pair:
294	    case (17, _): BackupIo.AlreadyExists{}
295	    case (code, message): BackupIo.Io{code, message}
296	
297	def 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	
305	def 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	
322	def 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	
341	def 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	# Exports a retained owner cursor through bounded checksummed backup blocks.
366	def 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)