admin: address sentinel PR review feedback

Switch RpcDzLedgerReader to use the nonblocking Solana RPC client so
tokio::try_join! actually runs fetches concurrently. Add solana tenant
filtering to the create command to match the find command.

Author: snormore

controlplane/doublezero-admin/src/cli/sentinel.rs

@@ -14,7 +14,9 @@ use doublezero_sentinel::{
 };
 use doublezero_serviceability::pda::get_tenant_pda;
 use serde::Serialize;
-use solana_client::rpc_client::RpcClient;
+use solana_client::{
+    nonblocking::rpc_client::RpcClient as NonblockingRpcClient, rpc_client::RpcClient,
+};
 use solana_sdk::{
     commitment_config::CommitmentConfig,
     instruction::Instruction,
@@ -132,7 +134,7 @@ impl FindValidatorMulticastPublishersCommand {
             api_url: self.validator_metadata_url.clone(),
         };
         let dz_ledger = RpcDzLedgerReader::new(
-            RpcClient::new_with_commitment(
+            NonblockingRpcClient::new_with_commitment(
                 dzclient.get_rpc().clone(),
                 CommitmentConfig::confirmed(),
             ),
@@ -428,7 +430,7 @@ impl CreateValidatorMulticastPublishersCommand {
             api_url: self.validator_metadata_url.clone(),
         };
         let dz_ledger = RpcDzLedgerReader::new(
-            RpcClient::new_with_commitment(
+            NonblockingRpcClient::new_with_commitment(
                 dzclient.get_rpc().clone(),
                 CommitmentConfig::confirmed(),
             ),
@@ -445,9 +447,13 @@ impl CreateValidatorMulticastPublishersCommand {
             multicast_group_pk, self.multicast_group
         );
 
+        // Derive the solana tenant PDA to scope user queries.
+        let (solana_tenant_pk, _) = get_tenant_pda(&program_id, "solana");
+        let default_tenant_pk = Pubkey::default();
+
         // Fetch users and validator data.
         eprintln!("Fetching DZ Ledger users and validator metadata...");
-        let (all_users, validators) = tokio::try_join!(
+        let (all_users_unfiltered, validators) = tokio::try_join!(
             async {
                 dz_ledger
                     .fetch_all_dz_users()
@@ -462,6 +468,12 @@ impl CreateValidatorMulticastPublishersCommand {
             },
         )?;
 
+        // Scope to solana tenant (or default/unset tenant).
+        let all_users: Vec<_> = all_users_unfiltered
+            .into_iter()
+            .filter(|u| u.tenant_pk == solana_tenant_pk || u.tenant_pk == default_tenant_pk)
+            .collect();
+
         let device_labels: HashMap<Pubkey, String> = codes
             .as_ref()
             .map(|c| c.device_codes.clone())

crates/sentinel/src/dz_ledger_reader.rs

@@ -6,6 +6,7 @@ use doublezero_sdk::{
 };
 use solana_account_decoder::UiAccountEncoding;
 use solana_client::{
+    nonblocking::rpc_client::RpcClient as NonblockingRpcClient,
     rpc_client::RpcClient,
     rpc_config::{RpcAccountInfoConfig, RpcProgramAccountsConfig},
     rpc_filter::{Memcmp, RpcFilterType},
@@ -52,12 +53,12 @@ pub trait DzLedgerReader: Send + Sync {
 // ---------------------------------------------------------------------------
 
 pub struct RpcDzLedgerReader {
-    client: RpcClient,
+    client: NonblockingRpcClient,
     program_id: Pubkey,
 }
 
 impl RpcDzLedgerReader {
-    pub fn new(client: RpcClient, program_id: Pubkey) -> Self {
+    pub fn new(client: NonblockingRpcClient, program_id: Pubkey) -> Self {
         Self { client, program_id }
     }
 }
@@ -82,6 +83,7 @@ impl DzLedgerReader for RpcDzLedgerReader {
                     ..Default::default()
                 },
             )
+            .await
             .context("failed to fetch User accounts from DZ Ledger")?;
 
         let mut users = Vec::new();
@@ -126,6 +128,7 @@ impl DzLedgerReader for RpcDzLedgerReader {
                     ..Default::default()
                 },
             )
+            .await
             .context("failed to fetch MulticastGroup accounts from DZ Ledger")?;
 
         for (pk, account) in accounts {