change(state): Add an init function for a standalone `ReadStateService` (#8595)

* Adds an init_read_only() fn in zebra-state

* moves elasticsearch initialization to `FinalizedState::new_with_debug()`

* Updates callers of `FinalizedState::{new, new_with_debug}` to pass a bool to try enabling elasticsearch

---------

Co-authored-by: Alfredo Garcia <oxarbitrage@gmail.com>
This commit is contained in:
Arya 2024-06-12 09:35:13 -04:00 committed by GitHub
parent 139e1c3ed7
commit 1b5f9426f9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 116 additions and 82 deletions

View File

@ -50,27 +50,6 @@ where
} }
} }
impl From<SemanticallyVerifiedBlock> for ChainTipBlock {
fn from(prepared: SemanticallyVerifiedBlock) -> Self {
let SemanticallyVerifiedBlock {
block,
hash,
height,
new_outputs: _,
transaction_hashes,
} = prepared;
Self {
hash,
height,
time: block.header.time,
transactions: block.transactions.clone(),
transaction_hashes,
previous_block_hash: block.header.previous_block_hash,
}
}
}
impl SemanticallyVerifiedBlock { impl SemanticallyVerifiedBlock {
/// Returns a [`ContextuallyVerifiedBlock`] created from this block, /// Returns a [`ContextuallyVerifiedBlock`] created from this block,
/// with fake zero-valued spent UTXOs. /// with fake zero-valued spent UTXOs.

View File

@ -46,8 +46,10 @@ pub use request::{
}; };
pub use response::{KnownBlock, MinedTx, ReadResponse, Response}; pub use response::{KnownBlock, MinedTx, ReadResponse, Response};
pub use service::{ pub use service::{
chain_tip::{ChainTipChange, LatestChainTip, TipAction}, chain_tip::{ChainTipBlock, ChainTipChange, ChainTipSender, LatestChainTip, TipAction},
check, init, spawn_init, check, init, init_read_only,
non_finalized_state::NonFinalizedState,
spawn_init, spawn_init_read_only,
watch_receiver::WatchReceiver, watch_receiver::WatchReceiver,
OutputIndex, OutputLocation, TransactionIndex, TransactionLocation, OutputIndex, OutputLocation, TransactionIndex, TransactionLocation,
}; };
@ -76,7 +78,6 @@ pub use response::GetBlockTemplateChainInfo;
#[cfg(any(test, feature = "proptest-impl"))] #[cfg(any(test, feature = "proptest-impl"))]
pub use service::{ pub use service::{
arbitrary::{populated_state, CHAIN_TIP_UPDATE_WAIT_LIMIT}, arbitrary::{populated_state, CHAIN_TIP_UPDATE_WAIT_LIMIT},
chain_tip::{ChainTipBlock, ChainTipSender},
finalized_state::{RawBytes, KV, MAX_ON_DISK_HEIGHT}, finalized_state::{RawBytes, KV, MAX_ON_DISK_HEIGHT},
init_test, init_test_services, init_test, init_test_services,
}; };
@ -96,4 +97,4 @@ pub(crate) use config::hidden::{
write_database_format_version_to_disk, write_state_database_format_version_to_disk, write_database_format_version_to_disk, write_state_database_format_version_to_disk,
}; };
pub(crate) use request::ContextuallyVerifiedBlock; pub use request::ContextuallyVerifiedBlock;

View File

