termivar_scanner/
distributed.rs

1//! Bounded deterministic in-process scan coordination.
2//!
3//! ## Runtime scope
4//!
5//! - **Build:** opt-in via `distributed`.
6//! - **Execution:** no repository runtime caller (not on any default path).
7//! - **Default `termivar scan`:** no.
8//! - **Support:** experimental.
9//!
10//! This module is one revisioned, in-memory state machine. It provides no
11//! transport, authentication, durability, process recovery, or multi-node
12//! consensus. Callers supply a monotonic logical time and explicitly drive
13//! lease expiry and worker-loss recovery.
14//! Every command, including an idempotent terminal replay, must name the
15//! coordinator's current revision. Ownership tokens make a current-revision
16//! replay semantically idempotent; they do not bypass revision ordering.
17//! Tokens and receipts are deterministic logical CAS/idempotency fences within
18//! one caller-enforced coordinator epoch. They are not authentication material
19//! and are not cross-instance replay-resistant.
20//! Bounds cover retained data per instance; caller allocations, returned clones,
21//! instance count, and allocator exhaustion remain host-budgeted.
22
23mod coordinator;
24mod lease;
25mod limits;
26mod model;
27mod queue;
28mod recovery;
29mod results;
30mod worker;
31
32pub use coordinator::WorkerPool;
33pub use lease::{
34    CancellationOutcome, CompletionOutcome, CompletionReceipt, FailureOutcome, QueuedTaskFence,
35    ScanTask, StartOutcome, TaskLease, TaskOwnership,
36};
37pub use limits::{
38    DistributedLimits, MAX_ACTIVE_TASKS, MAX_AGGREGATE_ITEMS, MAX_HEARTBEAT_TIMEOUT_SECS,
39    MAX_IDENTIFIER_BYTES, MAX_LEASE_TTL_SECS, MAX_RESULTS, MAX_RESULT_BYTES, MAX_RETRIES,
40    MAX_TARGET_REF_BYTES, MAX_TASK_PHASES, MAX_TASK_RECORDS, MAX_TASK_TTL_SECS,
41    MAX_TOTAL_RESULT_BYTES, MAX_WORKERS, MAX_WORKER_CAPACITY, MAX_WORKER_TAGS,
42    UTILIZATION_BASIS_POINTS,
43};
44pub use model::{StateSnapshot, TaskPriority, TaskSpec, TaskStatus, Transition};
45pub use queue::TaskQueue;
46pub use recovery::RecoverySummary;
47pub use results::{AggregatedResult, ResultAggregator, ResultLimits, StoreResultOutcome};
48pub use worker::{WorkerNode, WorkerObservation, WorkerSpec, WorkerStatus, WorkerTag};
49
50#[cfg(test)]
51use std::collections::BTreeSet;
52use std::sync::{Arc, Mutex, MutexGuard};
53
54#[cfg(test)]
55use coordinator::{next_counter, next_u32};
56#[cfg(test)]
57use queue::QueueKey;
58use thiserror::Error;
59
60/// Typed state-machine failures. Every failure leaves state unchanged.
61#[derive(Debug, Clone, PartialEq, Eq, Error)]
62pub enum DistributedError {
63    #[error("invalid zero limit: {name}")]
64    InvalidLimit { name: &'static str },
65    #[error("invalid limit relationship: {reason}")]
66    InvalidLimitRelationship { reason: &'static str },
67    #[error("count limit {name}={actual} exceeds absolute maximum {maximum}")]
68    CountLimitExceedsMaximum {
69        name: &'static str,
70        actual: usize,
71        maximum: usize,
72    },
73    #[error("time limit {name}={actual} exceeds absolute maximum {maximum}")]
74    TimeLimitExceedsMaximum {
75        name: &'static str,
76        actual: u64,
77        maximum: u64,
78    },
79    #[error("retry limit {actual} exceeds absolute maximum {maximum}")]
80    RetryLimitExceedsMaximum { actual: u32, maximum: u32 },
81    #[error("invalid task: {reason}")]
82    InvalidTask { reason: &'static str },
83    #[error("invalid worker: {reason}")]
84    InvalidWorker { reason: &'static str },
85    #[error("revision conflict: expected {expected}, actual {actual}")]
86    RevisionConflict { expected: u64, actual: u64 },
87    #[error("logical time regressed from {current} to {proposed}")]
88    LogicalTimeRegression { current: u64, proposed: u64 },
89    #[error("monotonic counter exhausted: {counter}")]
90    CounterExhausted { counter: &'static str },
91    #[error("task already exists: {task_id}")]
92    TaskAlreadyExists { task_id: String },
93    #[error("task not found: {task_id}")]
94    TaskNotFound { task_id: String },
95    #[error("worker already exists: {worker_id}")]
96    WorkerAlreadyExists { worker_id: String },
97    #[error("worker not found: {worker_id}")]
98    WorkerNotFound { worker_id: String },
99    #[error("worker generation conflict: expected {expected}, actual {actual}")]
100    WorkerGenerationConflict { expected: u64, actual: u64 },
101    #[error("task record capacity reached: {limit}")]
102    TaskRecordCapacityReached { limit: usize },
103    #[error("active task capacity reached: {limit}")]
104    ActiveTaskCapacityReached { limit: usize },
105    #[error("queued task capacity reached: {limit}")]
106    QueuedTaskCapacityReached { limit: usize },
107    #[error("terminal reservation capacity reached: {limit}")]
108    TerminalCapacityReserved { limit: usize },
109    #[error("worker capacity reached: {limit}")]
110    WorkerCapacityReached { limit: usize },
111    #[error("no queued task is available")]
112    NoQueuedTask,
113    #[error("no eligible worker is available")]
114    NoAvailableWorker,
115    #[error("worker is not eligible: {worker_id}")]
116    WorkerUnavailable { worker_id: String },
117    #[error("worker is at capacity: {worker_id}")]
118    WorkerAtCapacity { worker_id: String },
119    #[error("task {task_id} is not queued (status: {status:?})")]
120    TaskNotQueued { task_id: String, status: TaskStatus },
121    #[error("invalid {operation} transition for task {task_id} from {status:?}")]
122    InvalidTransition {
123        task_id: String,
124        status: TaskStatus,
125        operation: &'static str,
126    },
127    #[error("stale or mismatched task ownership token: {task_id}")]
128    StaleOwnership { task_id: String },
129    #[error("lease expired for task {task_id}")]
130    LeaseExpired { task_id: String },
131    #[error("result already exists with different bytes: {task_id}")]
132    ConflictingResult { task_id: String },
133    #[error("result receipt does not match the occupied task ID: {task_id}")]
134    MismatchedResultReceipt { task_id: String },
135    #[error("result capacity reached: {limit}")]
136    ResultCapacityReached { limit: usize },
137    #[error("result size {actual} exceeds limit {limit}")]
138    ResultTooLarge { actual: usize, limit: usize },
139    #[error("retained result bytes {actual} exceed limit {limit}")]
140    TotalResultBytesExceeded { actual: usize, limit: usize },
141    #[error("aggregate request has {actual} items, limit is {limit}")]
142    AggregateItemLimitExceeded { actual: usize, limit: usize },
143    #[error("aggregate request repeats task {task_id}")]
144    DuplicateAggregateTask { task_id: String },
145    #[error("aggregate result is missing task {task_id}")]
146    MissingResult { task_id: String },
147    #[error("aggregate bytes {actual} exceed limit {limit}")]
148    AggregateBytesExceeded { actual: usize, limit: usize },
149    #[error("state invariant failed: {reason}")]
150    StateInvariant { reason: &'static str },
151    #[error("state lock is poisoned")]
152    StatePoisoned,
153}
154
155fn lock_state<T>(state: &Arc<Mutex<T>>) -> Result<MutexGuard<'_, T>, DistributedError> {
156    state.lock().map_err(|_| DistributedError::StatePoisoned)
157}
158
159#[cfg(test)]
160mod tests {
161    use super::*;
162
163    #[test]
164    fn integer_capacity_is_exact_at_basis_point_edges() {
165        let mut worker = WorkerNode {
166            worker_id: "worker".to_owned(),
167            generation: 1,
168            status: WorkerStatus::Healthy,
169            capacity: 10,
170            current_tasks: 0,
171            completed_tasks: 0,
172            last_heartbeat: 0,
173            cpu_basis_points: 5_000,
174            memory_basis_points: 0,
175            network_basis_points: 0,
176            tags: BTreeSet::new(),
177        };
178        assert_eq!(worker.effective_capacity(), 5);
179        worker.cpu_basis_points = 9_999;
180        assert_eq!(worker.effective_capacity(), 1);
181        worker.cpu_basis_points = 10_000;
182        assert_eq!(worker.effective_capacity(), 0);
183    }
184
185    #[test]
186    fn counter_overflow_is_typed() {
187        assert_eq!(
188            next_counter(u64::MAX, "test"),
189            Err(DistributedError::CounterExhausted { counter: "test" })
190        );
191        assert_eq!(
192            next_u32(u32::MAX, "test32"),
193            Err(DistributedError::CounterExhausted { counter: "test32" })
194        );
195    }
196
197    #[test]
198    fn invalid_limit_relationships_are_rejected() {
199        let limits = DistributedLimits {
200            max_task_records: 4,
201            max_active_tasks: 2,
202            max_queued_tasks: 3,
203            max_terminal_tasks: 2,
204            ..DistributedLimits::default()
205        };
206        assert_eq!(
207            WorkerPool::with_limits(limits).err(),
208            Some(DistributedError::InvalidLimitRelationship {
209                reason: "max_queued_tasks exceeds max_active_tasks"
210            })
211        );
212    }
213
214    #[test]
215    fn queue_key_orders_priority_then_ordinal_then_id() {
216        let mut keys = BTreeSet::new();
217        keys.insert(QueueKey::new(TaskPriority::Normal, 1, "z".to_owned()));
218        keys.insert(QueueKey::new(TaskPriority::Critical, 2, "b".to_owned()));
219        keys.insert(QueueKey::new(TaskPriority::Critical, 2, "a".to_owned()));
220        let ordered: Vec<String> = keys.into_iter().map(|key| key.task_id).collect();
221        assert_eq!(ordered, ["a", "b", "z"]);
222    }
223}
termivar_scanner/distributed.rs

termivar_scanner/
distributed.rs
