1	import Base
2	import ../Fs.bend as Fs
3	import ./OwnerGeneration.bend as Generation
4	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
5	
6	# Separates host failures, corrupt state, and generation exhaustion.
7	type Error is Data:
8	  Host{code: U32, message: String}
9	  Invalid{}
10	  Exhausted{}
11	
12	# Preserves the raw filesystem error in the module result type.
13	def host(error: U32 & String) -> Error:
14	  match error:
15	    case (code, message): Host{code, message}
16	
17	def path.pair.go(path: String, left: String, right: String) -> String & String:
18	  match path:
19	    case SNil{}: (String.reverse(left), String.reverse(right))
20	    case SCon{+character, rest}:
21	      path.pair.go(rest, SCon{character, left}, SCon{character, right})
22	
23	def path.pair(path: String) -> String & String:
24	  path.pair.go(path, SNil{}, SNil{})
25	
26	def path.triple.go(
27	  path: String,
28	  first: String,
29	  second: String,
30	  third: String
31	) -> String & String & String:
32	  match path:
33	    case SNil{}:
34	      (String.reverse(first), (String.reverse(second), String.reverse(third)))
35	    case SCon{+character, rest}:
36	      path.triple.go(rest, SCon{character, first}, SCon{character, second},
37	        SCon{character, third})
38	
39	def path.triple(path: String) -> String & String & String:
40	  path.triple.go(path, SNil{}, SNil{}, SNil{})
41	
42	def decode.nonzero(value: U32, is_zero: Bool) -> Result<&1, &1, Error, U32>:
43	  match is_zero:
44	    case True{}: Fail{Invalid{}}
45	    case False{}: Done{value}
46	
47	def decode.value(copies: U32 & U32) -> Result<&1, &1, Error, U32>:
48	  match copies:
49	    case (value, test): decode.nonzero(value, U32.is_zero(test))
50	
51	def decode.word(bytes: Bytes.Bytes) -> Result<&1, &1, Error, U32>:
52	  decode.value(Generation.u32.clone.pair(Generation.u32.read(bytes)))
53	
54	# Accepts only one nonzero, fixed-width owner generation.
55	def decode(bytes: Bytes.Bytes, size: Nat) -> Result<&1, &1, Error, U32>:
56	  match size:
57	    case 4n: decode.word(bytes)
58	    case _: Fail{Invalid{}}
59	
60	def read.bytes(
61	  file: File,
62	  result: Result<&1, &1, U32 & String, Bytes.Bytes>
63	) -> IO(Result<&1, &1, Error, U32>):
64	  match result:
65	    case Fail{error}:
66	      do IO<Result<&1, &1, Error, U32>>:
67	        _closed : Unit <- File.close(file)
68	        return Fail{host(error)}
69	    case Done{bytes}:
70	      do IO<Result<&1, &1, Error, U32>>:
71	        _closed : Unit <- File.close(file)
72	        return decode(bytes, 4n)
73	
74	def read.pair(
75	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
76	) -> IO(Result<&1, &1, Error, U32>):
77	  match pair:
78	    case (file, result): read.bytes(file, result)
79	
80	def read.size(
81	  file: File,
82	  result: Result<&1, &1, U32 & String, Nat>
83	) -> IO(Result<&1, &1, Error, U32>):
84	  match result:
85	    case Fail{error}:
86	      do IO<Result<&1, &1, Error, U32>>:
87	        _closed : Unit <- File.close(file)
88	        return Fail{host(error)}
89	    case Done{4n}:
90	      do IO<Result<&1, &1, Error, U32>>:
91	        pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
92	          Fs.read_at(file, 0, 4)
93	        read.pair(pair)
94	    case Done{_}:
95	      do IO<Result<&1, &1, Error, U32>>:
96	        _closed : Unit <- File.close(file)
97	        return Fail{Invalid{}}
98	
99	def read.opened(
100	  paths: String & String,
101	  opened: Result<&1, &1, U32 & String, File>
102	) -> IO(Result<&1, &1, Error, U32>):
103	  match paths:
104	    case (size_path, retained_path):
105	      match opened:
106	        case Fail{error}:
107	          IO.pure(Result<&1, &1, Error, U32>, Fail{host(error)})
108	        case Done{file}:
109	          do IO<Result<&1, &1, Error, U32>>:
110	            size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(size_path)
111	            read.size(file, size)
112	
113	def read.path(paths: String & String) -> IO(Result<&1, &1, Error, U32>):
114	  match paths:
115	    case (open_path, retained_path):
116	      do IO<Result<&1, &1, Error, U32>>:
117	        opened : Result<&1, &1, U32 & String, File> <- File.open(open_path, "r")
118	        read.opened(path.pair(retained_path), opened)
119	
120	# Reads and validates the persisted generation without decoding text.
121	def read(path: String) -> IO(Result<&1, &1, Error, U32>):
122	  read.path(path.pair(path))
123	
124	def write.completed(
125	  generation: U32,
126	  result: Result<&1, &1, U32 & String, Unit>
127	) -> IO(Result<&1, &1, Error, U32>):
128	  match result:
129	    case Fail{error}:
130	      IO.pure(Result<&1, &1, Error, U32>, Fail{host(error)})
131	    case Done{Unit{}}:
132	      IO.pure(Result<&1, &1, Error, U32>, Done{generation})
133	
134	def write.directory.synced(
135	  directory: String,
136	  generation: U32,
137	  result: Result<&1, &1, U32 & String, Unit>
138	) -> IO(Result<&1, &1, Error, U32>):
139	  match result:
140	    case Fail{error}:
141	      IO.pure(Result<&1, &1, Error, U32>, Fail{host(error)})
142	    case Done{Unit{}}:
143	      do IO<Result<&1, &1, Error, U32>>:
144	        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(directory)
145	        write.completed(generation, synced)
146	
147	def write.renamed.paths(
148	  directories: String & String & String,
149	  generation: U32,
150	  result: Result<&1, &1, U32 & String, Unit>
151	) -> IO(Result<&1, &1, Error, U32>):
152	  match directories:
153	    case (temp_directory, (final_directory, retained_directory)):
154	      match result:
155	        case Fail{error}:
156	          IO.pure(Result<&1, &1, Error, U32>, Fail{host(error)})
157	        case Done{Unit{}}:
158	          do IO<Result<&1, &1, Error, U32>>:
159	            renamed : Result<&1, &1, U32 & String, Unit> <- Fs.rename(temp_directory ++ "/OWNER.GEN.tmp", final_directory ++ "/OWNER.GEN")
160	            write.directory.synced(retained_directory, generation, renamed)
161	
162	def write.renamed(
163	  directory: String,
164	  generation: U32,
165	  result: Result<&1, &1, U32 & String, Unit>
166	) -> IO(Result<&1, &1, Error, U32>):
167	  write.renamed.paths(path.triple(directory), generation, result)
168	
169	def write.temporary.synced(
170	  directory: String,
171	  generation: U32,
172	  result: Result<&1, &1, U32 & String, Unit>
173	) -> IO(Result<&1, &1, Error, U32>):
174	  write.renamed(directory, generation, result)
175	
176	def write.result.paths(
177	  directories: String & String,
178	  generation: U32,
179	  result: Result<&1, &1, U32 & String, Unit>
180	) -> IO(Result<&1, &1, Error, U32>):
181	  match directories:
182	    case (temp_directory, retained_directory):
183	      match result:
184	        case Fail{error}:
185	          IO.pure(Result<&1, &1, Error, U32>, Fail{host(error)})
186	        case Done{Unit{}}:
187	          do IO<Result<&1, &1, Error, U32>>:
188	            synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(temp_directory ++ "/OWNER.GEN.tmp")
189	            write.temporary.synced(retained_directory, generation, synced)
190	
191	def write.result(
192	  directory: String,
193	  generation: U32,
194	  result: Result<&1, &1, U32 & String, Unit>
195	) -> IO(Result<&1, &1, Error, U32>):
196	  write.result.paths(path.pair(directory), generation, result)
197	
198	def write.closed(
199	  directory: String,
200	  generation: U32,
201	  result: File & Result<&1, &1, U32 & String, Unit>
202	) -> IO(Result<&1, &1, Error, U32>):
203	  match result:
204	    case (file, result):
205	      do IO<Result<&1, &1, Error, U32>>:
206	        _closed : Unit <- File.close(file)
207	        write.result(directory, generation, result)
208	
209	def write.opened.generation(
210	  directory: String,
211	  file: File,
212	  generations: U32 & U32
213	) -> IO(Result<&1, &1, Error, U32>):
214	  match generations:
215	    case (encode_generation, retained_generation):
216	      do IO<Result<&1, &1, Error, U32>>:
217	        result : File & Result<&1, &1, U32 & String, Unit> <-
218	          Fs.write_bytes(file, Generation.u32.bytes(encode_generation))
219	        write.closed(directory, retained_generation, result)
220	
221	def write.opened(
222	  directory: String,
223	  generation: U32,
224	  opened: Result<&1, &1, U32 & String, File>
225	) -> IO(Result<&1, &1, Error, U32>):
226	  match opened:
227	    case Fail{error}:
228	      IO.pure(Result<&1, &1, Error, U32>, Fail{host(error)})
229	    case Done{file}:
230	      write.opened.generation(directory, file,
231	        Generation.u32.clone.pair(generation))
232	
233	def write.paths(
234	  paths: String & String,
235	  generation: U32
236	) -> IO(Result<&1, &1, Error, U32>):
237	  match paths:
238	    case (open_directory, retained_directory):
239	      do IO<Result<&1, &1, Error, U32>>:
240	        opened : Result<&1, &1, U32 & String, File> <-
241	          File.open(open_directory ++ "/OWNER.GEN.tmp", "w")
242	        write.opened(retained_directory, generation, opened)
243	
244	# Publishes a generation through a synced temporary file and atomic rename.
245	def write(directory: String, generation: U32) -> IO(Result<&1, &1, Error, U32>):
246	  write.paths(path.pair(directory), generation)
247	
248	def next.decision(
249	  directory: String,
250	  decision: Generation.Decision
251	) -> IO(Result<&1, &1, Error, U32>):
252	  match decision:
253	    case Generation.GenerationExhausted{}:
254	      IO.pure(Result<&1, &1, Error, U32>, Fail{Exhausted{}})
255	    case Generation.GenerationReady{generation}: write(directory, generation)
256	
257	def next.read(
258	  directory: String,
259	  result: Result<&1, &1, Error, U32>
260	) -> IO(Result<&1, &1, Error, U32>):
261	  match result:
262	    case Fail{Invalid{}}:
263	      IO.pure(Result<&1, &1, Error, U32>, Fail{Invalid{}})
264	    case Fail{Host{code, message}}:
265	      IO.pure(Result<&1, &1, Error, U32>, Fail{Host{code, message}})
266	    case Fail{Exhausted{}}:
267	      IO.pure(Result<&1, &1, Error, U32>, Fail{Exhausted{}})
268	    case Done{current}:
269	      next.decision(directory, Generation.next(Some{current}))
270	
271	def initialize.exists(
272	  directories: String & String,
273	  result: Result<&1, &1, U32 & String, Bool>
274	) -> IO(Result<&1, &1, Error, U32>):
275	  match directories:
276	    case (check_directory, retained_directory):
277	      match result:
278	        case Fail{error}:
279	          IO.pure(Result<&1, &1, Error, U32>, Fail{host(error)})
280	        case Done{False{}}: write(retained_directory, 1)
281	        case Done{True{}}:
282	          do IO<Result<&1, &1, Error, U32>>:
283	            current : Result<&1, &1, Error, U32> <- read(check_directory ++ "/OWNER.GEN")
284	            next.read(retained_directory, current)
285	
286	def initialize.paths(paths: String & String) -> IO(Result<&1, &1, Error, U32>):
287	  match paths:
288	    case (check_directory, retained_directory):
289	      do IO<Result<&1, &1, Error, U32>>:
290	        result : Result<&1, &1, U32 & String, Bool> <- Fs.exists(check_directory ++ "/OWNER.GEN")
291	        initialize.exists(path.pair(retained_directory), result)
292	
293	# Starts at one or advances the persisted value before owner publication.
294	def initialize(directory: String) -> IO(Result<&1, &1, Error, U32>):
295	  initialize.paths(path.pair(directory))