feat: validate time events and propagate proofs through aggregator

Author: smrz2001

event-svc/src/event/service.rs

@@ -12,7 +12,9 @@ use super::{
 };
 use async_trait::async_trait;
 use ceramic_core::{EventId, Network, NodeId, SerializeExt};
-use ceramic_pipeline::{concluder::TimeProof, ConclusionData, ConclusionEvent, ConclusionInit, ConclusionTime};
+use ceramic_pipeline::{
+    concluder::TimeProof, ConclusionData, ConclusionEvent, ConclusionInit, ConclusionTime,
+};
 use ceramic_sql::sqlite::SqlitePool;
 use cid::Cid;
 use futures::stream::BoxStream;
@@ -402,30 +404,22 @@ impl EventService {
 
         match event {
             ceramic_event::unvalidated::Event::Time(time_event) => {
-                let proof = match self.discover_chain_proof(&time_event).await {
-                    Ok(proof) => Some(proof),
-                    Err(error) => {
-                        tracing::warn!(
-                            ?event_cid,
-                            ?error,
-                            "Failed to discover chain proof for time event"
-                        );
-                        None
-                    }
-                };
+                let proof = self.discover_chain_proof(&time_event).await.map_err(|e| {
+                    Error::new_app(anyhow::anyhow!("Failed to discover chain proof: {:?}", e))
+                })?;
 
                 Ok(ConclusionEvent::Time(ConclusionTime {
                     event_cid,
                     init,
                     previous: vec![*time_event.prev()],
                     order: delivered as u64,
-                    time_proof: proof.map(|p| TimeProof {
-                        before: p
+                    time_proof: TimeProof {
+                        before: proof
                             .timestamp
                             .try_into()
                             .expect("conclusion timestamp overflow"),
-                        chain_id: p.chain_id,
-                    }),
+                        chain_id: proof.chain_id,
+                    },
                 }))
             }
             ceramic_event::unvalidated::Event::Signed(signed_event) => {
@@ -638,12 +632,6 @@ pub enum ValidationError {
         key: EventId,
         reason: String,
     },
-    /// 'Soft error' -> should not kill recon conversation but should not be persisted
-    /// A time event could not be validated because no RPC provider was available
-    SoftError {
-        key: EventId,
-        reason: String,
-    },
 }
 
 #[derive(Debug, PartialEq, Eq, Default)]

event-svc/src/event/validator/event.rs

@@ -4,7 +4,6 @@ use ceramic_core::{Cid, EventId, NodeId};
 use ceramic_event::unvalidated;
 use ipld_core::ipld::Ipld;
 use recon::ReconItem;
-use tokio::try_join;
 
 use crate::{
     blockchain::eth_rpc,
@@ -117,6 +116,7 @@ impl ValidatedEvents {
         self.valid.extend(other.valid);
         self.invalid.extend(other.invalid);
         self.unvalidated.extend(other.unvalidated);
+        self.proofs.extend(other.proofs);
     }
 }
 
@@ -150,38 +150,34 @@ impl EventValidator {
     }
 
     /// Validates the events with the given validation requirement
-    /// If the [`ValidationRequirement`] is None, it just returns every event as valid
+    /// Regardless of the validation requirement, time events are always validated.
+    /// If the [`ValidationRequirement`] is None, it just returns every data event as valid.
     pub(crate) async fn validate_events(
         &self,
         validation_req: Option<&ValidationRequirement>,
         parsed_events: Vec<UnvalidatedEvent>,
     ) -> Result<ValidatedEvents> {
-        let validation_req = if let Some(req) = validation_req {
-            req
-        } else {
-            // we don't validate so we just return done
-            return Ok(ValidatedEvents {
-                valid: parsed_events
-                    .into_iter()
-                    .map(ValidatedEvent::from_unvalidated_unchecked)
-                    .collect(),
-                unvalidated: Vec::new(),
-                invalid: Vec::new(),
-                proofs: Vec::new(),
-            });
-        };
-
         let mut validated = ValidatedEvents::new_with_expected_valid(parsed_events.len());
-        // partition the events by type of validation needed and delegate to validators
+
+        // Partition the events by type of validation needed and delegate to validators
         let grouped = GroupedEvents::from(parsed_events);
 
-        let (validated_signed, validated_time) = try_join!(
-            self.validate_signed_events(grouped.signed_batch, validation_req),
-            self.validate_time_events(grouped.time_batch)
-        )?;
-        validated.extend_with(validated_signed);
+        // Time events are always validated
+        let validated_time = self.validate_time_events(grouped.time_batch).await?;
         validated.extend_with(validated_time);
 
+        if let Some(req) = validation_req {
+            let validated_signed = self
+                .validate_signed_events(grouped.signed_batch, req)
+                .await?;
+            validated.extend_with(validated_signed);
+        } else {
+            // Return all data events as valid
+            validated
+                .valid
+                .extend(Vec::<ValidatedEvent>::from(grouped.signed_batch));
+        };
+
         if !validated.invalid.is_empty() {
             tracing::warn!(count=%validated.invalid.len(), "invalid events discovered");
         }

event-svc/src/event/validator/grouped.rs

@@ -104,11 +104,22 @@ pub struct TimeValidationBatch(pub(crate) Vec<Time>);
 
 #[derive(Debug)]
 pub struct SignedValidationBatch {
-    pub data: Vec<SignedData>,
-    pub init: Vec<SignedInit>,
+    pub(crate) data: Vec<SignedData>,
+    pub(crate) init: Vec<SignedInit>,
     /// Possibly needed to verify data event controllers as these
     /// don't yet exist on disk, but the signatures aren't checked.
-    pub unsigned: Vec<Unsigned>,
+    pub(crate) unsigned: Vec<Unsigned>,
+}
+
+impl From<SignedValidationBatch> for Vec<ValidatedEvent> {
+    fn from(value: SignedValidationBatch) -> Self {
+        let mut events =
+            Vec::with_capacity(value.data.len() + value.init.len() + value.unsigned.len());
+        events.extend(value.data.into_iter().map(Into::into));
+        events.extend(value.init.into_iter().map(Into::into));
+        events.extend(value.unsigned.into_iter().map(Into::into));
+        events
+    }
 }
 
 impl From<Vec<UnvalidatedEvent>> for GroupedEvents {

one/src/daemon.rs

@@ -8,8 +8,7 @@ use anyhow::{anyhow, bail, Result};
 use ceramic_anchor_remote::RemoteCas;
 use ceramic_anchor_service::AnchorService;
 use ceramic_core::NodeKey;
-use ceramic_event_svc::eth_rpc::HttpEthRpc;
-use ceramic_event_svc::{ChainInclusionProvider, EventService};
+use ceramic_event_svc::EventService;
 use ceramic_interest_svc::InterestService;
 use ceramic_kubo_rpc::Multiaddr;
 use ceramic_metrics::{config::Config as MetricsConfig, MetricsHandle};
@@ -276,56 +275,6 @@ pub struct DaemonOpts {
     object_store_url: url::Url,
 }
 
-async fn get_eth_rpc_providers(
-    ethereum_rpc_urls: Vec<String>,
-    network: &Network,
-) -> Result<Vec<ChainInclusionProvider>> {
-    let ethereum_rpc_urls = if ethereum_rpc_urls.is_empty() {
-        network.default_rpc_urls()?
-    } else {
-        ethereum_rpc_urls
-    };
-
-    let mut providers = Vec::new();
-    for url in ethereum_rpc_urls {
-        match HttpEthRpc::try_new(&url).await {
-            Ok(provider) => {
-                let provider_chain = provider.chain_id();
-                if network
-                    .supported_chain_ids()
-                    .is_none_or(|ids| ids.contains(provider_chain))
-                {
-                    info!(
-                        "Using ethereum rpc provider for chain: {} with url: {}",
-                        provider.chain_id(),
-                        provider.url()
-                    );
-                    let provider: ChainInclusionProvider = Arc::new(provider);
-                    providers.push(provider);
-                } else {
-                    warn!("Eth RPC provider {} uses chainid {} which isn't supported by Ceramic network {:?}", url, provider_chain,network);
-                }
-            }
-            Err(err) => {
-                warn!("failed to create RPC client with url: '{url}': {:?}", err);
-            }
-        }
-    }
-
-    if providers.is_empty() {
-        if *network == Network::Local || *network == Network::InMemory {
-            warn!("No usable ethereum RPC provided for network {}. All TimeEvent validation will fail", network.name());
-        } else {
-            bail!(
-                "No usable ethereum RPC configured for network {}",
-                network.name()
-            );
-        }
-    }
-
-    Ok(providers)
-}
-
 fn spawn_database_optimizer(
     sqlite_pool: SqlitePool,
     mut shutdown: ShutdownSignal,
@@ -422,7 +371,10 @@ pub async fn run(opts: DaemonOpts) -> Result<()> {
         None
     };
 
-    let rpc_providers = get_eth_rpc_providers(opts.ethereum_rpc_urls, &opts.network).await?;
+    let rpc_providers = opts
+        .network
+        .get_eth_rpc_providers(opts.ethereum_rpc_urls)
+        .await?;
 
     // Construct services from pool
     let peer_svc = Arc::new(PeerService::new(sqlite_pool.clone()));

one/src/lib.rs

@@ -9,8 +9,9 @@ mod migrations;
 mod network;
 mod query;
 
-use anyhow::{anyhow, Result};
+use anyhow::{anyhow, bail, Result};
 use ceramic_core::ssi::caip2::ChainId;
+use ceramic_event_svc::{eth_rpc::HttpEthRpc, ChainInclusionProvider};
 use ceramic_metrics::config::Config as MetricsConfig;
 use ceramic_sql::sqlite::{SqliteOpts, SqlitePool};
 use clap::{Args, Parser, Subcommand, ValueEnum};
@@ -22,8 +23,7 @@ use multihash_codetable::Code;
 use multihash_derive::Hasher;
 use shutdown::Shutdown;
 use signal_hook_tokio::Signals;
-use std::str::FromStr;
-use std::{env, path::PathBuf};
+use std::{env, path::PathBuf, str::FromStr, sync::Arc};
 use tokio::io::AsyncReadExt;
 use tracing::{debug, error, info, warn};
 
@@ -121,7 +121,7 @@ impl Network {
     }
 
     /// Return the default ethereum rpc providers for each network.
-    pub fn default_rpc_urls(&self) -> Result<Vec<String>> {
+    fn default_rpc_urls(&self) -> Result<Vec<String>> {
         match self {
             Network::Mainnet => {
                 anyhow::bail!("no Ethereum RPC URLs specified for Mainnet")
@@ -151,7 +151,7 @@ impl Network {
     }
 
     /// return the allowed chain ids for this network. or None for any
-    pub fn supported_chain_ids(&self) -> Option<Vec<ChainId>> {
+    fn supported_chain_ids(&self) -> Option<Vec<ChainId>> {
         match self {
             Network::Mainnet => Some(vec![
                 ChainId::from_str("eip155:1").expect("eip155:1 is a valid chain")
@@ -168,7 +168,7 @@ impl Network {
     }
 
     /// Get the network as a unique name.
-    pub fn name(&self) -> String {
+    fn name(&self) -> String {
         match self {
             Network::Mainnet => "mainnet".to_owned(),
             Network::TestnetClay => "testnet-clay".to_owned(),
@@ -177,6 +177,56 @@ impl Network {
             Network::InMemory => "inmemory".to_owned(),
         }
     }
+
+    pub async fn get_eth_rpc_providers(
+        &self,
+        ethereum_rpc_urls: Vec<String>,
+    ) -> Result<Vec<ChainInclusionProvider>> {
+        let ethereum_rpc_urls = if ethereum_rpc_urls.is_empty() {
+            self.default_rpc_urls()?
+        } else {
+            ethereum_rpc_urls
+        };
+
+        let mut providers = Vec::new();
+        for url in ethereum_rpc_urls {
+            match HttpEthRpc::try_new(&url).await {
+                Ok(provider) => {
+                    let provider_chain = provider.chain_id();
+                    if self
+                        .supported_chain_ids()
+                        .is_none_or(|ids| ids.contains(provider_chain))
+                    {
+                        info!(
+                            "Using ethereum rpc provider for chain: {} with url: {}",
+                            provider.chain_id(),
+                            provider.url()
+                        );
+                        let provider: ChainInclusionProvider = Arc::new(provider);
+                        providers.push(provider);
+                    } else {
+                        warn!("Eth RPC provider {} uses chainid {} which isn't supported by Ceramic network {:?}", url, provider_chain, self);
+                    }
+                }
+                Err(err) => {
+                    warn!("failed to create RPC client with url: '{url}': {:?}", err);
+                }
+            }
+        }
+
+        if providers.is_empty() {
+            if *self == Network::Local || *self == Network::InMemory {
+                warn!("No usable ethereum RPC provided for network {}. All TimeEvent validation will fail", self.name());
+            } else {
+                bail!(
+                    "No usable ethereum RPC configured for network {}",
+                    self.name()
+                );
+            }
+        }
+
+        Ok(providers)
+    }
 }
 
 /// The default storage directory to use if none is provided. In order:

one/src/migrations.rs

@@ -123,6 +123,15 @@ pub struct FromIpfsOpts {
     /// For example, on mainnet, events are expected to have a chain id of 'eip155:1'.
     #[clap(long, env = "CERAMIC_ONE_VALIDATE_CHAIN")]
     validate_chain: bool,
+
+    /// Ethereum RPC URLs used for time events validation. Required when connecting to mainnet and uses fallback URLs if not specified for other networks.
+    #[arg(
+        long,
+        use_value_delimiter = true,
+        value_delimiter = ',',
+        env = "CERAMIC_ONE_ETHEREUM_RPC_URLS"
+    )]
+    ethereum_rpc_urls: Vec<String>,
 }
 
 impl From<&FromIpfsOpts> for DBOpts {
@@ -165,13 +174,18 @@ async fn from_ipfs(opts: FromIpfsOpts) -> Result<()> {
     let network = opts.network.to_network(&opts.local_network_id)?;
     let db_opts: DBOpts = (&opts).into();
     let sqlite_pool = db_opts.get_sqlite_pool(SqliteOpts::default()).await?;
+    let rpc_providers = opts
+        .network
+        .get_eth_rpc_providers(opts.ethereum_rpc_urls)
+        .await?;
+
     // TODO: feature flags here? or just remove this entirely when enabling
     let event_svc = Arc::new(
         EventService::try_new(
             sqlite_pool,
             ceramic_event_svc::UndeliveredEventReview::Skip,
             false,
-            vec![],
+            rpc_providers,
         )
         .await?,
     );