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
13 changes: 13 additions & 0 deletions node/cli-opt/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,19 @@ pub struct LazyLoadingConfig {
pub runtime_override: Option<PathBuf>,
pub delay_between_requests: u32,
pub max_retries_per_request: u32,
/// Seconds to pause after printing the lazy-loading startup disclaimer before
/// the service actually starts. Gives an interactive operator time to read the
/// warning; set to `0` in automated environments (e.g. CI) where the extra
/// delay eats into the test runner's node-startup timeout budget.
pub startup_delay_seconds: u64,
/// Optional directory used to persist fetched fork state between runs.
///
/// State at a given block hash is immutable, so caching it on disk lets
/// restarts (e.g. CI retries) reuse previously fetched state instead of
/// re-querying the (potentially rate-limited) fork endpoint. When set and no
/// explicit `from_block` is given, the resolved fork block is pinned in this
/// directory so retries fork from the exact same block.
pub cache_path: Option<PathBuf>,
}
/// Extra args that are passed when creating a new node spec.
#[derive(Clone)]
Expand Down
21 changes: 21 additions & 0 deletions node/cli/src/cli.rs
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,27 @@ pub struct RunCmd {
#[clap(long, default_value = "10")]
pub lazy_loading_max_retries_per_request: u32,

/// Seconds to wait after printing the lazy-loading startup disclaimer before
/// starting the service.
///
/// Defaults to 10 seconds so an interactive operator can read the warning. Set
/// to `0` in automated environments (e.g. CI) where this delay would eat into
/// the test runner's node-startup timeout budget.
#[cfg(feature = "lazy-loading")]
#[clap(long, default_value = "10")]
pub lazy_loading_startup_delay: u64,

/// Optional directory used to persist fetched fork state between runs.
///
/// State at a given block hash is immutable, so caching it on disk lets
/// restarts (for example CI retries that share a working directory) reuse
/// previously fetched state instead of re-querying the fork endpoint. When
/// set and no explicit `--lazy-loading-block` is given, the resolved fork
/// block is pinned inside this directory so retries fork from the same block.
#[cfg(feature = "lazy-loading")]
#[clap(long, value_name = "PATH", value_parser, alias = "fork-cache-path")]
pub lazy_loading_cache_path: Option<PathBuf>,

/// When blocks should be sealed in the dev service.
///
/// Options are "instant", "manual", or timer interval in milliseconds
Expand Down
2 changes: 2 additions & 0 deletions node/cli/src/command.rs
Original file line number Diff line number Diff line change
Expand Up @@ -758,6 +758,8 @@ pub fn run() -> Result<()> {
runtime_override: cli.run.lazy_loading_runtime_override,
delay_between_requests: cli.run.lazy_loading_delay_between_requests,
max_retries_per_request: cli.run.lazy_loading_max_retries_per_request,
startup_delay_seconds: cli.run.lazy_loading_startup_delay,
cache_path: cli.run.lazy_loading_cache_path,
};

let spec_builder = lazy_loading::spec_builder();
Expand Down
5 changes: 3 additions & 2 deletions node/service/src/lazy_loading/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ mod helpers;
mod lock;
mod manual_sealing;
mod rpc_client;
mod state_cache;
mod state_overrides;
pub mod substrate_backend;

Expand Down Expand Up @@ -392,7 +393,7 @@ where
&lazy_loading_config,
)?;

let start_delay = 10;
let start_delay = lazy_loading_config.startup_delay_seconds;
let lazy_loading_startup_disclaimer = format!(
r#"

Expand All @@ -414,7 +415,7 @@ where
The service will start in {start_delay} seconds...

"#,
rpc = lazy_loading_config.state_rpc,
rpc = state_cache::redact_url(lazy_loading_config.state_rpc.as_str()),
fork_block = backend.fork_checkpoint.number
);

Expand Down
114 changes: 103 additions & 11 deletions node/service/src/lazy_loading/rpc_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,13 @@
// You should have received a copy of the GNU General Public License
// along with Moonbeam. If not, see <http://www.gnu.org/licenses/>.

use super::state_cache::StateCache;
use cumulus_primitives_core::BlockT;
use fc_rpc_v2_api::types::H256;
use jsonrpsee::http_client::HttpClient;
use moonbeam_core_primitives::BlockNumber;
use serde::de::DeserializeOwned;
use serde::Serialize;
use sp_api::__private::HeaderT;
use sp_rpc::list::ListOrValue;
use sp_rpc::number::NumberOrHex;
Expand All @@ -28,30 +30,76 @@ use std::future::Future;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio_retry::strategy::FixedInterval;
use tokio_retry::strategy::{jitter, ExponentialBackoff};
use tokio_retry::Retry;

/// Upper bound for a single retry back-off delay. Keeps retries from stalling
/// indefinitely while still giving a rate-limited endpoint time to recover.
const MAX_RETRY_BACKOFF: Duration = Duration::from_secs(30);
/// Floor for the exponential back-off base so that even a tiny configured
/// `delay_between_requests` still produces a meaningful back-off between retries.
const MIN_RETRY_BACKOFF_BASE_MS: u64 = 100;

#[derive(Debug, Clone)]
pub struct RPC {
http_client: HttpClient,
delay_between_requests_ms: u32,
max_retries_per_request: u32,
counter: Arc<AtomicU64>,
cache: Option<Arc<StateCache>>,
}

