feat(database): Add metadata table for upgradability and avoiding database switchups (#387)

- closes #357


  Introduce the `metadata` table where we track the currently used database schema. This will later allow us to implement "migrations", i.e. updating the database schema without requiring the user to resync.

Furthermore, when we create the database, we store the domain type used in order to avoid working with a database that exists, but was previously used on another network.

The existing `block` table is removed and it's column moved to `metadata` in order to have only one table with data that exists only once.

Author: dknopik

Cargo.lock

@@ -187,9 +187,14 @@ impl Client {
                 NetworkDatabase::new_as_impostor(
                     &config.global_config.data_dir.database_file(),
                     impostor,
+                    config.global_config.ssv_network.ssv_domain_type,
                 )
             } else {
-                NetworkDatabase::new(&config.global_config.data_dir.database_file(), &pubkey)
+                NetworkDatabase::new(
+                    &config.global_config.data_dir.database_file(),
+                    &pubkey,
+                    config.global_config.ssv_network.ssv_domain_type,
+                )
             }
             .map_err(|e| format!("Unable to open Anchor database: {e}"))?,
         );
@@ -429,7 +434,7 @@ impl Client {
         let signature_collector = SignatureCollectorManager::new(
             processor_senders.clone(),
             operator_id.clone(),
-            config.global_config.ssv_network.ssv_domain_type.clone(),
+            config.global_config.ssv_network.ssv_domain_type,
             message_sender.clone(),
             slot_clock.clone(),
         )
@@ -441,7 +446,7 @@ impl Client {
             operator_id.clone(),
             slot_clock.clone(),
             message_sender,
-            config.global_config.ssv_network.ssv_domain_type.clone(),
+            config.global_config.ssv_network.ssv_domain_type,
         )
         .map_err(|e| format!("Unable to initialize qbft manager: {e:?}"))?;
 

anchor/client/src/lib.rs

@@ -1,6 +1,11 @@
 use std::str::FromStr;
 
-#[derive(Clone, Debug, Default, PartialEq)]
+use rusqlite::{
+    ToSql,
+    types::{FromSql, FromSqlError, FromSqlResult, ToSqlOutput, Value, ValueRef},
+};
+
+#[derive(Clone, Copy, Debug, Default, PartialEq)]
 pub struct DomainType(pub [u8; 4]);
 
 impl FromStr for DomainType {
@@ -28,3 +33,19 @@ impl From<[u8; 4]> for DomainType {
         Self(bytes)
     }
 }
+
+impl FromSql for DomainType {
+    fn column_result(value: ValueRef<'_>) -> FromSqlResult<Self> {
+        let value = value.as_i64()?;
+        let value = u32::try_from(value).map_err(|_| FromSqlError::InvalidType)?;
+        Ok(value.to_le_bytes().into())
+    }
+}
+
+impl ToSql for DomainType {
+    fn to_sql(&self) -> rusqlite::Result<ToSqlOutput<'_>> {
+        Ok(ToSqlOutput::Owned(Value::Integer(
+            u32::from_le_bytes(self.0).into(),
+        )))
+    }
+}

anchor/common/ssv_types/src/domain_type.rs

@@ -1,6 +1,5 @@
 use std::{
     collections::{HashMap, HashSet},
-    fs::File,
     path::Path,
     time::Duration,
 };
@@ -10,7 +9,10 @@ use once_cell::sync::OnceCell;
 use openssl::{pkey::Public, rsa::Rsa};
 use r2d2_sqlite::SqliteConnectionManager;
 use rusqlite::{Transaction, params};
