1	import Base
2	import ./ByteSstFile.bend as ByteSstFile
3	import ./ByteMemTable.bend as ByteMemTable
4	import ./Fs.bend as Fs
5	import ./scan/ByteBlockReader.bend as ByteBlockReader
6	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7	
8	def path.clone.reverse(path: String, reversed: String) -> String:
9	  match path:
10	    case SNil{}: reversed
11	    case SCon{character, rest}:
12	      path.clone.reverse(rest, SCon{character, reversed})
13	
14	def path.clone(path: String) -> String:
15	  path.clone.reverse(path.clone.reverse(path, SNil{}), SNil{})
16	
17	def nat.clone.pair.step(pair: Nat & Nat) -> Nat & Nat:
18	  match pair:
19	    case (left, right): (Nat.add(left, 1n), Nat.add(right, 1n))
20	
21	def nat.clone.pair(value: Nat) -> Nat & Nat:
22	  match value:
23	    case 0n: (0n, 0n)
24	    case 1n+rest: nat.clone.pair.step(nat.clone.pair(rest))
25	
26	def path.clone.pair.scalar(
27	  first: String,
28	  second: String,
29	  scalars: Nat & Nat
30	) -> String & String:
31	  match scalars:
32	    case (left, right):
33	      (SCon{Chr{U32.from_nat(left)}, first},
34	        SCon{Chr{U32.from_nat(right)}, second})
35	
36	def path.clone.pair.build(
37	  path: String,
38	  copies: String & String
39	) -> String & String:
40	  match path:
41	    case SNil{}:
42	      match copies:
43	        case (left, right): (path.clone.reverse(left, SNil{}), path.clone.reverse(right, SNil{}))
44	    case SCon{Chr{+scalar}, rest}:
45	      match copies:
46	        case (left, right):
47	          path.clone.pair.build(rest,
48	            path.clone.pair.scalar(left, right,
49	              nat.clone.pair(U32.to_nat(scalar))))
50	
51	def path.clone.pair(path: String) -> String & String:
52	  path.clone.pair.build(path, (SNil{}, SNil{}))
53	
54	# Carries the reader or its latest step through blockwise recovery.
55	type ReadState is Type:
56	  ReadNeed{reader: ByteBlockReader.Reader, table: ByteMemTable.Table}
57	  ReadStep{result: Result<&1, &1, U32 & String, ByteBlockReader.Step>,
58	    table: ByteMemTable.Table}
59	
60	# Carries only bounded reader state and compact file metadata during validation.
61	type ValidationState is Type:
62	  ValidateNeed{reader: ByteBlockReader.Reader,
63	    metadata: ByteBlockReader.Metadata}
64	  ValidateStep{result: Result<&1, &1, U32 & String, ByteBlockReader.Step>,
65	    metadata: ByteBlockReader.Metadata}
66	  ValidatePage{reader: ByteBlockReader.Reader,
67	    metadata: ByteBlockReader.Metadata, entries: List<&1, ByteMemTable.Entry>}
68	
69	# Classifies malformed or oversized byte SST files as corruption.
70	def corruption(message: String) -> U32 & String:
71	  (4294967292, message)
72	
73	def validate.page(
74	  entries: List<&1, ByteMemTable.Entry>,
75	  metadata: ByteBlockReader.Metadata
76	) -> ByteBlockReader.Metadata:
77	  match entries:
78	    case Nil{}: metadata
79	    case Con{ByteMemTable.Entry{Bytes.Bytes{+key_len, _}, val}, tail}:
80	      match val:
81	        case None{}: validate.page(tail, metadata)
82	        case Some{Bytes.Bytes{+value_len, _}}:
83	          match metadata:
84	            case ByteBlockReader.Metadata{size, total, block_count, index_offset,
85	                live_entries, payload_bytes}:
86	              validate.page(tail, ByteBlockReader.Metadata{size, total, block_count,
87	                index_offset, Nat.add(live_entries, 1n),
88	                Nat.add(payload_bytes, Nat.add(U32.to_nat(key_len),
89	                  U32.to_nat(value_len)))})
90	
91	# Advances full-file validation without retaining decoded SST entries.
92	def validate.loop(
93	  fuel: Nat,
94	  state: ValidationState
95	) -> IO(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>):
96	  match fuel:
97	    case 0n:
98	      match state:
99	        case ValidateNeed{reader, _}:
100	          do IO<Result<&1, &1, U32 & String, ByteBlockReader.Metadata>>:
101	            _closed : Result<&1, &1, U32 & String, Unit> <-
102	              ByteBlockReader.close(reader)
103	            return Fail{corruption("byte SST validation exceeded its block budget")}
104	        case ValidatePage{reader, _, _}:
105	          do IO<Result<&1, &1, U32 & String, ByteBlockReader.Metadata>>:
106	            _closed : Result<&1, &1, U32 & String, Unit> <-
107	              ByteBlockReader.close(reader)
108	            return Fail{corruption("byte SST validation exceeded its block budget")}
109	        case ValidateStep{result, _}:
110	          match result:
111	            case Done{ByteBlockReader.Block{reader, _}}:
112	              do IO<Result<&1, &1, U32 & String, ByteBlockReader.Metadata>>:
113	                _closed : Result<&1, &1, U32 & String, Unit> <-
114	                  ByteBlockReader.close(reader)
115	                return Fail{corruption("byte SST validation exceeded its block budget")}
116	            case Fail{error}:
117	              IO.pure(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>, Fail{error})
118	            case _: IO.pure(
119	              Result<&1, &1, U32 & String, ByteBlockReader.Metadata>,
120	              Fail{corruption("byte SST validation exhausted its transition limit")})
121	    case 1n+rest:
122	      match state:
123	        case ValidateNeed{reader, metadata}:
124	          do IO<Result<&1, &1, U32 & String, ByteBlockReader.Metadata>>:
125	            result : Result<&1, &1, U32 & String, ByteBlockReader.Step> <-
126	              ByteBlockReader.next(reader)
127	            validate.loop(rest, ValidateStep{result, metadata})
128	        case ValidatePage{reader, metadata, entries}:
129	          validate.loop(rest, ValidateNeed{reader, validate.page(entries, metadata)})
130	        case ValidateStep{result, metadata}:
131	          match result:
132	            case Fail{error}:
133	              IO.pure(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>, Fail{error})
134	            case Done{ByteBlockReader.End{}}:
135	              IO.pure(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>, Done{metadata})
136	            case Done{ByteBlockReader.Block{reader, entries}}:
137	              validate.loop(rest, ValidatePage{reader, metadata, entries})
138	
139	# Starts metadata-only validation with the reader's bounded transition budget.
140	def validate.budget(
141	  metadata: ByteBlockReader.Metadata,
142	  pair: Nat & ByteBlockReader.Reader
143	) -> IO(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>):
144	  match pair:
145	    case (fuel, reader):
146	      validate.loop(fuel, ValidateNeed{reader, metadata})
147	
148	# Preserves metadata while deriving the reader budget from its parsed header.
149	def validate.metadata(
150	  pair: ByteBlockReader.Metadata & ByteBlockReader.Reader
151	) -> IO(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>):
152	  match pair:
153	    case (metadata, reader):
154	      validate.budget(metadata, ByteBlockReader.transition_budget(reader))
155	
156	# Extracts metadata before consuming the reader through the validation pass.
157	def validate.opened(
158	  opened: Result<&1, &1, U32 & String, ByteBlockReader.Reader>
159	) -> IO(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>):
160	  match opened:
161	    case Fail{error}:
162	      IO.pure(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>, Fail{error})
163	    case Done{reader}:
164	      validate.metadata(ByteBlockReader.metadata(reader))
165	
166	# Validates an SST incrementally and returns only authenticated metadata.
167	def validate(+path: String) -> IO(Result<&1, &1, U32 & String, ByteBlockReader.Metadata>):
168	  do IO<Result<&1, &1, U32 & String, ByteBlockReader.Metadata>>:
169	    opened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <-
170	      ByteBlockReader.open(path)
171	    validate.opened(opened)
172	
173	def read.fold(
174	  entries: List<&1, ByteMemTable.Entry>,
175	  table: ByteMemTable.Table
176	) -> ByteMemTable.Table:
177	  match entries:
178	    case Nil{}: table
179	    case Con{ByteMemTable.Entry{key, val}, tail}:
180	      match val:
181	        case None{}: read.fold(tail, ByteMemTable.del(table, key))
182	        case Some{value}: read.fold(tail, ByteMemTable.put(table, key, value))
183	
184	def read.loop(fuel: Nat, state: ReadState) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
185	  match fuel:
186	    case 0n:
187	      match state:
188	        case ReadNeed{reader, _}:
189	          do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
190	            _closed : Result<&1, &1, U32 & String, Unit> <- ByteBlockReader.close(reader)
191	            return Fail{corruption("byte SST block count exceeded reader limit")}
192	        case ReadStep{result, _}:
193	          match result:
194	            case Done{ByteBlockReader.Block{reader, _}}:
195	              do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
196	                _closed : Result<&1, &1, U32 & String, Unit> <- ByteBlockReader.close(reader)
197	                return Fail{corruption("byte SST block count exceeded reader limit")}
198	            case _: IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>,
199	              Fail{corruption("byte SST reader exhausted its transition limit")})
200	    case 1n+rest:
201	      match state:
202	        case ReadNeed{reader, table}:
203	          do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
204	            result : Result<&1, &1, U32 & String, ByteBlockReader.Step> <-
205	              ByteBlockReader.next(reader)
206	            read.loop(rest, ReadStep{result, table})
207	        case ReadStep{result, table}:
208	          match result:
209	            case Fail{error}:
210	              IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{error})
211	            case Done{ByteBlockReader.End{}}:
212	              IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Done{table})
213	            case Done{ByteBlockReader.Block{reader, entries}}:
214	              read.loop(rest, ReadNeed{reader, read.fold(entries, table)})
215	
216	def read.loop.start(pair: Nat & ByteBlockReader.Reader) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
217	  match pair:
218	    case (fuel, reader):
219	      read.loop(fuel, ReadNeed{reader, ByteMemTable.empty()})
220	
221	def read.opened(
222	  opened: Result<&1, &1, U32 & String, ByteBlockReader.Reader>
223	) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
224	  match opened:
225	    case Fail{error}:
226	      IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{error})
227	    case Done{reader}:
228	      read.loop.start(ByteBlockReader.transition_budget(reader))
229	
230	# Syncs the parent directory after SST rename publication.
231	def write.directory.synced(
232	  +directory: String,
233	  result: Result<&1, &1, U32 & String, Unit>
234	) -> IO(Result<&1, &1, U32 & String, Unit>):
235	  match result:
236	    case Fail{error}:
237	      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
238	    case Done{Unit{}}:
239	      Fs.fsync(directory)
240	
241	# Atomically renames the synced temporary SST and syncs its directory.
242	def write.published(
243	  +directory: String,
244	  temporary: String,
245	  final_path: String,
246	  result: Result<&1, &1, U32 & String, Unit>
247	) -> IO(Result<&1, &1, U32 & String, Unit>):
248	  match result:
249	    case Fail{error}:
250	      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
251	    case Done{Unit{}}:
252	      do IO<Result<&1, &1, U32 & String, Unit>>:
253	        renamed : Result<&1, &1, U32 & String, Unit> <-
254	          Fs.rename(temporary, final_path)
255	        write.directory.synced(directory, renamed)
256	
257	# Syncs the temporary SST contents before publication.
258	def write.synced(
259	  +directory: String,
260	  temporaries: String & String,
261	  final_path: String,
262	  result: Result<&1, &1, U32 & String, Unit>
263	) -> IO(Result<&1, &1, U32 & String, Unit>):
264	  match temporaries:
265	    case (for_sync, for_rename):
266	      match result:
267	        case Fail{error}:
268	          IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
269	        case Done{Unit{}}:
270	          do IO<Result<&1, &1, U32 & String, Unit>>:
271	            synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(for_sync)
272	            write.published(directory, for_rename, final_path, synced)
273	
274	# Closes the completed temporary-file write before syncing it.
275	def write.written(
276	  +directory: String,
277	  temporary: String,
278	  final_path: String,
279	  pair: File & Result<&1, &1, U32 & String, Unit>
280	) -> IO(Result<&1, &1, U32 & String, Unit>):
281	  match pair:
282	    case (file, result):
283	      do IO<Result<&1, &1, U32 & String, Unit>>:
284	        _closed : Unit <- File.close(file)
285	        write.synced(directory, path.clone.pair(temporary), final_path, result)
286	
287	# Writes the encoded SST bytes to the temporary file.
288	def write.encoded(
289	  +directory: String,
290	  temporary: String,
291	  final_path: String,
292	  bytes: Bytes.Bytes,
293	  opened: Result<&1, &1, U32 & String, File>
294	) -> IO(Result<&1, &1, U32 & String, Unit>):
295	  match opened:
296	    case Fail{error}:
297	      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
298	    case Done{file}:
299	      do IO<Result<&1, &1, U32 & String, Unit>>:
300	        pair : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(file, bytes)
301	        write.written(directory, temporary, final_path, pair)
302	
303	# Closes the temporary file when encoding rejects sorted input.
304	def write.encoder.failed(
305	  opened: Result<&1, &1, U32 & String, File>
306	) -> IO(Result<&1, &1, U32 & String, Unit>):
307	  match opened:
308	    case Fail{error}:
309	      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
310	    case Done{file}:
311	      do IO<Result<&1, &1, U32 & String, Unit>>:
312	        _closed : Unit <- File.close(file)
313	        return Fail{corruption("byte SST encoder rejected its sorted input")}
314	
315	# Dispatches a validated encoding result after opening the temporary file.
316	def write.opened(
317	  +directory: String,
318	  temporary: String,
319	  final_path: String,
320	  encoded: Result<&1, &1, ByteSstFile.Error, Bytes.Bytes>,
321	  opened: Result<&1, &1, U32 & String, File>
322	) -> IO(Result<&1, &1, U32 & String, Unit>):
323	  match encoded:
324	    case Fail{_}:
325	      write.encoder.failed(opened)
326	    case Done{bytes}:
327	      write.encoded(directory, temporary, final_path, bytes, opened)
328	
329	def path.temporary.final(copies: String & String) -> String & String:
330	  match copies:
331	    case (temporary_base, final_path):
332	      (temporary_base ++ ".tmp", final_path)
333	
334	def write_atomic_at.opened(
335	  +directory: String,
336	  temporary_copies: String & String,
337	  final_path: String,
338	  entries: List<&1, ByteMemTable.Entry>
339	) -> IO(Result<&1, &1, U32 & String, Unit>):
340	  match temporary_copies:
341	    case (temporary, retained_temporary):
342	      do IO<Result<&1, &1, U32 & String, Unit>>:
343	        opened : Result<&1, &1, U32 & String, File> <- File.open(temporary, "w")
344	        write.opened(directory, retained_temporary, final_path,
345	          ByteSstFile.encode_file(entries, 0), opened)
346	
347	def write_atomic_at.paths(
348	  +directory: String,
349	  paths: String & String,
350	  entries: List<&1, ByteMemTable.Entry>
351	) -> IO(Result<&1, &1, U32 & String, Unit>):
352	  match paths:
353	    case (temporary, final_path):
354	      write_atomic_at.opened(directory, path.clone.pair(temporary),
355	        final_path, entries)
356	
357	# Atomically publishes one bounded byte SST at a caller-selected path.
358	def write_atomic_at(
359	  +directory: String,
360	  final_path: String,
361	  entries: List<&1, ByteMemTable.Entry>
362	) -> IO(Result<&1, &1, U32 & String, Unit>):
363	  write_atomic_at.paths(directory,
364	    path.temporary.final(path.clone.pair(final_path)), entries)
365	
366	# Encodes and atomically publishes one bounded byte SST snapshot.
367	def write_atomic(
368	  +directory: String,
369	  entries: List<&1, ByteMemTable.Entry>
370	) -> IO(Result<&1, &1, U32 & String, Unit>):
371	  write_atomic_at(directory, directory ++ "/SST", entries)
372	
373	# Reads one decoded block at a time and returns only after footer validation.
374	def read(+path: String) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
375	  do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
376	    opened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <-
377	      ByteBlockReader.open(path)
378	    read.opened(opened)