Merge pull request #1934 from input-output-hk/ensemble/1900/aggregato…

…r_signatures_buffer

Aggregator single signatures buffer

CHANGELOG.md

@@ -11,6 +11,8 @@ As a minor extension, we have adopted a slightly different versioning convention
 
 - Optimizations of the state machine used by the signer to create individual signatures.
 
+- Support for buffering of incoming single signatures by the aggregator if it can not aggregate them yet
+
 - Crates versions:
 
 | Crate | Version |

mithril-aggregator/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-aggregator"
-version = "0.5.63"
+version = "0.5.64"
 description = "A Mithril Aggregator server"
 authors = { workspace = true }
 edition = { workspace = true }

mithril-aggregator/src/database/migration.rs

@@ -758,6 +758,23 @@ pragma foreign_keys=true;
             26,
             r#"
 create unique index signed_entity_unique_index on signed_entity(signed_entity_type_id, beacon);
+"#,
+        ),
+        // Migration 27
+        SqlMigration::new(
+            27,
+            r#"
+create table buffered_single_signature (
+    signed_entity_type_id           integer     not null,
+    party_id                        text        not null,
+    lottery_indexes                 json        not null,
+    signature                       text        not null,
+    created_at                      text        not null,
+    primary key (signed_entity_type_id, party_id)
+);
+
+create index buffered_single_signature_signed_entity_type_id on buffered_single_signature(signed_entity_type_id);
+create index buffered_single_signature_party_id_index on buffered_single_signature(party_id);
 "#,
         ),
     ]

mithril-aggregator/src/database/query/buffered_single_signature/delete_buffered_single_signature.rs

