diff --git a/.agents/context/crates/magicblock-chainlink.md b/.agents/context/crates/magicblock-chainlink.md index 6fa6e0918..fc178679c 100644 --- a/.agents/context/crates/magicblock-chainlink.md +++ b/.agents/context/crates/magicblock-chainlink.md @@ -145,6 +145,8 @@ Post-clone materialization metrics are emitted through `chainlink_clone_material Empty placeholders are created in `RemoteAccountProvider::try_get_multi` when RPC returns `None` and the pubkey is included in `mark_empty_if_not_found`; the provider converts the missing account into a zero-lamport, default-owner, empty-data account and emits `converted_to_empty`. Placeholder clone stages (`clone_submitted`, `clone_submit_failed`, `observed_in_bank_after_ensure`, and `still_missing_after_ensure`) are emitted only when the account clone request has that exact empty-placeholder shape. The `later_refetched` stage is deliberately not emitted yet because detecting repeated same-pubkey placeholders with retained pubkey state would add unbounded memory/cardinality risk; use group 7 sketches or sampled logs for repeated-same-pubkey detection instead. +Unique pubkey estimates are emitted through `chainlink_unique_pubkeys_estimate{origin,stage,window}` using a bounded HyperLogLog-like estimator with rotating minute buckets. Memory is bounded by active `(origin, stage)` series and 60 fixed-size sketches per series, not by the number of observed pubkeys. Stage labels reuse existing `magicblock-metrics` label enums when an exact label already exists; only labels missing from current shared enums live in Chainlink's local `UniquePubkeyStage`. The wired stages are `requested_by_rpc`, `bank_miss`, `remote_fetch`, `remote_not_found`, `subscription_already_present`, `subscription_added`, `subscription_evicted`, `clone_failed`, `still_missing_after_ensure`, and `companion_fetch`. Observations hash `Pubkey` values inside the estimator only; pubkeys and signatures must never become Prometheus labels. Export is throttled by `EXPORT_INTERVAL_SECONDS`, so estimates may be stale when no observations occur. + ### Remote fetch `RemoteAccountProvider::try_get_multi` subscribes before fetching so subscription updates that arrive during the fetch can win over stale RPC data. It: diff --git a/.agents/context/crates/magicblock-metrics.md b/.agents/context/crates/magicblock-metrics.md index ed34b98fb..494a836d9 100644 --- a/.agents/context/crates/magicblock-metrics.md +++ b/.agents/context/crates/magicblock-metrics.md @@ -209,6 +209,7 @@ System/storage gauge updates are driven from `magicblock-api/src/tickers.rs` at | `inc_chainlink_subscription_registration_accounts(origin, subscription_reason, outcome)` | `chainlink_subscription_registration_accounts_total` (`mbv_chainlink_subscription_registration_accounts_total`) classifies Chainlink account subscription registration attempts by `{origin,subscription_reason,outcome}`. Origin is `AccountFetchOrigin` label values plus `internal`; subscription reasons are `direct_account`, `delegation_record`, `program_data`, `undelegation_tracking`, `ata_projection`; outcomes are `already_present`, `added_below_capacity`, `evicted_candidate`, `subscribe_error`, `unsubscribe_evicted_error`, `rejected_and_unsubscribed`, `unsubscribe_rejected_error`. | | `inc_chainlink_subscription_release_accounts(reason, outcome)` | `chainlink_subscription_release_accounts_total` (`mbv_chainlink_subscription_release_accounts_total`) classifies Chainlink account subscription release attempts by `{reason,outcome}` with the same five reason values and outcomes `unsubscribed`, `already_absent`, `unsubscribe_failed`, `retained_intentionally`, `retained_other_reasons`. | | `inc_chainlink_subscription_cleanup_accounts(cleanup_source, outcome)` | `chainlink_subscription_cleanup_accounts_total` (`mbv_chainlink_subscription_cleanup_accounts_total`) classifies Chainlink account subscription cleanup actions by `{cleanup_source,outcome}`; cleanup sources are `normal_release`, `manual_unsubscribe`, `capacity_eviction`, `rejected_new_subscription`, `delegated_account_silent`, `reconciler`; outcomes are `unsubscribed`, `already_absent`, `unsubscribe_failed`, `removal_update_failed`, `retained_intentionally`. | +| `set_chainlink_unique_pubkeys_estimate(origin, stage, window, estimate)` | `chainlink_unique_pubkeys_estimate{origin,stage,window}` (`mbv_chainlink_unique_pubkeys_estimate`) sets approximate unique Chainlink pubkeys by bounded origin/stage labels and rolling windows `1m`, `5m`, and `1h`. The `stage` argument intentionally accepts `LabelValue` so current-master label enums are reused instead of duplicating stage enums. Example PromQL: `mbv_chainlink_unique_pubkeys_estimate{stage="remote_fetch", window="5m"}`. | | `inc_account_fetches_success(count)` | Successful network account fetch count. | | `inc_account_fetches_failed(count)` | Failed network account fetch count. | | `inc_account_fetches_found(origin, count)` | Network fetches that found accounts, labelled by `AccountFetchOrigin`. | @@ -233,6 +234,7 @@ Important caveats: - The clone lifecycle counters replace the stale clone-cache and pending-clone gauges removed by the eviction-vs-get metrics cleanup; use the counters for clone observability rather than reintroducing those gauges. - Empty-placeholder stages are bounded enum labels: `converted_to_empty` when Chainlink converts a remote `None` into the zero-lamport/default-owner/empty-data placeholder, `clone_submitted` after a placeholder clone is submitted, `clone_submit_failed` if that clone submission fails, `observed_in_bank_after_ensure` when the post-clone materialization check sees the placeholder in bank, and `still_missing_after_ensure` when the cloner returned success but the placeholder is still not visible. `later_refetched` is reserved for a future sampled/sketch implementation and is not emitted by the current code because retaining per-pubkey state would create unbounded memory/cardinality risk. - Subscription lifecycle counters are for Chainlink registration, release, and cleanup outcome classification. Call sites must use the provided enum/static labels only; do not add pubkey, signature, raw error, endpoint URL, or other free-form labels. +- Unique Chainlink pubkey estimates use bounded `{origin,stage,window}` labels. The `stage` wrapper parameter accepts any `LabelValue` so existing bounded label enums can be reused; do not add duplicate metric-stage enums in `magicblock-metrics`. Pubkeys and signatures must never be labels; pubkeys are hashed only inside the Chainlink estimator. - `AccountFetchOrigin::SendTransaction(Signature)` intentionally labels as only `send_transaction`; the signature is available through `signature()` for logging/correlation but must not become a Prometheus label. - `chainlink_pending_fetch_waiters_gauge` is incremented only for waiter joins, not for owner calls awaiting their own operation. It must be decremented on success, failure, cancellation, timeout, and waiter-drop paths. - Pending-fetch labels are low-cardinality enum/static labels only: `origin` uses `AccountFetchOrigin`, `layer` uses `fetch_cloner` or `remote_account_provider`, and `outcome` uses exactly `owned`, `joined_existing`, `owner_succeeded`, `owner_failed`, `owner_cancelled`, `resolved_by_subscription_update`, or `rpc_fetch_completed_after_update`. @@ -358,6 +360,7 @@ It is implemented for: - `Result` where both sides implement `LabelValue`, - `AccountFetchOrigin`, - Chainlink clone lifecycle label enums (`ChainlinkCloneRemoteResult`, `ChainlinkCloneIntent`, `ChainlinkCloneOutcome`, `ChainlinkCloneMaterializationOutcome`, `ChainlinkEmptyPlaceholderStage`), +- Chainlink unique pubkey window labels (`ChainlinkUniquePubkeyWindow`), - subscription lifecycle label enums (`SubscriptionRegistrationOrigin`, `SubscriptionReasonLabel`, `SubscriptionRegistrationOutcome`, `SubscriptionReleaseOutcome`, `SubscriptionCleanupSource`, `SubscriptionCleanupOutcome`), - downstream consumer types such as committor execution outputs and errors. diff --git a/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs b/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs index 2546c98b2..478c97586 100644 --- a/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs +++ b/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs @@ -76,6 +76,7 @@ use crate::{ blacklisted_accounts::{ blacklisted_accounts, programs_not_to_subscribe, }, + unique_pubkey_estimator::{UniquePubkeyEstimator, UniquePubkeyStage}, }, cloner::{ errors::{ClonerError, ClonerResult}, @@ -309,6 +310,12 @@ where &self.remote_account_provider } + pub(crate) fn unique_pubkey_estimator( + &self, + ) -> &Arc { + self.remote_account_provider.unique_pubkey_estimator() + } + #[cfg(test)] fn has_pending_request(&self, pubkey: &Pubkey) -> bool { self.pending_requests.contains(pubkey) @@ -810,6 +817,11 @@ where clone_intent, ChainlinkCloneOutcome::CloneFailed, ); + self.unique_pubkey_estimator().observe( + fetch_origin, + ChainlinkCloneOutcome::CloneFailed, + &pubkey, + ); metrics::inc_chainlink_clone_accounts_total( fetch_origin, remote_result, @@ -927,6 +939,11 @@ where clone_intent, ChainlinkCloneOutcome::CloneFailed, ); + self.unique_pubkey_estimator().observe( + fetch_origin, + ChainlinkCloneOutcome::CloneFailed, + &program_id, + ); metrics::inc_chainlink_clone_accounts_total( fetch_origin, remote_result, @@ -1155,6 +1172,11 @@ where clone_intent, ChainlinkCloneOutcome::CloneFailed, ); + self.unique_pubkey_estimator().observe( + fetch_origin, + ChainlinkCloneOutcome::CloneFailed, + &pubkey, + ); metrics::inc_chainlink_clone_accounts_total( fetch_origin, remote_result, @@ -2403,6 +2425,14 @@ where subscription_slot.max(self.remote_account_provider.chain_slot()) }); + self.remote_account_provider + .unique_pubkey_estimator() + .observe_many( + fetch_origin, + UniquePubkeyStage::RemoteFetch, + pubkeys.iter(), + ); + let accs = self .remote_account_provider .try_get_multi( @@ -2476,6 +2506,13 @@ where clone_as_empty, not_found, } = pipeline::partition_not_found(mark_empty_if_not_found, not_found); + self.remote_account_provider + .unique_pubkey_estimator() + .observe_many( + fetch_origin, + UniquePubkeyStage::RemoteNotFound, + not_found.iter().map(|(pubkey, _)| pubkey), + ); // For accounts we couldn't find we cannot do anything. We will let code depending // on them to be in the bank fail on its own @@ -2908,6 +2945,13 @@ where let count = pubkeys.len(); trace!(count, "Fetching and cloning accounts with dedup"); } + self.remote_account_provider + .unique_pubkey_estimator() + .observe_many( + fetch_origin, + UniquePubkeyStage::RequestedByRpc, + pubkeys.iter().copied(), + ); let mut in_bank = HashSet::new(); let mut extra_mark_empty = vec![]; @@ -2924,6 +2968,9 @@ where let mut undelegating_checks: Vec<(Pubkey, AccountSharedData)> = vec![]; for pubkey in pubkeys.iter() { if force_refresh_pubkeys.contains(*pubkey) { + self.remote_account_provider + .unique_pubkey_estimator() + .observe(fetch_origin, UniquePubkeyStage::BankMiss, pubkey); forced_refresh_remote_required_count += 1; continue; } @@ -2954,6 +3001,9 @@ where in_bank.insert(**pubkey); } } else { + self.remote_account_provider + .unique_pubkey_estimator() + .observe(fetch_origin, UniquePubkeyStage::BankMiss, pubkey); bank_miss_remote_required_count += 1; } } @@ -2997,6 +3047,13 @@ where pubkey = %pubkey, "Account completed undelegation which was missed and is fetched again" ); + self.remote_account_provider + .unique_pubkey_estimator() + .observe( + fetch_origin, + UniquePubkeyStage::BankMiss, + &pubkey, + ); bank_hit_undelegating_refresh_required_count += 1; metrics::inc_unstuck_undelegation_count(); if let RefreshDecision::YesAndMarkEmptyIfNotFound = @@ -3182,6 +3239,8 @@ where fetch_origin: AccountFetchOrigin, ) -> task::JoinHandle> { let provider = self.remote_account_provider.clone(); + let unique_pubkey_estimator = + provider.unique_pubkey_estimator().clone(); let bank = self.accounts_bank.clone(); let fetch_count = self.fetch_count.clone(); task::spawn(async move { @@ -3194,6 +3253,11 @@ where // Increment fetch counter for testing deduplication (2 accounts: pubkey + delegation_record_pubkey) fetch_count.fetch_add(2, Ordering::Relaxed); + unique_pubkey_estimator.observe_many( + fetch_origin, + UniquePubkeyStage::CompanionFetch, + [&pubkey, &companion_pubkey], + ); provider .try_get_multi_until_slots_match( diff --git a/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs b/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs index 0335f687e..1aae9ed1b 100644 --- a/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs +++ b/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs @@ -4,8 +4,10 @@ use dlp_api::state::DelegationRecord; use magicblock_metrics::metrics::{ chainlink_pending_fetch_accounts_value, chainlink_pending_fetch_waiters_gauge_value, - chainlink_pending_fetch_waiters_value, ChainlinkPendingFetchLayer, - ChainlinkPendingFetchOutcome, + chainlink_pending_fetch_waiters_value, + chainlink_unique_pubkeys_estimate_value, ChainlinkCloneOutcome, + ChainlinkPendingFetchLayer, ChainlinkPendingFetchOutcome, + ChainlinkUniquePubkeyWindow, }; use solana_account::{ Account, AccountSharedData, ReadableAccount, WritableAccount, @@ -31,6 +33,7 @@ use crate::{ accounts_bank::mock::AccountsBankStub, assert_not_cloned, assert_not_subscribed, assert_subscribed, assert_subscribed_without_delegation_record, + chainlink::unique_pubkey_estimator::UniquePubkeyStage, remote_account_provider::{ chain_pubsub_client::mock::ChainPubsubClientMock, chain_slot::ChainSlot, pubsub_common::SubscriptionSource, @@ -2309,6 +2312,97 @@ async fn test_parallel_fetch_prevention_multiple_accounts() { ); } +#[tokio::test] +async fn test_unique_pubkey_metrics_observe_requested_and_remote_fetch() { + init_logger(); + const TEST_SLOT: u64 = 100; + let validator_keypair = Keypair::new(); + let pubkey1 = random_pubkey(); + let pubkey2 = random_pubkey(); + + let FetcherTestCtx { fetch_cloner, .. } = + setup([], TEST_SLOT, validator_keypair.insecure_clone()).await; + + fetch_cloner + .fetch_and_clone_accounts_with_dedup( + &[pubkey1, pubkey2], + None, + None, + AccountFetchOrigin::GetAccount, + ) + .await + .unwrap(); + + fetch_cloner + .unique_pubkey_estimator() + .force_export_for_tests(0); + + assert!( + chainlink_unique_pubkeys_estimate_value( + &AccountFetchOrigin::GetAccount, + &UniquePubkeyStage::RequestedByRpc, + ChainlinkUniquePubkeyWindow::OneMinute, + ) >= 2 + ); + assert!( + chainlink_unique_pubkeys_estimate_value( + &AccountFetchOrigin::GetAccount, + &UniquePubkeyStage::RemoteFetch, + ChainlinkUniquePubkeyWindow::OneMinute, + ) >= 2 + ); +} + +#[tokio::test] +async fn test_unique_pubkey_metrics_observe_clone_failed() { + init_logger(); + const TEST_SLOT: u64 = 100; + let validator_keypair = Keypair::new(); + let pubkey = random_pubkey(); + let account_owner = random_pubkey(); + let account = Account { + lamports: 1_000_000, + data: vec![1, 2, 3], + owner: account_owner, + executable: false, + rent_epoch: 0, + }; + + let FetcherTestCtx { + fetch_cloner, + cloner, + .. + } = setup( + [(pubkey, account)], + TEST_SLOT, + validator_keypair.insecure_clone(), + ) + .await; + + cloner.set_fail_next_clone(true); + fetch_cloner + .fetch_and_clone_accounts_with_dedup( + &[pubkey], + None, + None, + AccountFetchOrigin::GetAccount, + ) + .await + .expect_err("clone failure should be returned"); + + fetch_cloner + .unique_pubkey_estimator() + .force_export_for_tests(0); + + assert!( + chainlink_unique_pubkeys_estimate_value( + &AccountFetchOrigin::GetAccount, + &ChainlinkCloneOutcome::CloneFailed, + ChainlinkUniquePubkeyWindow::OneMinute, + ) >= 1 + ); +} + #[tokio::test] async fn test_cold_accounts_fetch_is_batched_into_single_rpc_call() { init_logger(); diff --git a/magicblock-chainlink/src/chainlink/mod.rs b/magicblock-chainlink/src/chainlink/mod.rs index 2f4a844c3..b2d0225a5 100644 --- a/magicblock-chainlink/src/chainlink/mod.rs +++ b/magicblock-chainlink/src/chainlink/mod.rs @@ -9,7 +9,9 @@ use fetch_cloner::FetchCloner; use magicblock_accounts_db::{traits::AccountsBank, AccountsDb}; use magicblock_aml::RiskService; use magicblock_config::config::ChainLinkConfig; -use magicblock_metrics::metrics::AccountFetchOrigin; +use magicblock_metrics::metrics::{ + AccountFetchOrigin, ChainlinkCloneMaterializationOutcome, +}; use solana_account::{AccountSharedData, ReadableAccount}; use solana_commitment_config::CommitmentConfig; use solana_keypair::Keypair; @@ -35,6 +37,7 @@ mod blacklisted_accounts; pub mod config; pub mod errors; pub mod fetch_cloner; +pub(crate) mod unique_pubkey_estimator; pub use blacklisted_accounts::*; @@ -550,6 +553,15 @@ impl fetch_origin, ) .await?; + let still_missing_after_ensure = pubkeys + .iter() + .filter(|pubkey| self.accounts_bank.get_account(pubkey).is_none()) + .collect::>(); + fetch_cloner.unique_pubkey_estimator().observe_many( + fetch_origin, + ChainlinkCloneMaterializationOutcome::StillMissingAfterEnsure, + still_missing_after_ensure, + ); trace!("Fetched and cloned accounts"); Ok(result) } @@ -605,15 +617,20 @@ mod tests { use std::{sync::Arc, time::Duration}; use magicblock_accounts_db::traits::AccountsBank; - use magicblock_metrics::metrics::AccountFetchOrigin; + use magicblock_metrics::metrics::{ + chainlink_unique_pubkeys_estimate_value, AccountFetchOrigin, + ChainlinkCloneMaterializationOutcome, ChainlinkUniquePubkeyWindow, + }; use solana_account::AccountSharedData; + use solana_keypair::Keypair; use solana_message::legacy::Message; use solana_pubkey::Pubkey; use solana_transaction::{sanitized::SanitizedTransaction, Transaction}; use tokio::sync::mpsc; use super::{ - errors::ChainlinkError, InnerChainlink, ReplicationModeAwareChainlink, + errors::ChainlinkError, FetchCloner, InnerChainlink, + ReplicationModeAwareChainlink, }; use crate::{ accounts_bank::mock::AccountsBankStub, @@ -685,6 +702,30 @@ mod tests { (accounts_bank, chainlink) } + async fn enabled_chainlink( + ) -> (Arc, TestReplicationModeAwareChainlink) { + let accounts_bank = Arc::new(AccountsBankStub::default()); + let cloner = Arc::new(ClonerStub::new(accounts_bank.clone())); + let remote_account_provider = test_remote_account_provider().await; + let (_subscription_tx, subscription_rx) = mpsc::channel(1_000); + let fetch_cloner = FetchCloner::new( + &remote_account_provider, + &accounts_bank, + &cloner, + Keypair::new(), + subscription_rx, + None, + None, + ); + let chainlink = + InnerChainlink::try_new(&accounts_bank, Some(fetch_cloner)) + .expect("enabled Chainlink should be constructed"); + ( + accounts_bank, + TestReplicationModeAwareChainlink::enabled(chainlink), + ) + } + #[tokio::test] async fn disabled_mode_ensure_accounts_is_noop() { let (accounts_bank, chainlink) = disabled_chainlink(); @@ -702,6 +743,32 @@ mod tests { assert!(!chainlink.is_watching(&pubkey)); } + #[tokio::test] + async fn test_unique_pubkey_metrics_observe_still_missing_after_ensure() { + init_logger(); + let (_accounts_bank, chainlink) = enabled_chainlink().await; + let pubkey = Pubkey::new_unique(); + + chainlink + .ensure_accounts(&[pubkey], None, AccountFetchOrigin::GetAccount) + .await + .expect("missing account ensure should complete"); + + chainlink + .fetch_cloner() + .expect("enabled Chainlink should have a fetch cloner") + .unique_pubkey_estimator() + .force_export_for_tests(0); + + assert!( + chainlink_unique_pubkeys_estimate_value( + &AccountFetchOrigin::GetAccount, + &ChainlinkCloneMaterializationOutcome::StillMissingAfterEnsure, + ChainlinkUniquePubkeyWindow::OneMinute, + ) >= 1 + ); + } + #[tokio::test] async fn disabled_mode_fetch_accounts_returns_none_values() { let (_accounts_bank, chainlink) = disabled_chainlink(); diff --git a/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs new file mode 100644 index 000000000..144dd04e5 --- /dev/null +++ b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs @@ -0,0 +1,435 @@ +use std::{ + collections::{hash_map::DefaultHasher, HashMap}, + hash::{Hash, Hasher}, + sync::{ + atomic::{AtomicU64, Ordering}, + Mutex, + }, + time::{SystemTime, UNIX_EPOCH}, +}; + +use magicblock_metrics::metrics::{ + self, ChainlinkUniquePubkeyWindow, LabelValue, +}; +use solana_pubkey::Pubkey; + +const HLL_PRECISION: u8 = 12; +const HLL_REGISTER_COUNT: usize = 1 << HLL_PRECISION; +const BUCKET_SECONDS: u64 = 60; +const BUCKET_COUNT: usize = 60; +const EXPORT_INTERVAL_SECONDS: u64 = 15; +const WINDOWS: [ChainlinkUniquePubkeyWindow; 3] = [ + ChainlinkUniquePubkeyWindow::OneMinute, + ChainlinkUniquePubkeyWindow::FiveMinutes, + ChainlinkUniquePubkeyWindow::OneHour, +]; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum UniquePubkeyStage { + RequestedByRpc, + BankMiss, + RemoteFetch, + RemoteNotFound, + SubscriptionAlreadyPresent, + SubscriptionAdded, + SubscriptionEvicted, + CompanionFetch, +} + +impl LabelValue for UniquePubkeyStage { + fn value(&self) -> &str { + match self { + Self::RequestedByRpc => "requested_by_rpc", + Self::BankMiss => "bank_miss", + Self::RemoteFetch => "remote_fetch", + Self::RemoteNotFound => "remote_not_found", + Self::SubscriptionAlreadyPresent => "subscription_already_present", + Self::SubscriptionAdded => "subscription_added", + Self::SubscriptionEvicted => "subscription_evicted", + Self::CompanionFetch => "companion_fetch", + } + } +} + +const UNIQUE_PUBKEY_STAGES: [UniquePubkeyStage; 8] = [ + UniquePubkeyStage::RequestedByRpc, + UniquePubkeyStage::BankMiss, + UniquePubkeyStage::RemoteFetch, + UniquePubkeyStage::RemoteNotFound, + UniquePubkeyStage::SubscriptionAlreadyPresent, + UniquePubkeyStage::SubscriptionAdded, + UniquePubkeyStage::SubscriptionEvicted, + UniquePubkeyStage::CompanionFetch, +]; + +#[derive(Debug)] +pub(crate) struct UniquePubkeyEstimator { + series: Mutex>>, + last_export_epoch_seconds: AtomicU64, +} + +impl Default for UniquePubkeyEstimator { + fn default() -> Self { + debug_assert_unique_stage_labels(); + Self { + series: Mutex::default(), + last_export_epoch_seconds: AtomicU64::default(), + } + } +} + +impl UniquePubkeyEstimator { + pub(crate) fn observe( + &self, + origin: impl LabelValue, + stage: impl LabelValue, + pubkey: &Pubkey, + ) { + self.observe_at(origin, stage, pubkey, now_epoch_seconds()); + } + + pub(crate) fn observe_many<'a>( + &self, + origin: impl LabelValue + Copy, + stage: impl LabelValue + Copy, + pubkeys: impl IntoIterator, + ) { + let now_epoch_seconds = now_epoch_seconds(); + for pubkey in pubkeys { + self.observe_at(origin, stage, pubkey, now_epoch_seconds); + } + } + + fn observe_at( + &self, + origin: impl LabelValue, + stage: impl LabelValue, + pubkey: &Pubkey, + now_epoch_seconds: u64, + ) { + let pubkey_hash = hash_pubkey(pubkey); + let epoch_minute = now_epoch_seconds / BUCKET_SECONDS; + let origin_label = origin.value(); + let stage_label = stage.value(); + let should_export = self.should_export(now_epoch_seconds); + + let mut series = + self.series.lock().unwrap_or_else(|err| err.into_inner()); + let stage_series = series.entry(origin_label.to_string()).or_default(); + let pubkey_series = stage_series + .entry(stage_label.to_string()) + .or_insert_with(|| { + UniquePubkeySeries::new(origin_label, stage_label) + }); + pubkey_series.observe_hash(epoch_minute, pubkey_hash); + + // Snapshot the estimates while the lock is held, then drop the guard + // before performing Prometheus writes so exports never block other + // observe/observe_many callers across origin/stage pairs. + let export_estimates = should_export + .then(|| collect_series_estimates(&series, epoch_minute)); + drop(series); + + if let Some(estimates) = export_estimates { + write_series_estimates(estimates); + } + } + + fn should_export(&self, now_epoch_seconds: u64) -> bool { + let last_export_epoch_seconds = + self.last_export_epoch_seconds.load(Ordering::Relaxed); + if now_epoch_seconds.saturating_sub(last_export_epoch_seconds) + < EXPORT_INTERVAL_SECONDS + { + return false; + } + self.last_export_epoch_seconds + .compare_exchange( + last_export_epoch_seconds, + now_epoch_seconds, + Ordering::AcqRel, + Ordering::Relaxed, + ) + .is_ok() + } + + #[cfg(any(test, feature = "dev-context"))] + #[cfg_attr(feature = "dev-context", allow(dead_code))] + pub fn force_export_for_tests(&self, now_epoch_seconds: u64) { + let epoch_minute = now_epoch_seconds / BUCKET_SECONDS; + let series = self.series.lock().unwrap_or_else(|err| err.into_inner()); + export_series(&series, epoch_minute); + } +} + +#[derive(Debug)] +struct UniquePubkeySeries { + origin_label: String, + stage_label: String, + buckets: Vec, +} + +impl UniquePubkeySeries { + fn new(origin_label: &str, stage_label: &str) -> Self { + let buckets = + (0..BUCKET_COUNT).map(|_| MinuteBucket::default()).collect(); + Self { + origin_label: origin_label.to_string(), + stage_label: stage_label.to_string(), + buckets, + } + } + + fn observe_hash(&mut self, epoch_minute: u64, pubkey_hash: u64) { + let bucket_index = (epoch_minute as usize) % BUCKET_COUNT; + let bucket = &mut self.buckets[bucket_index]; + if bucket.epoch_minute != epoch_minute { + bucket.epoch_minute = epoch_minute; + bucket.sketch.clear(); + } + bucket.sketch.insert_hash(pubkey_hash); + } + + fn estimate_window( + &self, + now_epoch_minute: u64, + window_minutes: u64, + ) -> f64 { + let mut merged = HllSketch::default(); + for bucket in &self.buckets { + if now_epoch_minute.saturating_sub(bucket.epoch_minute) + < window_minutes + { + merged.merge(&bucket.sketch); + } + } + merged.estimate() + } +} + +#[derive(Debug, Default)] +struct MinuteBucket { + epoch_minute: u64, + sketch: HllSketch, +} + +#[derive(Debug, Clone)] +struct HllSketch { + registers: [u8; HLL_REGISTER_COUNT], +} + +impl Default for HllSketch { + fn default() -> Self { + Self { + registers: [0; HLL_REGISTER_COUNT], + } + } +} + +impl HllSketch { + fn clear(&mut self) { + self.registers = [0; HLL_REGISTER_COUNT]; + } + + fn insert_hash(&mut self, hash: u64) { + let index_mask = HLL_REGISTER_COUNT as u64 - 1; + let register_index = (hash & index_mask) as usize; + let remainder = hash >> HLL_PRECISION; + let remaining_bits = u64::BITS - HLL_PRECISION as u32; + let rank = if remainder == 0 { + remaining_bits + } else { + (remainder.leading_zeros() - HLL_PRECISION as u32 + 1) + .min(remaining_bits) + } as u8; + self.registers[register_index] = + self.registers[register_index].max(rank); + } + + fn merge(&mut self, other: &Self) { + for (register, other_register) in + self.registers.iter_mut().zip(other.registers.iter()) + { + *register = (*register).max(*other_register); + } + } + + fn estimate(&self) -> f64 { + let m = HLL_REGISTER_COUNT as f64; + let alpha = 0.7213 / (1.0 + 1.079 / m); + let mut harmonic_sum = 0.0; + let mut zero_registers = 0_u64; + for register in self.registers { + harmonic_sum += 2_f64.powi(-(register as i32)); + if register == 0 { + zero_registers += 1; + } + } + let raw_estimate = alpha * m * m / harmonic_sum; + if raw_estimate <= 2.5 * m && zero_registers > 0 { + m * (m / zero_registers as f64).ln() + } else { + raw_estimate + } + } +} + +struct SeriesEstimate { + origin_label: String, + stage_label: String, + window: ChainlinkUniquePubkeyWindow, + estimate: u64, +} + +/// Computes the per-window estimates for every series. This is the only part of +/// the export that reads shared series state, so it is meant to run while the +/// `series` lock is held; the returned snapshot can then be written to +/// Prometheus without the guard. +fn collect_series_estimates( + series: &HashMap>, + now_epoch_minute: u64, +) -> Vec { + let mut estimates = Vec::new(); + for stage_series in series.values() { + for pubkey_series in stage_series.values() { + for window in WINDOWS { + let estimate = pubkey_series + .estimate_window(now_epoch_minute, window.minutes()) + .round() as u64; + estimates.push(SeriesEstimate { + origin_label: pubkey_series.origin_label.clone(), + stage_label: pubkey_series.stage_label.clone(), + window, + estimate, + }); + } + } + } + estimates +} + +/// Writes a previously collected snapshot to Prometheus. This does not touch the +/// shared series state, so it must be called after the `series` lock is dropped. +fn write_series_estimates(estimates: Vec) { + for estimate in estimates { + metrics::set_chainlink_unique_pubkeys_estimate( + &estimate.origin_label, + &estimate.stage_label, + estimate.window, + estimate.estimate, + ); + } +} + +fn export_series( + series: &HashMap>, + now_epoch_minute: u64, +) { + write_series_estimates(collect_series_estimates(series, now_epoch_minute)); +} + +fn debug_assert_unique_stage_labels() { + for (index, stage) in UNIQUE_PUBKEY_STAGES.iter().enumerate() { + for other_stage in UNIQUE_PUBKEY_STAGES.iter().skip(index + 1) { + debug_assert_ne!(stage.value(), other_stage.value()); + } + } +} + +fn hash_pubkey(pubkey: &Pubkey) -> u64 { + let mut hasher = DefaultHasher::new(); + pubkey.hash(&mut hasher); + hasher.finish() +} + +fn now_epoch_seconds() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +#[cfg(test)] +mod tests { + use magicblock_metrics::metrics::{ + chainlink_unique_pubkeys_estimate_value, AccountFetchOrigin, + }; + use solana_signature::Signature; + + use super::*; + + fn pubkey_from_u64(value: u64) -> Pubkey { + let mut bytes = [0; 32]; + bytes[..8].copy_from_slice(&value.to_le_bytes()); + Pubkey::new_from_array(bytes) + } + + #[test] + fn hll_estimate_for_known_set_is_within_reasonable_error() { + let mut sketch = HllSketch::default(); + for value in 0..10_000 { + sketch.insert_hash(hash_pubkey(&pubkey_from_u64(value))); + } + + let estimate = sketch.estimate(); + let error_ratio = ((estimate - 10_000.0) / 10_000.0).abs(); + assert!( + error_ratio <= 0.10, + "estimate {estimate} should be within 10% of 10,000" + ); + } + + #[test] + fn bucket_rotation_drops_old_window_entries() { + let estimator = UniquePubkeyEstimator::default(); + let origin = AccountFetchOrigin::GetAccount; + let stage = UniquePubkeyStage::RequestedByRpc; + estimator.observe_at(origin, stage, &pubkey_from_u64(1), 0); + estimator.observe_at(origin, stage, &pubkey_from_u64(2), 120); + estimator.force_export_for_tests(120); + + assert_eq!( + chainlink_unique_pubkeys_estimate_value( + &origin, + &stage, + ChainlinkUniquePubkeyWindow::OneMinute, + ), + 1 + ); + assert!( + chainlink_unique_pubkeys_estimate_value( + &origin, + &stage, + ChainlinkUniquePubkeyWindow::FiveMinutes, + ) >= 2 + ); + } + + #[test] + fn send_transaction_origin_does_not_include_signature() { + let estimator = UniquePubkeyEstimator::default(); + let stage = UniquePubkeyStage::RemoteFetch; + estimator.observe_at( + AccountFetchOrigin::SendTransaction(Signature::new_unique()), + stage, + &pubkey_from_u64(10), + 0, + ); + estimator.observe_at( + AccountFetchOrigin::SendTransaction(Signature::new_unique()), + stage, + &pubkey_from_u64(11), + 0, + ); + estimator.force_export_for_tests(0); + + assert_eq!( + chainlink_unique_pubkeys_estimate_value( + &AccountFetchOrigin::SendTransaction(Signature::new_unique()), + &stage, + ChainlinkUniquePubkeyWindow::OneMinute, + ), + 2 + ); + } +} diff --git a/magicblock-chainlink/src/remote_account_provider/mod.rs b/magicblock-chainlink/src/remote_account_provider/mod.rs index 6ea82b3ad..868d9f600 100644 --- a/magicblock-chainlink/src/remote_account_provider/mod.rs +++ b/magicblock-chainlink/src/remote_account_provider/mod.rs @@ -88,6 +88,9 @@ use magicblock_metrics::{ pub use remote_account::{ResolvedAccount, ResolvedAccountSharedData}; use crate::{ + chainlink::unique_pubkey_estimator::{ + UniquePubkeyEstimator, UniquePubkeyStage, + }, errors::ChainlinkResult, remote_account_provider::{ chain_updates_client::ChainUpdatesClient, @@ -554,6 +557,8 @@ pub struct RemoteAccountProvider { subscription_forwarder: Arc>, + unique_pubkey_estimator: Arc, + /// Task that periodically updates the active subscriptions gauge _active_subscriptions_task_handle: Option>, } @@ -752,6 +757,7 @@ impl RemoteAccountProvider { lrucache_subscribed_accounts, capacity_eviction_protection: Arc::new(RwLock::new(None)), subscription_forwarder: Arc::new(subscription_forwarder), + unique_pubkey_estimator: Arc::default(), removed_account_tx, removed_account_rx: Mutex::new(Some(removed_account_rx)), _active_subscriptions_task_handle: active_subscriptions_updater, @@ -775,6 +781,12 @@ impl RemoteAccountProvider { } } + pub(crate) fn unique_pubkey_estimator( + &self, + ) -> &Arc { + &self.unique_pubkey_estimator + } + pub async fn try_new_from_endpoints( endpoints: &Endpoints, commitment: CommitmentConfig, @@ -1712,6 +1724,11 @@ impl RemoteAccountProvider { reason.into(), SubscriptionRegistrationOutcome::AlreadyPresent, ); + self.unique_pubkey_estimator.observe( + origin, + UniquePubkeyStage::SubscriptionAlreadyPresent, + pubkey, + ); } AddAccountOutcome::Added => { inc_chainlink_subscription_registration_accounts( @@ -1719,6 +1736,11 @@ impl RemoteAccountProvider { reason.into(), SubscriptionRegistrationOutcome::AddedBelowCapacity, ); + self.unique_pubkey_estimator.observe( + origin, + UniquePubkeyStage::SubscriptionAdded, + pubkey, + ); } AddAccountOutcome::Evicted(evicted) => { trace!(evicted = %evicted, "Evicting account"); @@ -1772,6 +1794,16 @@ impl RemoteAccountProvider { reason.into(), SubscriptionRegistrationOutcome::EvictedCandidate, ); + self.unique_pubkey_estimator.observe( + origin, + UniquePubkeyStage::SubscriptionEvicted, + &evicted, + ); + self.unique_pubkey_estimator.observe( + origin, + UniquePubkeyStage::SubscriptionAdded, + pubkey, + ); self.subscription_ownership.lock().await.remove(&evicted); // Inform upstream so it can remove it from the store. Failure @@ -1946,6 +1978,11 @@ impl RemoteAccountProvider { reason.into(), SubscriptionRegistrationOutcome::AlreadyPresent, ); + self.unique_pubkey_estimator.observe( + origin, + UniquePubkeyStage::SubscriptionAlreadyPresent, + pubkey, + ); return Ok(()); } drop(ownership); diff --git a/magicblock-chainlink/src/remote_account_provider/tests.rs b/magicblock-chainlink/src/remote_account_provider/tests.rs index a5bfdb7a3..bd21002f2 100644 --- a/magicblock-chainlink/src/remote_account_provider/tests.rs +++ b/magicblock-chainlink/src/remote_account_provider/tests.rs @@ -10,8 +10,9 @@ use magicblock_metrics::metrics::{ chainlink_pending_fetch_waiters_value, chainlink_subscription_cleanup_accounts_value, chainlink_subscription_registration_accounts_value, - chainlink_subscription_release_accounts_value, ChainlinkPendingFetchLayer, - ChainlinkPendingFetchOutcome, + chainlink_subscription_release_accounts_value, + chainlink_unique_pubkeys_estimate_value, ChainlinkPendingFetchLayer, + ChainlinkPendingFetchOutcome, ChainlinkUniquePubkeyWindow, }; use solana_account::Account; use solana_system_interface::program as system_program; @@ -19,6 +20,7 @@ use tokio::sync::mpsc; use super::*; use crate::{ + chainlink::unique_pubkey_estimator::UniquePubkeyStage, remote_account_provider::{ chain_pubsub_client::mock::ChainPubsubClientMock, chain_slot::ChainSlot, }, @@ -2270,6 +2272,49 @@ async fn test_registration_metric_already_present_on_duplicate_acquire() { assert_eq!(after - before, 1); } +#[tokio::test] +async fn test_unique_pubkey_metrics_observe_subscription_added_and_duplicate() { + init_logger(); + let _metric_guard = SUBSCRIPTION_LIFECYCLE_METRIC_TEST_GUARD.lock().await; + + let pubkey = solana_pubkey::Pubkey::new_unique(); + let account = Account { + lamports: 1_000_000, + data: vec![], + owner: system_program::id(), + executable: false, + rent_epoch: 0, + }; + let ProviderTestCtx { provider, .. } = + setup_provider(pubkey, account).await; + + provider + .acquire_subscription(&pubkey, SubscriptionReason::DirectAccount) + .await + .unwrap(); + provider + .acquire_subscription(&pubkey, SubscriptionReason::DirectAccount) + .await + .unwrap(); + + provider.unique_pubkey_estimator().force_export_for_tests(0); + + assert!( + chainlink_unique_pubkeys_estimate_value( + &SubscriptionRegistrationOrigin::Internal, + &UniquePubkeyStage::SubscriptionAdded, + ChainlinkUniquePubkeyWindow::OneMinute, + ) >= 1 + ); + assert!( + chainlink_unique_pubkeys_estimate_value( + &SubscriptionRegistrationOrigin::Internal, + &UniquePubkeyStage::SubscriptionAlreadyPresent, + ChainlinkUniquePubkeyWindow::OneMinute, + ) >= 1 + ); +} + #[tokio::test] async fn test_registration_metric_preserves_fetch_origin() { init_logger(); diff --git a/magicblock-metrics/src/metrics/mod.rs b/magicblock-metrics/src/metrics/mod.rs index 11b7f2ad9..47dee19a7 100644 --- a/magicblock-metrics/src/metrics/mod.rs +++ b/magicblock-metrics/src/metrics/mod.rs @@ -10,8 +10,9 @@ pub use types::{ BankPrecheckReason, ChainlinkCloneIntent, ChainlinkCloneMaterializationOutcome, ChainlinkCloneOutcome, ChainlinkCloneRemoteResult, ChainlinkEmptyPlaceholderStage, - ChainlinkPendingFetchLayer, ChainlinkPendingFetchOutcome, LabelValue, - Outcome, SubscriptionCleanupOutcome, SubscriptionCleanupSource, + ChainlinkPendingFetchLayer, ChainlinkPendingFetchOutcome, + ChainlinkUniquePubkeyWindow, LabelValue, Outcome, + SubscriptionCleanupOutcome, SubscriptionCleanupSource, SubscriptionReasonLabel, SubscriptionRegistrationOrigin, SubscriptionRegistrationOutcome, SubscriptionReleaseOutcome, }; @@ -206,6 +207,16 @@ lazy_static::lazy_static! { &["cleanup_source", "outcome"], ) .unwrap(); + + static ref CHAINLINK_UNIQUE_PUBKEYS_ESTIMATE: IntGaugeVec = + IntGaugeVec::new( + Opts::new( + "chainlink_unique_pubkeys_estimate", + "Approximate unique Chainlink pubkeys by origin, stage, and rolling window", + ), + &["origin", "stage", "window"], + ) + .unwrap(); static ref PROGRAM_SUBSCRIPTION_DISCOVERED_DLP_UPDATE_DELEGATED_ELSEWHERE_COUNT: IntCounter = IntCounter::new( "program_subscription_discovered_dlp_update_delegated_elsewhere_count", "DLP-owned subscription updates that, after fetching the delegation record, were found delegated to another validator and dropped", ).unwrap(); @@ -732,6 +743,7 @@ pub(crate) fn register() { register!(CHAINLINK_SUBSCRIPTION_REGISTRATION_ACCOUNTS_TOTAL); register!(CHAINLINK_SUBSCRIPTION_RELEASE_ACCOUNTS_TOTAL); register!(CHAINLINK_SUBSCRIPTION_CLEANUP_ACCOUNTS_TOTAL); + register!(CHAINLINK_UNIQUE_PUBKEYS_ESTIMATE); register!(PROGRAM_SUBSCRIPTION_DISCOVERED_DLP_UPDATE_DELEGATED_ELSEWHERE_COUNT); register!(PROGRAM_SUBSCRIPTION_ACCOUNT_UPDATES_COUNT); register!(ACCOUNT_SUBSCRIPTION_ACCOUNT_UPDATES_COUNT); @@ -952,6 +964,33 @@ pub fn inc_chainlink_subscription_cleanup_accounts( .inc(); } +pub fn set_chainlink_unique_pubkeys_estimate( + origin: &impl LabelValue, + stage: &impl LabelValue, + window: ChainlinkUniquePubkeyWindow, + estimate: u64, +) { + CHAINLINK_UNIQUE_PUBKEYS_ESTIMATE + .with_label_values(&[origin.value(), stage.value(), window.value()]) + .set(estimate as i64); +} + +#[cfg(any(test, feature = "dev-context"))] +pub fn chainlink_unique_pubkeys_estimate_value( + origin: &impl LabelValue, + stage: &impl LabelValue, + window: ChainlinkUniquePubkeyWindow, +) -> i64 { + CHAINLINK_UNIQUE_PUBKEYS_ESTIMATE + .get_metric_with_label_values(&[ + origin.value(), + stage.value(), + window.value(), + ]) + .map(|m| m.get()) + .unwrap_or(0) +} + #[cfg(any(test, feature = "dev-context"))] pub fn chainlink_subscription_registration_accounts_value( origin: SubscriptionRegistrationOrigin, diff --git a/magicblock-metrics/src/metrics/types.rs b/magicblock-metrics/src/metrics/types.rs index 934c8ee6b..92e356224 100644 --- a/magicblock-metrics/src/metrics/types.rs +++ b/magicblock-metrics/src/metrics/types.rs @@ -116,6 +116,43 @@ impl LabelValue for AccountFetchOrigin { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum ChainlinkUniquePubkeyWindow { + OneMinute, + FiveMinutes, + OneHour, +} + +impl ChainlinkUniquePubkeyWindow { + pub fn as_str(&self) -> &'static str { + match self { + Self::OneMinute => "1m", + Self::FiveMinutes => "5m", + Self::OneHour => "1h", + } + } + + pub fn minutes(&self) -> u64 { + match self { + Self::OneMinute => 1, + Self::FiveMinutes => 5, + Self::OneHour => 60, + } + } +} + +impl fmt::Display for ChainlinkUniquePubkeyWindow { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +impl LabelValue for ChainlinkUniquePubkeyWindow { + fn value(&self) -> &str { + self.as_str() + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ChainlinkPendingFetchLayer { FetchCloner,