src/ByteSstIo.bend

mylsm-lsm-store@0.6.0.1 · 378 lines · 33 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/ByteSstFile.bend as ByteSstFile
3import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
4import mylsm-lsm-store@0.6.0.1/src/Fs.bend as Fs
5import mylsm-lsm-store@0.6.0.1/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)