Fetch and store transactions from pd storage

crates/bin/pcli/src/command/view/tx.rs

@@ -255,7 +255,11 @@ impl TxCmd {
         let tx_info = if let Ok(tx_info) = app.view().transaction_info_by_hash(hash).await {
             tx_info
         } else {
-            println!("Transaction not found in view service, fetching from fullnode...");
+            if !self.raw {
+                println!("Transaction not found in view service, fetching from fullnode...");
+            } else {
+                tracing::info!("Transaction not found in view service, fetching from fullnode...");
+            }
             // Fall back to fetching from fullnode
             let mut client = app.tendermint_proxy_client().await?;
             let rsp = client

crates/core/app/Cargo.toml

@@ -16,7 +16,9 @@ penumbra-keys = { path = "../keys" }
 penumbra-num = { path = "../num" }
 penumbra-component = { path = "../component/component" }
 penumbra-chain = { path = "../component/chain", features = ["component"] }
-penumbra-shielded-pool = { path = "../component/shielded-pool", features = ["component"] }
+penumbra-shielded-pool = { path = "../component/shielded-pool", features = [
+    "component",
+] }
 penumbra-stake = { path = "../component/stake" }
 penumbra-governance = { path = "../component/governance" }
 penumbra-sct = { path = "../component/sct" }
@@ -63,7 +65,9 @@ tendermint = "0.34.0"
 tendermint-proto = "0.34.0"
 tendermint-light-client-verifier = "0.34.0"
 ibc-types = { version = "0.10.0", default-features = false }
-ibc-proto = { version = "0.39.0", default-features = false, features = ["server"] }
+ibc-proto = { version = "0.39.0", default-features = false, features = [
+    "server",
+] }
 
 [dev-dependencies]
 ed25519-consensus = "2"

crates/core/app/src/app/mod.rs

@@ -17,12 +17,14 @@ use penumbra_governance::component::{Governance, StateReadExt as _};
 use penumbra_governance::StateWriteExt as _;
 use penumbra_ibc::component::{IBCComponent, StateWriteExt as _};
 use penumbra_ibc::StateReadExt as _;
+use penumbra_proto::core::app::v1alpha1::TransactionsByHeightResponse;
 use penumbra_proto::DomainType;
 use penumbra_sct::component::SctManager;
 use penumbra_shielded_pool::component::{NoteManager, ShieldedPool};
 use penumbra_stake::component::{Staking, StateReadExt as _, StateWriteExt as _, ValidatorUpdates};
 use penumbra_storage::{ArcStateDeltaExt, Snapshot, StateDelta, StateRead, StateWrite, Storage};
 use penumbra_transaction::Transaction;
+use prost::Message as _;
 use tendermint::abci::{self, Event};
 use tendermint::validator::Update;
 use tracing::Instrument;
@@ -300,6 +302,14 @@ impl App {
             .state
             .try_begin_transaction()
             .expect("state Arc should be present and unique");
+
+        // Index the transaction:
+        let height = state_tx.get_block_height().await?;
+        let transaction = Arc::as_ref(&tx).clone();
+        state_tx
+            .put_block_transaction(height, transaction.into())
+            .await?;
+
         tx.execute(&mut state_tx).await?;
 
         // At this point, we've completed execution successfully with no errors,
@@ -677,6 +687,25 @@ pub trait StateReadExt: StateRead {
             stake_params,
         })
     }
+
+    async fn transactions_by_height(
+        &self,
+        block_height: u64,
+    ) -> Result<TransactionsByHeightResponse> {
+        let transactions = match self
+            .nonverifiable_get_raw(state_key::transactions_by_height(block_height).as_bytes())
+            .await?
+        {
+            Some(transactions) => transactions,
+            None => TransactionsByHeightResponse {
+                transactions: vec![],
+                block_height,
+            }
+            .encode_to_vec(),
+        };
+
+        Ok(TransactionsByHeightResponse::decode(&transactions[..])?)
+    }
 }
 
 impl<
