Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 65 additions & 1 deletion crates/chain-gateway/src/chain_gateway.rs
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
use std::future::Future;
use std::path::Path;

use near_account_id::AccountId;
Expand Down Expand Up @@ -153,7 +154,7 @@ fn run_node(
ready_sender: tokio::sync::oneshot::Sender<RunNodeResult>,
near_config: nearcore::NearConfig,
home_dir: &Path,
shutdown_receiver: tokio::sync::oneshot::Receiver<()>,
mut shutdown_receiver: tokio::sync::oneshot::Receiver<()>,
streamer_setup: Option<StreamerSetup>,
) {
let rt = tokio::runtime::Builder::new_multi_thread()
Expand Down Expand Up @@ -185,6 +186,24 @@ fn run_node(
let rpc_handler = NearRpcActorHandle::new(near_node.rpc_handler);

let stream = if let Some((indexer, streamer_config)) = indexer_and_params {
// Don't start the streamer until synced, or its `LatestSynced`
// cursor pins at genesis. Raced against shutdown to stay responsive.
tracing::info!(
"chain-gateway waiting for node to finish syncing before streaming blocks."
);
let synced = await_sync_or_shutdown(client.wait_for_full_sync(), async {
let _ = (&mut shutdown_receiver).await;
})
.await;
if !synced {
tracing::info!("shutdown requested before sync completed; exiting startup.");
let _ = ready_sender.send(Err(ChainGatewayError::StartupFailed {
msg: "shutdown requested before node finished syncing".to_string(),
}));
actor_system.stop();
return;
}

let raw_stream: Receiver<StreamerMessage> = indexer.streamer();
match event_subscriber::streamer::start(
streamer_config,
Expand Down Expand Up @@ -232,3 +251,48 @@ struct StreamerSetup {
indexer_config: near_indexer::IndexerConfig,
near_config: NearConfig,
}

async fn await_sync_or_shutdown(
sync: impl Future<Output = ()>,
shutdown: impl Future<Output = ()>,
) -> bool {
tokio::select! {
_ = sync => true,
_ = shutdown => false,
}
}

#[cfg(test)]
#[expect(non_snake_case)]
mod tests {
use super::await_sync_or_shutdown;
use std::future::pending;

#[tokio::test]
async fn await_sync_or_shutdown__should_return_true_when_sync_completes_first() {
// Given
let sync = async {};
let shutdown = pending::<()>();

// When
let synced = await_sync_or_shutdown(sync, shutdown).await;

// Then
assert!(synced);
}

/// A shutdown (or abandoned startup) arriving while the node is still
/// syncing must win the race rather than block on sync.
#[tokio::test]
async fn await_sync_or_shutdown__should_return_false_when_shutdown_first() {
// Given
let sync = pending::<()>();
let shutdown = async {};

// When
let synced = await_sync_or_shutdown(sync, shutdown).await;

// Then
assert!(!synced);
}
}
77 changes: 75 additions & 2 deletions crates/node/src/indexer/real.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ use near_account_id::AccountId;
use near_async::ActorSystem;
use near_indexer::Indexer;
use near_mpc_contract_interface::types::ProtocolContractState;
use std::future::Future;
use std::path::PathBuf;
use std::sync::Arc;
#[cfg(feature = "network-hardship-simulation")]
Expand Down Expand Up @@ -110,15 +111,41 @@ pub fn spawn_real_indexer(

let indexer = Indexer::from_near_node(near_indexer_config, near_config, &near_node);

let stream = indexer.streamer();
Comment thread
gilcu3 marked this conversation as resolved.

let indexer_state = Arc::new(IndexerState::new(
near_node.view_client,
near_node.client,
near_node.rpc_handler,
mpc_indexer_config.mpc_contract_id.clone(),
));

tracing::info!("Indexer waiting for node to finish syncing before streaming blocks.");

// Defer all indexing work until the node has finished syncing. On a
// node that state-syncs from scratch, neard sits at genesis until
// sync completes; starting the streamer then pins its cursor at
// genesis (`LatestSynced`), below the node's block tail, so it can
// never reach the chain tip. (#3623)
//
// The wait is raced against `shutdown_token` so a SIGTERM during the
// (possibly long) initial state sync still tears the thread down
// cleanly instead of blocking until sync finishes.
if !await_sync_or_shutdown(
indexer_state.client.wait_for_full_sync(),
&shutdown_token,
)
.await
{
tracing::info!(
"Indexer thread received shutdown signal before sync completed; exiting."
);
let _ = indexer_exit_sender.send(Ok(()));
return;
}

// The node is fully synced by this point, so `LatestSynced` resolves
// to the chain tip rather than genesis.
let stream = indexer.streamer();

let txn_sender_result = TransactionProcessorHandle::start_transaction_processor(
my_near_account_id_clone,
account_secret_key.clone(),
Expand Down Expand Up @@ -289,3 +316,49 @@ pub fn spawn_real_indexer(
foreign_chain_policy_reader,
}
}

async fn await_sync_or_shutdown(
sync: impl Future<Output = ()>,
shutdown: &CancellationToken,
) -> bool {
tokio::select! {
_ = sync => true,
_ = shutdown.cancelled() => false,
}
}

#[cfg(test)]
#[expect(non_snake_case)]
mod tests {
use super::await_sync_or_shutdown;
use std::future::pending;
use tokio_util::sync::CancellationToken;

#[tokio::test]
async fn await_sync_or_shutdown__should_return_true_when_sync_completes_first() {
// Given
let shutdown = CancellationToken::new();

// When
let synced = await_sync_or_shutdown(async {}, &shutdown).await;

// Then
assert!(synced);
}

/// The dominant production path: a SIGTERM arrives while the node is still
/// syncing, so the wait must yield to shutdown rather than block on sync.
#[tokio::test]
async fn await_sync_or_shutdown__should_return_false_when_shutdown_during_sync() {
// Given
let shutdown = CancellationToken::new();
let shutdown_clone = shutdown.clone();
tokio::spawn(async move { shutdown_clone.cancel() });

// When
let synced = await_sync_or_shutdown(pending::<()>(), &shutdown).await;

// Then
assert!(!synced);
}
}
Loading