-use ssv_types::{Cluster, ClusterId, CommitteeId, Operator, OperatorId, Share, ValidatorMetadata};
+use ssv_types::{
+    Cluster, ClusterId, CommitteeId, Operator, OperatorId, Share, ValidatorMetadata,
+    domain_type::DomainType,
+};
 use tokio::sync::{
     watch,
     watch::{Receiver, Ref},
@@ -76,6 +78,7 @@ mod cluster_operations;
 mod error;
 mod keysplit_operations;
 mod operator_operations;
+mod schema;
 mod share_operations;
 mod sql_operations;
 mod state;
@@ -140,8 +143,12 @@ pub struct NetworkDatabase {
 
 impl NetworkDatabase {
     /// Construct a new NetworkDatabase at the given path and the Public Key of the current operator
-    pub fn new(path: &Path, pubkey: &Rsa<Public>) -> Result<Self, DatabaseError> {
-        let conn_pool = Self::open_or_create(path)?;
+    pub fn new(
+        path: &Path,
+        pubkey: &Rsa<Public>,
+        domain: DomainType,
+    ) -> Result<Self, DatabaseError> {
+        let conn_pool = Self::open_or_create(path, domain)?;
         let operator = PubkeyOrId::Pubkey(pubkey.clone());
         let state = watch::Sender::new(NetworkState::new_with_state(&conn_pool, &operator)?);
         Ok(Self {
@@ -152,8 +159,12 @@ impl NetworkDatabase {
     }
 
     /// Act as if we had the pubkey of a certain operator
-    pub fn new_as_impostor(path: &Path, operator: &OperatorId) -> Result<Self, DatabaseError> {
-        let conn_pool = Self::open_or_create(path)?;
+    pub fn new_as_impostor(
+        path: &Path,
+        operator: &OperatorId,
+        domain: DomainType,
+    ) -> Result<Self, DatabaseError> {
+        let conn_pool = Self::open_or_create(path, domain)?;
         let operator = PubkeyOrId::Id(*operator);
         let state = watch::Sender::new(NetworkState::new_with_state(&conn_pool, &operator)?);
         Ok(Self {
@@ -186,12 +197,9 @@ impl NetworkDatabase {
     }
 
     // Open an existing database at the given `path`, or create one if none exists.
-    fn open_or_create(path: &Path) -> Result<Pool, DatabaseError> {
-        if path.exists() {
-            Self::open_conn_pool(path)
-        } else {
-            Self::create(path)
-        }
+    fn open_or_create(path: &Path, domain: DomainType) -> Result<Pool, DatabaseError> {
+        schema::ensure_up_to_date(path, domain)?;
+        Self::open_conn_pool(path)
     }
 
     // Build a new connection pool
@@ -205,23 +213,6 @@ impl NetworkDatabase {
         Ok(conn_pool)
     }
 
-    // Create a database at the given path.
-    fn create(path: &Path) -> Result<Pool, DatabaseError> {
-        let _file = File::options()
-            .write(true)
-            .read(true)
-            .create_new(true)
-            .open(path)?;
-
-        // restrict file permissions
-        let conn_pool = Self::open_conn_pool(path)?;
-        let conn = conn_pool.get()?;
-
-        // create all of the tables
-        conn.execute_batch(include_str!("table_schema.sql"))?;
-        Ok(conn_pool)
-    }
-
     // Open a new connection
     pub fn connection(&self) -> Result<PoolConn, DatabaseError> {
         Ok(self.conn_pool.get()?)

anchor/database/src/lib.rs

@@ -0,0 +1,147 @@
+use std::path::Path;
+
+use rusqlite::{Connection, types::Value};
+use ssv_types::domain_type::DomainType;
+
+use crate::{DatabaseError, sql_operations};
+
+type SchemaVersion = u32;
+
+struct Metadata {
+    schema_version: SchemaVersion,
+    domain: DomainType,
+}
+
+enum UpgradeAction {
+    UpToDate,
+    // allow dead code until there are upgrade scripts
+    #[allow(dead_code)]
+    DoUpdate {
+        script: &'static str,
+        new_version: SchemaVersion,
+    },
+    Outdated,
+    Future,
+}
+
+enum DatabaseType {
+    /// If the Option is none, the database is from an older version of Anchor where we did not
+    /// track the schema version yet. We can change the type to "SchemaVersion" at some point and
+    /// treat older versions as "Unknown".
+    Anchor(Option<SchemaVersion>),
+    IncorrectDomain(DomainType),
+    Unknown,
+}
+
+/// Ensure that there is an up-to-date database available at `db_path`. Also check or set the
+/// domain type to ensure the database is for the correct network.
+pub fn ensure_up_to_date(
+    db_path: impl AsRef<Path>,
+    domain: DomainType,
+) -> Result<(), DatabaseError> {
+    let db_path = db_path.as_ref();
+    let is_new_file = !db_path.exists();
+    let conn = Connection::open(db_path)?;
+
+    let mut schema_version = if is_new_file {
+        create_initial_schema(&conn, domain)?
+    } else {
+        match determine_database_type(&conn, domain) {
+            DatabaseType::Anchor(schema_version) => schema_version,
+            DatabaseType::Unknown => {
+                // We do not know what this is. Let's be safe and error out.
+                return Err(DatabaseError::AlreadyPresent(
+                    "Unknown database schema".to_string(),
+                ));
+            }
+            DatabaseType::IncorrectDomain(domain) => {
+                return Err(DatabaseError::AlreadyPresent(format!(
+                    "Existing database for different network: {domain:?}"
+                )));
+            }
+        }
+    };
+
+    // Upgrade scripts are step by step, so we need to loop until we are up to date.
+    loop {
+        match get_upgrade_action(schema_version) {
+            UpgradeAction::UpToDate => return Ok(()),
+            UpgradeAction::DoUpdate {
+                script,
+                new_version,
+            } => {
+                conn.execute_batch(script)?;
+                schema_version = Some(new_version);
+            }
+            UpgradeAction::Outdated => {
+                return Err(DatabaseError::AlreadyPresent(
+                    "Database is outdated - please remove \"anchor_db.sqlite\" or use another data dir.".to_string(),
+                ));
+            }
+            UpgradeAction::Future => {
+                return Err(DatabaseError::AlreadyPresent(
+                    "Database schema is newer than supported by this version of Anchor".to_string(),
+                ));
+            }
+        }
+    }
+}
+
+fn determine_database_type(conn: &Connection, domain: DomainType) -> DatabaseType {
+    let result = conn.query_row(sql_operations::GET_METADATA, [], |row| {
+        Ok(Metadata {
+            schema_version: row.get("schema_version")?,
+            domain: row.get("domain_type")?,
+        })
+    });
+
+    match result {
+        Ok(metadata) => {
+            if metadata.domain == domain {
+                DatabaseType::Anchor(Some(metadata.schema_version))
+            } else {
+                DatabaseType::IncorrectDomain(metadata.domain)
+            }
+        }
+        Err(_) => {
+            // Something failed - this might be a non-Anchor or legacy Anchor database.
+            // To check, try to get the block from the old table before `metadata` was introduced.
+            let legacy = conn
+                .query_row(sql_operations::GET_LEGACY_BLOCK, [], |row| {
+                    // Check if there is the expected column and no further columns.
+                    Ok(
+                        row.get::<_, u64>("block_number").is_ok()
+                            && row.get::<_, Value>(1).is_err(),
+                    )
+                })
+                .unwrap_or(false);
+
+            if legacy {
+                DatabaseType::Anchor(None)
+            } else {
+                DatabaseType::Unknown
+            }
+        }
+    }
+}
+
+// Before release, update the return value of this function if the initial table schema was changed.
+fn create_initial_schema(
+    conn: &rusqlite::Connection,
+    domain: DomainType,
+) -> Result<Option<SchemaVersion>, DatabaseError> {
+    conn.execute_batch(include_str!("table_schema.sql"))?;
+    conn.execute(sql_operations::INSERT_METADATA, [&domain])?;
+    Ok(Some(0))
+}
+
+// Register upgrade scripts in this function and mark the current version. Define any versions for
+// which the schema is not upgradable as "Recreate" and all versions after the current version as
+// "Future".
+fn get_upgrade_action(version: Option<SchemaVersion>) -> UpgradeAction {
+    match version {
+        None => UpgradeAction::Outdated,
+        Some(0) => UpgradeAction::UpToDate,
+        Some(1..) => UpgradeAction::Future,
+    }
+}

anchor/database/src/schema.rs

@@ -1,3 +1,8 @@
+// Metadata
+pub const INSERT_METADATA: &str = r#"INSERT INTO metadata (domain_type) VALUES (?1)"#;
+pub const GET_METADATA: &str = r#"SELECT schema_version, domain_type FROM metadata"#;
+pub const GET_LEGACY_BLOCK: &str = r#"SELECT * FROM block"#;
+
 // Operator
 pub const INSERT_OPERATOR: &str = r#"
     INSERT INTO operators 
@@ -81,8 +86,8 @@ pub const SET_INDEX: &str = r#"
 "#;
 
 // Blocks
-pub const UPDATE_BLOCK_NUMBER: &str = r#"UPDATE block SET block_number = ?1"#;
-pub const GET_BLOCK_NUMBER: &str = r#"SELECT block_number FROM block"#;
+pub const UPDATE_BLOCK_NUMBER: &str = r#"UPDATE metadata SET block_number = ?1"#;
+pub const GET_BLOCK_NUMBER: &str = r#"SELECT block_number FROM metadata"#;
 
 // Nonce
 pub const GET_ALL_NONCES: &str = r#"SELECT owner, nonce FROM owners"#;

anchor/database/src/sql_operations.rs

@@ -1,7 +1,17 @@
-CREATE TABLE block (
+-- SCHEMA VERSION 0
+
+-- we should avoid removing columns from this to keep compatibility between anchor Versions
+CREATE TABLE metadata (
+    schema_version INTEGER NOT NULL DEFAULT 0, 
+    domain_type INTEGER NOT NULL,
     block_number INTEGER NOT NULL DEFAULT 0 CHECK (block_number >= 0)
 );
-INSERT INTO block (block_number) VALUES (0);
+CREATE TRIGGER unique_metadata
+    BEFORE INSERT ON metadata
+    WHEN (SELECT COUNT(*) FROM metadata) >= 1
+BEGIN
+    SELECT RAISE(FAIL, 'we can only have one metadata row');
+END;
 
 CREATE TABLE owners (
     owner TEXT PRIMARY KEY NOT NULL,