@@ -692,3 +721,32 @@ impl<
     > StateReadExt for T
 {
 }
+
+#[async_trait]
+pub trait StateWriteExt: StateWrite {
+    /// Stores the transactions that occurred during a CometBFT block.
+    /// This is used to create a durable transaction log for clients to retrieve;
+    /// the CometBFT `get_block_by_height` RPC call will only return data for blocks
+    /// since the last checkpoint, so we need to store the transactions separately.
+    async fn put_block_transaction(
+        &mut self,
+        height: u64,
+        transaction: penumbra_proto::core::transaction::v1alpha1::Transaction,
+    ) -> Result<()> {
+        // Extend the existing transactions with the new one.
+        let mut transactions_response = self.transactions_by_height(height).await?;
+        transactions_response.transactions = transactions_response
+            .transactions
+            .into_iter()
+            .chain(std::iter::once(transaction))
+            .collect();
+
+        self.nonverifiable_put_raw(
+            state_key::transactions_by_height(height).into(),
+            transactions_response.encode_to_vec(),
+        );
+        Ok(())
+    }
+}
+
+impl<T: StateWrite + ?Sized> StateWriteExt for T {}

crates/core/app/src/app/state_key.rs

@@ -1,3 +1,7 @@
 pub fn app_state() -> &'static str {
     "genesis/app_state"
 }
+
+pub fn transactions_by_height(block_height: u64) -> String {
+    format!("cometbft-data/transactions_by_height/{block_height:020}")
+}

crates/core/app/src/lib.rs

@@ -5,14 +5,19 @@ mod mock_client;
 mod temp_storage_ext;
 
 pub use action_handler::ActionHandler;
+pub use app::StateWriteExt;
 pub use dao_ext::DaoStateReadExt;
 pub use mock_client::MockClient;
 pub use temp_storage_ext::TempStorageExt;
 
 use once_cell::sync::Lazy;
 
-pub static SUBSTORE_PREFIXES: Lazy<Vec<String>> =
-    Lazy::new(|| vec![penumbra_ibc::IBC_SUBSTORE_PREFIX.to_string()]);
+pub static SUBSTORE_PREFIXES: Lazy<Vec<String>> = Lazy::new(|| {
+    vec![
+        penumbra_ibc::IBC_SUBSTORE_PREFIX.to_string(),
+        penumbra_chain::COMETBFT_SUBSTORE_PREFIX.to_string(),
+    ]
+});
 
 pub mod app;
 pub mod genesis;

crates/core/app/src/rpc.rs

@@ -1,6 +1,7 @@
 use penumbra_chain::component::StateReadExt as _;
 use penumbra_proto::core::app::v1alpha1::{
     query_service_server::QueryService, AppParametersRequest, AppParametersResponse,
+    TransactionsByHeightRequest, TransactionsByHeightResponse,
 };
 use penumbra_storage::Storage;
 use tonic::Status;
