1	import Base
2	import ./Backup.bend as Backup
3	import ./PublicationPolicy.bend as PublicationPolicy
4	import ../ByteMemTable.bend as ByteMemTable
5	import ../CrashPoint.bend as CrashPoint
6	import ../Fs.bend as Fs
7	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
8	
9	# Distinguishes pre-publication IO failures from an uncertain durable commit.
10	type 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.
18	type 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	
23	def 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	
28	def 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.
32	def 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	
38	def 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	
48	def 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	
53	def 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	
70	def 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	
84	def 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	
101	def 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	
116	def 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	
134	def 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	
149	def 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	
164	def 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.
175	def 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	
205	def 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	
219	def 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	
236	def 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.
253	def 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	
262	def 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.
271	def export(+destination: String, entries: List<&1, ByteMemTable.Entry>) -> IO(Result<&1, &1, Error, Unit>):
272	  start.result(destination, Backup.start(entries))