Synchronise importer changes from signer to aggregator

Author: Alenar

mithril-aggregator/src/database/cardano_transaction_migration.rs

@@ -57,6 +57,19 @@ vacuum;
             4,
             r#"
 create index block_number_index on cardano_tx(block_number);
+"#,
+        ),
+        // Migration 5
+        // Add `block_range_root` table
+        SqlMigration::new(
+            5,
+            r#"
+create table block_range_root (
+    start         integer   not null,
+    end           integer   not null,
+    merkle_root   text      not null,
+    primary key (start, end)
+);
 "#,
         ),
     ]

mithril-aggregator/src/database/provider/block_range_root/get_block_range_root.rs

@@ -0,0 +1,34 @@
+use mithril_persistence::sqlite::{Provider, SourceAlias, SqLiteEntity, SqliteConnection};
+
+use crate::database::record::BlockRangeRootRecord;
+
+/// Simple queries to retrieve [BlockRangeRootRecord] from the sqlite database.
+pub struct GetBlockRangeRootProvider<'client> {
+    connection: &'client SqliteConnection,
+}
+
+impl<'client> GetBlockRangeRootProvider<'client> {
+    #[cfg(test)]
+    /// Create a new instance
+    pub fn new(connection: &'client SqliteConnection) -> Self {
+        Self { connection }
+    }
+}
+
+#[cfg(test)]
+impl mithril_persistence::sqlite::GetAllCondition for GetBlockRangeRootProvider<'_> {}
+
+impl<'client> Provider<'client> for GetBlockRangeRootProvider<'client> {
+    type Entity = BlockRangeRootRecord;
+
+    fn get_connection(&'client self) -> &'client SqliteConnection {
+        self.connection
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        let aliases = SourceAlias::new(&[("{:block_range_root:}", "block_range_root")]);
+        let projection = Self::Entity::get_projection().expand(aliases);
+
+        format!("select {projection} from block_range_root where {condition} order by rowid")
+    }
+}

mithril-aggregator/src/database/provider/block_range_root/get_interval_without_block_range_provider.rs

@@ -0,0 +1,43 @@
+use mithril_persistence::sqlite::{
+    Provider, SourceAlias, SqLiteEntity, SqliteConnection, WhereCondition,
+};
+
+use crate::database::record::IntervalWithoutBlockRangeRootRecord;
+
+/// Query that return the interval of block numbers that does not have a block range root.
+pub struct GetIntervalWithoutBlockRangeRootProvider<'client> {
+    connection: &'client SqliteConnection,
+}
+
+impl<'client> GetIntervalWithoutBlockRangeRootProvider<'client> {
+    /// Create a new instance
+    pub fn new(connection: &'client SqliteConnection) -> Self {
+        Self { connection }
+    }
+
+    pub fn get_interval_without_block_range_condition(&self) -> WhereCondition {
+        WhereCondition::default()
+    }
+}
+
+impl<'client> Provider<'client> for GetIntervalWithoutBlockRangeRootProvider<'client> {
+    type Entity = IntervalWithoutBlockRangeRootRecord;
+
+    fn get_connection(&'client self) -> &'client SqliteConnection {
+        self.connection
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        let aliases = SourceAlias::new(&[
+            ("{:interval_start:}", "interval_start"),
+            ("{:interval_end:}", "interval_end"),
+        ]);
+        let projection = Self::Entity::get_projection().expand(aliases);
+
+        format!(
+            "select {projection} from \
+        (select max(end) as start from block_range_root where {condition}) as interval_start, \
+        (select max(block_number) as end from cardano_tx where {condition}) as interval_end"
+        )
+    }
+}

mithril-aggregator/src/database/provider/block_range_root/insert_block_range.rs

