From 254fe989b526da6bad6fcbd268499dfcb6c5b6fa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Wo=C5=9B?= Date: Mon, 4 Sep 2023 16:15:51 +0200 Subject: [PATCH 1/8] refactor fuel-indexer run command to take advantage of tokio::task::JoinSet --- Cargo.lock | 1 - packages/fuel-indexer/Cargo.toml | 1 - packages/fuel-indexer/src/commands/run.rs | 100 ++++++++++------------ 3 files changed, 43 insertions(+), 59 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9779e287a..d8a0e6cfb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3227,7 +3227,6 @@ dependencies = [ "sqlx", "thiserror", "tokio", - "tokio-util", "tracing", "wasmer", "wasmer-middlewares", diff --git a/packages/fuel-indexer/Cargo.toml b/packages/fuel-indexer/Cargo.toml index edf9fc26f..334653a3a 100644 --- a/packages/fuel-indexer/Cargo.toml +++ b/packages/fuel-indexer/Cargo.toml @@ -36,7 +36,6 @@ itertools = "0.10" sqlx = { version = "0.6", features = ["bigdecimal"] } thiserror = { workspace = true } tokio = { features = ["macros", "rt-multi-thread", "sync", "process"], workspace = true } -tokio-util = { workspace = true } tracing = { workspace = true } wasmer = "4" wasmer-middlewares = "4" diff --git a/packages/fuel-indexer/src/commands/run.rs b/packages/fuel-indexer/src/commands/run.rs index 1cdc347f2..cbf030a26 100644 --- a/packages/fuel-indexer/src/commands/run.rs +++ b/packages/fuel-indexer/src/commands/run.rs @@ -8,55 +8,47 @@ use fuel_indexer_lib::{ }; use tokio::signal::unix::{signal, Signal, SignalKind}; use tokio::sync::mpsc::channel; -use tokio_util::sync::CancellationToken; use tracing::info; #[cfg(feature = "api-server")] use fuel_indexer_api_server::api::WebApi; -// Returns a CancellationToken which will be notified when a shutdown signal -// have been reveived. -async fn shutdown_signal_handler() -> anyhow::Result { - let cancel_token = tokio_util::sync::CancellationToken::new(); - +// Returns a future which completes when a shutdown signal has been reveived. +fn shutdown_signal_handler() -> std::io::Result> { let mut sighup: Signal = signal(SignalKind::hangup())?; let mut sigterm: Signal = signal(SignalKind::terminate())?; let mut sigint: Signal = signal(SignalKind::interrupt())?; - tokio::spawn({ - let cancel_token = cancel_token.clone(); - - async move { - #[cfg(unix)] - { - tokio::select! { - _ = sighup.recv() => { - info!("Received SIGHUP. Stopping services."); - } - _ = sigterm.recv() => { - info!("Received SIGTERM. Stopping services."); - } - _ = sigint.recv() => { - info!("Received SIGINT. Stopping services."); - } + let future = async move { + #[cfg(unix)] + { + tokio::select! { + _ = sighup.recv() => { + info!("Received SIGHUP. Stopping services."); + } + _ = sigterm.recv() => { + info!("Received SIGTERM. Stopping services."); + } + _ = sigint.recv() => { + info!("Received SIGINT. Stopping services."); } - cancel_token.cancel(); } + } - #[cfg(not(unix))] - { - signal::ctrl_c().await?; - info!("Received CTRL+C. Stopping services."); - } + #[cfg(not(unix))] + { + signal::ctrl_c().await?; + info!("Received CTRL+C. Stopping services."); } - }); + }; - Ok(cancel_token) + Ok(future) } pub async fn exec(args: IndexerArgs) -> anyhow::Result<()> { - // for graceful shutdown - let cancel_token = shutdown_signal_handler().await?; + let mut subsystems: tokio::task::JoinSet<()> = tokio::task::JoinSet::new(); + + subsystems.spawn(shutdown_signal_handler()?); let IndexerArgs { manifest, @@ -138,16 +130,20 @@ pub async fn exec(args: IndexerArgs) -> anyhow::Result<()> { } } - let service_handle = tokio::spawn(service.run()); + subsystems.spawn(service.run()); #[cfg(feature = "api-server")] - let web_handle = tokio::spawn(WebApi::build_and_run(config.clone(), pool, tx)); - - #[cfg(not(feature = "api-server"))] - let web_handle = tokio::spawn(futures::future::ready(())); + subsystems.spawn({ + let config = config.clone(); + async { + if let Err(e) = WebApi::build_and_run(config, pool, tx).await { + tracing::error!("Api Server failed: {e}"); + } + } + }); #[cfg(feature = "fuel-core-lib")] - let node_handle = { + { use fuel_core::service::{Config, FuelService}; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; @@ -156,28 +152,18 @@ pub async fn exec(args: IndexerArgs) -> anyhow::Result<()> { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 4000), ..Config::local_node() }; - tokio::spawn(async move { - let node = FuelService::new_node(config).await.unwrap(); - Some(node) - }) - } else { - tokio::spawn(futures::future::ready(None)) + subsystems.spawn(async move { + if let Err(e) = FuelService::new_node(config).await { + tracing::error!("Fuen Node failed: {e}"); + }; + }); } }; - #[cfg(not(feature = "fuel-core-lib"))] - let node_handle = tokio::spawn(futures::future::ready(())); - - // spawn application as separate task - tokio::spawn({ - let cancel_token = cancel_token.clone(); - async move { - let _ = tokio::join!(service_handle, node_handle, web_handle); - cancel_token.cancel(); - } - }); - - cancel_token.cancelled().await; + // Each subsystem runs its own loop, and we require all subsystems for the + // Indexer service to operate correctly. If any of the subsystems stops + // running, the entire Indexer Service exits. + subsystems.join_next().await; if embedded_database { let name = postgres_database.unwrap_or(defaults::POSTGRES_DATABASE.to_string()); From 80074af6cb41b530c5f7b44f3f9d2f845843ae62 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Wo=C5=9B?= Date: Mon, 4 Sep 2023 17:23:18 +0200 Subject: [PATCH 2/8] fuel-indexer: refactor run, service, and executor --- packages/fuel-indexer-benchmarks/src/lib.rs | 7 +- .../tests/graphql_server.rs | 92 +++---- packages/fuel-indexer-tests/tests/service.rs | 5 +- packages/fuel-indexer/src/commands/run.rs | 7 +- packages/fuel-indexer/src/executor.rs | 175 +++++-------- packages/fuel-indexer/src/service.rs | 245 ++++++++---------- 6 files changed, 217 insertions(+), 314 deletions(-) diff --git a/packages/fuel-indexer-benchmarks/src/lib.rs b/packages/fuel-indexer-benchmarks/src/lib.rs index 22eafb10f..474b53643 100644 --- a/packages/fuel-indexer-benchmarks/src/lib.rs +++ b/packages/fuel-indexer-benchmarks/src/lib.rs @@ -7,7 +7,7 @@ use fuel_indexer::{ use fuel_indexer_database::IndexerConnectionPool; use fuel_indexer_lib::config::DatabaseConfig; use fuel_indexer_tests::fixtures::TestPostgresDb; -use std::{str::FromStr, sync::atomic::AtomicBool}; +use std::str::FromStr; /// Location of Fuel node to be used for block retrieval. pub const NODE_URL: &str = "beta-4.fuel.network:80"; @@ -73,7 +73,6 @@ pub fn create_wasm_indexer_benchmark( .unwrap(); let blocks = rt.block_on(get_blocks(start_block, num_blocks)).unwrap(); c.bench_function(name, move |b| { - let kill_switch = std::sync::Arc::new(AtomicBool::new(false)); b.iter_batched( // This setup function is run prior to each iteration of // the benchmark; this ensures that there is a fresh WASM @@ -92,9 +91,7 @@ pub fn create_wasm_indexer_benchmark( (executor, blocks.clone()) }) }, - |(mut ex, blocks)| { - rt.block_on(ex.handle_events(kill_switch.clone(), blocks)) - }, + |(mut ex, blocks)| rt.block_on(ex.handle_events(blocks)), criterion::BatchSize::SmallInput, ) }); diff --git a/packages/fuel-indexer-tests/tests/graphql_server.rs b/packages/fuel-indexer-tests/tests/graphql_server.rs index a15310459..0a33ae791 100644 --- a/packages/fuel-indexer-tests/tests/graphql_server.rs +++ b/packages/fuel-indexer-tests/tests/graphql_server.rs @@ -1,6 +1,4 @@ -use fuel_indexer_tests::fixtures::{ - mock_request, setup_web_test_components, WebTestComponents, -}; +use fuel_indexer_tests::fixtures::{mock_request, setup_web_test_components}; use fuel_indexer_utils::uid; use hyper::header::CONTENT_TYPE; use serde_json::{Number, Value}; @@ -8,17 +6,13 @@ use std::collections::HashMap; #[actix_web::test] async fn test_entity_with_required_and_optional_fields() { - let WebTestComponents { - server, - db: _db, - client, - .. - } = setup_web_test_components(None).await; + let test = setup_web_test_components(None).await; mock_request("/block").await; // All required - let resp = client + let resp = test + .client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body(r#"{ "query": "query { blockentity { id height timestamp }}" }"#) @@ -37,7 +31,7 @@ async fn test_entity_with_required_and_optional_fields() { assert!(data[1]["timestamp"].as_u64().unwrap() > 0); // Optionals - let resp = client + let resp = test.client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body(r#"{ "query": "query { optionentity { int_required int_optional_some addr_optional_none }}"}"#) @@ -53,22 +47,18 @@ async fn test_entity_with_required_and_optional_fields() { assert_eq!(data[0]["int_optional_some"], Value::from(Number::from(999))); assert_eq!(data[0]["addr_optional_none"], Value::from(None::<&str>)); - server.abort(); + test.server.abort(); } #[actix_web::test] async fn test_entity_with_foreign_keys() { - let WebTestComponents { - server, - db: _db, - client, - .. - } = setup_web_test_components(None).await; + let test = setup_web_test_components(None).await; mock_request("/block").await; // Implicit foreign keys - let resp = client + let resp = test + .client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body(r#"{ "query": "query { txentity { block { id height } id timestamp } }" }"#) @@ -86,7 +76,8 @@ async fn test_entity_with_foreign_keys() { assert!(data[0]["block"]["height"].as_i64().unwrap() > 0); // Explicit foreign keys - let resp = client + let resp = test + .client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body( @@ -110,17 +101,12 @@ async fn test_entity_with_foreign_keys() { Some("Republic of Indexia") ); - server.abort(); + test.server.abort(); } #[actix_web::test] async fn test_deeply_nested_entity() { - let WebTestComponents { - server, - db: _db, - client, - .. - } = setup_web_test_components(None).await; + let test = setup_web_test_components(None).await; mock_request("/deeply_nested").await; @@ -185,7 +171,8 @@ async fn test_deeply_nested_entity() { }", )]); - let resp = client + let resp = test + .client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .json(&deeply_nested_query) @@ -245,17 +232,12 @@ async fn test_deeply_nested_entity() { Some("Ava") ); - server.abort(); + test.server.abort(); } #[actix_web::test] async fn test_filtering() { - let WebTestComponents { - server, - db: _db, - client, - .. - } = setup_web_test_components(None).await; + let test = setup_web_test_components(None).await; mock_request("/ping").await; @@ -263,7 +245,7 @@ async fn test_filtering() { let _id = uid([1]).to_string(); // ID selection - let resp = client + let resp = test.client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body(r#"{ "query": "query { filterentity(id: \"4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a\") { id foola maybe_null_bar bazoo } }" }"#) @@ -284,7 +266,7 @@ async fn test_filtering() { assert_eq!(data[0]["bazoo"].as_i64(), Some(1)); // Set membership - let resp = client + let resp = test.client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body( @@ -308,7 +290,7 @@ async fn test_filtering() { assert_eq!(data[1]["bazoo"].as_i64(), Some(5)); // Non-null - let resp = client + let resp = test.client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body( @@ -332,7 +314,7 @@ async fn test_filtering() { assert_eq!(data[1]["bazoo"].as_i64(), Some(1000)); // Complex comparison - let resp = client + let resp = test.client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body( @@ -356,7 +338,7 @@ async fn test_filtering() { assert_eq!(data[1]["bazoo"].as_i64(), Some(5)); // Simple comparison - let resp = client + let resp = test.client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body(r#"{ "query": "query { filterentity(filter: { bazoo: { lt: 1000 } } ) { id foola maybe_null_bar bazoo } }" }"#) @@ -378,7 +360,7 @@ async fn test_filtering() { assert_eq!(data[1]["bazoo"].as_i64(), Some(5)); // Nested filters - let resp = client + let resp = test.client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body( @@ -402,7 +384,7 @@ async fn test_filtering() { assert_eq!(data[0]["inner_entity"]["inner_baz"].as_u64(), Some(600)); // Multiple filters on single entity - let resp = client + let resp = test.client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body( @@ -422,7 +404,7 @@ async fn test_filtering() { assert_eq!(data[0]["bazoo"].as_i64(), Some(1)); // Negation - let resp = client + let resp = test.client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body( @@ -441,21 +423,16 @@ async fn test_filtering() { assert_eq!(data[0]["maybe_null_bar"].as_i64(), Some(456)); assert_eq!(data[0]["bazoo"].as_i64(), Some(1000)); - server.abort(); + test.server.abort(); } #[actix_web::test] async fn test_sorting() { - let WebTestComponents { - server, - db: _db, - client, - .. - } = setup_web_test_components(None).await; + let test = setup_web_test_components(None).await; mock_request("/ping").await; - let resp = client + let resp = test.client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body( @@ -476,21 +453,16 @@ async fn test_sorting() { assert_eq!(data[2]["id"].as_str().unwrap(), uid([1]).to_string()); assert_eq!(data[2]["foola"].as_str(), Some("beep")); - server.abort(); + test.server.abort(); } #[actix_web::test] async fn test_aliasing_and_pagination() { - let WebTestComponents { - server, - db: _db, - client, - .. - } = setup_web_test_components(None).await; + let test = setup_web_test_components(None).await; mock_request("/ping").await; - let resp = client + let resp = test.client .post("http://127.0.0.1:29987/api/graph/fuel_indexer_test/index1") .header(CONTENT_TYPE, "application/graphql".to_owned()) .body( @@ -510,5 +482,5 @@ async fn test_aliasing_and_pagination() { ); assert_eq!(data[0]["page_info"]["pages"].as_i64(), Some(3)); - server.abort(); + test.server.abort(); } diff --git a/packages/fuel-indexer-tests/tests/service.rs b/packages/fuel-indexer-tests/tests/service.rs index 20a306aff..28c29add1 100644 --- a/packages/fuel-indexer-tests/tests/service.rs +++ b/packages/fuel-indexer-tests/tests/service.rs @@ -3,7 +3,6 @@ use fuel_indexer::{Executor, IndexerConfig, WasmIndexExecutor}; use fuel_indexer_lib::{config::DatabaseConfig, manifest::Manifest}; use fuel_indexer_tests::fixtures::TestPostgresDb; use std::str::FromStr; -use std::sync::atomic::AtomicBool; #[tokio::test] async fn test_wasm_executor_can_meter_execution() { @@ -58,11 +57,9 @@ async fn test_wasm_executor_can_meter_execution() { .await .unwrap(); - let kill_switch = std::sync::Arc::new(AtomicBool::new(false)); - let blocks: Vec = vec![]; - if let Err(e) = executor.handle_events(kill_switch, blocks.clone()).await { + if let Err(e) = executor.handle_events(blocks.clone()).await { if let fuel_indexer::IndexerError::RuntimeError(e) = e { if let Some(e) = e.to_trap() { assert_eq!(e, wasmer_types::TrapCode::UnreachableCodeReached); diff --git a/packages/fuel-indexer/src/commands/run.rs b/packages/fuel-indexer/src/commands/run.rs index cbf030a26..8d233b128 100644 --- a/packages/fuel-indexer/src/commands/run.rs +++ b/packages/fuel-indexer/src/commands/run.rs @@ -130,7 +130,12 @@ pub async fn exec(args: IndexerArgs) -> anyhow::Result<()> { } } - subsystems.spawn(service.run()); + subsystems.spawn(async { + let result = service.run().await; + if let Err(e) = result { + tracing::error!("Indexer Service failed: {e}"); + } + }); #[cfg(feature = "api-server")] subsystems.spawn({ diff --git a/packages/fuel-indexer/src/executor.rs b/packages/fuel-indexer/src/executor.rs index 67267494d..726e34954 100644 --- a/packages/fuel-indexer/src/executor.rs +++ b/packages/fuel-indexer/src/executor.rs @@ -3,11 +3,7 @@ use crate::{ database::Database, ffi, queries::ClientExt, IndexerConfig, IndexerError, IndexerResult, }; -use async_std::{ - fs::File, - io::ReadExt, - sync::{Arc, Mutex}, -}; +use async_std::sync::{Arc, Mutex}; use async_trait::async_trait; use fuel_core_client::client::{ pagination::{PageDirection, PaginatedResult, PaginationRequest}, @@ -33,7 +29,7 @@ use std::{ sync::atomic::{AtomicBool, Ordering}, }; use tokio::{ - task::{spawn_blocking, JoinHandle}, + task::spawn_blocking, time::{sleep, Duration}, }; use tracing::{debug, error, info, warn}; @@ -78,24 +74,23 @@ impl From for Vec { // types in `fuel_core_client` don't compile to WASM. pub fn run_executor( config: &IndexerConfig, - manifest: &Manifest, mut executor: T, - kill_switch: Arc, ) -> impl Future { // TODO: https://github.com/FuelLabs/fuel-indexer/issues/286 - let end_block = manifest.end_block(); + let end_block = executor.manifest().end_block(); let stop_idle_indexers = config.stop_idle_indexers; - let indexer_uid = manifest.uid(); + let indexer_uid = executor.manifest().uid(); let node_block_page_size = config.node_block_page_size; - let fuel_node_addr = manifest + let fuel_node_addr = executor + .manifest() .fuel_client() .map(|x| x.to_string()) .unwrap_or(config.fuel_node.to_string()); // Where should we initially start when fetching blocks from the client? - let mut cursor = manifest.start_block().map(|x| { + let mut cursor = executor.manifest().start_block().map(|x| { if x > 1 { let decremented = x - 1; decremented.to_string() @@ -138,7 +133,7 @@ pub fn run_executor( loop { // If something else has signaled that this indexer should stop, then stop. - if kill_switch.load(Ordering::SeqCst) { + if executor.kill_switch().load(Ordering::SeqCst) { info!("Kill switch flipped, stopping Indexer({indexer_uid}). <('.')>"); break; } @@ -183,9 +178,7 @@ pub fn run_executor( } // The client responded with actual blocks, so attempt to index them. - let result = executor - .handle_events(kill_switch.clone(), block_info) - .await; + let result = executor.handle_events(block_info).await; if let Err(e) = result { // Run time metering is deterministic. There is no point in retrying. @@ -229,7 +222,7 @@ pub fn run_executor( cursor = next_cursor; // Again, check if something else has signaled that this indexer should stop, then stop. - if kill_switch.load(Ordering::SeqCst) { + if executor.kill_switch().load(Ordering::SeqCst) { info!("Kill switch flipped, stopping Indexer({indexer_uid}). <('.')>"); break; } @@ -480,11 +473,11 @@ pub trait Executor where Self: Sized, { - async fn handle_events( - &mut self, - kill_switch: Arc, - blocks: Vec, - ) -> IndexerResult<()>; + async fn handle_events(&mut self, blocks: Vec) -> IndexerResult<()>; + + fn manifest(&self) -> &Manifest; + + fn kill_switch(&self) -> &Arc; } /// WASM indexer runtime environment responsible for fetching/saving data to and from the database. @@ -535,6 +528,9 @@ where /// Function that handles events. handle_events_fn: fn(Vec, Arc>) -> F, + + /// Kill switch. When set to true, the indexer must stop execution. + kill_switch: Arc, } impl NativeIndexExecutor @@ -557,31 +553,23 @@ where ) .await?; db.load_schema(version).await?; + let kill_switch = Arc::new(AtomicBool::new(false)); Ok(Self { db: Arc::new(Mutex::new(db)), manifest: manifest.to_owned(), handle_events_fn, + kill_switch, }) } /// Create a new `NativeIndexExecutor`. - pub async fn create> + Send + 'static>( + pub async fn create( config: &IndexerConfig, manifest: &Manifest, pool: IndexerConnectionPool, - handle_events: fn(Vec, Arc>) -> T, - ) -> IndexerResult<(JoinHandle<()>, ExecutorSource, Arc)> { - let executor = - NativeIndexExecutor::new(manifest, pool.clone(), config, handle_events) - .await?; - let kill_switch = Arc::new(AtomicBool::new(false)); - let handle = tokio::spawn(run_executor( - config, - manifest, - executor, - kill_switch.clone(), - )); - Ok((handle, ExecutorSource::Manifest, kill_switch)) + handle_events: fn(Vec, Arc>) -> F, + ) -> IndexerResult { + NativeIndexExecutor::new(manifest, pool.clone(), config, handle_events).await } } @@ -591,11 +579,7 @@ where F: Future> + Send, { /// Handle events for native executor. - async fn handle_events( - &mut self, - kill_switch: Arc, - blocks: Vec, - ) -> IndexerResult<()> { + async fn handle_events(&mut self, blocks: Vec) -> IndexerResult<()> { self.db.lock().await.start_transaction().await?; let res = (self.handle_events_fn)(blocks, self.db.clone()).await; let uid = self.manifest.uid(); @@ -605,7 +589,7 @@ where return Err(IndexerError::NativeExecutionRuntimeError); } else { // Do not commit if kill switch has been triggered. - if kill_switch.load(Ordering::SeqCst) { + if self.kill_switch.load(Ordering::SeqCst) { self.db.lock().await.revert_transaction().await?; } else { self.db.lock().await.commit_transaction().await?; @@ -613,6 +597,14 @@ where } Ok(()) } + + fn kill_switch(&self) -> &Arc { + &self.kill_switch + } + + fn manifest(&self) -> &Manifest { + &self.manifest + } } /// WASM executors are the primary means of execution. @@ -639,6 +631,9 @@ pub struct WasmIndexExecutor { /// Manifest of the indexer. manifest: Manifest, + + /// Kill switch. When set to true, the indexer must stop execution. + kill_switch: Arc, } impl WasmIndexExecutor { @@ -715,6 +710,8 @@ impl WasmIndexExecutor { db.lock().await.load_schema(schema_version).await?; + let kill_switch = Arc::new(AtomicBool::new(false)); + Ok(WasmIndexExecutor { instance, _module: module, @@ -722,6 +719,7 @@ impl WasmIndexExecutor { db: db.clone(), metering_points: config.metering_points, manifest: manifest.clone(), + kill_switch, }) } @@ -742,76 +740,21 @@ impl WasmIndexExecutor { pub async fn create( config: &IndexerConfig, manifest: &Manifest, - exec_source: ExecutorSource, pool: IndexerConnectionPool, schema_version: String, - ) -> IndexerResult<(JoinHandle<()>, ExecutorSource, Arc)> { - let killer = Arc::new(AtomicBool::new(false)); + wasm_bytes: impl AsRef<[u8]>, + ) -> IndexerResult { let uid = manifest.uid(); - match &exec_source { - ExecutorSource::Manifest => match manifest.module() { - crate::Module::Wasm(ref module) => { - let mut bytes = Vec::::new(); - let mut file = File::open(module).await?; - file.read_to_end(&mut bytes).await?; - - match WasmIndexExecutor::new( - config, - manifest, - bytes.clone(), - pool, - schema_version, - ) - .await - { - Ok(executor) => { - let handle = tokio::spawn(run_executor( - config, - manifest, - executor, - killer.clone(), - )); - - Ok((handle, ExecutorSource::Registry(bytes), killer)) - } - Err(e) => { - error!( - "Could not instantiate WasmIndexExecutor({uid}) from ExecutorSource::Manifest: {e:?}." - ); - Err(IndexerError::WasmExecutionInstantiationError) - } - } - } - crate::Module::Native => { - Err(IndexerError::NativeExecutionInstantiationError) - } - }, - ExecutorSource::Registry(bytes) => { - match WasmIndexExecutor::new( - config, - manifest, - bytes, - pool, - schema_version, - ) - .await - { - Ok(executor) => { - let handle = tokio::spawn(run_executor( - config, - manifest, - executor, - killer.clone(), - )); - - Ok((handle, exec_source, killer)) - } - Err(e) => { - error!("Could not instantiate WasmIndexExecutor({uid}) from ExecutorSource::Registry: {e:?}."); - Err(IndexerError::WasmExecutionInstantiationError) - } - } + match WasmIndexExecutor::new(config, manifest, wasm_bytes, pool, schema_version) + .await + { + Ok(executor) => Ok(executor), + Err(e) => { + error!( + "Could not instantiate WasmIndexExecutor({uid}) from ExecutorSource::Manifest: {e:?}." + ); + Err(IndexerError::WasmExecutionInstantiationError) } } } @@ -868,11 +811,7 @@ impl WasmIndexExecutor { #[async_trait] impl Executor for WasmIndexExecutor { /// Trigger a WASM event handler, passing in a serialized event struct. - async fn handle_events( - &mut self, - kill_switch: Arc, - blocks: Vec, - ) -> IndexerResult<()> { + async fn handle_events(&mut self, blocks: Vec) -> IndexerResult<()> { if blocks.is_empty() { return Ok(()); } @@ -923,7 +862,7 @@ impl Executor for WasmIndexExecutor { } } else { // Do not commit if kill switch has been triggered. - if kill_switch.load(Ordering::SeqCst) { + if self.kill_switch.load(Ordering::SeqCst) { self.db.lock().await.revert_transaction().await?; } else { self.db.lock().await.commit_transaction().await?; @@ -932,4 +871,12 @@ impl Executor for WasmIndexExecutor { Ok(()) } + + fn kill_switch(&self) -> &Arc { + &self.kill_switch + } + + fn manifest(&self) -> &Manifest { + &self.manifest + } } diff --git a/packages/fuel-indexer/src/service.rs b/packages/fuel-indexer/src/service.rs index 85cc5f29d..319befb71 100644 --- a/packages/fuel-indexer/src/service.rs +++ b/packages/fuel-indexer/src/service.rs @@ -1,24 +1,21 @@ use crate::{ - executor::{ExecutorSource, NativeIndexExecutor, WasmIndexExecutor}, - Database, IndexerConfig, IndexerError, IndexerResult, Manifest, + executor::{NativeIndexExecutor, WasmIndexExecutor}, + Database, Executor, IndexerConfig, IndexerError, IndexerResult, Manifest, }; use async_std::sync::{Arc, Mutex}; +use async_std::{fs::File, io::ReadExt}; use fuel_indexer_database::{ queries, types::IndexerAssetType, IndexerConnection, IndexerConnectionPool, }; use fuel_indexer_lib::{defaults, utils::ServiceRequest}; use fuel_indexer_schema::db::manager::SchemaManager; use fuel_indexer_types::fuel::BlockData; -use futures::{ - stream::{FuturesUnordered, StreamExt}, - Future, -}; +use futures::Future; use std::collections::HashMap; use std::marker::Send; use std::sync::atomic::{AtomicBool, Ordering}; use tokio::{ sync::mpsc::Receiver, - task::JoinHandle, time::{sleep, Duration}, }; use tracing::{debug, error, info, warn}; @@ -34,8 +31,8 @@ pub struct IndexerService { /// Schema manager used to manage the database schema. manager: SchemaManager, - /// Handles to the spawned indexers. - handles: HashMap>, + /// Tasks for the spawned indexers. + tasks: tokio::task::JoinSet<()>, /// Channel used to receive `ServiceRequest`s. rx: Receiver, @@ -57,8 +54,8 @@ impl IndexerService { config, pool, manager, - handles: HashMap::default(), killers: HashMap::default(), + tasks: tokio::task::JoinSet::new(), rx, }) } @@ -128,17 +125,29 @@ impl IndexerService { let start_block = get_start_block(&mut conn, &manifest).await?; manifest.set_start_block(start_block); - let (handle, exec_source, killer) = WasmIndexExecutor::create( + let wasm_bytes = match manifest.module() { + crate::Module::Wasm(ref module) => { + let mut bytes = Vec::::new(); + let mut file = File::open(module).await?; + file.read_to_end(&mut bytes).await?; + bytes + } + crate::Module::Native => { + return Err(IndexerError::NativeExecutionInstantiationError) + } + }; + + let executor = WasmIndexExecutor::create( &self.config, &manifest, - ExecutorSource::Manifest, self.pool.clone(), schema_version, + wasm_bytes.clone(), ) .await?; let mut items = vec![ - (IndexerAssetType::Wasm, exec_source.into()), + (IndexerAssetType::Wasm, wasm_bytes), (IndexerAssetType::Manifest, manifest.clone().into()), (IndexerAssetType::Schema, schema_bytes), ]; @@ -166,8 +175,8 @@ impl IndexerService { manifest.namespace(), manifest.identifier() ); - self.handles.insert(manifest.uid(), handle); - self.killers.insert(manifest.uid(), killer); + + self.start_executor(executor); Ok(()) } @@ -183,18 +192,18 @@ impl IndexerService { let start_block = get_start_block(&mut conn, &manifest).await.unwrap_or(1); manifest.set_start_block(start_block); - if let Ok((handle, _module_bytes, killer)) = WasmIndexExecutor::create( + if let Ok(executor) = WasmIndexExecutor::create( &self.config, &manifest, - ExecutorSource::Registry(assets.wasm.bytes), self.pool.clone(), assets.schema.digest, + assets.wasm.bytes, ) .await { info!("Registered Indexer({})", manifest.uid()); - self.handles.insert(manifest.uid(), handle); - self.killers.insert(manifest.uid(), killer); + + self.start_executor(executor); } else { error!( "Failed to register Indexer({}) from registry.", @@ -237,7 +246,7 @@ impl IndexerService { manifest.set_start_block(start_block); let uid = manifest.uid(); - let (handle, _module_bytes, killer) = NativeIndexExecutor::::create( + let executor = NativeIndexExecutor::::create( &self.config, &manifest, self.pool.clone(), @@ -247,131 +256,107 @@ impl IndexerService { info!("Registered NativeIndex({})", uid); - self.handles.insert(uid.clone(), handle); - self.killers.insert(uid, killer); + self.start_executor(executor); + Ok(()) } - /// Kick it off! - pub async fn run(self) { - let IndexerService { - handles, - rx, - pool, - config, - killers, - .. - } = self; - - let futs = Arc::new(Mutex::new(FuturesUnordered::from_iter( - handles.into_values(), - ))); - - let _ = tokio::spawn(create_service_task( - rx, - config.clone(), - pool.clone(), - futs.clone(), - killers, - )) - .await - .unwrap(); - - while let Some(fut) = futs.lock().await.next().await { - info!("Retired a future {fut:?}"); - } - } -} + /// Kick it off! Run the indexer service loop, listening to service messages primarily coming from the web API. + pub async fn run(mut self) -> IndexerResult<()> { + loop { + match self.rx.try_recv() { + Ok(service_request) => match service_request { + ServiceRequest::Reload(request) => { + let mut conn = self.pool.acquire().await?; + + match queries::get_indexer_id( + &mut conn, + &request.namespace, + &request.identifier, + ) + .await + { + Ok(id) => { + let assets = + queries::latest_assets_for_indexer(&mut conn, &id) + .await?; + let mut manifest = + Manifest::try_from(&assets.manifest.bytes)?; + + let start_block = + get_start_block(&mut conn, &manifest).await?; + manifest.set_start_block(start_block); + + if let Some(killer_for_prev_executor) = + self.killers.remove(&manifest.uid()) + { + let uid = manifest.uid(); + info!("Indexer({uid}) is being replaced. Stopping previous version of Indexer({uid})."); + killer_for_prev_executor + .store(true, Ordering::SeqCst); + } -/// Create a tokio task used to listen to service messages primarily coming from the web API. -async fn create_service_task( - mut rx: Receiver, - config: IndexerConfig, - pool: IndexerConnectionPool, - futs: Arc>>>, - mut killers: HashMap>, -) -> IndexerResult<()> { - loop { - let futs = futs.lock().await; - match rx.try_recv() { - Ok(service_request) => match service_request { - ServiceRequest::Reload(request) => { - let mut conn = pool.acquire().await?; - - match queries::get_indexer_id( - &mut conn, - &request.namespace, - &request.identifier, - ) - .await - { - Ok(id) => { - let assets = - queries::latest_assets_for_indexer(&mut conn, &id) - .await?; - let mut manifest = - Manifest::try_from(&assets.manifest.bytes)?; - - let start_block = - get_start_block(&mut conn, &manifest).await?; - manifest.set_start_block(start_block); - - match WasmIndexExecutor::create( - &config, - &manifest, - ExecutorSource::Registry(assets.wasm.bytes), - pool.clone(), - assets.schema.digest, - ) - .await - { - Ok((handle, _module_bytes, killer)) => { - futs.push(handle); - - if let Some(killer_for_prev_executor) = - killers.insert(manifest.uid(), killer) - { - let uid = manifest.uid(); - info!("Indexer({uid}) was replaced. Stopping previous version of Indexer({uid})."); - killer_for_prev_executor - .store(true, Ordering::SeqCst); + match WasmIndexExecutor::create( + &self.config, + &manifest, + self.pool.clone(), + assets.schema.digest, + assets.wasm.bytes, + ) + .await + { + Ok(executor) => self.start_executor(executor), + Err(e) => { + error!( + "Failed to reload Indexer({}.{}): {e:?}", + &request.namespace, &request.identifier + ); + return Ok(()); } } - Err(e) => { - error!( - "Failed to reload Indexer({}.{}): {e:?}", - &request.namespace, &request.identifier - ); - return Ok(()); - } } - } - Err(e) => { - error!( - "Failed to find Indexer({}.{}): {}", - &request.namespace, &request.identifier, e - ); + Err(e) => { + error!( + "Failed to find Indexer({}.{}): {}", + &request.namespace, &request.identifier, e + ); - continue; + continue; + } } } - } - ServiceRequest::Stop(request) => { - let uid = format!("{}.{}", request.namespace, request.identifier); - - if let Some(killer) = killers.remove(&uid) { - killer.store(true, Ordering::SeqCst); - } else { - warn!("Stop Indexer: No indexer with the name Indexer({uid})"); + ServiceRequest::Stop(request) => { + let uid = format!("{}.{}", request.namespace, request.identifier); + + if let Some(killer) = self.killers.remove(&uid) { + killer.store(true, Ordering::SeqCst); + } else { + warn!( + "Stop Indexer: No indexer with the name Indexer({uid})" + ); + } } + }, + Err(e) => { + debug!("No service request to handle: {e:?}."); + sleep(Duration::from_secs(defaults::IDLE_SERVICE_WAIT_SECS)).await; } - }, - Err(e) => { - debug!("No service request to handle: {e:?}."); - sleep(Duration::from_secs(defaults::IDLE_SERVICE_WAIT_SECS)).await; } } } + + // Spawn and register a tokio::task running the Executor loop, as well as + // the kill switch and the abort handle.ß + fn start_executor(&mut self, executor: T) { + let uid = executor.manifest().uid(); + + self.killers + .insert(uid.clone(), executor.kill_switch().clone()); + + self + .tasks + .spawn(crate::executor::run_executor(&self.config, executor)); + } } /// Determine the starting block for this indexer. From d97ccbcc1963cd36d12a1f6f88f3bfd41f6ae55b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Wo=C5=9B?= Date: Mon, 4 Sep 2023 17:54:29 +0200 Subject: [PATCH 3/8] fuel-indexer: prune finished tasks from tasks set --- packages/fuel-indexer/src/service.rs | 141 ++++++++++++++------------- 1 file changed, 71 insertions(+), 70 deletions(-) diff --git a/packages/fuel-indexer/src/service.rs b/packages/fuel-indexer/src/service.rs index 319befb71..3e33515b7 100644 --- a/packages/fuel-indexer/src/service.rs +++ b/packages/fuel-indexer/src/service.rs @@ -264,82 +264,84 @@ impl IndexerService { /// Kick it off! Run the indexer service loop, listening to service messages primarily coming from the web API. pub async fn run(mut self) -> IndexerResult<()> { loop { - match self.rx.try_recv() { - Ok(service_request) => match service_request { - ServiceRequest::Reload(request) => { - let mut conn = self.pool.acquire().await?; - - match queries::get_indexer_id( - &mut conn, - &request.namespace, - &request.identifier, - ) - .await - { - Ok(id) => { - let assets = - queries::latest_assets_for_indexer(&mut conn, &id) - .await?; - let mut manifest = - Manifest::try_from(&assets.manifest.bytes)?; - - let start_block = - get_start_block(&mut conn, &manifest).await?; - manifest.set_start_block(start_block); - - if let Some(killer_for_prev_executor) = - self.killers.remove(&manifest.uid()) - { - let uid = manifest.uid(); - info!("Indexer({uid}) is being replaced. Stopping previous version of Indexer({uid})."); - killer_for_prev_executor - .store(true, Ordering::SeqCst); - } + tokio::select! { + // Calling join_next will remove finished tasks from the set. + Some(Err(e)) = self.tasks.join_next() => { + error!("Error retiring indexer task {e}"); + } + Some(service_request) = self.rx.recv() => { + match service_request { + ServiceRequest::Reload(request) => { + let mut conn = self.pool.acquire().await?; + + match queries::get_indexer_id( + &mut conn, + &request.namespace, + &request.identifier, + ) + .await + { + Ok(id) => { + let assets = + queries::latest_assets_for_indexer(&mut conn, &id) + .await?; + let mut manifest = + Manifest::try_from(&assets.manifest.bytes)?; + + let start_block = + get_start_block(&mut conn, &manifest).await?; + manifest.set_start_block(start_block); + + if let Some(killer_for_prev_executor) = + self.killers.remove(&manifest.uid()) + { + let uid = manifest.uid(); + info!("Indexer({uid}) is being replaced. Stopping previous version of Indexer({uid})."); + killer_for_prev_executor + .store(true, Ordering::SeqCst); + } - match WasmIndexExecutor::create( - &self.config, - &manifest, - self.pool.clone(), - assets.schema.digest, - assets.wasm.bytes, - ) - .await - { - Ok(executor) => self.start_executor(executor), - Err(e) => { - error!( - "Failed to reload Indexer({}.{}): {e:?}", - &request.namespace, &request.identifier - ); - return Ok(()); + match WasmIndexExecutor::create( + &self.config, + &manifest, + self.pool.clone(), + assets.schema.digest, + assets.wasm.bytes, + ) + .await + { + Ok(executor) => self.start_executor(executor), + Err(e) => { + error!( + "Failed to reload Indexer({}.{}): {e:?}", + &request.namespace, &request.identifier + ); + return Ok(()); + } } } - } - Err(e) => { - error!( - "Failed to find Indexer({}.{}): {}", - &request.namespace, &request.identifier, e - ); + Err(e) => { + error!( + "Failed to find Indexer({}.{}): {}", + &request.namespace, &request.identifier, e + ); - continue; + continue; + } } } - } - ServiceRequest::Stop(request) => { - let uid = format!("{}.{}", request.namespace, request.identifier); - - if let Some(killer) = self.killers.remove(&uid) { - killer.store(true, Ordering::SeqCst); - } else { - warn!( - "Stop Indexer: No indexer with the name Indexer({uid})" - ); + ServiceRequest::Stop(request) => { + let uid = format!("{}.{}", request.namespace, request.identifier); + + if let Some(killer) = self.killers.remove(&uid) { + killer.store(true, Ordering::SeqCst); + } else { + warn!( + "Stop Indexer: No indexer with the name Indexer({uid})" + ); + } } } - }, - Err(e) => { - debug!("No service request to handle: {e:?}."); - sleep(Duration::from_secs(defaults::IDLE_SERVICE_WAIT_SECS)).await; } } } @@ -353,8 +355,7 @@ impl IndexerService { self.killers .insert(uid.clone(), executor.kill_switch().clone()); - self - .tasks + self.tasks .spawn(crate::executor::run_executor(&self.config, executor)); } } From 87ee7030e0f48827e4dcd62656be0d63457eba97 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Wo=C5=9B?= Date: Tue, 5 Sep 2023 10:59:36 +0200 Subject: [PATCH 4/8] clippy and fmt --- packages/fuel-indexer/src/service.rs | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/packages/fuel-indexer/src/service.rs b/packages/fuel-indexer/src/service.rs index 3e33515b7..b816ae559 100644 --- a/packages/fuel-indexer/src/service.rs +++ b/packages/fuel-indexer/src/service.rs @@ -7,18 +7,15 @@ use async_std::{fs::File, io::ReadExt}; use fuel_indexer_database::{ queries, types::IndexerAssetType, IndexerConnection, IndexerConnectionPool, }; -use fuel_indexer_lib::{defaults, utils::ServiceRequest}; +use fuel_indexer_lib::utils::ServiceRequest; use fuel_indexer_schema::db::manager::SchemaManager; use fuel_indexer_types::fuel::BlockData; use futures::Future; use std::collections::HashMap; use std::marker::Send; use std::sync::atomic::{AtomicBool, Ordering}; -use tokio::{ - sync::mpsc::Receiver, - time::{sleep, Duration}, -}; -use tracing::{debug, error, info, warn}; +use tokio::sync::mpsc::Receiver; +use tracing::{error, info, warn}; /// Primary service used to run one or many indexers. pub struct IndexerService { From 13c7746c9a36b92b9e3a5378e5c8b4124cae9768 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Wo=C5=9B?= Date: Tue, 12 Sep 2023 10:21:47 +0200 Subject: [PATCH 5/8] review feedback --- packages/fuel-indexer/src/commands/run.rs | 8 +++++--- packages/fuel-indexer/src/executor.rs | 2 +- packages/fuel-indexer/src/service.rs | 2 +- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/packages/fuel-indexer/src/commands/run.rs b/packages/fuel-indexer/src/commands/run.rs index 8d233b128..d41c5a080 100644 --- a/packages/fuel-indexer/src/commands/run.rs +++ b/packages/fuel-indexer/src/commands/run.rs @@ -13,7 +13,7 @@ use tracing::info; #[cfg(feature = "api-server")] use fuel_indexer_api_server::api::WebApi; -// Returns a future which completes when a shutdown signal has been reveived. +// Returns a future which completes when a shutdown signal has been received. fn shutdown_signal_handler() -> std::io::Result> { let mut sighup: Signal = signal(SignalKind::hangup())?; let mut sigterm: Signal = signal(SignalKind::terminate())?; @@ -159,7 +159,7 @@ pub async fn exec(args: IndexerArgs) -> anyhow::Result<()> { }; subsystems.spawn(async move { if let Err(e) = FuelService::new_node(config).await { - tracing::error!("Fuen Node failed: {e}"); + tracing::error!("Fuel Node failed: {e}"); }; }); } @@ -168,7 +168,9 @@ pub async fn exec(args: IndexerArgs) -> anyhow::Result<()> { // Each subsystem runs its own loop, and we require all subsystems for the // Indexer service to operate correctly. If any of the subsystems stops // running, the entire Indexer Service exits. - subsystems.join_next().await; + if let Some(_) = subsystems.join_next().await { + subsystems.shutdown().await; + } if embedded_database { let name = postgres_database.unwrap_or(defaults::POSTGRES_DATABASE.to_string()); diff --git a/packages/fuel-indexer/src/executor.rs b/packages/fuel-indexer/src/executor.rs index 726e34954..c750c4683 100644 --- a/packages/fuel-indexer/src/executor.rs +++ b/packages/fuel-indexer/src/executor.rs @@ -752,7 +752,7 @@ impl WasmIndexExecutor { Ok(executor) => Ok(executor), Err(e) => { error!( - "Could not instantiate WasmIndexExecutor({uid}) from ExecutorSource::Manifest: {e:?}." + "Could not instantiate WasmIndexExecutor({uid}): {e:?}." ); Err(IndexerError::WasmExecutionInstantiationError) } diff --git a/packages/fuel-indexer/src/service.rs b/packages/fuel-indexer/src/service.rs index 3d3cbdfec..a7a01e49f 100644 --- a/packages/fuel-indexer/src/service.rs +++ b/packages/fuel-indexer/src/service.rs @@ -345,7 +345,7 @@ impl IndexerService { } // Spawn and register a tokio::task running the Executor loop, as well as - // the kill switch and the abort handle.ß + // the kill switch and the abort handle. fn start_executor(&mut self, executor: T) { let uid = executor.manifest().uid(); From 2ee3ffa027c05cbf702968840331c8c9e8c3f22f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Wo=C5=9B?= Date: Tue, 12 Sep 2023 10:23:22 +0200 Subject: [PATCH 6/8] remove comment --- .../postgres/migrations/20230911123600_remove_versions.down.sql | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/fuel-indexer-database/postgres/migrations/20230911123600_remove_versions.down.sql b/packages/fuel-indexer-database/postgres/migrations/20230911123600_remove_versions.down.sql index 63d324d38..bc3e95469 100644 --- a/packages/fuel-indexer-database/postgres/migrations/20230911123600_remove_versions.down.sql +++ b/packages/fuel-indexer-database/postgres/migrations/20230911123600_remove_versions.down.sql @@ -1,4 +1,3 @@ --- Active: 1693382282533@@127.0.0.1@5432@postgres ALTER TABLE index_asset_registry_manifest ADD COLUMN version integer not null; ALTER TABLE index_asset_registry_schema ADD COLUMN version integer not null; ALTER TABLE index_asset_registry_wasm ADD COLUMN version integer not null; \ No newline at end of file From 4f7d466beb2e73ed687976de8642621e9cd8c248 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Wo=C5=9B?= Date: Tue, 12 Sep 2023 10:24:28 +0200 Subject: [PATCH 7/8] fmt --- packages/fuel-indexer/src/executor.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/packages/fuel-indexer/src/executor.rs b/packages/fuel-indexer/src/executor.rs index c750c4683..9b5de4e09 100644 --- a/packages/fuel-indexer/src/executor.rs +++ b/packages/fuel-indexer/src/executor.rs @@ -751,9 +751,7 @@ impl WasmIndexExecutor { { Ok(executor) => Ok(executor), Err(e) => { - error!( - "Could not instantiate WasmIndexExecutor({uid}): {e:?}." - ); + error!("Could not instantiate WasmIndexExecutor({uid}): {e:?}."); Err(IndexerError::WasmExecutionInstantiationError) } } From 54253e4fe2313ba527de3446c3ac0a2c29044281 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Wo=C5=9B?= Date: Tue, 12 Sep 2023 11:26:52 +0200 Subject: [PATCH 8/8] clippy --- packages/fuel-indexer/src/commands/run.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/fuel-indexer/src/commands/run.rs b/packages/fuel-indexer/src/commands/run.rs index d41c5a080..e8b49c202 100644 --- a/packages/fuel-indexer/src/commands/run.rs +++ b/packages/fuel-indexer/src/commands/run.rs @@ -168,7 +168,7 @@ pub async fn exec(args: IndexerArgs) -> anyhow::Result<()> { // Each subsystem runs its own loop, and we require all subsystems for the // Indexer service to operate correctly. If any of the subsystems stops // running, the entire Indexer Service exits. - if let Some(_) = subsystems.join_next().await { + if subsystems.join_next().await.is_some() { subsystems.shutdown().await; }