src/ByteSstIo.bend
mylsm-lsm-store@0.6.0.0 · 378 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/ByteSstFile.bend as ByteSstFile
3import mylsm-lsm-store@0.6.0.0/src/ByteMemTable.bend as ByteMemTable
4import mylsm-lsm-store@0.6.0.0/src/Fs.bend as Fs
5import mylsm-lsm-store@0.6.0.0/src/scan/ByteBlockReader.bend as ByteBlockReader
6import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7
8def path.clone.reverse(path: String, reversed: String) -> String:
9 match path:
10 case SNil{}: reversed
11 case SCon{character, rest}:
12 path.clone.reverse(rest, SCon{character, reversed})
13
14def path.clone(path: String) -> String:
15 path.clone.reverse(path.clone.reverse(path, SNil{}), SNil{})
16
17def nat.clone.pair.step(pair: Nat & Nat) -> Nat & Nat:
18 match pair:
19 case (left, right): (Nat.add(left, 1n), Nat.add(right, 1n))
20
21def nat.clone.pair(value: Nat) -> Nat & Nat:
22 match value:
23 case 0n: (0n, 0n)
24 case 1n+rest: nat.clone.pair.step(nat.clone.pair(rest))
25
26def path.clone.pair.scalar(
27 first: String,
28 second: String,
29 scalars: Nat & Nat
30) -> String & String:
31 match scalars:
32 case (left, right):
33 (SCon{Chr{U32.from_nat(left)}, first},
34 SCon{Chr{U32.from_nat(right)}, second})
35
36def path.clone.pair.build(
37 path: String,
38 copies: String & String
39) -> String & String:
40 match path:
41 case SNil{}:
42 match copies:
43 case (left, right): (path.clone.reverse(left, SNil{}), path.clone.reverse(right, SNil{}))
44 case SCon{Chr{+scalar}, rest}:
45 match copies:
46 case (left, right):
47 path.clone.pair.build(rest,
48 path.clone.pair.scalar(left, right,
49 nat.clone.pair(U32.to_nat(scalar))))
50
51def path.clone.pair(path: String) -> String & String:
52 path.clone.pair.build(path, (SNil{}, SNil{}))
53
54# Carries the reader or its latest step through blockwise recovery.
55type ReadState is Type:
56 ReadNeed{reader: ByteBlockReader.Reader, table: ByteMemTable.Table}
57 ReadStep{result: Result<&1, &1, U32 & String, ByteBlockReader.Step>,
58 table: ByteMemTable.Table}
59
60# Carries only bounded reader state and compact file metadata during validation.
61type ValidationState is Type:
62 ValidateNeed{reader: ByteBlockReader.Reader,
63 metadata: ByteBlockReader.Metadata}
64 ValidateStep{result: Result<&1, &1, U32 & String, ByteBlockReader.Step>,
65 metadata: ByteBlockReader.Metadata}
66 ValidatePage{reader: ByteBlockReader.Reader,
67 metadata: ByteBlockReader.Metadata, entries: List<&1, ByteMemTable.Entry>}
68
69# Classifies malformed or oversized byte SST files as corruption.
70def corruption(message: String) -> U32 & String:
71 (4294967292, message)
72
73def validate.page(
74 entries: List<&1, ByteMemTable.Entry>,
75 metadata: ByteBlockReader.Metadata
76) -> ByteBlockReader.Metadata:
77 match entries:
78 case Nil{}: metadata
79 case Con{ByteMemTable.Entry{Bytes.Bytes{+key_len, _}, val}, tail}:
80 match val:
81 case None{}: validate.page(tail, metadata)
82 case Some{Bytes.Bytes{+value_len, _}}:
83 match metadata:
84 case ByteBlockReader.Metadata{size, total, block_count, index_offset,
85 live_entries, payload_bytes}:
86 validate.page(tail, ByteBlockReader.Metadata{size, total, block_count,
87 index_offset, Nat.add(live_entries, 1n),
88 Nat.add(payload_bytes, Nat.add(U32.to_nat(key_len),
89 U32.to_nat(value_len)))})
90
91# Advances full-file validation without retaining decoded SST entries.
92def validate.loop(
93 fuel: Nat,
94 state: ValidationState
95) -> IO(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>):
96 match fuel:
97 case 0n:
98 match state:
99 case ValidateNeed{reader, _}:
100 do IO<Result<&1, &1, U32 & String, ByteBlockReader.Metadata>>:
101 _closed : Result<&1, &1, U32 & String, Unit> <-
102 ByteBlockReader.close(reader)
103 return Fail{corruption("byte SST validation exceeded its block budget")}
104 case ValidatePage{reader, _, _}:
105 do IO<Result<&1, &1, U32 & String, ByteBlockReader.Metadata>>:
106 _closed : Result<&1, &1, U32 & String, Unit> <-
107 ByteBlockReader.close(reader)
108 return Fail{corruption("byte SST validation exceeded its block budget")}
109 case ValidateStep{result, _}:
110 match result:
111 case Done{ByteBlockReader.Block{reader, _}}:
112 do IO<Result<&1, &1, U32 & String, ByteBlockReader.Metadata>>:
113 _closed : Result<&1, &1, U32 & String, Unit> <-
114 ByteBlockReader.close(reader)
115 return Fail{corruption("byte SST validation exceeded its block budget")}
116 case Fail{error}:
117 IO.pure(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>, Fail{error})
118 case _: IO.pure(
119 Result<&1, &1, U32 & String, ByteBlockReader.Metadata>,
120 Fail{corruption("byte SST validation exhausted its transition limit")})
121 case 1n+rest:
122 match state:
123 case ValidateNeed{reader, metadata}:
124 do IO<Result<&1, &1, U32 & String, ByteBlockReader.Metadata>>:
125 result : Result<&1, &1, U32 & String, ByteBlockReader.Step> <-
126 ByteBlockReader.next(reader)
127 validate.loop(rest, ValidateStep{result, metadata})
128 case ValidatePage{reader, metadata, entries}:
129 validate.loop(rest, ValidateNeed{reader, validate.page(entries, metadata)})
130 case ValidateStep{result, metadata}:
131 match result:
132 case Fail{error}:
133 IO.pure(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>, Fail{error})
134 case Done{ByteBlockReader.End{}}:
135 IO.pure(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>, Done{metadata})
136 case Done{ByteBlockReader.Block{reader, entries}}:
137 validate.loop(rest, ValidatePage{reader, metadata, entries})
138
139# Starts metadata-only validation with the reader's bounded transition budget.
140def validate.budget(
141 metadata: ByteBlockReader.Metadata,
142 pair: Nat & ByteBlockReader.Reader
143) -> IO(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>):
144 match pair:
145 case (fuel, reader):
146 validate.loop(fuel, ValidateNeed{reader, metadata})
147
148# Preserves metadata while deriving the reader budget from its parsed header.
149def validate.metadata(
150 pair: ByteBlockReader.Metadata & ByteBlockReader.Reader
151) -> IO(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>):
152 match pair:
153 case (metadata, reader):
154 validate.budget(metadata, ByteBlockReader.transition_budget(reader))
155
156# Extracts metadata before consuming the reader through the validation pass.
157def validate.opened(
158 opened: Result<&1, &1, U32 & String, ByteBlockReader.Reader>
159) -> IO(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>):
160 match opened:
161 case Fail{error}:
162 IO.pure(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>, Fail{error})
163 case Done{reader}:
164 validate.metadata(ByteBlockReader.metadata(reader))
165
166# Validates an SST incrementally and returns only authenticated metadata.
167def validate(+path: String) -> IO(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>):
168 do IO<Result<&1, &1, U32 & String, ByteBlockReader.Metadata>>:
169 opened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <-
170 ByteBlockReader.open(path)
171 validate.opened(opened)
172
173def read.fold(
174 entries: List<&1, ByteMemTable.Entry>,
175 table: ByteMemTable.Table
176) -> ByteMemTable.Table:
177 match entries:
178 case Nil{}: table
179 case Con{ByteMemTable.Entry{key, val}, tail}:
180 match val:
181 case None{}: read.fold(tail, ByteMemTable.del(table, key))
182 case Some{value}: read.fold(tail, ByteMemTable.put(table, key, value))
183
184def read.loop(fuel: Nat, state: ReadState) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
185 match fuel:
186 case 0n:
187 match state:
188 case ReadNeed{reader, _}:
189 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
190 _closed : Result<&1, &1, U32 & String, Unit> <- ByteBlockReader.close(reader)
191 return Fail{corruption("byte SST block count exceeded reader limit")}
192 case ReadStep{result, _}:
193 match result:
194 case Done{ByteBlockReader.Block{reader, _}}:
195 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
196 _closed : Result<&1, &1, U32 & String, Unit> <- ByteBlockReader.close(reader)
197 return Fail{corruption("byte SST block count exceeded reader limit")}
198 case _: IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>,
199 Fail{corruption("byte SST reader exhausted its transition limit")})
200 case 1n+rest:
201 match state:
202 case ReadNeed{reader, table}:
203 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
204 result : Result<&1, &1, U32 & String, ByteBlockReader.Step> <-
205 ByteBlockReader.next(reader)
206 read.loop(rest, ReadStep{result, table})
207 case ReadStep{result, table}:
208 match result:
209 case Fail{error}:
210 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{error})
211 case Done{ByteBlockReader.End{}}:
212 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Done{table})
213 case Done{ByteBlockReader.Block{reader, entries}}:
214 read.loop(rest, ReadNeed{reader, read.fold(entries, table)})
215
216def read.loop.start(pair: Nat & ByteBlockReader.Reader) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
217 match pair:
218 case (fuel, reader):
219 read.loop(fuel, ReadNeed{reader, ByteMemTable.empty()})
220
221def read.opened(
222 opened: Result<&1, &1, U32 & String, ByteBlockReader.Reader>
223) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
224 match opened:
225 case Fail{error}:
226 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{error})
227 case Done{reader}:
228 read.loop.start(ByteBlockReader.transition_budget(reader))
229
230# Syncs the parent directory after SST rename publication.
231def write.directory.synced(
232 +directory: String,
233 result: Result<&1, &1, U32 & String, Unit>
234) -> IO(Result<&1, &1, U32 & String, Unit>):
235 match result:
236 case Fail{error}:
237 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
238 case Done{Unit{}}:
239 Fs.fsync(directory)
240
241# Atomically renames the synced temporary SST and syncs its directory.
242def write.published(
243 +directory: String,
244 temporary: String,
245 final_path: String,
246 result: Result<&1, &1, U32 & String, Unit>
247) -> IO(Result<&1, &1, U32 & String, Unit>):
248 match result:
249 case Fail{error}:
250 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
251 case Done{Unit{}}:
252 do IO<Result<&1, &1, U32 & String, Unit>>:
253 renamed : Result<&1, &1, U32 & String, Unit> <-
254 Fs.rename(temporary, final_path)
255 write.directory.synced(directory, renamed)
256
257# Syncs the temporary SST contents before publication.
258def write.synced(
259 +directory: String,
260 temporaries: String & String,
261 final_path: String,
262 result: Result<&1, &1, U32 & String, Unit>
263) -> IO(Result<&1, &1, U32 & String, Unit>):
264 match temporaries:
265 case (for_sync, for_rename):
266 match result:
267 case Fail{error}:
268 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
269 case Done{Unit{}}:
270 do IO<Result<&1, &1, U32 & String, Unit>>:
271 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(for_sync)
272 write.published(directory, for_rename, final_path, synced)
273
274# Closes the completed temporary-file write before syncing it.
275def write.written(
276 +directory: String,
277 temporary: String,
278 final_path: String,
279 pair: File & Result<&1, &1, U32 & String, Unit>
280) -> IO(Result<&1, &1, U32 & String, Unit>):
281 match pair:
282 case (file, result):
283 do IO<Result<&1, &1, U32 & String, Unit>>:
284 _closed : Unit <- File.close(file)
285 write.synced(directory, path.clone.pair(temporary), final_path, result)
286
287# Writes the encoded SST bytes to the temporary file.
288def write.encoded(
289 +directory: String,
290 temporary: String,
291 final_path: String,
292 bytes: Bytes.Bytes,
293 opened: Result<&1, &1, U32 & String, File>
294) -> IO(Result<&1, &1, U32 & String, Unit>):
295 match opened:
296 case Fail{error}:
297 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
298 case Done{file}:
299 do IO<Result<&1, &1, U32 & String, Unit>>:
300 pair : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(file, bytes)
301 write.written(directory, temporary, final_path, pair)
302
303# Closes the temporary file when encoding rejects sorted input.
304def write.encoder.failed(
305 opened: Result<&1, &1, U32 & String, File>
306) -> IO(Result<&1, &1, U32 & String, Unit>):
307 match opened:
308 case Fail{error}:
309 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
310 case Done{file}:
311 do IO<Result<&1, &1, U32 & String, Unit>>:
312 _closed : Unit <- File.close(file)
313 return Fail{corruption("byte SST encoder rejected its sorted input")}
314
315# Dispatches a validated encoding result after opening the temporary file.
316def write.opened(
317 +directory: String,
318 temporary: String,
319 final_path: String,
320 encoded: Result<&1, &1, ByteSstFile.Error, Bytes.Bytes>,
321 opened: Result<&1, &1, U32 & String, File>
322) -> IO(Result<&1, &1, U32 & String, Unit>):
323 match encoded:
324 case Fail{_}:
325 write.encoder.failed(opened)
326 case Done{bytes}:
327 write.encoded(directory, temporary, final_path, bytes, opened)
328
329def path.temporary.final(copies: String & String) -> String & String:
330 match copies:
331 case (temporary_base, final_path):
332 (temporary_base ++ ".tmp", final_path)
333
334def write_atomic_at.opened(
335 +directory: String,
336 temporary_copies: String & String,
337 final_path: String,
338 entries: List<&1, ByteMemTable.Entry>
339) -> IO(Result<&1, &1, U32 & String, Unit>):
340 match temporary_copies:
341 case (temporary, retained_temporary):
342 do IO<Result<&1, &1, U32 & String, Unit>>:
343 opened : Result<&1, &1, U32 & String, File> <- File.open(temporary, "w")
344 write.opened(directory, retained_temporary, final_path,
345 ByteSstFile.encode_file(entries, 0), opened)
346
347def write_atomic_at.paths(
348 +directory: String,
349 paths: String & String,
350 entries: List<&1, ByteMemTable.Entry>
351) -> IO(Result<&1, &1, U32 & String, Unit>):
352 match paths:
353 case (temporary, final_path):
354 write_atomic_at.opened(directory, path.clone.pair(temporary),
355 final_path, entries)
356
357# Atomically publishes one bounded byte SST at a caller-selected path.
358def write_atomic_at(
359 +directory: String,
360 final_path: String,
361 entries: List<&1, ByteMemTable.Entry>
362) -> IO(Result<&1, &1, U32 & String, Unit>):
363 write_atomic_at.paths(directory,
364 path.temporary.final(path.clone.pair(final_path)), entries)
365
366# Encodes and atomically publishes one bounded byte SST snapshot.
367def write_atomic(
368 +directory: String,
369 entries: List<&1, ByteMemTable.Entry>
370) -> IO(Result<&1, &1, U32 & String, Unit>):
371 write_atomic_at(directory, directory ++ "/SST", entries)
372
373# Reads one decoded block at a time and returns only after footer validation.
374def read(+path: String) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
375 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
376 opened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <-
377 ByteBlockReader.open(path)
378 read.opened(opened)