Coverage Report

Created: 2025-05-30 16:37

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/build/source/nativelink-util/src/action_messages.rs
Line
Count
Source
1
// Copyright 2024 The NativeLink Authors. All rights reserved.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//    http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14
15
use core::cmp::Ordering;
16
use core::convert::Into;
17
use core::hash::Hash;
18
use core::time::Duration;
19
use std::collections::HashMap;
20
use std::time::SystemTime;
21
22
use nativelink_error::{Error, ResultExt, error_if, make_input_err};
23
use nativelink_metric::{
24
    MetricFieldData, MetricKind, MetricPublishKnownKindData, MetricsComponent, publish,
25
};
26
use nativelink_proto::build::bazel::remote::execution::v2::{
27
    Action, ActionResult as ProtoActionResult, ExecuteOperationMetadata, ExecuteRequest,
28
    ExecuteResponse, ExecutedActionMetadata, FileNode, LogFile, OutputDirectory, OutputFile,
29
    OutputSymlink, SymlinkNode, execution_stage,
30
};
31
use nativelink_proto::google::longrunning::Operation;
32
use nativelink_proto::google::longrunning::operation::Result as LongRunningResult;
33
use nativelink_proto::google::rpc::Status;
34
use prost::Message;
35
use prost::bytes::Bytes;
36
use prost_types::Any;
37
use serde::ser::Error as SerdeError;
38
use serde::{Deserialize, Serialize};
39
use uuid::Uuid;
40
41
use crate::common::{DigestInfo, HashMapExt, VecExt};
42
use crate::digest_hasher::DigestHasherFunc;
43
44
/// Default priority remote execution jobs will get when not provided.
45
pub const DEFAULT_EXECUTION_PRIORITY: i32 = 0;
46
47
/// Exit code sent if there is an internal error.
48
pub const INTERNAL_ERROR_EXIT_CODE: i32 = -178;
49
50
/// Holds an id that is unique to the client for a requested operation.
51
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
52
pub enum OperationId {
53
    Uuid(Uuid),
54
    String(String),
55
}
56
57
impl OperationId {
58
1
    pub fn into_string(self) -> String {
59
1
        match self {
60
1
            Self::Uuid(uuid) => uuid.to_string(),
61
0
            Self::String(name) => name,
62
        }
63
1
    }
64
}
65
66
impl Default for OperationId {
67
96
    fn default() -> Self {
68
96
        Self::Uuid(Uuid::new_v4())
69
96
    }
70
}
71
72
impl core::fmt::Display for OperationId {
73
78
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
74
78
        match self {
75
71
            Self::Uuid(uuid) => uuid.fmt(f),
76
7
            Self::String(name) => f.write_str(name),
77
        }
78
78
    }
79
}
80
81
impl MetricsComponent for OperationId {
82
0
    fn publish(
83
0
        &self,
84
0
        _kind: MetricKind,
85
0
        _field_metadata: MetricFieldData,
86
0
    ) -> Result<MetricPublishKnownKindData, nativelink_metric::Error> {
87
0
        Ok(MetricPublishKnownKindData::String(self.to_string()))
88
0
    }
89
}
90
91
impl From<&str> for OperationId {
92
28
    fn from(value: &str) -> Self {
93
28
        Uuid::parse_str(value).map_or_else(|_| Self::String(
value7
.
to_string7
()), Self::Uuid)
94
28
    }
95
}
96
97
impl From<String> for OperationId {
98
8
    fn from(value: String) -> Self {
99
8
        Uuid::parse_str(&value).map_or(Self::String(value), Self::Uuid)
100
8
    }
101
}
102
103
impl TryFrom<Bytes> for OperationId {
104
    type Error = Error;
105
106
0
    fn try_from(value: Bytes) -> Result<Self, Self::Error> {
107
        // This is an optimized path to attempt to do the conversion in-place
108
        // to avoid an extra allocation/copy.
109
0
        match value.try_into_mut() {
110
            // We are the only reference to the Bytes, so we can convert it into a Vec<u8>
111
            // for free then convert the Vec<u8> to a String for free too.
112
0
            Ok(value) => {
113
0
                let value = String::from_utf8(value.into()).map_err(|e| {
114
0
                    make_input_err!(
115
                        "Failed to convert bytes to string in try_from<Bytes> for OperationId : {e:?}"
116
                    )
117
0
                })?;
118
0
                Ok(Self::from(value))
119
            }
120
            // We could not take ownership of the Bytes, so we may need to copy our data.
121
0
            Err(value) => {
122
0
                let value = core::str::from_utf8(&value).map_err(|e| {
123
0
                    make_input_err!(
124
                        "Failed to convert bytes to string in try_from<Bytes> for OperationId : {e:?}"
125
                    )
126
0
                })?;
127
0
                Ok(Self::from(value))
128
            }
129
        }
130
0
    }
131
}
132
133
/// Unique id of worker.
134
#[derive(Default, Eq, PartialEq, Hash, Clone, Serialize, Deserialize)]
135
pub struct WorkerId(pub String);
136
137
impl MetricsComponent for WorkerId {
138
0
    fn publish(
139
0
        &self,
140
0
        _kind: MetricKind,
141
0
        _field_metadata: MetricFieldData,
142
0
    ) -> Result<MetricPublishKnownKindData, nativelink_metric::Error> {
143
0
        Ok(MetricPublishKnownKindData::String(self.0.clone()))
144
0
    }
145
}
146
147
impl core::fmt::Display for WorkerId {
148
25
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
149
25
        f.write_fmt(format_args!("{}", self.0))
150
25
    }
151
}
152
153
impl core::fmt::Debug for WorkerId {
154
5
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
155
5
        core::fmt::Display::fmt(&self, f)
156
5
    }