@@ -0,0 +1,65 @@
+use std::iter::repeat;
+
+use sqlite::Value;
+
+use mithril_common::StdResult;
+use mithril_persistence::sqlite::{
+    Provider, SourceAlias, SqLiteEntity, SqliteConnection, WhereCondition,
+};
+
+use crate::database::record::BlockRangeRootRecord;
+
+/// Query to insert [BlockRangeRootRecord] in the sqlite database
+pub struct InsertBlockRangeRootProvider<'client> {
+    connection: &'client SqliteConnection,
+}
+
+impl<'client> InsertBlockRangeRootProvider<'client> {
+    /// Create a new instance
+    pub fn new(connection: &'client SqliteConnection) -> Self {
+        Self { connection }
+    }
+
+    /// Condition to insert multiples records.
+    pub fn get_insert_many_condition(
+        &self,
+        block_range_records: Vec<BlockRangeRootRecord>,
+    ) -> StdResult<WhereCondition> {
+        let columns = "(start, end, merkle_root)";
+        let values_columns: Vec<&str> = repeat("(?*, ?*, ?*)")
+            .take(block_range_records.len())
+            .collect();
+
+        let values: StdResult<Vec<Value>> =
+            block_range_records
+                .into_iter()
+                .try_fold(vec![], |mut vec, record| {
+                    vec.append(&mut vec![
+                        Value::Integer(record.range.start.try_into()?),
+                        Value::Integer(record.range.end.try_into()?),
+                        Value::String(record.merkle_root.to_hex()),
+                    ]);
+                    Ok(vec)
+                });
+
+        Ok(WhereCondition::new(
+            format!("{columns} values {}", values_columns.join(", ")).as_str(),
+            values?,
+        ))
+    }
+}
+
+impl<'client> Provider<'client> for InsertBlockRangeRootProvider<'client> {
+    type Entity = BlockRangeRootRecord;
+
+    fn get_connection(&'client self) -> &'client SqliteConnection {
+        self.connection
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        let aliases = SourceAlias::new(&[("{:block_range_root:}", "block_range_root")]);
+        let projection = Self::Entity::get_projection().expand(aliases);
+
+        format!("insert or ignore into block_range_root {condition} returning {projection}")
+    }
+}

mithril-aggregator/src/database/provider/block_range_root/mod.rs

@@ -0,0 +1,8 @@
+mod get_block_range_root;
+mod get_interval_without_block_range_provider;
+mod insert_block_range;
+
+#[cfg(test)]
+pub use get_block_range_root::*;
+pub use get_interval_without_block_range_provider::*;
+pub use insert_block_range::*;

mithril-aggregator/src/database/provider/cardano_transaction/get_cardano_transaction.rs

@@ -1,6 +1,8 @@
+use std::ops::Range;
+
 use sqlite::Value;
 
-use mithril_common::entities::{ImmutableFileNumber, TransactionHash};
+use mithril_common::entities::{BlockNumber, ImmutableFileNumber, TransactionHash};
 use mithril_persistence::sqlite::{
     Provider, SourceAlias, SqLiteEntity, SqliteConnection, WhereCondition,
 };
@@ -47,6 +49,20 @@ impl<'client> GetCardanoTransactionProvider<'client> {
             vec![Value::Integer(beacon as i64)],
         )
     }
+
+    pub fn get_transaction_between_blocks_condition(
+        &self,
+        range: Range<BlockNumber>,
+    ) -> WhereCondition {
+        WhereCondition::new(
+            "block_number >= ?*",
+            vec![Value::Integer(range.start as i64)],
+        )
+        .and_where(WhereCondition::new(
+            "block_number < ?*",
+            vec![Value::Integer(range.end as i64)],
+        ))
+    }
 }
 
 impl<'client> Provider<'client> for GetCardanoTransactionProvider<'client> {

mithril-aggregator/src/database/provider/mod.rs

@@ -1,4 +1,5 @@
 //! Aggregator related database providers
+mod block_range_root;
 mod cardano_transaction;
 mod certificate;
 mod epoch_setting;
@@ -9,6 +10,7 @@ mod signer_registration;
 mod single_signature;
 mod stake_pool;
 
+pub use block_range_root::*;
 pub use cardano_transaction::*;
 pub use certificate::*;
 pub use epoch_setting::*;

mithril-aggregator/src/database/record/block_range_root.rs

