1	import Base
2	import ./hub_sha/sha256.bend as SHA
3	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
4	
5	# Represents a legacy or generation-addressed byte-store table layout.
6	type ManifestValue is Data:
7	  Empty{}
8	  HasSst{}
9	  GenerationEmpty{}
10	  GenerationSst{file_id: U32}
11	
12	# Distinguishes malformed bytes, unknown formats, and comparator mismatch.
13	type Error is Data:
14	  Malformed{}
15	  UnsupportedVersion{generation: U32}
16	  ComparatorMismatch{generation: U32}
17	  ChecksumMismatch{}
18	
19	# Reports the next generation identifier or exhaustion without wraparound.
20	type FileIdDecision is Data:
21	  FileIdAvailable{id: U32}
22	  FileIdExhausted{}
23	
24	# Unwraps a fixed manifest-format literal.
25	def literal(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
26	  match value:
27	    case Some{bytes}: bytes
28	    case None{}: Bytes.new(0)
29	
30	# Returns the byte-manifest family identifier shared by generations four and five.
31	def magic() -> Bytes.Bytes:
32	  literal(Bytes.from_hex("4d594c534d344d00"))
33	
34	# Encodes one unsigned big-endian field.
35	def u32.bytes(value: U32) -> Bytes.Bytes:
36	  Bytes.set.u32be(Bytes.new(4), 0, value)
37	
38	# Returns the exact fixed file name referenced by the single-run manifest.
39	def sst.name() -> Bytes.Bytes:
40	  literal(Bytes.from_hex("535354"))
41	
42	# Names the SST file of one generation.
43	def sst.generation.name(file_id: U32) -> String:
44	  "SST." ++ Bytes.to_hex(u32.bytes(file_id))
45	
46	def sst.generation.path(directory: String, file_id: U32) -> String:
47	  directory ++ "/" ++ sst.generation.name(file_id)
48	
49	# Creates two owned copies for hashing and output.
50	def clone.pair(+len: U32, copies: Array<U32> & Array<U32>) -> Bytes.Bytes & Bytes.Bytes:
51	  match copies:
52	    case (left, right): (Bytes.Bytes{len, left}, Bytes.Bytes{len, right})
53	
54	# Clones a packed buffer for the manifest checksum.
55	def clone.split(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes:
56	  match bytes:
57	    case Bytes.Bytes{+len, buf}:
58	      clone.pair(len, Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
59	
60	# Packs the three independently owned manifest copies.
61	def clone.triple.parts(
62	  +len: U32,
63	  first: Array<U32>,
64	  second: Array<U32>,
65	  third: Array<U32>
66	) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
67	  (Bytes.Bytes{len, first}, Bytes.Bytes{len, second}, Bytes.Bytes{len, third})
68	
69	# Selects the retained copy from the final byte duplication.
70	def clone.triple.finish(
71	  +len: U32,
72	  second: Array<U32>,
73	  copies: Array<U32> & Array<U32>
74	) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
75	  match copies:
76	    case (first, third): clone.triple.parts(len, first, second, third)
77	
78	# Adds an independently copied third buffer.
79	def clone.triple.first(
80	  +len: U32,
81	  copies: Array<U32> & Array<U32>
82	) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
83	  match copies:
84	    case (first, second):
85	      clone.triple.finish(len, second,
86	        Bytes.copy.bytes(len, first, Bytes.alloc(len), 0, 0))
87	
88	# Creates three owned copies for independent header-field checks.
89	def clone.triple(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
90	  match bytes:
91	    case Bytes.Bytes{+len, buf}:
92	      clone.triple.first(len,
93	        Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
94	
95	# Builds the checksummed body for one manifest value.
96	def body(value: ManifestValue) -> Bytes.Bytes:
97	  match value:
98	    case Empty{}:
99	      Bytes.concat([magic(), u32.bytes(4), u32.bytes(1), u32.bytes(1), u32.bytes(0)])
100	    case HasSst{}:
101	      Bytes.concat([magic(), u32.bytes(4), u32.bytes(1), u32.bytes(1),
102	        u32.bytes(1), u32.bytes(3), sst.name()])
103	    case GenerationEmpty{}:
104	      Bytes.concat([magic(), u32.bytes(5), u32.bytes(1), u32.bytes(1),
105	        u32.bytes(0)])
106	    case GenerationSst{file_id}:
107	      Bytes.concat([magic(), u32.bytes(5), u32.bytes(1), u32.bytes(1),
108	        u32.bytes(1), u32.bytes(file_id)])
109	
110	# Appends the digest of all preceding manifest bytes.
111	def encode.body.copies(copies: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
112	  match copies:
113	    case (encoded, for_hash):
114	      Bytes.concat([encoded, SHA.sha256_packed_bytes(for_hash)])
115	
116	# Appends the digest of all preceding manifest bytes.
117	def encode.body(packed: Bytes.Bytes) -> Bytes.Bytes:
118	  encode.body.copies(clone.split(packed))
119	
120	# Encodes the currently supported comparator and table layout.
121	def encode(value: ManifestValue) -> Bytes.Bytes:
122	  encode.body(body(value))
123	
124	def u32.read.value(pair: Bytes.Bytes & Maybe<&2, U32>) -> U32:
125	  match pair:
126	    case (_, Some{value}): value
127	    case _: 0
128	
129	def u32.read(input: Bytes.Bytes) -> U32:
130	  u32.read.value(Bytes.get.u32be(input, 0))
131	
132	def u32.clone.pair.bytes(copies: Bytes.Bytes & Bytes.Bytes) -> U32 & U32:
133	  match copies:
134	    case (left, right):
135	      (u32.read(left), u32.read(right))
136	
137	def clone.file.id(file_id: U32) -> U32 & U32:
138	  u32.clone.pair.bytes(clone.split(u32.bytes(file_id)))
139	
140	def next.file.id.current.result(next_id: U32, is_max: Bool) -> FileIdDecision:
141	  match is_max:
142	    case True{}: FileIdExhausted{}
143	    case False{}: FileIdAvailable{U32.add(next_id, 1)}
144	
145	def next.file.id.current.copies(copies: U32 & U32) -> FileIdDecision:
146	  match copies:
147	    case (next_id, max_check):
148	      next.file.id.current.result(next_id,
149	        U32.is_eq(max_check, u32.read(literal(Bytes.from_hex("ffffffff")))))
150	
151	def next.file.id.current(file_id: U32) -> FileIdDecision:
152	  next.file.id.current.copies(
153	    u32.clone.pair.bytes(clone.split(u32.bytes(file_id))))
154	
155	def next.file.id(value: ManifestValue) -> FileIdDecision:
156	  match value:
157	    case Empty{}: FileIdAvailable{1}
158	    case HasSst{}: FileIdAvailable{1}
159	    case GenerationEmpty{}: FileIdAvailable{1}
160	    case GenerationSst{file_id}: next.file.id.current(file_id)
161	
162	# Returns the second packed buffer from a slice operation.
163	def slice.result(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
164	  match pair:
165	    case (_, sliced): sliced
166	
167	# Compares two packed values while preserving the equality result.
168	def equal(pair: Bytes.Bytes & Bytes.Bytes & Bool) -> Bool:
169	  match pair:
170	    case (_, _, same): same
171	
172	# Returns the decoded value only when its body checksum is valid.
173	def decode.checksum.compare(
174	  value: ManifestValue,
175	  valid: Bool
176	) -> Result<&1, &1, Error, ManifestValue>:
177	  match valid:
178	    case True{}: Done{value}
179	    case False{}: Fail{ChecksumMismatch{}}
180	
181	# Hashes one owned copy and compares the other with the stored digest.
182	def decode.checksum.copies(
183	  value: ManifestValue,
184	  +offset: U32,
185	  copies: Bytes.Bytes & Bytes.Bytes
186	) -> Result<&1, &1, Error, ManifestValue>:
187	  match copies:
188	    case (for_hash, for_digest):
189	      decode.checksum.compare(value,
190	        equal(Bytes.eq(SHA.sha256_packed_bytes(slice.result(
191	          Bytes.slice(for_hash, 0, offset))),
192	          slice.result(Bytes.slice(for_digest, offset, 32)))))
193	
194	# Rejects a truncated or extended manifest before hashing it.
195	def decode.checksum.size(
196	  input: Bytes.Bytes,
197	  value: ManifestValue,
198	  +offset: U32,
199	  exact: Bool
200	) -> Result<&1, &1, Error, ManifestValue>:
201	  match exact:
202	    case False{}: Fail{Malformed{}}
203	    case True{}:
204	      decode.checksum.copies(value, offset, clone.split(input))
205	
206	# Compares the measured file size with its selected canonical layout.
207	def decode.checksum.measured(
208	  value: ManifestValue,
209	  +offset: U32,
210	  +expected_size: U32,
211	  measured: Bytes.Bytes & U32
212	) -> Result<&1, &1, Error, ManifestValue>:
213	  match measured:
214	    case (input, size):
215	      decode.checksum.size(input, value, offset, U32.is_eq(size, expected_size))
216	
217	# Checks exact size and the SHA-256 checksum over the canonical body.
218	def decode.checksum(
219	  input: Bytes.Bytes,
220	  value: ManifestValue,
221	  +offset: U32,
222	  +expected_size: U32
223	) -> Result<&1, &1, Error, ManifestValue>:
224	  decode.checksum.measured(value, offset, expected_size, Bytes.length(input))
225	
226	# Checks the fixed table name and its exact encoded length.
227	def decode.sst.name.valid(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, ManifestValue>:
228	  match valid:
229	    case False{}: Fail{Malformed{}}
230	    case True{}: decode.checksum(input, HasSst{}, 31, 63)
231	
232	# Checks the fixed SST name using independent packed-byte copies.
233	def decode.sst_name.copies(
234	  copies: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes
235	) -> Result<&1, &1, Error, ManifestValue>:
236	  match copies:
237	    case (input, length_bytes, name_bytes):
238	      decode.sst.name.valid(input,
239	        equal(Bytes.eq(slice.result(Bytes.slice(length_bytes, 24, 4)), u32.bytes(3))) &&
240	          equal(Bytes.eq(slice.result(Bytes.slice(name_bytes, 28, 3)), sst.name())))
241	
242	# Validates the only permitted SST file name.
243	def decode.sst_name(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
244	  decode.sst_name.copies(clone.triple(input))
245	
246	# Selects the empty or one-table layout from its declared file count.
247	def decode.sst_count.value(
248	  input: Bytes.Bytes,
249	  field: Maybe<&2, U32>
250	) -> Result<&1, &1, Error, ManifestValue>:
251	  match field:
252	    case None{}: Fail{Malformed{}}
253	    case Some{0}: decode.checksum(input, Empty{}, 24, 56)
254	    case Some{1}: decode.sst_name(input)
255	    case _: Fail{Malformed{}}
256	
257	def decode.sst_count.field(
258	  pair: Bytes.Bytes & Maybe<&2, U32>
259	) -> Result<&1, &1, Error, ManifestValue>:
260	  match pair:
261	    case (input, field): decode.sst_count.value(input, field)
262	
263	def decode.sst_count(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
264	  decode.sst_count.field(Bytes.get.u32be(input, 20))
265	
266	# Requires the canonical single level used by the byte durable engine.
267	def decode.level_count.value(
268	  input: Bytes.Bytes,
269	  field: Maybe<&2, U32>
270	) -> Result<&1, &1, Error, ManifestValue>:
271	  match field:
272	    case None{}: Fail{Malformed{}}
273	    case Some{1}: decode.sst_count(input)
274	    case _: Fail{Malformed{}}
275	
276	def decode.level_count.field(
277	  pair: Bytes.Bytes & Maybe<&2, U32>
278	) -> Result<&1, &1, Error, ManifestValue>:
279	  match pair:
280	    case (input, field): decode.level_count.value(input, field)
281	
282	def decode.level_count(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
283	  decode.level_count.field(Bytes.get.u32be(input, 16))
284	
285	# Rejects manifests written for a different key comparator.
286	def decode.comparator.value(
287	  input: Bytes.Bytes,
288	  field: Maybe<&2, U32>
289	) -> Result<&1, &1, Error, ManifestValue>:
290	  match field:
291	    case None{}: Fail{Malformed{}}
292	    case Some{1}: decode.level_count(input)
293	    case Some{generation}: Fail{ComparatorMismatch{generation}}
294	
295	def decode.comparator.field(
296	  pair: Bytes.Bytes & Maybe<&2, U32>
297	) -> Result<&1, &1, Error, ManifestValue>:
298	  match pair:
299	    case (input, field): decode.comparator.value(input, field)
300	
301	def decode.comparator(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
302	  decode.comparator.field(Bytes.get.u32be(input, 12))
303	
304	def decode.v5.sst.value(
305	  input: Bytes.Bytes,
306	  field: Maybe<&2, U32>
307	) -> Result<&1, &1, Error, ManifestValue>:
308	  match field:
309	    case None{}: Fail{Malformed{}}
310	    case Some{file_id}:
311	      decode.checksum(input, GenerationSst{file_id}, 28, 60)
312	
313	def decode.v5.sst.field(
314	  pair: Bytes.Bytes & Maybe<&2, U32>
315	) -> Result<&1, &1, Error, ManifestValue>:
316	  match pair:
317	    case (input, field): decode.v5.sst.value(input, field)
318	
319	def decode.v5.sst.count.value(
320	  input: Bytes.Bytes,
321	  count_field: Maybe<&2, U32>
322	) -> Result<&1, &1, Error, ManifestValue>:
323	  match count_field:
324	    case None{}: Fail{Malformed{}}
325	    case Some{0}:
326	      decode.checksum(input, GenerationEmpty{}, 24, 56)
327	    case Some{1}:
328	      decode.v5.sst.field(Bytes.get.u32be(input, 24))
329	    case _: Fail{Malformed{}}
330	
331	def decode.v5.sst.count.field(
332	  pair: Bytes.Bytes & Maybe<&2, U32>
333	) -> Result<&1, &1, Error, ManifestValue>:
334	  match pair:
335	    case (input, field): decode.v5.sst.count.value(input, field)
336	
337	def decode.v5.level_count.value(
338	  input: Bytes.Bytes,
339	  field: Maybe<&2, U32>
340	) -> Result<&1, &1, Error, ManifestValue>:
341	  match field:
342	    case None{}: Fail{Malformed{}}
343	    case Some{1}:
344	      decode.v5.sst.count.field(Bytes.get.u32be(input, 20))
345	    case _: Fail{Malformed{}}
346	
347	def decode.v5.level_count.field(
348	  pair: Bytes.Bytes & Maybe<&2, U32>
349	) -> Result<&1, &1, Error, ManifestValue>:
350	  match pair:
351	    case (input, field): decode.v5.level_count.value(input, field)
352	
353	def decode.v5.level_count(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
354	  decode.v5.level_count.field(Bytes.get.u32be(input, 16))
355	
356	def decode.v5.comparator.value(
357	  input: Bytes.Bytes,
358	  field: Maybe<&2, U32>
359	) -> Result<&1, &1, Error, ManifestValue>:
360	  match field:
361	    case None{}: Fail{Malformed{}}
362	    case Some{1}:
363	      decode.v5.level_count(input)
364	    case Some{generation}: Fail{ComparatorMismatch{generation}}
365	
366	def decode.v5.comparator.field(
367	  pair: Bytes.Bytes & Maybe<&2, U32>
368	) -> Result<&1, &1, Error, ManifestValue>:
369	  match pair:
370	    case (input, field): decode.v5.comparator.value(input, field)
371	
372	def decode.v5.comparator(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
373	  decode.v5.comparator.field(Bytes.get.u32be(input, 12))
374	
375	# Rejects unknown storage generations before comparator validation.
376	def decode.version.value(
377	  input: Bytes.Bytes,
378	  field: Maybe<&2, U32>
379	) -> Result<&1, &1, Error, ManifestValue>:
380	  match field:
381	    case None{}: Fail{Malformed{}}
382	    case Some{4}: decode.comparator(input)
383	    case Some{5}: decode.v5.comparator(input)
384	    case Some{generation}: Fail{UnsupportedVersion{generation}}
385	
386	def decode.version.field(
387	  pair: Bytes.Bytes & Maybe<&2, U32>
388	) -> Result<&1, &1, Error, ManifestValue>:
389	  match pair:
390	    case (input, field): decode.version.value(input, field)
391	
392	def decode.version(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
393	  decode.version.field(Bytes.get.u32be(input, 8))
394	
395	# Checks the magic before interpreting version fields.
396	def decode.magic(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, ManifestValue>:
397	  match valid:
398	    case False{}: Fail{Malformed{}}
399	    case True{}: decode.version(input)
400	
401	def decode.magic.pair(pair: Bytes.Bytes & Bool) -> Result<&1, &1, Error, ManifestValue>:
402	  match pair:
403	    case (input, valid): decode.magic(input, valid)
404	
405	def decode.magic.copies(
406	  copies: Bytes.Bytes & Bytes.Bytes
407	) -> Result<&1, &1, Error, ManifestValue>:
408	  match copies:
409	    case (input, for_magic):
410	      decode.magic.pair((input,
411	        equal(Bytes.eq(slice.result(Bytes.slice(for_magic, 0, 8)), magic()))))
412	
413	# Validates the size before reading fixed header fields.
414	def decode.sized.check_magic(bytes: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
415	  decode.magic.copies(clone.split(bytes))
416	
417	def decode.sized.valid(bytes: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, ManifestValue>:
418	  match valid:
419	    case False{}: Fail{Malformed{}}
420	    case True{}: decode.sized.check_magic(bytes)
421	
422	# Validates the size before reading fixed header fields.
423	def decode.sized(input: Bytes.Bytes & U32) -> Result<&1, &1, Error, ManifestValue>:
424	  match input:
425	    case (bytes, size):
426	      match size:
427	        case 56: decode.sized.valid(bytes, True{})
428	        case 63: decode.sized.valid(bytes, True{})
429	        case 60: decode.sized.valid(bytes, True{})
430	        case _: decode.sized.valid(bytes, False{})
431	
432	# Decodes bounded generation-four and generation-five byte manifests.
433	def decode(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
434	  decode.sized(Bytes.length(input))
435	
436	def retired.current(value: ManifestValue) -> Maybe<&2, String>:
437	  match value:
438	    case GenerationSst{file_id}: Some{sst.generation.name(file_id)}
439	    case _: None{}
440	
441	def retired.published(current: Maybe<&2, String>, +name: String) -> Bool:
442	  match current:
443	    case None{}: False{}
444	    case Some{published}: String.eq(published, name)
445	
446	def retired.push(is_retired: Bool, name: String, rest: List<&2, String>) -> List<&2, String>:
447	  match is_retired:
448	    case True{}: Con{name, rest}
449	    case False{}: rest
450	
451	def retired.go(names: List<&2, String>, +current: Maybe<&2, String>) -> List<&2, String>:
452	  match names:
453	    case Nil{}: Nil{}
454	    case Con{+name, rest}:
455	      retired.push(String.starts_with(name, "SST.") &&
456	        Bool.not(retired.published(current, name)), name, retired.go(rest, current))
457	
458	# Lists the SST files in a directory listing that the manifest no longer references.
459	def retired(value: ManifestValue, names: List<&2, String>) -> List<&2, String>:
460	  retired.go(names, retired.current(value))