157
}
158
159
impl From<WorkerId> for String {
160
88
    fn from(val: WorkerId) -> Self {
161
88
        val.0
162
88
    }
163
}
164
165
impl From<String> for WorkerId {
166
8
    fn from(s: String) -> Self {
167
8
        Self(s)
168
8
    }
169
}
170
171
/// Holds the information needed to uniquely identify an action
172
/// and if it is cacheable or not.
173
#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize)]
174
pub enum ActionUniqueQualifier {
175
    /// The action is cacheable.
176
    Cacheable(ActionUniqueKey),
177
    /// The action is uncacheable.
178
    Uncacheable(ActionUniqueKey),
179
}
180
181
impl MetricsComponent for ActionUniqueQualifier {
182
0
    fn publish(
183
0
        &self,
184
0
        _kind: MetricKind,
185
0
        field_metadata: MetricFieldData,
186
0
    ) -> Result<MetricPublishKnownKindData, nativelink_metric::Error> {
187
0
        let (cacheable, action) = match self {
188
0
            Self::Cacheable(action) => (true, action),
189
0
            Self::Uncacheable(action) => (false, action),
190
        };
191
0
        publish!(
192
0
            cacheable,
193
0
            &cacheable,
194
0
            MetricKind::Default,
195
0
            "If the action is cacheable.",
196
0
            ""
197
        );
198
0
        action.publish(MetricKind::Component, field_metadata)?;
199
0
        Ok(MetricPublishKnownKindData::Component)
200
0
    }
201
}
202
203
impl ActionUniqueQualifier {
204
    /// Get the `instance_name` of the action.
205
0
    pub const fn instance_name(&self) -> &String {
206
0
        match self {
207
0
            Self::Cacheable(action) | Self::Uncacheable(action) => &action.instance_name,
208
        }
209
0
    }
210
211
    /// Get the digest function of the action.
212
11
    pub const fn digest_function(&self) -> DigestHasherFunc {
213
11
        match self {
214
11
            Self::Cacheable(action) | Self::Uncacheable(
action0
) => action.digest_function,
215
        }
216
11
    }
217
218
    /// Get the digest of the action.
219
115
    pub const fn digest(&self) -> DigestInfo {
220
115
        match self {
221
115
            Self::Cacheable(
action110
) | Self::Uncacheable(
action5
) => action.digest,
222
        }
223
115
    }
224
}
225
226
impl core::fmt::Display for ActionUniqueQualifier {
227
3
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
228
3
        let (cacheable, unique_key) = match self {
229
3
            Self::Cacheable(action) => (true, action),
230
0
            Self::Uncacheable(action) => (false, action),
231
        };
232
3
        f.write_fmt(format_args!(
233
            // Note: We use underscores because it makes escaping easier
234
            // for redis.
235
3
            "{}_{}_{}_{}_{}",
236
            unique_key.instance_name,
237
            unique_key.digest_function,
238
3
            unique_key.digest.packed_hash(),
239
3
            unique_key.digest.size_bytes(),
240
3
            if cacheable { 'c' } else { 
'u'0
},
  Branch (240:16): [True: 3, False: 0]
  Branch (240:16): [Folded - Ignored]
241
        ))
242
3
    }
243
}
244
245
/// This is a utility struct used to make it easier to match `ActionInfos` in a
246
/// `HashMap` without needing to construct an entire `ActionInfo`.
247
#[derive(Debug, Clone, Eq, PartialEq, Hash, Serialize, Deserialize, MetricsComponent)]
248
pub struct ActionUniqueKey {
249
    /// Name of instance group this action belongs to.
250
    #[metric(help = "Name of instance group this action belongs to.")]
251
    pub instance_name: String,
252
    /// The digest function this action expects.
253
    #[metric(help = "The digest function this action expects.")]
254
    pub digest_function: DigestHasherFunc,
255
    /// Digest of the underlying `Action`.
256
    #[metric(help = "Digest of the underlying Action.")]
257
    pub digest: DigestInfo,
258
}
259
260
impl core::fmt::Display for ActionUniqueKey {
261
0
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
262
0
        f.write_fmt(format_args!(
263
0
            "{}/{}/{}",
264
            self.instance_name, self.digest_function, self.digest,
265
        ))
266
0
    }
