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