termivar_scanner/distributed/
model.rs

1use super::limits::{MAX_IDENTIFIER_BYTES, MAX_TARGET_REF_BYTES, MAX_TASK_PHASES};
2use super::DistributedError;
3
4/// Task priority. Higher priorities are selected first.
5#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
6pub enum TaskPriority {
7    Low = 1,
8    Normal = 2,
9    High = 3,
10    Critical = 4,
11}
12
13/// Exact task lifecycle state.
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum TaskStatus {
16    Queued,
17    Leased,
18    Running,
19    Completed,
20    Failed,
21    Cancelled,
22    Expired,
23}
24
25impl TaskStatus {
26    pub fn as_str(self) -> &'static str {
27        match self {
28            Self::Queued => "queued",
29            Self::Leased => "leased",
30            Self::Running => "running",
31            Self::Completed => "completed",
32            Self::Failed => "failed",
33            Self::Cancelled => "cancelled",
34            Self::Expired => "expired",
35        }
36    }
37
38    pub(super) fn is_active(self) -> bool {
39        matches!(self, Self::Queued | Self::Leased | Self::Running)
40    }
41}
42
43/// Fresh task admission data. `target_ref` is opaque; this module never opens it.
44#[derive(Clone, PartialEq, Eq)]
45pub struct TaskSpec {
46    pub task_id: String,
47    pub scan_id: String,
48    pub target_ref: String,
49    pub phases: Vec<u8>,
50    pub priority: TaskPriority,
51}
52
53impl std::fmt::Debug for TaskSpec {
54    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
55        formatter
56            .debug_struct("TaskSpec")
57            .field("task_id", &self.task_id)
58            .field("scan_id", &self.scan_id)
59            .field("target_ref_bytes", &self.target_ref.len())
60            .field("phases", &self.phases)
61            .field("priority", &self.priority)
62            .finish()
63    }
64}
65
66impl TaskSpec {
67    pub fn new(
68        task_id: impl Into<String>,
69        scan_id: impl Into<String>,
70        target_ref: impl Into<String>,
71        phases: Vec<u8>,
72        priority: TaskPriority,
73    ) -> Self {
74        Self {
75            task_id: task_id.into(),
76            scan_id: scan_id.into(),
77            target_ref: target_ref.into(),
78            phases,
79            priority,
80        }
81    }
82}
83
84pub(super) fn validate_task_spec(spec: &TaskSpec) -> Result<(), DistributedError> {
85    validate_identifier(&spec.task_id, "task_id")?;
86    validate_identifier(&spec.scan_id, "scan_id")?;
87    if spec.target_ref.is_empty() {
88        return Err(DistributedError::InvalidTask {
89            reason: "target_ref is empty",
90        });
91    }
92    if spec.target_ref.len() > MAX_TARGET_REF_BYTES {
93        return Err(DistributedError::InvalidTask {
94            reason: "target_ref is too long",
95        });
96    }
97    if spec.phases.len() > MAX_TASK_PHASES {
98        return Err(DistributedError::InvalidTask {
99            reason: "too many phases",
100        });
101    }
102    Ok(())
103}
104
105pub(super) fn validate_identifier(
106    value: &str,
107    field: &'static str,
108) -> Result<(), DistributedError> {
109    if value.is_empty() {
110        return Err(DistributedError::InvalidTask {
111            reason: match field {
112                "task_id" => "task_id is empty",
113                "scan_id" => "scan_id is empty",
114                _ => "identifier is empty",
115            },
116        });
117    }
118    if value.len() > MAX_IDENTIFIER_BYTES {
119        return Err(DistributedError::InvalidTask {
120            reason: match field {
121                "task_id" => "task_id is too long",
122                "scan_id" => "scan_id is too long",
123                _ => "identifier is too long",
124            },
125        });
126    }
127    if !identifier_is_safe(value) {
128        return Err(DistributedError::InvalidTask {
129            reason: match field {
130                "task_id" => "task_id contains unsafe characters",
131                "scan_id" => "scan_id contains unsafe characters",
132                _ => "identifier contains unsafe characters",
133            },
134        });
135    }
136    Ok(())
137}
138
139fn identifier_is_safe(value: &str) -> bool {
140    value
141        .bytes()
142        .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':'))
143}
144
145pub(super) fn validate_task_command_id(value: &str) -> Result<(), DistributedError> {
146    if value.is_empty() || value.len() > MAX_IDENTIFIER_BYTES || !identifier_is_safe(value) {
147        return Err(DistributedError::InvalidTask {
148            reason: "task command identifier is invalid",
149        });
150    }
151    Ok(())
152}
153
154pub(super) fn validate_worker_command_id(value: &str) -> Result<(), DistributedError> {
155    if value.is_empty() || value.len() > MAX_IDENTIFIER_BYTES || !identifier_is_safe(value) {
156        return Err(DistributedError::InvalidWorker {
157            reason: "worker command identifier is invalid",
158        });
159    }
160    Ok(())
161}
162
163/// A successful command result and the revision after it.
164#[derive(Debug, Clone, PartialEq, Eq)]
165pub struct Transition<T> {
166    pub revision: u64,
167    pub value: T,
168}
169
170/// Bounded coordinator state summary.
171#[derive(Debug, Clone, Copy, PartialEq, Eq)]
172pub struct StateSnapshot {
173    pub revision: u64,
174    pub logical_time: u64,
175    pub task_records: usize,
176    pub active_tasks: usize,
177    pub queued_tasks: usize,
178    pub terminal_tasks: usize,
179    pub workers: usize,
180}
termivar_scanner/distributed/model.rs

termivar_scanner/distributed/
model.rs