267
}
268
269
/// Information needed to execute an action. This struct is used over bazel's proto `Action`
270
/// for simplicity and offers a `salt`, which is useful to ensure during hashing (for dicts)
271
/// to ensure we never match against another `ActionInfo` (when a task should never be cached).
272
/// This struct must be 100% compatible with `ExecuteRequest` struct in `remote_execution.proto`
273
/// except for the salt field.
274
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, MetricsComponent)]
275
pub struct ActionInfo {
276
    /// Digest of the underlying `Command`.
277
    #[metric(help = "Digest of the underlying Command.")]
278
    pub command_digest: DigestInfo,
279
    /// Digest of the underlying `Directory`.
280
    #[metric(help = "Digest of the underlying Directory.")]
281
    pub input_root_digest: DigestInfo,
282
    /// Timeout of the action.
283
    #[metric(help = "Timeout of the action.")]
284
    pub timeout: Duration,
285
    /// The properties rules that must be applied when finding a worker that can run this action.
286
    #[metric(group = "platform_properties")]
287
    pub platform_properties: HashMap<String, String>,
288
    /// The priority of the action. Higher value means it should execute faster.
289
    #[metric(help = "The priority of the action. Higher value means it should execute faster.")]
290
    pub priority: i32,
291
    /// When this action started to be loaded from the CAS.
292
    #[metric(help = "When this action started to be loaded from the CAS.")]
293
    pub load_timestamp: SystemTime,
294
    /// When this action was created.
295
    #[metric(help = "When this action was created.")]
296
    pub insert_timestamp: SystemTime,
297
    /// Info used to uniquely identify this `ActionInfo` and if it is cacheable.
298
    /// This is primarily used to join actions/operations together using this key.
299
    #[metric(help = "Info used to uniquely identify this ActionInfo and if it is cacheable.")]
300
    pub unique_qualifier: ActionUniqueQualifier,
301
}
302
303
impl ActionInfo {
304
    #[inline]
305
0
    pub const fn instance_name(&self) -> &String {
306
0
        self.unique_qualifier.instance_name()
307
0
    }
308
309
    /// Returns the underlying digest of the `Action`.
310
    #[inline]
311
74
    pub const fn digest(&self) -> DigestInfo {
312
74
        self.unique_qualifier.digest()
313
74
    }
314
315
16
    pub fn try_from_action_and_execute_request(
316
16
        execute_request: ExecuteRequest,
317
16
        action: Action,
318
16
        load_timestamp: SystemTime,
319
16
        queued_timestamp: SystemTime,
320
16
    ) -> Result<Self, Error> {
321
16
        let unique_key = ActionUniqueKey {
322
16
            instance_name: execute_request.instance_name,
323
16
            digest_function: DigestHasherFunc::try_from(execute_request.digest_function)
324
16
                .err_tip(|| format!(
"Could not find digest_function in try_from_action_and_execute_request {:?}"0
, execute_request.digest_function))
?0
,
325
16
            digest: execute_request
326
16
                .action_digest
327
16
                .err_tip(|| "Expected action_digest to exist on ExecuteRequest")
?0
328
16
                .try_into()
?0
,
329
        };
330
16
        let unique_qualifier = if execute_request.skip_cache_lookup {
  Branch (330:35): [True: 0, False: 16]
  Branch (330:35): [Folded - Ignored]
331
0
            ActionUniqueQualifier::Uncacheable(unique_key)
332
        } else {
333
16
            ActionUniqueQualifier::Cacheable(unique_key)
334
        };
335
336
16
        let proto_properties = action.platform.unwrap_or_default();
337
16
        let mut platform_properties = HashMap::with_capacity(proto_properties.properties.len());
338
17
        for 
property1
in proto_properties.properties {
339
1
            platform_properties.insert(property.name, property.value);
340
1
        }
341
342
        Ok(Self {
343
16
            command_digest: action
344
16
                .command_digest
345
16
                .err_tip(|| "Expected command_digest to exist on Action")
?0
346
16
                .try_into()
?0
,
347
16
            input_root_digest: action
348
16
                .input_root_digest
349
16
                .err_tip(|| "Expected input_root_digest to exist on Action")
?0
350
16
                .try_into()
?0
,
351
16
            timeout: action
352
16
                .timeout
353
16
                .unwrap_or_default()
354
16
                .try_into()
355
16
                .map_err(|_| make_input_err!("Failed convert proto duration to system duration"))
?0
,
356
16
            platform_properties,
357
16
            priority: execute_request
358
16
                .execution_policy
359
16
                .unwrap_or_default()
360
16
                .priority,
361
16
            load_timestamp,
362
16
            insert_timestamp: queued_timestamp,
363
16
            unique_qualifier,
364
        })
365
16
    }
366
}
367
368
impl From<&ActionInfo> for ExecuteRequest {
369
32
    fn from(val: &ActionInfo) -> Self {
370
32
        let digest = val.digest().into();
371
32
        let (skip_cache_lookup, unique_qualifier) = match &val.unique_qualifier {
372
27
            ActionUniqueQualifier::Cacheable(unique_qualifier) => (false, unique_qualifier),
373
5
            ActionUniqueQualifier::Uncacheable(unique_qualifier) => (true, unique_qualifier),
374
        };
375
32
        Self {
376
32
            instance_name: unique_qualifier.instance_name.clone(),
377
32
            action_digest: Some(digest),
378
32
            skip_cache_lookup,
379
32
            execution_policy: None,     // Not used in the worker.
380
32
            results_cache_policy: None, // Not used in the worker.
381
32
            digest_function: unique_qualifier.digest_function.proto_digest_func().into(),
382
32
        }
383
32
    }
384
}
385
386
/// Simple utility struct to determine if a string is representing a full path or
387
/// just the name of the file.
388
/// This is in order to be able to reuse the same struct instead of building different
389
/// structs when converting `FileInfo` -> {`OutputFile`, `FileNode`} and other similar
390
/// structs.
391
#[derive(Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
392
pub enum NameOrPath {
393
    Name(String),
394
    Path(String),
395
}
396
397
impl PartialOrd for NameOrPath {
398
0
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
399
0
        Some(self.cmp(other))
400
0
    }
