daml_grpc/data/
completion.rs

1use std::convert::TryFrom;
2use std::time::Duration;
3
4use chrono::{DateTime, Utc};
5
6use crate::data::offset::DamlLedgerOffset;
7use crate::data::{DamlError, DamlResult};
8use crate::grpc_protobuf::com::daml::ledger::api::v2::completion::DeduplicationPeriod;
9use crate::grpc_protobuf::com::daml::ledger::api::v2::completion_stream_response::CompletionResponse;
10use crate::grpc_protobuf::com::daml::ledger::api::v2::{
11    Completion, CompletionStreamResponse, OffsetCheckpoint, SynchronizerTime,
12};
13use crate::grpc_protobuf::google::rpc::Status;
14use crate::util;
15use crate::util::Required;
16
17/// One element in a `CompletionStream`. v2 sends *either* a [`DamlCompletion`]
18/// (the participant's verdict on a submission) *or* a
19/// [`DamlOffsetCheckpoint`] (a periodic offset marker used to detect
20/// timeouts and to checkpoint stream resumption); never both in the same
21/// message.
22#[derive(Debug, Eq, PartialEq, Clone)]
23pub enum DamlCompletionResponse {
24    Completion(DamlCompletion),
25    OffsetCheckpoint(DamlOffsetCheckpoint),
26}
27
28impl TryFrom<CompletionStreamResponse> for DamlCompletionResponse {
29    type Error = DamlError;
30
31    fn try_from(response: CompletionStreamResponse) -> DamlResult<Self> {
32        match response.completion_response.req()? {
33            CompletionResponse::Completion(c) => Ok(Self::Completion(DamlCompletion::try_from(c)?)),
34            CompletionResponse::OffsetCheckpoint(c) => Ok(Self::OffsetCheckpoint(DamlOffsetCheckpoint::try_from(c)?)),
35        }
36    }
37}
38
39#[derive(Debug, Eq, PartialEq, Clone, Default)]
40pub struct DamlCompletion {
41    pub command_id: String,
42    pub status: DamlStatus,
43    /// The id of the resulting transaction or reassignment. v2 generalised
44    /// v1's `transaction_id` because a command may now produce a
45    /// non-transaction update (e.g. a reassignment).
46    pub update_id: String,
47    /// v2 renames v1's `application_id`.
48    pub user_id: String,
49    pub act_as: Vec<String>,
50    pub submission_id: String,
51    pub deduplication_period: Option<DamlCompletionDeduplicationPeriod>,
52    /// Offset at which the participant emitted this completion. Use this
53    /// in a follow-up `CompletionStreamRequest::begin_exclusive` to resume
54    /// the stream after a disconnect.
55    pub offset: DamlLedgerOffset,
56    /// The synchronizer that ordered the underlying confirmation request,
57    /// plus its record time at the corresponding offset.
58    pub synchronizer_time: Option<DamlSynchronizerTime>,
59    /// Traffic cost paid by this participant for the submission. Zero for
60    /// pre-ordering rejections; see proto docs for caveats.
61    pub paid_traffic_cost: i64,
62}
63
64impl TryFrom<Completion> for DamlCompletion {
65    type Error = DamlError;
66
67    fn try_from(c: Completion) -> DamlResult<Self> {
68        Ok(Self {
69            command_id: c.command_id,
70            // Per proto, `status` is documented as optional but is set on
71            // every completion the participant emits in practice — treat
72            // absence as a wire-protocol violation.
73            status: DamlStatus::from(c.status.req()?),
74            update_id: c.update_id,
75            user_id: c.user_id,
76            act_as: c.act_as,
77            submission_id: c.submission_id,
78            deduplication_period: c
79                .deduplication_period
80                .map(DamlCompletionDeduplicationPeriod::try_from)
81                .transpose()?,
82            offset: DamlLedgerOffset::new(c.offset),
83            synchronizer_time: c.synchronizer_time.map(DamlSynchronizerTime::try_from).transpose()?,
84            paid_traffic_cost: c.paid_traffic_cost,
85        })
86    }
87}
88
89/// Periodic offset marker emitted in the completion (and update) streams.
90///
91/// Lets clients (a) detect commands that have likely timed out (no
92/// completion received before `synchronizer_times` advanced past the
93/// command's max record time) and (b) checkpoint stream position so a
94/// later subscription can resume from the same point.
95#[derive(Debug, Eq, PartialEq, Clone, Default)]
96pub struct DamlOffsetCheckpoint {
97    pub offset: DamlLedgerOffset,
98    pub synchronizer_times: Vec<DamlSynchronizerTime>,
99}
100
101impl TryFrom<OffsetCheckpoint> for DamlOffsetCheckpoint {
102    type Error = DamlError;
103
104    fn try_from(c: OffsetCheckpoint) -> DamlResult<Self> {
105        Ok(Self {
106            offset: DamlLedgerOffset::new(c.offset),
107            synchronizer_times: c
108                .synchronizer_times
109                .into_iter()
110                .map(DamlSynchronizerTime::try_from)
111                .collect::<DamlResult<Vec<_>>>()?,
112        })
113    }
114}
115
116/// A `(synchronizer_id, record_time)` pair, attached to checkpoints and
117/// completions so clients can reason about per-synchronizer freshness.
118#[derive(Debug, Eq, PartialEq, Clone, Default)]
119pub struct DamlSynchronizerTime {
120    pub synchronizer_id: String,
121    pub record_time: DateTime<Utc>,
122}
123
124impl TryFrom<SynchronizerTime> for DamlSynchronizerTime {
125    type Error = DamlError;
126
127    fn try_from(s: SynchronizerTime) -> DamlResult<Self> {
128        Ok(Self {
129            synchronizer_id: s.synchronizer_id,
130            record_time: util::from_grpc_timestamp(&s.record_time.req()?)?,
131        })
132    }
133}
134
135#[derive(Debug, Eq, PartialEq, Clone, Default)]
136pub struct DamlStatus {
137    pub code: i32,
138    pub message: String,
139    /// Structured error details as a list of `google.protobuf.Any`.
140    /// Downstream decoders can match on `type_url` to recover the
141    /// concrete Daml error payload (e.g. `com.daml.error.ErrorInfo`).
142    pub details: Vec<prost_types::Any>,
143}
144
145impl From<Status> for DamlStatus {
146    fn from(status: Status) -> Self {
147        Self {
148            code: status.code,
149            message: status.message,
150            details: status.details,
151        }
152    }
153}
154
155#[derive(Debug, Eq, PartialEq, Clone)]
156pub enum DamlCompletionDeduplicationPeriod {
157    /// Completion-stream offset (exclusive). v2 changed this from a
158    /// stringified offset to a real `int64`.
159    DeduplicationOffset(i64),
160    DeduplicationDuration(Duration),
161}
162
163impl TryFrom<DeduplicationPeriod> for DamlCompletionDeduplicationPeriod {
164    type Error = DamlError;
165
166    fn try_from(p: DeduplicationPeriod) -> DamlResult<Self> {
167        Ok(match p {
168            DeduplicationPeriod::DeduplicationOffset(offset) => Self::DeduplicationOffset(offset),
169            DeduplicationPeriod::DeduplicationDuration(duration) => {
170                Self::DeduplicationDuration(util::from_grpc_duration(&duration)?)
171            },
172        })
173    }
174}
daml_grpc/data/completion.rs

daml_grpc/data/
completion.rs
