1 import Base 2 import ../ByteMemTable.bend as ByteMemTable 3 import ../MemTable.bend as MemTable 4 import ../StorageBytes.bend as StorageBytes 5 import ../Wal.bend as Wal 6 import ../coordinator/QueuePolicy.bend as QueuePolicy 7 import ./FormatInspect.bend as FormatInspect 8 import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes 9 10 # Reports the result of verifying all legacy WAL and SST files. 11 type SourceValidation is Data: 12 LegacyFilesVerified{entries: Nat, encoded_bytes: Nat} 13 LegacyFilesRejected{} 14 SourceNotChecked{} 15 16 # Classifies safe migration preflight failures. 17 type 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. 26 type 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. 31 type 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. 40 type PublicationDecision is Data: 41 PublicationAdvanced{state: Publication} 42 PublicationRejected{} 43 44 # Carries legacy record conversion across bounded state-machine steps. 45 type 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 50 def 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 56 def legacy.string.clone(value: String) -> String & String: 57 legacy.string.clone.go(value, SNil{}, SNil{}) 58 59 def canonical.entries(table: ByteMemTable.Table) -> List<&1, ByteMemTable.Entry>: 60 ByteMemTable.compact.live(ByteMemTable.ordered_entries(table)) 61 62 def 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 69 def 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 76 def 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 84 def 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 93 def 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 102 def 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 109 def 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 139 def 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 149 def legacy.entries.measure( 150 entries: List<&2, MemTable.Entry> 151 ) -> List<&2, MemTable.Entry> & Nat: 152 legacy.entries.measure.go(entries, Nil{}, 0n) 153 154 def 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 163 def 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 169 def 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 179 def 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 184 def 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 191 def 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 197 def 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 204 def 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 212 def 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 220 def 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 227 def 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. 235 def 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. 241 def 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. 247 def 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. 253 def 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. 271 def 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. 277 def 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. 284 def 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. 291 def 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. 311 def 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. 328 def 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. 341 def 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. 359 def 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)