401
}
402
403
impl Ord for NameOrPath {
404
0
    fn cmp(&self, other: &Self) -> Ordering {
405
0
        let self_lexical_name = match self {
406
0
            Self::Name(name) => name,
407
0
            Self::Path(path) => path,
408
        };
409
0
        let other_lexical_name = match other {
410
0
            Self::Name(name) => name,
411
0
            Self::Path(path) => path,
412
        };
413
0
        self_lexical_name.cmp(other_lexical_name)
414
0
    }
415
}
416
417
/// Represents an individual file and associated metadata.
418
/// This struct must be 100% compatible with `OutputFile` and `FileNode` structs
419
/// in `remote_execution.proto`.
420
#[derive(Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
421
pub struct FileInfo {
422
    pub name_or_path: NameOrPath,
423
    pub digest: DigestInfo,
424
    pub is_executable: bool,
425
}
426
427
impl TryFrom<FileInfo> for FileNode {
428
    type Error = Error;
429
430
3
    fn try_from(val: FileInfo) -> Result<Self, Error> {
431
3
        match val.name_or_path {
432
0
            NameOrPath::Path(_) => Err(make_input_err!(
433
0
                "Cannot return a FileInfo that uses a NameOrPath::Path(), it must be a NameOrPath::Name()"
434
0
            )),
435
3
            NameOrPath::Name(name) => Ok(Self {
436
3
                name,
437
3
                digest: Some((&val.digest).into()),
438
3
                is_executable: val.is_executable,
439
3
                node_properties: None, // Not supported.
440
3
            }),
441
        }
442
3
    }
443
}
444
445
impl TryFrom<OutputFile> for FileInfo {
446
    type Error = Error;
447
448
2
    fn try_from(output_file: OutputFile) -> Result<Self, Error> {
449
        Ok(Self {
450
2
            name_or_path: NameOrPath::Path(output_file.path),
451
2
            digest: output_file
452
2
                .digest
453
2
                .err_tip(|| "Expected digest to exist on OutputFile")
?0
454
2
                .try_into()
?0
,
455
2
            is_executable: output_file.is_executable,
456
        })
457
2
    }
458
}
459
460
impl TryFrom<FileInfo> for OutputFile {
461
    type Error = Error;
462
463
7
    fn try_from(val: FileInfo) -> Result<Self, Error> {
464
7
        match val.name_or_path {
465
0
            NameOrPath::Name(_) => Err(make_input_err!(
466
0
                "Cannot return a FileInfo that uses a NameOrPath::Name(), it must be a NameOrPath::Path()"
467
0
            )),
468
7
            NameOrPath::Path(path) => Ok(Self {
469
7
                path,
470
7
                digest: Some((&val.digest).into()),
471
7
                is_executable: val.is_executable,
472
7
                contents: Bytes::default(),
473
7
                node_properties: None, // Not supported.
474
7
            }),
475
        }
476
7
    }
477
}
478
479
/// Represents an individual symlink file and associated metadata.
480
/// This struct must be 100% compatible with `SymlinkNode` and `OutputSymlink`.
481
#[derive(Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
482
pub struct SymlinkInfo {
483
    pub name_or_path: NameOrPath,
484
    pub target: String,
485
}
486
487
impl TryFrom<SymlinkNode> for SymlinkInfo {
488
    type Error = Error;
489
490
0
    fn try_from(symlink_node: SymlinkNode) -> Result<Self, Error> {
491
0
        Ok(Self {
492
0
            name_or_path: NameOrPath::Name(symlink_node.name),
493
0
            target: symlink_node.target,
494
0
        })
495
0
    }
496
}
497
498
impl TryFrom<SymlinkInfo> for SymlinkNode {
499
    type Error = Error;
500
501
1
    fn try_from(val: SymlinkInfo) -> Result<Self, Error> {
502
1
        match val.name_or_path {
503
0
            NameOrPath::Path(_) => Err(make_input_err!(
504
0
                "Cannot return a SymlinkInfo that uses a NameOrPath::Path(), it must be a NameOrPath::Name()"
505
0
            )),
506
1
            NameOrPath::Name(name) => Ok(Self {
507
1
                name,
508
1
                target: val.target,
509
1
                node_properties: None, // Not supported.
510
1
            }),
511
        }
512
1
    }
513
}
514
515
impl TryFrom<OutputSymlink> for SymlinkInfo {
516
    type Error = Error;
517
518
2
    fn try_from(output_symlink: OutputSymlink) -> Result<Self, Error> {
519
2
        Ok(Self {
520
2
            name_or_path: NameOrPath::Path(output_symlink.path),
521
2
            target: output_symlink.target,
522
2
        })
523
2
    }
524
}
525
526
impl TryFrom<SymlinkInfo> for OutputSymlink {
527
    type Error = Error;
528
529
2
    fn try_from(val: SymlinkInfo) -> Result<Self, Error> {
530
2
        match val.name_or_path {
531
2
            NameOrPath::Path(path) => {
532
2
                Ok(Self {
533
2
                    path,
534
2
                    target: val.target,
535
2
                    node_properties: None, // Not supported.
536
2
                })
537
            }
538
0
            NameOrPath::Name(_) => Err(make_input_err!(
539
0
                "Cannot return a SymlinkInfo that uses a NameOrPath::Name(), it must be a NameOrPath::Path()"
540
0
            )),
541
        }
542
2
    }
543
}
544
545
/// Represents an individual directory file and associated metadata.
546
/// This struct must be 100% compatible with `SymlinkNode` and `OutputSymlink`.
547
#[derive(Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
548
pub struct DirectoryInfo {
549
    pub path: String,
550
    pub tree_digest: DigestInfo,
551
}
552
553
impl TryFrom<OutputDirectory> for DirectoryInfo {
554
    type Error = Error;
555
556
2
    fn try_from(output_directory: OutputDirectory) -> Result<Self, Error> {
557
        Ok(Self {
558
2
            path: output_directory.path,
559
2
            tree_digest: output_directory
560
2
                .tree_digest
561
2
                .err_tip(|| "Expected tree_digest to exist in OutputDirectory")
?0
562
2
                .try_into()
?0
,
563
        })
564
2
    }
565
}
566
567
impl From<DirectoryInfo> for OutputDirectory {
568
1
    fn from(val: DirectoryInfo) -> Self {
569
1
        Self {
570
1
            path: val.path,
571
1
            tree_digest: Some(val.tree_digest.into()),
572
1
            is_topologically_sorted: false,
573
1
        }
574
1
    }
575
}
576
577
/// Represents the metadata associated with the execution result.
578
/// This struct must be 100% compatible with `ExecutedActionMetadata`.
579
#[derive(Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
580
pub struct ExecutionMetadata {
581
    pub worker: String,
582
    pub queued_timestamp: SystemTime,
583
    pub worker_start_timestamp: SystemTime,
584
    pub worker_completed_timestamp: SystemTime,
585
    pub input_fetch_start_timestamp: SystemTime,
586
    pub input_fetch_completed_timestamp: SystemTime,
587
    pub execution_start_timestamp: SystemTime,
588
    pub execution_completed_timestamp: SystemTime,
589
    pub output_upload_start_timestamp: SystemTime,
590
    pub output_upload_completed_timestamp: SystemTime,
591
}
592
593
impl Default for ExecutionMetadata {
594
1
    fn default() -> Self {
595
1
        Self {
596
1
            worker: String::new(),
597
1
            queued_timestamp: SystemTime::UNIX_EPOCH,
598
1
            worker_start_timestamp: SystemTime::UNIX_EPOCH,
599
1
            worker_completed_timestamp: SystemTime::UNIX_EPOCH,
600
1
            input_fetch_start_timestamp: SystemTime::UNIX_EPOCH,
601
1
            input_fetch_completed_timestamp: SystemTime::UNIX_EPOCH,
602
1
            execution_start_timestamp: SystemTime::UNIX_EPOCH,
603
1
            execution_completed_timestamp: SystemTime::UNIX_EPOCH,
604
1
            output_upload_start_timestamp: SystemTime::UNIX_EPOCH,
605
1
            output_upload_completed_timestamp: SystemTime::UNIX_EPOCH,
606
1
        }
607
1
    }
608
}
609
610
impl From<ExecutionMetadata> for ExecutedActionMetadata {
611
17
    fn from(val: ExecutionMetadata) -> Self {
612
        Self {
613
17
            worker: val.worker,
614
17
            queued_timestamp: Some(val.queued_timestamp.into()),
615
17
            worker_start_timestamp: Some(val.worker_start_timestamp.into()),
616
17
            worker_completed_timestamp: Some(val.worker_completed_timestamp.into()),
617
17
            input_fetch_start_timestamp: Some(val.input_fetch_start_timestamp.into()),
618
17
            input_fetch_completed_timestamp: Some(val.input_fetch_completed_timestamp.into()),
619
17
            execution_start_timestamp: Some(val.execution_start_timestamp.into()),
620
17
            execution_completed_timestamp: Some(val.execution_completed_timestamp.into()),
621
17
            output_upload_start_timestamp: Some(val.output_upload_start_timestamp.into()),
622
17
            output_upload_completed_timestamp: Some(val.output_upload_completed_timestamp.into()),
623
17
            virtual_execution_duration: val
624
17
                .execution_completed_timestamp
625
17
                .duration_since(val.execution_start_timestamp)
626
17
                .ok()
627
17
                .and_then(|duration| prost_types::Duration::try_from(duration).ok()),
628
17
            auxiliary_metadata: Vec::default(),
629
        }
630
17
    }
631
}
632
633
impl TryFrom<ExecutedActionMetadata> for ExecutionMetadata {
634
    type Error = Error;
635
636
3
    fn try_from(eam: ExecutedActionMetadata) -> Result<Self, Error> {
637
        Ok(Self {
638
3
            worker: eam.worker,
639
3
            queued_timestamp: eam
640
3
                .queued_timestamp
641
3
                .err_tip(|| "Expected queued_timestamp to exist in ExecutedActionMetadata")
?0
642
3
                .try_into()
?0
,
643
3
            worker_start_timestamp: eam
644
3
                .worker_start_timestamp
645
3
                .err_tip(|| "Expected worker_start_timestamp to exist in ExecutedActionMetadata")
?0
646
3
                .try_into()
?0
,
647
3
            worker_completed_timestamp: eam
648
3
                .worker_completed_timestamp
649
3
                .err_tip(|| 
{0
650
0
                    "Expected worker_completed_timestamp to exist in ExecutedActionMetadata"
651
0
                })?
652
3
                .try_into()
?0
,
653
3
            input_fetch_start_timestamp: eam
654
3
                .input_fetch_start_timestamp
655
3
                .err_tip(|| 
{0
656
0
                    "Expected input_fetch_start_timestamp to exist in ExecutedActionMetadata"
657
0
                })?
658
3
                .try_into()
?0
,
659
3
            input_fetch_completed_timestamp: eam
660
3
                .input_fetch_completed_timestamp
661
3
                .err_tip(|| 
{0
662
0
                    "Expected input_fetch_completed_timestamp to exist in ExecutedActionMetadata"
663
0
                })?
664
3
                .try_into()
?0
,
665
3
            execution_start_timestamp: eam
666
3
                .execution_start_timestamp
667
3
                .err_tip(|| 
{0
668
0
                    "Expected execution_start_timestamp to exist in ExecutedActionMetadata"
669
0
                })?
670
3
                .try_into()
?0
,
671
3
            execution_completed_timestamp: eam
672
3
                .execution_completed_timestamp
673
3
                .err_tip(|| 
{0
674
0
                    "Expected execution_completed_timestamp to exist in ExecutedActionMetadata"
675
0
                })?
676
3
                .try_into()
?0
,
677
3
            output_upload_start_timestamp: eam
678
3
                .output_upload_start_timestamp
679
3
                .err_tip(|| 
{0
680
0
                    "Expected output_upload_start_timestamp to exist in ExecutedActionMetadata"
681
0
                })?
682
3
                .try_into()
?0
,
683
3
            output_upload_completed_timestamp: eam
684
3
                .output_upload_completed_timestamp
685
3
                .err_tip(|| 
{0
686
0
                    "Expected output_upload_completed_timestamp to exist in ExecutedActionMetadata"
687
0
                })?
688
3
                .try_into()
?0
,
689
        })
690
3
    }
691
}
692
693
/// Represents the results of an execution.
694
/// This struct must be 100% compatible with `ActionResult` in `remote_execution.proto`.
695
#[derive(Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
696
pub struct ActionResult {
697
    pub output_files: Vec<FileInfo>,
698
    pub output_folders: Vec<DirectoryInfo>,
699
    pub output_directory_symlinks: Vec<SymlinkInfo>,
700
    pub output_file_symlinks: Vec<SymlinkInfo>,
701
    pub exit_code: i32,
702
    pub stdout_digest: DigestInfo,
703
    pub stderr_digest: DigestInfo,
704
    pub execution_metadata: ExecutionMetadata,
705
    pub server_logs: HashMap<String, DigestInfo>,
706
    pub error: Option<Error>,
707
    pub message: String,
708
}
709
710
impl Default for ActionResult {
711
7
    fn default() -> Self {
712
7
        Self {
713
7
            output_files: Vec::default(),
714
7
            output_folders: Vec::default(),
715
7
            output_directory_symlinks: Vec::default(),
716
7
            output_file_symlinks: Vec::default(),
717
7
            exit_code: INTERNAL_ERROR_EXIT_CODE,
718
7
            stdout_digest: DigestInfo::new([0u8; 32], 0),
719
7
            stderr_digest: DigestInfo::new([0u8; 32], 0),
720
7
            execution_metadata: ExecutionMetadata {
721
7
                worker: String::new(),
722
7
                queued_timestamp: SystemTime::UNIX_EPOCH,
723
7
                worker_start_timestamp: SystemTime::UNIX_EPOCH,
724
7
                worker_completed_timestamp: SystemTime::UNIX_EPOCH,
725
7
                input_fetch_start_timestamp: SystemTime::UNIX_EPOCH,
726
7
                input_fetch_completed_timestamp: SystemTime::UNIX_EPOCH,
727
7
                execution_start_timestamp: SystemTime::UNIX_EPOCH,
728
7
                execution_completed_timestamp: SystemTime::UNIX_EPOCH,
729
7
                output_upload_start_timestamp: SystemTime::UNIX_EPOCH,
730
7
                output_upload_completed_timestamp: SystemTime::UNIX_EPOCH,
731
7
            },
732
7
            server_logs: HashMap::default(),
733
7
            error: None,
734
7
            message: String::new(),
735
7
        }
736
7
    }
737
}
738
739
/// The execution status/stage. This should match `ExecutionStage::Value` in `remote_execution.proto`.
740
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize)]
741
#[allow(
742
    clippy::large_enum_variant,
