feat: audit entire sync of pre-merge chain

entity/src/lib.rs

@@ -16,5 +16,6 @@ pub mod key_value;
 pub mod node;
 pub mod record;
 pub mod state_roots;
+pub mod sync_audit;
 pub mod test;
 pub mod utils;

entity/src/sync_audit/mod.rs

@@ -0,0 +1,3 @@
+pub mod sync_audit;
+pub mod sync_audit_error;
+pub mod sync_audit_segment;

entity/src/sync_audit/sync_audit.rs

@@ -0,0 +1,62 @@
+//! `SeaORM` Entity for sync_audit
+use anyhow::Result;
+use chrono::{DateTime, Utc};
+use sea_orm::{entity::prelude::*, ActiveValue::NotSet, Set};
+
+#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
+#[sea_orm(table_name = "sync_audit")]
+pub struct Model {
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub started_at: DateTime<Utc>,
+    pub completed_at: Option<DateTime<Utc>>,
+    pub status: SyncAuditStatus,
+    pub segment_size: i32,
+}
+
+#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumIter, DeriveActiveEnum)]
+#[sea_orm(rs_type = "i32", db_type = "Integer")]
+pub enum SyncAuditStatus {
+    InProgress = 0,
+    Completed = 1,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(has_many = "super::sync_audit_segment::Entity")]
+    SyncAuditSegment,
+}
+
+impl ActiveModelBehavior for ActiveModel {}
+
+impl Related<super::sync_audit_segment::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::SyncAuditSegment.def()
+    }
+}
+
+pub async fn create(segment_size: i32, conn: &DatabaseConnection) -> Result<Model> {
+    let active = ActiveModel {
+        id: NotSet,
+        started_at: Set(Utc::now()),
+        completed_at: Set(None),
+        status: Set(SyncAuditStatus::InProgress),
+        segment_size: Set(segment_size),
+    };
+    Ok(active.insert(conn).await?)
+}
+
+pub async fn get(id: i32, conn: &DatabaseConnection) -> Result<Option<Model>> {
+    Ok(Entity::find_by_id(id).one(conn).await?)
+}
+
+pub async fn mark_complete(id: i32, conn: &DatabaseConnection) -> Result<()> {
+    let audit = get(id, conn)
+        .await?
+        .ok_or_else(|| anyhow::anyhow!("Audit not found"))?;
+    let mut audit: ActiveModel = audit.into();
+    audit.completed_at = Set(Some(Utc::now()));
+    audit.status = Set(SyncAuditStatus::Completed);
+    audit.update(conn).await?;
+    Ok(())
+}

entity/src/sync_audit/sync_audit_error.rs

@@ -0,0 +1,52 @@
+//! `SeaORM` Entity for sync_audit_error
+use anyhow::Result;
+use chrono::{DateTime, Utc};
+use sea_orm::{entity::prelude::*, ActiveValue::NotSet, Set};
+
+#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
+#[sea_orm(table_name = "sync_audit_error")]
+pub struct Model {
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub sync_audit_segment_id: i32,
+    pub block_number: i32,
+    pub error_type: Option<String>,
+    pub error_message: Option<String>,
+    pub created_at: DateTime<Utc>,
+}
+
+impl Related<super::sync_audit_segment::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::SyncAuditSegment.def()
+    }
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::sync_audit_segment::Entity",
+        from = "Column::SyncAuditSegmentId",
+        to = "super::sync_audit_segment::Column::Id"
+    )]
+    SyncAuditSegment,
+}
+
+impl ActiveModelBehavior for ActiveModel {}
+
+pub async fn create(
+    sync_audit_segment_id: i32,
+    block_number: i32,
+    error_type: Option<String>,
+    error_message: Option<String>,
+    conn: &DatabaseConnection,
+) -> Result<Model> {
+    let active = ActiveModel {
+        id: NotSet,
+        sync_audit_segment_id: Set(sync_audit_segment_id),
+        block_number: Set(block_number),
+        error_type: Set(error_type),
+        error_message: Set(error_message),
+        created_at: Set(Utc::now()),
+    };
+    Ok(active.insert(conn).await?)
+}

glados-audit/Cargo.toml

@@ -21,6 +21,7 @@ eth_trie = "0.5.0"
 ethportal-api.workspace = true
 glados-core.workspace = true
 migration.workspace = true
+futures.workspace = true
 rand.workspace = true
 sea-orm.workspace = true
 serde_json.workspace= true

glados-audit/src/cli.rs

@@ -110,6 +110,9 @@ pub struct Args {
     #[arg(long, action(ArgAction::Append))]
     pub portal_client: Vec<String>,
 
+    #[arg(long, default_value = "false")]
+    pub sync: bool,
+
     #[command(subcommand)]
     pub subcommand: Option<Command>,
 }
