Skip to main content

ethrex_storage/
migrations.rs

1use std::io::Write;
2use std::path::Path;
3use std::time::{Duration, Instant};
4
5use ethrex_common::H256;
6use ethrex_common::types::{BlockHash, BlockNumber, Index};
7use ethrex_rlp::decode::RLPDecode;
8use ethrex_rlp::encode::RLPEncode;
9
10use crate::api::tables::{RECEIPTS, RECEIPTS_V2, TRANSACTION_LOCATIONS};
11use crate::api::{StorageBackend, StorageWriteBatch};
12use crate::error::StoreError;
13use crate::store::receipt_key;
14use crate::{STORE_METADATA_FILENAME, STORE_SCHEMA_VERSION};
15
16use super::store::StoreMetadata;
17
18/// A migration function that upgrades the database schema by one version.
19///
20/// Receives a reference to the storage backend so it can read/write data
21/// as needed for the migration.
22pub type MigrationFn = fn(backend: &dyn StorageBackend) -> Result<(), StoreError>;
23
24/// Migration functions indexed by source version.
25///
26/// `MIGRATIONS[i]` upgrades the schema from version `(i + 1)` to `(i + 2)`.
27/// For example:
28/// - `MIGRATIONS[0]` upgrades v1 → v2
29/// - `MIGRATIONS[1]` upgrades v2 → v3
30///
31/// **Invariant**: `MIGRATIONS.len() == (STORE_SCHEMA_VERSION - 1) as usize`
32/// (empty when `STORE_SCHEMA_VERSION == 1`, one entry when it's 2, etc.)
33pub const MIGRATIONS: &[MigrationFn] = &[migrate_1_to_2, migrate_2_to_3, migrate_3_to_4];
34
35// Compile-time check: the number of migration functions must match the number
36// of version gaps (i.e. STORE_SCHEMA_VERSION - 1).
37const _: () = assert!(
38    MIGRATIONS.len() == (STORE_SCHEMA_VERSION - 1) as usize,
39    "MIGRATIONS length must equal STORE_SCHEMA_VERSION - 1"
40);
41
42/// Returns the migration function that upgrades from `version` to `version + 1`.
43fn migration_for_version(version: u64) -> MigrationFn {
44    MIGRATIONS[(version - 1) as usize]
45}
46
47/// v3 → v4: no data change.
48///
49/// `ACCOUNT_CODES` values carry their JUMPDEST positions as a bitmap rather than an RLP
50/// list of `u32` offsets. Both forms are readable, so a v3 database needs no rewriting
51/// and this migration only moves the version marker.
52///
53/// The bump exists for the other direction: a v3 binary cannot decode a bitmap, so the
54/// store refuses to open a v4 database (`IncompatibleDBVersion`) instead of letting it
55/// fail on the first code read.
56fn migrate_3_to_4(_backend: &dyn StorageBackend) -> Result<(), StoreError> {
57    Ok(())
58}
59
60/// Minimum interval between migration progress log lines.
61const PROGRESS_LOG_INTERVAL: Duration = Duration::from_secs(10);
62
63/// Per-second processing rate for progress logs. Returns 0 when `elapsed` is
64/// zero so the division can never produce `inf` or `NaN`.
65fn entries_per_second(count: u64, elapsed: Duration) -> f64 {
66    let secs = elapsed.as_secs_f64();
67    if secs > 0.0 { count as f64 / secs } else { 0.0 }
68}
69
70/// Runs all pending migrations from `current_version` up to `STORE_SCHEMA_VERSION`.
71///
72/// Each migration is applied one version at a time, and the metadata file is
73/// updated (with fsync) after each successful step for crash safety.
74///
75/// Returns `Ok(())` if `current_version == STORE_SCHEMA_VERSION` (no-op).
76/// If `current_version > STORE_SCHEMA_VERSION` (older binary against a newer
77/// database), it returns `IncompatibleDBVersion` without touching the database:
78/// there is no downgrade path, and continuing would fail on the first read of a
79/// format this binary cannot decode.
80pub fn run_pending_migrations(
81    backend: &dyn StorageBackend,
82    db_path: &Path,
83    current_version: u64,
84) -> Result<(), StoreError> {
85    if current_version > STORE_SCHEMA_VERSION {
86        return Err(StoreError::IncompatibleDBVersion {
87            found: current_version,
88            expected: STORE_SCHEMA_VERSION,
89        });
90    }
91
92    let pending = STORE_SCHEMA_VERSION - current_version;
93    if pending == 0 {
94        return Ok(());
95    }
96
97    tracing::info!(
98        "Database schema is at v{current_version}, latest is v{STORE_SCHEMA_VERSION}; running {pending} migration(s). This may take a while on large databases"
99    );
100
101    for version in current_version..STORE_SCHEMA_VERSION {
102        let target = version + 1;
103
104        tracing::info!("Running schema migration v{version} → v{target}");
105        let start = Instant::now();
106
107        migration_for_version(version)(backend).map_err(|e| StoreError::MigrationFailed {
108            from: version,
109            to: target,
110            reason: e.to_string(),
111        })?;
112
113        // Persist the new version to metadata.json after each migration step
114        write_metadata_version(db_path, target).map_err(|e| StoreError::MigrationFailed {
115            from: version,
116            to: target,
117            reason: format!("failed to write metadata: {e}"),
118        })?;
119
120        tracing::info!(
121            "Schema migration v{version} → v{target} completed in {:.1}s",
122            start.elapsed().as_secs_f64()
123        );
124    }
125
126    Ok(())
127}
128
129/// Writes the schema version to metadata.json using write-to-temp-then-rename
130/// for crash safety. On POSIX filesystems `rename` is atomic, so the metadata
131/// file is never left in a partial/truncated state.
132// TODO: move metadata persistence into the StorageBackend abstraction so we
133// don't need to pass `db_path` around.
134fn write_metadata_version(db_path: &Path, version: u64) -> Result<(), StoreError> {
135    let metadata_path = db_path.join(STORE_METADATA_FILENAME);
136    let tmp_path = db_path.join(format!("{}.tmp", STORE_METADATA_FILENAME));
137    let metadata = StoreMetadata::new(version);
138    let serialized = serde_json::to_string_pretty(&metadata)?;
139
140    let mut file = std::fs::File::create(&tmp_path)?;
141    file.write_all(serialized.as_bytes())?;
142    file.sync_all()?;
143    std::fs::rename(&tmp_path, &metadata_path)?;
144
145    Ok(())
146}
147
148/// Migrates the RECEIPTS table from RLP-encoded `(BlockHash, u64)` keys
149/// to raw `block_hash (32B) || index (8B big-endian u64)` keys in a new
150/// `receipts_v2` column family.
151///
152/// This two-CF approach copies entries from the old `receipts` CF to
153/// `receipts_v2` with the new key format. The old `receipts` CF is **not**
154/// deleted here — `Store::new()` calls `drop_obsolete_cfs()` right after
155/// this migration returns, which drops it in the same startup.
156///
157/// Crash safety: if interrupted, metadata still says v1, so the migration
158/// restarts from scratch on next boot. Duplicate puts to `receipts_v2` are
159/// idempotent.
160fn migrate_1_to_2(backend: &dyn StorageBackend) -> Result<(), StoreError> {
161    const BATCH_SIZE: usize = 10_000;
162
163    let txn = backend.begin_read()?;
164    let iter = txn.prefix_iterator(RECEIPTS, &[])?;
165
166    let mut batch: Vec<(Vec<u8>, Vec<u8>)> = Vec::with_capacity(BATCH_SIZE);
167    let mut migrated: u64 = 0;
168    let start = Instant::now();
169    let mut last_progress_log = Instant::now();
170
171    for result in iter {
172        let (key, value) = result?;
173
174        let (block_hash, index) = match <(H256, u64)>::decode(&key) {
175            Ok(decoded) => decoded,
176            Err(_) => {
177                tracing::warn!(
178                    "Schema migration v1 → v2: skipping receipts key that failed RLP decode (len={})",
179                    key.len()
180                );
181                continue;
182            }
183        };
184
185        let new_key = receipt_key(&block_hash, index);
186        batch.push((new_key, value.to_vec()));
187
188        if batch.len() >= BATCH_SIZE {
189            let count = batch.len() as u64;
190            let mut tx = backend.begin_write()?;
191            tx.put_batch(RECEIPTS_V2, std::mem::take(&mut batch))?;
192            tx.commit()?;
193            migrated += count;
194            if last_progress_log.elapsed() >= PROGRESS_LOG_INTERVAL {
195                let rate = entries_per_second(migrated, start.elapsed());
196                tracing::info!(
197                    "Schema migration v1 → v2: {migrated} receipt entries migrated so far ({rate:.0} entries/s)"
198                );
199                last_progress_log = Instant::now();
200            }
201        }
202    }
203
204    // Flush remaining entries.
205    if !batch.is_empty() {
206        let count = batch.len() as u64;
207        let mut tx = backend.begin_write()?;
208        tx.put_batch(RECEIPTS_V2, batch)?;
209        tx.commit()?;
210        migrated += count;
211    }
212
213    tracing::info!("Schema migration v1 → v2: migrated {migrated} receipt entries in total");
214    Ok(())
215}
216
217type TxLocation = (BlockNumber, BlockHash, Index);
218
219/// Rewrites `TRANSACTION_LOCATIONS` from the v2 composite-key schema
220/// (`key = tx_hash || block_hash`, `value = (block_number, block_hash, index)`)
221/// to the v3 schema (`key = tx_hash`, `value = Vec<(block_number, block_hash, index)>`).
222///
223/// Streams the old table in lex order, grouping consecutive entries by tx_hash
224/// (composite keys with the same 32-byte prefix are adjacent — both backends
225/// iterate sorted). Flushes each group as an atomic write batch (merge the new
226/// key + delete the old composite keys), chunking commits to bound memory.
227/// Skips any already-migrated 32-byte keys it encounters.
228///
229/// Crash-resume is safe by construction: the new value is written with `merge`,
230/// not `put`, so if a tx_hash ever has both a v3 value and leftover v2 siblings
231/// (e.g. a non-atomic backend), the resumed run unions them (deduping by
232/// block_hash) instead of overwriting. The merge operator that already backs
233/// the live write path makes this free.
234fn migrate_2_to_3(backend: &dyn StorageBackend) -> Result<(), StoreError> {
235    const GROUPS_PER_COMMIT: usize = 50_000;
236
237    let read = backend.begin_read()?;
238    // Empty prefix → full-table scan. Both backends yield keys in sorted order,
239    // which the same-prefix grouping below relies on.
240    let iter = read.prefix_iterator(TRANSACTION_LOCATIONS, &[])?;
241
242    let mut write_batch = backend.begin_write()?;
243    let mut groups_in_batch: usize = 0;
244    let mut current: Option<(H256, Vec<TxLocation>, Vec<Vec<u8>>)> = None;
245    let mut total_groups: u64 = 0;
246    let mut total_old_entries: u64 = 0;
247    let start = Instant::now();
248    let mut last_progress_log = Instant::now();
249
250    for result in iter {
251        let (key, value) = result?;
252
253        // Already-migrated entries (32-byte tx_hash keys, from a prior partial run): skip.
254        if key.len() == 32 {
255            continue;
256        }
257        if key.len() != 64 {
258            return Err(StoreError::Custom(format!(
259                "unexpected TRANSACTION_LOCATIONS key length {} during migration",
260                key.len()
261            )));
262        }
263
264        total_old_entries += 1;
265        if last_progress_log.elapsed() >= PROGRESS_LOG_INTERVAL {
266            let rate = entries_per_second(total_old_entries, start.elapsed());
267            tracing::info!(
268                "Schema migration v2 → v3: {total_old_entries} transaction location entries processed so far ({rate:.0} entries/s)"
269            );
270            last_progress_log = Instant::now();
271        }
272
273        let tx_hash = H256::from_slice(&key[..32]);
274        let location = TxLocation::decode(&value)?;
275        let key_vec = key.into_vec();
276
277        match &mut current {
278            Some((h, locs, keys_to_delete)) if *h == tx_hash => {
279                locs.push(location);
280                keys_to_delete.push(key_vec);
281            }
282            _ => {
283                if let Some((h, locs, keys_to_delete)) = current.take() {
284                    flush_tx_location_group(&mut *write_batch, h, locs, keys_to_delete)?;
285                    total_groups += 1;
286                    groups_in_batch += 1;
287                    if groups_in_batch >= GROUPS_PER_COMMIT {
288                        write_batch.commit()?;
289                        // Re-acquire instead of relying on post-commit reuse
290                        // of the trait object (works today via mem::take in
291                        // RocksDB and a no-op InMemory commit, but it's not
292                        // a documented contract on `StorageWriteBatch`).
293                        write_batch = backend.begin_write()?;
294                        groups_in_batch = 0;
295                    }
296                }
297                current = Some((tx_hash, vec![location], vec![key_vec]));
298            }
299        }
300    }
301
302    if let Some((h, locs, keys_to_delete)) = current {
303        flush_tx_location_group(&mut *write_batch, h, locs, keys_to_delete)?;
304        total_groups += 1;
305    }
306
307    // Final commit. `groups_in_batch` is not bumped/reset here intentionally
308    // — the post-loop flush is followed immediately by a commit, after which
309    // the variable goes out of scope.
310    write_batch.commit()?;
311
312    tracing::info!(
313        "Schema migration v2 → v3: rewrote {} transaction location entries into {} transaction records",
314        total_old_entries,
315        total_groups
316    );
317    Ok(())
318}
319
320fn flush_tx_location_group(
321    write_batch: &mut dyn StorageWriteBatch,
322    tx_hash: H256,
323    locations: Vec<TxLocation>,
324    composite_keys: Vec<Vec<u8>>,
325) -> Result<(), StoreError> {
326    // Use `merge`, not `put`: the operand is the same `Vec` type as the value,
327    // so a re-processed group unions with any existing v3 value (dedup by
328    // block_hash) instead of overwriting it. The composite-key deletes ride in
329    // the same batch, so the group is applied atomically.
330    write_batch.merge(
331        TRANSACTION_LOCATIONS,
332        tx_hash.as_bytes(),
333        &locations.encode_to_vec(),
334    )?;
335    for key in composite_keys {
336        write_batch.delete(TRANSACTION_LOCATIONS, &key)?;
337    }
338    Ok(())
339}
340
341#[cfg(test)]
342mod tests {
343    use super::*;
344
345    #[test]
346    fn migrations_length_matches_schema_version() {
347        assert_eq!(
348            MIGRATIONS.len(),
349            (STORE_SCHEMA_VERSION - 1) as usize,
350            "MIGRATIONS array length must be STORE_SCHEMA_VERSION - 1"
351        );
352    }
353
354    #[test]
355    fn run_pending_migrations_noop_when_current() {
356        // When current_version == STORE_SCHEMA_VERSION, nothing should happen.
357        // We use a dummy in-memory backend since no migrations will actually run.
358        let backend = crate::backend::in_memory::InMemoryBackend::open().unwrap();
359        let temp_dir = tempfile::tempdir().unwrap();
360
361        // Write initial metadata
362        write_metadata_version(temp_dir.path(), STORE_SCHEMA_VERSION).unwrap();
363
364        let result = run_pending_migrations(&backend, temp_dir.path(), STORE_SCHEMA_VERSION);
365        assert!(result.is_ok());
366    }
367
368    #[test]
369    fn run_pending_migrations_rejects_a_database_ahead_of_the_binary() {
370        // A database stamped by a newer ethrex must be refused, not "migrated"
371        // downwards or silently opened: this binary cannot decode formats
372        // introduced after its schema version.
373        let backend = crate::backend::in_memory::InMemoryBackend::open().unwrap();
374        let temp_dir = tempfile::tempdir().unwrap();
375        let newer = STORE_SCHEMA_VERSION + 1;
376        write_metadata_version(temp_dir.path(), newer).unwrap();
377
378        let result = run_pending_migrations(&backend, temp_dir.path(), newer);
379
380        assert!(
381            matches!(
382                result,
383                Err(StoreError::IncompatibleDBVersion { found, expected })
384                    if found == newer && expected == STORE_SCHEMA_VERSION
385            ),
386            "expected IncompatibleDBVersion, got {result:?}"
387        );
388        // The marker is left as the newer binary wrote it, so that binary can
389        // still open the database.
390        let contents =
391            std::fs::read_to_string(temp_dir.path().join(STORE_METADATA_FILENAME)).unwrap();
392        let metadata: StoreMetadata = serde_json::from_str(&contents).unwrap();
393        assert_eq!(metadata.schema_version, newer);
394    }
395
396    #[test]
397    fn fresh_store_creates_correct_metadata() {
398        let temp_dir = tempfile::tempdir().unwrap();
399
400        write_metadata_version(temp_dir.path(), STORE_SCHEMA_VERSION).unwrap();
401
402        let metadata_path = temp_dir.path().join(STORE_METADATA_FILENAME);
403        let contents = std::fs::read_to_string(&metadata_path).unwrap();
404        let metadata: StoreMetadata = serde_json::from_str(&contents).unwrap();
405        assert_eq!(metadata.schema_version, STORE_SCHEMA_VERSION);
406    }
407
408    #[test]
409    fn migrate_1_to_2_converts_rlp_keys_to_fixed_width() {
410        use crate::api::StorageBackend;
411        use ethrex_common::types::{Receipt, TxType};
412        use ethrex_rlp::encode::RLPEncode;
413
414        let backend = crate::backend::in_memory::InMemoryBackend::open().unwrap();
415
416        let block_hash = H256::random();
417        let receipts: Vec<Receipt> = (0..5)
418            .map(|i| Receipt::new(TxType::Legacy, true, (i + 1) * 21000, vec![]))
419            .collect();
420
421        // Seed old-format RLP keys: (BlockHash, u64).encode_to_vec()
422        {
423            let mut tx = backend.begin_write().unwrap();
424            let batch: Vec<(Vec<u8>, Vec<u8>)> = receipts
425                .iter()
426                .enumerate()
427                .map(|(i, r)| {
428                    let old_key = (block_hash, i as u64).encode_to_vec();
429                    let value = r.encode_to_vec();
430                    (old_key, value)
431                })
432                .collect();
433            tx.put_batch(RECEIPTS, batch).unwrap();
434            tx.commit().unwrap();
435        }
436
437        // Verify old keys exist
438        {
439            let txn = backend.begin_read().unwrap();
440            let old_key = (block_hash, 0u64).encode_to_vec();
441            assert!(txn.get(RECEIPTS, &old_key).unwrap().is_some());
442        }
443
444        // Run migration
445        migrate_1_to_2(&backend).unwrap();
446
447        // Verify new fixed-width keys exist in RECEIPTS_V2
448        let txn = backend.begin_read().unwrap();
449        for i in 0..5u64 {
450            let new_key = receipt_key(&block_hash, i);
451            let value = txn
452                .get(RECEIPTS_V2, &new_key)
453                .unwrap()
454                .expect("new key should exist in RECEIPTS_V2 after migration");
455            let decoded = Receipt::decode(value.as_ref()).unwrap();
456            assert_eq!(decoded, receipts[i as usize]);
457
458            // Old keys should still be in RECEIPTS (drop_obsolete_cfs runs after migration)
459            let old_key = (block_hash, i).encode_to_vec();
460            assert!(
461                txn.get(RECEIPTS, &old_key).unwrap().is_some(),
462                "old key should still exist in RECEIPTS (dropped after migration)"
463            );
464        }
465    }
466
467    /// Seeds the backend with one entry under the v2 composite-key schema:
468    /// `key = tx_hash || block_hash`, `value = (block_number, block_hash, index)`.
469    fn seed_old_entry(
470        backend: &dyn StorageBackend,
471        tx_hash: H256,
472        block_number: BlockNumber,
473        block_hash: BlockHash,
474        index: Index,
475    ) {
476        let mut composite_key = Vec::with_capacity(64);
477        composite_key.extend_from_slice(tx_hash.as_bytes());
478        composite_key.extend_from_slice(block_hash.as_bytes());
479        let value = (block_number, block_hash, index).encode_to_vec();
480
481        let mut batch = backend.begin_write().unwrap();
482        batch
483            .put(TRANSACTION_LOCATIONS, &composite_key, &value)
484            .unwrap();
485        batch.commit().unwrap();
486    }
487
488    fn read_new_entry(
489        backend: &dyn StorageBackend,
490        tx_hash: H256,
491    ) -> Option<Vec<(BlockNumber, BlockHash, Index)>> {
492        let read = backend.begin_read().unwrap();
493        let bytes = read
494            .get(TRANSACTION_LOCATIONS, tx_hash.as_bytes())
495            .unwrap()?;
496        Some(<Vec<(BlockNumber, BlockHash, Index)>>::decode(&bytes).unwrap())
497    }
498
499    fn h256(byte: u8) -> H256 {
500        H256::from_low_u64_be(byte as u64)
501    }
502
503    #[test]
504    fn migrate_2_to_3_empty_table() {
505        let backend = crate::backend::in_memory::InMemoryBackend::open().unwrap();
506        migrate_2_to_3(&backend).unwrap();
507        // Nothing to assert other than no error and no spurious entries.
508        assert!(read_new_entry(&backend, h256(1)).is_none());
509    }
510
511    #[test]
512    fn migrate_2_to_3_single_entry_per_hash() {
513        let backend = crate::backend::in_memory::InMemoryBackend::open().unwrap();
514        seed_old_entry(&backend, h256(1), 100, h256(0x10), 0);
515        seed_old_entry(&backend, h256(2), 101, h256(0x11), 5);
516        seed_old_entry(&backend, h256(3), 102, h256(0x12), 7);
517
518        migrate_2_to_3(&backend).unwrap();
519
520        assert_eq!(
521            read_new_entry(&backend, h256(1)).unwrap(),
522            vec![(100u64, h256(0x10), 0u64)]
523        );
524        assert_eq!(
525            read_new_entry(&backend, h256(2)).unwrap(),
526            vec![(101u64, h256(0x11), 5u64)]
527        );
528        assert_eq!(
529            read_new_entry(&backend, h256(3)).unwrap(),
530            vec![(102u64, h256(0x12), 7u64)]
531        );
532
533        // Old composite-key entries are gone.
534        let read = backend.begin_read().unwrap();
535        let iter = read.prefix_iterator(TRANSACTION_LOCATIONS, &[]).unwrap();
536        for entry in iter {
537            let (key, _) = entry.unwrap();
538            assert_eq!(key.len(), 32, "leftover non-migrated key: {:?}", key);
539        }
540    }
541
542    #[test]
543    fn migrate_2_to_3_multi_block_per_hash() {
544        let backend = crate::backend::in_memory::InMemoryBackend::open().unwrap();
545        // Same tx hash appears in three different blocks (reorg scenario).
546        seed_old_entry(&backend, h256(0xAA), 100, h256(0x10), 3);
547        seed_old_entry(&backend, h256(0xAA), 100, h256(0x11), 4);
548        seed_old_entry(&backend, h256(0xAA), 101, h256(0x12), 5);
549
550        migrate_2_to_3(&backend).unwrap();
551
552        let mut got = read_new_entry(&backend, h256(0xAA)).unwrap();
553        got.sort();
554        let mut expected = vec![
555            (100u64, h256(0x10), 3u64),
556            (100u64, h256(0x11), 4u64),
557            (101u64, h256(0x12), 5u64),
558        ];
559        expected.sort();
560        assert_eq!(got, expected);
561    }
562
563    #[test]
564    fn migrate_2_to_3_is_idempotent_on_partial_state() {
565        // Simulate a crash-resume: the backend already has a v3 32-byte entry
566        // for one tx hash (from a previously-completed chunk), and v2 composite
567        // entries for another tx hash (still pending).
568        let backend = crate::backend::in_memory::InMemoryBackend::open().unwrap();
569
570        // Already-migrated v3 entry for h256(1).
571        {
572            let v3_value: Vec<(BlockNumber, BlockHash, Index)> =
573                vec![(100, h256(0x10), 0), (100, h256(0x11), 0)];
574            let mut batch = backend.begin_write().unwrap();
575            batch
576                .put(
577                    TRANSACTION_LOCATIONS,
578                    h256(1).as_bytes(),
579                    &v3_value.encode_to_vec(),
580                )
581                .unwrap();
582            batch.commit().unwrap();
583        }
584        // Pending v2 entries for h256(2).
585        seed_old_entry(&backend, h256(2), 200, h256(0x20), 0);
586        seed_old_entry(&backend, h256(2), 200, h256(0x21), 1);
587
588        migrate_2_to_3(&backend).unwrap();
589
590        // h256(1)'s already-migrated entry is unchanged.
591        assert_eq!(
592            read_new_entry(&backend, h256(1)).unwrap(),
593            vec![(100u64, h256(0x10), 0u64), (100u64, h256(0x11), 0u64)]
594        );
595
596        // h256(2) is now migrated.
597        let mut got = read_new_entry(&backend, h256(2)).unwrap();
598        got.sort();
599        let mut expected = vec![(200u64, h256(0x20), 0u64), (200u64, h256(0x21), 1u64)];
600        expected.sort();
601        assert_eq!(got, expected);
602
603        // No leftover 64-byte keys.
604        let read = backend.begin_read().unwrap();
605        let iter = read.prefix_iterator(TRANSACTION_LOCATIONS, &[]).unwrap();
606        for entry in iter {
607            let (key, _) = entry.unwrap();
608            assert_eq!(key.len(), 32);
609        }
610    }
611
612    /// The pathological case flagged in review: a single tx_hash has BOTH a v3
613    /// value (from a prior partial run) AND leftover v2 composite keys. Because
614    /// `flush_tx_location_group` uses `merge` (not `put`), the resumed migration
615    /// must UNION the leftover entries into the existing v3 value, not overwrite
616    /// it — no locations may be lost.
617    #[test]
618    fn migrate_2_to_3_unions_same_hash_mixed_state() {
619        let backend = crate::backend::in_memory::InMemoryBackend::open().unwrap();
620        let tx = h256(0x42);
621
622        // Pre-existing v3 value for `tx` (one block already migrated).
623        {
624            let v3_value: Vec<(BlockNumber, BlockHash, Index)> = vec![(100, h256(0x10), 0)];
625            let mut batch = backend.begin_write().unwrap();
626            batch
627                .merge(
628                    TRANSACTION_LOCATIONS,
629                    tx.as_bytes(),
630                    &v3_value.encode_to_vec(),
631                )
632                .unwrap();
633            batch.commit().unwrap();
634        }
635        // Leftover v2 composite entries for the SAME tx (different blocks).
636        seed_old_entry(&backend, tx, 101, h256(0x11), 3);
637        seed_old_entry(&backend, tx, 102, h256(0x12), 7);
638
639        migrate_2_to_3(&backend).unwrap();
640
641        let mut got = read_new_entry(&backend, tx).unwrap();
642        got.sort();
643        let mut expected = vec![
644            (100u64, h256(0x10), 0u64), // pre-existing v3 entry survives
645            (101u64, h256(0x11), 3u64),
646            (102u64, h256(0x12), 7u64),
647        ];
648        expected.sort();
649        assert_eq!(
650            got, expected,
651            "merge must union, not overwrite, on mixed state"
652        );
653    }
654}