Merge pull request #6271 from nymtech/bugfix/data-observatory

Fix migrations in the Data Observatory

Cargo.lock

@@ -6302,7 +6302,7 @@ dependencies = [
 
 [[package]]
 name = "nym-data-observatory"
-version = "1.0.0"
+version = "1.0.1"
 dependencies = [
  "anyhow",
  "async-trait",

common/nyxd-scraper-psql/src/storage/block_storage.rs

@@ -13,7 +13,7 @@ use nyxd_scraper_shared::storage::helpers::log_db_operation_time;
 use nyxd_scraper_shared::storage::{NyxdScraperStorage, NyxdScraperStorageError};
 use sqlx::types::time::{OffsetDateTime, PrimitiveDateTime};
 use tokio::time::Instant;
-use tracing::{debug, error, info, instrument, warn};
+use tracing::{debug, error, info, instrument};
 
 #[derive(Clone)]
 pub struct PostgresScraperStorage {
@@ -22,7 +22,10 @@ pub struct PostgresScraperStorage {
 
 impl PostgresScraperStorage {
     #[instrument]
-    pub async fn init(connection_string: &str) -> Result<Self, PostgresScraperError> {
+    pub async fn init(
+        connection_string: &str,
+        run_migrations: &bool,
+    ) -> Result<Self, PostgresScraperError> {
         debug!("initialising scraper database with '{connection_string}'",);
 
         let connection_pool = match sqlx::PgPool::connect(connection_string).await {
@@ -33,12 +36,13 @@ impl PostgresScraperStorage {
             }
         };
 
-        if let Err(err) = sqlx::migrate!("./sql_migrations")
-            .run(&connection_pool)
-            .await
-        {
-            warn!("Failed to initialize SQLx database: {err}");
-            // return Err(err.into());
+        if *run_migrations {
+            if let Err(err) = sqlx::migrate!("./sql_migrations")
+                .run(&connection_pool)
+                .await
+            {
+                return Err(err.into());
+            }
         }
 
         info!("Database migration finished!");
@@ -192,8 +196,11 @@ impl PostgresScraperStorage {
 impl NyxdScraperStorage for PostgresScraperStorage {
     type StorageTransaction = PostgresStorageTransaction;
 
-    async fn initialise(storage: &str) -> Result<Self, NyxdScraperStorageError> {
-        PostgresScraperStorage::init(storage)
+    async fn initialise(
+        storage: &str,
+        run_migrations: &bool,
+    ) -> Result<Self, NyxdScraperStorageError> {
+        PostgresScraperStorage::init(storage, run_migrations)
             .await
             .map_err(NyxdScraperStorageError::from)
     }

common/nyxd-scraper-shared/src/scraper/mod.rs

@@ -48,6 +48,8 @@ pub struct Config {
     pub store_precommits: bool,
 
     pub start_block: StartingBlockOpts,
+
+    pub run_migrations: bool,
 }
 
 pub struct NyxdScraperBuilder<S> {
@@ -161,7 +163,7 @@ where
 
     pub async fn new(config: Config) -> Result<Self, ScraperError> {
         config.pruning_options.validate()?;
-        let storage = S::initialise(&config.database_storage).await?;
+        let storage = S::initialise(&config.database_storage, &config.run_migrations).await?;
         let rpc_client = RpcClient::new(&config.rpc_url)?;
 
         Ok(NyxdScraper {

common/nyxd-scraper-shared/src/storage/mod.rs

@@ -33,7 +33,10 @@ pub trait NyxdScraperStorage: Clone + Sized {
     type StorageTransaction: NyxdScraperTransaction;
 
     /// Either connection string (postgres) or storage path (sqlite)
-    async fn initialise(storage: &str) -> Result<Self, NyxdScraperStorageError>;
+    async fn initialise(
+        storage: &str,
+        run_migrations: &bool,
+    ) -> Result<Self, NyxdScraperStorageError>;
 
     async fn begin_processing_tx(
         &self,

common/nyxd-scraper-sqlite/src/storage/block_storage.rs

@@ -207,7 +207,10 @@ impl SqliteScraperStorage {
 impl NyxdScraperStorage for SqliteScraperStorage {
     type StorageTransaction = SqliteStorageTransaction;
 
-    async fn initialise(storage: &str) -> Result<Self, NyxdScraperStorageError> {
+    async fn initialise(
+        storage: &str,
+        _run_migrations: &bool,
+    ) -> Result<Self, NyxdScraperStorageError> {
         SqliteScraperStorage::init(storage)
             .await
             .map_err(NyxdScraperStorageError::from)

nym-data-observatory/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "nym-data-observatory"
-version = "1.0.0"
+version = "1.0.1"
 authors.workspace = true
 repository.workspace = true
 homepage.workspace = true

nym-data-observatory/src/chain_scraper/mod.rs

@@ -1,8 +1,7 @@
 use crate::cli::commands::run::Args;
 use crate::db::DbPool;
 use nyxd_scraper_psql::{PostgresNyxdScraper, PruningOptions};
-use std::fs;
-use tracing::{info, warn};
+use tracing::info;
 
 pub(crate) mod webhook;
 
@@ -13,11 +12,6 @@ pub(crate) async fn run_chain_scraper(
 ) -> anyhow::Result<PostgresNyxdScraper> {
     let use_best_effort_start_height = args.start_block_height.is_some();
 
-    if args.nuke_db {
-        warn!("☢️☢️☢️ NUKING THE SCRAPER DATABASE");
-        fs::remove_file(config.chain_scraper_connection_string())?;
-    }
-
     let database_storage = config
         .chain_scraper_connection_string
         .clone()
@@ -34,6 +28,7 @@ pub(crate) async fn run_chain_scraper(
             start_block_height: args.start_block_height,
             use_best_effort_start_height,
         },
+        run_migrations: false, // ignore the base migrations
     })
     .with_msg_module(crate::modules::wasm::WasmModule::new(connection_pool))
     .with_tx_module(webhook::WebhookModule::new(config.clone())?);

nym-data-observatory/src/cli/commands/run/args.rs

@@ -15,9 +15,6 @@ pub(crate) struct Args {
     #[arg(long, env = NYXD_SCRAPER_START_HEIGHT)]
     pub(crate) start_block_height: Option<u32>,
 
-    #[arg(long, env = NYXD_SCRAPER_UNSAFE_NUKE_DB, default_value = "false")]
-    pub(crate) nuke_db: bool,
-
     /// (Override) Postgres connection string for chain scraper history
     #[arg(long, env = NYM_DATA_OBSERVATORY_DB_URL, alias = "db_url")]
     pub(crate) db_connection_string: Option<String>,

nym-data-observatory/src/cli/commands/run/mod.rs

@@ -123,7 +123,6 @@ pub(crate) async fn execute(args: Args, http_port: u16) -> Result<(), NymDataObs
             w.watch_for_chain_message_types
         );
     }
-    info!("nuke_db: {}", args.nuke_db);
 
     let storage = db::Storage::init(db_connection_string).await?;
     let watcher_pool = storage.pool_owned();

nym-data-observatory/src/db/mod.rs

@@ -1,6 +1,7 @@
 use anyhow::{Result, anyhow};
-use sqlx::{Postgres, postgres::PgConnectOptions};
+use sqlx::{Postgres, migrate::Migrator, postgres::PgConnectOptions};
 use std::str::FromStr;
+use tracing::info;
 
 pub(crate) mod models;
 pub(crate) mod queries {
@@ -8,6 +9,8 @@ pub(crate) mod queries {
     pub mod wasm;
 }
 
+static MIGRATOR: Migrator = sqlx::migrate!("./migrations");
+
 pub(crate) type DbPool = sqlx::Pool<Postgres>;
 
 pub(crate) struct Storage {
@@ -22,6 +25,13 @@ impl Storage {
             .await
             .map_err(|err| anyhow!("Failed to connect to {}: {}", &connection_url, err))?;
 
+        MIGRATOR
+            .run(&pool)
+            .await
+            .map_err(|err| anyhow!("Failed to run migrations: {}", err))?;
+
+        info!("✅ Successfully migrated the database");
+
         Ok(Storage { pool })
     }
 

nym-data-observatory/src/env.rs

@@ -16,8 +16,6 @@ pub mod vars {
     pub const NYXD_SCRAPER_USE_BEST_EFFORT_START_HEIGHT: &str =
         "NYXD_SCRAPER_USE_BEST_EFFORT_START_HEIGHT";
 
-    pub const NYXD_SCRAPER_UNSAFE_NUKE_DB: &str = "NYXD_SCRAPER_UNSAFE_NUKE_DB";
-
     pub const NYM_DATA_OBSERVATORY_ID_ARG: &str = "NYM_DATA_OBSERVATORY_ID";
     pub const NYM_DATA_OBSERVATORY_OUTPUT_ARG: &str = "NYM_DATA_OBSERVATORY_OUTPUT";
 

nym-validator-rewarder/src/config/mod.rs

@@ -135,6 +135,7 @@ impl Config {
                 start_block_height: None,
                 use_best_effort_start_height: true,
             },
+            run_migrations: true,
         })
     }
 

nyx-chain-watcher/src/chain_scraper/mod.rs

@@ -60,6 +60,7 @@ pub(crate) async fn run_chain_scraper(
             start_block_height,
             use_best_effort_start_height,
         },
+        run_migrations: true,
     })
     .with_msg_module(BankScraperModule::new(
         db_pool,