test fixtures

src/load.rs

@@ -1,7 +1,7 @@
 use crate::{
     extract_transactions::extract_current_transactions,
     load_tx_cypher::{self, BatchTxReturn},
-    queue,
+    queue::{self, clear_queue, push_queue_from_archive_map},
     scan::{ArchiveMap, ManifestInfo},
 };
 
@@ -10,7 +10,12 @@ use log::info;
 use neo4rs::Graph;
 
 /// takes all the archives from a map, and tries to load them sequentially
-pub async fn ingest_all(archive_map: &ArchiveMap, pool: &Graph) -> Result<()> {
+pub async fn ingest_all(archive_map: &ArchiveMap, pool: &Graph, force_queue: bool) -> Result<()> {
+    if force_queue {
+        clear_queue(pool).await?;
+        push_queue_from_archive_map(archive_map, pool).await?;
+    }
+
     // get queue of any batch which has any incomplete batches
     let pending = queue::get_queued(pool).await?;
     info!("pending archives: {}", pending.len());
@@ -23,11 +28,15 @@ pub async fn ingest_all(archive_map: &ArchiveMap, pool: &Graph) -> Result<()> {
         );
 
         if pending.contains(&m.archive_id) {
-            info!("load incomplete: {}", m.archive_dir.display());
+            info!("archive queued: {}", m.archive_dir.display());
+            let batch_tx_return = try_load_one_archive(m, pool).await?;
+            println!("SUCCESS: {}", batch_tx_return);
+        } else {
+            info!(
+                "archive complete (or not in queue): {}",
+                m.archive_dir.display()
+            );
         }
-
-        let batch_tx_return = try_load_one_archive(m, pool).await?;
-        println!("SUCCESS: {}", batch_tx_return);
     }
 
     Ok(())

src/queue.rs

@@ -1,6 +1,8 @@
 use anyhow::{bail, Context, Result};
 use neo4rs::Graph;
 
+use crate::scan::ArchiveMap;
+
 pub async fn update_task(
     pool: &Graph,
     archive_id: &str,
@@ -59,7 +61,8 @@ pub async fn is_complete(pool: &Graph, archive_id: &str, batch: usize) -> Result
     let cypher_string = format!(
         r#"
         MATCH (a:Queue {{ archive_id: "{}", batch: {} }})
-        RETURN DISTINCT a.completed
+        WHERE a.completed = false
+        RETURN COUNT(a) = 0 AS is_complete;
       "#,
         archive_id, batch
     );
@@ -73,14 +76,14 @@ pub async fn is_complete(pool: &Graph, archive_id: &str, batch: usize) -> Result
 
     if let Some(row) = res.next().await? {
         // Extract `archive_id` as a String
-        Ok(row.get::<bool>("a.completed").ok())
+        Ok(row.get::<bool>("is_complete").ok())
     } else {
         bail!("not found")
     }
 }
 
 // clear queue
-pub async fn clear(pool: &Graph) -> Result<()> {
+pub async fn clear_queue(pool: &Graph) -> Result<()> {
     let cypher_string = r#"
         MATCH (a:Queue)
         DELETE a
@@ -95,3 +98,11 @@ pub async fn clear(pool: &Graph) -> Result<()> {
         .context("execute query error")?;
     Ok(())
 }
+
+pub async fn push_queue_from_archive_map(map: &ArchiveMap, pool: &Graph) -> Result<()> {
+    for (_, a) in map.0.iter() {
+        // set at least one batch of each archive_id to false, so it gets picked up in the queue
+        update_task(pool, &a.archive_id, false, 0).await?;
+    }
+    Ok(())
+}

src/warehouse_cli.rs

@@ -25,6 +25,9 @@ pub struct WarehouseCli {
     #[clap(long, short('p'))]
     /// db password
     db_password: Option<String>,
+    #[clap(long, short('q'))]
+    /// force clear queue
+    clear_queue: bool,
 
     #[clap(subcommand)]
     command: Sub,
@@ -71,7 +74,7 @@ impl WarehouseCli {
                 let map = scan_dir_archive(start_path, archive_content.to_owned())?;
                 let pool = try_db_connection_pool(self).await?;
                 neo4j_init::maybe_create_indexes(&pool).await?;
-                ingest_all(&map, &pool).await?;
+                ingest_all(&map, &pool, self.clear_queue).await?;
             }
             Sub::LoadOne { archive_dir } => {
                 match scan_dir_archive(archive_dir, None)?.0.get(archive_dir) {

tests/fixtures/v5/state_ver_119757649.17a8/state.manifest

@@ -0,0 +1 @@
+{"version":119757649,"root_hash":"d2e6b6c474529cd3c3ea76cf211a528ee28bb13d7bcd8dcbef0ecf4c43bc3703","chunks":[{"first_idx":0,"last_idx":17338,"first_key":"000131122524ba9f4a13bd90a8b13c5d03ab621649c8100c6bbf7846fe8eaf0f","last_key":"fffed52c1dd93cc7aa5b79a8699df0c33eb4d65ab8d2a21bbe79f9d05ad62377","blobs":"state_ver_119757649.17a8/0-.chunk","proof":"state_ver_119757649.17a8/0-17338.proof"}],"proof":"state_ver_119757649.17a8/state.proof"}

tests/fixtures/v6/epoch_ending_79-.166d/epoch_ending.manifest.unzip

@@ -0,0 +1 @@
+{"first_epoch":79,"last_epoch":79,"waypoints":["33217173:632e6be8f7c0e6abccd101deba8aa109ea8ce36186978b5ae4656a510eddc5ea"],"chunks":[{"first_epoch":79,"last_epoch":79,"ledger_infos":"epoch_ending_79-.166d/79-.chunk"}]}

tests/fixtures/v6/state_epoch_54_ver_18120170.96e8/state.manifest

@@ -0,0 +1 @@
+{"version":18120170,"epoch":54,"root_hash":"ad97d4fda9cce3869b1753d1b6fd850d08134d123895d3845d87fced55a00b38","chunks":[{"first_idx":0,"last_idx":121418,"first_key":"0000acadcb83f9bdcffcbb63d132be250f8018a731137db7e3aab5af2e3bc8bd","last_key":"fffff7e8a7aff9e730227f0decbef95c00ccdd60744654b8c4799a80bec23e5a","blobs":"state_epoch_54_ver_18120170.96e8/0-.chunk","proof":"state_epoch_54_ver_18120170.96e8/0-121418.proof"}],"proof":"state_epoch_54_ver_18120170.96e8/state.proof"}

tests/fixtures/v6/state_epoch_79_ver_33217173.795d/state.manifest

@@ -0,0 +1 @@
+{"version":33217173,"epoch":79,"root_hash":"d5fc08dbe1756f9399be8edb26d194ec1c97442139849556bb9652a52b374995","chunks":[{"first_idx":0,"last_idx":141533,"first_key":"0000acadcb83f9bdcffcbb63d132be250f8018a731137db7e3aab5af2e3bc8bd","last_key":"fffff7e8a7aff9e730227f0decbef95c00ccdd60744654b8c4799a80bec23e5a","blobs":"state_epoch_79_ver_33217173.795d/0-.chunk","proof":"state_epoch_79_ver_33217173.795d/0-141533.proof"}],"proof":"state_epoch_79_ver_33217173.795d/state.proof"}

tests/fixtures/v6/transaction_9900001-.e469/transaction.manifest

@@ -0,0 +1 @@
+{"first_version":9900001,"last_version":10000000,"chunks":[{"first_version":9900001,"last_version":10000000,"transactions":"transaction_9900001-.e469/9900001-.chunk","proof":"transaction_9900001-.e469/9900001-10000000.proof"}]}

tests/fixtures/v7/epoch_ending_116-.be9b/epoch_ending.manifest

@@ -0,0 +1 @@
+{"first_epoch":116,"last_epoch":116,"waypoints":["38180075:98d7eb16747c1441350c4a1a6aa8b2d504c30ce2520807f2adc87d5a06d4f70f"],"chunks":[{"first_epoch":116,"last_epoch":116,"ledger_infos":"epoch_ending_116-.be9b/116-.chunk"}]}

tests/fixtures/v7/state_epoch_116_ver_38180075.05af/state.manifest

@@ -0,0 +1 @@
+{"version":38180075,"epoch":116,"root_hash":"b4c9918ddb62469cc3e7e7b2a01b43aeac803470913b3a89afdcc44078df8d8a","chunks":[{"first_idx":0,"last_idx":130241,"first_key":"0000acadcb83f9bdcffcbb63d132be250f8018a731137db7e3aab5af2e3bc8bd","last_key":"fffff7e8a7aff9e730227f0decbef95c00ccdd60744654b8c4799a80bec23e5a","blobs":"state_epoch_116_ver_38180075.05af/0-.chunk","proof":"state_epoch_116_ver_38180075.05af/0-130241.proof"}],"proof":"state_epoch_116_ver_38180075.05af/state.proof"}

tests/fixtures/v7/transaction_38100001-.541f/transaction.manifest

@@ -0,0 +1 @@
+{"first_version":38100001,"last_version":38200000,"chunks":[{"first_version":38100001,"last_version":38200000,"transactions":"transaction_38100001-.541f/38100001-.chunk","proof":"transaction_38100001-.541f/38100001-38200000.proof"}]}

tests/support/fixtures.rs

@@ -1,9 +1,23 @@
 use std::path::PathBuf;
 
+pub fn v5_fixtures_path() -> PathBuf {
+    let p = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+    p.join("tests/fixtures/v5")
+}
+
+pub fn v7_fixtures_path() -> PathBuf {
+    let p = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+    p.join("tests/fixtures/v7")
+}
+
+pub fn v7_fixtures_gzipped() -> PathBuf {
+    let p = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+    p.join("tests/fixtures/v7/transaction_38100001-.541f_gzipped")
+}
+
 pub fn v5_state_manifest_fixtures_path() -> PathBuf {
     let p = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
-    let project_root = p.parent().unwrap();
-    let dir = project_root.join("compatibility/fixtures/v5/state_ver_119757649.17a8");
+    let dir = p.join("tests/fixtures/v5/state_ver_119757649.17a8");
     assert!(
         &dir.exists(),
         "fixtures for backup archive cannot be found at path {}",
@@ -18,9 +32,9 @@ pub fn v7_state_manifest_fixtures_path() -> PathBuf {
         .canonicalize()
         .unwrap();
     assert!(&p.exists(), "not at the cargo manifest dir");
-    let project_root = p.parent().unwrap();
-    assert!(&project_root.exists(), "cannot find project root dir");
-    let dir = project_root.join("tools/storage/fixtures/v7/state_epoch_116_ver_38180075.05af");
+
+    assert!(&p.exists(), "cannot find project root dir");
+    let dir = p.join("tests/fixtures/v7/state_epoch_116_ver_38180075.05af");
     assert!(
         &dir.exists(),
         "fixtures for backup archive cannot be found at path {}",
@@ -34,9 +48,9 @@ pub fn v7_tx_manifest_fixtures_path() -> PathBuf {
         .canonicalize()
         .unwrap();
     assert!(&p.exists(), "not at the cargo manifest dir");
-    let project_root = p.parent().unwrap();
-    assert!(&project_root.exists(), "cannot find project root dir");
-    let dir = project_root.join("tools/storage/fixtures/v7/transaction_38100001-.541f");
+
+    assert!(&p.exists(), "cannot find project root dir");
+    let dir = p.join("tests/fixtures/v7/transaction_38100001-.541f");
     assert!(
         &dir.exists(),
         "fixtures for backup archive cannot be found at path {}",
@@ -50,9 +64,9 @@ pub fn v6_tx_manifest_fixtures_path() -> PathBuf {
         .canonicalize()
         .unwrap();
     assert!(&p.exists(), "not at the cargo manifest dir");
-    let project_root = p.parent().unwrap();
-    assert!(&project_root.exists(), "cannot find project root dir");
-    let dir = project_root.join("tools/storage/fixtures/v6/transaction_9900001-.e469");
+    let p = p.parent().unwrap();
+    assert!(&p.exists(), "cannot find project root dir");
+    let dir = p.join("tests/fixtures/v6/transaction_9900001-.e469");
     assert!(
         &dir.exists(),
         "fixtures for backup archive cannot be found at path {}",

tests/test_scan_dirs.rs

@@ -1,29 +1,13 @@
+mod support;
+
 use anyhow::Result;
 use libra_forensic_db::{scan::scan_dir_archive, unzip_temp::make_temp_unzipped};
-use std::path::PathBuf;
-
-fn v5_fixtures_path() -> PathBuf {
-    let p = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
-    let project_root = p.parent().unwrap();
-    project_root.join("compatibility/fixtures")
-}
-
-fn v7_fixtures_path() -> PathBuf {
-    let p = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
-    let project_root = p.parent().unwrap();
-    project_root.join("tools/storage/fixtures/v7")
-}
-
-fn v7_fixtures_gzipped() -> PathBuf {
-    let p = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
-    let project_root = p.parent().unwrap();
-    project_root.join("tools/storage/fixtures/v7/transaction_38100001-.541f_gzipped")
-}
+use support::fixtures;
 
 #[test]
 
 fn test_scan_dir_for_v5_manifests() -> Result<()> {
-    let start_here = v5_fixtures_path();
+    let start_here = fixtures::v5_fixtures_path();
 
     let s = scan_dir_archive(&start_here, None)?;
 
@@ -33,7 +17,7 @@ fn test_scan_dir_for_v5_manifests() -> Result<()> {
 
 #[test]
 fn test_scan_dir_for_v7_manifests() -> Result<()> {
-    let start_here = v7_fixtures_path();
+    let start_here = fixtures::v7_fixtures_path();
 
     let s = scan_dir_archive(&start_here, None)?;
 
@@ -45,7 +29,7 @@ fn test_scan_dir_for_v7_manifests() -> Result<()> {
 
 #[test]
 fn test_scan_dir_for_compressed_v7_manifests() -> Result<()> {
-    let start_here = v7_fixtures_gzipped();
+    let start_here = fixtures::v7_fixtures_gzipped();
 
     let archives = scan_dir_archive(&start_here, None)?;