743
    reason = "TODO box the two relevant variants in a breaking release. Unfulfilled on nightly"
744
)]
745
pub enum ActionStage {
746
    /// Stage is unknown.
747
    Unknown,
748
    /// Checking the cache to see if action exists.
749
    CacheCheck,
750
    /// Action has been accepted and waiting for worker to take it.
751
    Queued,
752
    // TODO(aaronmondal) We need a way to know if the job was sent to a worker, but hasn't begun
753
    // execution yet.
754
    /// Worker is executing the action.
755
    Executing,
756
    /// Worker completed the work with result.
757
    Completed(ActionResult),
758
    /// Result was found from cache, don't decode the proto just to re-encode it.
759
    #[serde(serialize_with = "serialize_proto_result", skip_deserializing)]
760
    // The serialization step decodes this to an ActionResult which is serializable.
761
    // Since it will always be serialized as an ActionResult, we do not need to support
762
    // deserialization on this type at all.
763
    // In theory, serializing this should never happen so performance shouldn't be affected.
764
    CompletedFromCache(ProtoActionResult),
765
}
766
767
0
fn serialize_proto_result<S>(v: &ProtoActionResult, serializer: S) -> Result<S::Ok, S::Error>
768
0
where
769
0
    S: serde::Serializer,
770
{
771
0
    let s = ActionResult::try_from(v.clone()).map_err(S::Error::custom)?;
772
0
    s.serialize(serializer)
773
0
}
774
775
impl ActionStage {
776
90
    pub const fn has_action_result(&self) -> bool {
777
90
        match self {
778
77
            Self::Unknown | Self::CacheCheck | Self::Queued | Self::Executing => false,
779
13
            Self::Completed(_) | Self::CompletedFromCache(_) => true,
780
        }
781
90
    }
782
783
    /// Returns true if the worker considers the action done and no longer needs to be tracked.
784
    // Note: This function is separate from `has_action_result()` to not mix the concept of
785
    //       "finished" with "has a result".
786
89
    pub const fn is_finished(&self) -> bool {
787
89
        self.has_action_result()
788
89
    }
789
790
    /// Returns if the stage enum is the same as the other stage enum, but
791
    /// does not compare the values of the enum.
792
39
    pub const fn is_same_stage(&self, other: &Self) -> bool {
793
39
        matches!(
794
39
            (self, other),
795
            (Self::Unknown, Self::Unknown)
796
                | (Self::CacheCheck, Self::CacheCheck)
797
                | (Self::Queued, Self::Queued)
798
                | (Self::Executing, Self::Executing)
799
                | (Self::Completed(_), Self::Completed(_))
800
                | (Self::CompletedFromCache(_), Self::CompletedFromCache(_))
801
        )
802
39
    }
803
}
804
805
impl MetricsComponent for ActionStage {
806
0
    fn publish(
807
0
        &self,
808
0
        _kind: MetricKind,
809
0
        _field_metadata: MetricFieldData,
810
0
    ) -> Result<MetricPublishKnownKindData, nativelink_metric::Error> {
811
0
        let value = match self {
812
0
            Self::Unknown => "Unknown".to_string(),
813
0
            Self::CacheCheck => "CacheCheck".to_string(),
814
0
            Self::Queued => "Queued".to_string(),
815
0
            Self::Executing => "Executing".to_string(),
816
0
            Self::Completed(_) => "Completed".to_string(),
817
0
            Self::CompletedFromCache(_) => "CompletedFromCache".to_string(),
818
        };
819
0
        Ok(MetricPublishKnownKindData::String(value))
820
0
    }
821
}
822
823
impl From<&ActionStage> for execution_stage::Value {
824
1
    fn from(val: &ActionStage) -> Self {
825
1
        match val {
826
0
            ActionStage::Unknown => Self::Unknown,
827
0
            ActionStage::CacheCheck => Self::CacheCheck,
828
0
            ActionStage::Queued => Self::Queued,
829
0
            ActionStage::Executing => Self::Executing,
830
1
            ActionStage::Completed(_) | ActionStage::CompletedFromCache(_) => Self::Completed,
831
        }
832
1
    }
833
}
834
835
11
pub fn to_execute_response(action_result: ActionResult) -> ExecuteResponse {
836
11
    fn logs_from(server_logs: HashMap<String, DigestInfo>) -> HashMap<String, LogFile> {
837
11
        let mut logs = HashMap::with_capacity(server_logs.len());
838
12
        for (
k1
,
v1
) in server_logs {
839
1
            logs.insert(
840
1
                k.clone(),
841
1
                LogFile {
842
1
                    digest: Some(v.into()),
843
1
                    human_readable: false,
844
1
                },
845
1
            );
846
1
        }
847
11
        logs
848
11
    }
849
850
11
    let status = Some(
851
11
        action_result
852
11
            .error
853
11
            .clone()
854
11
            .map_or_else(Status::default, Into::into),
855
11
    );
856
11
    let message = action_result.message.clone();
857
11
    ExecuteResponse {
858
11
        server_logs: logs_from(action_result.server_logs.clone()),
859
11
        result: action_result.try_into().ok(),
860
11
        cached_result: false,
861
11
        status,
862
11
        message,
863
11
    }
864
11
}
865
866
impl From<ActionStage> for ExecuteResponse {
867
6
    fn from(val: ActionStage) -> Self {
868
6
        match val {
869
            // We don't have an execute response if we don't have the results. It is defined
870
            // behavior to return an empty proto struct.
871
            ActionStage::Unknown
872
            | ActionStage::CacheCheck
873
            | ActionStage::Queued
874
0
            | ActionStage::Executing => Self::default(),
875
6
            ActionStage::Completed(action_result) => to_execute_response(action_result),
876
            // Handled separately as there are no server logs and the action
877
            // result is already in Proto format.
878
0
            ActionStage::CompletedFromCache(proto_action_result) => Self {
879
0
                server_logs: HashMap::new(),
880
0
                result: Some(proto_action_result),
881
0
                cached_result: true,
882
0
                status: Some(Status::default()),
883
0
                message: String::new(), // Will be populated later if applicable.
884
0
            },
885
        }
886
6
    }
887
}
888
889
impl TryFrom<ActionResult> for ProtoActionResult {
890
    type Error = Error;
891
892
17
    fn try_from(val: ActionResult) -> Result<Self, Error> {
893
17
        let mut output_symlinks = Vec::with_capacity(
894
17
            val.output_file_symlinks.len() + val.output_directory_symlinks.len(),
895
        );
896
17
        output_symlinks.extend_from_slice(val.output_file_symlinks.as_slice());
897
17
        output_symlinks.extend_from_slice(val.output_directory_symlinks.as_slice());
898
899
        Ok(Self {
900
17
            output_files: val
901
17
                .output_files
902
17
                .into_iter()
903
17
                .map(TryInto::try_into)
904
17
                .collect::<Result<_, _>>()
?0
,
905
17
            output_file_symlinks: val
906
17
                .output_file_symlinks
907
17
                .into_iter()
908
17
                .map(TryInto::try_into)
909
17
                .collect::<Result<_, _>>()
?0
,
910
17
            output_symlinks: output_symlinks
911
17
                .into_iter()
912
17
                .map(TryInto::try_into)
913
17
                .collect::<Result<_, _>>()
?0
,
914
17
            output_directories: val
915
17
                .output_folders
916
17
                .into_iter()
917
17
                .map(TryInto::try_into)
918
17
                .collect::<Result<_, _>>()
?0
,
919
17
            output_directory_symlinks: val
920
17
                .output_directory_symlinks
921
17
                .into_iter()
922
17
                .map(TryInto::try_into)
923
17
                .collect::<Result<_, _>>()
?0
,
924
17
            exit_code: val.exit_code,
925
17
            stdout_raw: Bytes::default(),
926
17
            stdout_digest: Some(val.stdout_digest.into()),
927
17
            stderr_raw: Bytes::default(),
928
17
            stderr_digest: Some(val.stderr_digest.into()),
929
17
            execution_metadata: Some(val.execution_metadata.into()),
930
        })
931
17
    }
932
}
933
934
impl TryFrom<ProtoActionResult> for ActionResult {
935
    type Error = Error;
936
937
0
    fn try_from(val: ProtoActionResult) -> Result<Self, Error> {
938
0
        let output_file_symlinks = val
939
0
            .output_file_symlinks
940
0
            .into_iter()
941
0
            .map(|output_symlink| {
942
0
                SymlinkInfo::try_from(output_symlink)
943
0
                    .err_tip(|| "Output File Symlinks could not be converted to SymlinkInfo")
944
0
            })
945
0
            .collect::<Result<Vec<_>, _>>()?;
946
947
0
        let output_directory_symlinks = val
948
0
            .output_directory_symlinks
949
0
            .into_iter()
950
0
            .map(|output_symlink| {
951
0
                SymlinkInfo::try_from(output_symlink)
952
0
                    .err_tip(|| "Output File Symlinks could not be converted to SymlinkInfo")
953
0
            })
954
0
            .collect::<Result<Vec<_>, _>>()?;
955
956
0
        let output_files = val
957
0
            .output_files
958
0
            .into_iter()
959
0
            .map(|output_file| {
960
0
                output_file
961
0
                    .try_into()
962
0
                    .err_tip(|| "Output File could not be converted")
963
0
            })
964
0
            .collect::<Result<Vec<_>, _>>()?;
965
966
0
        let output_folders = val
967
0
            .output_directories
968
0
            .into_iter()
969
0
            .map(|output_directory| {
970
0
                output_directory
971
0
                    .try_into()
972
0
                    .err_tip(|| "Output File could not be converted")
973
0
            })
974
0
            .collect::<Result<Vec<_>, _>>()?;
975
976
        Ok(Self {
977
0
            output_files,
978
0
            output_folders,
979
0
            output_file_symlinks,
980
0
            output_directory_symlinks,
981
0
            exit_code: val.exit_code,
982
0
            stdout_digest: val
983
0
                .stdout_digest
984
0
                .err_tip(|| "Expected stdout_digest to be set on ExecuteResponse msg")?
985
0
                .try_into()?,
986
0
            stderr_digest: val
987
0
                .stderr_digest
988
0
                .err_tip(|| "Expected stderr_digest to be set on ExecuteResponse msg")?
989
0
                .try_into()?,
990
0
            execution_metadata: val
991
0
                .execution_metadata
992
0
                .err_tip(|| "Expected execution_metadata to be set on ExecuteResponse msg")?
993
0
                .try_into()?,
994
0
            server_logs: HashMap::default(),
995
0
            error: None,
996
0
            message: String::new(),
997
        })
998
0
    }
999
}
1000
1001
impl TryFrom<ExecuteResponse> for ActionStage {
1002
    type Error = Error;
1003
1004
3
    fn try_from(execute_response: ExecuteResponse) -> Result<Self, Error> {
1005
3
        let proto_action_result = execute_response
1006
3
            .result
1007
3
            .err_tip(|| "Expected result to be set on ExecuteResponse msg")
?0
;
1008
3
        let action_result = ActionResult {
1009
3
            output_files: proto_action_result
1010
3
                .output_files
1011
3
                .try_map(TryInto::try_into)
?0
,
1012
3
            output_directory_symlinks: proto_action_result
1013
3
                .output_directory_symlinks
1014
3
                .try_map(TryInto::try_into)
?0
,
1015
3
            output_file_symlinks: proto_action_result
1016
3
                .output_file_symlinks
1017
3
                .try_map(TryInto::try_into)
?0
,
1018
3
            output_folders: proto_action_result
1019
3
                .output_directories
1020
3
                .try_map(TryInto::try_into)
?0
,
1021
3
            exit_code: proto_action_result.exit_code,
1022
1023
3
            stdout_digest: proto_action_result
1024
3
                .stdout_digest
1025
3
                .err_tip(|| "Expected stdout_digest to be set on ExecuteResponse msg")
?0
1026
3
                .try_into()
?0
,
1027
3
            stderr_digest: proto_action_result
1028
3
                .stderr_digest
1029
3
                .err_tip(|| "Expected stderr_digest to be set on ExecuteResponse msg")
?0
1030
3
                .try_into()
?0
,
1031
3
            execution_metadata: proto_action_result
1032
3
                .execution_metadata
1033
3
                .err_tip(|| "Expected execution_metadata to be set on ExecuteResponse msg")
?0
1034
3
                .try_into()
?0
,
1035
3
            server_logs: execute_response.server_logs.try_map(|v| 
{2
1036
2
                v.digest
1037
2
                    .err_tip(|| "Expected digest to be set on LogFile msg")
?0
1038
2
                    .try_into()
1039
2
            })
?0
,
1040
3
            error: execute_response
1041
3
                .status
1042
3
                .clone()
1043
3
                .and_then(|v| if v.code == 0 { 
None1
} else {
Some(v.into())2
}),
  Branch (1043:34): [True: 1, False: 2]
  Branch (1043:34): [Folded - Ignored]
1044
3
            message: execute_response.message,
1045
        };
1046
1047
3
        if execute_response.cached_result {
  Branch (1047:12): [True: 0, False: 3]
  Branch (1047:12): [Folded - Ignored]
1048
0
            return Ok(Self::CompletedFromCache(action_result.try_into()?));
1049
3
        }
1050
3
        Ok(Self::Completed(action_result))
1051
3
    }
1052
}
1053
1054
// TODO: Should be able to remove this after tokio-rs/prost#299
1055
trait TypeUrl: Message {
1056
    const TYPE_URL: &'static str;
1057
}
1058
1059
impl TypeUrl for ExecuteResponse {
1060
    const TYPE_URL: &'static str =
1061
        "type.googleapis.com/build.bazel.remote.execution.v2.ExecuteResponse";
1062
}
1063
1064
impl TypeUrl for ExecuteOperationMetadata {
1065
    const TYPE_URL: &'static str =
1066
        "type.googleapis.com/build.bazel.remote.execution.v2.ExecuteOperationMetadata";
1067
}
1068
1069
2
fn from_any<T>(message: &Any) -> Result<T, Error>
1070
2
where
1071
2
    T: TypeUrl + Default,
