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
18pub type MigrationFn = fn(backend: &dyn StorageBackend) -> Result<(), StoreError>;
23
24pub const MIGRATIONS: &[MigrationFn] = &[migrate_1_to_2, migrate_2_to_3, migrate_3_to_4];
34
35const _: () = assert!(
38 MIGRATIONS.len() == (STORE_SCHEMA_VERSION - 1) as usize,
39 "MIGRATIONS length must equal STORE_SCHEMA_VERSION - 1"
40);
41
42fn migration_for_version(version: u64) -> MigrationFn {
44 MIGRATIONS[(version - 1) as usize]
45}
46
47fn migrate_3_to_4(_backend: &dyn StorageBackend) -> Result<(), StoreError> {
57 Ok(())
58}
59
60const PROGRESS_LOG_INTERVAL: Duration = Duration::from_secs(10);
62
63fn 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
70pub 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 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
129fn 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
148fn 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 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
219fn 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 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 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 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 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 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 let backend = crate::backend::in_memory::InMemoryBackend::open().unwrap();
359 let temp_dir = tempfile::tempdir().unwrap();
360
361 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 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 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 {
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 {
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 migrate_1_to_2(&backend).unwrap();
446
447 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 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 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 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 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 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 let backend = crate::backend::in_memory::InMemoryBackend::open().unwrap();
569
570 {
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 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 assert_eq!(
592 read_new_entry(&backend, h256(1)).unwrap(),
593 vec![(100u64, h256(0x10), 0u64), (100u64, h256(0x11), 0u64)]
594 );
595
596 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 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 #[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 {
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 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), (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}