@@ -21,6 +22,27 @@ impl Server {
 
 #[tonic::async_trait]
 impl QueryService for Server {
+    #[instrument(skip(self, request))]
+    async fn transactions_by_height(
+        &self,
+        request: tonic::Request<TransactionsByHeightRequest>,
+    ) -> Result<tonic::Response<TransactionsByHeightResponse>, Status> {
+        let state = self.storage.latest_snapshot();
+        state
+            .check_chain_id(&request.get_ref().chain_id)
+            .await
+            .map_err(|e| tonic::Status::unknown(format!("chain_id not OK: {e}")))?;
+        let request_inner = request.into_inner();
+        let block_height = request_inner.block_height;
+
+        let tx_response = state
+            .transactions_by_height(block_height)
+            .await
+            .map_err(|e| tonic::Status::internal(format!("transaction response bad: {e}")))?;
+
+        Ok(tonic::Response::new(tx_response))
+    }
+
     #[instrument(skip(self, request))]
     async fn app_parameters(
         &self,

crates/core/component/chain/src/effect_hash.rs

@@ -1,4 +1,4 @@
-use penumbra_proto::{penumbra::core::component::chain::v1alpha1 as pb_crypto, DomainType};
+use penumbra_proto::{penumbra::core::transaction::v1alpha1 as pb_transaction, DomainType};
 
 /// Something that can be hashed to produce an [`EffectHash`].
 pub trait EffectingData {
@@ -11,6 +11,8 @@ pub trait EffectingData {
 /// This includes, e.g., the commitments to new output notes created by the
 /// transaction, or nullifiers spent by the transaction, but does not include
 /// _authorizing data_ such as signatures or zk proofs.
+///
+/// TODO: move this to the transaction crate, can be done when we ditch the chain component.
 #[derive(Clone, Copy, Eq, PartialEq)]
 pub struct EffectHash(pub [u8; 64]);
 
@@ -35,20 +37,20 @@ impl std::fmt::Debug for EffectHash {
 }
 
 impl DomainType for EffectHash {
-    type Proto = pb_crypto::EffectHash;
+    type Proto = pb_transaction::EffectHash;
 }
 
-impl From<EffectHash> for pb_crypto::EffectHash {
+impl From<EffectHash> for pb_transaction::EffectHash {
     fn from(msg: EffectHash) -> Self {
         Self {
             inner: msg.0.to_vec(),
         }
     }
 }
 
-impl TryFrom<pb_crypto::EffectHash> for EffectHash {
+impl TryFrom<pb_transaction::EffectHash> for EffectHash {
     type Error = anyhow::Error;
-    fn try_from(value: pb_crypto::EffectHash) -> Result<Self, Self::Error> {
+    fn try_from(value: pb_transaction::EffectHash) -> Result<Self, Self::Error> {
         Ok(Self(value.inner.try_into().map_err(|_| {
             anyhow::anyhow!("incorrect length for effect hash")
         })?))

crates/core/component/chain/src/lib.rs

@@ -9,6 +9,9 @@ pub use transaction::TransactionContext;
 mod epoch;
 mod note_source;
 
+/// The substore prefix used for storing historic CometBFT block data.
+pub static COMETBFT_SUBSTORE_PREFIX: &'static str = "cometbft-data/";
+
 #[cfg_attr(docsrs, doc(cfg(feature = "component")))]
 #[cfg(feature = "component")]
 pub mod component;

crates/narsil/narsil/src/ledger/app.rs

@@ -1,10 +1,11 @@
 use anyhow::Result;
+use penumbra_chain::component::StateReadExt as _;
 use std::sync::Arc;
 
 // TODO: we should not have dependencies on penumbra_chain in narsil
 // and instead implement narsil-specific state accessors or extract
 // the common accessors elsewhere to avoid mingling penumbra-specific logic.
-use penumbra_app::genesis;
+use penumbra_app::{genesis, StateWriteExt as _};
 use penumbra_proto::{core::transaction::v1alpha1::Transaction, Message};
 use penumbra_storage::{ArcStateDeltaExt, RootHash, Snapshot, StateDelta, Storage};
 use tendermint::{abci, validator::Update};
@@ -51,8 +52,16 @@ impl App {
         self.deliver_tx(tx).await
     }
 
-    pub async fn deliver_tx(&mut self, _tx: Arc<Transaction>) -> Result<Vec<abci::Event>> {
-        Ok(vec![])
+    pub async fn deliver_tx(&mut self, tx: Arc<Transaction>) -> Result<Vec<abci::Event>> {
+        let mut state_tx = self
+            .state
+            .try_begin_transaction()
+            .expect("state Arc should not be referenced elsewhere");
+
+        let height = state_tx.get_block_height().await?;
+        let transaction = Arc::as_ref(&tx).clone();
+        state_tx.put_block_transaction(height, transaction).await?;
+        Ok(state_tx.apply().1)
     }
 
     pub async fn end_block(&mut self, _end_block: &abci::request::EndBlock) -> Vec<abci::Event> {