logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-07-25 23:23:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'arjentix/full-bedrock-integration' into Pravdyvy/indexer-state-management

Browse Source
This commit is contained in:
Pravdyvy
2026-02-03 11:11:19 +02:00
parent 8fa8cc33d8 71a7905a58
commit e1df915357
98 changed files with 6933 additions and 686 deletions
Download Patch File Download Diff File Expand all files Collapse all files
integration_tests/src/config.rs
+39
View File
@@ -0,0 +1,39 @@
use std::net::SocketAddr;
use anyhow::Result;
use indexer_service::{BackoffConfig, BedrockClientConfig, ChannelId, IndexerConfig};
use url::Url;
pub fn indexer_config(bedrock_addr: SocketAddr) -> IndexerConfig {
let channel_id: [u8; 32] = [0u8, 1]
.repeat(16)
.try_into()
.unwrap_or_else(|_| unreachable!());
let channel_id = ChannelId::try_from(channel_id).expect("Failed to create channel ID");
IndexerConfig {
resubscribe_interval_millis: 1000,
backoff: BackoffConfig {
start_delay_millis: 100,
max_retries: 10,
},
bedrock_client_config: BedrockClientConfig {
addr: addr_to_http_url(bedrock_addr).expect("Failed to convert bedrock addr to URL"),
auth: None,
},
channel_id,
}
}
fn addr_to_http_url(addr: SocketAddr) -> Result<Url> {
// Convert 0.0.0.0 to 127.0.0.1 for client connections
// When binding to port 0, the server binds to 0.0.0.0:<random_port>
// but clients need to connect to 127.0.0.1:<port> to work reliably
let url_string = if addr.ip().is_unspecified() {
format!("http://127.0.0.1:{}", addr.port())
} else {
format!("http://{addr}")
};
url_string.parse().map_err(Into::into)
}
integration_tests/src/lib.rs
+32 -130
View File
@@ -2,24 +2,24 @@
use std::{net::SocketAddr, path::PathBuf, sync::LazyLock};
use actix_web::dev::ServerHandle;
use anyhow::{Context, Result};
use base64::{Engine, engine::general_purpose::STANDARD as BASE64};
use common::{
sequencer_client::SequencerClient,
transaction::{EncodedTransaction, NSSATransaction},
};
use futures::FutureExt as _;
use indexer_core::{IndexerCore, config::IndexerConfig};
use indexer_service::IndexerHandle;
use log::debug;
use nssa::PrivacyPreservingTransaction;
use nssa_core::Commitment;
use sequencer_core::config::SequencerConfig;
use sequencer_runner::SequencerHandle;
use tempfile::TempDir;
use tokio::task::JoinHandle;
use url::Url;
use wallet::{WalletCore, config::WalletConfigOverrides};
mod config;
// TODO: Remove this and control time from tests
pub const TIME_TO_WAIT_FOR_BLOCK_SECONDS: u64 = 12;
@@ -38,19 +38,17 @@ static LOGGER: LazyLock<()> = LazyLock::new(env_logger::init);
/// It's memory and logically safe to create multiple instances of this struct in parallel tests,
/// as each instance uses its own temporary directories for sequencer and wallet data.
pub struct TestContext {
sequencer_server_handle: ServerHandle,
sequencer_loop_handle: JoinHandle<Result<()>>,
sequencer_retry_pending_blocks_handle: JoinHandle<Result<()>>,
indexer_loop_handle: Option<JoinHandle<Result<()>>>,
sequencer_client: SequencerClient,
wallet: WalletCore,
_sequencer_handle: SequencerHandle,
_indexer_handle: IndexerHandle,
_temp_sequencer_dir: TempDir,
_temp_wallet_dir: TempDir,
_temp_indexer_dir: Option<TempDir>,
}
impl TestContext {
/// Create new test context in detached mode. Default.
/// Create new test context.
pub async fn new() -> Result<Self> {
let manifest_dir = env!("CARGO_MANIFEST_DIR");
@@ -60,51 +58,26 @@ impl TestContext {
let sequencer_config = SequencerConfig::from_path(&sequencer_config_path)
.context("Failed to create sequencer config from file")?;
Self::new_with_sequencer_and_maybe_indexer_configs(sequencer_config, None).await
Self::new_with_sequencer_config(sequencer_config).await
}
/// Create new test context in local bedrock node attached mode.
pub async fn new_bedrock_local_attached() -> Result<Self> {
let manifest_dir = env!("CARGO_MANIFEST_DIR");
let sequencer_config_path = PathBuf::from(manifest_dir)
.join("configs/sequencer/bedrock_local_attached/sequencer_config.json");
let sequencer_config = SequencerConfig::from_path(&sequencer_config_path)
.context("Failed to create sequencer config from file")?;
let indexer_config_path =
PathBuf::from(manifest_dir).join("configs/indexer/indexer_config.json");
let indexer_config = IndexerConfig::from_path(&indexer_config_path)
.context("Failed to create indexer config from file")?;
Self::new_with_sequencer_and_maybe_indexer_configs(sequencer_config, Some(indexer_config))
.await
}
/// Create new test context with custom sequencer config and maybe indexer config.
/// Create new test context with custom sequencer config.
///
/// `home` and `port` fields of the provided config will be overridden to meet tests parallelism
/// requirements.
pub async fn new_with_sequencer_and_maybe_indexer_configs(
sequencer_config: SequencerConfig,
indexer_config: Option<IndexerConfig>,
) -> Result<Self> {
pub async fn new_with_sequencer_config(sequencer_config: SequencerConfig) -> Result<Self> {
// Ensure logger is initialized only once
*LOGGER;
debug!("Test context setup");
let (
sequencer_server_handle,
sequencer_addr,
sequencer_loop_handle,
sequencer_retry_pending_blocks_handle,
temp_sequencer_dir,
) = Self::setup_sequencer(sequencer_config)
.await
.context("Failed to setup sequencer")?;
let bedrock_addr = todo!();
let indexer_config = config::indexer_config(bedrock_addr);
let (_sequencer_handle, sequencer_addr, temp_sequencer_dir) =
Self::setup_sequencer(sequencer_config)
.await
.context("Failed to setup sequencer")?;
// Convert 0.0.0.0 to 127.0.0.1 for client connections
// When binding to port 0, the server binds to 0.0.0.0:<random_port>
@@ -124,57 +97,23 @@ impl TestContext {
)
.context("Failed to create sequencer client")?;
if let Some(mut indexer_config) = indexer_config {
indexer_config.sequencer_client_config.addr =
Url::parse(&sequencer_addr).context("Failed to parse sequencer addr")?;
let _indexer_handle = indexer_service::run_server(indexer_config, 0)
.await
.context("Failed to run Indexer Service")?;
let temp_indexer_dir =
tempfile::tempdir().context("Failed to create temp dir for indexer home")?;
debug!("Using temp indexer home at {:?}", temp_indexer_dir.path());
indexer_config.home = temp_indexer_dir.path().to_owned();
let indexer_core = IndexerCore::new(indexer_config).await?;
let indexer_loop_handle = Some(tokio::spawn(async move {
indexer_core.subscribe_parse_block_stream().await
}));
Ok(Self {
sequencer_server_handle,
sequencer_loop_handle,
sequencer_retry_pending_blocks_handle,
indexer_loop_handle,
sequencer_client,
wallet,
_temp_sequencer_dir: temp_sequencer_dir,
_temp_wallet_dir: temp_wallet_dir,
_temp_indexer_dir: Some(temp_indexer_dir),
})
} else {
Ok(Self {
sequencer_server_handle,
sequencer_loop_handle,
sequencer_retry_pending_blocks_handle,
indexer_loop_handle: None,
sequencer_client,
wallet,
_temp_sequencer_dir: temp_sequencer_dir,
_temp_wallet_dir: temp_wallet_dir,
_temp_indexer_dir: None,
})
}
Ok(Self {
sequencer_client,
wallet,
_sequencer_handle,
_indexer_handle,
_temp_sequencer_dir: temp_sequencer_dir,
_temp_wallet_dir: temp_wallet_dir,
})
}
async fn setup_sequencer(
mut config: SequencerConfig,
) -> Result<(
ServerHandle,
SocketAddr,
JoinHandle<Result<()>>,
JoinHandle<Result<()>>,
TempDir,
)> {
) -> Result<(SequencerHandle, SocketAddr, TempDir)> {
let temp_sequencer_dir =
tempfile::tempdir().context("Failed to create temp dir for sequencer home")?;
@@ -186,20 +125,10 @@ impl TestContext {
// Setting port to 0 lets the OS choose a free port for us
config.port = 0;
let (
sequencer_server_handle,
sequencer_addr,
sequencer_loop_handle,
sequencer_retry_pending_blocks_handle,
) = sequencer_runner::startup_sequencer(config).await?;
let (sequencer_handle, sequencer_addr) =
sequencer_runner::startup_sequencer(config).await?;
Ok((
sequencer_server_handle,
sequencer_addr,
sequencer_loop_handle,
sequencer_retry_pending_blocks_handle,
temp_sequencer_dir,
))
Ok((sequencer_handle, sequencer_addr, temp_sequencer_dir))
}
async fn setup_wallet(sequencer_addr: String) -> Result<(WalletCore, TempDir)> {
@@ -251,33 +180,6 @@ impl TestContext {
}
}
impl Drop for TestContext {
fn drop(&mut self) {
debug!("Test context cleanup");
let Self {
sequencer_server_handle,
sequencer_loop_handle,
sequencer_retry_pending_blocks_handle,
indexer_loop_handle,
sequencer_client: _,
wallet: _,
_temp_sequencer_dir,
_temp_wallet_dir,
_temp_indexer_dir,
} = self;
sequencer_loop_handle.abort();
sequencer_retry_pending_blocks_handle.abort();
if let Some(indexer_loop_handle) = indexer_loop_handle {
indexer_loop_handle.abort();
}
// Can't wait here as Drop can't be async, but anyway stop signal should be sent
sequencer_server_handle.stop(true).now_or_never();
}
}
pub fn format_public_account_id(account_id: &str) -> String {
format!("Public/{account_id}")
}