Merge remote-tracking branch 'origin/hm/head-change-revert' into hm/sqlite-indexer

Author: hanabi1224

src/chain/store/chain_store.rs

@@ -6,8 +6,6 @@ use super::{
     index::{ChainIndex, ResolveNullTipset},
     tipset_tracker::TipsetTracker,
 };
-use crate::db::{EthMappingsStore, EthMappingsStoreExt};
-use crate::interpreter::{BlockMessages, VMTrace};
 use crate::libp2p_bitswap::{BitswapStoreRead, BitswapStoreReadWrite};
 use crate::message::{ChainMessage, Message as MessageTrait, SignedMessage};
 use crate::networks::{ChainConfig, Height};
@@ -23,7 +21,15 @@ use crate::{
     blocks::{CachingBlockHeader, Tipset, TipsetKey, TxMeta},
     db::HeaviestTipsetKeyProvider,
 };
+use crate::{
+    db::{EthMappingsStore, EthMappingsStoreExt},
+    rpc::chain::PathChange,
+};
 use crate::{fil_cns, utils::cache::SizeTrackingLruCache};
+use crate::{
+    interpreter::{BlockMessages, VMTrace},
+    rpc::chain::PathChanges,
+};
 use ahash::{HashMap, HashMapExt, HashSet};
 use anyhow::Context as _;
 use cid::Cid;
@@ -47,17 +53,16 @@ pub type ChainEpochDelta = ChainEpoch;
 
 /// `Enum` for `pubsub` channel that defines message type variant and data
 /// contained in message type.
