daml_grpc/
ledger_client.rs

1use std::time::Duration;
2
3#[cfg(test)]
4use tonic::transport::Uri;
5use tonic::transport::{Certificate, Channel, ClientTlsConfig};
6use tracing::{debug, instrument};
7
8use crate::data::{DamlError, DamlResult};
9use crate::service::DamlTimeService;
10use crate::service::{
11    DamlCommandCompletionService, DamlCommandService, DamlCommandSubmissionService, DamlContractService,
12    DamlEventQueryService, DamlPackageService, DamlStateService, DamlUpdateService, DamlVersionService,
13};
14#[cfg(feature = "admin")]
15use crate::service::{
16    DamlCommandInspectionService, DamlIdentityProviderConfigService, DamlPackageManagementService,
17    DamlParticipantPruningService, DamlPartyManagementService, DamlUserManagementService,
18};
19
20const DEFAULT_TIMEOUT_SECS: u64 = 5;
21const DEFAULT_CONNECT_TIMEOUT_SECS: u64 = 5;
22
23/// Connection configuration for a [`DamlGrpcClient`].
24///
25/// v2 dropped v1's `ledger_id` discovery flow — every request runs
26/// against the connected participant directly, so there's no
27/// `LedgerIdentityService` round-trip at connect time and no
28/// reset-and-wait timeout needed for that flow.
29#[derive(Debug, Default)]
30pub struct DamlGrpcClientConfig {
31    uri: String,
32    timeout: Duration,
33    connect_timeout: Option<Duration>,
34    concurrency_limit: Option<usize>,
35    rate_limit: Option<(u64, Duration)>,
36    initial_stream_window_size: Option<u32>,
37    initial_connection_window_size: Option<u32>,
38    tcp_keepalive: Option<Duration>,
39    tcp_nodelay: bool,
40    tls_config: Option<DamlGrpcTlsConfig>,
41    auth_token: Option<String>,
42}
43
44#[derive(Debug)]
45pub struct DamlGrpcTlsConfig {
46    ca_cert: Option<Vec<u8>>,
47}
48
49/// Construct a [`DamlGrpcClient`].
50pub struct DamlGrpcClientBuilder {
51    config: DamlGrpcClientConfig,
52}
53
54impl DamlGrpcClientBuilder {
55    pub fn uri(uri: impl Into<String>) -> Self {
56        Self {
57            config: DamlGrpcClientConfig {
58                uri: uri.into(),
59                timeout: Duration::from_secs(DEFAULT_TIMEOUT_SECS),
60                connect_timeout: Some(Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS)),
61                ..DamlGrpcClientConfig::default()
62            },
63        }
64    }
65
66    /// The network timeout.
67    pub fn timeout(self, timeout: Duration) -> Self {
68        Self {
69            config: DamlGrpcClientConfig {
70                timeout,
71                ..self.config
72            },
73        }
74    }
75
76    /// The connection timeout.
77    pub fn connect_timeout(self, connect_timeout: Option<Duration>) -> Self {
78        Self {
79            config: DamlGrpcClientConfig {
80                connect_timeout,
81                ..self.config
82            },
83        }
84    }
85
86    pub fn concurrency_limit(self, concurrency_limit: usize) -> Self {
87        Self {
88            config: DamlGrpcClientConfig {
89                concurrency_limit: Some(concurrency_limit),
90                ..self.config
91            },
92        }
93    }
94
95    pub fn rate_limit(self, rate_limit: (u64, Duration)) -> Self {
96        Self {
97            config: DamlGrpcClientConfig {
98                rate_limit: Some(rate_limit),
99                ..self.config
100            },
101        }
102    }
103
104    pub fn initial_stream_window_size(self, initial_stream_window_size: u32) -> Self {
105        Self {
106            config: DamlGrpcClientConfig {
107                initial_stream_window_size: Some(initial_stream_window_size),
108                ..self.config
109            },
110        }
111    }
112
113    pub fn initial_connection_window_size(self, initial_connection_window_size: u32) -> Self {
114        Self {
115            config: DamlGrpcClientConfig {
116                initial_connection_window_size: Some(initial_connection_window_size),
117                ..self.config
118            },
119        }
120    }
121
122    pub fn tcp_keepalive(self, tcp_keepalive: Duration) -> Self {
123        Self {
124            config: DamlGrpcClientConfig {
125                tcp_keepalive: Some(tcp_keepalive),
126                ..self.config
127            },
128        }
129    }
130
131    pub fn tcp_nodelay(self, tcp_nodelay: bool) -> Self {
132        Self {
133            config: DamlGrpcClientConfig {
134                tcp_nodelay,
135                ..self.config
136            },
137        }
138    }
139
140    /// Enable TLS, pinning the participant's CA certificate. Pass the
141    /// PEM-encoded CA bytes. For system-root verification instead, use
142    /// [`with_tls_system_roots`](Self::with_tls_system_roots).
143    pub fn with_tls(self, ca_cert: impl Into<Vec<u8>>) -> Self {
144        Self {
145            config: DamlGrpcClientConfig {
146                tls_config: Some(DamlGrpcTlsConfig {
147                    ca_cert: Some(ca_cert.into()),
148                }),
149                ..self.config
150            },
151        }
152    }
153
154    /// Enable TLS, verifying the participant against the platform's
155    /// system root store. For a pinned CA certificate, use
156    /// [`with_tls`](Self::with_tls) instead.
157    pub fn with_tls_system_roots(self) -> Self {
158        Self {
159            config: DamlGrpcClientConfig {
160                tls_config: Some(DamlGrpcTlsConfig {
161                    ca_cert: None,
162                }),
163                ..self.config
164            },
165        }
166    }
167
168    pub fn with_auth(self, auth_token: String) -> Self {
169        Self {
170            config: DamlGrpcClientConfig {
171                auth_token: Some(auth_token),
172                ..self.config
173            },
174        }
175    }
176
177    pub async fn connect(self) -> DamlResult<DamlGrpcClient> {
178        DamlGrpcClient::connect(self.config).await
179    }
180}
181
182/// Daml v2 ledger client connection. A thin handle around a tonic
183/// [`Channel`] that hands out per-service clients on demand. Cheap to
184/// hold — service factories clone the channel rather than opening
185/// new ones.
186#[derive(Debug)]
187pub struct DamlGrpcClient {
188    config: DamlGrpcClientConfig,
189    channel: Channel,
190}
191
192impl DamlGrpcClient {
193    /// Open a channel and connect.
194    #[instrument]
195    pub async fn connect(config: DamlGrpcClientConfig) -> DamlResult<Self> {
196        debug!("connecting to {}", config.uri);
197        let channel = Self::make_channel(&config).await?;
198        Ok(Self {
199            config,
200            channel,
201        })
202    }
203
204    pub const fn config(&self) -> &DamlGrpcClientConfig {
205        &self.config
206    }
207
208    /// Retrieve a [`DamlPackageService`] for querying the Daml-LF
209    /// packages supported by the participant.
210    pub fn package_service(&self) -> DamlPackageService<'_> {
211        DamlPackageService::new(self.channel.clone(), self.config.auth_token.as_deref())
212    }
213
214    /// Retrieve a [`DamlCommandSubmissionService`] for fire-and-forget
215    /// command submissions. Completion is observed separately through
216    /// [`command_completion_service`](Self::command_completion_service).
217    pub fn command_submission_service(&self) -> DamlCommandSubmissionService<'_> {
218        DamlCommandSubmissionService::new(self.channel.clone(), self.config.auth_token.as_deref())
219    }
220
221    /// Retrieve a [`DamlCommandCompletionService`] for observing the
222    /// asynchronous outcome (success or rejection) of command
223    /// submissions, plus periodic `OffsetCheckpoint` markers.
224    pub fn command_completion_service(&self) -> DamlCommandCompletionService<'_> {
225        DamlCommandCompletionService::new(self.channel.clone(), self.config.auth_token.as_deref())
226    }
227
228    /// Retrieve a [`DamlUpdateService`] for reading the participant's
229    /// update stream — transactions, reassignments, and topology
230    /// transactions, paginated or open-ended. v2's replacement for
231    /// v1's `TransactionService`.
232    pub fn update_service(&self) -> DamlUpdateService<'_> {
233        DamlUpdateService::new(self.channel.clone(), self.config.auth_token.as_deref())
234    }
235
236    /// Retrieve a [`DamlStateService`] for snapshotting the active
237    /// contract set, listing connected synchronizers, reading the
238    /// ledger end, and querying pruning watermarks. v2's
239    /// replacement for v1's `ActiveContractsService`.
240    pub fn state_service(&self) -> DamlStateService<'_> {
241        DamlStateService::new(self.channel.clone(), self.config.auth_token.as_deref())
242    }
243
244    /// Retrieve a [`DamlEventQueryService`] for per-contract event
245    /// lookup (create + consuming-archive halves) by contract id.
246    pub fn event_query_service(&self) -> DamlEventQueryService<'_> {
247        DamlEventQueryService::new(self.channel.clone(), self.config.auth_token.as_deref())
248    }
249
250    /// Retrieve a [`DamlContractService`] for contract-payload
251    /// lookup by id. Experimental / alpha per the proto; prefer
252    /// [`event_query_service`](Self::event_query_service) or
253    /// [`state_service`](Self::state_service) for stable surfaces.
254    pub fn contract_service(&self) -> DamlContractService<'_> {
255        DamlContractService::new(self.channel.clone(), self.config.auth_token.as_deref())
256    }
257
258    /// Retrieve a [`DamlCommandService`] for synchronous command
259    /// submission: submit and wait for the participant's verdict in
260    /// a single RPC.
261    pub fn command_service(&self) -> DamlCommandService<'_> {
262        DamlCommandService::new(self.channel.clone(), self.config.auth_token.as_deref())
263    }
264
265    /// Retrieve a [`DamlVersionService`] for querying the participant's
266    /// Ledger API version.
267    pub fn version_service(&self) -> DamlVersionService<'_> {
268        DamlVersionService::new(self.channel.clone(), self.config.auth_token.as_deref())
269    }
270
271    /// Retrieve a [`DamlPackageManagementService`] for inspecting
272    /// known packages, uploading DARs, validating DARs, and adjusting
273    /// the participant's package-vetting topology.
274    #[cfg(feature = "admin")]
275    pub fn package_management_service(&self) -> DamlPackageManagementService<'_> {
276        DamlPackageManagementService::new(self.channel.clone(), self.config.auth_token.as_deref())
277    }
278
279    /// Retrieve a [`DamlPartyManagementService`] for inspecting and
280    /// administering participant-local party state.
281    #[cfg(feature = "admin")]
282    pub fn party_management_service(&self) -> DamlPartyManagementService<'_> {
283        DamlPartyManagementService::new(self.channel.clone(), self.config.auth_token.as_deref())
284    }
285
286    /// Retrieve a [`DamlUserManagementService`] for managing
287    /// participant users and their rights.
288    #[cfg(feature = "admin")]
289    pub fn user_management_service(&self) -> DamlUserManagementService<'_> {
290        DamlUserManagementService::new(self.channel.clone(), self.config.auth_token.as_deref())
291    }
292
293    /// Retrieve a [`DamlIdentityProviderConfigService`] for managing
294    /// runtime-configured Identity Provider configurations.
295    #[cfg(feature = "admin")]
296    pub fn identity_provider_config_service(&self) -> DamlIdentityProviderConfigService<'_> {
297        DamlIdentityProviderConfigService::new(self.channel.clone(), self.config.auth_token.as_deref())
298    }
299
300    /// Retrieve a [`DamlCommandInspectionService`] for debugging
301    /// in-flight commands on the participant. Alpha; only available
302    /// when the participant advertises
303    /// `experimental.command_inspection_service.supported` in its
304    /// `VersionService.GetLedgerApiVersion` feature descriptor.
305    #[cfg(feature = "admin")]
306    pub fn command_inspection_service(&self) -> DamlCommandInspectionService<'_> {
307        DamlCommandInspectionService::new(self.channel.clone(), self.config.auth_token.as_deref())
308    }
309
310    /// Retrieve a [`DamlParticipantPruningService`] for truncating
311    /// older portions of the participant-local ledger view.
312    #[cfg(feature = "admin")]
313    pub fn participant_pruning_service(&self) -> DamlParticipantPruningService<'_> {
314        DamlParticipantPruningService::new(self.channel.clone(), self.config.auth_token.as_deref())
315    }
316
317    /// Retrieve a [`DamlTimeService`] for reading and advancing the
318    /// participant's static-time clock. Only meaningful when the
319    /// participant is configured for static time (see the
320    /// `experimental.static_time` flag in
321    /// `VersionService::GetLedgerApiVersion`).
322    ///
323    /// v2 dropped the `sandbox` feature gate — `TimeService` is part
324    /// of every Canton participant's `testing` API; static-vs-
325    /// wallclock is a server-side config switch, not a client
326    /// build flag.
327    pub fn time_service(&self) -> DamlTimeService<'_> {
328        DamlTimeService::new(self.channel.clone(), self.config.auth_token.as_deref())
329    }
330
331    async fn make_channel(config: &DamlGrpcClientConfig) -> DamlResult<Channel> {
332        let mut endpoint = Channel::from_shared(config.uri.clone())?;
333        if let Some(limit) = config.concurrency_limit {
334            endpoint = endpoint.concurrency_limit(limit);
335        }
336        if let Some((limit, duration)) = config.rate_limit {
337            endpoint = endpoint.rate_limit(limit, duration);
338        }
339        if let Some(size) = config.initial_stream_window_size {
340            endpoint = endpoint.initial_stream_window_size(size);
341        }
342        if let Some(size) = config.initial_connection_window_size {
343            endpoint = endpoint.initial_connection_window_size(size);
344        }
345        if let Some(duration) = config.tcp_keepalive {
346            endpoint = endpoint.tcp_keepalive(Some(duration));
347        }
348        endpoint = endpoint.tcp_nodelay(config.tcp_nodelay);
349        endpoint = endpoint.timeout(config.timeout);
350        if let Some(duration) = config.connect_timeout {
351            endpoint = endpoint.connect_timeout(duration);
352        }
353        match &config.tls_config {
354            Some(DamlGrpcTlsConfig {
355                ca_cert: Some(cert),
356            }) => {
357                endpoint = endpoint.tls_config(ClientTlsConfig::new().ca_certificate(Certificate::from_pem(cert)))?;
358            },
359            Some(DamlGrpcTlsConfig {
360                ca_cert: None,
361            }) => {
362                endpoint = endpoint.tls_config(ClientTlsConfig::new())?;
363            },
364            _ => {},
365        }
366
367        endpoint.connect().await.map_err(DamlError::from)
368    }
369
370    #[cfg(test)]
371    pub(crate) fn dummy_for_testing() -> Self {
372        DamlGrpcClient {
373            config: DamlGrpcClientConfig::default(),
374            channel: Channel::builder(Uri::from_static("http://dummy.for.testing")).connect_lazy(),
375        }
376    }
377}
daml_grpc/ledger_client.rs

daml_grpc/
ledger_client.rs