@@ -0,0 +1,61 @@
+use sqlite::Row;
+
+use mithril_common::crypto_helper::MKTreeNode;
+use mithril_common::entities::BlockRange;
+use mithril_persistence::sqlite::{HydrationError, Projection, SqLiteEntity};
+
+use crate::database::record::hydrator::try_to_u64;
+
+/// Block range root record is the representation of block range with its merkle root precomputed.
+#[derive(Debug, PartialEq, Clone)]
+pub struct BlockRangeRootRecord {
+    /// Range of block numbers covered
+    pub range: BlockRange,
+    /// Merkle root of the block range, computed from the list of all transactions that are
+    /// included in the range
+    pub merkle_root: MKTreeNode,
+}
+
+impl From<(BlockRange, MKTreeNode)> for BlockRangeRootRecord {
+    fn from(value: (BlockRange, MKTreeNode)) -> Self {
+        Self {
+            range: value.0,
+            merkle_root: value.1,
+        }
+    }
+}
+
+impl From<BlockRangeRootRecord> for (BlockRange, MKTreeNode) {
+    fn from(value: BlockRangeRootRecord) -> Self {
+        (value.range, value.merkle_root)
+    }
+}
+
+impl SqLiteEntity for BlockRangeRootRecord {
+    fn hydrate(row: Row) -> Result<Self, HydrationError>
+    where
+        Self: Sized,
+    {
+        let start = try_to_u64("block_range.start", row.read::<i64, _>(0))?;
+        let _end = try_to_u64("block_range.end", row.read::<i64, _>(1))?;
+        let merkle_root = row.read::<&str, _>(2);
+
+        Ok(Self {
+            range: BlockRange::from_block_number(start),
+            merkle_root: MKTreeNode::from_hex(merkle_root)
+                .map_err(|e| HydrationError::InvalidData(
+                    format!(
+                        "Field block_range.merkle_root (value={merkle_root}) is incompatible with hex representation. Error = {e}")
+                )
+            )?,
+        })
+    }
+
+    fn get_projection() -> Projection {
+        Projection::from(&[
+            ("start", "{:block_range_root:}.start", "int"),
+            ("end", "{:block_range_root:}.end", "int"),
+            ("merkle_root", "{:block_range_root:}.merkle_root", "text"),
+        ])
+    }
+}

mithril-aggregator/src/database/record/certificate.rs

@@ -17,6 +17,8 @@ use mithril_persistence::{
     sqlite::{HydrationError, Projection, SqLiteEntity},
 };
 
+use crate::database::record::hydrator;
+
 era_deprecate!("Remove immutable_file_number");
 /// Certificate record is the representation of a stored certificate.
 #[derive(Debug, PartialEq, Clone)]
@@ -312,7 +314,7 @@ impl SqLiteEntity for CertificateRecord {
         let network = row.read::<&str, _>(6).to_string();
         let immutable_file_number = row.read::<i64, _>(7);
         let signed_entity_type_id = row.read::<i64, _>(8);
-        let signed_entity_beacon_string = super::read_signed_entity_beacon_column(&row, 9);
+        let signed_entity_beacon_string = hydrator::read_signed_entity_beacon_column(&row, 9);
         let protocol_version = row.read::<&str, _>(10).to_string();
         let protocol_parameters_string = row.read::<&str, _>(11);
         let protocol_message_string = row.read::<&str, _>(12);

mithril-aggregator/src/database/record/interval_without_block_range_root.rs

@@ -0,0 +1,39 @@
+use sqlite::Row;
+
+use mithril_common::entities::BlockNumber;
+use mithril_persistence::sqlite::{HydrationError, Projection, SqLiteEntity};
+
+use crate::database::record::hydrator::try_to_u64;
+
+/// Interval of block numbers without block ranges root.
+pub struct IntervalWithoutBlockRangeRootRecord {
+    /// Start of the interval
+    pub start: Option<BlockNumber>,
+    /// End of the interval
+    pub end: Option<BlockNumber>,
+}
+
+impl SqLiteEntity for IntervalWithoutBlockRangeRootRecord {
+    fn hydrate(row: Row) -> Result<Self, HydrationError>
+    where
+        Self: Sized,
+    {
+        let start = row
+            .read::<Option<i64>, _>(0)
+            .map(|v| try_to_u64("interval_start.start", v))
+            .transpose()?;
+        let end = row
+            .read::<Option<i64>, _>(1)
+            .map(|v| try_to_u64("interval_end.end", v))
+            .transpose()?;
+
+        Ok(Self { start, end })
+    }
+
+    fn get_projection() -> Projection {
+        Projection::from(&[
+            ("start", "{:interval_start:}.start", "int"),
+            ("end", "{:interval_end:}.end", "int"),
+        ])
+    }
+}