src/transfer/Migrate.bend
mylsm-lsm-store@0.6.0.0 · 370 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.0/src/MemTable.bend as MemTable
4import mylsm-lsm-store@0.6.0.0/src/StorageBytes.bend as StorageBytes
5import mylsm-lsm-store@0.6.0.0/src/Wal.bend as Wal
6import mylsm-lsm-store@0.6.0.0/src/coordinator/QueuePolicy.bend as QueuePolicy
7import mylsm-lsm-store@0.6.0.0/src/transfer/FormatInspect.bend as FormatInspect
8import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
9
10# Reports the result of verifying all legacy WAL and SST files.
11type SourceValidation is Data:
12 LegacyFilesVerified{entries: Nat, encoded_bytes: Nat}
13 LegacyFilesRejected{}
14 SourceNotChecked{}
15
16# Classifies safe migration preflight failures.
17type Rejection is Data:
18 UnsupportedSource{}
19 InvalidManifest{}
20 SourceFilesInvalid{}
21 SourceFilesNotVerified{}
22 DestinationExists{}
23 ResourceLimit{}
24
25# Admits a fully verified source or rejects it before destination mutation.
26type Decision is Data:
27 MigrateReady{entries: Nat, encoded_bytes: Nat}
28 MigrateRejected{reason: Rejection}
29
30# Tracks the migration stages that guard atomic destination publication.
31type Publication is Data:
32 SourceValidated{}
33 DestinationStaged{}
34 RecordsCopied{}
35 DestinationValidated{}
36 Published{}
37 Aborted{}
38
39# Reports an allowed stage transition or a rejected transition.
40type PublicationDecision is Data:
41 PublicationAdvanced{state: Publication}
42 PublicationRejected{}
43
44# Carries legacy record conversion across bounded state-machine steps.
45type LegacyApplyState is Type:
46 LegacyApplyNeed{entries: List<&2, MemTable.Entry>, table: ByteMemTable.Table}
47 LegacyApplyConverted{tail: List<&2, MemTable.Entry>, table: ByteMemTable.Table,
48 result: Result<&1, &1, U32 & String, ByteMemTable.Entry>}
49
50def legacy.string.clone.go(value: String, left: String, right: String) -> String & String:
51 match value:
52 case SNil{}: (String.reverse(left), String.reverse(right))
53 case SCon{+character, rest}:
54 legacy.string.clone.go(rest, SCon{character, left}, SCon{character, right})
55
56def legacy.string.clone(value: String) -> String & String:
57 legacy.string.clone.go(value, SNil{}, SNil{})
58
59def canonical.entries(table: ByteMemTable.Table) -> List<&1, ByteMemTable.Entry>:
60 ByteMemTable.compact.live(ByteMemTable.ordered_entries(table))
61
62def legacy.value.bytes.result(
63 result: Result<&1, &1, U32 & String, Bytes.Bytes>
64) -> Result<&1, &1, U32 & String, Maybe<&1, Bytes.Bytes>>:
65 match result:
66 case Fail{error}: Fail{error}
67 case Done{bytes}: Done{Some{bytes}}
68
69def legacy.value.bytes(
70 value: Maybe<&2, String>
71) -> Result<&1, &1, U32 & String, Maybe<&1, Bytes.Bytes>>:
72 match value:
73 case None{}: Done{None{}}
74 case Some{text}: legacy.value.bytes.result(StorageBytes.from_string(text))
75
76def legacy.entry.bytes.value(
77 key: Bytes.Bytes,
78 result: Result<&1, &1, U32 & String, Maybe<&1, Bytes.Bytes>>
79) -> Result<&1, &1, U32 & String, ByteMemTable.Entry>:
80 match result:
81 case Fail{error}: Fail{error}
82 case Done{value}: Done{ByteMemTable.Entry{key, value}}
83
84def legacy.entry.bytes.key(
85 value: Maybe<&2, String>,
86 pair: String & Result<&1, &1, U32 & String, Bytes.Bytes>
87) -> Result<&1, &1, U32 & String, ByteMemTable.Entry>:
88 match pair:
89 case (_, Fail{error}): Fail{error}
90 case (_, Done{key_bytes}):
91 legacy.entry.bytes.value(key_bytes, legacy.value.bytes(value))
92
93def legacy.entry.bytes.start(
94 value: Maybe<&2, String>,
95 pair: String & String
96) -> Result<&1, &1, U32 & String, ByteMemTable.Entry>:
97 match pair:
98 case (source, retained):
99 legacy.entry.bytes.key(value,
100 (retained, StorageBytes.from_string(source)))
101
102def legacy.entry.bytes(
103 entry: MemTable.Entry
104) -> Result<&1, &1, U32 & String, ByteMemTable.Entry>:
105 match entry:
106 case MemTable.Entry{key, value}:
107 legacy.entry.bytes.start(value, legacy.string.clone(key))
108
109def legacy.apply.loop(
110 fuel: Nat,
111 state: LegacyApplyState
112) -> Result<&1, &1, U32 & String, ByteMemTable.Table>:
113 match fuel:
114 case 0n:
115 match state:
116 case LegacyApplyNeed{_, table}: Done{table}
117 case LegacyApplyConverted{_, _, Fail{error}}: Fail{error}
118 case LegacyApplyConverted{_, table, Done{_}}: Done{table}
119 case 1n+rest:
120 match state:
121 case LegacyApplyNeed{entries, table}:
122 match entries:
123 case Nil{}: Done{table}
124 case Con{entry, tail}:
125 legacy.apply.loop(rest, LegacyApplyConverted{tail, table,
126 legacy.entry.bytes(entry)})
127 case LegacyApplyConverted{tail, table, result}:
128 match result:
129 case Fail{error}: Fail{error}
130 case Done{ByteMemTable.Entry{key, value}}:
131 match value:
132 case None{}:
133 legacy.apply.loop(rest, LegacyApplyNeed{tail,
134 ByteMemTable.del(table, key)})
135 case Some{bytes}:
136 legacy.apply.loop(rest, LegacyApplyNeed{tail,
137 ByteMemTable.put(table, key, bytes)})
138
139def legacy.entries.measure.go(
140 entries: List<&2, MemTable.Entry>,
141 reversed: List<&2, MemTable.Entry>,
142 count: Nat
143) -> List<&2, MemTable.Entry> & Nat:
144 match entries:
145 case Nil{}: (List.reverse(&2, MemTable.Entry, reversed), count)
146 case Con{entry, tail}:
147 legacy.entries.measure.go(tail, Con{entry, reversed}, Nat.add(count, 1n))
148
149def legacy.entries.measure(
150 entries: List<&2, MemTable.Entry>
151) -> List<&2, MemTable.Entry> & Nat:
152 legacy.entries.measure.go(entries, Nil{}, 0n)
153
154def legacy.apply.measured(
155 pair: List<&2, MemTable.Entry> & Nat,
156 table: ByteMemTable.Table
157) -> Result<&1, &1, U32 & String, ByteMemTable.Table>:
158 match pair:
159 case (entries, count):
160 legacy.apply.loop(Nat.add(Nat.mul(2n, count), 1n),
161 LegacyApplyNeed{entries, table})
162
163def legacy.apply.chronological(
164 entries: List<&2, MemTable.Entry>,
165 table: ByteMemTable.Table
166) -> Result<&1, &1, U32 & String, ByteMemTable.Table>:
167 legacy.apply.measured(legacy.entries.measure(entries), table)
168
169def legacy.batch.entries(mutations: List<&2, Wal.Mut>) -> List<&2, MemTable.Entry>:
170 match mutations:
171 case Nil{}: Nil{}
172 case Con{mutation, tail}:
173 match mutation:
174 case Wal.Put{key, val}:
175 Con{MemTable.Entry{key, Some{val}}, legacy.batch.entries(tail)}
176 case Wal.Del{key}:
177 Con{MemTable.Entry{key, None{}}, legacy.batch.entries(tail)}
178
179def legacy.mutations.count(mutations: List<&2, Wal.Mut>) -> Nat:
180 match mutations:
181 case Nil{}: 0n
182 case Con{_, tail}: Nat.add(1n, legacy.mutations.count(tail))
183
184def legacy.batches.mutation.count(batches: List<&2, Wal.Batch>) -> Nat:
185 match batches:
186 case Nil{}: 0n
187 case Con{Wal.Batch{muts}, tail}:
188 Nat.add(legacy.mutations.count(muts),
189 legacy.batches.mutation.count(tail))
190
191def legacy.source.mutation.count(
192 sst_entries: Nat,
193 wal_batches: List<&2, Wal.Batch>
194) -> Nat:
195 Nat.add(sst_entries, legacy.batches.mutation.count(wal_batches))
196
197def legacy.batches.entries(batches: List<&2, Wal.Batch>) -> List<&2, MemTable.Entry>:
198 match batches:
199 case Nil{}: Nil{}
200 case Con{Wal.Batch{muts}, tail}:
201 List.append(&2, MemTable.Entry, legacy.batch.entries(muts),
202 legacy.batches.entries(tail))
203
204def legacy.sources.chronological(
205 sst_newest_first: List<&2, MemTable.Entry>,
206 wal_batches: List<&2, Wal.Batch>
207) -> List<&2, MemTable.Entry>:
208 List.append(&2, MemTable.Entry,
209 List.reverse(&2, MemTable.Entry, sst_newest_first),
210 legacy.batches.entries(wal_batches))
211
212def legacy.snapshot.table(
213 sst_newest_first: List<&2, MemTable.Entry>,
214 wal_batches: List<&2, Wal.Batch>
215) -> Result<&1, &1, U32 & String, ByteMemTable.Table>:
216 legacy.apply.chronological(
217 legacy.sources.chronological(sst_newest_first, wal_batches),
218 ByteMemTable.empty())
219
220def legacy.snapshot.canonicalize(
221 result: Result<&1, &1, U32 & String, ByteMemTable.Table>
222) -> Result<&1, &1, U32 & String, List<&1, ByteMemTable.Entry>>:
223 match result:
224 case Fail{error}: Fail{error}
225 case Done{table}: Done{canonical.entries(table)}
226
227def legacy.snapshot.entries(
228 sst_newest_first: List<&2, MemTable.Entry>,
229 wal_batches: List<&2, Wal.Batch>
230) -> Result<&1, &1, U32 & String, List<&1, ByteMemTable.Entry>>:
231 legacy.snapshot.canonicalize(
232 legacy.snapshot.table(sst_newest_first, wal_batches))
233
234# Begins publication only for a source admitted by the migration plan.
235def begin(decision: Decision) -> PublicationDecision:
236 match decision:
237 case MigrateReady{_, _}: PublicationAdvanced{SourceValidated{}}
238 case MigrateRejected{_}: PublicationRejected{}
239
240# Stages a sibling destination only after source validation.
241def stage_destination(state: Publication) -> PublicationDecision:
242 match state:
243 case SourceValidated{}: PublicationAdvanced{DestinationStaged{}}
244 case _: PublicationRejected{}
245
246# Marks the copy complete only for an initialized staging destination.
247def records_copied(state: Publication) -> PublicationDecision:
248 match state:
249 case DestinationStaged{}: PublicationAdvanced{RecordsCopied{}}
250 case _: PublicationRejected{}
251
252# Marks the staged database verified only when count, digest, and order agree.
253def destination_validated(
254 state: Publication,
255 counts: Nat & Nat,
256 checks: Bool & Bool
257) -> PublicationDecision:
258 match state:
259 case RecordsCopied{}:
260 match counts:
261 case (source_entries, destination_entries):
262 match checks:
263 case (digest_equal, canonical_order):
264 Bool.pick(PublicationDecision,
265 Nat.is_eq(source_entries, destination_entries) && digest_equal && canonical_order,
266 PublicationAdvanced{DestinationValidated{}},
267 PublicationRejected{})
268 case _: PublicationRejected{}
269
270# Allows final publication only after destination verification.
271def publish(state: Publication) -> PublicationDecision:
272 match state:
273 case DestinationValidated{}: PublicationAdvanced{Published{}}
274 case _: PublicationRejected{}
275
276# Aborts an unpublished migration without retracting a published destination.
277def abort(state: Publication) -> PublicationDecision:
278 match state:
279 case Published{}: PublicationRejected{}
280 case Aborted{}: PublicationRejected{}
281 case _: PublicationAdvanced{Aborted{}}
282
283# Accepts only the durable API and storage identities promised for migration.
284def source.supported(identity: FormatInspect.Identity) -> Bool:
285 match identity:
286 case FormatInspect.Identity{package_version, storage_version, generation}:
287 String.eq(package_version, "0.5.0.0") &&
288 String.eq(storage_version, "0.4.0.0") && U32.is_eq(generation, 3)
289
290# Applies count and byte limits after duplicating the affine byte count.
291def gate.destination.measured(
292 max_entries: Nat,
293 max_encoded_bytes: Nat,
294 destination_missing: Bool,
295 entry_counts: Nat & Nat,
296 byte_counts: Nat & Nat
297) -> Decision:
298 match entry_counts:
299 case (result_entries, measured_entries):
300 match byte_counts:
301 case (result_bytes, measured_bytes):
302 Bool.pick(Decision, destination_missing,
303 Bool.pick(Decision,
304 Nat.is_le(measured_entries, max_entries) &&
305 Nat.is_le(measured_bytes, max_encoded_bytes),
306 MigrateReady{result_entries, result_bytes},
307 MigrateRejected{ResourceLimit{}}),
308 MigrateRejected{DestinationExists{}})
309
310# Checks verified record and byte counts against configured migration limits.
311def gate.destination(
312 validation: SourceValidation,
313 destination_missing: Bool,
314 max_entries: Nat,
315 max_encoded_bytes: Nat
316) -> Decision:
317 match validation:
318 case LegacyFilesVerified{entries, encoded_bytes}:
319 gate.destination.measured(max_entries, max_encoded_bytes,
320 destination_missing, QueuePolicy.nat.clone.pair(entries),
321 QueuePolicy.nat.clone.pair(encoded_bytes))
322 case LegacyFilesRejected{}:
323 MigrateRejected{SourceFilesInvalid{}}
324 case SourceNotChecked{}:
325 MigrateRejected{SourceFilesNotVerified{}}
326
327# Applies the source compatibility check before destination admission.
328def gate.source(
329 validation: SourceValidation,
330 destination_missing: Bool,
331 max_entries: Nat,
332 max_encoded_bytes: Nat,
333 supported: Bool
334) -> Decision:
335 Bool.pick(Decision, supported,
336 gate.destination(validation, destination_missing, max_entries,
337 max_encoded_bytes),
338 MigrateRejected{UnsupportedSource{}})
339
340# Requires a valid manifest before considering the legacy file verification.
341def gate.manifest(
342 validation: SourceValidation,
343 destination_missing: Bool,
344 max_entries: Nat,
345 max_encoded_bytes: Nat,
346 supported: Bool,
347 integrity: FormatInspect.Integrity
348) -> Decision:
349 match integrity:
350 case FormatInspect.Valid{}:
351 gate.source(validation, destination_missing, max_entries,
352 max_encoded_bytes, supported)
353 case FormatInspect.ResourceLimit{}:
354 MigrateRejected{ResourceLimit{}}
355 case _:
356 MigrateRejected{InvalidManifest{}}
357
358# Chooses compatibility, validation, destination, and resource preconditions.
359def plan(
360 inspection: FormatInspect.Inspection,
361 validation: SourceValidation,
362 destination_missing: Bool,
363 max_entries: Nat,
364 max_encoded_bytes: Nat
365) -> Decision:
366 match inspection:
367 case FormatInspect.Inspection{identity, integrity}:
368 gate.manifest(validation, destination_missing, max_entries,
369 max_encoded_bytes,
370 source.supported(identity), integrity)