1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
// Closing Steps — what a chain does once one of its nodes has failed.
//
// The first failure is the chain's verdict. Every later node is skipped
// (no events) except those marked `always`, which run in order like
// `finally`: the first gets the files the failed step handed over, each
// next one gets the previous one's output. A closing step that fails joins
// the verdict after the first failure and hands over its own files. A run
// that passed its time limit runs no closing step: nothing starts past it.
use crate::errors::BntoError;
use crate::pipeline::PipelineFile;
use super::leaf_output::{node_output, pipeline_files};
use super::{ChainPosition, PipelineContext, PipelineNodeRef, execute_node};
/// A chain after one of its nodes failed.
pub(super) struct ChainFailure {
/// The node whose failure is the verdict (a container's id when the
/// failure was inside it).
pub(super) node_id: String,
/// The first failure, plus any closing step's that failed after it.
error: BntoError,
/// What the next closing step receives.
files: Vec<PipelineFile>,
}
impl ChainFailure {
/// The chain's first failure, at `node_id`, keeping the files it handed over.
pub(super) fn new(node_id: &str, error: BntoError) -> Self {
let (error, files) = split(error);
Self {
node_id: node_id.to_string(),
error,
files,
}
}
/// Run each `always` node of `rest` (placed by `position`) in order,
/// skipping the others. Closing steps record their output metadata in
/// `ctx`, so later ones can read it.
pub(super) fn close<F: Fn() -> u64 + Copy>(
mut self,
ctx: &mut PipelineContext<F>,
rest: &[PipelineNodeRef],
file_offset: usize,
position: ChainPosition,
) -> Self {
if self.error.is_timeout() {
return self;
}
for (i, node) in rest.iter().enumerate().filter(|(_, n)| n.always) {
let index = position.first_index + i;
self = self.run_step(ctx, node, index, position.total_nodes, file_offset);
}
self
}
/// Run `node` as a closing step on `files` (a graph step's fan-in)
/// rather than on what the previous closing step handed over.
pub(super) fn close_with<F: Fn() -> u64 + Copy>(
mut self,
ctx: &mut PipelineContext<F>,
node: PipelineNodeRef,
files: Vec<PipelineFile>,
position: ChainPosition,
file_offset: usize,
) -> Self {
self.files = files;
if self.error.is_timeout() {
return self;
}
let (index, total) = (position.first_index, position.total_nodes);
self.run_step(ctx, node, index, total, file_offset)
}
/// The files handed over so far, leaving none.
pub(super) fn take_files(&mut self) -> Vec<PipelineFile> {
std::mem::take(&mut self.files)
}
/// The same verdict, handing over `files` instead.
pub(super) fn handing_over(mut self, files: Vec<PipelineFile>) -> Self {
self.files = files;
self
}
/// Run one closing step on the files handed over so far.
fn run_step<F: Fn() -> u64 + Copy>(
mut self,
ctx: &mut PipelineContext<F>,
node: PipelineNodeRef,
index: usize,
total: usize,
file_offset: usize,
) -> Self {
let files = std::mem::take(&mut self.files);
match execute_node(ctx, node, files, index, total, file_offset) {
Ok(result) => {
let count_keys = ctx.count_keys;
ctx.node_outputs.add(result.node_outputs, count_keys);
self.files = result.output_files;
self
}
Err(error) => self.closing_failed(&node.id, error),
}
}
/// Join a closing step's failure to the verdict, after the first one.
fn closing_failed(self, node_id: &str, error: BntoError) -> Self {
let (error, files) = split(error);
let error = BntoError::ClosingStepFailed {
first: Box::new(self.error),
node_id: node_id.to_string(),
error: Box::new(error),
};
Self {
node_id: self.node_id,
error,
files,
}
}
/// The verdict, still handing over the chain's last files, for the
/// closing steps of an enclosing chain.
pub(super) fn into_error(self) -> BntoError {
if self.files.is_empty() {
return self.error;
}
BntoError::with_output(self.error, node_output(self.files))
}
/// The verdict alone, once no closing step is left anywhere.
pub(super) fn into_verdict(self) -> BntoError {
self.error
}
}
/// An error and the pipeline files it handed over.
fn split(error: BntoError) -> (BntoError, Vec<PipelineFile>) {
let (error, output) = error.into_parts();
(error, output.map(pipeline_files).unwrap_or_default())
}