1	import Base
2	import ../Manifest.bend as Manifest
3	import ../StorageBytes.bend as StorageBytes
4	import ../Fs.bend as Fs
5	import ../CrashPoint.bend as CrashPoint
6	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7	
8	# Represent ManifestReadState data used by the memtable flush.
9	type ManifestReadState is Type:
10	  ManifestNeed{file: File, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
11	  ManifestGot{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
12	
13	def write_tail.chmod(
14	  result: Result<&1, &1, U32 & String, Unit>,
15	  +path: String
16	) -> IO(Result<&1, &1, U32 & String, Unit>):
17	  match result:
18	    case Fail{error}:
19	      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
20	    case Done{Unit{}}:
21	      Fs.chmod(path, U32.from_nat(384n))
22	
23	def write_tail.fsync(
24	  result: Result<&1, &1, U32 & String, Unit>,
25	  +path: String
26	) -> IO(Result<&1, &1, U32 & String, Unit>):
27	  match result:
28	    case Fail{error}:
29	      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
30	    case Done{Unit{}}:
31	      do IO<Result<&1, &1, U32 & String, Unit>>:
32	        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path)
33	        write_tail.chmod(synced, path)
34	
35	# Writes the remaining bytes of a file buffer.
36	def write_tail(
37	  fr: File & Result<&1, &1, U32 & String, Unit>,
38	  +path: String
39	) -> IO(Result<&1, &1, U32 & String, Unit>):
40	  match fr:
41	    case (file, result):
42	      do IO<Result<&1, &1, U32 & String, Unit>>:
43	        _closed : Unit <- File.close(file)
44	        write_tail.fsync(result, path)
45	
46	def write_file.opened(
47	  opened: Result<&1, &1, U32 & String, File>,
48	  +path: String,
49	  +data: String
50	) -> IO(Result<&1, &1, U32 & String, Unit>):
51	  match opened:
52	    case Fail{error}:
53	      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
54	    case Done{file}:
55	      do IO<Result<&1, &1, U32 & String, Unit>>:
56	        written : File & Result<&1, &1, U32 & String, Unit> <- File.write(file, data)
57	        write_tail(written, path)
58	
59	# Creates and syncs one storage file.
60	def write_file(+path: String, +data: String) -> IO(Result<&1, &1, U32 & String, Unit>):
61	  do IO<Result<&1, &1, U32 & String, Unit>>:
62	    opened : Result<&1, &1, U32 & String, File> <- File.open(path, "w")
63	    write_file.opened(opened, path, data)
64	
65	def write_manifest.bytes.opened(
66	  opened: Result<&1, &1, U32 & String, File>,
67	  +path: String,
68	  bytes: Bytes.Bytes
69	) -> IO(Result<&1, &1, U32 & String, Unit>):
70	  match opened:
71	    case Fail{error}:
72	      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
73	    case Done{file}:
74	      do IO<Result<&1, &1, U32 & String, Unit>>:
75	        written : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(file, bytes)
76	        write_tail(written, path)
77	
78	def write_manifest.bytes(+path: String, bytes: Bytes.Bytes) -> IO(Result<&1, &1, U32 & String, Unit>):
79	  do IO<Result<&1, &1, U32 & String, Unit>>:
80	    opened : Result<&1, &1, U32 & String, File> <- File.open(path, "w")
81	    write_manifest.bytes.opened(opened, path, bytes)
82	
83	def write_manifest.encoded(
84	  path: String,
85	  encoded: Result<&1, &1, Manifest.Error, Bytes.Bytes>
86	) -> IO(Result<&1, &1, U32 & String, Unit>):
87	  match encoded:
88	    case Fail{_}:
89	      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{(U32.from_nat(1n), "cannot encode manifest")})
90	    case Done{bytes}:
91	      write_manifest.bytes(path, bytes)
92	
93	# Write manifest for the memtable flush.
94	def write_manifest(path: String, manifest: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Unit>):
95	  write_manifest.encoded(path, Manifest.serialize(manifest))
96	
97	# Read tail3 for the memtable flush.
98	def read_tail3(fr: (File & Result<&1, &1, U32 & String, String>)) -> IO(Result<&1, &1, U32 & String, String>):
99	  match fr:
100	    case (f3, r1):
101	      do IO<Result<&1, &1, U32 & String, String>>:
102	        _cls : Unit <- File.close(f3)
103	        IO.pure(Result<&1, &1, U32 & String, String>, r1)
104	
105	
106	# Base File.read decodes each returned chunk independently. A chunk boundary can
107	# split a multi-byte UTF-8 character, so table files are read in one bounded call
108	# Reads a bounded storage file and returns its bytes.
109	def read_file.opened(
110	  opened: Result<&1, &1, U32 & String, File>
111	) -> IO(Result<&1, &1, U32 & String, String>):
112	  match opened:
113	    case Fail{error}:
114	      IO.pure(Result<&1, &1, U32 & String, String>, Fail{error})
115	    case Done{file}:
116	      do IO<Result<&1, &1, U32 & String, String>>:
117	        result : File & Result<&1, &1, U32 & String, String> <- File.read(file, U32.from_nat(4294967295n))
118	        read_tail3(result)
119	
120	# Reads a bounded storage file and returns its bytes.
121	def read_file(+path: String) -> IO(Result<&1, &1, U32 & String, String>):
122	  do IO<Result<&1, &1, U32 & String, String>>:
123	    opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
124	    read_file.opened(opened)
125	
126	# Classifies the Manifest parse result before recovery.
127	def mfst_parsed(parsed: Manifest.ParseResult) -> Result<&1, &1, U32 & String, Manifest.Manifest>:
128	  match parsed:
129	    case Manifest.Parsed{manifest}:
130	      Done{manifest}
131	    case Manifest.UnsupportedVersion{_}:
132	      Fail{(Manifest.unsupported_version_error_code(), "unsupported manifest version")}
133	    case Manifest.Malformed{}:
134	      Fail{(Manifest.malformed_error_code(), "malformed manifest")}
135	    case Manifest.ParseLimitExceeded{}:
136	      Fail{(Manifest.parse_limit_error_code(), "manifest size limit exceeded")}
137	
138	def manifest.read.result(
139	  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
140	) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
141	  match pair:
142	    case (file, Fail{error}):
143	      do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
144	        _closed : Unit <- File.close(file)
145	        return Fail{error}
146	    case (file, Done{bytes}):
147	      do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
148	        _closed : Unit <- File.close(file)
149	        return mfst_parsed(Manifest.parse_result(bytes))
150	
151	def manifest.read.loop(fuel: Nat, state: ManifestReadState) -> IO(File & Result<&1, &1, U32 & String, Bytes.Bytes>):
152	  match fuel:
153	    case 0n:
154	      match state:
155	        case ManifestNeed{file, +remaining, chunks}:
156	          match remaining:
157	            case 0n:
158	              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Done{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks))}))
159	            case 1n+_:
160	              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(7n), "bounded manifest read exhausted")}))
161	        case ManifestGot{pair, _, _}:
162	          match pair:
163	            case (file, Fail{error}):
164	              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{error}))
165	            case (file, Done{_}):
166	              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(7n), "bounded manifest read exhausted")}))
167	    case 1n+rest:
168	      match state:
169	        case ManifestNeed{file, +remaining, chunks}:
170	          match remaining:
171	            case 0n:
172	              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Done{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks))}))
173	            case 1n+_:
174	              do IO<File & Result<&1, &1, U32 & String, Bytes.Bytes>>:
175	                pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <- Fs.read_bytes(file, U32.from_nat(remaining))
176	                manifest.read.loop(rest, ManifestGot{pair, remaining, chunks})
177	        case ManifestGot{pair, +remaining, chunks}:
178	          match pair:
179	            case (file, Fail{error}):
180	              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{error}))
181	            case (file, Done{Bytes.Bytes{0, _}}):
182	              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(8n), "truncated manifest file")}))
183	            case (file, Done{Bytes.Bytes{+len, buf}}):
184	              manifest.read.loop(rest, ManifestNeed{file, Nat.sub(remaining, U32.to_nat(len)), Con{Bytes.Bytes{len, buf}, chunks}})
185	
186	def manifest.read.size.run(file: File, +size: Nat) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
187	  do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
188	    pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <- manifest.read.loop(Nat.add(size, 1n), ManifestNeed{file, size, Nil{}})
189	    manifest.read.result(pair)
190	
191	def manifest.read.size.valid(
192	  file: File,
193	  size: Nat,
194	  within: Bool
195	) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
196	  match within:
197	    case False{}:
198	      do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
199	        _closed : Unit <- File.close(file)
200	        return Fail{(U32.from_nat(9n), "manifest too large")}
201	    case True{}:
202	      manifest.read.size.run(file, size)
203	
204	def manifest.read.size(
205	  file: File,
206	  result: Result<&1, &1, U32 & String, Nat>
207	) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
208	  match result:
209	    case Fail{error}:
210	      do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
211	        _closed : Unit <- File.close(file)
212	        return Fail{error}
213	    case Done{+size}:
214	      manifest.read.size.valid(file, size, Nat.is_le(size, U32.to_nat(StorageBytes.MAX_MANIFEST_BYTES())))
215	
216	# Handle mfst open error in the memtable flush.
217	def mfst_open_error(
218	  missing: Bool,
219	  +code: U32
220	) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
221	  match missing:
222	    case True{}:
223	      IO.pure(Result<&1, &1, U32 & String, Manifest.Manifest>, Done{Manifest.M{Nil{}}})
224	    case False{}:
225	      IO.pure(Result<&1, &1, U32 & String, Manifest.Manifest>, Fail{(code, "cannot open manifest")})
226	
227	# Handle mfst opened in the memtable flush.
228	def mfst_opened(
229	  path: String,
230	  result: Result<&1, &1, U32 & String, File>
231	) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
232	  match result:
233	    case Fail{(+code, _)}:
234	      mfst_open_error(U32.is_eq(code, 2), code)
235	    case Done{file}:
236	      do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
237	        size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path)
238	        manifest.read.size(file, size)
239	
240	# Handle load mfst in the memtable flush.
241	def load_mfst(+path: String) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
242	  do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
243	    opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
244	    mfst_opened(path, opened)
245	
246	# Handle dir tail in the memtable flush.
247	def dir_tail(res: Result<&1, &1, U32 & String, Nat>, +path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
248	  match res:
249	    case Done{n}:
250	      IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
251	    case Fail{e}:
252	      Fs.make_dir(path)
253	
254	# Handle ensure dir in the memtable flush.
255	def ensure_dir(+path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
256	  do IO<Result<&1, &1, U32 & String, Unit>>:
257	    r : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(path)
258	    dir_tail(r, path)
259	
260	# Handle ldir of in the database recovery effects.
261	def ldir_of(idx: Nat) -> Maybe<&2, String>:
262	  match idx:
263	    case 0n:
264	      Some{"l0"}
265	    case 1n+m:
266	      match m:
267	        case 0n:
268	          Some{"l1"}
269	        case 1n+p:
270	          match p:
271	            case 0n:
272	              Some{"l2"}
273	            case 1n+q:
274	              match q:
275	                case 0n:
276	                  Some{"l3"}
277	                case 1n+r:
278	                  None{}