1072
{
1073
0
    error_if!(
1074
2
        message.type_url != T::TYPE_URL,
  Branch (1074:9): [True: 0, False: 1]
  Branch (1074:9): [True: 0, False: 1]
  Branch (1074:9): [Folded - Ignored]
1075
        "Incorrect type when decoding Any. {} != {}",
1076
        message.type_url,
1077
0
        T::TYPE_URL.to_string()
1078
    );
1079
2
    Ok(T::decode(message.value.as_slice())
?0
)
1080
2
}
1081
1082
2
fn to_any<T>(message: &T) -> Any
1083
2
where
1084
2
    T: TypeUrl,
1085
{
1086
2
    Any {
1087
2
        type_url: T::TYPE_URL.to_string(),
1088
2
        value: message.encode_to_vec(),
1089
2
    }
1090
2
}
1091
1092
/// Current state of the action.
1093
/// This must be 100% compatible with `Operation` in `google/longrunning/operations.proto`.
1094
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize, MetricsComponent)]
1095
pub struct ActionState {
1096
    #[metric(help = "The current stage of the action.")]
1097
    pub stage: ActionStage,
1098
    #[metric(help = "The unique identifier of the action.")]
1099
    pub client_operation_id: OperationId,
1100
    #[metric(help = "The digest of the action.")]
1101
    pub action_digest: DigestInfo,
1102
}
1103
1104
impl ActionState {
1105
1
    pub fn try_from_operation(
1106
1
        operation: Operation,
1107
1
        client_operation_id: OperationId,
1108
1
    ) -> Result<Self, Error> {
1109
1
        let metadata = from_any::<ExecuteOperationMetadata>(
1110
1
            &operation
1111
1
                .metadata
1112
1
                .err_tip(|| "No metadata in upstream operation")
?0
,
1113
        )
1114
1
        .err_tip(|| "Could not decode metadata in upstream operation")
?0
;
1115
1116
1
        let stage = match execution_stage::Value::try_from(metadata.stage).err_tip(|| 
{0
1117
0
            format!(
1118
0
                "Could not convert {} to execution_stage::Value",
1119
                metadata.stage
1120
            )
1121
0
        })? {
1122
0
            execution_stage::Value::Unknown => ActionStage::Unknown,
1123
0
            execution_stage::Value::CacheCheck => ActionStage::CacheCheck,
1124
0
            execution_stage::Value::Queued => ActionStage::Queued,
1125
0
            execution_stage::Value::Executing => ActionStage::Executing,
1126
            execution_stage::Value::Completed => {
1127
1
                let execute_response = operation
1128
1
                    .result
1129
1
                    .err_tip(|| "No result data for completed upstream action")
?0
;
1130
1
                match execute_response {
1131
0
                    LongRunningResult::Error(error) => ActionStage::Completed(ActionResult {
1132
0
                        error: Some(error.into()),
1133
0
                        ..ActionResult::default()
1134
0
                    }),
1135
1
                    LongRunningResult::Response(response) => {
1136
                        // Could be Completed, CompletedFromCache or Error.
1137
1
                        from_any::<ExecuteResponse>(&response)
1138
1
                            .err_tip(|| 
{0
1139
0
                                "Could not decode result structure for completed upstream action"
1140
0
                            })?
1141
1
                            .try_into()
?0
1142
                    }
1143
                }
1144
            }
1145
        };
1146
1147
1
        let action_digest = metadata
1148
1
            .action_digest
1149
1
            .err_tip(|| "No action_digest in upstream operation")
?0
1150
1
            .try_into()
1151
1
            .err_tip(|| "Could not convert action_digest into DigestInfo")
?0
;
1152
1153
1
        Ok(Self {
1154
1
            stage,
1155
1
            client_operation_id,
1156
1
            action_digest,
1157
1
        })
1158
1
    }
1159
1160
1
    pub fn as_operation(&self, client_operation_id: OperationId) -> Operation {
1161
1
        let stage = Into::<execution_stage::Value>::into(&self.stage) as i32;
1162
1
        let name = client_operation_id.into_string();
1163
1164
1
        let result = if self.stage.has_action_result() {
  Branch (1164:25): [True: 1, False: 0]
  Branch (1164:25): [Folded - Ignored]
1165
1
            let execute_response: ExecuteResponse = self.stage.clone().into();
1166
1
            Some(LongRunningResult::Response(to_any(&execute_response)))
1167
        } else {
1168
0
            None
1169
        };
1170
1
        let digest = Some(self.action_digest.into());
1171
1172
1
        let metadata = ExecuteOperationMetadata {
1173
1
            stage,
1174
1
            action_digest: digest,
1175
1
            // TODO(aaronmondal) We should support stderr/stdout streaming.
1176
1
            stdout_stream_name: String::default(),
1177
1
            stderr_stream_name: String::default(),
1178
1
            partial_execution_metadata: None,
1179
1
        };
1180
1181
1
        Operation {
1182
1
            name,
1183
1
            metadata: Some(to_any(&metadata)),
1184
1
            done: result.is_some(),
1185
1
            result,
1186
1
        }
1187
1
    }
1188
}