canton_ledger/submission.rs
1//! A submission whose identity is known before it is sent.
2//!
3//! Submitting a command is not a request whose failure means "it did not
4//! happen". A dropped connection, a timeout, or a retry the participant
5//! de-duplicated all produce an error while the command may have committed
6//! perfectly well. The only way back to the outcome is the
7//! [`ChangeId`](crate::ChangeId), read from the completion stream — so the
8//! change ID has to exist *before* the submission, not be a value returned by
9//! the call that failed.
10
11use std::time::Duration;
12
13use canton_proto::com::daml::ledger::api::v2 as pb;
14
15use crate::client::CantonClient;
16use crate::command::ChangeId;
17use crate::request::TransactionShape;
18use canton_core::Result;
19
20/// A command with its identity fixed, ready to send and to recover.
21///
22/// Built by [`CantonClient::submission`]. The three send methods mirror the
23/// client's own; [`Submission::recover`] is the one that has no equivalent
24/// without a handle, because it needs the change ID the send did not return.
25#[derive(Clone, Debug)]
26pub struct Submission {
27 client: CantonClient,
28 change_id: ChangeId,
29 commands: pb::Commands,
30 shape: TransactionShape,
31}
32
33impl Submission {
34 pub(crate) fn new(
35 client: CantonClient,
36 change_id: ChangeId,
37 commands: pb::Commands,
38 shape: TransactionShape,
39 ) -> Self {
40 Self {
41 client,
42 change_id,
43 commands,
44 shape,
45 }
46 }
47
48 /// The command's complete identity — available before anything is sent.
49 #[must_use]
50 pub fn change_id(&self) -> &ChangeId {
51 &self.change_id
52 }
53
54 /// Submit without waiting (`CommandSubmissionService.Submit`).
55 ///
56 /// # Errors
57 /// Returns an [`Error`](canton_core::Error) if authentication or the RPC
58 /// fails. A failure here is *ambiguous*: use [`Self::recover`].
59 pub async fn submit(&self) -> Result<()> {
60 self.client.submit_commands(self.commands.clone()).await
61 }
62
63 /// Submit and wait for the completion (`CommandService.SubmitAndWait`).
64 ///
65 /// # Errors
66 /// Returns an [`Error`](canton_core::Error) if authentication fails or the
67 /// command is rejected. A transport failure is ambiguous: use
68 /// [`Self::recover`].
69 pub async fn submit_and_wait(&self) -> Result<pb::SubmitAndWaitResponse> {
70 self.client
71 .submit_and_wait_commands(self.commands.clone())
72 .await
73 }
74
75 /// Submit and wait for the committed transaction.
76 ///
77 /// # Errors
78 /// Returns an [`Error`](canton_core::Error) if authentication fails, the
79 /// command is rejected, or the response carries no transaction. A transport
80 /// failure is ambiguous: use [`Self::recover`].
81 pub async fn submit_and_wait_for_transaction(&self) -> Result<pb::Transaction> {
82 self.client
83 .submit_and_wait_for_transaction_commands(self.commands.clone(), self.shape)
84 .await
85 }
86
87 /// Read this command's outcome back from the completion stream, matching on
88 /// the whole change ID.
89 ///
90 /// `begin_offset` must be an offset from **before** the submission —
91 /// [`CantonClient::ledger_end`] taken beforehand is the usual source, since
92 /// a completion that has already gone past cannot be read again.
93 ///
94 /// # Errors
95 /// Returns [`Error::Timeout`](canton_core::Error::Timeout) if no completion
96 /// arrives within `timeout` (which, for a command that never reached the
97 /// participant, is the correct answer), or
98 /// [`Error::CommandRejected`](canton_core::Error::CommandRejected) if the
99 /// ledger rejected it.
100 pub async fn recover(&self, begin_offset: i64, timeout: Duration) -> Result<pb::Completion> {
101 self.client
102 .await_completion(&self.change_id, begin_offset, timeout)
103 .await
104 }
105}
106
107/// A JSON-transport command with its identity fixed, ready to send and to
108/// recover — the JSON lane's [`Submission`].
109///
110/// Built by [`JsonClient::submission`](crate::JsonClient::submission).
111#[derive(Clone, Debug)]
112pub struct JsonSubmission {
113 client: crate::JsonClient,
114 commands: crate::JsonCommands,
115 change_id: ChangeId,
116}
117
118impl JsonSubmission {
119 pub(crate) fn new(client: crate::JsonClient, commands: crate::JsonCommands) -> Self {
120 let change_id = commands.change_id();
121 Self {
122 client,
123 commands,
124 change_id,
125 }
126 }
127
128 /// The command's complete identity — available before anything is sent.
129 #[must_use]
130 pub fn change_id(&self) -> &ChangeId {
131 &self.change_id
132 }
133
134 /// Submit without waiting (`POST /v2/commands/async/submit`).
135 ///
136 /// # Errors
137 /// Returns an [`Error`](canton_core::Error) if authentication or the
138 /// request fails. A failure is *ambiguous*: use [`Self::recover`].
139 pub async fn submit(&self) -> Result<()> {
140 self.client.submit(&self.commands).await
141 }
142
143 /// Submit and wait for the completion.
144 ///
145 /// # Errors
146 /// Returns an [`Error`](canton_core::Error) if authentication fails or the
147 /// command is rejected.
148 pub async fn submit_and_wait(&self) -> Result<crate::JsonSubmitAndWaitResponse> {
149 self.client.submit_and_wait(&self.commands).await
150 }
151
152 /// Submit and wait for the committed transaction.
153 ///
154 /// # Errors
155 /// Returns an [`Error`](canton_core::Error) if authentication fails or the
156 /// command is rejected.
157 pub async fn submit_and_wait_for_transaction(&self) -> Result<crate::JsonSubmitResponse> {
158 self.client
159 .submit_and_wait_for_transaction(&self.commands)
160 .await
161 }
162
163 /// Read this command's outcome back from the completion stream, matching on
164 /// the whole change ID.
165 ///
166 /// The JSON transport carries completions over the WebSocket, so this needs
167 /// the `ws` feature. `begin_offset` must be from **before** the submission;
168 /// [`JsonClient::ledger_end`](crate::JsonClient::ledger_end) taken
169 /// beforehand is the usual source.
170 ///
171 /// # Errors
172 /// Returns [`Error::Timeout`](canton_core::Error::Timeout) if no completion
173 /// arrives within `timeout`, or
174 /// [`Error::CommandRejected`](canton_core::Error::CommandRejected) if the
175 /// ledger rejected the command.
176 #[cfg(feature = "ws")]
177 #[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
178 #[allow(clippy::large_futures)] // the WS handshake state is inherently large; awaited once.
179 pub async fn recover(&self, begin_offset: i64, timeout: Duration) -> Result<serde_json::Value> {
180 use canton_core::Error;
181 use tokio_stream::StreamExt as _;
182
183 // Boxed: the WS handshake state makes both this future and the timeout
184 // around it several tens of kilobytes, which is not something to leave
185 // on a caller's stack.
186 let scan = Box::pin(async {
187 let stream = self
188 .client
189 .ws_completions(self.change_id.act_as().to_vec(), begin_offset)
190 .await?;
191 tokio::pin!(stream);
192 while let Some(item) = stream.next().await {
193 let frame = item?;
194 let Some(completion) = crate::ws::completion_value(&frame) else {
195 continue;
196 };
197 if !self.change_id.matches_json(completion) {
198 continue;
199 }
200 // A non-OK status on the completion is the ledger rejecting the
201 // command, which is an answer — the same one the gRPC lane
202 // reports as `CommandRejected`.
203 if let Some(status) = completion.get("status") {
204 let code = status.get("code").and_then(serde_json::Value::as_i64);
205 if code.is_some_and(|code| code != 0) {
206 return Err(Error::CommandRejected {
207 code: code.unwrap_or_default().to_string(),
208 message: status
209 .get("message")
210 .and_then(serde_json::Value::as_str)
211 .unwrap_or_default()
212 .to_string(),
213 });
214 }
215 }
216 return Ok(completion.clone());
217 }
218 Err(Error::UnexpectedResponse(format!(
219 "completion stream ended before command {} was seen",
220 self.change_id.command_id()
221 )))
222 });
223
224 Box::pin(tokio::time::timeout(timeout, scan))
225 .await
226 .map_err(|_| canton_core::Error::Timeout)?
227 }
228}