@ -24,15 +24,6 @@ use std::{
time::{Duration, Instant}, time::{Duration, Instant},
}; };
#[cfg(feature = "elasticsearch")]
use elasticsearch::{
auth::Credentials::Basic,
cert::CertificateValidation,
http::transport::{SingleNodeConnectionPool, TransportBuilder},
http::Url,
Elasticsearch,
};
use futures::future::FutureExt; use futures::future::FutureExt;
use tokio::sync::{oneshot, watch}; use tokio::sync::{oneshot, watch};
use tower::{util::BoxService, Service, ServiceExt}; use tower::{util::BoxService, Service, ServiceExt};
@ -319,29 +310,12 @@ impl StateService {
checkpoint_verify_concurrency_limit: usize, checkpoint_verify_concurrency_limit: usize,
) -> (Self, ReadStateService, LatestChainTip, ChainTipChange) { ) -> (Self, ReadStateService, LatestChainTip, ChainTipChange) {
let timer = CodeTimer::start(); let timer = CodeTimer::start();
let finalized_state = FinalizedState::new(
&config,
network,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
let finalized_state = { true,
let conn_pool = SingleNodeConnectionPool::new(
Url::parse(config.elasticsearch_url.as_str())
.expect("configured elasticsearch url is invalid"),
); );
let transport = TransportBuilder::new(conn_pool)
.cert_validation(CertificateValidation::None)
.auth(Basic(
config.clone().elasticsearch_username,
config.clone().elasticsearch_password,
))
.build()
.expect("elasticsearch transport builder should not fail");
let elastic_db = Some(Elasticsearch::new(transport));
FinalizedState::new(&config, network, elastic_db)
};
#[cfg(not(feature = "elasticsearch"))]
let finalized_state = { FinalizedState::new(&config, network) };
timer.finish(module_path!(), line!(), "opening finalized state database"); timer.finish(module_path!(), line!(), "opening finalized state database");
let timer = CodeTimer::start(); let timer = CodeTimer::start();
@ -387,7 +361,7 @@ impl StateService {
let read_service = ReadStateService::new( let read_service = ReadStateService::new(
&finalized_state, &finalized_state,
block_write_task, Some(block_write_task),
non_finalized_state_receiver, non_finalized_state_receiver,
); );
@ -828,14 +802,14 @@ impl ReadStateService {
/// and a watch channel for updating the shared recent non-finalized chain. /// and a watch channel for updating the shared recent non-finalized chain.
pub(crate) fn new( pub(crate) fn new(
finalized_state: &FinalizedState, finalized_state: &FinalizedState,
block_write_task: Arc<std::thread::JoinHandle<()>>, block_write_task: Option<Arc<std::thread::JoinHandle<()>>>,
non_finalized_state_receiver: watch::Receiver<NonFinalizedState>, non_finalized_state_receiver: watch::Receiver<NonFinalizedState>,
) -> Self { ) -> Self {
let read_service = Self { let read_service = Self {
network: finalized_state.network(), network: finalized_state.network(),
db: finalized_state.db.clone(), db: finalized_state.db.clone(),
non_finalized_state_receiver: WatchReceiver::new(non_finalized_state_receiver), non_finalized_state_receiver: WatchReceiver::new(non_finalized_state_receiver),
block_write_task: Some(block_write_task), block_write_task,
}; };
tracing::debug!("created new read-only state service"); tracing::debug!("created new read-only state service");
@ -1945,6 +1919,52 @@ pub fn init(
) )
} }
/// Initialize a read state service from the provided [`Config`].
/// Returns a read-only state service,
///
/// Each `network` has its own separate on-disk database.
///
/// To share access to the state, clone the returned [`ReadStateService`].
pub fn init_read_only(
config: Config,
network: &Network,
) -> (
ReadStateService,
ZebraDb,
tokio::sync::watch::Sender<NonFinalizedState>,
) {
let finalized_state = FinalizedState::new_with_debug(
&config,
network,
true,
#[cfg(feature = "elasticsearch")]
false,
true,
);
let (non_finalized_state_sender, non_finalized_state_receiver) =
tokio::sync::watch::channel(NonFinalizedState::new(network));
(
ReadStateService::new(&finalized_state, None, non_finalized_state_receiver),
finalized_state.db.clone(),
non_finalized_state_sender,
)
}
/// Calls [`init_read_only`] with the provided [`Config`] and [`Network`] from a blocking task.
/// Returns a [`tokio::task::JoinHandle`] with a read state service and chain tip sender.
pub fn spawn_init_read_only(
config: Config,
network: &Network,
) -> tokio::task::JoinHandle<(
ReadStateService,
ZebraDb,
tokio::sync::watch::Sender<NonFinalizedState>,
)> {
let network = network.clone();
tokio::task::spawn_blocking(move || init_read_only(config, &network))
}
/// Calls [`init`] with the provided [`Config`] and [`Network`] from a blocking task. /// Calls [`init`] with the provided [`Config`] and [`Network`] from a blocking task.
/// Returns a [`tokio::task::JoinHandle`] with a boxed state service, /// Returns a [`tokio::task::JoinHandle`] with a boxed state service,
/// a read state service, and receivers for state chain tip updates. /// a read state service, and receivers for state chain tip updates.

View File

@ -107,15 +107,15 @@ impl From<ContextuallyVerifiedBlock> for ChainTipBlock {
} }
} }
impl From<CheckpointVerifiedBlock> for ChainTipBlock { impl From<SemanticallyVerifiedBlock> for ChainTipBlock {
fn from(finalized: CheckpointVerifiedBlock) -> Self { fn from(prepared: SemanticallyVerifiedBlock) -> Self {
let CheckpointVerifiedBlock(SemanticallyVerifiedBlock { let SemanticallyVerifiedBlock {
block, block,
hash, hash,
height, height,
new_outputs: _,
transaction_hashes, transaction_hashes,
.. } = prepared;
}) = finalized;
Self { Self {
hash, hash,
@ -128,6 +128,12 @@ impl From<CheckpointVerifiedBlock> for ChainTipBlock {
} }
} }
impl From<CheckpointVerifiedBlock> for ChainTipBlock {
fn from(CheckpointVerifiedBlock(prepared): CheckpointVerifiedBlock) -> Self {
prepared.into()
}
}
/// A sender for changes to the non-finalized and finalized chain tips. /// A sender for changes to the non-finalized and finalized chain tips.
#[derive(Debug)] #[derive(Debug)]
pub struct ChainTipSender { pub struct ChainTipSender {

View File

@ -142,14 +142,14 @@ impl FinalizedState {
pub fn new( pub fn new(
config: &Config, config: &Config,
network: &Network, network: &Network,
#[cfg(feature = "elasticsearch")] elastic_db: Option<elasticsearch::Elasticsearch>, #[cfg(feature = "elasticsearch")] enable_elastic_db: bool,
) -> Self { ) -> Self {
Self::new_with_debug( Self::new_with_debug(
config, config,
network, network,
false, false,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
elastic_db, enable_elastic_db,
false, false,
) )
} }
@ -162,9 +162,37 @@ impl FinalizedState {
config: &Config, config: &Config,
network: &Network, network: &Network,
debug_skip_format_upgrades: bool, debug_skip_format_upgrades: bool,
#[cfg(feature = "elasticsearch")] elastic_db: Option<elasticsearch::Elasticsearch>, #[cfg(feature = "elasticsearch")] enable_elastic_db: bool,
read_only: bool, read_only: bool,
) -> Self { ) -> Self {
#[cfg(feature = "elasticsearch")]
let elastic_db = if enable_elastic_db {
use elasticsearch::{
auth::Credentials::Basic,
cert::CertificateValidation,
http::transport::{SingleNodeConnectionPool, TransportBuilder},
http::Url,
Elasticsearch,
};
let conn_pool = SingleNodeConnectionPool::new(
Url::parse(config.elasticsearch_url.as_str())
.expect("configured elasticsearch url is invalid"),
);
let transport = TransportBuilder::new(conn_pool)
.cert_validation(CertificateValidation::None)
.auth(Basic(
config.clone().elasticsearch_username,
config.clone().elasticsearch_password,
))
.build()
.expect("elasticsearch transport builder should not fail");
Some(Elasticsearch::new(transport))
} else {
None
};
let db = ZebraDb::new( let db = ZebraDb::new(
config, config,
STATE_DATABASE_KIND, STATE_DATABASE_KIND,

View File

@ -62,7 +62,7 @@ fn test_raw_rocksdb_column_families_with_network(network: Network) {
&Config::ephemeral(), &Config::ephemeral(),
&network, &network,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
None, false,
); );
// Snapshot the column family names // Snapshot the column family names

View File

@ -24,7 +24,7 @@ fn blocks_with_v5_transactions() -> Result<()> {
.and_then(|v| v.parse().ok()) .and_then(|v| v.parse().ok())
.unwrap_or(DEFAULT_PARTIAL_CHAIN_PROPTEST_CASES)), .unwrap_or(DEFAULT_PARTIAL_CHAIN_PROPTEST_CASES)),
|((chain, count, network, _history_tree) in PreparedChain::default())| { |((chain, count, network, _history_tree) in PreparedChain::default())| {
let mut state = FinalizedState::new(&Config::ephemeral(), &network, #[cfg(feature = "elasticsearch")] None); let mut state = FinalizedState::new(&Config::ephemeral(), &network, #[cfg(feature = "elasticsearch")] false);
let mut height = Height(0); let mut height = Height(0);
// use `count` to minimize test failures, so they are easier to diagnose // use `count` to minimize test failures, so they are easier to diagnose
for block in chain.iter().take(count) { for block in chain.iter().take(count) {
@ -65,7 +65,7 @@ fn all_upgrades_and_wrong_commitments_with_fake_activation_heights() -> Result<(
.unwrap_or(DEFAULT_PARTIAL_CHAIN_PROPTEST_CASES)), .unwrap_or(DEFAULT_PARTIAL_CHAIN_PROPTEST_CASES)),
|((chain, _count, network, _history_tree) in PreparedChain::default().with_valid_commitments().no_shrink())| { |((chain, _count, network, _history_tree) in PreparedChain::default().with_valid_commitments().no_shrink())| {
let mut state = FinalizedState::new(&Config::ephemeral(), &network, #[cfg(feature = "elasticsearch")] None); let mut state = FinalizedState::new(&Config::ephemeral(), &network, #[cfg(feature = "elasticsearch")] false);
let mut height = Height(0); let mut height = Height(0);
let heartwood_height = NetworkUpgrade::Heartwood.activation_height(&network).unwrap(); let heartwood_height = NetworkUpgrade::Heartwood.activation_height(&network).unwrap();
let heartwood_height_plus1 = (heartwood_height + 1).unwrap(); let heartwood_height_plus1 = (heartwood_height + 1).unwrap();

View File

@ -169,7 +169,7 @@ fn test_block_and_transaction_data_with_network(network: Network) {
&Config::ephemeral(), &Config::ephemeral(),
&network, &network,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
None, false,
); );
// Assert that empty databases are the same, regardless of the network. // Assert that empty databases are the same, regardless of the network.

View File

@ -479,7 +479,7 @@ fn rejection_restores_internal_state_genesis() -> Result<()> {
} }
))| { ))| {
let mut state = NonFinalizedState::new(&network); let mut state = NonFinalizedState::new(&network);
let finalized_state = FinalizedState::new(&Config::ephemeral(), &network, #[cfg(feature = "elasticsearch")] None); let finalized_state = FinalizedState::new(&Config::ephemeral(), &network, #[cfg(feature = "elasticsearch")] false);
let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool(); let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool();
finalized_state.set_finalized_value_pool(fake_value_pool); finalized_state.set_finalized_value_pool(fake_value_pool);

View File

@ -157,7 +157,7 @@ fn best_chain_wins_for_network(network: Network) -> Result<()> {
&Config::ephemeral(), &Config::ephemeral(),
&network, &network,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
None, false,
); );
state.commit_new_chain(block2.prepare(), &finalized_state)?; state.commit_new_chain(block2.prepare(), &finalized_state)?;
@ -194,7 +194,7 @@ fn finalize_pops_from_best_chain_for_network(network: Network) -> Result<()> {
&Config::ephemeral(), &Config::ephemeral(),
&network, &network,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
None, false,
); );
let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool(); let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool();
@ -245,7 +245,7 @@ fn commit_block_extending_best_chain_doesnt_drop_worst_chains_for_network(
&Config::ephemeral(), &Config::ephemeral(),
&network, &network,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
None, false,
); );
let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool(); let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool();
@ -289,7 +289,7 @@ fn shorter_chain_can_be_best_chain_for_network(network: Network) -> Result<()> {
&Config::ephemeral(), &Config::ephemeral(),
&network, &network,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
None, false,
); );
let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool(); let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool();
@ -334,7 +334,7 @@ fn longer_chain_with_more_work_wins_for_network(network: Network) -> Result<()>
&Config::ephemeral(), &Config::ephemeral(),
&network, &network,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
None, false,
); );
let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool(); let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool();
@ -378,7 +378,7 @@ fn equal_length_goes_to_more_work_for_network(network: Network) -> Result<()> {
&Config::ephemeral(), &Config::ephemeral(),
&network, &network,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
None, false,
); );
let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool(); let fake_value_pool = ValueBalance::<NonNegative>::fake_populated_pool();
@ -426,7 +426,7 @@ fn history_tree_is_updated_for_network_upgrade(
&Config::ephemeral(), &Config::ephemeral(),
&network, &network,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
None, false,
); );
state state
@ -525,7 +525,7 @@ fn commitment_is_validated_for_network_upgrade(network: Network, network_upgrade
&Config::ephemeral(), &Config::ephemeral(),
&network, &network,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
None, false,
); );
state state

View File

@ -101,7 +101,7 @@ pub(crate) fn new_state_with_mainnet_genesis(
// The tests that use this setup function also commit invalid blocks to the state. // The tests that use this setup function also commit invalid blocks to the state.
true, true,
#[cfg(feature = "elasticsearch")] #[cfg(feature = "elasticsearch")]
None, false,
false, false,
); );
let non_finalized_state = NonFinalizedState::new(&network); let non_finalized_state = NonFinalizedState::new(&network);