src/SstStreamIo.bend
mylsm-lsm-store@0.5.0.0 · 528 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.5.0.0/src/Fs.bend as Fs
3import mylsm-lsm-store@0.5.0.0/src/MemTable.bend as MemTable
4import mylsm-lsm-store@0.5.0.0/src/Sstable.bend as Sstable
5import mylsm-lsm-store@0.5.0.0/src/SstFile.bend as SstFile
6import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7
8# Returns the typed corruption code for SST codec failures.
9def corruption_error_code() -> U32:
10 4294967292
11
12# Map pure format errors to the storage IO error domain.
13def codec_error(error: SstFile.Error) -> U32 & String:
14 match error:
15 case SstFile.InvalidLevel{}:
16 (corruption_error_code(), "invalid SST level")
17 case SstFile.InvalidOrder{}:
18 (corruption_error_code(), "SST keys are not strictly ordered")
19 case SstFile.TooLarge{}:
20 (corruption_error_code(), "SST exceeds a size limit")
21 case SstFile.Malformed{}:
22 (corruption_error_code(), "malformed v3 SST")
23 case SstFile.InvalidUtf8{}:
24 (corruption_error_code(), "invalid UTF-8 in v3 SST")
25 case SstFile.ChecksumMismatch{}:
26 (corruption_error_code(), "v3 SST checksum mismatch")
27
28# Represent ExactResult data used by the streaming SSTable filesystem effects.
29type ExactResult is Type:
30 ExactResult{bytes: Bytes.Bytes, complete: Bool}
31
32# Represent ExactState data used by the streaming SSTable filesystem effects.
33type ExactState is Type:
34 ExactNeed{file: File, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
35 ExactRead{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
36 remaining: Nat, chunks: List<&1, Bytes.Bytes>}
37
38def exact.complete(
39 file: File,
40 chunks: List<&1, Bytes.Bytes>,
41 complete: Bool
42) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
43 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
44 (file, Done{ExactResult{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks)), complete}}))
45
46def exact.loop(
47 fuel: Nat,
48 state: ExactState
49) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
50 match fuel:
51 case 0n:
52 match state:
53 case ExactNeed{file, remaining, chunks}:
54 match remaining:
55 case 0n:
56 exact.complete(file, chunks, True{})
57 case 1n+_:
58 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
59 (file, Fail{(U32.from_nat(7n), "bounded read exhausted")}))
60 case ExactRead{pair, _, _}:
61 match pair:
62 case (file, Fail{error}):
63 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{error}))
64 case (file, Done{_}):
65 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
66 (file, Fail{(U32.from_nat(7n), "bounded read exhausted")}))
67 case 1n+rest:
68 match state:
69 case ExactNeed{file, +remaining, chunks}:
70 match remaining:
71 case 0n:
72 exact.complete(file, chunks, True{})
73 case 1n+_:
74 do IO<File & Result<&1, &1, U32 & String, ExactResult>>:
75 next : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
76 Fs.read_bytes(file, U32.from_nat(Nat.min(1048576n, remaining)))
77 exact.loop(rest, ExactRead{next, remaining, chunks})
78 case ExactRead{pair, remaining, chunks}:
79 match pair:
80 case (file, Fail{error}):
81 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{error}))
82 case (file, Done{Bytes.Bytes{0, _}}):
83 exact.complete(file, chunks, False{})
84 case (file, Done{Bytes.Bytes{+len, buf}}):
85 exact.loop(rest, ExactNeed{file, (remaining - U32.to_nat(len) : Nat),
86 Con{Bytes.Bytes{len, buf}, chunks}})
87
88def exact.start(file: File, +needed: Nat) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
89 exact.loop(Nat.add(Nat.mul(2n, needed), 1n), ExactNeed{file, needed, Nil{}})
90
91# Represent StreamReadState data used by the streaming SSTable filesystem effects.
92type StreamReadState is Type:
93 ReadHeader{file: File, size: Nat}
94 ReadHeaderResult{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>, size: Nat}
95 ReadBlocks{file: File, size: Nat, used: Nat, parser: SstFile.StreamState}
96 ReadBlockHeaderResult{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
97 size: Nat, used: Nat, parser: SstFile.StreamState}
98 ReadBlockBody{file: File, size: Nat, used: Nat,
99 parser: SstFile.StreamState, header: Bytes.Bytes, tail_len: U32}
100 ReadBlockBodyResult{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
101 size: Nat, used: Nat, parser: SstFile.StreamState,
102 header: Bytes.Bytes, tail_len: U32}
103 ReadFooterResult{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
104 size: Nat, used: Nat, parser: SstFile.StreamState}
105 ReadClose{file: File, result: Result<&1, &1, U32 & String, Sstable.Table>}
106
107def read.exact.complete(bytes: Bytes.Bytes, complete: Bool) -> Result<&1, &1, U32 & String, Bytes.Bytes>:
108 match complete:
109 case True{}:
110 Done{bytes}
111 case False{}:
112 Fail{(corruption_error_code(), "truncated SST")}
113
114def read.exact.state(
115 pair: File & Result<&1, &1, U32 & String, ExactResult>
116) -> File & Result<&1, &1, U32 & String, Bytes.Bytes>:
117 match pair:
118 case (file, Fail{error}):
119 (file, Fail{error})
120 case (file, Done{ExactResult{bytes, complete}}):
121 (file, read.exact.complete(bytes, complete))
122
123def read.header.parsed(
124 file: File,
125 +size: Nat,
126 result: Result<&1, &1, SstFile.Error, SstFile.StreamState>
127) -> StreamReadState:
128 match result:
129 case Fail{error}:
130 ReadClose{file, Fail{codec_error(error)}}
131 case Done{parser}:
132 ReadBlocks{file, size, 20n, parser}
133
134def read.header.state(
135 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
136 +size: Nat
137) -> StreamReadState:
138 match pair:
139 case (file, Fail{error}):
140 ReadClose{file, Fail{error}}
141 case (file, Done{bytes}):
142 read.header.parsed(file, size, SstFile.stream.header(bytes))
143
144def read.block.header.bound(
145 file: File,
146 +size: Nat,
147 +used: Nat,
148 parser: SstFile.StreamState,
149 header: Bytes.Bytes,
150 +tail_len: U32,
151 valid: Bool
152) -> StreamReadState:
153 match valid:
154 case False{}:
155 ReadClose{file, Fail{(corruption_error_code(), "truncated SST block")}}
156 case True{}:
157 ReadBlockBody{file, size, used, parser, header, tail_len}
158
159def read.block.header.size(
160 file: File,
161 +size: Nat,
162 +used: Nat,
163 parser: SstFile.StreamState,
164 header: Bytes.Bytes,
165 tail: Maybe<&2, U32>
166) -> StreamReadState:
167 match tail:
168 case None{}:
169 ReadClose{file, Fail{(corruption_error_code(), "invalid SST block size")}}
170 case Some{+tail_len}:
171 read.block.header.bound(file, size, used, parser, header, tail_len,
172 Nat.is_le(Nat.add(Nat.add(used, 8n), U32.to_nat(tail_len)),
173 Nat.sub(size, 40n)))
174
175def read.block.header.tail(
176 saved: Bytes.Bytes,
177 tail: Maybe<&2, U32>
178) -> Bytes.Bytes & Maybe<&2, U32>:
179 (saved, tail)
180
181def read.block.header.inspected(
182 pair: Bytes.Bytes & Bytes.Bytes
183) -> Bytes.Bytes & Maybe<&2, U32>:
184 match pair:
185 case (saved, inspect):
186 read.block.header.tail(saved, SstFile.stream.block.tail_size(inspect))
187
188def read.block.header.copied(
189 file: File,
190 +size: Nat,
191 +used: Nat,
192 parser: SstFile.StreamState,
193 inspected: Bytes.Bytes & Maybe<&2, U32>
194) -> StreamReadState:
195 match inspected:
196 case (saved, None{}):
197 ReadClose{file, Fail{(corruption_error_code(), "invalid SST block size")}}
198 case (saved, Some{+tail_len}):
199 read.block.header.size(file, size, used, parser, saved,
200 Some{tail_len})
201
202def read.block.header.copy(
203 file: File,
204 +size: Nat,
205 +used: Nat,
206 parser: SstFile.StreamState,
207 header: Bytes.Bytes
208) -> StreamReadState:
209 read.block.header.copied(file, size, used, parser,
210 read.block.header.inspected(SstFile.bytes.clone.split(header)))
211
212def read.block.header.state(
213 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
214 +size: Nat,
215 +used: Nat,
216 parser: SstFile.StreamState
217) -> StreamReadState:
218 match pair:
219 case (file, Fail{error}):
220 ReadClose{file, Fail{error}}
221 case (file, Done{header}):
222 read.block.header.copy(file, size, used, parser, header)
223
224def read.block.body.parsed(
225 file: File,
226 +size: Nat,
227 +used: Nat,
228 +tail_len: U32,
229 result: Result<&1, &1, SstFile.Error, SstFile.StreamState>
230) -> StreamReadState:
231 match result:
232 case Fail{error}:
233 ReadClose{file, Fail{codec_error(error)}}
234 case Done{parser}:
235 ReadBlocks{file, size, Nat.add(used,
236 Nat.add(8n, U32.to_nat(tail_len))), parser}
237
238def read.block.body.state(
239 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
240 +size: Nat,
241 +used: Nat,
242 parser: SstFile.StreamState,
243 header: Bytes.Bytes,
244 +tail_len: U32
245) -> StreamReadState:
246 match pair:
247 case (file, Fail{error}):
248 ReadClose{file, Fail{error}}
249 case (file, Done{body}):
250 read.block.body.parsed(file, size, used, tail_len,
251 SstFile.stream.block(parser, Bytes.concat([header, body])))
252
253def read.footer.parsed(
254 file: File,
255 result: Result<&1, &1, SstFile.Error, Sstable.Table>
256) -> StreamReadState:
257 match result:
258 case Fail{error}:
259 ReadClose{file, Fail{codec_error(error)}}
260 case Done{table}:
261 ReadClose{file, Done{table}}
262
263def read.footer.size(
264 file: File,
265 parser: SstFile.StreamState,
266 bytes: Bytes.Bytes,
267 valid: Bool
268) -> StreamReadState:
269 match valid:
270 case False{}:
271 ReadClose{file, Fail{(corruption_error_code(), "malformed SST size")}}
272 case True{}:
273 read.footer.parsed(file, SstFile.stream.footer(parser, bytes))
274
275def read.footer.state(
276 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
277 +size: Nat,
278 +used: Nat,
279 parser: SstFile.StreamState
280) -> StreamReadState:
281 match pair:
282 case (file, Fail{error}):
283 ReadClose{file, Fail{error}}
284 case (file, Done{bytes}):
285 read.footer.size(file, parser, bytes,
286 Nat.is_eq(Nat.add(used, 40n), size))
287
288def read.exhausted.file(file: File) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
289 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
290 _closed : Unit <- File.close(file)
291 return Fail{(U32.from_nat(7n), "bounded SST read exhausted")}
292
293def read.exhausted.pair(
294 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
295) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
296 match pair:
297 case (file, _):
298 read.exhausted.file(file)
299
300def read.exhausted(state: StreamReadState) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
301 match state:
302 case ReadHeader{file, _}:
303 read.exhausted.file(file)
304 case ReadHeaderResult{pair, _}:
305 read.exhausted.pair(pair)
306 case ReadBlocks{file, _, _, _}:
307 read.exhausted.file(file)
308 case ReadBlockHeaderResult{pair, _, _, _}:
309 read.exhausted.pair(pair)
310 case ReadBlockBody{file, _, _, _, _, _}:
311 read.exhausted.file(file)
312 case ReadBlockBodyResult{pair, _, _, _, _, _}:
313 read.exhausted.pair(pair)
314 case ReadFooterResult{pair, _, _, _}:
315 read.exhausted.pair(pair)
316 case ReadClose{file, result}:
317 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
318 _closed : Unit <- File.close(file)
319 return result
320
321def read.loop(fuel: Nat, state: StreamReadState) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
322 match fuel:
323 case 0n:
324 read.exhausted(state)
325 case 1n+rest:
326 match state:
327 case ReadHeader{file, size}:
328 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
329 pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 20n)
330 read.loop(rest, read.header.state(read.exact.state(pair), size))
331 case ReadHeaderResult{pair, size}:
332 read.loop(rest, read.header.state(pair, size))
333 case ReadBlocks{file, size, used, parser}:
334 match parser:
335 case SstFile.StreamState{_, _, _, 0, _, _, _, _}:
336 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
337 pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 40n)
338 read.loop(rest, read.footer.state(read.exact.state(pair),
339 size, used, parser))
340 case SstFile.StreamState{+level, +total, +block_count, +remaining,
341 +entries_left, +previous, entries, hashes}:
342 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
343 pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 8n)
344 read.loop(rest, read.block.header.state(read.exact.state(pair),
345 size, used, SstFile.StreamState{level, total, block_count,
346 remaining, entries_left, previous, entries, hashes}))
347 case ReadBlockHeaderResult{pair, size, used, parser}:
348 read.loop(rest, read.block.header.state(pair, size, used, parser))
349 case ReadBlockBody{file, size, used, parser, header, +tail_len}:
350 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
351 pair : File & Result<&1, &1, U32 & String, ExactResult> <-
352 exact.start(file, U32.to_nat(tail_len))
353 read.loop(rest, ReadBlockBodyResult{read.exact.state(pair),
354 size, used, parser, header, tail_len})
355 case ReadBlockBodyResult{pair, size, used, parser, header, tail_len}:
356 read.loop(rest, read.block.body.state(pair, size, used, parser,
357 header, tail_len))
358 case ReadFooterResult{pair, size, used, parser}:
359 read.loop(rest, read.footer.state(pair, size, used, parser))
360 case ReadClose{file, result}:
361 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
362 _closed : Unit <- File.close(file)
363 return result
364
365def read.table.opened(
366 +size: Nat,
367 opened: Result<&1, &1, U32 & String, File>
368) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
369 match opened:
370 case Fail{error}:
371 IO.pure(Result<&1, &1, U32 & String, Sstable.Table>, Fail{error})
372 case Done{file}:
373 read.loop(Nat.add(Nat.mul(2n, size), 16n), ReadHeader{file, size})
374
375def read.table.size.checked(
376 +path: String,
377 +size: Nat,
378 valid: Bool
379) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
380 match valid:
381 case False{}:
382 IO.pure(Result<&1, &1, U32 & String, Sstable.Table>,
383 Fail{(corruption_error_code(), "malformed or oversized SST")})
384 case True{}:
385 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
386 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
387 read.table.opened(size, opened)
388
389def read.table.size(+path: String, +size: Nat) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
390 read.table.size.checked(path, size,
391 Nat.is_le(60n, size) && Nat.is_le(size, 4294967295n))
392
393# Read table for the streaming SSTable filesystem effects.
394def read.table.file_size(
395 +path: String,
396 result: Result<&1, &1, U32 & String, Nat>
397) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
398 match result:
399 case Fail{error}:
400 IO.pure(Result<&1, &1, U32 & String, Sstable.Table>, Fail{error})
401 case Done{size}:
402 read.table.size(path, size)
403
404# Reads and validates one SSTable while preserving IO errors.
405def read_table(+path: String) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
406 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
407 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path)
408 read.table.file_size(path, size)
409
410def write.finish.chmod(
411 result: Result<&1, &1, U32 & String, Unit>
412) -> IO(Result<&1, &1, U32 & String, Unit>):
413 match result:
414 case Fail{error}:
415 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
416 case Done{Unit{}}:
417 IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
418
419
420def write.finish.synced(
421 +path: String,
422 result: Result<&1, &1, U32 & String, Unit>
423) -> IO(Result<&1, &1, U32 & String, Unit>):
424 match result:
425 case Fail{error}:
426 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
427 case Done{Unit{}}:
428 do IO<Result<&1, &1, U32 & String, Unit>>:
429 changed : Result<&1, &1, U32 & String, Unit> <-
430 Fs.chmod(path, U32.from_nat(384n))
431 write.finish.chmod(changed)
432
433def write.finish(
434 file: File,
435 +path: String
436) -> IO(Result<&1, &1, U32 & String, Unit>):
437 do IO<Result<&1, &1, U32 & String, Unit>>:
438 _closed : Unit <- File.close(file)
439 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path)
440 write.finish.synced(path, synced)
441
442# Linear writer state keeps the file handle paired with the encoder result.
443type WriteState is Type:
444 WriteNext{file: File, encoder: SstFile.Encoder}
445 WriteOutput{file: File,
446 pair: SstFile.Encoder & Maybe<&1, Bytes.Bytes>}
447 WriteResult{pair: File & Result<&1, &1, U32 & String, Unit>,
448 encoder: SstFile.Encoder}
449
450def write.step(
451 fuel: Nat,
452 +path: String,
453 state: WriteState
454) -> IO(Result<&1, &1, U32 & String, Unit>):
455 match fuel:
456 case 0n:
457 match state:
458 case WriteNext{file, _}:
459 do IO<Result<&1, &1, U32 & String, Unit>>:
460 _closed : Unit <- File.close(file)
461 return Fail{(U32.from_nat(7n), "table exceeds packed write limit")}
462 case WriteOutput{file, _}:
463 do IO<Result<&1, &1, U32 & String, Unit>>:
464 _closed : Unit <- File.close(file)
465 return Fail{(U32.from_nat(7n), "table exceeds packed write limit")}
466 case WriteResult{pair, _}:
467 match pair:
468 case (file, _):
469 do IO<Result<&1, &1, U32 & String, Unit>>:
470 _closed : Unit <- File.close(file)
471 return Fail{(U32.from_nat(7n), "table exceeds packed write limit")}
472 case 1n+rest:
473 match state:
474 case WriteNext{file, encoder}:
475 write.step(rest, path,
476 WriteOutput{file, SstFile.next_chunk(encoder)})
477 case WriteOutput{file, pair}:
478 match pair:
479 case (_, None{}):
480 write.finish(file, path)
481 case (next, Some{bytes}):
482 do IO<Result<&1, &1, U32 & String, Unit>>:
483 written : File & Result<&1, &1, U32 & String, Unit> <-
484 Fs.write_bytes(file, bytes)
485 write.step(rest, path, WriteResult{written, next})
486 case WriteResult{pair, encoder}:
487 match pair:
488 case (file, Fail{error}):
489 do IO<Result<&1, &1, U32 & String, Unit>>:
490 _closed : Unit <- File.close(file)
491 return Fail{error}
492 case (file, Done{Unit{}}):
493 write.step(rest, path, WriteNext{file, encoder})
494
495def write.encoder.opened(
496 +path: String,
497 fuel: Nat,
498 encoder: SstFile.Encoder,
499 opened: Result<&1, &1, U32 & String, File>
500) -> IO(Result<&1, &1, U32 & String, Unit>):
501 match opened:
502 case Fail{error}:
503 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
504 case Done{file}:
505 write.step(fuel, path, WriteNext{file, encoder})
506
507def write.encoder(
508 +path: String,
509 fuel: Nat,
510 result: Result<&1, &1, SstFile.Error, SstFile.Encoder>
511) -> IO(Result<&1, &1, U32 & String, Unit>):
512 match result:
513 case Fail{error}:
514 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{codec_error(error)})
515 case Done{encoder}:
516 do IO<Result<&1, &1, U32 & String, Unit>>:
517 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "w")
518 write.encoder.opened(path, fuel, encoder, opened)
519
520# Encode and write one SST as packed chunks, then sync it before publication.
521def write_table(
522 +path: String,
523 +entries: List<&2, MemTable.Entry>,
524 +level: U32
525) -> IO(Result<&1, &1, U32 & String, Unit>):
526 write.encoder(path,
527 Nat.add(Nat.mul(3n, Nat.add(List.length(&2, MemTable.Entry, entries), 2n)), 2n),
528 SstFile.new_encoder(entries, level))