1	import Base
2	import ../ByteDurableDb.bend as Db
3	import ../ByteMemTable.bend as ByteMemTable
4	import ../ByteWal.bend as ByteWal
5	import ../CrashPoint.bend as CrashPoint
6	import ../Fs.bend as Fs
7	import ./Restore.bend as Restore
8	import ./PublicationPolicy.bend as PublicationPolicy
9	import ./RestorePolicy.bend as RestorePolicy
10	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
11	
12	# Classifies exchange validation, staging, publication, and durability failures.
13	type Error is Data:
14	  InvalidFormat{}
15	  ResourceLimit{}
16	  InvalidOrder{}
17	  ChecksumMismatch{}
18	  CountMismatch{}
19	  Storage{error: Db.Error}
20	  Io{code: U32, message: String}
21	  DestinationExists{}
22	  TemporaryExists{}
23	  VerificationFailed{}
24	  CommitUnknown{code: U32, message: String}
25	
26	# Retains a single staged record and database handle between WAL writes.
27	type WriteState is Type:
28	  WriteNeed{handle: Db.Handle, entries: List<&1, ByteMemTable.Entry>}
29	  WriteBatch{handle: Db.Handle, page: WritePage}
30	  WriteResult{pair: Db.Handle & Result<&1, &1, Db.Error, Unit>,
31	    rest: List<&1, ByteMemTable.Entry>}
32	
33	# Carries one bounded mutation page and the remaining ordered entries.
34	type WritePage is Type:
35	  WritePage{batch: List<&1, ByteMemTable.Entry>, rest: List<&1, ByteMemTable.Entry>}
36	
37	def decode.error(error: Restore.Error) -> Error:
38	  match error:
39	    case Restore.InvalidFormat{}: InvalidFormat{}
40	    case Restore.ResourceLimit{}: ResourceLimit{}
41	    case Restore.InvalidOrder{}: InvalidOrder{}
42	    case Restore.ChecksumMismatch{}: ChecksumMismatch{}
43	    case Restore.CountMismatch{}: CountMismatch{}
44	
45	def write.finish(pair: Db.Handle & Result<&1, &1, Db.Error, Unit>) -> WriteState:
46	  WriteResult{pair, Nil{}}
47	
48	def write.result(
49	  rest: List<&1, ByteMemTable.Entry>,
50	  pair: Db.Handle & Result<&1, &1, Db.Error, Unit>
51	) -> WriteState:
52	  WriteResult{pair, rest}
53	
54	def write.page.go(
55	  fuel: Nat,
56	  entries: List<&1, ByteMemTable.Entry>,
57	  reversed: List<&1, ByteMemTable.Entry>
58	) -> WritePage:
59	  match fuel:
60	    case 0n: WritePage{List.reverse(&1, ByteMemTable.Entry, reversed), entries}
61	    case 1n+rest:
62	      match entries:
63	        case Nil{}: WritePage{List.reverse(&1, ByteMemTable.Entry, reversed), Nil{}}
64	        case Con{entry, tail}: write.page.go(rest, tail, Con{entry, reversed})
65	
66	def write.page.ready(
67	  measured: List<&1, ByteMemTable.Entry> & Nat
68	) -> WritePage:
69	  match measured:
70	    case (entries, count):
71	      write.page.go(Nat.min(RestorePolicy.write.page.size(), count), entries, Nil{})
72	
73	def write.page(entries: List<&1, ByteMemTable.Entry>) -> WritePage:
74	  write.page.ready(ByteMemTable.count.keep(entries))
75	
76	def write.mutations(entries: List<&1, ByteMemTable.Entry>) -> List<&1, ByteWal.Mutation>:
77	  match entries:
78	    case Nil{}: Nil{}
79	    case Con{ByteMemTable.Entry{key, value}, tail}:
80	      match value:
81	        case None{}: Con{ByteWal.Delete{key}, write.mutations(tail)}
82	        case Some{value}: Con{ByteWal.Put{key, value}, write.mutations(tail)}
83	
84	def write.loop(fuel: Nat, state: WriteState) -> IO(Db.Handle & Result<&1, &1, Db.Error, Unit>):
85	  match fuel:
86	    case 0n:
87	      match state:
88	        case WriteNeed{handle, _}:
89	          IO.pure(Db.Handle & Result<&1, &1, Db.Error, Unit>,
90	            (handle, Fail{Db.ResourceLimit{}}))
91	        case WriteBatch{handle, _}:
92	          IO.pure(Db.Handle & Result<&1, &1, Db.Error, Unit>,
93	            (handle, Fail{Db.ResourceLimit{}}))
94	        case WriteResult{pair, _}: IO.pure(Db.Handle & Result<&1, &1, Db.Error, Unit>, pair)
95	    case 1n+rest_fuel:
96	      match state:
97	        case WriteNeed{handle, entries}:
98	          match entries:
99	            case Nil{}:
100	              IO.pure(Db.Handle & Result<&1, &1, Db.Error, Unit>,
101	                (handle, Done{Unit{}}))
102	            case Con{_, _}:
103	              write.loop(rest_fuel, WriteBatch{handle, write.page(entries)})
104	        case WriteBatch{handle, WritePage{batch, rest}}:
105	          do IO<Db.Handle & Result<&1, &1, Db.Error, Unit>>:
106	            pair : Db.Handle & Result<&1, &1, Db.Error, Unit> <-
107	              Db.write_batch(handle, write.mutations(batch))
108	            write.loop(rest_fuel, WriteResult{pair, rest})
109	        case WriteResult{pair, rest}:
110	          match pair:
111	            case (handle, Fail{error}):
112	              IO.pure(Db.Handle & Result<&1, &1, Db.Error, Unit>,
113	                (handle, Fail{error}))
114	            case (handle, Done{Unit{}}):
115	              write.loop(rest_fuel, WriteNeed{handle, rest})
116	
117	def remove.ignored(path: String) -> IO(Unit):
118	  do IO<Unit>:
119	    _removed : Result<&1, &1, U32 & String, Unit> <- Fs.remove(path)
120	    return Unit{}
121	
122	def cleanup.directory(+path: String) -> IO(Unit):
123	  do IO<Unit>:
124	    remove.ignored(path ++ "/WAL")
125	    remove.ignored(path ++ "/WAL.tmp")
126	    remove.ignored(path ++ "/SST")
127	    remove.ignored(path ++ "/SST.tmp")
128	    remove.ignored(path ++ "/LOCK")
129	    remove.ignored(path)
130	
131	def cleanup.result(
132	  path: String,
133	  result: Result<&1, &1, Db.Error, Unit>
134	) -> IO(Result<&1, &1, Error, Unit>):
135	  match result:
136	    case Fail{error}:
137	      do IO<Result<&1, &1, Error, Unit>>:
138	        cleanup.directory(path)
139	        return Fail{Storage{error}}
140	    case Done{Unit{}}:
141	      do IO<Result<&1, &1, Error, Unit>>:
142	        cleanup.directory(path)
143	        return Done{Unit{}}
144	
145	def cleanup.handle(
146	  path: String,
147	  handle: Db.Handle,
148	  error: Db.Error
149	) -> IO(Result<&1, &1, Error, Unit>):
150	  do IO<Result<&1, &1, Error, Unit>>:
151	    _closed : Result<&1, &1, Db.Error, Unit> <- Db.close(handle)
152	    cleanup.result(path, Fail{error})
153	
154	def flush.closed(
155	  path: String,
156	  result: Result<&1, &1, Db.Error, Unit>
157	) -> IO(Result<&1, &1, Error, Unit>):
158	  match result:
159	    case Fail{error}: cleanup.result(path, Fail{error})
160	    case Done{Unit{}}: IO.pure(Result<&1, &1, Error, Unit>, Done{Unit{}})
161	
162	def flush.result(
163	  path: String,
164	  pair: Db.Handle & Result<&1, &1, Db.Error, Unit>
165	) -> IO(Result<&1, &1, Error, Unit>):
166	  match pair:
167	    case (handle, Fail{error}): cleanup.handle(path, handle, error)
168	    case (handle, Done{Unit{}}):
169	      do IO<Result<&1, &1, Error, Unit>>:
170	        closed : Result<&1, &1, Db.Error, Unit> <- Db.close(handle)
171	        result : Result<&1, &1, Error, Unit> <- flush.closed(path, closed)
172	        return result
173	
174	
175	def write.result.path(
176	  path: String,
177	  pair: Db.Handle & Result<&1, &1, Db.Error, Unit>
178	) -> IO(Result<&1, &1, Error, Unit>):
179	  match pair:
180	    case (handle, Fail{error}): cleanup.handle(path, handle, error)
181	    case (handle, Done{Unit{}}):
182	      do IO<Result<&1, &1, Error, Unit>>:
183	        flushed : Db.Handle & Result<&1, &1, Db.Error, Unit> <- Db.flush(handle)
184	        result : Result<&1, &1, Error, Unit> <- flush.result(path, flushed)
185	        return result
186	
187	# Rejects a staged database whose recovered entry count differs from the exchange.
188	def verify_count(path: String, equal: Bool) -> IO(Result<&1, &1, Error, Unit>):
189	  match equal:
190	    case False{}:
191	      do IO<Result<&1, &1, Error, Unit>>:
192	        cleanup.directory(path)
193	        return Fail{VerificationFailed{}}
194	    case True{}: IO.pure(Result<&1, &1, Error, Unit>, Done{Unit{}})
195	
196	def verified.closed(
197	  path: String,
198	  expected: Nat,
199	  count: Nat,
200	  closed: Result<&1, &1, Db.Error, Unit>
201	) -> IO(Result<&1, &1, Error, Unit>):
202	  match closed:
203	    case Fail{error}:
204	      do IO<Result<&1, &1, Error, Unit>>:
205	        cleanup.directory(path)
206	        return Fail{Storage{error}}
207	    case Done{Unit{}}:
208	      verify_count(path, Nat.is_eq(count, expected))
209	
210	def path.reverse(path: String, reversed: String) -> String:
211	  match path:
212	    case SNil{}: reversed
213	    case SCon{character, rest}:
214	      path.reverse(rest, SCon{character, reversed})
215	
216	def path.clone(path: String) -> String:
217	  path.reverse(path.reverse(path, SNil{}), SNil{})
218	
219	
220	def verified.stats(
221	  path: String,
222	  expected: Nat,
223	  pair: Db.Handle & Db.Stats
224	) -> IO(Result<&1, &1, Error, Unit>):
225	  match pair:
226	    case (handle, Db.Stats{entries, _}):
227	      do IO<Result<&1, &1, Error, Unit>>:
228	        closed : Result<&1, &1, Db.Error, Unit> <- Db.close(handle)
229	        verified.closed(path, expected, entries, closed)
230	
231	# Opens and checks the staged database before publication.
232	def verified(
233	  path: String,
234	  expected: Nat,
235	  opened: Result<&1, &1, Db.Error, Db.Handle>
236	) -> IO(Result<&1, &1, Error, Unit>):
237	  match opened:
238	    case Fail{error}:
239	      do IO<Result<&1, &1, Error, Unit>>:
240	        cleanup.directory(path)
241	        return Fail{Storage{error}}
242	    case Done{handle}: verified.stats(path, expected, Db.stats(handle))
243	
244	# Validates an exchange, builds a sibling database, verifies it, then publishes it.
245	def publish.parent.sync.failed(
246	  disposition: PublicationPolicy.Failure,
247	  error: U32 & String
248	) -> IO(Result<&1, &1, Error, Unit>):
249	  match disposition:
250	    case PublicationPolicy.CommitUnknown{}:
251	      match error:
252	        case (code, message):
253	          IO.pure(Result<&1, &1, Error, Unit>,
254	            Fail{CommitUnknown{code, "parent directory sync failed: " ++ message}})
255	    case PublicationPolicy.KnownFailure{}:
256	      match error:
257	        case (code, message):
258	          IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
259	
260	def publish.parent.synced(
261	  result: Result<&1, &1, U32 & String, Unit>
262	) -> IO(Result<&1, &1, Error, Unit>):
263	  match result:
264	    case Fail{error}:
265	      publish.parent.sync.failed(
266	        PublicationPolicy.failure(PublicationPolicy.AfterRename{}), error)
267	    case Done{Unit{}}:
268	      do IO<Result<&1, &1, Error, Unit>>:
269	        _checkpoint : Result<&1, &1, U32 & String, Unit> <-
270	          CrashPoint.hit("restore.parent_synced")
271	        return Done{Unit{}}
272	
273	def publish.rename.failed(
274	  +temporary: String,
275	  disposition: PublicationPolicy.Failure,
276	  error: U32 & String
277	) -> IO(Result<&1, &1, Error, Unit>):
278	  match disposition:
279	    case PublicationPolicy.KnownFailure{}:
280	      match error:
281	        case (code, message):
282	          do IO<Result<&1, &1, Error, Unit>>:
283	            cleanup.directory(temporary)
284	            return Fail{Io{code, "exclusive rename failed: " ++ message}}
285	    case PublicationPolicy.CommitUnknown{}:
286	      match error:
287	        case (code, message):
288	          IO.pure(Result<&1, &1, Error, Unit>, Fail{CommitUnknown{code, message}})
289	
290	def publish.renamed(
291	  +destination: String,
292	  +temporary: String,
293	  result: Result<&1, &1, U32 & String, Unit>
294	) -> IO(Result<&1, &1, Error, Unit>):
295	  match result:
296	    case Fail{error}:
297	      publish.rename.failed(temporary,
298	        PublicationPolicy.failure(PublicationPolicy.BeforeRename{}), error)
299	    case Done{Unit{}}:
300	      do IO<Result<&1, &1, Error, Unit>>:
301	        _checkpoint : Result<&1, &1, U32 & String, Unit> <-
302	          CrashPoint.hit("restore.renamed")
303	        synced : Result<&1, &1, U32 & String, Unit> <-
304	          Fs.fsync(path.clone(destination ++ "/.."))
305	        published : Result<&1, &1, Error, Unit> <- publish.parent.synced(synced)
306	        return published
307	
308	def publish.synced(
309	  +destination: String,
310	  +temporary: String,
311	  result: Result<&1, &1, U32 & String, Unit>
312	) -> IO(Result<&1, &1, Error, Unit>):
313	  match result:
314	    case Fail{(code, message)}:
315	      do IO<Result<&1, &1, Error, Unit>>:
316	        cleanup.directory(temporary)
317	        return Fail{Io{code, "staged directory sync failed: " ++ message}}
318	    case Done{Unit{}}:
319	      do IO<Result<&1, &1, Error, Unit>>:
320	        _checkpoint : Result<&1, &1, U32 & String, Unit> <-
321	          CrashPoint.hit("restore.before_rename")
322	        renamed : Result<&1, &1, U32 & String, Unit> <-
323	          Fs.rename_noreplace(path.clone(temporary), path.clone(destination))
324	        published : Result<&1, &1, Error, Unit> <- publish.renamed(destination, temporary, renamed)
325	        return published
326	
327	def publish.verified(
328	  +destination: String,
329	  +temporary: String,
330	  result: Result<&1, &1, Error, Unit>
331	) -> IO(Result<&1, &1, Error, Unit>):
332	  match result:
333	    case Fail{error}: IO.pure(Result<&1, &1, Error, Unit>, Fail{error})
334	    case Done{Unit{}}:
335	      do IO<Result<&1, &1, Error, Unit>>:
336	        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path.clone(temporary))
337	        published : Result<&1, &1, Error, Unit> <- publish.synced(destination, temporary, synced)
338	        return published
339	
340	# Reopens staged data and checks that recovery sees the expected record count.
341	def publish(
342	  +destination: String,
343	  +temporary: String,
344	  expected: Nat
345	) -> IO(Result<&1, &1, Error, Unit>):
346	  do IO<Result<&1, &1, Error, Unit>>:
347	    opened : Result<&1, &1, Db.Error, Db.Handle> <- Db.open_existing(path.clone(temporary))
348	    verified : Result<&1, &1, Error, Unit> <- verified(temporary, expected, opened)
349	    published : Result<&1, &1, Error, Unit> <- publish.verified(destination, temporary, verified)
350	    return published
351	
352	def publish.result(
353	  destination: String,
354	  temporary: String,
355	  expected: Nat,
356	  populated: Result<&1, &1, Error, Unit>
357	) -> IO(Result<&1, &1, Error, Unit>):
358	  match populated:
359	    case Fail{error}: IO.pure(Result<&1, &1, Error, Unit>, Fail{error})
360	    case Done{Unit{}}: publish(destination, temporary, expected)
361	
362	
363	def temporary.create(
364	  +destination: String,
365	  +temporary: String,
366	  created: Result<&1, &1, Db.Error, Db.Handle>,
367	  measured: List<&1, ByteMemTable.Entry> & Nat
368	) -> IO(Result<&1, &1, Error, Unit>):
369	  match created:
370	    case Fail{error}: IO.pure(Result<&1, &1, Error, Unit>, Fail{Storage{error}})
371	    case Done{handle}:
372	      match measured:
373	        case (entries, +count):
374	          do IO<Result<&1, &1, Error, Unit>>:
375	            written : Db.Handle & Result<&1, &1, Db.Error, Unit> <-
376	              write.loop(RestorePolicy.write.fuel(count),
377	                WriteNeed{handle, entries})
378	            populated : Result<&1, &1, Error, Unit> <- write.result.path(temporary, written)
379	            published : Result<&1, &1, Error, Unit> <-
380	              publish.result(destination, temporary, count, populated)
381	            return published
382	
383	
384	def temporary.decoded(
385	  +destination: String,
386	  +temporary: String,
387	  decoded: Result<&1, &1, Restore.Error, Restore.Decoded>
388	) -> IO(Result<&1, &1, Error, Unit>):
389	  match decoded:
390	    case Fail{error}:
391	      IO.pure(Result<&1, &1, Error, Unit>, Fail{decode.error(error)})
392	    case Done{Restore.Decoded{entries}}:
393	      do IO<Result<&1, &1, Error, Unit>>:
394	        created : Result<&1, &1, Db.Error, Db.Handle> <- Db.create(path.clone(temporary))
395	        result : Result<&1, &1, Error, Unit> <- temporary.create(destination,
396	          temporary, created, ByteMemTable.count.keep(entries))
397	        return result
398	
399	
400	def temporary.checked(
401	  +destination: String,
402	  +temporary: String,
403	  input: Bytes.Bytes,
404	  result: Result<&1, &1, U32 & String, Bool>
405	) -> IO(Result<&1, &1, Error, Unit>):
406	  match result:
407	    case Fail{(code, message)}:
408	      IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
409	    case Done{True{}}:
410	      IO.pure(Result<&1, &1, Error, Unit>, Fail{TemporaryExists{}})
411	    case Done{False{}}:
412	      temporary.decoded(destination, temporary, Restore.decode(input))
413	
414	def destination.checked(
415	  +destination: String,
416	  +temporary: String,
417	  input: Bytes.Bytes,
418	  result: Result<&1, &1, U32 & String, Bool>
419	) -> IO(Result<&1, &1, Error, Unit>):
420	  match result:
421	    case Fail{(code, message)}:
422	      IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
423	    case Done{True{}}:
424	      IO.pure(Result<&1, &1, Error, Unit>, Fail{DestinationExists{}})
425	    case Done{False{}}:
426	      do IO<Result<&1, &1, Error, Unit>>:
427	        exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path.clone(temporary))
428	        temporary.checked(destination, temporary, input, exists)
429	
430	
431	# Validates an exchange before staging and atomically publishing a database.
432	def restore(
433	  +destination: String,
434	  input: Bytes.Bytes
435	) -> IO(Result<&1, &1, Error, Unit>):
436	  +temporary = path.clone(destination ++ ".restore-060.tmp")
437	  do IO<Result<&1, &1, Error, Unit>>:
438	    exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path.clone(destination))
439	    destination.checked(destination, temporary, input, exists)
440	
441	def source.read.length(
442	  +destination: String,
443	  input: Bytes.Bytes,
444	  equal: Bool
445	) -> IO(Result<&1, &1, Error, Unit>):
446	  match equal:
447	    case False{}: IO.pure(Result<&1, &1, Error, Unit>, Fail{InvalidFormat{}})
448	    case True{}: restore(destination, input)
449	
450	def source.read.checked(
451	  +destination: String,
452	  +expected_size: Nat,
453	  input: Bytes.Bytes
454	) -> IO(Result<&1, &1, Error, Unit>):
455	  match input:
456	    case Bytes.Bytes{+len, _}:
457	      source.read.length(destination, input,
458	        Nat.is_eq(U32.to_nat(len), expected_size))
459	
460	def source.read.result(
461	  +destination: String,
462	  +expected_size: Nat,
463	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
464	) -> IO(Result<&1, &1, Error, Unit>):
465	  match pair:
466	    case (file, Fail{(code, message)}):
467	      do IO<Result<&1, &1, Error, Unit>>:
468	        _closed : Unit <- File.close(file)
469	        return Fail{Io{code, message}}
470	    case (file, Done{input}):
471	      do IO<Result<&1, &1, Error, Unit>>:
472	        _closed : Unit <- File.close(file)
473	        source.read.checked(destination, expected_size, input)
474	
475	def source.opened(
476	  +destination: String,
477	  +size: Nat,
478	  opened: Result<&1, &1, U32 & String, File>
479	) -> IO(Result<&1, &1, Error, Unit>):
480	  match opened:
481	    case Fail{(code, message)}:
482	      IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
483	    case Done{file}:
484	      do IO<Result<&1, &1, Error, Unit>>:
485	        pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
486	          Fs.read_at(file, 0, U32.from_nat(size))
487	        source.read.result(destination, size, pair)
488	
489	def source.sized.checked(
490	  +destination: String,
491	  +source: String,
492	  size: Nat,
493	  valid: Bool
494	) -> IO(Result<&1, &1, Error, Unit>):
495	  match valid:
496	    case False{}:
497	      IO.pure(Result<&1, &1, Error, Unit>, Fail{ResourceLimit{}})
498	    case True{}:
499	      do IO<Result<&1, &1, Error, Unit>>:
500	        opened : Result<&1, &1, U32 & String, File> <- File.open(source, "r")
501	        source.opened(destination, size, opened)
502	
503	def source.sized(
504	  +destination: String,
505	  +source: String,
506	  +size: Nat
507	) -> IO(Result<&1, &1, Error, Unit>):
508	  source.sized.checked(destination, source, size,
509	    RestorePolicy.exchange.fits(RestorePolicy.max.exchange.bytes(), size))
510	
511	def source.size(
512	  +destination: String,
513	  +source: String,
514	  measured: Result<&1, &1, U32 & String, Nat>
515	) -> IO(Result<&1, &1, Error, Unit>):
516	  match measured:
517	    case Fail{(code, message)}:
518	      IO.pure(Result<&1, &1, Error, Unit>, Fail{Io{code, message}})
519	    case Done{size}: source.sized(destination, source, size)
520	
521	# Reads a bounded backup bundle file and restores its validated exchange.
522	def restore_from(
523	  +destination: String,
524	  +bundle: String
525	) -> IO(Result<&1, &1, Error, Unit>):
526	  do IO<Result<&1, &1, Error, Unit>>:
527	    measured : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(bundle ++ "/BACKUP")
528	    source.size(destination, bundle ++ "/BACKUP", measured)