canton_admin/
topology.rs

1//! Canton admin-API client: topology **read**.
2//!
3//! Wraps `TopologyManagerReadService` (served on the Canton admin API port,
4//! e.g. `:3902`) for the M1 read surface: party→participant mappings, namespace
5//! delegations, and vetted packages.
6
7use std::sync::Arc;
8
9use canton_core::auth::{self, Intercepted};
10use canton_core::telemetry::{self, TRANSPORT_GRPC};
11use canton_core::{Config, Result};
12use canton_proto::com::digitalasset::canton::protocol::v30 as protocol;
13use canton_proto::com::digitalasset::canton::topology::admin::v30 as topo;
14use tonic::transport::Channel;
15use topo::topology_manager_read_service_client::TopologyManagerReadServiceClient;
16
17/// Which topology store to query.
18///
19/// `#[non_exhaustive]` so new store kinds can be added without a breaking
20/// change.
21#[derive(Clone, Debug, PartialEq, Eq)]
22#[non_exhaustive]
23pub enum Store {
24    /// The node's authorized store (its own local topology state).
25    Authorized,
26    /// A synchronizer store, addressed by its physical synchronizer id.
27    Synchronizer(String),
28    /// A temporary store, by name.
29    Temporary(String),
30}
31
32impl Store {
33    fn to_store_id(&self) -> topo::StoreId {
34        use topo::store_id::Store as S;
35        let store = match self {
36            Store::Authorized => S::Authorized(topo::store_id::Authorized {}),
37            Store::Synchronizer(physical_id) => S::Synchronizer(topo::Synchronizer {
38                kind: Some(topo::synchronizer::Kind::PhysicalId(physical_id.clone())),
39            }),
40            Store::Temporary(name) => {
41                S::Temporary(topo::store_id::Temporary { name: name.clone() })
42            }
43        };
44        topo::StoreId { store: Some(store) }
45    }
46}
47
48/// One topology mapping result: the mapping `item` plus its
49/// [`BaseResult`](topo::BaseResult) `context` (store, validity window, serial,
50/// …). `#[non_exhaustive]` so fields can be added without a breaking change.
51#[derive(Clone, Debug, PartialEq)]
52#[non_exhaustive]
53pub struct Entry<T> {
54    /// Topology metadata for this mapping (store, validity window, serial, …).
55    pub context: topo::BaseResult,
56    /// The mapping item.
57    pub item: T,
58}
59
60/// A client for Canton's `TopologyManagerReadService` (admin API).
61///
62/// On LocalNet the admin API is unauthenticated; in hardened deployments set a
63/// bearer token via [`Config::with_oidc`](canton_core::Config::with_oidc).
64#[derive(Clone, Debug)]
65pub struct TopologyClient {
66    channel: Channel,
67    config: Arc<Config>,
68}
69
70impl TopologyClient {
71    /// Build a lazily-connected client for the admin API endpoint (e.g.
72    /// `http://localhost:3902`).
73    ///
74    /// # Errors
75    /// Returns [`Error::InvalidRequest`](canton_core::Error::InvalidRequest) if
76    /// the endpoint is not a valid URI.
77    pub fn connect_lazy(config: Config) -> Result<Self> {
78        Ok(Self {
79            channel: config.connect_channel()?,
80            config: Arc::new(config),
81        })
82    }
83
84    async fn intercepted(&self) -> Result<Intercepted> {
85        auth::intercepted(&self.channel, self.config.auth()).await
86    }
87
88    async fn with_retry<T, F, Fut>(&self, op: F) -> Result<T>
89    where
90        F: FnMut() -> Fut,
91        Fut: std::future::Future<Output = Result<T>>,
92    {
93        canton_core::retry::run_with_retry(self.config.retry(), op).await
94    }
95
96    /// A `BaseQuery` for the current (head) state of `store`, no proposals.
97    fn base_query(store: &Store) -> topo::BaseQuery {
98        topo::BaseQuery {
99            store: Some(store.to_store_id()),
100            proposals: false,
101            operation: 0, // TOPOLOGY_CHANGE_OP_UNSPECIFIED
102            time_query: Some(topo::base_query::TimeQuery::HeadState(())),
103            filter_signed_key: String::new(),
104            protocol_version: None,
105        }
106    }
107
108    /// Read party→participant mappings (`ListPartyToParticipant`). Empty filters
109    /// match all.
110    ///
111    /// # Errors
112    /// Returns an [`Error`](canton_core::Error) if the RPC fails.
113    pub async fn list_party_to_participant(
114        &self,
115        store: Store,
116        filter_party: &str,
117        filter_participant: &str,
118    ) -> Result<Vec<Entry<protocol::PartyToParticipant>>> {
119        let filter_party = filter_party.to_string();
120        let filter_participant = filter_participant.to_string();
121        telemetry::instrument("list_party_to_participant", TRANSPORT_GRPC, async move {
122            self.with_retry(|| {
123                let (store, filter_party, filter_participant) = (
124                    store.clone(),
125                    filter_party.clone(),
126                    filter_participant.clone(),
127                );
128                async move {
129                    let mut client =
130                        TopologyManagerReadServiceClient::new(self.intercepted().await?);
131                    let response = client
132                        .list_party_to_participant(topo::ListPartyToParticipantRequest {
133                            base_query: Some(Self::base_query(&store)),
134                            filter_party,
135                            filter_participant,
136                        })
137                        .await?
138                        .into_inner();
139                    Ok(collect_entries(
140                        response.results.into_iter().map(|r| (r.context, r.item)),
141                    ))
142                }
143            })
144            .await
145        })
146        .await
147    }
148
149    /// Read namespace delegations (`ListNamespaceDelegation`). Empty filters
150    /// match all.
151    ///
152    /// # Errors
153    /// Returns an [`Error`](canton_core::Error) if the RPC fails.
154    pub async fn list_namespace_delegations(
155        &self,
156        store: Store,
157        filter_namespace: &str,
158        filter_target_key_fingerprint: &str,
159    ) -> Result<Vec<Entry<protocol::NamespaceDelegation>>> {
160        let filter_namespace = filter_namespace.to_string();
161        let filter_target_key_fingerprint = filter_target_key_fingerprint.to_string();
162        telemetry::instrument("list_namespace_delegations", TRANSPORT_GRPC, async move {
163            self.with_retry(|| {
164                let (store, filter_namespace, filter_target_key_fingerprint) = (
165                    store.clone(),
166                    filter_namespace.clone(),
167                    filter_target_key_fingerprint.clone(),
168                );
169                async move {
170                    let mut client =
171                        TopologyManagerReadServiceClient::new(self.intercepted().await?);
172                    let response = client
173                        .list_namespace_delegation(topo::ListNamespaceDelegationRequest {
174                            base_query: Some(Self::base_query(&store)),
175                            filter_namespace,
176                            filter_target_key_fingerprint,
177                        })
178                        .await?
179                        .into_inner();
180                    Ok(collect_entries(
181                        response.results.into_iter().map(|r| (r.context, r.item)),
182                    ))
183                }
184            })
185            .await
186        })
187        .await
188    }
189
190    /// Read vetted packages (`ListVettedPackages`). Empty participant filter
191    /// matches all.
192    ///
193    /// Vetted packages are stored per-synchronizer: query with
194    /// [`Store::Synchronizer`] (the synchronizer's physical id) rather than
195    /// [`Store::Authorized`], or the result is empty.
196    ///
197    /// # Errors
198    /// Returns an [`Error`](canton_core::Error) if the RPC fails.
199    pub async fn list_vetted_packages(
200        &self,
201        store: Store,
202        filter_participant: &str,
203    ) -> Result<Vec<Entry<protocol::VettedPackages>>> {
204        let filter_participant = filter_participant.to_string();
205        telemetry::instrument("list_vetted_packages", TRANSPORT_GRPC, async move {
206            self.with_retry(|| {
207                let (store, filter_participant) = (store.clone(), filter_participant.clone());
208                async move {
209                    let mut client =
210                        TopologyManagerReadServiceClient::new(self.intercepted().await?);
211                    let response = client
212                        .list_vetted_packages(topo::ListVettedPackagesRequest {
213                            base_query: Some(Self::base_query(&store)),
214                            filter_participant,
215                        })
216                        .await?
217                        .into_inner();
218                    Ok(collect_entries(
219                        response.results.into_iter().map(|r| (r.context, r.item)),
220                    ))
221                }
222            })
223            .await
224        })
225        .await
226    }
227}
228
229/// Keep only results that carry both a context and an item, pairing them.
230fn collect_entries<T>(
231    results: impl Iterator<Item = (Option<topo::BaseResult>, Option<T>)>,
232) -> Vec<Entry<T>> {
233    results
234        .filter_map(|(context, item)| {
235            Some(Entry {
236                context: context?,
237                item: item?,
238            })
239        })
240        .collect()
241}
242
243#[cfg(test)]
244mod tests {
245    use super::*;
246
247    #[test]
248    fn store_maps_to_the_right_store_id_variant() {
249        use topo::store_id::Store as S;
250
251        assert!(matches!(
252            Store::Authorized.to_store_id().store,
253            Some(S::Authorized(_))
254        ));
255
256        match Store::Synchronizer("pid-1".to_string()).to_store_id().store {
257            Some(S::Synchronizer(s)) => assert_eq!(
258                s.kind,
259                Some(topo::synchronizer::Kind::PhysicalId("pid-1".to_string()))
260            ),
261            other => panic!("expected a synchronizer store, got {other:?}"),
262        }
263
264        match Store::Temporary("scratch".to_string()).to_store_id().store {
265            Some(S::Temporary(t)) => assert_eq!(t.name, "scratch"),
266            other => panic!("expected a temporary store, got {other:?}"),
267        }
268    }
269
270    #[test]
271    fn base_query_asks_for_head_state_without_proposals() {
272        let query = TopologyClient::base_query(&Store::Authorized);
273        assert!(matches!(
274            query.time_query,
275            Some(topo::base_query::TimeQuery::HeadState(()))
276        ));
277        assert!(!query.proposals, "reads default to committed state");
278        assert!(query.store.is_some());
279    }
280
281    #[test]
282    fn collect_entries_drops_partial_results() {
283        let ctx = topo::BaseResult::default();
284        let rows = vec![
285            (Some(ctx.clone()), Some(7u8)),
286            (None, Some(8u8)),         // missing context -> dropped
287            (Some(ctx.clone()), None), // missing item -> dropped
288        ];
289        let kept = collect_entries(rows.into_iter());
290        assert_eq!(kept.len(), 1);
291        assert_eq!(kept[0].item, 7);
292    }
293}
canton_admin/topology.rs

canton_admin/
topology.rs
