cortex/models/
tasks.rs

1#![allow(clippy::implicit_hasher, clippy::extra_unused_lifetimes)]
2use diesel::result::Error;
3use diesel::*;
4
5use super::{Corpus, Service};
6use crate::concerns::{CortexDeletable, CortexInsertable};
7use crate::helpers::TaskStatus;
8use crate::schema::tasks;
9
10// Tasks
11
12#[derive(
13  Identifiable, Queryable, Associations, AsChangeset, Clone, Debug, PartialEq, Eq, QueryableByName,
14)]
15#[diesel(table_name = tasks)]
16#[diesel(belongs_to(Corpus, foreign_key = corpus_id))]
17#[diesel(belongs_to(Service, foreign_key = service_id))]
18/// A `CorTeX` task, for a given corpus-service pair
19pub struct Task {
20  /// task primary key, auto-incremented by postgresql
21  pub id: i64,
22  /// id of the service owning this task
23  pub service_id: i32,
24  /// id of the corpus hosting this task
25  pub corpus_id: i32,
26  /// current processing status of this task
27  pub status: i32,
28  /// entry path on the file system
29  pub entry: String,
30}
31
32#[derive(Insertable, Debug, Clone)]
33#[diesel(table_name = tasks)]
34/// A new task, to be inserted into `CorTeX`
35pub struct NewTask {
36  /// id of the service owning this task
37  pub service_id: i32,
38  /// id of the corpus hosting this task
39  pub corpus_id: i32,
40  /// current processing status of this task
41  pub status: i32,
42  /// entry path on the file system
43  pub entry: String,
44}
45
46impl CortexInsertable for NewTask {
47  fn create(&self, connection: &mut PgConnection) -> Result<usize, Error> {
48    insert_into(tasks::table).values(self).execute(connection)
49  }
50}
51
52impl CortexDeletable for Task {
53  fn delete_by(&self, connection: &mut PgConnection, field: &str) -> Result<usize, Error> {
54    match field {
55      "entry" => self.delete_by_entry(connection),
56      "service_id" => self.delete_by_service_id(connection),
57      "id" => self.delete_by_id(connection),
58      _ => Err(Error::QueryBuilderError(
59        format!("unknown Task model field: {field}").into(),
60      )),
61    }
62  }
63}
64impl Task {
65  /// Delete task by entry
66  pub fn delete_by_entry(&self, connection: &mut PgConnection) -> Result<usize, Error> {
67    use crate::schema::tasks::dsl::entry;
68    delete(tasks::table.filter(entry.eq(&self.entry))).execute(connection)
69  }
70
71  /// Delete all tasks matching this task's service id
72  pub fn delete_by_service_id(&self, connection: &mut PgConnection) -> Result<usize, Error> {
73    use crate::schema::tasks::dsl::service_id;
74    delete(tasks::table.filter(service_id.eq(&self.service_id))).execute(connection)
75  }
76
77  /// Delete task by id
78  pub fn delete_by_id(&self, connection: &mut PgConnection) -> Result<usize, Error> {
79    use crate::schema::tasks::dsl::id;
80    delete(tasks::table.filter(id.eq(self.id))).execute(connection)
81  }
82
83  /// Find task by id, error if none
84  pub fn find(taskid: i64, connection: &mut PgConnection) -> Result<Task, Error> {
85    tasks::table.find(taskid).first(connection)
86  }
87
88  /// Count of TODO tasks for **real conversion services** — the **pending-conversion backlog**
89  /// (work waiting to be dispatched to the fleet). The single operational "is the fleet keeping
90  /// up?" number, shared by `/metrics` (`cortex_tasks_todo`), the admin dashboard, and `cortex
91  /// status`. One count over `tasks`; bounded, and fast via the partial `todo_index` once a running
92  /// dispatcher drains the backlog. `0` on a query error (best-effort, matching the report paths).
93  ///
94  /// Excludes the magic infrastructure services (1=init, 2=import; real services are id > 2): their
95  /// rows are status-0 (TODO) **placeholders** that exist for other services to activate over and
96  /// never receive real conversion work, so counting them inflated the backlog with tasks that are
97  /// never truly pending.
98  pub fn count_todo(connection: &mut PgConnection) -> i64 {
99    tasks::table
100      .filter(tasks::status.eq(TaskStatus::TODO.raw()))
101      .filter(tasks::service_id.gt(2))
102      .count()
103      .get_result(connection)
104      .unwrap_or(0)
105  }
106
107  /// Find task by entry, error if none
108  pub fn find_by_entry(entry: &str, connection: &mut PgConnection) -> Result<Task, Error> {
109    tasks::table
110      .filter(tasks::entry.eq(entry))
111      .first(connection)
112  }
113
114  /// Find task by name-suffix of an entry, error if none
115  pub fn find_by_name(
116    name: &str,
117    corpus: &Corpus,
118    service: &Service,
119    connection: &mut PgConnection,
120  ) -> Result<Task, Error> {
121    use crate::schema::tasks::dsl::{corpus_id, service_id};
122    tasks::table
123      .filter(corpus_id.eq(corpus.id))
124      .filter(service_id.eq(service.id))
125      .filter(tasks::entry.like(&format!("%{name}.zip")))
126      .first(connection)
127  }
128}
129
130impl CortexDeletable for NewTask {
131  fn delete_by(&self, connection: &mut PgConnection, field: &str) -> Result<usize, Error> {
132    match field {
133      "entry" => self.delete_by_entry(connection),
134      "service_id" => self.delete_by_service_id(connection),
135      _ => Err(Error::QueryBuilderError(
136        format!("unknown Task model field: {field}").into(),
137      )),
138    }
139  }
140}
141
142impl NewTask {
143  fn delete_by_entry(&self, connection: &mut PgConnection) -> Result<usize, Error> {
144    use crate::schema::tasks::dsl::entry;
145    delete(tasks::table.filter(entry.eq(&self.entry))).execute(connection)
146  }
147  fn delete_by_service_id(&self, connection: &mut PgConnection) -> Result<usize, Error> {
148    use crate::schema::tasks::dsl::service_id;
149    delete(tasks::table.filter(service_id.eq(&self.service_id))).execute(connection)
150  }
151  /// Creates the task unless already present in the DB (entry conflict)
152  pub fn create_if_new(&self, connection: &mut PgConnection) -> Result<usize, Error> {
153    insert_into(tasks::table)
154      .values(self)
155      .on_conflict_do_nothing()
156      .execute(connection)
157  }
158}
cortex/models/tasks.rs

cortex/models/
tasks.rs
