cortex/concerns/
mark_rerun.rs

1use diesel::result::Error;
2use diesel::*;
3
4use crate::helpers::TaskStatus;
5use crate::models::{LogError, LogFatal, LogInfo, LogInvalid, LogWarning, Task};
6use crate::schema::tasks;
7
8/// Task reruns by a variety of selector granularity
9pub trait MarkRerun {
10  /// Most-specific rerun query, via both category and what filter
11  fn mark_rerun_by_what(
12    mark: i32,
13    corpus_id: i32,
14    service_id: i32,
15    rerun_category: &str,
16    rerun_what: &str,
17    connection: &mut PgConnection,
18  ) -> Result<usize, Error>;
19  /// Mid-specificity `category`-filtered reruns
20  fn mark_rerun_by_category(
21    mark: i32,
22    corpus_id: i32,
23    service_id: i32,
24    rerun_category: &str,
25    connection: &mut PgConnection,
26  ) -> Result<usize, Error>;
27}
28
29/// Info level reruns
30impl MarkRerun for LogInfo {
31  fn mark_rerun_by_what(
32    mark: i32,
33    corpus_id: i32,
34    service_id: i32,
35    rerun_category: &str,
36    rerun_what: &str,
37    connection: &mut PgConnection,
38  ) -> Result<usize, Error> {
39    use crate::schema::log_infos::dsl::{category, log_infos, task_id, what};
40    let task_ids_to_rerun = log_infos
41      .filter(category.eq(rerun_category))
42      .filter(what.eq(rerun_what))
43      .select(task_id)
44      .distinct();
45
46    update(tasks::table)
47      .filter(tasks::corpus_id.eq(&corpus_id))
48      .filter(tasks::service_id.eq(&service_id))
49      .filter(tasks::id.eq_any(task_ids_to_rerun))
50      .set(tasks::status.eq(mark))
51      .execute(connection)
52  }
53
54  fn mark_rerun_by_category(
55    mark: i32,
56    corpus_id: i32,
57    service_id: i32,
58    rerun_category: &str,
59    connection: &mut PgConnection,
60  ) -> Result<usize, Error> {
61    use crate::schema::log_infos::dsl::{category, log_infos, task_id};
62    let task_ids_to_rerun = log_infos
63      .filter(category.eq(rerun_category))
64      .select(task_id)
65      .distinct();
66
67    update(tasks::table)
68      .filter(tasks::corpus_id.eq(&corpus_id))
69      .filter(tasks::service_id.eq(&service_id))
70      .filter(tasks::id.eq_any(task_ids_to_rerun))
71      .set(tasks::status.eq(mark))
72      .execute(connection)
73  }
74}
75
76/// Warning level reruns
77impl MarkRerun for LogWarning {
78  fn mark_rerun_by_what(
79    mark: i32,
80    corpus_id: i32,
81    service_id: i32,
82    rerun_category: &str,
83    rerun_what: &str,
84    connection: &mut PgConnection,
85  ) -> Result<usize, Error> {
86    use crate::schema::log_warnings::dsl::{category, log_warnings, task_id, what};
87    let task_ids_to_rerun = log_warnings
88      .filter(category.eq(rerun_category))
89      .filter(what.eq(rerun_what))
90      .select(task_id)
91      .distinct();
92
93    update(tasks::table)
94      .filter(tasks::corpus_id.eq(&corpus_id))
95      .filter(tasks::service_id.eq(&service_id))
96      .filter(tasks::id.eq_any(task_ids_to_rerun))
97      .set(tasks::status.eq(mark))
98      .execute(connection)
99  }
100
101  fn mark_rerun_by_category(
102    mark: i32,
103    corpus_id: i32,
104    service_id: i32,
105    rerun_category: &str,
106    connection: &mut PgConnection,
107  ) -> Result<usize, Error> {
108    use crate::schema::log_warnings::dsl::{category, log_warnings, task_id};
109    let task_ids_to_rerun = log_warnings
110      .filter(category.eq(rerun_category))
111      .select(task_id)
112      .distinct();
113
114    update(tasks::table)
115      .filter(tasks::corpus_id.eq(&corpus_id))
116      .filter(tasks::service_id.eq(&service_id))
117      .filter(tasks::id.eq_any(task_ids_to_rerun))
118      .set(tasks::status.eq(mark))
119      .execute(connection)
120  }
121}
122
123/// Error level reruns
124impl MarkRerun for LogError {
125  fn mark_rerun_by_what(
126    mark: i32,
127    corpus_id: i32,
128    service_id: i32,
129    rerun_category: &str,
130    rerun_what: &str,
131    connection: &mut PgConnection,
132  ) -> Result<usize, Error> {
133    use crate::schema::log_errors::dsl::{category, log_errors, task_id, what};
134    let task_ids_to_rerun = log_errors
135      .filter(category.eq(rerun_category))
136      .filter(what.eq(rerun_what))
137      .select(task_id)
138      .distinct();
139
140    update(tasks::table)
141      .filter(tasks::corpus_id.eq(&corpus_id))
142      .filter(tasks::service_id.eq(&service_id))
143      .filter(tasks::id.eq_any(task_ids_to_rerun))
144      .set(tasks::status.eq(mark))
145      .execute(connection)
146  }
147
148  fn mark_rerun_by_category(
149    mark: i32,
150    corpus_id: i32,
151    service_id: i32,
152    rerun_category: &str,
153    connection: &mut PgConnection,
154  ) -> Result<usize, Error> {
155    use crate::schema::log_errors::dsl::{category, log_errors, task_id};
156    let task_ids_to_rerun = log_errors
157      .filter(category.eq(rerun_category))
158      .select(task_id)
159      .distinct();
160
161    update(tasks::table)
162      .filter(tasks::corpus_id.eq(&corpus_id))
163      .filter(tasks::service_id.eq(&service_id))
164      .filter(tasks::id.eq_any(task_ids_to_rerun))
165      .set(tasks::status.eq(mark))
166      .execute(connection)
167  }
168}
169/// Fatal level reruns
170impl MarkRerun for LogFatal {
171  fn mark_rerun_by_what(
172    mark: i32,
173    corpus_id: i32,
174    service_id: i32,
175    rerun_category: &str,
176    rerun_what: &str,
177    connection: &mut PgConnection,
178  ) -> Result<usize, Error> {
179    use crate::schema::log_fatals::dsl::{category, log_fatals, task_id, what};
180    let task_ids_to_rerun = log_fatals
181      .filter(category.eq(rerun_category))
182      .filter(what.eq(rerun_what))
183      .select(task_id)
184      .distinct();
185
186    update(tasks::table)
187      .filter(tasks::corpus_id.eq(&corpus_id))
188      .filter(tasks::service_id.eq(&service_id))
189      .filter(tasks::id.eq_any(task_ids_to_rerun))
190      .set(tasks::status.eq(mark))
191      .execute(connection)
192  }
193
194  fn mark_rerun_by_category(
195    mark: i32,
196    corpus_id: i32,
197    service_id: i32,
198    rerun_category: &str,
199    connection: &mut PgConnection,
200  ) -> Result<usize, Error> {
201    use crate::schema::log_fatals::dsl::{category, log_fatals, task_id};
202    use diesel::sql_types::BigInt;
203    if rerun_category == "no_messages" {
204      let no_messages_query_string = "SELECT * FROM tasks t WHERE ".to_string()
205        + "service_id=$1 and corpus_id=$2 and status=$3 and "
206        + "NOT EXISTS (SELECT null FROM log_fatals where log_fatals.task_id=t.id)";
207      let tasks_to_rerun: Vec<Task> = sql_query(no_messages_query_string)
208        .bind::<BigInt, i64>(i64::from(service_id))
209        .bind::<BigInt, i64>(i64::from(corpus_id))
210        .bind::<BigInt, i64>(i64::from(TaskStatus::Fatal.raw()))
211        .get_results(connection)
212        .unwrap_or_default();
213      let task_ids_to_rerun: Vec<i64> = tasks_to_rerun.iter().map(|t| t.id).collect();
214      update(tasks::table)
215        .filter(tasks::corpus_id.eq(&corpus_id))
216        .filter(tasks::service_id.eq(&service_id))
217        .filter(tasks::id.eq_any(task_ids_to_rerun))
218        .set(tasks::status.eq(mark))
219        .execute(connection)
220    } else {
221      let task_ids_to_rerun = log_fatals
222        .filter(category.eq(rerun_category))
223        .select(task_id)
224        .distinct();
225
226      update(tasks::table)
227        .filter(tasks::corpus_id.eq(&corpus_id))
228        .filter(tasks::service_id.eq(&service_id))
229        .filter(tasks::id.eq_any(task_ids_to_rerun))
230        .set(tasks::status.eq(mark))
231        .execute(connection)
232    }
233  }
234}
235
236/// Invalid level reruns
237impl MarkRerun for LogInvalid {
238  fn mark_rerun_by_what(
239    mark: i32,
240    corpus_id: i32,
241    service_id: i32,
242    rerun_category: &str,
243    rerun_what: &str,
244    connection: &mut PgConnection,
245  ) -> Result<usize, Error> {
246    use crate::schema::log_invalids::dsl::{category, log_invalids, task_id, what};
247    let task_ids_to_rerun = log_invalids
248      .filter(category.eq(rerun_category))
249      .filter(what.eq(rerun_what))
250      .select(task_id)
251      .distinct();
252
253    update(tasks::table)
254      .filter(tasks::corpus_id.eq(&corpus_id))
255      .filter(tasks::service_id.eq(&service_id))
256      .filter(tasks::id.eq_any(task_ids_to_rerun))
257      .set(tasks::status.eq(mark))
258      .execute(connection)
259  }
260
261  fn mark_rerun_by_category(
262    mark: i32,
263    corpus_id: i32,
264    service_id: i32,
265    rerun_category: &str,
266    connection: &mut PgConnection,
267  ) -> Result<usize, Error> {
268    use crate::schema::log_invalids::dsl::{category, log_invalids, task_id};
269    let task_ids_to_rerun = log_invalids
270      .filter(category.eq(rerun_category))
271      .select(task_id)
272      .distinct();
273
274    update(tasks::table)
275      .filter(tasks::corpus_id.eq(&corpus_id))
276      .filter(tasks::service_id.eq(&service_id))
277      .filter(tasks::id.eq_any(task_ids_to_rerun))
278      .set(tasks::status.eq(mark))
279      .execute(connection)
280  }
281}
cortex/concerns/mark_rerun.rs

cortex/concerns/
mark_rerun.rs
