intermediate changes

history/src/block_stores/postgres_block_store.rs

@@ -32,6 +32,14 @@ pub struct PostgresBlockStore {
 }
 
 impl PostgresBlockStore {
+    pub fn new(session_cache: PostgresSessionCache, epoch_cache: EpochCache) -> Self {
+        Self {
+            session_cache,
+            epoch_cache,
+            postgres_data: Arc::new(RwLock::new(PostgresData { from_slot: 0, to_slot: 0, current_epoch: 0 })),
+        }
+    }
+
     pub async fn start_new_epoch(&self, schema: &String) -> Result<()> {
         // create schema for new epoch
         let session = self

lite-rpc/Cargo.toml

@@ -41,6 +41,7 @@ async-trait = { workspace = true }
 tokio = { version = "1.28.2", features = ["full", "fs"]}
 tokio-postgres = { version = "0.7.8", features = ["with-chrono-0_4"] }
 chrono = { workspace = true }
+itertools = { workspace = true }
 
 solana-lite-rpc-core = { workspace = true }
 solana-lite-rpc-services = { workspace = true }

lite-rpc/src/bridge.rs

@@ -3,19 +3,17 @@ use crate::{
     jsonrpsee_subscrption_handler_sink::JsonRpseeSubscriptionHandlerSink,
     rpc::LiteRpcServer,
 };
-use solana_sdk::{epoch_info::EpochInfo, transaction::VersionedTransaction, message::{v0::Message, VersionedMessage}};
-
+use solana_sdk::epoch_info::EpochInfo;
 use solana_lite_rpc_services::{
     transaction_service::TransactionService, tx_sender::TXS_IN_CHANNEL,
 };
-
 use anyhow::Context;
 use jsonrpsee::{core::SubscriptionResult, server::ServerBuilder, PendingSubscriptionSink};
 use log::info;
 use prometheus::{opts, register_int_counter, IntCounter};
 use solana_lite_rpc_core::{
     stores::{block_information_store::BlockInformation, data_cache::DataCache, tx_store::TxProps},
-    AnyhowJoinHandle, encoding::BASE64,
+    AnyhowJoinHandle,
 };
 use solana_lite_rpc_history::history::History;
 use solana_rpc_client::nonblocking::rpc_client::RpcClient;
@@ -30,6 +28,7 @@ use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey, slot_histo
 use solana_transaction_status::{TransactionStatus, UiConfirmedBlock, EncodedTransactionWithStatusMeta, UiTransactionStatusMeta, option_serializer::OptionSerializer};
 use std::{str::FromStr, sync::Arc};
 use tokio::net::ToSocketAddrs;
+use itertools::Itertools;
 
 lazy_static::lazy_static! {
     static ref RPC_SEND_TX: IntCounter =
@@ -340,14 +339,13 @@ impl LiteRpcServer for LiteBridge {
                         solana_transaction_status::TransactionDetails::Accounts => (false, false, true),
                     };
 
-                    if include_accounts || include_rewards {
-                        block.transactions.iter().map(|transaction_info| {
-                            let signature = transaction_info.signature.clone();
+                    if is_full || include_accounts || include_rewards {
+                        Some(block.transactions.iter().map(|transaction_info| {
                             EncodedTransactionWithStatusMeta {
                                 version: None,
                                 meta: Some(UiTransactionStatusMeta {
-                                    err: transaction_info.err,
-                                    status: transaction_info.err.map_or(Ok(()), |x| Err(x)),
+                                    err: transaction_info.err.clone(),
+                                    status: transaction_info.err.clone().map_or(Ok(()), |x| Err(x)),
                                     fee: 0,
                                     pre_balances: vec![],
                                     post_balances: vec![],
@@ -360,9 +358,9 @@ impl LiteRpcServer for LiteBridge {
                                     return_data: OptionSerializer::None,
                                     compute_units_consumed: transaction_info.cu_consumed.map_or( OptionSerializer::None, |x| OptionSerializer::Some(x)),
                                 }),
-                                transaction: solana_transaction_status::EncodedTransaction::Binary(transaction_info.message, solana_transaction_status::TransactionBinaryEncoding::Base64)
+                                transaction: solana_transaction_status::EncodedTransaction::Binary(transaction_info.message.clone(), solana_transaction_status::TransactionBinaryEncoding::Base64)
                             }
-                        }).collect_vec()
+                        }).collect_vec())
                     } else {
                         None
                     }
@@ -375,7 +373,7 @@ impl LiteRpcServer for LiteBridge {
                 parent_slot: block.parent_slot,
                 transactions,
                 signatures: None,
-                rewards: None,
+                rewards: block.rewards,
                 block_time: Some(block.block_time as i64),
                 block_height: Some(block.block_height),
             }))

lite-rpc/src/main.rs

@@ -28,9 +28,11 @@ use solana_lite_rpc_core::structures::{
     epoch::EpochCache, identity_stakes::IdentityStakes, notifications::NotificationSender,
     produced_block::ProducedBlock,
 };
+use solana_lite_rpc_core::traits::block_storage_interface::BlockStorageImpl;
 use solana_lite_rpc_core::types::BlockStream;
 use solana_lite_rpc_core::AnyhowJoinHandle;
 use solana_lite_rpc_history::block_stores::inmemory_block_store::InmemoryBlockStore;
+use solana_lite_rpc_history::block_stores::postgres_block_store::PostgresBlockStore;
 use solana_lite_rpc_history::history::History;
 use solana_lite_rpc_history::postgres::postgres_session::PostgresSessionCache;
 use solana_lite_rpc_services::data_caching_service::DataCachingService;
@@ -62,32 +64,23 @@ async fn get_latest_block(
 }
 
 pub async fn start_postgres(
-    enable: bool,
-) -> anyhow::Result<(Option<NotificationSender>, AnyhowJoinHandle)> {
-    if !enable {
-        return Ok((
-            None,
-            tokio::spawn(async {
-                std::future::pending::<()>().await;
-                unreachable!()
-            }),
-        ));
-    }
+    postgres_session_cache: PostgresSessionCache,
+) -> anyhow::Result<(NotificationSender, AnyhowJoinHandle)> {
 
     let (postgres_send, postgres_recv) = mpsc::unbounded_channel();
-
-    let postgres_session_cache = PostgresSessionCache::new().await?;
     let postgres = PostgresLogger::start(postgres_session_cache, postgres_recv);
 
-    Ok((Some(postgres_send), postgres))
+    Ok((postgres_send, postgres))
 }
 
 pub async fn start_lite_rpc(args: Args, rpc_client: Arc<RpcClient>) -> anyhow::Result<()> {
     let Args {
         lite_rpc_ws_addr,
         lite_rpc_http_addr,
         fanout_size,
-        enable_postgres_logging: enable_postgres,
+        enable_postgres_logging,
+        enable_fetch_blocks_from_postgres,
+        enable_save_blocks_in_postgres,
         prometheus_addr,
         identity_keypair,
         maximum_retries_per_tx,
@@ -137,7 +130,7 @@ pub async fn start_lite_rpc(args: Args, rpc_client: Arc<RpcClient>) -> anyhow::R
             store: Arc::new(DashMap::new()),
             save_for_additional_slots: NB_SLOTS_TRANSACTIONS_TO_CACHE,
         },
-        epoch_data,
+        epoch_data: epoch_data.clone(),
     };
 
     let lata_cache_service = DataCachingService {
@@ -153,8 +146,22 @@ pub async fn start_lite_rpc(args: Args, rpc_client: Arc<RpcClient>) -> anyhow::R
         vote_account_notifier,
     );
     drop(blocks_notifier);
+    let postgres_session_cache = if enable_fetch_blocks_from_postgres || enable_postgres_logging {
+         Some(PostgresSessionCache::new().await?)
+    } else {
+        None
+    };
 
-    let (notification_channel, postgres) = start_postgres(enable_postgres).await?;
+    let (notification_channel, postgres) = if enable_postgres_logging {
+        let (channel, postgres) = start_postgres(postgres_session_cache.clone().unwrap()).await?;
+        (Some(channel), postgres)
+    } else {
+        let jh = tokio::spawn(async {
+            std::future::pending::<()>().await;
+            unreachable!()
+        });
+        (None, jh)
+    };
 
     let tpu_config = TpuServiceConfig {
         fanout_slots: fanout_size,
@@ -200,8 +207,15 @@ pub async fn start_lite_rpc(args: Args, rpc_client: Arc<RpcClient>) -> anyhow::R
 
     let support_service = tokio::spawn(async move { spawner.spawn_support_services().await });
 
+    let block_storage: BlockStorageImpl = if enable_save_blocks_in_postgres {
+        let session_cache =  postgres_session_cache.unwrap();
+        Arc::new(PostgresBlockStore::new(session_cache, epoch_data.clone()))
+    } else {
+        Arc::new(InmemoryBlockStore::new(1024))
+    };
+
     let history = History {
-        block_storage: Arc::new(InmemoryBlockStore::new(1024)),
+        block_storage,
     };
 
     let bridge_service = tokio::spawn(
@@ -240,7 +254,7 @@ fn get_args() -> Args {
 
     dotenv().ok();
 
-    args.enable_postgres = args.enable_postgres
+    args.enable_postgres_logging = args.enable_postgres_logging
         || if let Ok(enable_postgres_env_var) = env::var("PG_ENABLED") {
             enable_postgres_env_var != "false"
         } else {