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