-#[derive(Clone, Debug)]
-pub enum HeadChange {
-    Apply(Tipset),
-}
+pub type HeadChange = PathChange<Tipset>;
+
+pub type HeadChanges = PathChanges<Tipset>;
 
 /// Stores chain data such as heaviest tipset and cached tipset info at each
 /// epoch. This structure is thread-safe, and all caches are wrapped in a mutex
 /// to allow a consistent `ChainStore` to be shared across tasks.
 pub struct ChainStore<DB> {
     /// Publisher for head change events
-    publisher: Publisher<HeadChange>,
+    publisher: Publisher<HeadChanges>,
 
     /// key-value `datastore`.
     db: Arc<DB>,
@@ -66,7 +71,7 @@ pub struct ChainStore<DB> {
     heaviest_tipset_key_provider: Arc<dyn HeaviestTipsetKeyProvider + Sync + Send>,
 
     /// Heaviest tipset cache
-    heaviest_tipset_cache: Arc<RwLock<Option<Tipset>>>,
+    heaviest_tipset_cache: Arc<RwLock<Tipset>>,
 
     /// Used as a cache for tipset `lookbacks`.
     chain_index: Arc<ChainIndex<Arc<DB>>>,
@@ -124,14 +129,24 @@ where
         let (publisher, _) = broadcast::channel(SINK_CAP);
         let chain_index = Arc::new(ChainIndex::new(Arc::clone(&db)));
         let validated_blocks = Mutex::new(HashSet::default());
-
+        let head = if let Some(head_tsk) = heaviest_tipset_key_provider
+            .heaviest_tipset_key()
+            .context("failed to load head tipset key")?
+            && let Some(head) = chain_index
+                .load_tipset(&head_tsk)
+                .context("failed to load head tipset")?
+        {
+            head
+        } else {
+            Tipset::from(&genesis_block_header)
+        };
         let cs = Self {
             publisher,
             chain_index,
             tipset_tracker: TipsetTracker::new(Arc::clone(&db), chain_config.clone()),
             db,
             heaviest_tipset_key_provider,
-            heaviest_tipset_cache: Default::default(),
+            heaviest_tipset_cache: Arc::new(RwLock::new(head)),
             genesis_block_header,
             validated_blocks,
             eth_mappings,
@@ -142,14 +157,31 @@ where
     }
 
     /// Sets heaviest tipset
-    pub fn set_heaviest_tipset(&self, ts: Tipset) -> Result<(), Error> {
+    pub fn set_heaviest_tipset(&self, head: Tipset) -> Result<(), Error> {
+        head.key().save(self.blockstore())?;
         self.heaviest_tipset_key_provider
-            .set_heaviest_tipset_key(ts.key())?;
-        *self.heaviest_tipset_cache.write() = Some(ts.clone());
-        ts.key().save(self.blockstore())?;
-        if self.publisher.send(HeadChange::Apply(ts)).is_err() {
-            debug!("did not publish head change, no active receivers");
+            .set_heaviest_tipset_key(head.key())?;
+        let old_head = std::mem::replace(&mut *self.heaviest_tipset_cache.write(), head.clone());
+
+        let changes = match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) {
+            Ok(changes) => changes,
+            Err(e) => {
+                // Do not warn when the old head is genesis
+                if old_head.epoch() > 0 {
+                    warn!("failed to get chain path changes: {e}");
+                }
+                // Fallback to single apply
+                PathChanges {
+                    applies: vec![head],
+                    reverts: vec![],
+                }
+            }
+        };
+
+        if self.publisher.send(changes).is_err() {
+            debug!("did not publish changes, no active receivers");
         }
+
         Ok(())
     }
 
@@ -200,16 +232,7 @@ where
 
     /// Returns the currently tracked heaviest tipset.
     pub fn heaviest_tipset(&self) -> Tipset {
-        if let Some(ts) = &*self.heaviest_tipset_cache.read() {
-            return ts.clone();
-        }
-        let tsk = self
-            .heaviest_tipset_key_provider
-            .heaviest_tipset_key()
-            .unwrap_or_else(|_| TipsetKey::from(nunny::vec![*self.genesis_block_header.cid()]));
-        self.chain_index
-            .load_required_tipset(&tsk)
-            .expect("failed to load heaviest tipset")
+        self.heaviest_tipset_cache.read().clone()
     }
 
     /// Returns the genesis tipset.
@@ -218,7 +241,7 @@ where
     }
 
     /// Returns a reference to the publisher of head changes.
-    pub fn publisher(&self) -> &Publisher<HeadChange> {
+    pub fn publisher(&self) -> &Publisher<HeadChanges> {
         &self.publisher
     }
 

src/chain/store/indexer.rs

@@ -14,7 +14,7 @@ use sqlx::Row as _;
 
 use crate::{
     blocks::Tipset,
-    chain::{ChainStore, HeadChange, index::ResolveNullTipset},
+    chain::{ChainStore, HeadChanges, index::ResolveNullTipset},
     message::{ChainMessage, SignedMessage},
     rpc::{
         chain::types::ChainIndexValidation,
@@ -122,18 +122,18 @@ where
 
     pub async fn index_loop(
         &self,
-        mut head_change_subscriber: tokio::sync::broadcast::Receiver<HeadChange>,
+        mut head_change_subscriber: tokio::sync::broadcast::Receiver<HeadChanges>,
     ) -> anyhow::Result<()> {
         loop {
-            match head_change_subscriber.recv().await? {
-                HeadChange::Apply(ts) => {
-                    if let Err(e) = self.index_tipset(&ts).await {
-                        tracing::warn!(
-                            "failed to index new head@{}({}): {e}",
-                            ts.epoch(),
-                            ts.key()
-                        );
-                    }
+            let HeadChanges { reverts, applies } = head_change_subscriber.recv().await?;
+            for ts in reverts {
+                if let Err(e) = self.revert_tipset(&ts).await {
+                    tracing::warn!("failed to index new head@{}({}): {e}", ts.epoch(), ts.key());
+                }
+            }
+            for ts in applies {
+                if let Err(e) = self.index_tipset(&ts).await {
+                    tracing::warn!("failed to index new head@{}({}): {e}", ts.epoch(), ts.key());
                 }
             }
         }
@@ -556,6 +556,27 @@ where
         Ok(())
     }
 
+    pub async fn revert_tipset(&self, ts: &Tipset) -> anyhow::Result<()> {
+        let tsk_cid_bytes = ts.key().cid()?.to_bytes();
+        // Because of deferred execution in Filecoin, events at tipset T are reverted when a tipset T+1 is reverted.
+        // However, the tipet `T` itself is not reverted.
+        let pts = Tipset::load_required(self.cs.blockstore(), ts.parents())?;
+        let events_tsk_cid_bytes = pts.key().cid()?.to_bytes();
+        let mut tx = self.db.begin().await?;
+        sqlx::query(self.stmts.update_events_to_reverted)
+            .bind(&tsk_cid_bytes)
+            .execute(tx.deref_mut())
+            .await?;
+        // events are indexed against the message inclusion tipset, not the message execution tipset.
+        // So we need to revert the events for the message inclusion tipset
+        sqlx::query(self.stmts.update_events_to_reverted)
+            .bind(&events_tsk_cid_bytes)
+            .execute(tx.deref_mut())
+            .await?;
+        tx.commit().await?;
+        Ok(())
+    }
+
     pub async fn index_tipset(&self, ts: &Tipset) -> anyhow::Result<()> {
         let mut tx = self.db.begin().await?;
         self.index_tipset_and_parent_events_with_tx(&mut tx, ts)

src/chain/store/indexer/ddls.rs

@@ -56,7 +56,9 @@ pub struct PreparedStatements {
     pub insert_eth_tx_hash: &'static str,
     pub insert_tipset_message: &'static str,
     pub update_tipset_to_non_reverted: &'static str,
+    pub update_tipset_to_reverted: &'static str,
     pub update_events_to_non_reverted: &'static str,
+    pub update_events_to_reverted: &'static str,
     pub get_msg_id_for_msg_cid_and_tipset: &'static str,
     pub insert_event: &'static str,
     pub insert_event_entry: &'static str,
@@ -81,7 +83,10 @@ impl Default for PreparedStatements {
         let insert_tipset_message = "INSERT INTO tipset_message (tipset_key_cid, height, reverted, message_cid, message_index) VALUES (?, ?, ?, ?, ?) ON CONFLICT (tipset_key_cid, message_cid) DO UPDATE SET reverted = 0";
         let update_tipset_to_non_reverted =
             "UPDATE tipset_message SET reverted = 0 WHERE tipset_key_cid = ?";
+        let update_tipset_to_reverted =
+            "UPDATE tipset_message SET reverted = 1 WHERE tipset_key_cid = ?";
         let update_events_to_non_reverted = "UPDATE event SET reverted = 0 WHERE message_id IN (SELECT id FROM tipset_message WHERE tipset_key_cid = ?)";
+        let update_events_to_reverted = "UPDATE event SET reverted = 1 WHERE message_id IN (SELECT id FROM tipset_message WHERE height >= ?)";
         let get_msg_id_for_msg_cid_and_tipset = "SELECT id FROM tipset_message WHERE tipset_key_cid = ? AND message_cid = ? AND reverted = 0";
         let insert_event = "INSERT INTO event (message_id, event_index, emitter_id, emitter_addr, reverted) VALUES (?, ?, ?, ?, ?)";
         let insert_event_entry = "INSERT INTO event_entry (event_id, indexed, flags, key, codec, value) VALUES (?, ?, ?, ?, ?, ?)";
@@ -102,7 +107,9 @@ impl Default for PreparedStatements {
             insert_eth_tx_hash,
             insert_tipset_message,
             update_tipset_to_non_reverted,
+            update_tipset_to_reverted,
             update_events_to_non_reverted,
+            update_events_to_reverted,
             get_msg_id_for_msg_cid_and_tipset,
             insert_event,
             insert_event_entry,

src/daemon/mod.rs

@@ -7,7 +7,6 @@ pub mod db_util;
 pub mod main;
 
 use crate::blocks::Tipset;
-use crate::chain::HeadChange;
 use crate::chain::index::ResolveNullTipset;
 use crate::chain_sync::network_context::SyncNetworkContext;
 use crate::chain_sync::{ChainFollower, SyncStatus};
@@ -507,21 +506,19 @@ fn maybe_start_indexer_service(
     {
         // Old indexer
         {
-            let mut receiver = ctx.state_manager.chain_store().publisher().subscribe();
+            let mut head_changes_subscriber =
+                ctx.state_manager.chain_store().publisher().subscribe();
             let chain_store = ctx.state_manager.chain_store().clone();
             services.spawn(async move {
                 tracing::info!("Starting indexer service");
-
                 // Continuously listen for head changes
                 loop {
-                    let HeadChange::Apply(ts) = receiver.recv().await?;
-
-                    tracing::debug!("Indexing tipset {}", ts.key());
-
-                    let delegated_messages =
-                        chain_store.headers_delegated_messages(ts.block_headers().iter())?;
-
-                    chain_store.process_signed_messages(&delegated_messages)?;
+                    for ts in head_changes_subscriber.recv().await?.applies {
+                        tracing::debug!("Indexing tipset {}", ts.key());
+                        let delegated_messages =
+                            chain_store.headers_delegated_messages(ts.block_headers().iter())?;
+                        chain_store.process_signed_messages(&delegated_messages)?;
+                    }
                 }
             });
 
@@ -544,11 +541,11 @@ fn maybe_start_indexer_service(
         // New SQLITE indexer
         if let Some(indexer) = &ctx.chain_indexer {
             services.spawn({
-                let head_change_subscriber =
+                let head_changes_subscriber =
                     ctx.state_manager.chain_store().publisher().subscribe();
                 let indexer = indexer.clone();
                 async move {
-                    if let Err(e) = indexer.index_loop(head_change_subscriber).await {
+                    if let Err(e) = indexer.index_loop(head_changes_subscriber).await {
                         tracing::warn!("indexer stopped unexpectedly: {e}");
                     }
                     Ok(())

src/db/car/many.rs

@@ -130,12 +130,12 @@ impl<WriterT> ManyCar<WriterT> {
         Ok(())
     }
 
-    pub fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey> {
-        self.read_only
+    pub fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
+        Ok(self
+            .read_only
             .read()
             .peek()
-            .map(|w| AnyCar::heaviest_tipset_key(&w.car))
-            .context("ManyCar store doesn't have a heaviest tipset key")
+            .map(|w| AnyCar::heaviest_tipset_key(&w.car)))
     }
 
     pub fn heaviest_tipset(&self) -> anyhow::Result<Tipset> {
@@ -252,9 +252,9 @@ impl<WriterT: EthMappingsStore> EthMappingsStore for ManyCar<WriterT> {
 }
 
 impl<T: Blockstore + SettingsStore> super::super::HeaviestTipsetKeyProvider for ManyCar<T> {
-    fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey> {
+    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
         match SettingsStoreExt::read_obj::<TipsetKey>(self, crate::db::setting_keys::HEAD_KEY)? {
-            Some(tsk) => Ok(tsk),
+            Some(tsk) => Ok(Some(tsk)),
             None => self.heaviest_tipset_key(),
         }
     }

src/db/gc/snapshot.rs

@@ -282,7 +282,7 @@ where
             tracing::warn!("{e}");
         }
 
-        *self.memory_db_head_key.write() = db.heaviest_tipset_key().ok();
+        *self.memory_db_head_key.write() = db.heaviest_tipset_key()?;
         db.unsubscribe_write_ops();
         match joinset.join_next().await {
             Some(Ok(map)) => {

src/db/memory.rs

@@ -154,9 +154,8 @@ impl BitswapStoreReadWrite for MemoryDB {
 }
 
 impl super::HeaviestTipsetKeyProvider for MemoryDB {
-    fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey> {
-        SettingsStoreExt::read_obj::<TipsetKey>(self, crate::db::setting_keys::HEAD_KEY)?
-            .context("head key not found")
+    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
+        SettingsStoreExt::read_obj::<TipsetKey>(self, crate::db::setting_keys::HEAD_KEY)
     }
 
     fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()> {

src/db/mod.rs

@@ -239,14 +239,14 @@ impl<T: PersistentStore> PersistentStore for &Arc<T> {
 
 pub trait HeaviestTipsetKeyProvider {
     /// Returns the currently tracked heaviest tipset.
-    fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey>;
+    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>>;
 
     /// Sets heaviest tipset.
     fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()>;
 }
 
 impl<T: HeaviestTipsetKeyProvider> HeaviestTipsetKeyProvider for Arc<T> {
-    fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey> {
+    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
         self.as_ref().heaviest_tipset_key()
     }
 

src/db/parity_db.rs

@@ -173,9 +173,8 @@ impl SettingsStore for ParityDb {
 }
 
 impl super::HeaviestTipsetKeyProvider for ParityDb {
-    fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey> {
-        super::SettingsStoreExt::read_obj::<TipsetKey>(self, super::setting_keys::HEAD_KEY)?
-            .context("head key not found")
+    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
+        super::SettingsStoreExt::read_obj::<TipsetKey>(self, super::setting_keys::HEAD_KEY)
     }
 
     fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()> {