@@ -0,0 +1,119 @@
+use sqlite::Value;
+
+use mithril_common::entities::{PartyId, SignedEntityTypeDiscriminants};
+use mithril_persistence::sqlite::{Query, SourceAlias, SqLiteEntity, WhereCondition};
+
+use crate::database::record::BufferedSingleSignatureRecord;
+
+/// Query to delete old [BufferedSingleSignatureRecord] from the sqlite database
+pub struct DeleteBufferedSingleSignatureQuery {
+    condition: WhereCondition,
+}
+
+impl DeleteBufferedSingleSignatureQuery {
+    pub fn by_discriminant_and_party_ids(
+        signed_entity_type_discriminant: SignedEntityTypeDiscriminants,
+        party_ids: Vec<PartyId>,
+    ) -> Self {
+        let ids_values = party_ids.into_iter().map(Value::String).collect();
+
+        Self {
+            condition: WhereCondition::new(
+                "signed_entity_type_id = ?*",
+                vec![Value::Integer(
+                    signed_entity_type_discriminant.index() as i64
+                )],
+            )
+            .and_where(WhereCondition::where_in("party_id", ids_values)),
+        }
+    }
+}
+
+impl Query for DeleteBufferedSingleSignatureQuery {
+    type Entity = BufferedSingleSignatureRecord;
+
+    fn filters(&self) -> WhereCondition {
+        self.condition.clone()
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        // it is important to alias the fields with the same name as the table
+        // since the table cannot be aliased in a RETURNING statement in SQLite.
+        let projection = Self::Entity::get_projection().expand(SourceAlias::new(&[(
+            "{:buffered_single_signature:}",
+            "buffered_single_signature",
+        )]));
+
+        format!("delete from buffered_single_signature where {condition} returning {projection}")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use mithril_common::entities::SignedEntityTypeDiscriminants::{
+        CardanoTransactions, MithrilStakeDistribution,
+    };
+    use mithril_persistence::sqlite::ConnectionExtensions;
+
+    use crate::database::query::GetBufferedSingleSignatureQuery;
+    use crate::database::record::strip_buffered_sigs_date;
+    use crate::database::test_helper::{insert_buffered_single_signatures, main_db_connection};
+
+    use super::*;
+
+    #[test]
+    fn test_delete_buffered_single_signature_records_by_discriminant_and_party_ids() {
+        let connection = main_db_connection().unwrap();
+        let records = BufferedSingleSignatureRecord::fakes(&[
+            ("party_1", MithrilStakeDistribution),
+            ("party_2", MithrilStakeDistribution),
+            ("party_3", MithrilStakeDistribution),
+            ("party_1", CardanoTransactions),
+            ("party_2", CardanoTransactions),
+        ]);
+        insert_buffered_single_signatures(&connection, records.clone()).unwrap();
+
+        let cursor = connection
+            .fetch(
+                DeleteBufferedSingleSignatureQuery::by_discriminant_and_party_ids(
+                    MithrilStakeDistribution,
+                    vec!["party_1".into(), "party_3".into()],
+                ),
+            )
+            .unwrap();
+        assert_eq!(2, cursor.count());
+
+        let remaining_records: Vec<BufferedSingleSignatureRecord> = connection
+            .fetch_collect(GetBufferedSingleSignatureQuery::all())
+            .unwrap();
+        assert_eq!(
+            strip_buffered_sigs_date(&BufferedSingleSignatureRecord::fakes(&[
+                ("party_2", CardanoTransactions),
+                ("party_1", CardanoTransactions),
+                ("party_2", MithrilStakeDistribution),
+            ])),
+            strip_buffered_sigs_date(&remaining_records)
+        );
+
+        let cursor = connection
+            .fetch(
+                DeleteBufferedSingleSignatureQuery::by_discriminant_and_party_ids(
+                    CardanoTransactions,
+                    vec!["party_1".into(), "party_2".into()],
+                ),
+            )
+            .unwrap();
+        assert_eq!(2, cursor.count());
+
+        let remaining_records: Vec<BufferedSingleSignatureRecord> = connection
+            .fetch_collect(GetBufferedSingleSignatureQuery::all())
+            .unwrap();
+        assert_eq!(
+            strip_buffered_sigs_date(&BufferedSingleSignatureRecord::fakes(&[(
+                "party_2",
+                MithrilStakeDistribution
+            ),])),
+            strip_buffered_sigs_date(&remaining_records)
+        );
+    }
+}

mithril-aggregator/src/database/query/buffered_single_signature/get_buffered_single_signature.rs

@@ -0,0 +1,119 @@
+use sqlite::Value;
+
+use mithril_common::entities::SignedEntityTypeDiscriminants;
+use mithril_persistence::sqlite::{Query, SourceAlias, SqLiteEntity, WhereCondition};
+
+use crate::database::record::BufferedSingleSignatureRecord;
+
+/// Simple queries to retrieve [BufferedSingleSignatureRecord] from the sqlite database.
+pub struct GetBufferedSingleSignatureQuery {
+    condition: WhereCondition,
+}
+
+impl GetBufferedSingleSignatureQuery {
+    #[cfg(test)]
+    pub(crate) fn all() -> Self {
+        Self {
+            condition: WhereCondition::default(),
+        }
+    }
+
+    pub fn by_discriminant(signed_entity_type_discriminant: SignedEntityTypeDiscriminants) -> Self {
+        Self {
+            condition: WhereCondition::new(
+                "signed_entity_type_id = ?*",
+                vec![Value::Integer(
+                    signed_entity_type_discriminant.index() as i64
+                )],
+            ),
+        }
+    }
+}
+
+impl Query for GetBufferedSingleSignatureQuery {
+    type Entity = BufferedSingleSignatureRecord;
+
+    fn filters(&self) -> WhereCondition {
+        self.condition.clone()
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        let aliases = SourceAlias::new(&[("{:buffered_single_signature:}", "b")]);
+        let projection = Self::Entity::get_projection().expand(aliases);
+        format!("select {projection} from buffered_single_signature as b where {condition} order by ROWID desc")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use mithril_common::entities::SignedEntityTypeDiscriminants::{
+        CardanoImmutableFilesFull, CardanoTransactions, MithrilStakeDistribution,
+    };
+    use mithril_persistence::sqlite::ConnectionExtensions;
+
+    use crate::database::test_helper::{insert_buffered_single_signatures, main_db_connection};
+
+    use super::*;
+
+    #[test]
+    fn test_get_all() {
+        let connection = main_db_connection().unwrap();
+        let records = BufferedSingleSignatureRecord::fakes(&[
+            ("party1", MithrilStakeDistribution),
+            ("party2", CardanoTransactions),
+            ("party3", MithrilStakeDistribution),
+        ]);
+        insert_buffered_single_signatures(&connection, records.clone()).unwrap();
+
+        let stored_records: Vec<BufferedSingleSignatureRecord> = connection
+            .fetch_collect(GetBufferedSingleSignatureQuery::all())
+            .unwrap();
+
+        assert_eq!(
+            records.into_iter().rev().collect::<Vec<_>>(),
+            stored_records
+        );
+    }
+
+    #[test]
+    fn test_get_buffered_single_signature_records_by_discriminant() {
+        let connection = main_db_connection().unwrap();
+        let msd_records = BufferedSingleSignatureRecord::fakes(&[
+            ("party1", MithrilStakeDistribution),
+            ("party2", MithrilStakeDistribution),
+        ]);
+        let ctx_records = BufferedSingleSignatureRecord::fakes(&[("party3", CardanoTransactions)]);
+        insert_buffered_single_signatures(
+            &connection,
+            [msd_records.clone(), ctx_records.clone()].concat(),
+        )
+        .unwrap();
+
+        let stored_msd_records: Vec<BufferedSingleSignatureRecord> = connection
+            .fetch_collect(GetBufferedSingleSignatureQuery::by_discriminant(
+                MithrilStakeDistribution,
+            ))
+            .unwrap();
+        assert_eq!(
+            msd_records.into_iter().rev().collect::<Vec<_>>(),
+            stored_msd_records
+        );
+
+        let stored_ctx_records: Vec<BufferedSingleSignatureRecord> = connection
+            .fetch_collect(GetBufferedSingleSignatureQuery::by_discriminant(
+                CardanoTransactions,
+            ))
+            .unwrap();
+        assert_eq!(
+            ctx_records.into_iter().rev().collect::<Vec<_>>(),
+            stored_ctx_records
+        );
+
+        let cursor = connection
+            .fetch(GetBufferedSingleSignatureQuery::by_discriminant(
+                CardanoImmutableFilesFull,
+            ))
+            .unwrap();
+        assert_eq!(0, cursor.count());
+    }
+}