diff --git a/Cargo.lock b/Cargo.lock index 050292e82..cde7ee674 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -881,6 +881,7 @@ checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" name = "control-panel" version = "0.4.0" dependencies = [ + "async-trait", "candid", "candid_parser", "canfund", diff --git a/core/control-panel/impl/Cargo.toml b/core/control-panel/impl/Cargo.toml index 835acfc3b..5b280ea9e 100644 --- a/core/control-panel/impl/Cargo.toml +++ b/core/control-panel/impl/Cargo.toml @@ -15,6 +15,7 @@ bench = false [dependencies] candid = { workspace = true } +async-trait = { workspace = true } canfund = { workspace = true } orbit-essentials = { path = '../../../libs/orbit-essentials', version = '0.2.0' } hex = { workspace = true } diff --git a/core/control-panel/impl/src/services/canister.rs b/core/control-panel/impl/src/services/canister.rs index dc720bbb1..a25522508 100644 --- a/core/control-panel/impl/src/services/canister.rs +++ b/core/control-panel/impl/src/services/canister.rs @@ -3,16 +3,19 @@ use crate::core::{canister_config, write_canister_config, CallContext}; use crate::errors::CanisterError; use crate::repositories::{UserRepository, USER_REPOSITORY}; use crate::SYSTEM_VERSION; +use canfund::errors::Error as CanfundError; use canfund::manager::options::{CyclesThreshold, FundManagerOptions, FundStrategy}; use canfund::manager::RegisterOpts; use canfund::operations::fetch::{FetchCyclesBalance, FetchCyclesBalanceFromPrometheusMetrics}; use canfund::FundManager; use control_panel_api::UploadCanisterModulesInput; +use ic_cdk::api::call::RejectionCode; +use ic_cdk::api::management_canister::main::CanisterId; use lazy_static::lazy_static; use orbit_essentials::api::ServiceResult; use orbit_essentials::repository::Repository; use std::cell::RefCell; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; lazy_static! { @@ -25,6 +28,152 @@ thread_local! { pub static FUND_MANAGER: RefCell = RefCell::new(FundManager::new()); } +/// Upper bound on a station's self-reported cycles balance. +/// +/// The balance is read from the station's own `/metrics`, so it is chosen by whoever controls that +/// station's code. canfund derives a consumption rate from consecutive readings as +/// `(previous - current) * 1_000_000_000 / elapsed`, and that multiplication is not saturating. +/// Keeping every accepted reading at or below this bound keeps the product inside `u128` for any +/// pair of readings, so a crafted balance cannot trap the shared monitoring round. +const MAX_REPORTED_STATION_CYCLES: u128 = u128::MAX / 1_000_000_000; + +/// Rejects self-reported cycles balances large enough to break the arithmetic canfund performs on +/// them. +/// +/// A rejected reading is surfaced as a fetch failure, which canfund already records per canister +/// without aborting the round, so one misbehaving station cannot stop the rest being funded. +struct BoundedCyclesFetcher { + inner: T, +} + +impl BoundedCyclesFetcher { + fn new(inner: T) -> Self { + Self { inner } + } +} + +#[async_trait::async_trait] +impl FetchCyclesBalance for BoundedCyclesFetcher { + async fn fetch_cycles_balance(&self, canister_id: CanisterId) -> Result { + let cycles = self.inner.fetch_cycles_balance(canister_id).await?; + + if cycles > MAX_REPORTED_STATION_CYCLES { + return Err(CanfundError::MetricsHttpRequestFailed { + code: RejectionCode::CanisterError, + reason: format!( + "canister {canister_id} reported an implausible cycles balance of {cycles}" + ), + }); + } + + Ok(cycles) + } +} + +/// How stale a cached reading may be before it stops being usable. +/// +/// The monitoring round runs daily, so a reading is normally about a day old by the time it is +/// used. This allows one missed refresh before a station stops being funded on old data. +const MAX_CACHED_BALANCE_AGE_NS: u64 = 3 * 24 * 60 * 60 * 1_000_000_000; + +#[derive(Clone)] +struct CachedBalance { + cycles: u128, + fetched_at: u64, +} + +impl CachedBalance { + fn is_usable_at(&self, now: u64) -> bool { + now.saturating_sub(self.fetched_at) <= MAX_CACHED_BALANCE_AGE_NS + } +} + +thread_local! { + /// Last successfully read balance per monitored canister. + static CYCLES_BALANCE_CACHE: RefCell> = + RefCell::new(HashMap::new()); + + /// Canisters with a refresh already in flight. A canister that never replies stays here, which + /// is what stops the round from opening a new call to it every day. + static REFRESHES_IN_FLIGHT: RefCell> = RefCell::new(HashSet::new()); +} + +/// Serves balances from a local cache and refreshes them out of band. +/// +/// The monitoring round awaits every registered canister's fetch inside one `join_all` while +/// holding the process lock, so a canister that accepts the call and never replies stalls the +/// round forever and the lock is never released. Nothing on the platform gets funded again. +/// +/// Reading a balance is decoupled from the round instead: the fetch returns immediately from +/// cache and the actual call happens in a spawned task. A canister that never replies only ever +/// starves its own cache entry, and every other canister is funded as normal. +struct CachedCyclesFetcher { + inner: Arc, +} + +impl CachedCyclesFetcher { + fn new(inner: Arc) -> Self { + Self { inner } + } + + fn spawn_refresh(&self, canister_id: CanisterId) { + let already_running = + REFRESHES_IN_FLIGHT.with(|running| !running.borrow_mut().insert(canister_id)); + + if already_running { + return; + } + + let inner = Arc::clone(&self.inner); + + crate::core::ic_cdk::spawn(async move { + let fetched = inner.fetch_cycles_balance(canister_id).await; + + REFRESHES_IN_FLIGHT.with(|running| { + running.borrow_mut().remove(&canister_id); + }); + + match fetched { + Ok(cycles) => CYCLES_BALANCE_CACHE.with(|cache| { + cache.borrow_mut().insert( + canister_id, + CachedBalance { + cycles, + fetched_at: time(), + }, + ); + }), + Err(err) => print(format!( + "Failed to refresh the cycles balance of {canister_id}: {err}" + )), + } + }); + } +} + +#[async_trait::async_trait] +impl FetchCyclesBalance for CachedCyclesFetcher { + async fn fetch_cycles_balance(&self, canister_id: CanisterId) -> Result { + self.spawn_refresh(canister_id); + + let cached = CYCLES_BALANCE_CACHE + .with(|cache| cache.borrow().get(&canister_id).cloned()) + .ok_or_else(|| CanfundError::MetricsHttpRequestFailed { + code: RejectionCode::CanisterError, + reason: format!("no cycles balance recorded yet for canister {canister_id}"), + })?; + + if !cached.is_usable_at(time()) { + return Err(CanfundError::MetricsHttpRequestFailed { + code: RejectionCode::CanisterError, + reason: format!("cycles balance for canister {canister_id} is stale"), + }); + } + + Ok(cached.cycles) + } +} + #[derive(Default, Debug)] pub struct CanisterService { user_repository: Arc, @@ -82,10 +231,14 @@ impl CanisterService { } pub fn create_station_cycles_fetcher(&self) -> Arc { - Arc::new(FetchCyclesBalanceFromPrometheusMetrics::new( - "/metrics".to_string(), - "station_canister_cycles_balance".to_string(), - )) + // Bounded rejects an implausible reading before it is cached; cached keeps a station that + // never replies from stalling the shared monitoring round. + Arc::new(CachedCyclesFetcher::new(Arc::new( + BoundedCyclesFetcher::new(FetchCyclesBalanceFromPrometheusMetrics::new( + "/metrics".to_string(), + "station_canister_cycles_balance".to_string(), + )), + ))) } // Monitor the cycles of active canisters that have been deployed by the control panel @@ -143,3 +296,124 @@ impl CanisterService { }; } } + +#[cfg(test)] +mod bounded_cycles_fetcher_tests { + use super::*; + + struct StubFetcher(u128); + + #[async_trait::async_trait] + impl FetchCyclesBalance for StubFetcher { + async fn fetch_cycles_balance( + &self, + _canister_id: CanisterId, + ) -> Result { + Ok(self.0) + } + } + + #[tokio::test] + async fn accepts_a_plausible_balance() { + let fetcher = BoundedCyclesFetcher::new(StubFetcher(500_000_000_000)); + + assert_eq!( + fetcher + .fetch_cycles_balance(CanisterId::anonymous()) + .await + .unwrap(), + 500_000_000_000 + ); + } + + #[tokio::test] + async fn accepts_a_balance_at_the_bound() { + let fetcher = BoundedCyclesFetcher::new(StubFetcher(MAX_REPORTED_STATION_CYCLES)); + + assert!(fetcher + .fetch_cycles_balance(CanisterId::anonymous()) + .await + .is_ok()); + } + + #[tokio::test] + async fn rejects_a_balance_that_would_overflow_the_consumption_rate() { + let fetcher = BoundedCyclesFetcher::new(StubFetcher(u128::MAX)); + + assert!(fetcher + .fetch_cycles_balance(CanisterId::anonymous()) + .await + .is_err()); + } + + /// The bound has to be tight enough that the largest difference between two accepted readings + /// still survives the `* 1_000_000_000` canfund applies to it. + #[test] + fn bound_keeps_the_consumption_rate_arithmetic_in_range() { + assert!(MAX_REPORTED_STATION_CYCLES + .checked_mul(1_000_000_000) + .is_some()); + } + + /// A canister with no reading yet must not be funded on invented data. + #[tokio::test] + async fn reports_a_failure_when_nothing_has_been_read_yet() { + let canister_id = CanisterId::from_slice(&[9; 29]); + CYCLES_BALANCE_CACHE.with(|cache| cache.borrow_mut().remove(&canister_id)); + + let fetcher = CachedCyclesFetcher::new(Arc::new(StubFetcher(1_000))); + + assert!(fetcher.fetch_cycles_balance(canister_id).await.is_err()); + } + + #[tokio::test] + async fn serves_a_fresh_cached_reading() { + let canister_id = CanisterId::from_slice(&[10; 29]); + CYCLES_BALANCE_CACHE.with(|cache| { + cache.borrow_mut().insert( + canister_id, + CachedBalance { + cycles: 700_000_000_000, + fetched_at: time(), + }, + ); + }); + + let fetcher = CachedCyclesFetcher::new(Arc::new(StubFetcher(1_000))); + + assert_eq!( + fetcher.fetch_cycles_balance(canister_id).await.unwrap(), + 700_000_000_000 + ); + } + + /// A canister that stops replying goes stale rather than being funded forever on an old value. + #[test] + fn a_cached_reading_expires_once_it_passes_the_maximum_age() { + let reading = CachedBalance { + cycles: 700_000_000_000, + fetched_at: 1_000, + }; + + assert!(reading.is_usable_at(1_000)); + assert!(reading.is_usable_at(1_000 + MAX_CACHED_BALANCE_AGE_NS)); + assert!(!reading.is_usable_at(1_000 + MAX_CACHED_BALANCE_AGE_NS + 1)); + } + + /// Without this, a canister that never replies would accumulate one open call per round. + #[test] + fn does_not_start_a_second_refresh_while_one_is_in_flight() { + let canister_id = CanisterId::from_slice(&[12; 29]); + REFRESHES_IN_FLIGHT.with(|running| { + running.borrow_mut().insert(canister_id); + }); + + let fetcher = CachedCyclesFetcher::new(Arc::new(StubFetcher(1_000))); + fetcher.spawn_refresh(canister_id); + + assert_eq!( + REFRESHES_IN_FLIGHT.with(|running| running.borrow().len()), + 1 + ); + } +}