src/ByteManifest.bend

mylsm-lsm-store@0.6.0.1 · 460 lines · 71 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/hub_sha/sha256.bend as SHA
3import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
4
5# Represents a legacy or generation-addressed byte-store table layout.
6type ManifestValue is Data:
7  Empty{}
8  HasSst{}
9  GenerationEmpty{}
10  GenerationSst{file_id: U32}
11
12# Distinguishes malformed bytes, unknown formats, and comparator mismatch.
13type 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.
20type FileIdDecision is Data:
21  FileIdAvailable{id: U32}
22  FileIdExhausted{}
23
24# Unwraps a fixed manifest-format literal.
25def 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.
31def magic() -> Bytes.Bytes:
32  literal(Bytes.from_hex("4d594c534d344d00"))
33
34# Encodes one unsigned big-endian field.
35def 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.
39def sst.name() -> Bytes.Bytes:
40  literal(Bytes.from_hex("535354"))
41
42# Names the SST file of one generation.
43def sst.generation.name(file_id: U32) -> String:
44  "SST." ++ Bytes.to_hex(u32.bytes(file_id))
45
46def 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.
50def 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.
55def 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.
61def 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.
70def 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.
79def 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.
89def 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.
96def 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.
111def 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.
117def encode.body(packed: Bytes.Bytes) -> Bytes.Bytes:
118  encode.body.copies(clone.split(packed))
119
120# Encodes the currently supported comparator and table layout.
121def encode(value: ManifestValue) -> Bytes.Bytes:
122  encode.body(body(value))
123
124def u32.read.value(pair: Bytes.Bytes & Maybe<&2, U32>) -> U32:
125  match pair:
126    case (_, Some{value}): value
127    case _: 0
128
129def u32.read(input: Bytes.Bytes) -> U32:
130  u32.read.value(Bytes.get.u32be(input, 0))
131
132def 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
137def clone.file.id(file_id: U32) -> U32 & U32:
138  u32.clone.pair.bytes(clone.split(u32.bytes(file_id)))
139
140def 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
145def 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
151def 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
155def 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.
163def 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.
168def 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.
173def 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.
182def 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.
195def 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.
207def 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.
218def 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.
227def 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.
233def 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.
243def 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.
247def 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
257def 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
263def 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.
267def 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
276def 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
282def 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.
286def 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
295def 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
301def decode.comparator(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
302  decode.comparator.field(Bytes.get.u32be(input, 12))
303
304def 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
313def 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
319def 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
331def 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
337def 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
347def 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
353def decode.v5.level_count(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
354  decode.v5.level_count.field(Bytes.get.u32be(input, 16))
355
356def 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
366def 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
372def 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.
376def 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
386def 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
392def 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.
396def 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
401def decode.magic.pair(pair: Bytes.Bytes & Bool) -> Result<&1, &1, Error, ManifestValue>:
402  match pair:
403    case (input, valid): decode.magic(input, valid)
404
405def 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.
414def decode.sized.check_magic(bytes: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
415  decode.magic.copies(clone.split(bytes))
416
417def 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.
423def 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.
433def decode(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
434  decode.sized(Bytes.length(input))
435
436def 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
441def 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
446def 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
451def 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.
459def retired(value: ManifestValue, names: List<&2, String>) -> List<&2, String>:
460  retired.go(names, retired.current(value))