@@ -141,6 +144,7 @@ impl Default for Args {
             state: false,
             state_strategy: None,
             portal_client: vec!["ipc:////tmp/trin-jsonrpc.ipc".to_owned()],
+            sync: false,
             subcommand: None,
             stats_recording_period: 300,
         }

glados-audit/src/lib.rs

@@ -14,6 +14,7 @@ use std::{
     thread::available_parallelism,
     vec,
 };
+use sync_audit::run_sync_audit;
 
 use tokio::{
     sync::mpsc::{self, Receiver},
@@ -40,6 +41,7 @@ pub mod cli;
 pub(crate) mod selection;
 mod state;
 pub mod stats;
+pub mod sync_audit;
 pub(crate) mod validation;
 
 /// Configuration created from CLI arguments.
@@ -59,6 +61,8 @@ pub struct AuditConfig {
     pub state: bool,
     /// Specific state audit strategies to run.
     pub state_strategies: Vec<StateSelectionStrategy>,
+    /// Run 4444 sync audits.
+    pub sync: bool,
     /// Weight for each strategy.
     pub weights: HashMap<HistorySelectionStrategy, u8>,
     /// Number requests to a Portal node active at the same time.
@@ -127,6 +131,7 @@ impl AuditConfig {
             beacon_strategies: args.beacon_strategy.unwrap_or_default(),
             state: args.state,
             state_strategies: args.state_strategy.unwrap_or_default(),
+            sync: args.sync,
         })
     }
 }
@@ -168,6 +173,10 @@ pub async fn run_glados_command(conn: DatabaseConnection, command: cli::Command)
 }
 
 pub async fn run_glados_audit(conn: DatabaseConnection, config: AuditConfig) {
+    if config.sync {
+        run_sync_audit(config.clone(), &conn).await.unwrap();
+        return ();
+    }
     // if state network is enabled, run state audits
     if config.state {
         spawn_state_audit(conn.clone(), config.clone()).await;

glados-audit/src/validation.rs

@@ -35,7 +35,7 @@ pub fn content_is_valid(content: &content::Model, content_bytes: &[u8]) -> bool
     }
 }
 
-fn validate_beacon(content_key: &BeaconContentKey, content_bytes: &[u8]) -> bool {
+pub fn validate_beacon(content_key: &BeaconContentKey, content_bytes: &[u8]) -> bool {
     let content: BeaconContentValue = match BeaconContentValue::decode(content_key, content_bytes) {
         Ok(c) => c,
         Err(e) => {
@@ -72,7 +72,7 @@ fn validate_beacon(content_key: &BeaconContentKey, content_bytes: &[u8]) -> bool
     }
 }
 
-fn validate_history(content_key: &HistoryContentKey, content_bytes: &[u8]) -> bool {
+pub fn validate_history(content_key: &HistoryContentKey, content_bytes: &[u8]) -> bool {
     // check deserialization is valid
     let content: HistoryContentValue = match HistoryContentValue::decode(content_key, content_bytes)
     {

glados-core/src/jsonrpc.rs

@@ -247,29 +247,10 @@ impl PortalApi {
     ) -> Result<(Option<Content>, QueryTrace), JsonRpcError> {
         match content.protocol_id {
             content::SubProtocol::History => {
-                match HistoryNetworkApiClient::trace_get_content(
-                    &self.client,
-                    HistoryContentKey::try_from_bytes(&content.content_key)?,
-                )
+                self.get_history_content_with_trace(HistoryContentKey::try_from_bytes(
+                    &content.content_key,
+                )?)
                 .await
-                {
-                    Ok(TraceContentInfo { content, trace, .. }) => Ok((
-                        Some(Content {
-                            raw: content.into(),
-                        }),
-                        trace,
-                    )),
-                    Err(err) => match err.into() {
-                        JsonRpcError::ContentNotFound { trace } => {
-                            if let Some(trace) = trace {
-                                Ok((None, trace))
-                            } else {
-                                Err(JsonRpcError::MissingQueryTrace)
-                            }
-                        }
-                        err => Err(err),
-                    },
-                }
             }
             content::SubProtocol::State => {
                 match StateNetworkApiClient::trace_get_content(
@@ -323,4 +304,28 @@ impl PortalApi {
             }
         }
     }
+
+    pub async fn get_history_content_with_trace(
+        self,
+        key: HistoryContentKey,
+    ) -> Result<(Option<Content>, QueryTrace), JsonRpcError> {
+        match HistoryNetworkApiClient::trace_get_content(&self.client, key).await {
+            Ok(TraceContentInfo { content, trace, .. }) => Ok((
+                Some(Content {
+                    raw: content.into(),
+                }),
+                trace,
+            )),
+            Err(err) => match err.into() {
+                JsonRpcError::ContentNotFound { trace } => {
+                    if let Some(trace) = trace {
+                        Ok((None, trace))
+                    } else {
+                        Err(JsonRpcError::MissingQueryTrace)
+                    }
+                }
+                err => Err(err),
+            },
+        }
+    }
 }