From 407a2fe55689209c20a83fc5a8ca2e6498492974 Mon Sep 17 00:00:00 2001 From: Thorsten Lorenz Date: Thu, 2 Jul 2026 10:30:57 +0800 Subject: [PATCH 01/10] feat: add chainlink unique pubkey metric api --- .agents/context/crates/magicblock-metrics.md | 3 ++ magicblock-metrics/src/metrics/mod.rs | 43 +++++++++++++++++++- magicblock-metrics/src/metrics/types.rs | 37 +++++++++++++++++ 3 files changed, 81 insertions(+), 2 deletions(-) diff --git a/.agents/context/crates/magicblock-metrics.md b/.agents/context/crates/magicblock-metrics.md index eb3e851a8..9f9b0655d 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`. | @@ -229,6 +230,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. ### RPC and aperture @@ -340,6 +342,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-metrics/src/metrics/mod.rs b/magicblock-metrics/src/metrics/mod.rs index c0b4efd78..e0d75c26d 100644 --- a/magicblock-metrics/src/metrics/mod.rs +++ b/magicblock-metrics/src/metrics/mod.rs @@ -9,8 +9,9 @@ pub use types::{ AccountClone, AccountCommit, AccountFetchOrigin, BankPrecheckOutcome, BankPrecheckReason, ChainlinkCloneIntent, ChainlinkCloneMaterializationOutcome, ChainlinkCloneOutcome, - ChainlinkCloneRemoteResult, ChainlinkEmptyPlaceholderStage, LabelValue, - Outcome, SubscriptionCleanupOutcome, SubscriptionCleanupSource, + ChainlinkCloneRemoteResult, ChainlinkEmptyPlaceholderStage, + ChainlinkUniquePubkeyWindow, LabelValue, Outcome, + SubscriptionCleanupOutcome, SubscriptionCleanupSource, SubscriptionReasonLabel, SubscriptionRegistrationOrigin, SubscriptionRegistrationOutcome, SubscriptionReleaseOutcome, }; @@ -205,6 +206,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(); @@ -684,6 +695,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); @@ -900,6 +912,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 0816032c8..23ef04a94 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 BankPrecheckOutcome { BankHitNoFetch, From d7791abeb5279a370edd1d422a1b711b316b2166 Mon Sep 17 00:00:00 2001 From: Thorsten Lorenz Date: Thu, 2 Jul 2026 10:44:54 +0800 Subject: [PATCH 02/10] Add Chainlink unique pubkey estimator --- .../context/crates/magicblock-chainlink.md | 2 + .../src/chainlink/fetch_cloner/mod.rs | 43 ++ magicblock-chainlink/src/chainlink/mod.rs | 1 + .../src/chainlink/unique_pubkey_estimator.rs | 407 ++++++++++++++++++ .../src/remote_account_provider/mod.rs | 10 + 5 files changed, 463 insertions(+) create mode 100644 magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs diff --git a/.agents/context/crates/magicblock-chainlink.md b/.agents/context/crates/magicblock-chainlink.md index 717de01c0..8c8d74627 100644 --- a/.agents/context/crates/magicblock-chainlink.md +++ b/.agents/context/crates/magicblock-chainlink.md @@ -143,6 +143,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`. 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/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs b/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs index f86d17a74..d5eef52cd 100644 --- a/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs +++ b/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs @@ -75,6 +75,7 @@ use crate::{ blacklisted_accounts::{ blacklisted_accounts, programs_not_to_subscribe, }, + unique_pubkey_estimator::UniquePubkeyStage, }, cloner::{ errors::{ClonerError, ClonerResult}, @@ -2371,6 +2372,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( @@ -2444,6 +2453,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 @@ -2876,6 +2892,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![]; @@ -2892,6 +2915,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; } @@ -2922,6 +2948,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; } } @@ -2965,6 +2994,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 = @@ -3150,6 +3186,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 { @@ -3162,6 +3200,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/mod.rs b/magicblock-chainlink/src/chainlink/mod.rs index 2f4a844c3..cbd9ff4f3 100644 --- a/magicblock-chainlink/src/chainlink/mod.rs +++ b/magicblock-chainlink/src/chainlink/mod.rs @@ -35,6 +35,7 @@ mod blacklisted_accounts; pub mod config; pub mod errors; pub mod fetch_cloner; +pub(crate) mod unique_pubkey_estimator; pub use blacklisted_accounts::*; 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..410fa5f14 --- /dev/null +++ b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs @@ -0,0 +1,407 @@ +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(); + let estimator = Self { + series: Mutex::default(), + last_export_epoch_seconds: AtomicU64::default(), + }; + #[cfg(feature = "dev-context")] + estimator.force_export_for_tests(0); + estimator + } +} + +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()); + if !series.contains_key(origin_label) { + series.insert(origin_label.to_string(), HashMap::new()); + } + let stage_series = series + .get_mut(origin_label) + .expect("origin series was just inserted"); + if !stage_series.contains_key(stage_label) { + stage_series.insert( + stage_label.to_string(), + UniquePubkeySeries::new(origin_label, stage_label), + ); + } + let pubkey_series = stage_series + .get_mut(stage_label) + .expect("stage series was just inserted"); + pubkey_series.observe_hash(epoch_minute, pubkey_hash); + + if should_export { + export_series(&series, epoch_minute); + } + } + + 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"))] + 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 + } + } +} + +fn export_series( + series: &HashMap>, + now_epoch_minute: u64, +) { + 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; + metrics::set_chainlink_unique_pubkeys_estimate( + &pubkey_series.origin_label, + &pubkey_series.stage_label, + window, + estimate, + ); + } + } + } +} + +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_eq!( + 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 7c6e85284..26aa4d1d2 100644 --- a/magicblock-chainlink/src/remote_account_provider/mod.rs +++ b/magicblock-chainlink/src/remote_account_provider/mod.rs @@ -82,6 +82,7 @@ use magicblock_metrics::{ pub use remote_account::{ResolvedAccount, ResolvedAccountSharedData}; use crate::{ + chainlink::unique_pubkey_estimator::UniquePubkeyEstimator, errors::ChainlinkResult, remote_account_provider::{ chain_updates_client::ChainUpdatesClient, @@ -506,6 +507,8 @@ pub struct RemoteAccountProvider { subscription_forwarder: Arc>, + unique_pubkey_estimator: Arc, + /// Task that periodically updates the active subscriptions gauge _active_subscriptions_task_handle: Option>, } @@ -704,6 +707,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, @@ -727,6 +731,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, From 9c37094d790ef7530e4c1fed4834de662d94c54d Mon Sep 17 00:00:00 2001 From: Thorsten Lorenz Date: Thu, 2 Jul 2026 10:52:36 +0800 Subject: [PATCH 03/10] Wire Chainlink unique pubkey outcome stages --- .../context/crates/magicblock-chainlink.md | 2 +- .../src/chainlink/fetch_cloner/mod.rs | 13 +++++++++- magicblock-chainlink/src/chainlink/mod.rs | 13 +++++++++- .../src/remote_account_provider/mod.rs | 24 ++++++++++++++++++- 4 files changed, 48 insertions(+), 4 deletions(-) diff --git a/.agents/context/crates/magicblock-chainlink.md b/.agents/context/crates/magicblock-chainlink.md index 8c8d74627..8fd8ea2f6 100644 --- a/.agents/context/crates/magicblock-chainlink.md +++ b/.agents/context/crates/magicblock-chainlink.md @@ -143,7 +143,7 @@ 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`. 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. +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 diff --git a/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs b/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs index d5eef52cd..dcaadd693 100644 --- a/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs +++ b/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs @@ -75,7 +75,7 @@ use crate::{ blacklisted_accounts::{ blacklisted_accounts, programs_not_to_subscribe, }, - unique_pubkey_estimator::UniquePubkeyStage, + unique_pubkey_estimator::{UniquePubkeyEstimator, UniquePubkeyStage}, }, cloner::{ errors::{ClonerError, ClonerResult}, @@ -309,6 +309,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) @@ -779,6 +785,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, diff --git a/magicblock-chainlink/src/chainlink/mod.rs b/magicblock-chainlink/src/chainlink/mod.rs index cbd9ff4f3..55c510d40 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; @@ -551,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) } diff --git a/magicblock-chainlink/src/remote_account_provider/mod.rs b/magicblock-chainlink/src/remote_account_provider/mod.rs index 26aa4d1d2..e2f552dea 100644 --- a/magicblock-chainlink/src/remote_account_provider/mod.rs +++ b/magicblock-chainlink/src/remote_account_provider/mod.rs @@ -82,7 +82,9 @@ use magicblock_metrics::{ pub use remote_account::{ResolvedAccount, ResolvedAccountSharedData}; use crate::{ - chainlink::unique_pubkey_estimator::UniquePubkeyEstimator, + chainlink::unique_pubkey_estimator::{ + UniquePubkeyEstimator, UniquePubkeyStage, + }, errors::ChainlinkResult, remote_account_provider::{ chain_updates_client::ChainUpdatesClient, @@ -1631,6 +1633,11 @@ impl RemoteAccountProvider { reason.into(), SubscriptionRegistrationOutcome::AlreadyPresent, ); + self.unique_pubkey_estimator.observe( + origin, + UniquePubkeyStage::SubscriptionAlreadyPresent, + pubkey, + ); } AddAccountOutcome::Added => { inc_chainlink_subscription_registration_accounts( @@ -1638,6 +1645,11 @@ impl RemoteAccountProvider { reason.into(), SubscriptionRegistrationOutcome::AddedBelowCapacity, ); + self.unique_pubkey_estimator.observe( + origin, + UniquePubkeyStage::SubscriptionAdded, + pubkey, + ); } AddAccountOutcome::Evicted(evicted) => { trace!(evicted = %evicted, "Evicting account"); @@ -1691,6 +1703,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 From 5694fed75306d8d1897231a4f5c0c459407031ca Mon Sep 17 00:00:00 2001 From: Thorsten Lorenz Date: Thu, 2 Jul 2026 16:35:45 +0800 Subject: [PATCH 04/10] fix record duplicate subscription unique pubkeys --- .../src/chainlink/fetch_cloner/tests.rs | 96 +++++++++++++++++++ magicblock-chainlink/src/chainlink/mod.rs | 59 +++++++++++- .../src/remote_account_provider/mod.rs | 5 + .../src/remote_account_provider/tests.rs | 45 +++++++++ 4 files changed, 203 insertions(+), 2 deletions(-) diff --git a/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs b/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs index f22157fa2..7477ce201 100644 --- a/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs +++ b/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs @@ -1,6 +1,10 @@ use std::{collections::HashMap, sync::Arc, time::Duration}; use dlp_api::state::DelegationRecord; +use magicblock_metrics::metrics::{ + chainlink_unique_pubkeys_estimate_value, ChainlinkCloneOutcome, + ChainlinkUniquePubkeyWindow, +}; use solana_account::{ Account, AccountSharedData, ReadableAccount, WritableAccount, }; @@ -25,6 +29,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, @@ -2279,6 +2284,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 55c510d40..b2d0225a5 100644 --- a/magicblock-chainlink/src/chainlink/mod.rs +++ b/magicblock-chainlink/src/chainlink/mod.rs @@ -617,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, @@ -697,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(); @@ -714,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/remote_account_provider/mod.rs b/magicblock-chainlink/src/remote_account_provider/mod.rs index e2f552dea..3d0fa7602 100644 --- a/magicblock-chainlink/src/remote_account_provider/mod.rs +++ b/magicblock-chainlink/src/remote_account_provider/mod.rs @@ -1887,6 +1887,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 7e2480612..4f5b1c5a8 100644 --- a/magicblock-chainlink/src/remote_account_provider/tests.rs +++ b/magicblock-chainlink/src/remote_account_provider/tests.rs @@ -8,6 +8,7 @@ use magicblock_metrics::metrics::{ chainlink_subscription_cleanup_accounts_value, chainlink_subscription_registration_accounts_value, chainlink_subscription_release_accounts_value, + chainlink_unique_pubkeys_estimate_value, ChainlinkUniquePubkeyWindow, }; use solana_account::Account; use solana_system_interface::program as system_program; @@ -15,6 +16,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, }, @@ -1964,6 +1966,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(); From d15f87e590fa1ccee40c006bfa0c5ada203f86d4 Mon Sep 17 00:00:00 2001 From: Thorsten Lorenz Date: Thu, 2 Jul 2026 18:31:07 +0800 Subject: [PATCH 05/10] fix: track clone failure pubkeys Amp-Thread-ID: https://ampcode.com/threads/T-019f225e-8224-765c-ba4e-6985c3edd79f Co-authored-by: Amp --- magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs b/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs index 23018b70d..478c97586 100644 --- a/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs +++ b/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs @@ -939,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, @@ -1167,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, From 5ac82f14aca1996650ca5fa307553bfbbb684bf2 Mon Sep 17 00:00:00 2001 From: Thorsten Lorenz Date: Thu, 2 Jul 2026 18:35:22 +0800 Subject: [PATCH 06/10] fix: remove no-op pubkey export Amp-Thread-ID: https://ampcode.com/threads/T-019f225e-8224-765c-ba4e-6985c3edd79f Co-authored-by: Amp --- .../src/chainlink/unique_pubkey_estimator.rs | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs index 410fa5f14..6362599ff 100644 --- a/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs +++ b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs @@ -71,13 +71,10 @@ pub(crate) struct UniquePubkeyEstimator { impl Default for UniquePubkeyEstimator { fn default() -> Self { debug_assert_unique_stage_labels(); - let estimator = Self { + Self { series: Mutex::default(), last_export_epoch_seconds: AtomicU64::default(), - }; - #[cfg(feature = "dev-context")] - estimator.force_export_for_tests(0); - estimator + } } } @@ -159,6 +156,7 @@ impl UniquePubkeyEstimator { } #[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()); From 65c7c618b266a76c78d2e835883a5cddaac34710 Mon Sep 17 00:00:00 2001 From: Thorsten Lorenz Date: Thu, 2 Jul 2026 18:38:55 +0800 Subject: [PATCH 07/10] perf: drop series mutex before prometheus writes Amp-Thread-ID: https://ampcode.com/threads/T-019f2266-6bc4-729a-9f45-96e55b52e7aa Co-authored-by: Amp --- .../src/chainlink/unique_pubkey_estimator.rs | 56 ++++++++++++++++--- 1 file changed, 48 insertions(+), 8 deletions(-) diff --git a/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs index 6362599ff..3137bfd14 100644 --- a/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs +++ b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs @@ -132,8 +132,15 @@ impl UniquePubkeyEstimator { .expect("stage series was just inserted"); pubkey_series.observe_hash(epoch_minute, pubkey_hash); - if should_export { - export_series(&series, epoch_minute); + // 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); } } @@ -276,25 +283,58 @@ impl HllSketch { } } -fn export_series( +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; - metrics::set_chainlink_unique_pubkeys_estimate( - &pubkey_series.origin_label, - &pubkey_series.stage_label, + 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() { From 4b320e9d2b69137fc51081ccc193b6554127bfe7 Mon Sep 17 00:00:00 2001 From: Thorsten Lorenz Date: Thu, 2 Jul 2026 18:41:45 +0800 Subject: [PATCH 08/10] fix: use entry for pubkey series Amp-Thread-ID: https://ampcode.com/threads/T-019f2266-6bc4-729a-9f45-96e55b52e7aa Co-authored-by: Amp --- .../src/chainlink/unique_pubkey_estimator.rs | 19 +++++-------------- 1 file changed, 5 insertions(+), 14 deletions(-) diff --git a/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs index 3137bfd14..ff3e7b5b0 100644 --- a/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs +++ b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs @@ -115,21 +115,12 @@ impl UniquePubkeyEstimator { let mut series = self.series.lock().unwrap_or_else(|err| err.into_inner()); - if !series.contains_key(origin_label) { - series.insert(origin_label.to_string(), HashMap::new()); - } - let stage_series = series - .get_mut(origin_label) - .expect("origin series was just inserted"); - if !stage_series.contains_key(stage_label) { - stage_series.insert( - stage_label.to_string(), - UniquePubkeySeries::new(origin_label, stage_label), - ); - } + let stage_series = series.entry(origin_label.to_string()).or_default(); let pubkey_series = stage_series - .get_mut(stage_label) - .expect("stage series was just inserted"); + .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 From 7a8456a4217f81532c5445d62cea5f2d5fb1b7b9 Mon Sep 17 00:00:00 2001 From: Thorsten Lorenz Date: Thu, 2 Jul 2026 18:51:38 +0800 Subject: [PATCH 09/10] test: relax shared pubkey gauge asserts Amp-Thread-ID: https://ampcode.com/threads/T-019f2272-19c8-7408-84e8-451c3c36ec8a Co-authored-by: Amp --- .../src/chainlink/unique_pubkey_estimator.rs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs index ff3e7b5b0..8532efc2e 100644 --- a/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs +++ b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs @@ -388,21 +388,19 @@ mod tests { estimator.observe_at(origin, stage, &pubkey_from_u64(2), 120); estimator.force_export_for_tests(120); - assert_eq!( + assert!( chainlink_unique_pubkeys_estimate_value( &origin, &stage, ChainlinkUniquePubkeyWindow::OneMinute, - ), - 1 + ) >= 1 ); - assert_eq!( + assert!( chainlink_unique_pubkeys_estimate_value( &origin, &stage, ChainlinkUniquePubkeyWindow::FiveMinutes, - ), - 2 + ) >= 2 ); } From 770d453eb61cab2adf4cc19f2b0a5b87adbe7a62 Mon Sep 17 00:00:00 2001 From: Thorsten Lorenz Date: Mon, 6 Jul 2026 10:27:29 +0200 Subject: [PATCH 10/10] test: tighten pubkey window regression --- .../src/chainlink/unique_pubkey_estimator.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs index 8532efc2e..144dd04e5 100644 --- a/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs +++ b/magicblock-chainlink/src/chainlink/unique_pubkey_estimator.rs @@ -388,12 +388,13 @@ mod tests { estimator.observe_at(origin, stage, &pubkey_from_u64(2), 120); estimator.force_export_for_tests(120); - assert!( + assert_eq!( chainlink_unique_pubkeys_estimate_value( &origin, &stage, ChainlinkUniquePubkeyWindow::OneMinute, - ) >= 1 + ), + 1 ); assert!( chainlink_unique_pubkeys_estimate_value(