From 3bd29a5d44855e54c806bc2ab0ba7c9339e3cd2e Mon Sep 17 00:00:00 2001 From: worldisonline <242876766+worldisonline@users.noreply.github.com> Date: Mon, 20 Jul 2026 23:13:14 +0200 Subject: [PATCH] Add exec job store --- src/exec.rs | 761 ++++++++++++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 1 + 2 files changed, 762 insertions(+) create mode 100644 src/exec.rs diff --git a/src/exec.rs b/src/exec.rs new file mode 100644 index 00000000..1139b97c --- /dev/null +++ b/src/exec.rs @@ -0,0 +1,761 @@ +use std::{ + collections::{HashMap, VecDeque}, + sync::Mutex, + time::{Duration, Instant, SystemTime}, +}; + +use thiserror::Error; +use uuid::Uuid; + +pub const MAX_SOURCE_SIZE_BYTES: usize = 1024 * 1024; +pub const MAX_PENDING_JOBS: usize = 16; +pub const MAX_RETAINED_TERMINAL_JOBS: usize = 64; + +pub const CLAIM_TIMEOUT: Duration = Duration::from_secs(30); +pub const EXECUTION_TIMEOUT: Duration = Duration::from_secs(30); +pub const TERMINAL_RETENTION: Duration = Duration::from_secs(5 * 60); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ExecJobState { + Pending, + Claimed, + Succeeded, + Failed, + TimedOut, +} + +impl ExecJobState { + pub fn is_terminal(self) -> bool { + matches!(self, Self::Succeeded | Self::Failed | Self::TimedOut) + } +} + +#[derive(Debug, Clone, PartialEq)] +pub enum ExecValue { + Nil, + String(String), + Number(f64), + Boolean(bool), + Array(Vec), + Table(Vec<(String, ExecValue)>), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ExecLogLevel { + Print, + Warn, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ExecLog { + pub level: ExecLogLevel, + pub message: String, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct ExecJob { + pub id: Uuid, + pub script_name: String, + pub source: String, + pub state: ExecJobState, + pub created_at: SystemTime, + pub claim_deadline: Instant, + pub execution_deadline: Option, + pub result: Option, + pub logs: Option>, + pub runtime_error: Option, + pub traceback: Option, + + terminal_at: Option, +} + +impl ExecJob { + pub fn is_terminal(&self) -> bool { + self.state.is_terminal() + } +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum ExecJobStoreError { + #[error("exec source is {size} bytes, exceeding the {limit}-byte limit")] + SourceTooLarge { size: usize, limit: usize }, + + #[error("exec pending queue is full (limit: {limit})")] + PendingQueueFull { limit: usize }, + + #[error("exec job {id} does not exist")] + UnknownJob { id: Uuid }, + + #[error("exec job {id} cannot be completed from state {state:?}")] + JobNotClaimed { id: Uuid, state: ExecJobState }, + + #[error("exec job {id} was already completed with state {state:?}")] + DuplicateCompletion { id: Uuid, state: ExecJobState }, +} + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct ExecCleanupStats { + pub timed_out_pending: usize, + pub timed_out_claimed: usize, + pub removed_terminal: usize, +} + +#[derive(Debug)] +pub struct ExecJobStore { + inner: Mutex, + claim_timeout: Duration, + execution_timeout: Duration, + terminal_retention: Duration, +} + +#[derive(Debug, Default)] +struct ExecJobStoreInner { + jobs: HashMap, + pending_jobs: VecDeque, + claimed_job: Option, + terminal_jobs: VecDeque, +} + +impl ExecJobStore { + pub fn new() -> Self { + Self::with_durations(CLAIM_TIMEOUT, EXECUTION_TIMEOUT, TERMINAL_RETENTION) + } + + pub fn submit( + &self, + script_name: impl Into, + source: impl Into, + ) -> Result { + self.submit_at( + script_name.into(), + source.into(), + SystemTime::now(), + Instant::now(), + ) + } + + pub fn claim_next(&self) -> Option { + self.claim_next_at(Instant::now()) + } + + pub fn complete_success( + &self, + id: Uuid, + result: Option, + logs: Option>, + ) -> Result { + self.complete_success_at(id, result, logs, Instant::now()) + } + + pub fn complete_failure( + &self, + id: Uuid, + runtime_error: Option, + traceback: Option, + logs: Option>, + ) -> Result { + self.complete_failure_at(id, runtime_error, traceback, logs, Instant::now()) + } + + pub fn get(&self, id: Uuid) -> Result { + let inner = self.inner.lock().unwrap(); + + inner + .jobs + .get(&id) + .cloned() + .ok_or(ExecJobStoreError::UnknownJob { id }) + } + + pub fn cleanup_expired(&self) -> ExecCleanupStats { + self.cleanup_expired_at(Instant::now()) + } + + fn with_durations( + claim_timeout: Duration, + execution_timeout: Duration, + terminal_retention: Duration, + ) -> Self { + Self { + inner: Mutex::new(ExecJobStoreInner::default()), + claim_timeout, + execution_timeout, + terminal_retention, + } + } + + fn submit_at( + &self, + script_name: String, + source: String, + created_at: SystemTime, + now: Instant, + ) -> Result { + let source_size = source.len(); + if source_size > MAX_SOURCE_SIZE_BYTES { + return Err(ExecJobStoreError::SourceTooLarge { + size: source_size, + limit: MAX_SOURCE_SIZE_BYTES, + }); + } + + let mut inner = self.inner.lock().unwrap(); + if inner.pending_jobs.len() >= MAX_PENDING_JOBS { + return Err(ExecJobStoreError::PendingQueueFull { + limit: MAX_PENDING_JOBS, + }); + } + + let id = loop { + let candidate = Uuid::new_v4(); + if !inner.jobs.contains_key(&candidate) { + break candidate; + } + }; + + let job = ExecJob { + id, + script_name, + source, + state: ExecJobState::Pending, + created_at, + claim_deadline: now + self.claim_timeout, + execution_deadline: None, + result: None, + logs: None, + runtime_error: None, + traceback: None, + terminal_at: None, + }; + + inner.jobs.insert(id, job.clone()); + inner.pending_jobs.push_back(id); + + Ok(job) + } + + fn claim_next_at(&self, now: Instant) -> Option { + let mut inner = self.inner.lock().unwrap(); + if inner.claimed_job.is_some() { + return None; + } + + while let Some(id) = inner.pending_jobs.pop_front() { + let claimed_job = { + let Some(job) = inner.jobs.get_mut(&id) else { + continue; + }; + + if job.state != ExecJobState::Pending { + continue; + } + + job.state = ExecJobState::Claimed; + job.execution_deadline = Some(now + self.execution_timeout); + job.clone() + }; + + inner.claimed_job = Some(id); + return Some(claimed_job); + } + + None + } + + fn complete_success_at( + &self, + id: Uuid, + result: Option, + logs: Option>, + now: Instant, + ) -> Result { + let mut inner = self.inner.lock().unwrap(); + ensure_job_can_complete(&inner, id)?; + + { + let job = inner.jobs.get_mut(&id).unwrap(); + job.state = ExecJobState::Succeeded; + job.source.clear(); + job.result = result; + job.logs = logs; + job.runtime_error = None; + job.traceback = None; + job.terminal_at = Some(now); + } + + inner.claimed_job = None; + track_terminal_job(&mut inner, id); + + Ok(inner.jobs.get(&id).unwrap().clone()) + } + + fn complete_failure_at( + &self, + id: Uuid, + runtime_error: Option, + traceback: Option, + logs: Option>, + now: Instant, + ) -> Result { + let mut inner = self.inner.lock().unwrap(); + ensure_job_can_complete(&inner, id)?; + + { + let job = inner.jobs.get_mut(&id).unwrap(); + job.state = ExecJobState::Failed; + job.source.clear(); + job.result = None; + job.logs = logs; + job.runtime_error = runtime_error; + job.traceback = traceback; + job.terminal_at = Some(now); + } + + inner.claimed_job = None; + track_terminal_job(&mut inner, id); + + Ok(inner.jobs.get(&id).unwrap().clone()) + } + + fn cleanup_expired_at(&self, now: Instant) -> ExecCleanupStats { + let mut inner = self.inner.lock().unwrap(); + let mut stats = ExecCleanupStats::default(); + + let pending_count = inner.pending_jobs.len(); + for _ in 0..pending_count { + let id = inner.pending_jobs.pop_front().unwrap(); + let is_expired = inner + .jobs + .get(&id) + .is_some_and(|job| job.state == ExecJobState::Pending && now >= job.claim_deadline); + + if is_expired { + stats.removed_terminal += mark_job_timed_out(&mut inner, id, now); + stats.timed_out_pending += 1; + } else if inner.jobs.contains_key(&id) { + inner.pending_jobs.push_back(id); + } + } + + if let Some(id) = inner.claimed_job { + let is_expired = inner.jobs.get(&id).is_some_and(|job| { + job.state == ExecJobState::Claimed + && job + .execution_deadline + .is_some_and(|deadline| now >= deadline) + }); + + if is_expired { + stats.removed_terminal += mark_job_timed_out(&mut inner, id, now); + inner.claimed_job = None; + stats.timed_out_claimed += 1; + } + } + + while let Some(id) = inner.terminal_jobs.front().copied() { + let should_remove = match inner.jobs.get(&id) { + Some(job) => job.terminal_at.is_some_and(|terminal_at| { + now.saturating_duration_since(terminal_at) >= self.terminal_retention + }), + None => true, + }; + + if !should_remove { + break; + } + + inner.terminal_jobs.pop_front(); + if inner.jobs.remove(&id).is_some() { + stats.removed_terminal += 1; + } + } + + stats.removed_terminal += enforce_terminal_limit(&mut inner); + stats + } +} + +impl Default for ExecJobStore { + fn default() -> Self { + Self::new() + } +} + +fn ensure_job_can_complete(inner: &ExecJobStoreInner, id: Uuid) -> Result<(), ExecJobStoreError> { + let job = inner + .jobs + .get(&id) + .ok_or(ExecJobStoreError::UnknownJob { id })?; + + match job.state { + ExecJobState::Claimed => Ok(()), + state if state.is_terminal() => Err(ExecJobStoreError::DuplicateCompletion { id, state }), + state => Err(ExecJobStoreError::JobNotClaimed { id, state }), + } +} + +fn mark_job_timed_out(inner: &mut ExecJobStoreInner, id: Uuid, now: Instant) -> usize { + let job = inner + .jobs + .get_mut(&id) + .expect("queued exec job should exist in the job map"); + + job.state = ExecJobState::TimedOut; + job.source.clear(); + job.result = None; + job.logs = None; + job.runtime_error = None; + job.traceback = None; + job.terminal_at = Some(now); + + track_terminal_job(inner, id) +} + +fn track_terminal_job(inner: &mut ExecJobStoreInner, id: Uuid) -> usize { + inner.terminal_jobs.push_back(id); + enforce_terminal_limit(inner) +} + +fn enforce_terminal_limit(inner: &mut ExecJobStoreInner) -> usize { + let mut removed = 0; + + while inner.terminal_jobs.len() > MAX_RETAINED_TERMINAL_JOBS { + let id = inner.terminal_jobs.pop_front().unwrap(); + if inner.jobs.remove(&id).is_some() { + removed += 1; + } + } + + removed +} + +#[cfg(test)] +mod test { + use std::sync::{Arc, Barrier}; + + use super::*; + + const TEST_CLAIM_TIMEOUT: Duration = Duration::from_secs(10); + const TEST_EXECUTION_TIMEOUT: Duration = Duration::from_secs(20); + const TEST_TERMINAL_RETENTION: Duration = Duration::from_secs(30); + + fn test_store() -> ExecJobStore { + ExecJobStore::with_durations( + TEST_CLAIM_TIMEOUT, + TEST_EXECUTION_TIMEOUT, + TEST_TERMINAL_RETENTION, + ) + } + + fn submit_at(store: &ExecJobStore, name: &str, source: &str, now: Instant) -> ExecJob { + store + .submit_at( + name.to_owned(), + source.to_owned(), + SystemTime::UNIX_EPOCH, + now, + ) + .unwrap() + } + + #[test] + fn submit_adds_a_pending_job() { + let store = test_store(); + let now = Instant::now(); + let job = submit_at(&store, "hello.lua", "return 'hello'", now); + + assert_ne!(job.id, Uuid::nil()); + assert_eq!(job.script_name, "hello.lua"); + assert_eq!(job.source, "return 'hello'"); + assert_eq!(job.state, ExecJobState::Pending); + assert_eq!(job.created_at, SystemTime::UNIX_EPOCH); + assert_eq!(job.claim_deadline, now + TEST_CLAIM_TIMEOUT); + assert_eq!(job.execution_deadline, None); + assert_eq!(job.result, None); + assert_eq!(job.logs, None); + assert_eq!(job.runtime_error, None); + assert_eq!(job.traceback, None); + assert_eq!(store.get(job.id).unwrap(), job); + } + + #[test] + fn claim_returns_the_oldest_pending_job() { + let store = test_store(); + let now = Instant::now(); + let first = submit_at(&store, "first.lua", "return 1", now); + submit_at(&store, "second.lua", "return 2", now); + + let claimed = store.claim_next_at(now + Duration::from_secs(1)).unwrap(); + + assert_eq!(claimed.id, first.id); + assert_eq!(claimed.state, ExecJobState::Claimed); + assert_eq!( + claimed.execution_deadline, + Some(now + Duration::from_secs(1) + TEST_EXECUTION_TIMEOUT) + ); + assert_eq!(store.get(first.id).unwrap(), claimed); + } + + #[test] + fn claim_prevents_a_second_active_claim() { + let store = test_store(); + let now = Instant::now(); + let first = submit_at(&store, "first.lua", "return 1", now); + let second = submit_at(&store, "second.lua", "return 2", now); + + assert_eq!(store.claim_next_at(now).unwrap().id, first.id); + assert_eq!(store.claim_next_at(now), None); + assert_eq!(store.get(second.id).unwrap().state, ExecJobState::Pending); + + store + .complete_success_at(first.id, None, None, now) + .unwrap(); + assert_eq!(store.claim_next_at(now).unwrap().id, second.id); + } + + #[test] + fn concurrent_claims_only_return_one_job() { + let store = Arc::new(test_store()); + let now = Instant::now(); + let job = submit_at(&store, "atomic.lua", "return nil", now); + let start = Arc::new(Barrier::new(9)); + let mut workers = Vec::new(); + + for _ in 0..8 { + let store = Arc::clone(&store); + let start = Arc::clone(&start); + workers.push(std::thread::spawn(move || { + start.wait(); + store.claim_next_at(now) + })); + } + + start.wait(); + let claims: Vec<_> = workers + .into_iter() + .filter_map(|worker| worker.join().unwrap()) + .collect(); + + assert_eq!(claims.len(), 1); + assert_eq!(claims[0].id, job.id); + } + + #[test] + fn complete_success_stores_result_and_logs() { + let store = test_store(); + let now = Instant::now(); + let job = submit_at(&store, "success.lua", "return true", now); + store.claim_next_at(now).unwrap(); + let logs = vec![ExecLog { + level: ExecLogLevel::Print, + message: "done".to_owned(), + }]; + + let completed = store + .complete_success_at( + job.id, + Some(ExecValue::Boolean(true)), + Some(logs.clone()), + now, + ) + .unwrap(); + + assert_eq!(completed.state, ExecJobState::Succeeded); + assert!(completed.source.is_empty()); + assert_eq!(completed.result, Some(ExecValue::Boolean(true))); + assert_eq!(completed.logs, Some(logs)); + assert_eq!(completed.runtime_error, None); + assert_eq!(completed.traceback, None); + } + + #[test] + fn complete_failure_stores_error_traceback_and_logs() { + let store = test_store(); + let now = Instant::now(); + let job = submit_at(&store, "failure.lua", "error('oh no')", now); + store.claim_next_at(now).unwrap(); + let logs = vec![ExecLog { + level: ExecLogLevel::Warn, + message: "before failure".to_owned(), + }]; + + let completed = store + .complete_failure_at( + job.id, + Some("oh no".to_owned()), + Some("traceback".to_owned()), + Some(logs.clone()), + now, + ) + .unwrap(); + + assert_eq!(completed.state, ExecJobState::Failed); + assert!(completed.source.is_empty()); + assert_eq!(completed.result, None); + assert_eq!(completed.logs, Some(logs)); + assert_eq!(completed.runtime_error.as_deref(), Some("oh no")); + assert_eq!(completed.traceback.as_deref(), Some("traceback")); + } + + #[test] + fn duplicate_completion_is_rejected() { + let store = test_store(); + let now = Instant::now(); + let job = submit_at(&store, "duplicate.lua", "return nil", now); + store.claim_next_at(now).unwrap(); + store + .complete_success_at(job.id, Some(ExecValue::Nil), None, now) + .unwrap(); + + assert_eq!( + store.complete_failure_at(job.id, None, None, None, now), + Err(ExecJobStoreError::DuplicateCompletion { + id: job.id, + state: ExecJobState::Succeeded, + }) + ); + } + + #[test] + fn unknown_job_is_rejected() { + let store = test_store(); + let unknown_id = Uuid::new_v4(); + + assert_eq!( + store.get(unknown_id), + Err(ExecJobStoreError::UnknownJob { id: unknown_id }) + ); + assert_eq!( + store.complete_success(unknown_id, None, None), + Err(ExecJobStoreError::UnknownJob { id: unknown_id }) + ); + } + + #[test] + fn cleanup_times_out_pending_jobs() { + let store = test_store(); + let now = Instant::now(); + let job = submit_at(&store, "pending.lua", "return nil", now); + + let stats = store.cleanup_expired_at(now + TEST_CLAIM_TIMEOUT); + + assert_eq!(stats.timed_out_pending, 1); + assert_eq!(stats.timed_out_claimed, 0); + assert_eq!(store.get(job.id).unwrap().state, ExecJobState::TimedOut); + assert_eq!(store.claim_next_at(now + TEST_CLAIM_TIMEOUT), None); + } + + #[test] + fn cleanup_times_out_claimed_jobs() { + let store = test_store(); + let now = Instant::now(); + let job = submit_at(&store, "claimed.lua", "return nil", now); + let claim_time = now + Duration::from_secs(1); + store.claim_next_at(claim_time).unwrap(); + + let stats = store.cleanup_expired_at(claim_time + TEST_EXECUTION_TIMEOUT); + + assert_eq!(stats.timed_out_pending, 0); + assert_eq!(stats.timed_out_claimed, 1); + let timed_out = store.get(job.id).unwrap(); + assert_eq!(timed_out.state, ExecJobState::TimedOut); + assert_eq!( + timed_out.execution_deadline, + Some(claim_time + TEST_EXECUTION_TIMEOUT) + ); + } + + #[test] + fn cleanup_removes_old_terminal_jobs() { + let store = test_store(); + let now = Instant::now(); + let job = submit_at(&store, "old.lua", "return nil", now); + store.claim_next_at(now).unwrap(); + store.complete_success_at(job.id, None, None, now).unwrap(); + + assert_eq!( + store.cleanup_expired_at(now + TEST_TERMINAL_RETENTION - Duration::from_nanos(1)), + ExecCleanupStats::default() + ); + assert!(store.get(job.id).is_ok()); + + let stats = store.cleanup_expired_at(now + TEST_TERMINAL_RETENTION); + assert_eq!(stats.removed_terminal, 1); + assert_eq!( + store.get(job.id), + Err(ExecJobStoreError::UnknownJob { id: job.id }) + ); + } + + #[test] + fn pending_queue_limit_is_enforced() { + let store = test_store(); + let now = Instant::now(); + + for index in 0..MAX_PENDING_JOBS { + submit_at(&store, &format!("{index}.lua"), "return nil", now); + } + + assert_eq!( + store.submit_at( + "overflow.lua".to_owned(), + "return nil".to_owned(), + SystemTime::UNIX_EPOCH, + now, + ), + Err(ExecJobStoreError::PendingQueueFull { + limit: MAX_PENDING_JOBS, + }) + ); + + store.claim_next_at(now).unwrap(); + assert!(store + .submit_at( + "replacement.lua".to_owned(), + "return nil".to_owned(), + SystemTime::UNIX_EPOCH, + now, + ) + .is_ok()); + } + + #[test] + fn source_size_limit_is_enforced() { + let store = test_store(); + + assert!(store + .submit("maximum.lua", "x".repeat(MAX_SOURCE_SIZE_BYTES)) + .is_ok()); + assert_eq!( + store.submit("too-large.lua", "x".repeat(MAX_SOURCE_SIZE_BYTES + 1)), + Err(ExecJobStoreError::SourceTooLarge { + size: MAX_SOURCE_SIZE_BYTES + 1, + limit: MAX_SOURCE_SIZE_BYTES, + }) + ); + } + + #[test] + fn terminal_job_limit_evicts_the_oldest_job() { + let store = test_store(); + let now = Instant::now(); + let mut completed_ids = Vec::new(); + + for index in 0..=MAX_RETAINED_TERMINAL_JOBS { + let job = submit_at(&store, &format!("{index}.lua"), "return nil", now); + store.claim_next_at(now).unwrap(); + store.complete_success_at(job.id, None, None, now).unwrap(); + completed_ids.push(job.id); + } + + assert_eq!( + store.get(completed_ids[0]), + Err(ExecJobStoreError::UnknownJob { + id: completed_ids[0], + }) + ); + for id in &completed_ids[1..] { + assert!(store.get(*id).is_ok()); + } + } +} diff --git a/src/lib.rs b/src/lib.rs index f1947eaa..ca574581 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,6 +3,7 @@ #![recursion_limit = "1024"] pub mod cli; +pub mod exec; #[cfg(test)] mod tree_view;