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, Error, 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
70/// Build a gRPC service client on the authenticated channel, with this
71/// client's decode limit applied.
72///
73/// A macro rather than a function because `max_decoding_message_size` is an
74/// inherent method on each generated client — tonic exposes no trait for it —
75/// so there is nothing to be generic over. Keeping every construction site
76/// behind one expansion is the point: `tonic`'s 4 MiB default is small enough
77/// that a real ACS page trips it, and a new RPC added later would otherwise
78/// pick the default up silently.
79macro_rules! service {
80    ($self:ident, $ctor:expr) => {
81        $ctor($self.intercepted().await?)
82            .max_decoding_message_size($self.config.max_decoding_message_size())
83    };
84}
85
86impl TopologyClient {
87    /// Build a lazily-connected client for the admin API endpoint (e.g.
88    /// `http://localhost:3902`).
89    ///
90    /// # Errors
91    /// Returns [`Error::InvalidRequest`](canton_core::Error::InvalidRequest) if
92    /// the endpoint is not a valid URI.
93    pub fn connect_lazy(config: Config) -> Result<Self> {
94        Ok(Self {
95            channel: config.connect_channel()?,
96            config: Arc::new(config),
97        })
98    }
99
100    async fn intercepted(&self) -> Result<Intercepted> {
101        auth::intercepted(&self.channel, self.config.auth()).await
102    }
103
104    async fn with_retry<T, F, Fut>(&self, op: F) -> Result<T>
105    where
106        F: FnMut() -> Fut,
107        Fut: std::future::Future<Output = Result<T>>,
108    {
109        canton_core::retry::run_with_retry(self.config.retry(), op).await
110    }
111
112    /// A `BaseQuery` for the current (head) state of `store`, no proposals.
113    fn base_query(store: &Store) -> topo::BaseQuery {
114        topo::BaseQuery {
115            store: Some(store.to_store_id()),
116            proposals: false,
117            operation: 0, // TOPOLOGY_CHANGE_OP_UNSPECIFIED
118            time_query: Some(topo::base_query::TimeQuery::HeadState(())),
119            filter_signed_key: String::new(),
120            protocol_version: None,
121        }
122    }
123
124    /// Read party→participant mappings (`ListPartyToParticipant`). Empty filters
125    /// match all.
126    ///
127    /// # Errors
128    /// Returns an [`Error`](canton_core::Error) if the RPC fails.
129    pub async fn list_party_to_participant(
130        &self,
131        store: Store,
132        filter_party: &str,
133        filter_participant: &str,
134    ) -> Result<Vec<Entry<protocol::PartyToParticipant>>> {
135        let filter_party = filter_party.to_string();
136        let filter_participant = filter_participant.to_string();
137        telemetry::instrument("list_party_to_participant", TRANSPORT_GRPC, async move {
138            self.with_retry(|| {
139                let (store, filter_party, filter_participant) = (
140                    store.clone(),
141                    filter_party.clone(),
142                    filter_participant.clone(),
143                );
144                async move {
145                    let mut client = service!(self, TopologyManagerReadServiceClient::new);
146                    let response = client
147                        .list_party_to_participant(topo::ListPartyToParticipantRequest {
148                            base_query: Some(Self::base_query(&store)),
149                            filter_party,
150                            filter_participant,
151                        })
152                        .await?
153                        .into_inner();
154                    collect_entries(response.results.into_iter().map(|r| (r.context, r.item)))
155                }
156            })
157            .await
158        })
159        .await
160    }
161
162    /// Read namespace delegations (`ListNamespaceDelegation`). Empty filters
163    /// match all.
164    ///
165    /// # Errors
166    /// Returns an [`Error`](canton_core::Error) if the RPC fails.
167    pub async fn list_namespace_delegations(
168        &self,
169        store: Store,
170        filter_namespace: &str,
171        filter_target_key_fingerprint: &str,
172    ) -> Result<Vec<Entry<protocol::NamespaceDelegation>>> {
173        let filter_namespace = filter_namespace.to_string();
174        let filter_target_key_fingerprint = filter_target_key_fingerprint.to_string();
175        telemetry::instrument("list_namespace_delegations", TRANSPORT_GRPC, async move {
176            self.with_retry(|| {
177                let (store, filter_namespace, filter_target_key_fingerprint) = (
178                    store.clone(),
179                    filter_namespace.clone(),
180                    filter_target_key_fingerprint.clone(),
181                );
182                async move {
183                    let mut client = service!(self, TopologyManagerReadServiceClient::new);
184                    let response = client
185                        .list_namespace_delegation(topo::ListNamespaceDelegationRequest {
186                            base_query: Some(Self::base_query(&store)),
187                            filter_namespace,
188                            filter_target_key_fingerprint,
189                        })
190                        .await?
191                        .into_inner();
192                    collect_entries(response.results.into_iter().map(|r| (r.context, r.item)))
193                }
194            })
195            .await
196        })
197        .await
198    }
199
200    /// Read vetted packages (`ListVettedPackages`). Empty participant filter
201    /// matches all.
202    ///
203    /// Vetted packages are stored per-synchronizer: query with
204    /// [`Store::Synchronizer`] (the synchronizer's physical id) rather than
205    /// [`Store::Authorized`], or the result is empty.
206    ///
207    /// # Errors
208    /// Returns an [`Error`](canton_core::Error) if the RPC fails.
209    pub async fn list_vetted_packages(
210        &self,
211        store: Store,
212        filter_participant: &str,
213    ) -> Result<Vec<Entry<protocol::VettedPackages>>> {
214        let filter_participant = filter_participant.to_string();
215        telemetry::instrument("list_vetted_packages", TRANSPORT_GRPC, async move {
216            self.with_retry(|| {
217                let (store, filter_participant) = (store.clone(), filter_participant.clone());
218                async move {
219                    let mut client = service!(self, TopologyManagerReadServiceClient::new);
220                    let response = client
221                        .list_vetted_packages(topo::ListVettedPackagesRequest {
222                            base_query: Some(Self::base_query(&store)),
223                            filter_participant,
224                        })
225                        .await?
226                        .into_inner();
227                    collect_entries(response.results.into_iter().map(|r| (r.context, r.item)))
228                }
229            })
230            .await
231        })
232        .await
233    }
234}
235
236/// Keep only results that carry both a context and an item, pairing them.
237/// Assemble the entries of a topology response, refusing an incomplete one.
238///
239/// Both halves of an entry are `Required` in the proto, so a row missing
240/// either is a participant that answered something this client cannot read.
241/// Dropping such a row silently is the dangerous shape: topology reads answer
242/// questions like "which participants host this party", and a short answer is
243/// indistinguishable from a true one — the caller acts on a view of the
244/// network that is missing a member it was never told about.
245fn collect_entries<T>(
246    results: impl Iterator<Item = (Option<topo::BaseResult>, Option<T>)>,
247) -> Result<Vec<Entry<T>>> {
248    results
249        .map(|(context, item)| match (context, item) {
250            (Some(context), Some(item)) => Ok(Entry { context, item }),
251            (None, _) => Err(Error::UnexpectedResponse(
252                "a topology result carried no context; the response is incomplete".to_string(),
253            )),
254            (_, None) => Err(Error::UnexpectedResponse(
255                "a topology result carried no mapping; the response is incomplete".to_string(),
256            )),
257        })
258        .collect()
259}
260
261#[cfg(test)]
262#[allow(clippy::expect_used)]
263mod tests {
264    use super::*;
265
266    #[test]
267    fn store_maps_to_the_right_store_id_variant() {
268        use topo::store_id::Store as S;
269
270        assert!(matches!(
271            Store::Authorized.to_store_id().store,
272            Some(S::Authorized(_))
273        ));
274
275        match Store::Synchronizer("pid-1".to_string()).to_store_id().store {
276            Some(S::Synchronizer(s)) => assert_eq!(
277                s.kind,
278                Some(topo::synchronizer::Kind::PhysicalId("pid-1".to_string()))
279            ),
280            other => panic!("expected a synchronizer store, got {other:?}"),
281        }
282
283        match Store::Temporary("scratch".to_string()).to_store_id().store {
284            Some(S::Temporary(t)) => assert_eq!(t.name, "scratch"),
285            other => panic!("expected a temporary store, got {other:?}"),
286        }
287    }
288
289    #[test]
290    fn base_query_asks_for_head_state_without_proposals() {
291        let query = TopologyClient::base_query(&Store::Authorized);
292        assert!(matches!(
293            query.time_query,
294            Some(topo::base_query::TimeQuery::HeadState(()))
295        ));
296        assert!(!query.proposals, "reads default to committed state");
297        assert!(query.store.is_some());
298    }
299
300    #[test]
301    fn collect_entries_refuses_partial_results() {
302        let ctx = topo::BaseResult::default();
303
304        // A complete response reads normally.
305        let rows = vec![
306            (Some(ctx.clone()), Some(7u8)),
307            (Some(ctx.clone()), Some(8u8)),
308        ];
309        let kept = collect_entries(rows.into_iter()).expect("a complete response");
310        assert_eq!(kept.len(), 2);
311
312        // A row missing either half fails the whole read rather than shrinking
313        // it: a topology answer that is quietly short is worse than none.
314        let missing_context = vec![(Some(ctx.clone()), Some(7u8)), (None, Some(8u8))];
315        assert!(matches!(
316            collect_entries(missing_context.into_iter()),
317            Err(Error::UnexpectedResponse(_))
318        ));
319        let missing_item = vec![(Some(ctx.clone()), Some(7u8)), (Some(ctx), None::<u8>)];
320        assert!(matches!(
321            collect_entries(missing_item.into_iter()),
322            Err(Error::UnexpectedResponse(_))
323        ));
324    }
325}
canton_admin/topology.rs

canton_admin/
topology.rs