impl RPC {
pub fn new(
http_client: HttpClient,
delay_between_requests_ms: u32,
max_retries_per_request: u32,
cache: Option<Arc<StateCache>>,
) -> Self {
Self {
http_client,
delay_between_requests_ms,
max_retries_per_request,
counter: Default::default(),
cache,
}
}

/// Serve `request` from the on-disk cache when possible, otherwise perform it
/// over the network and persist a successful response.
///
/// Only call this for requests whose result is immutable for the given
/// `cache_key` (i.e. reads pinned to a concrete block hash/number). When no
/// cache is configured, or `cache_key` is `None`, this simply forwards to the
/// network with the usual retry behaviour.
fn cached_request<F, T>(
&self,
namespace: &str,
cache_key: Option<Vec<u8>>,
request: &dyn Fn() -> F,
) -> Result<T, jsonrpsee::core::ClientError>
where
F: Future<Output = Result<T, jsonrpsee::core::ClientError>>,
T: Serialize + DeserializeOwned,
{
if let (Some(cache), Some(key)) = (self.cache.as_ref(), cache_key.as_ref()) {
if let Some(bytes) = cache.get(namespace, key) {
if let Ok(value) = serde_json::from_slice::<T>(&bytes) {
return Ok(value);
}
}
}

let result = self.block_on(request)?;

if let (Some(cache), Some(key)) = (self.cache.as_ref(), cache_key.as_ref()) {
if let Ok(bytes) = serde_json::to_vec(&result) {
cache.put(namespace, key, &bytes);
}
}

Ok(result)
}
pub fn system_chain(&self) -> Result<String, jsonrpsee::core::ClientError> {
let request = &|| {
substrate_rpc_client::SystemApi::<H256, BlockNumber>::system_chain(&self.http_client)
Expand Down Expand Up @@ -145,7 +193,10 @@ impl RPC {
)
};

self.block_on(request)
let cache_key = at
.as_ref()
.and_then(|at| make_cache_key(at, &[key.0.as_slice()]));
self.cached_request("state_storage_hash", cache_key, request)
}

pub fn storage<
Expand All @@ -163,7 +214,10 @@ impl RPC {
)
};

self.block_on(request)
let cache_key = at
.as_ref()
.and_then(|at| make_cache_key(at, &[key.0.as_slice()]));
self.cached_request("state_storage", cache_key, request)
}

pub fn storage_keys_paged<
Expand All @@ -184,12 +238,21 @@ impl RPC {
at.clone(),
)
};
let result = self.block_on(request);

match result {
Ok(result) => Ok(result.iter().map(|item| item.0.clone()).collect()),
Err(err) => Err(err),
}
let cache_key = at.as_ref().and_then(|at| {
make_cache_key(
at,
&[
key.as_ref().map(|k| k.0.as_slice()).unwrap_or(&[]),
&count.to_le_bytes(),
start_key.as_ref().map(|k| k.0.as_slice()).unwrap_or(&[]),
],
)
});
let result: Vec<StorageKey> =
self.cached_request("state_storage_keys_paged", cache_key, request)?;

Ok(result.iter().map(|item| item.0.clone()).collect())
}

pub fn transaction_by_hash(
Expand Down Expand Up @@ -262,9 +325,21 @@ impl RPC {
// Explicit request delay, to avoid getting 429 errors
let _ = tokio::time::sleep(delay_between_requests).await;

// Retry request in case of failure
// The maximum number of retries is specified by `self.max_retries_per_request`
let retry_strategy = FixedInterval::new(delay_between_requests)
// Retry request in case of failure. The public fork endpoint
// (e.g. trace.api.moonbeam.network) can intermittently rate-limit
// (HTTP 429), time out, or drop connections under CI load. A fixed
// tiny interval would burn every retry within a few milliseconds
// without giving the endpoint time to recover, and retrying in
// lockstep across concurrent requests causes a thundering herd.
// Use exponential back-off with jitter instead, capped so a single
// request never stalls for too long. The maximum number of retries
// is specified by `self.max_retries_per_request`.
let backoff_base_ms =
(self.delay_between_requests_ms as u64).max(MIN_RETRY_BACKOFF_BASE_MS);
let retry_strategy = ExponentialBackoff::from_millis(2)
.factor(backoff_base_ms)
.max_delay(MAX_RETRY_BACKOFF)
.map(jitter)
.take(self.max_retries_per_request as usize);
let result = Retry::spawn(retry_strategy, f).await;

Expand All @@ -282,3 +357,20 @@ impl RPC {
})
}
}

/// Build a collision-resistant cache key from the block selector `at` and a set
/// of additional parts (e.g. the storage key and paging parameters). Each part
/// is length-prefixed so different argument combinations can never alias. Returns
/// `None` if the selector cannot be serialized, in which case the caller should
/// skip caching and go straight to the network.
fn make_cache_key<Hash: Serialize>(at: &Hash, parts: &[&[u8]]) -> Option<Vec<u8>> {
let at_bytes = serde_json::to_vec(at).ok()?;
let mut key = Vec::new();
key.extend_from_slice(&(at_bytes.len() as u32).to_le_bytes());
key.extend_from_slice(&at_bytes);
for part in parts {
key.extend_from_slice(&(part.len() as u32).to_le_bytes());
key.extend_from_slice(part);
}
Some(key)
}
Loading
Loading