From 6570a24db6399264a4b3c3acc8c368bbd153b7b0 Mon Sep 17 00:00:00 2001 From: Sarfaraz Nawaz Date: Wed, 1 Jul 2026 00:59:20 +0530 Subject: [PATCH 1/2] feat: Impl UndelegationRequest backfill loop --- .agents/context/crates/magicblock-api.md | 12 +-- .../context/crates/magicblock-chainlink.md | 12 ++- .agents/context/crates/magicblock-config.md | 5 +- .agents/specs/validator-specification.md | 10 ++- config.example.toml | 6 ++ magicblock-api/src/magic_validator.rs | 1 + .../src/chainlink/fetch_cloner/mod.rs | 62 ++++++++++++++- .../src/chainlink/fetch_cloner/tests.rs | 74 ++++++++++++++++- magicblock-chainlink/src/chainlink/mod.rs | 20 +++++ .../chain_rpc_client.rs | 45 ++++++++++- .../src/remote_account_provider/mod.rs | 45 ++++++++++- .../src/testing/rpc_client_mock.rs | 61 +++++++++++++- magicblock-config/src/config/chain.rs | 8 ++ magicblock-config/src/consts.rs | 3 + magicblock-config/src/tests.rs | 19 +++++ .../src/undelegation_request_service.rs | 79 ++++++++++++++++++- 16 files changed, 447 insertions(+), 15 deletions(-) diff --git a/.agents/context/crates/magicblock-api.md b/.agents/context/crates/magicblock-api.md index 63c650573..ad1386794 100644 --- a/.agents/context/crates/magicblock-api.md +++ b/.agents/context/crates/magicblock-api.md @@ -136,8 +136,9 @@ Caveats: 5. Standalone/primary nodes reset the bank and, for primaries, send a replication `Message::Reset`. 6. The intent execution service starts only after replay/reset and handles pending commit intent recovery before its normal accept loop. 7. Standalone nodes switch the scheduler to `SchedulerMode::Primary`; primary/replica modes spawn the replication service instead. -8. Standalone ephemeral nodes start background base-layer setup: funding check, magic fee vault init/delegation, optional startup fee claim, and optional domain registration. -9. The undelegation request service, claim-fees periodic task, intent execution service, ledger truncator, and primary-only task scheduler are started. +8. Non-replica validators start `UndelegationRequestService`; it consumes Chainlink observed requests and the configured polling backfill. Replicas keep Chainlink disabled and do not scan DLP requests. +9. Standalone ephemeral nodes start background base-layer setup: funding check, magic fee vault init/delegation, optional startup fee claim, and optional domain registration. +10. Claim-fees periodic task, intent execution service, ledger truncator, and primary-only task scheduler are started. ### Scheduled intent service flow @@ -219,9 +220,10 @@ The ledger stores a validator keypair file beside the blockstore parent. With `v 5. Ephemeral vault initialization must preserve the ephemeral flag and Magic Program owner. 6. Validator identity must remain privileged in local AccountsDb. 7. Committor service wiring must stay available before Magic Program execution can fetch commit nonces or schedule settlement work. -8. Do not add blocking RPC, slow I/O, or expensive serialization to scheduler/executor hot paths from this crate; keep such work in startup/background services where possible. -9. Shutdown must cancel/stop services before flushing AccountsDb and ledger. -10. Operator-facing config keys and base-layer registration/fee-vault behavior are compatibility-sensitive. +8. Owner-program DLP undelegation request polling must stay in the background processor path, not the slot ticker or transaction-execution hot path. +9. Do not add blocking RPC, slow I/O, or expensive serialization to scheduler/executor hot paths from this crate; keep such work in startup/background services where possible. +10. Shutdown must cancel/stop services before flushing AccountsDb and ledger. +11. Operator-facing config keys and base-layer registration/fee-vault behavior are compatibility-sensitive. ## Common change areas and what to inspect diff --git a/.agents/context/crates/magicblock-chainlink.md b/.agents/context/crates/magicblock-chainlink.md index 6fa6e0918..3595c482d 100644 --- a/.agents/context/crates/magicblock-chainlink.md +++ b/.agents/context/crates/magicblock-chainlink.md @@ -76,6 +76,7 @@ Important methods: - `fetch_accounts(pubkeys, fetch_origin)`: ensures accounts and then reads them from the local bank. - `accounts_delegated_on_base_and_er(pubkeys, fetch_origin)`: checks that each account is DLP-owned on base and represented as delegated/DLP-owned locally. - `undelegation_requested(pubkey)`: called by committor/account flows before an account is undelegated so Chainlink keeps watching for base-layer completion. +- `fetch_undelegation_requests()`: scans base-layer Delegation Program accounts for active `UndelegationRequest` PDAs using filtered `getProgramAccounts` and returns decoded `ObservedUndelegationRequest`s for `magicblock-accounts`. - `fetch_count()` / `is_watching()`: mainly observability/testing helpers. Disabled replication mode is intentionally conservative: @@ -274,6 +275,15 @@ Key behavior: - Non-advancing updates are ignored unless they represent a same-slot delegated refresh needed for undelegate/redelegate recovery. - Delegated updates cause direct subscription cleanup; undelegation-completion updates retain/directly ensure subscriptions as appropriate and release `UndelegationTracking` ownership. +### DLP undelegation request scanning + +Owner-program undelegation requests are discovered in two ways: + +- Live updates: DLP-owned `UndelegationRequest` account subscription/program-subscription updates are decoded in `FetchCloner::process_subscription_update` and broadcast as `ObservedUndelegationRequest`. +- Backfill scans: `FetchCloner::fetch_undelegation_requests` calls `getProgramAccounts` for `dlp_api::id()` with a `DataSize(UndelegationRequest::size_with_discriminator())` filter and a discriminator `memcmp` at offset `0`, then decodes each returned account with `UndelegationRequest::try_from_bytes_with_discriminator`. + +The scan uses Base64Zstd account encoding and gets a nearby base-chain slot for `observed_slot`. Malformed matching accounts are logged and skipped; a bad account must not abort the whole scan. Polling cadence is controlled by `chainlink.undelegation-request-poll-interval` in `magicblock-config` and consumed by `magicblock-accounts`. + ### Greedy discovery If a subscription update discovers a DLP-owned account absent from the bank, Chainlink may greedily fetch and clone it if the delegation record says it belongs to this validator (or is confined). This is especially important for delegated eATA discovery and owner-program subscriptions. @@ -368,7 +378,7 @@ Changing SubMux behavior can affect ordering, duplicate clone submissions, and p ## Lifecycle mode and configuration -`ChainlinkConfig` wraps `RemoteAccountProviderConfig` and currently includes `remove_confined_accounts`. +`ChainlinkConfig` wraps `RemoteAccountProviderConfig` and includes settings such as `remove_confined_accounts`, allowed program filters, resubscription delay, Range risk checks, and `undelegation_request_poll_interval` for the DLP request backfill consumer in `magicblock-accounts`. `RemoteAccountProviderConfig` includes: diff --git a/.agents/context/crates/magicblock-config.md b/.agents/context/crates/magicblock-config.md index f10971e50..0d9b7e4d1 100644 --- a/.agents/context/crates/magicblock-config.md +++ b/.agents/context/crates/magicblock-config.md @@ -169,6 +169,7 @@ Most structs use `rename_all = "kebab-case"`; environment variables are upper sn - `ledger.block-time` -> `MBV_LEDGER__BLOCK_TIME` - `task-scheduler.failed-task-cleanup-interval` -> `MBV_TASK_SCHEDULER__FAILED_TASK_CLEANUP_INTERVAL` - `chainlink.risk.request-timeout` -> `MBV_CHAINLINK__RISK__REQUEST_TIMEOUT` +- `chainlink.undelegation-request-poll-interval` -> `MBV_CHAINLINK__UNDELEGATION_REQUEST_POLL_INTERVAL` Do not introduce aliases casually. Operator docs, `config.example.toml`, integration tooling, and deployment configs depend on stable keys. @@ -182,7 +183,7 @@ Most config structs use `deny_unknown_fields`. This catches typos and stale conf ### Defaults are operational behavior -Defaults are not merely test conveniences. They set devnet remotes, local storage, development validator keypair, base fee, commit compute unit price, account DB size, ledger timing, metrics cadence, Chainlink monitoring capacity, Range risk defaults, task scheduler timings, and gRPC stream limits. Changing defaults can affect local developer flows, integration tests, startup performance, storage usage, and network traffic. +Defaults are not merely test conveniences. They set devnet remotes, local storage, development validator keypair, base fee, commit compute unit price, account DB size, ledger timing, metrics cadence, Chainlink monitoring capacity, DLP undelegation-request polling cadence, Range risk defaults, task scheduler timings, and gRPC stream limits. Changing defaults can affect local developer flows, integration tests, startup performance, storage usage, and network traffic. ### Config example is tested @@ -277,12 +278,14 @@ Inspect first: - `magicblock-config/src/config/chain.rs` and `grpc.rs`; - `magicblock-api/src/magic_validator.rs::init_chainlink`; - `magicblock-chainlink/src/remote_account_provider/`; +- `magicblock-services/src/undelegation_request_service.rs` for DLP undelegation-request polling; - `magicblock-aml` Range risk usage; - allowed-program filtering tests in Chainlink. Risks: - subscription limits and resubscription delay affect account sync throughput and provider load; +- undelegation-request polling adds periodic base-layer RPC load and should stay disabled with `0s` only when subscription backfill is not needed; - allowed-program semantics treat `None` and `Some(vec![])` as unrestricted in current Chainlink code; - risk config may add external I/O and should remain explicitly disabled by default. diff --git a/.agents/specs/validator-specification.md b/.agents/specs/validator-specification.md index 239cdf3eb..987dbc870 100644 --- a/.agents/specs/validator-specification.md +++ b/.agents/specs/validator-specification.md @@ -269,6 +269,13 @@ owner, rent payer, and commit nonce information must come from the delegation record/metadata. Validators must still verify that the request PDA matches the delegated account before scheduling. +The validator observes owner-program `UndelegationRequest` PDAs through +Chainlink and `magicblock-accounts`. Live request-account subscription updates +are the low-latency path. A background backfill loop also scans DLP-owned +accounts with a filtered `getProgramAccounts` every configured interval +(`chainlink.undelegation-request-poll-interval`, default `5m`) and feeds the +decoded requests into the same `ScheduleCommitAndUndelegate` scheduling path. + The current Delegation Program does not undelegate from finalize instructions. Commit/finalize-style instructions only commit/finalize state and record or preserve `DelegationMetadata.undelegation_requester`. The validator-side @@ -280,7 +287,8 @@ and request account cleanup. When metadata says `Validator` or is still `None` at task-build time, no request account is passed. `None` can be valid for validator-requested undelegation because commit and finalize task lists are built before the commit-stage transaction records the validator requester on -base. +base. `AlreadyUndelegated` remains a conflict for validator-requested +undelegation state, not a reason to drop an owner-program request. ### Callback discriminator diff --git a/config.example.toml b/config.example.toml index 4b7e04840..191dec256 100644 --- a/config.example.toml +++ b/config.example.toml @@ -263,6 +263,12 @@ remove-confined-accounts = false # Env: MBV_CHAINLINK__RESUBSCRIPTION_DELAY # resubscription-delay = "50ms" +# Period for polling DLP-owned UndelegationRequest accounts as a backfill for +# missed subscription updates. Set to "0s" to disable polling. +# Default: "5m" +# Env: MBV_CHAINLINK__UNDELEGATION_REQUEST_POLL_INTERVAL +# undelegation-request-poll-interval = "5m" + # ------------------------------------------------------------------------------ # Optional: Range Risk API validation for post-delegation actions signers # ------------------------------------------------------------------------------ diff --git a/magicblock-api/src/magic_validator.rs b/magicblock-api/src/magic_validator.rs index e23f806f8..43eb677f9 100644 --- a/magicblock-api/src/magic_validator.rs +++ b/magicblock-api/src/magic_validator.rs @@ -337,6 +337,7 @@ impl MagicValidator { dispatch.transaction_scheduler.clone(), identity_keypair.insecure_clone(), ledger.latest_block().clone(), + config.chainlink.undelegation_request_poll_interval, )) }); diff --git a/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs b/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs index 465fb8633..727cf4b4b 100644 --- a/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs +++ b/magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs @@ -10,7 +10,10 @@ use std::{ use dlp_api::{ pda::delegation_record_pda_from_delegated_account, - state::{DelegationRecord, UndelegationRequest}, + state::{ + discriminator::AccountDiscriminator, DelegationRecord, + UndelegationRequest, + }, }; use lru::LruCache; use magicblock_accounts_db::traits::AccountsBank; @@ -30,8 +33,13 @@ use magicblock_metrics::metrics::{ use parking_lot::Mutex as PlMutex; use scc::HashMap; use solana_account::{AccountSharedData, ReadableAccount}; +use solana_account_decoder_client_types::UiAccountEncoding; use solana_keypair::Keypair; use solana_pubkey::Pubkey; +use solana_rpc_client_api::{ + config::{RpcAccountInfoConfig, RpcProgramAccountsConfig}, + filter::{Memcmp, RpcFilterType}, +}; use solana_sdk_ids::system_program; use solana_signature::Signature; use solana_signer::Signer; @@ -334,6 +342,58 @@ where .await?) } + pub async fn fetch_undelegation_requests( + &self, + ) -> ChainlinkResult> { + let observed_slot = self.remote_account_provider.get_slot().await?; + let config = RpcProgramAccountsConfig { + filters: Some(vec![ + RpcFilterType::DataSize( + UndelegationRequest::size_with_discriminator() as u64, + ), + RpcFilterType::Memcmp(Memcmp::new_raw_bytes( + 0, + AccountDiscriminator::UndelegationRequest + .to_bytes() + .to_vec(), + )), + ]), + account_config: RpcAccountInfoConfig { + encoding: Some(UiAccountEncoding::Base64Zstd), + ..Default::default() + }, + ..Default::default() + }; + let accounts = self + .remote_account_provider + .get_program_accounts_with_config(&dlp_api::id(), config) + .await?; + + let mut requests = Vec::with_capacity(accounts.len()); + for (request_pda, account) in accounts { + let Ok(request) = + UndelegationRequest::try_from_bytes_with_discriminator( + &account.data, + ) + else { + warn!( + request_pda = %request_pda, + data_len = account.data.len(), + "Skipping malformed DLP undelegation request account" + ); + continue; + }; + requests.push(ObservedUndelegationRequest { + request_pda, + delegated_account: request.delegated_account, + expires_at_slot: request.expires_at_slot, + observed_slot, + }); + } + + Ok(requests) + } + pub fn cloner(&self) -> &Arc { &self.cloner } diff --git a/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs b/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs index 8c0547d96..3a4ce815c 100644 --- a/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs +++ b/magicblock-chainlink/src/chainlink/fetch_cloner/tests.rs @@ -1,6 +1,9 @@ use std::{collections::HashMap, sync::Arc, time::Duration}; -use dlp_api::state::DelegationRecord; +use dlp_api::{ + pda::undelegation_request_pda_from_delegated_account, + state::{DelegationRecord, UndelegationRequest}, +}; use magicblock_metrics::metrics::{ chainlink_pending_fetch_accounts_value, chainlink_pending_fetch_waiters_gauge_value, @@ -220,6 +223,29 @@ where } } +fn undelegation_request_data( + delegated_account: Pubkey, + expires_at_slot: u64, +) -> Vec { + let request = UndelegationRequest { + delegated_account, + expires_at_slot, + }; + let mut data = vec![0; UndelegationRequest::size_with_discriminator()]; + request.to_bytes_with_discriminator(&mut data).unwrap(); + data +} + +fn dlp_account(data: Vec) -> Account { + Account { + lamports: 1_000_000, + data, + owner: dlp_api::id(), + executable: false, + rent_epoch: 0, + } +} + fn insert_plain_ata_in_bank( accounts_bank: &Arc, ata_pubkey: Pubkey, @@ -577,6 +603,52 @@ fn pending_waiters_gauge_value() -> i64 { ) } +#[tokio::test] +async fn fetch_undelegation_requests_scans_filtered_dlp_accounts() { + let delegated_account = random_pubkey(); + let request_pda = + undelegation_request_pda_from_delegated_account(&delegated_account); + let request_data = undelegation_request_data(delegated_account, 20); + + let ctx = setup( + vec![ + (request_pda, dlp_account(request_data.clone())), + ( + random_pubkey(), + dlp_account(vec![ + 0; + UndelegationRequest::size_with_discriminator() + ]), + ), + ( + random_pubkey(), + Account { + owner: system_program::id(), + data: request_data, + ..Default::default() + }, + ), + ], + 42, + Keypair::new(), + ) + .await; + + let requests = ctx + .fetch_cloner + .fetch_undelegation_requests() + .await + .unwrap(); + + assert_eq!(ctx.rpc_client.program_account_fetches(), 1); + assert_eq!(requests.len(), 1); + let request = &requests[0]; + assert_eq!(request.request_pda, request_pda); + assert_eq!(request.delegated_account, delegated_account); + assert_eq!(request.expires_at_slot, 20); + assert_eq!(request.observed_slot, 42); +} + // ----------------- // Single Account Tests // ----------------- diff --git a/magicblock-chainlink/src/chainlink/mod.rs b/magicblock-chainlink/src/chainlink/mod.rs index 991800c84..111f62377 100644 --- a/magicblock-chainlink/src/chainlink/mod.rs +++ b/magicblock-chainlink/src/chainlink/mod.rs @@ -180,6 +180,17 @@ impl } } + pub async fn fetch_undelegation_requests( + &self, + ) -> ChainlinkResult> { + match self { + Self::Enabled(chainlink) => { + chainlink.fetch_undelegation_requests().await + } + Self::Disabled => Ok(Vec::new()), + } + } + pub fn fetch_count(&self) -> Option { match self { Self::Enabled(chainlink) => chainlink.fetch_count(), @@ -627,6 +638,15 @@ impl Ok(()) } + pub async fn fetch_undelegation_requests( + &self, + ) -> ChainlinkResult> { + let Some(fetch_cloner) = self.fetch_cloner() else { + return Ok(Vec::new()); + }; + fetch_cloner.fetch_undelegation_requests().await + } + pub fn fetch_cloner(&self) -> Option<&Arc>> { self.fetch_cloner.as_ref() } diff --git a/magicblock-chainlink/src/remote_account_provider/chain_rpc_client.rs b/magicblock-chainlink/src/remote_account_provider/chain_rpc_client.rs index f82f29b47..f1156cf82 100644 --- a/magicblock-chainlink/src/remote_account_provider/chain_rpc_client.rs +++ b/magicblock-chainlink/src/remote_account_provider/chain_rpc_client.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::{str::FromStr, sync::Arc}; use async_trait::async_trait; use solana_account::Account; @@ -7,7 +7,7 @@ use solana_pubkey::Pubkey; use solana_rpc_client::nonblocking::rpc_client::RpcClient; use solana_rpc_client_api::{ client_error, - config::RpcAccountInfoConfig, + config::{RpcAccountInfoConfig, RpcProgramAccountsConfig}, response::{Response, RpcResult}, }; @@ -31,6 +31,12 @@ pub trait ChainRpcClient: Send + Sync + Clone + 'static { pubkeys: &[Pubkey], config: RpcAccountInfoConfig, ) -> RpcResult>>; + + async fn get_program_accounts_with_config( + &self, + pubkey: &Pubkey, + config: RpcProgramAccountsConfig, + ) -> client_error::Result>; } // ----------------- @@ -97,4 +103,39 @@ impl ChainRpcClient for ChainRpcClientImpl { .collect(), }) } + + async fn get_program_accounts_with_config( + &self, + pubkey: &Pubkey, + config: RpcProgramAccountsConfig, + ) -> client_error::Result> { + self.rpc_client + .get_program_ui_accounts_with_config(pubkey, config) + .await? + .into_iter() + .map(|(pubkey, account)| { + let data = account.data.decode().ok_or_else(|| { + client_error::ErrorKind::Custom(format!( + "failed to decode program account {pubkey} data" + )) + })?; + let owner = + Pubkey::from_str(&account.owner).map_err(|err| { + client_error::ErrorKind::Custom(format!( + "failed to decode program account {pubkey} owner: {err}" + )) + })?; + Ok(( + pubkey, + Account { + lamports: account.lamports, + data, + owner, + executable: account.executable, + rent_epoch: account.rent_epoch, + }, + )) + }) + .collect() + } } diff --git a/magicblock-chainlink/src/remote_account_provider/mod.rs b/magicblock-chainlink/src/remote_account_provider/mod.rs index e630fbaa8..a63dfeae0 100644 --- a/magicblock-chainlink/src/remote_account_provider/mod.rs +++ b/magicblock-chainlink/src/remote_account_provider/mod.rs @@ -26,7 +26,8 @@ use solana_account_decoder_client_types::UiAccountEncoding; use solana_commitment_config::CommitmentConfig; use solana_pubkey::Pubkey; use solana_rpc_client_api::{ - client_error::ErrorKind, config::RpcAccountInfoConfig, + client_error::ErrorKind, + config::{RpcAccountInfoConfig, RpcProgramAccountsConfig}, custom_error::JSON_RPC_SERVER_ERROR_MIN_CONTEXT_SLOT_NOT_REACHED, request::RpcError, }; @@ -1022,6 +1023,48 @@ impl RemoteAccountProvider { self.lrucache_subscribed_accounts.promote_multi(pubkeys); } + pub(crate) async fn get_slot(&self) -> RemoteAccountProviderResult { + tokio::time::timeout(RPC_FETCH_TIMEOUT, self.rpc_client.get_slot()) + .await + .map_err(|_| { + RemoteAccountProviderError::AccountResolutionsFailed(format!( + "RPC call timeout fetching slot after {}ms", + RPC_FETCH_TIMEOUT.as_millis() + )) + })? + .map_err(|err| { + RemoteAccountProviderError::AccountResolutionsFailed(format!( + "RpcError fetching slot: {err:?}" + )) + }) + } + + pub(crate) async fn get_program_accounts_with_config( + &self, + pubkey: &Pubkey, + mut config: RpcProgramAccountsConfig, + ) -> RemoteAccountProviderResult> { + config.account_config.commitment = Some(self.rpc_client.commitment()); + + tokio::time::timeout(RPC_FETCH_TIMEOUT, async { + self.rpc_client + .get_program_accounts_with_config(pubkey, config) + .await + }) + .await + .map_err(|_| { + RemoteAccountProviderError::AccountResolutionsFailed(format!( + "RPC call timeout fetching program accounts for {pubkey} after {}ms", + RPC_FETCH_TIMEOUT.as_millis() + )) + })? + .map_err(|err| { + RemoteAccountProviderError::AccountResolutionsFailed(format!( + "RpcError fetching program accounts for {pubkey}: {err:?}" + )) + }) + } + pub(crate) fn set_capacity_eviction_protection(&self, predicate: F) where F: Fn(&Pubkey) -> CapacityEvictionProtection + Send + Sync + 'static, diff --git a/magicblock-chainlink/src/testing/rpc_client_mock.rs b/magicblock-chainlink/src/testing/rpc_client_mock.rs index bd2cb06e3..dcc5b70d6 100644 --- a/magicblock-chainlink/src/testing/rpc_client_mock.rs +++ b/magicblock-chainlink/src/testing/rpc_client_mock.rs @@ -20,7 +20,8 @@ use solana_pubkey::Pubkey; use solana_rpc_client_api::client_error; #[cfg(any(test, feature = "dev-context"))] use solana_rpc_client_api::{ - config::RpcAccountInfoConfig, + config::{RpcAccountInfoConfig, RpcProgramAccountsConfig}, + filter::RpcFilterType, response::{Response, RpcResponseContext, RpcResult}, }; #[cfg(any(test, feature = "dev-context"))] @@ -124,6 +125,7 @@ impl ChainRpcClientMockBuilder { current_slot: Arc::new(AtomicU64::new(self.current_slot)), single_account_fetches: Arc::::default(), multi_account_fetches: Arc::::default(), + program_account_fetches: Arc::::default(), block_fetches: Arc::new(AtomicBool::new(false)), fetch_block_notify: Arc::new(Notify::new()), multi_account_response_truncate: self @@ -151,6 +153,7 @@ pub struct ChainRpcClientMock { current_slot: Arc, single_account_fetches: Arc, multi_account_fetches: Arc, + program_account_fetches: Arc, block_fetches: Arc, fetch_block_notify: Arc, multi_account_response_truncate: Option, @@ -165,6 +168,7 @@ impl ChainRpcClientMock { current_slot: Arc::::default(), single_account_fetches: Arc::::default(), multi_account_fetches: Arc::::default(), + program_account_fetches: Arc::::default(), block_fetches: Arc::new(AtomicBool::new(false)), fetch_block_notify: Arc::new(Notify::new()), multi_account_response_truncate: None, @@ -279,6 +283,10 @@ impl ChainRpcClientMock { self.multi_account_fetches.load(Ordering::Relaxed) } + pub fn program_account_fetches(&self) -> u64 { + self.program_account_fetches.load(Ordering::Relaxed) + } + /// Account lookup shared by both fetch methods of the /// [ChainRpcClient] impl; does not touch the per-method call /// counters. @@ -317,6 +325,31 @@ impl ChainRpcClientMock { notified.await; } } + + fn matches_program_account_filters( + account: &Account, + config: &RpcProgramAccountsConfig, + ) -> bool { + let Some(filters) = config.filters.as_ref() else { + return true; + }; + + filters.iter().all(|filter| match filter { + RpcFilterType::DataSize(size) => account.data.len() as u64 == *size, + RpcFilterType::Memcmp(memcmp) => { + let offset = memcmp.offset(); + let Some(bytes) = memcmp.bytes() else { + return false; + }; + let end = offset.saturating_add(bytes.len()); + account + .data + .get(offset..end) + .is_some_and(|data| data == bytes.as_slice()) + } + RpcFilterType::TokenAccountState => true, + }) + } } #[cfg(any(test, feature = "dev-context"))] @@ -402,4 +435,30 @@ impl ChainRpcClient for ChainRpcClientMock { }; Ok(res) } + + async fn get_program_accounts_with_config( + &self, + pubkey: &Pubkey, + config: RpcProgramAccountsConfig, + ) -> client_error::Result> { + self.program_account_fetches.fetch_add(1, Ordering::Relaxed); + self.wait_if_fetches_blocked().await; + + let mut accounts = vec![]; + for (account_pubkey, AccountAtSlot { account, slot }) in + self.accounts.lock().unwrap().iter() + { + if account.owner != *pubkey { + continue; + } + if *slot < self.current_slot.load(Ordering::Relaxed) { + continue; + } + if !Self::matches_program_account_filters(account, &config) { + continue; + } + accounts.push((*account_pubkey, account.clone())); + } + Ok(accounts) + } } diff --git a/magicblock-config/src/config/chain.rs b/magicblock-config/src/config/chain.rs index ff9aa912e..3d0a624b1 100644 --- a/magicblock-config/src/config/chain.rs +++ b/magicblock-config/src/config/chain.rs @@ -66,6 +66,11 @@ pub struct ChainLinkConfig { #[serde(with = "humantime")] pub resubscription_delay: Duration, + /// Period for polling DLP-owned UndelegationRequest accounts. + /// Set to 0s to disable the polling backfill loop. + #[serde(with = "humantime")] + pub undelegation_request_poll_interval: Duration, + /// AML/Risk checks for post-delegation actions via Range API. pub risk: RiskConfig, } @@ -79,6 +84,9 @@ impl Default for ChainLinkConfig { resubscription_delay: Duration::from_millis( consts::DEFAULT_RESUBSCRIPTION_DELAY_MS, ), + undelegation_request_poll_interval: Duration::from_secs( + consts::DEFAULT_UNDELEGATION_REQUEST_POLL_INTERVAL_SECS, + ), risk: RiskConfig::default(), } } diff --git a/magicblock-config/src/consts.rs b/magicblock-config/src/consts.rs index c79d7fb05..9b532ba31 100644 --- a/magicblock-config/src/consts.rs +++ b/magicblock-config/src/consts.rs @@ -79,6 +79,9 @@ pub const DEFAULT_TASK_SCHEDULER_FAILED_TASK_CLEANUP_INTERVAL_SECS: u64 = /// Default delay in milliseconds between resubscribing to accounts after a pubsub reconnection pub const DEFAULT_RESUBSCRIPTION_DELAY_MS: u64 = 50; +/// Default period in seconds for scanning DLP undelegation request accounts. +pub const DEFAULT_UNDELEGATION_REQUEST_POLL_INTERVAL_SECS: u64 = 5 * 60; + /// Default capacity for the LRU cache of subscribed accounts pub const DEFAULT_MAX_MONITORED_ACCOUNTS: usize = 5_000; diff --git a/magicblock-config/src/tests.rs b/magicblock-config/src/tests.rs index 1268d9bc8..169f367fd 100644 --- a/magicblock-config/src/tests.rs +++ b/magicblock-config/src/tests.rs @@ -78,6 +78,12 @@ fn test_defaults_are_sane() { assert_eq!(config.grpc.max_old_unoptimized, 5); assert_eq!(config.grpc.max_subs_in_new, 400); assert_eq!(config.grpc.max_time_without_optimization_secs, 60); + assert_eq!( + config.chainlink.undelegation_request_poll_interval, + Duration::from_secs( + consts::DEFAULT_UNDELEGATION_REQUEST_POLL_INTERVAL_SECS + ) + ); } #[test] @@ -294,6 +300,7 @@ fn test_chainlink_config() { [chainlink] max-monitored-accounts = 5000 resubscription-delay = "50ms" + undelegation-request-poll-interval = "5m" [chainlink.risk] enabled = true @@ -311,6 +318,10 @@ fn test_chainlink_config() { config.chainlink.resubscription_delay, std::time::Duration::from_millis(50) ); + assert_eq!( + config.chainlink.undelegation_request_poll_interval, + std::time::Duration::from_secs(5 * 60) + ); assert!(config.chainlink.risk.enabled); assert_eq!( config.chainlink.risk.api_key, @@ -592,6 +603,10 @@ fn test_env_vars_full_coverage() { // --- Chainlink --- EnvVarGuard::new("MBV_CHAINLINK__MAX_MONITORED_ACCOUNTS", "123"), EnvVarGuard::new("MBV_CHAINLINK__RESUBSCRIPTION_DELAY", "150ms"), + EnvVarGuard::new( + "MBV_CHAINLINK__UNDELEGATION_REQUEST_POLL_INTERVAL", + "2m", + ), EnvVarGuard::new("MBV_CHAINLINK__RISK__ENABLED", "true"), EnvVarGuard::new("MBV_CHAINLINK__RISK__API_KEY", "env-range-token"), EnvVarGuard::new("MBV_CHAINLINK__RISK__CACHE_TTL", "45m"), @@ -666,6 +681,10 @@ fn test_env_vars_full_coverage() { config.chainlink.resubscription_delay, Duration::from_millis(150) ); + assert_eq!( + config.chainlink.undelegation_request_poll_interval, + Duration::from_secs(2 * 60) + ); assert!(config.chainlink.risk.enabled); assert_eq!( config.chainlink.risk.api_key, diff --git a/magicblock-services/src/undelegation_request_service.rs b/magicblock-services/src/undelegation_request_service.rs index 817acb4f9..228863fee 100644 --- a/magicblock-services/src/undelegation_request_service.rs +++ b/magicblock-services/src/undelegation_request_service.rs @@ -13,7 +13,7 @@ use solana_keypair::Keypair; use solana_signer::Signer; use solana_transaction::Transaction; use solana_transaction_error::TransactionError; -use tokio::sync::broadcast; +use tokio::{sync::broadcast, time::MissedTickBehavior}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; @@ -52,6 +52,7 @@ pub struct UndelegationRequestService { internal_transaction_scheduler: TransactionSchedulerHandle, validator_authority: Arc, latest_block_reader: Arc, + undelegation_request_poll_interval: Duration, } trait LatestBlockReader: Send + Sync { @@ -73,6 +74,7 @@ impl UndelegationRequestService { internal_transaction_scheduler: TransactionSchedulerHandle, validator_authority: Keypair, latest_block: impl LatestBlockProvider, + undelegation_request_poll_interval: Duration, ) -> Self { Self { chainlink, @@ -80,6 +82,7 @@ impl UndelegationRequestService { internal_transaction_scheduler, validator_authority: Arc::new(validator_authority), latest_block_reader: Arc::new(latest_block.clone()), + undelegation_request_poll_interval, } } @@ -128,6 +131,58 @@ impl UndelegationRequestService { } } + async fn undelegation_request_poll_processor( + poll_interval: Duration, + cancellation_token: CancellationToken, + chainlink: Arc, + internal_transaction_scheduler: TransactionSchedulerHandle, + validator_authority: Arc, + latest_block: Arc, + ) { + if poll_interval.is_zero() { + debug!( + "DLP undelegation request polling is disabled by configuration" + ); + return; + } + + let mut interval = tokio::time::interval(poll_interval); + interval.set_missed_tick_behavior(MissedTickBehavior::Delay); + + loop { + tokio::select! { + biased; + _ = cancellation_token.cancelled() => { + info!("Shutting down undelegation request poll processor"); + return; + } + _ = interval.tick() => {} + } + + let requests = match chainlink.fetch_undelegation_requests().await { + Ok(requests) => requests, + Err(err) => { + error!( + error = ?err, + "Failed to scan DLP undelegation requests" + ); + continue; + } + }; + for request in requests { + Self::process_observed_undelegation_request_with_retries( + request, + &chainlink, + &internal_transaction_scheduler, + validator_authority.as_ref(), + latest_block.as_ref(), + &cancellation_token, + ) + .await; + } + } + } + async fn process_observed_undelegation_request_with_retries( request: ObservedUndelegationRequest, chainlink: &ChainlinkImpl, @@ -290,6 +345,12 @@ impl UndelegationRequestService { pub fn start(self: &Arc) { let Some(requests) = self.chainlink.subscribe_undelegation_requests() else { + if !self.undelegation_request_poll_interval.is_zero() { + warn!( + "Cannot subscribe to DLP undelegation requests; falling back to polling only" + ); + } + self.spawn_undelegation_request_poll_processor(); return; }; tokio::spawn(Self::undelegation_request_processor( @@ -300,6 +361,22 @@ impl UndelegationRequestService { self.validator_authority.clone(), self.latest_block_reader.clone(), )); + self.spawn_undelegation_request_poll_processor(); + } + + fn spawn_undelegation_request_poll_processor(self: &Arc) { + if self.undelegation_request_poll_interval.is_zero() { + debug!("Skipping DLP undelegation request poll processor"); + return; + } + tokio::spawn(Self::undelegation_request_poll_processor( + self.undelegation_request_poll_interval, + self.cancellation_token.clone(), + self.chainlink.clone(), + self.internal_transaction_scheduler.clone(), + self.validator_authority.clone(), + self.latest_block_reader.clone(), + )); } pub fn stop(&self) { From 81bd61bef10edaee840409d2a5c56322022c6261 Mon Sep 17 00:00:00 2001 From: Sarfaraz Nawaz Date: Tue, 14 Jul 2026 15:45:44 +0530 Subject: [PATCH 2/2] address Gabriele's feedback --- .agents/context/crates/magicblock-services.md | 9 ++++++--- .../src/undelegation_request_service.rs | 19 +++++++++++++++++++ 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/.agents/context/crates/magicblock-services.md b/.agents/context/crates/magicblock-services.md index 1039f0f9f..a56a399f0 100644 --- a/.agents/context/crates/magicblock-services.md +++ b/.agents/context/crates/magicblock-services.md @@ -118,9 +118,10 @@ service.stop() For each observed request, the service: 1. verifies the request PDA matches the delegated account; -2. checks the delegated account is delegated on base and ER; -3. best-effort notifies Chainlink with `undelegation_requested`; -4. submits a local validator-signed `ScheduleCommitAndUndelegate` transaction with the validator as payer/signer, `MAGIC_CONTEXT_PUBKEY`, and each delegated account as writable non-signer. +2. asks Chainlink to materialize the delegated account if it is missing locally; +3. checks the delegated account is delegated on base and ER; +4. best-effort notifies Chainlink with `undelegation_requested`; +5. submits a local validator-signed `ScheduleCommitAndUndelegate` transaction with the validator as payer/signer, `MAGIC_CONTEXT_PUBKEY`, and each delegated account as writable non-signer. Transient delegation-check and local-scheduling failures are retried three times with short exponential backoff. Invalid request PDAs and non-delegated accounts are skipped. @@ -216,6 +217,8 @@ The undelegation request service is a trigger: it only schedules the local Magic The service validates the request PDA from delegated account before scheduling. It must not trust request-local owner, rent payer, or commit nonce fields; current request data carries the delegated account and expiry slot only. +Before checking base/ER delegation readiness, the service calls Chainlink `ensure_accounts` for the delegated account. This lets polling recover valid requests for accounts that are delegated on base but not yet present in the ER bank, instead of skipping the request until unrelated traffic clones the account. + The service logs expired requests but still schedules normal undelegation when the delegated account is still valid. This preserves the best chance of committing ER state and clearing lifecycle state instead of leaving timeout/rollback handling to a stale request. ## Important invariants diff --git a/magicblock-services/src/undelegation_request_service.rs b/magicblock-services/src/undelegation_request_service.rs index 228863fee..0742cfae3 100644 --- a/magicblock-services/src/undelegation_request_service.rs +++ b/magicblock-services/src/undelegation_request_service.rs @@ -278,6 +278,25 @@ impl UndelegationRequestService { return Ok(()); } + if let Err(err) = chainlink + .ensure_accounts( + &[request.delegated_account], + None, + AccountFetchOrigin::GetAccount, + ) + .await + { + error!( + request_pda = %request.request_pda, + delegated_account = %request.delegated_account, + error = ?err, + "Failed to materialize requested undelegation account" + ); + return Err(ObservedUndelegationRequestError::Transient( + "failed to materialize requested undelegation account", + )); + } + let delegated = match chainlink .accounts_delegated_on_base_and_er( &[request.delegated_account],