feat: add chainlink unique pubkey cardinality metric

.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:

.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<T, E>` 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.
 

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<UniquePubkeyEstimator> {
+        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<ChainlinkResult<AccountWithCompanion>> {
         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(

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();