src/coordinator/OwnerGenerationIo.bend

mylsm-lsm-store@0.6.0.1 · 295 lines · 33 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/OwnerGeneration.bend as Generation
4import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
5
6# Separates host failures, corrupt state, and generation exhaustion.
7type 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.
13def host(error: U32 & String) -> Error:
14  match error:
15    case (code, message): Host{code, message}
16
17def 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
23def path.pair(path: String) -> String & String:
24  path.pair.go(path, SNil{}, SNil{})
25
26def 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
39def path.triple(path: String) -> String & String & String:
40  path.triple.go(path, SNil{}, SNil{}, SNil{})
41
42def 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
47def 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
51def 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.
55def 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
60def 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
74def 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
80def 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
99def 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
113def 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.
121def read(path: String) -> IO(Result<&1, &1, Error, U32>):
122  read.path(path.pair(path))
123
124def 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
134def 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
147def 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
162def 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
169def 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
176def 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
191def 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
198def 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
209def 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
221def 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
233def 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.
245def write(directory: String, generation: U32) -> IO(Result<&1, &1, Error, U32>):
246  write.paths(path.pair(directory), generation)
247
248def 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
257def 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
271def 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
286def 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.
294def initialize(directory: String) -> IO(Result<&1, &1, Error, U32>):
295  initialize.paths(path.pair(directory))