1	import Base
2	import ./Fs.bend as Fs
3	import ./MemTable.bend as MemTable
4	import ./Sstable.bend as Sstable
5	import ./SstFile.bend as SstFile
6	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7	
8	# Returns the typed corruption code for SST codec failures.
9	def corruption_error_code() -> U32:
10	  4294967292
11	
12	# Map pure format errors to the storage IO error domain.
13	def 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.
29	type ExactResult is Type:
30	  ExactResult{bytes: Bytes.Bytes, complete: Bool}
31	
32	# Represent ExactState data used by the streaming SSTable filesystem effects.
33	type 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	
38	def 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	
46	def 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	
88	def 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.
92	type 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	
107	def 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	
114	def 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	
123	def 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	
134	def 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	
144	def 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	
159	def 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	
175	def read.block.header.tail(
176	  saved: Bytes.Bytes,
177	  tail: Maybe<&2, U32>
178	) -> Bytes.Bytes & Maybe<&2, U32>:
179	  (saved, tail)
180	
181	def 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	
188	def 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	
202	def 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	
212	def 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	
224	def 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	
238	def 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	
253	def 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	
263	def 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	
275	def 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	
288	def 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	
293	def 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	
300	def 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	
321	def 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	
365	def 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	
375	def 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	
389	def 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.
394	def 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.
405	def 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	
410	def 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	
420	def 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	
433	def 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.
443	type 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	
450	def 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	
495	def 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	
507	def 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.
521	def 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))