Add exec HTTP API

src/exec.rs

@@ -157,6 +157,16 @@ impl ExecJobStore {
         self.complete_failure_at(id, runtime_error, traceback, logs, Instant::now())
     }
 
+    pub fn complete_timeout(
+        &self,
+        id: Uuid,
+        runtime_error: Option<String>,
+        traceback: Option<String>,
+        logs: Option<Vec<ExecLog>>,
+    ) -> Result<ExecJob, ExecJobStoreError> {
+        self.complete_timeout_at(id, runtime_error, traceback, logs, Instant::now())
+    }
+
     pub fn get(&self, id: Uuid) -> Result<ExecJob, ExecJobStoreError> {
         let inner = self.inner.lock().unwrap();
 
@@ -317,6 +327,34 @@ impl ExecJobStore {
         Ok(inner.jobs.get(&id).unwrap().clone())
     }
 
+    fn complete_timeout_at(
+        &self,
+        id: Uuid,
+        runtime_error: Option<String>,
+        traceback: Option<String>,
+        logs: Option<Vec<ExecLog>>,
+        now: Instant,
+    ) -> Result<ExecJob, ExecJobStoreError> {
+        let mut inner = self.inner.lock().unwrap();
+        ensure_job_can_complete(&inner, id)?;
+
+        {
+            let job = inner.jobs.get_mut(&id).unwrap();
+            job.state = ExecJobState::TimedOut;
+            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();
@@ -597,6 +635,38 @@ mod test {
         assert_eq!(completed.traceback.as_deref(), Some("traceback"));
     }
 
+    #[test]
+    fn complete_timeout_stores_timeout_details() {
+        let store = test_store();
+        let now = Instant::now();
+        let job = submit_at(&store, "timeout.lua", "task.wait(60)", now);
+        store.claim_next_at(now).unwrap();
+        let logs = vec![ExecLog {
+            level: ExecLogLevel::Warn,
+            message: "still running".to_owned(),
+        }];
+
+        let completed = store
+            .complete_timeout_at(
+                job.id,
+                Some("execution timed out".to_owned()),
+                Some("traceback".to_owned()),
+                Some(logs.clone()),
+                now,
+            )
+            .unwrap();
+
+        assert_eq!(completed.state, ExecJobState::TimedOut);
+        assert!(completed.source.is_empty());
+        assert_eq!(completed.result, None);
+        assert_eq!(completed.logs, Some(logs));
+        assert_eq!(
+            completed.runtime_error.as_deref(),
+            Some("execution timed out")
+        );
+        assert_eq!(completed.traceback.as_deref(), Some("traceback"));
+    }
+
     #[test]
     fn duplicate_completion_is_rejected() {
         let store = test_store();

src/serve_session.rs

@@ -13,6 +13,7 @@ use thiserror::Error;
 
 use crate::{
     change_processor::ChangeProcessor,
+    exec::ExecJobStore,
     message_queue::MessageQueue,
     project::{Project, ProjectError},
     session_id::SessionId,
@@ -82,6 +83,9 @@ pub struct ServeSession {
     /// to be applied.
     message_queue: Arc<MessageQueue<AppliedPatchSet>>,
 
+    /// Execution jobs submitted to this serve session.
+    exec_job_store: ExecJobStore,
+
     /// A channel to send mutation requests on. These will be handled by the
     /// ChangeProcessor and trigger changes in the tree.
     tree_mutation_sender: Sender<PatchSet>,
@@ -143,6 +147,7 @@ impl ServeSession {
             root_project,
             tree,
             message_queue,
+            exec_job_store: ExecJobStore::new(),
             tree_mutation_sender,
             vfs,
         })
@@ -169,6 +174,10 @@ impl ServeSession {
         &self.message_queue
     }
 
+    pub fn exec_job_store(&self) -> &ExecJobStore {
+        &self.exec_job_store
+    }
+
     pub fn session_id(&self) -> SessionId {
         self.session_id
     }

src/web/api.rs

@@ -4,31 +4,73 @@
 use std::{collections::HashMap, fs, net::SocketAddr, path::PathBuf, str::FromStr, sync::Arc};
 
 use futures::{sink::SinkExt, stream::StreamExt};
-use hyper::{body, Body, Method, Request, Response, StatusCode};
+use hyper::{
+    body::{self, HttpBody},
+    header::{HeaderValue, CACHE_CONTROL},
+    Body, Method, Request, Response, StatusCode,
+};
 use hyper_tungstenite::{is_upgrade_request, tungstenite::Message, upgrade, HyperWebsocket};
 use opener::OpenError;
 use rbx_dom_weak::{
     types::{Ref, Variant},
     InstanceBuilder, UstrMap, WeakDom,
 };
+use uuid::Uuid;
 
 use crate::{
+    exec::{ExecJobStoreError, MAX_SOURCE_SIZE_BYTES},
     serve_session::ServeSession,
     snapshot::{InstanceWithMeta, PatchSet, PatchUpdate},
     web::{
         interface::{
-            ErrorResponse, Instance, MessagesPacket, OpenResponse, ReadResponse,
-            ServerInfoResponse, SocketPacket, SocketPacketBody, SocketPacketType, SubscribeMessage,
-            WriteRequest, WriteResponse, PROTOCOL_VERSION, SERVER_VERSION,
+            ErrorResponse, ExecJobClaimResponse, ExecJobCompletionRequest, ExecJobResponse,
+            ExecJobSubmissionRequest, ExecLog, Instance, MessagesPacket, OpenResponse,
+            ReadResponse, ServerInfoResponse, SocketPacket, SocketPacketBody, SocketPacketType,
+            SubscribeMessage, WriteRequest, WriteResponse, PROTOCOL_VERSION, SERVER_VERSION,
         },
         origin::canonical,
-        util::{deserialize_msgpack, msgpack, msgpack_ok, serialize_msgpack},
+        util::{deserialize_msgpack, msgpack, msgpack_ok, response, serialize_msgpack},
     },
     web_api::{
         InstanceUpdate, RefPatchRequest, RefPatchResponse, SerializeRequest, SerializeResponse,
     },
 };
 
+const EXEC_SUBMISSION_BODY_LIMIT_BYTES: usize = MAX_SOURCE_SIZE_BYTES + 64 * 1024;
+const EXEC_COMPLETION_BODY_LIMIT_BYTES: usize = 256 * 1024;
+
+enum ExecRoute {
+    Submit,
+    ClaimNext,
+    Status(String),
+    Complete(String),
+}
+
+impl ExecRoute {
+    fn parse(method: &Method, path: &str) -> Option<Self> {
+        match (method, path) {
+            (&Method::POST, "/api/exec/jobs") => return Some(Self::Submit),
+            (&Method::GET, "/api/exec/jobs/next") => return Some(Self::ClaimNext),
+            _ => {}
+        }
+
+        let remainder = path.strip_prefix("/api/exec/jobs/")?;
+        let mut segments = remainder.split('/');
+        let id = segments.next()?;
+        let suffix = segments.next();
+
+        if id.is_empty() || segments.next().is_some() {
+            return None;
+        }
+
+        match (method, suffix) {
+            (&Method::GET, None) => Some(Self::Status(id.to_owned())),
+            (&Method::POST, Some("complete")) => Some(Self::Complete(id.to_owned())),
+            _ => None,
+        }
+    }
+}
+
 pub async fn call(
     serve_session: Arc<ServeSession>,
     remote_addr: SocketAddr,
@@ -36,6 +78,21 @@ pub async fn call(
 ) -> Response<Body> {
     let service = ApiService::new(serve_session, remote_addr);
 
+    if let Some(route) = ExecRoute::parse(request.method(), request.uri().path()) {
+        if let Some(response) = service.reject_non_local("/api/exec") {
+            return response;
+        }
+
+        service.serve_session.exec_job_store().cleanup_expired();
+
+        return match route {
+            ExecRoute::Submit => service.handle_api_exec_submit(request).await,
+            ExecRoute::ClaimNext => service.handle_api_exec_claim_next(),
+            ExecRoute::Status(id) => service.handle_api_exec_status(&id),
+            ExecRoute::Complete(id) => service.handle_api_exec_complete(&id, request).await,
+        };
+    }
+
     match (request.method(), request.uri().path()) {
         (&Method::GET, "/api/rojo") => service.handle_api_rojo().await,
         (&Method::GET, path) if path.starts_with("/api/read/") => {
@@ -81,6 +138,144 @@ impl ApiService {
         }
     }
 
+    fn reject_non_local(&self, route: &str) -> Option<Response<Body>> {
+        if canonical(self.remote_addr.ip()).is_loopback() {
+            return None;
+        }
+
+        Some(msgpack(
+            ErrorResponse::forbidden(format!("{route} is only available to local clients")),
+            StatusCode::FORBIDDEN,
+        ))
+    }
+
+    async fn handle_api_exec_submit(&self, request: Request<Body>) -> Response<Body> {
+        let body = match read_exec_body(request.into_body(), EXEC_SUBMISSION_BODY_LIMIT_BYTES).await
+        {
+            Ok(body) => body,
+            Err(response) => return response,
+        };
+        let request: ExecJobSubmissionRequest = match deserialize_msgpack(&body) {
+            Ok(request) => request,
+            Err(err) => {
+                return msgpack(
+                    ErrorResponse::bad_request(format!("Invalid body: {err}")),
+                    StatusCode::BAD_REQUEST,
+                );
+            }
+        };
+
+        if request.script_name.is_empty() {
+            return msgpack(
+                ErrorResponse::bad_request("Exec script name must not be empty"),
+                StatusCode::BAD_REQUEST,
+            );
+        }
+
+        if request.script_name.chars().any(char::is_control) {
+            return msgpack(
+                ErrorResponse::bad_request("Exec script name must not contain control characters"),
+                StatusCode::BAD_REQUEST,
+            );
+        }
+
+        match self
+            .serve_session
+            .exec_job_store()
+            .submit(request.script_name, request.source)
+        {
+            Ok(job) => msgpack(ExecJobResponse::from(job), StatusCode::CREATED),
+            Err(err) => exec_store_error(err),
+        }
+    }
+
+    fn handle_api_exec_claim_next(&self) -> Response<Body> {
+        match self.serve_session.exec_job_store().claim_next() {
+            Some(job) => no_store(msgpack_ok(ExecJobClaimResponse::from(job))),
+            None => no_store(response(
+                StatusCode::NO_CONTENT,
+                "application/msgpack",
+                Body::empty(),
+            )),
+        }
+    }
+
+    fn handle_api_exec_status(&self, id: &str) -> Response<Body> {
+        let id = match Uuid::parse_str(id) {
+            Ok(id) => id,
+            Err(err) => {
+                return msgpack(
+                    ErrorResponse::bad_request(format!("Malformed exec job ID: {err}")),
+                    StatusCode::BAD_REQUEST,
+                );
+            }
+        };
+
+        match self.serve_session.exec_job_store().get(id) {
+            Ok(job) => no_store(msgpack_ok(ExecJobResponse::from(job))),
+            Err(err) => exec_store_error(err),
+        }
+    }
+
+    async fn handle_api_exec_complete(&self, id: &str, request: Request<Body>) -> Response<Body> {
+        let id = match Uuid::parse_str(id) {
+            Ok(id) => id,
+            Err(err) => {
+                return msgpack(
+                    ErrorResponse::bad_request(format!("Malformed exec job ID: {err}")),
+                    StatusCode::BAD_REQUEST,
+                );
+            }
+        };
+
+        let body = match read_exec_body(request.into_body(), EXEC_COMPLETION_BODY_LIMIT_BYTES).await
+        {
+            Ok(body) => body,
+            Err(response) => return response,
+        };
+        let request: ExecJobCompletionRequest = match deserialize_msgpack(&body) {
+            Ok(request) => request,
+            Err(err) => {
+                return msgpack(
+                    ErrorResponse::bad_request(format!("Invalid body: {err}")),
+                    StatusCode::BAD_REQUEST,
+                );
+            }
+        };
+
+        let store = self.serve_session.exec_job_store();
+        let result = match request {
+            ExecJobCompletionRequest::Success { result, logs } => {
+                store.complete_success(id, result.map(Into::into), into_stored_logs(logs))
+            }
+            ExecJobCompletionRequest::CompileFailure {
+                error,
+                traceback,
+                logs,
+            }
+            | ExecJobCompletionRequest::RuntimeFailure {
+                error,
+                traceback,
+                logs,
+            }
+            | ExecJobCompletionRequest::Rejected {
+                error,
+                traceback,
+                logs,
+            } => store.complete_failure(id, Some(error), traceback, into_stored_logs(logs)),
+            ExecJobCompletionRequest::Timeout {
+                error,
+                traceback,
+                logs,
+            } => store.complete_timeout(id, Some(error), traceback, into_stored_logs(logs)),
+        };
+
+        match result {
+            Ok(job) => no_store(msgpack_ok(ExecJobResponse::from(job))),
+            Err(err) => exec_store_error(err),
+        }
+    }
+
     /// Get a summary of information about the server
     async fn handle_api_rojo(&self) -> Response<Body> {
         let tree = self.serve_session.tree();
@@ -374,11 +569,8 @@ impl ApiService {
         // An IPv4 client reaching a dual-stack (`::`) bind appears as an
         // IPv4-mapped IPv6 peer (`::ffff:127.0.0.1`), so canonicalize to the bare
         // IPv4 form before the loopback test, matching `origin`'s handling.
-        if !canonical(self.remote_addr.ip()).is_loopback() {
-            return msgpack(
-                ErrorResponse::forbidden("/api/open is only available to local clients"),
-                StatusCode::FORBIDDEN,
-            );
+        if let Some(response) = self.reject_non_local("/api/open") {
+            return response;
         }
 
         let argument = &request.uri().path()["/api/open/".len()..];
@@ -456,6 +648,84 @@ impl ApiService {
     }
 }
 
+async fn read_exec_body(mut body: Body, limit: usize) -> Result<Vec<u8>, Response<Body>> {
+    if body
+        .size_hint()
+        .upper()
+        .is_some_and(|size| size > limit as u64)
+    {
+        return Err(exec_body_too_large(limit));
+    }
+
+    let capacity = body.size_hint().lower().min(limit as u64) as usize;
+    let mut bytes = Vec::with_capacity(capacity);
+
+    while let Some(chunk) = body.data().await {
+        let chunk = match chunk {
+            Ok(chunk) => chunk,
+            Err(err) => {
+                return Err(msgpack(
+                    ErrorResponse::bad_request(format!("Failed to read request body: {err}")),
+                    StatusCode::BAD_REQUEST,
+                ));
+            }
+        };
+
+        if bytes.len().saturating_add(chunk.len()) > limit {
+            return Err(exec_body_too_large(limit));
+        }
+
+        bytes.extend_from_slice(&chunk);
+    }
+
+    Ok(bytes)
+}
+
+fn exec_body_too_large(limit: usize) -> Response<Body> {
+    msgpack(
+        ErrorResponse::payload_too_large(format!(
+            "Exec request body exceeds the {limit}-byte limit"
+        )),
+        StatusCode::PAYLOAD_TOO_LARGE,
+    )
+}
+
+fn exec_store_error(error: ExecJobStoreError) -> Response<Body> {
+    let details = error.to_string();
+
+    match error {
+        ExecJobStoreError::SourceTooLarge { .. } => msgpack(
+            ErrorResponse::payload_too_large(details),
+            StatusCode::PAYLOAD_TOO_LARGE,
+        ),
+        ExecJobStoreError::PendingQueueFull { .. } => msgpack(
+            ErrorResponse::too_many_requests(details),
+            StatusCode::TOO_MANY_REQUESTS,
+        ),
+        ExecJobStoreError::UnknownJob { .. } => {
+            msgpack(ErrorResponse::not_found(details), StatusCode::NOT_FOUND)
+        }
+        ExecJobStoreError::JobNotClaimed { .. } | ExecJobStoreError::DuplicateCompletion { .. } => {
+            msgpack(ErrorResponse::conflict(details), StatusCode::CONFLICT)
+        }
+    }
+}
+
+fn into_stored_logs(logs: Vec<ExecLog>) -> Option<Vec<crate::exec::ExecLog>> {
+    if logs.is_empty() {
+        None
+    } else {
+        Some(logs.into_iter().map(Into::into).collect())
+    }
+}
+
+fn no_store(mut response: Response<Body>) -> Response<Body> {
+    response
+        .headers_mut()
+        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
+    response
+}
+
 /// If this instance is represented by a script, try to find the correct .lua or .luau
 /// file to open to edit it.
 fn pick_script_path(instance: InstanceWithMeta<'_>) -> Option<PathBuf> {
@@ -605,3 +875,50 @@ fn parent_requirements(class: &str) -> Option<&str> {
         _ => return None,
     })
 }
+
+#[cfg(test)]
+mod test {
+    use memofs::{StdBackend, Vfs};
+
+    use super::*;
+
+    #[tokio::test]
+    async fn exec_routes_reject_non_loopback_peers() {
+        let dir = tempfile::tempdir().unwrap();
+        std::fs::write(
+            dir.path().join("default.project.json"),
+            r#"{ "name": "test", "tree": { "$className": "Folder" } }"#,
+        )
+        .unwrap();
+
+        let start_path = std::fs::canonicalize(dir.path()).unwrap();
+        let vfs = Vfs::new(StdBackend::new().unwrap());
+        let serve_session = Arc::new(ServeSession::new(vfs, start_path).unwrap());
+        let id = Uuid::new_v4();
+        let routes = [
+            (Method::POST, "/api/exec/jobs".to_owned()),
+            (Method::GET, "/api/exec/jobs/next".to_owned()),
+            (Method::GET, format!("/api/exec/jobs/{id}")),
+            (Method::POST, format!("/api/exec/jobs/{id}/complete")),
+        ];
+
+        for (method, route) in routes {
+            let request = Request::builder()
+                .method(method)
+                .uri(route)
+                .body(Body::empty())
+                .unwrap();
+            let response = call(
+                Arc::clone(&serve_session),
+                "192.0.2.1:4567".parse().unwrap(),
+                request,
+            )
+            .await;
+
+            assert_eq!(response.status(), StatusCode::FORBIDDEN);
+            let body = body::to_bytes(response.into_body()).await.unwrap();
+            let error: serde_json::Value = deserialize_msgpack(&body).unwrap();
+            assert_eq!(error["kind"], "Forbidden");
+        }
+    }
+}

src/web/interface.rs

@@ -15,6 +15,10 @@ use serde::{Deserialize, Serialize};
 use strum::Display;
 
 use crate::{
+    exec::{
+        ExecJob as StoredExecJob, ExecJobState as StoredExecJobState, ExecLog as StoredExecLog,
+        ExecLogLevel as StoredExecLogLevel, ExecValue as StoredExecValue,
+    },
     session_id::SessionId,
     snapshot::{
         AppliedPatchSet, InstanceMetadata as RojoInstanceMetadata, InstanceWithMeta, RojoTree,
@@ -267,6 +271,223 @@ pub struct RefPatchResponse<'a> {
     pub patch: SubscribeMessage<'a>,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecJobSubmissionRequest {
+    pub script_name: String,
+    pub source: String,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub enum ExecJobState {
+    Pending,
+    Claimed,
+    Succeeded,
+    Failed,
+    TimedOut,
+}
+
+impl From<StoredExecJobState> for ExecJobState {
+    fn from(value: StoredExecJobState) -> Self {
+        match value {
+            StoredExecJobState::Pending => Self::Pending,
+            StoredExecJobState::Claimed => Self::Claimed,
+            StoredExecJobState::Succeeded => Self::Succeeded,
+            StoredExecJobState::Failed => Self::Failed,
+            StoredExecJobState::TimedOut => Self::TimedOut,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(tag = "kind", rename_all = "camelCase")]
+pub enum ExecValue {
+    Nil,
+    String { value: String },
+    Number { value: f64 },
+    Boolean { value: bool },
+    Array { value: Vec<ExecValue> },
+    Table { value: Vec<ExecTableEntry> },
+}
+
+impl From<StoredExecValue> for ExecValue {
+    fn from(value: StoredExecValue) -> Self {
+        match value {
+            StoredExecValue::Nil => Self::Nil,
+            StoredExecValue::String(value) => Self::String { value },
+            StoredExecValue::Number(value) => Self::Number { value },
+            StoredExecValue::Boolean(value) => Self::Boolean { value },
+            StoredExecValue::Array(value) => Self::Array {
+                value: value.into_iter().map(Into::into).collect(),
+            },
+            StoredExecValue::Table(value) => Self::Table {
+                value: value
+                    .into_iter()
+                    .map(|(key, value)| ExecTableEntry {
+                        key,
+                        value: value.into(),
+                    })
+                    .collect(),
+            },
+        }
+    }
+}
+
+impl From<ExecValue> for StoredExecValue {
+    fn from(value: ExecValue) -> Self {
+        match value {
+            ExecValue::Nil => Self::Nil,
+            ExecValue::String { value } => Self::String(value),
+            ExecValue::Number { value } => Self::Number(value),
+            ExecValue::Boolean { value } => Self::Boolean(value),
+            ExecValue::Array { value } => Self::Array(value.into_iter().map(Into::into).collect()),
+            ExecValue::Table { value } => Self::Table(
+                value
+                    .into_iter()
+                    .map(|entry| (entry.key, entry.value.into()))
+                    .collect(),
+            ),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecTableEntry {
+    pub key: String,
+    pub value: ExecValue,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub enum ExecLogLevel {
+    Print,
+    Warn,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecLog {
+    pub level: ExecLogLevel,
+    pub message: String,
+}
+
+impl From<StoredExecLog> for ExecLog {
+    fn from(value: StoredExecLog) -> Self {
+        Self {
+            level: match value.level {
+                StoredExecLogLevel::Print => ExecLogLevel::Print,
+                StoredExecLogLevel::Warn => ExecLogLevel::Warn,
+            },
+            message: value.message,
+        }
+    }
+}
+
+impl From<ExecLog> for StoredExecLog {
+    fn from(value: ExecLog) -> Self {
+        Self {
+            level: match value.level {
+                ExecLogLevel::Print => StoredExecLogLevel::Print,
+                ExecLogLevel::Warn => StoredExecLogLevel::Warn,
+            },
+            message: value.message,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecJobResponse {
+    pub job_id: String,
+    pub script_name: String,
+    pub state: ExecJobState,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub result: Option<ExecValue>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub logs: Option<Vec<ExecLog>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub traceback: Option<String>,
+}
+
+impl From<StoredExecJob> for ExecJobResponse {
+    fn from(value: StoredExecJob) -> Self {
+        Self {
+            job_id: value.id.to_string(),
+            script_name: value.script_name,
+            state: value.state.into(),
+            result: value.result.map(Into::into),
+            logs: value
+                .logs
+                .map(|logs| logs.into_iter().map(Into::into).collect()),
+            error: value.runtime_error,
+            traceback: value.traceback,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(rename_all = "camelCase")]
+pub struct ExecJobClaimResponse {
+    pub job_id: String,
+    pub script_name: String,
+    pub source: String,
+    pub state: ExecJobState,
+}
+
+impl From<StoredExecJob> for ExecJobClaimResponse {
+    fn from(value: StoredExecJob) -> Self {
+        Self {
+            job_id: value.id.to_string(),
+            script_name: value.script_name,
+            source: value.source,
+            state: value.state.into(),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(tag = "outcome", rename_all = "camelCase")]
+pub enum ExecJobCompletionRequest {
+    Success {
+        #[serde(default)]
+        result: Option<ExecValue>,
+        #[serde(default)]
+        logs: Vec<ExecLog>,
+    },
+    CompileFailure {
+        error: String,
+        #[serde(default)]
+        traceback: Option<String>,
+        #[serde(default)]
+        logs: Vec<ExecLog>,
+    },
+    RuntimeFailure {
+        error: String,
+        #[serde(default)]
+        traceback: Option<String>,
+        #[serde(default)]
+        logs: Vec<ExecLog>,
+    },
+    Rejected {
+        error: String,
+        #[serde(default)]
+        traceback: Option<String>,
+        #[serde(default)]
+        logs: Vec<ExecLog>,
+    },
+    Timeout {
+        error: String,
+        #[serde(default)]
+        traceback: Option<String>,
+        #[serde(default)]
+        logs: Vec<ExecLog>,
+    },
+}
+
 /// General response type returned from all Rojo routes
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -297,6 +518,27 @@ impl ErrorResponse {
         }
     }
 
+    pub fn conflict<S: Into<String>>(details: S) -> Self {
+        Self {
+            kind: ErrorResponseKind::Conflict,
+            details: details.into(),
+        }
+    }
+
+    pub fn payload_too_large<S: Into<String>>(details: S) -> Self {
+        Self {
+            kind: ErrorResponseKind::PayloadTooLarge,
+            details: details.into(),
+        }
+    }
+
+    pub fn too_many_requests<S: Into<String>>(details: S) -> Self {
+        Self {
+            kind: ErrorResponseKind::TooManyRequests,
+            details: details.into(),
+        }
+    }
+
     pub fn internal_error<S: Into<String>>(details: S) -> Self {
         Self {
             kind: ErrorResponseKind::InternalError,
@@ -310,5 +552,8 @@ pub enum ErrorResponseKind {
     NotFound,
     BadRequest,
     Forbidden,
+    Conflict,
+    PayloadTooLarge,
+    TooManyRequests,
     InternalError,
 }

tests/rojo_test/serve_util.rs

@@ -15,8 +15,8 @@ use tempfile::{tempdir, TempDir};
 
 use librojo::{
     web_api::{
-        ReadResponse, SerializeRequest, SerializeResponse, ServerInfoResponse, SocketPacket,
-        SocketPacketType,
+        ExecJobCompletionRequest, ExecJobSubmissionRequest, ReadResponse, SerializeRequest,
+        SerializeResponse, ServerInfoResponse, SocketPacket, SocketPacketType,
     },
     SessionId,
 };
@@ -276,6 +276,54 @@ impl TestServeSession {
             .expect("Failed to send request")
             .status()
     }
+
+    pub fn post_api_exec_job(
+        &self,
+        request: &ExecJobSubmissionRequest,
+    ) -> reqwest::blocking::Response {
+        let body = serialize_msgpack(request).unwrap();
+        self.api_request(reqwest::Method::POST, "/api/exec/jobs", Some(body))
+    }
+
+    pub fn get_api_exec_next(&self) -> reqwest::blocking::Response {
+        self.api_request(reqwest::Method::GET, "/api/exec/jobs/next", None)
+    }
+
+    pub fn get_api_exec_status(&self, id: &str) -> reqwest::blocking::Response {
+        self.api_request(reqwest::Method::GET, &format!("/api/exec/jobs/{id}"), None)
+    }
+
+    pub fn post_api_exec_complete(
+        &self,
+        id: &str,
+        request: &ExecJobCompletionRequest,
+    ) -> reqwest::blocking::Response {
+        let body = serialize_msgpack(request).unwrap();
+        self.api_request(
+            reqwest::Method::POST,
+            &format!("/api/exec/jobs/{id}/complete"),
+            Some(body),
+        )
+    }
+
+    pub fn api_request(
+        &self,
+        method: reqwest::Method,
+        path: &str,
+        body: Option<Vec<u8>>,
+    ) -> reqwest::blocking::Response {
+        let client = reqwest::blocking::Client::new();
+        let url = format!("http://localhost:{}{}", self.port, path);
+        let mut request = client.request(method, url);
+
+        if let Some(body) = body {
+            request = request
+                .header(reqwest::header::CONTENT_TYPE, "application/msgpack")
+                .body(body);
+        }
+
+        request.send().expect("Failed to send request")
+    }
 }
 
 fn serialize_msgpack<T: Serialize>(value: T) -> Result<Vec<u8>, rmp_serde::encode::Error> {

tests/tests/serve.rs

@@ -3,14 +3,333 @@ use std::fs;
 use insta::{assert_snapshot, assert_yaml_snapshot, with_settings};
 use rbx_dom_weak::types::Ref;
 use reqwest::StatusCode;
+use serde::de::DeserializeOwned;
 use tempfile::tempdir;
+use uuid::Uuid;
 
 use crate::rojo_test::{
     internable::InternAndRedact,
-    serve_util::{deserialize_msgpack, run_serve_test, serialize_to_xml_model},
+    serve_util::{deserialize_msgpack, run_serve_test, serialize_to_xml_model, TestServeSession},
 };
 
-use librojo::web_api::{SerializeResponse, SocketPacketType};
+use librojo::{
+    exec::{MAX_PENDING_JOBS, MAX_SOURCE_SIZE_BYTES},
+    web_api::{
+        ExecJobClaimResponse, ExecJobCompletionRequest, ExecJobResponse,
+        ExecJobState as ApiExecJobState, ExecJobSubmissionRequest, ExecLog, ExecLogLevel,
+        ExecValue, SerializeResponse, SocketPacketType,
+    },
+};
+
+#[test]
+fn exec_jobs_submit_claim_complete_and_release_slot() {
+    run_serve_test("empty", |session, _redactions| {
+        let submission_response = session.post_api_exec_job(&ExecJobSubmissionRequest {
+            script_name: "first.lua".to_owned(),
+            source: "return true".to_owned(),
+        });
+        assert_eq!(submission_response.status(), StatusCode::CREATED);
+        let submission_body = submission_response.bytes().unwrap();
+        let submission_value: serde_json::Value = deserialize_msgpack(&submission_body).unwrap();
+        assert!(submission_value.get("source").is_none());
+        let first: ExecJobResponse = deserialize_msgpack(&submission_body).unwrap();
+        let second = submit_exec_job(&session, "second.lua", "return 'second'");
+
+        assert_eq!(first.script_name, "first.lua");
+        assert_eq!(first.state, ApiExecJobState::Pending);
+        assert_eq!(second.state, ApiExecJobState::Pending);
+
+        let status_response = session.get_api_exec_status(&first.job_id);
+        assert_eq!(status_response.status(), StatusCode::OK);
+        let status_body = status_response.bytes().unwrap();
+        let status_value: serde_json::Value = deserialize_msgpack(&status_body).unwrap();
+        assert_eq!(status_value["state"], "pending");
+        assert!(status_value.get("source").is_none());
+
+        let claim_response = session.get_api_exec_next();
+        assert_eq!(claim_response.status(), StatusCode::OK);
+        let claim: ExecJobClaimResponse = decode_response(claim_response);
+        assert_eq!(claim.job_id, first.job_id);
+        assert_eq!(claim.source, "return true");
+        assert_eq!(claim.state, ApiExecJobState::Claimed);
+
+        let claimed_status: ExecJobResponse =
+            decode_response(session.get_api_exec_status(&first.job_id));
+        assert_eq!(claimed_status.state, ApiExecJobState::Claimed);
+
+        let blocked = session.get_api_exec_next();
+        assert_eq!(blocked.status(), StatusCode::NO_CONTENT);
+        assert!(blocked.bytes().unwrap().is_empty());
+
+        let expected_logs = vec![ExecLog {
+            level: ExecLogLevel::Print,
+            message: "finished".to_owned(),
+        }];
+        let completion = ExecJobCompletionRequest::Success {
+            result: Some(ExecValue::Boolean { value: true }),
+            logs: expected_logs.clone(),
+        };
+        let completion_response = session.post_api_exec_complete(&first.job_id, &completion);
+        assert_eq!(completion_response.status(), StatusCode::OK);
+        let completed: ExecJobResponse = decode_response(completion_response);
+        assert_eq!(completed.state, ApiExecJobState::Succeeded);
+        assert_eq!(completed.result, Some(ExecValue::Boolean { value: true }));
+        assert_eq!(completed.logs, Some(expected_logs));
+
+        let succeeded_status: ExecJobResponse =
+            decode_response(session.get_api_exec_status(&first.job_id));
+        assert_eq!(succeeded_status.state, ApiExecJobState::Succeeded);
+
+        let second_claim: ExecJobClaimResponse = decode_response(session.get_api_exec_next());
+        assert_eq!(second_claim.job_id, second.job_id);
+        assert_eq!(second_claim.source, "return 'second'");
+
+        let second_completion = ExecJobCompletionRequest::Success {
+            result: Some(ExecValue::Nil),
+            logs: Vec::new(),
+        };
+        assert_eq!(
+            session
+                .post_api_exec_complete(&second.job_id, &second_completion)
+                .status(),
+            StatusCode::OK,
+        );
+
+        let empty = session.get_api_exec_next();
+        assert_eq!(empty.status(), StatusCode::NO_CONTENT);
+        assert!(empty.bytes().unwrap().is_empty());
+    });
+}
+
+#[test]
+fn exec_jobs_record_failure_and_timeout_outcomes() {
+    run_serve_test("empty", |session, _redactions| {
+        let compile_job = submit_exec_job(&session, "compile.lua", "not valid luau");
+        assert_eq!(
+            decode_response::<ExecJobClaimResponse>(session.get_api_exec_next()).job_id,
+            compile_job.job_id,
+        );
+        let compile_failure = ExecJobCompletionRequest::CompileFailure {
+            error: "expected identifier".to_owned(),
+            traceback: None,
+            logs: Vec::new(),
+        };
+        let compile_response =
+            session.post_api_exec_complete(&compile_job.job_id, &compile_failure);
+        assert_eq!(compile_response.status(), StatusCode::OK);
+        let compile_status: ExecJobResponse = decode_response(compile_response);
+        assert_eq!(compile_status.state, ApiExecJobState::Failed);
+        assert_eq!(compile_status.error.as_deref(), Some("expected identifier"));
+        assert_eq!(compile_status.traceback, None);
+
+        assert_eq!(
+            session
+                .post_api_exec_complete(&compile_job.job_id, &compile_failure)
+                .status(),
+            StatusCode::CONFLICT,
+        );
+
+        let runtime_job = submit_exec_job(&session, "runtime.lua", "error('boom')");
+        decode_response::<ExecJobClaimResponse>(session.get_api_exec_next());
+        let expected_logs = vec![ExecLog {
+            level: ExecLogLevel::Warn,
+            message: "about to fail".to_owned(),
+        }];
+        let runtime_failure = ExecJobCompletionRequest::RuntimeFailure {
+            error: "boom".to_owned(),
+            traceback: Some("runtime.lua:1".to_owned()),
+            logs: expected_logs.clone(),
+        };
+        assert_eq!(
+            session
+                .post_api_exec_complete(&runtime_job.job_id, &runtime_failure)
+                .status(),
+            StatusCode::OK,
+        );
+        let runtime_status: ExecJobResponse =
+            decode_response(session.get_api_exec_status(&runtime_job.job_id));
+        assert_eq!(runtime_status.state, ApiExecJobState::Failed);
+        assert_eq!(runtime_status.error.as_deref(), Some("boom"));
+        assert_eq!(runtime_status.traceback.as_deref(), Some("runtime.lua:1"));
+        assert_eq!(runtime_status.logs, Some(expected_logs));
+
+        let rejected_job = submit_exec_job(&session, "rejected.lua", "return nil");
+        decode_response::<ExecJobClaimResponse>(session.get_api_exec_next());
+        let rejected = ExecJobCompletionRequest::Rejected {
+            error: "Rojo exec is only available in edit mode".to_owned(),
+            traceback: None,
+            logs: Vec::new(),
+        };
+        let rejected_status: ExecJobResponse =
+            decode_response(session.post_api_exec_complete(&rejected_job.job_id, &rejected));
+        assert_eq!(rejected_status.state, ApiExecJobState::Failed);
+
+        let timeout_job = submit_exec_job(&session, "timeout.lua", "task.wait(60)");
+        decode_response::<ExecJobClaimResponse>(session.get_api_exec_next());
+        let timeout = ExecJobCompletionRequest::Timeout {
+            error: "execution timed out".to_owned(),
+            traceback: Some("timeout.lua:1".to_owned()),
+            logs: Vec::new(),
+        };
+        let timeout_status: ExecJobResponse =
+            decode_response(session.post_api_exec_complete(&timeout_job.job_id, &timeout));
+        assert_eq!(timeout_status.state, ApiExecJobState::TimedOut);
+        assert_eq!(timeout_status.error.as_deref(), Some("execution timed out"));
+
+        let retained_timeout: ExecJobResponse =
+            decode_response(session.get_api_exec_status(&timeout_job.job_id));
+        assert_eq!(retained_timeout.state, ApiExecJobState::TimedOut);
+    });
+}
+
+#[test]
+fn exec_jobs_validate_requests_routes_and_limits() {
+    run_serve_test("empty", |session, _redactions| {
+        let pending = submit_exec_job(&session, "pending.lua", "return nil");
+        let success = ExecJobCompletionRequest::Success {
+            result: None,
+            logs: Vec::new(),
+        };
+        assert_eq!(
+            session
+                .post_api_exec_complete(&pending.job_id, &success)
+                .status(),
+            StatusCode::CONFLICT,
+        );
+
+        assert_eq!(
+            session.get_api_exec_status("not-a-uuid").status(),
+            StatusCode::BAD_REQUEST,
+        );
+        let unknown_id = Uuid::new_v4().to_string();
+        assert_eq!(
+            session.get_api_exec_status(&unknown_id).status(),
+            StatusCode::NOT_FOUND,
+        );
+        assert_eq!(
+            session
+                .post_api_exec_complete(&unknown_id, &success)
+                .status(),
+            StatusCode::NOT_FOUND,
+        );
+
+        decode_response::<ExecJobClaimResponse>(session.get_api_exec_next());
+        let malformed_path = format!("/api/exec/jobs/{}/complete", pending.job_id);
+        assert_eq!(
+            session
+                .api_request(reqwest::Method::POST, &malformed_path, Some(vec![0xc1]))
+                .status(),
+            StatusCode::BAD_REQUEST,
+        );
+
+        let oversized_completion = ExecJobCompletionRequest::RuntimeFailure {
+            error: "x".repeat(300 * 1024),
+            traceback: None,
+            logs: Vec::new(),
+        };
+        assert_eq!(
+            session
+                .post_api_exec_complete(&pending.job_id, &oversized_completion)
+                .status(),
+            StatusCode::PAYLOAD_TOO_LARGE,
+        );
+        assert_eq!(
+            session
+                .post_api_exec_complete(&pending.job_id, &success)
+                .status(),
+            StatusCode::OK,
+        );
+
+        for (method, path) in [
+            (reqwest::Method::GET, "/api/exec/jobs/"),
+            (reqwest::Method::GET, "/api/exec/jobs/next/extra"),
+            (
+                reqwest::Method::POST,
+                "/api/exec/jobs/not-a-uuid/complete/extra",
+            ),
+        ] {
+            assert_eq!(
+                session.api_request(method, path, None).status(),
+                StatusCode::NOT_FOUND,
+            );
+        }
+        assert_eq!(
+            session
+                .api_request(reqwest::Method::POST, "/api/exec/jobs/next", None)
+                .status(),
+            StatusCode::NOT_FOUND,
+        );
+
+        assert_eq!(
+            session
+                .post_api_exec_job(&ExecJobSubmissionRequest {
+                    script_name: String::new(),
+                    source: "return nil".to_owned(),
+                })
+                .status(),
+            StatusCode::BAD_REQUEST,
+        );
+        assert_eq!(
+            session
+                .post_api_exec_job(&ExecJobSubmissionRequest {
+                    script_name: "bad\nname.lua".to_owned(),
+                    source: "return nil".to_owned(),
+                })
+                .status(),
+            StatusCode::BAD_REQUEST,
+        );
+        assert_eq!(
+            session
+                .api_request(reqwest::Method::POST, "/api/exec/jobs", Some(vec![0xc1]))
+                .status(),
+            StatusCode::BAD_REQUEST,
+        );
+        assert_eq!(
+            session
+                .post_api_exec_job(&ExecJobSubmissionRequest {
+                    script_name: "large.lua".to_owned(),
+                    source: "x".repeat(MAX_SOURCE_SIZE_BYTES + 1),
+                })
+                .status(),
+            StatusCode::PAYLOAD_TOO_LARGE,
+        );
+
+        for index in 0..MAX_PENDING_JOBS {
+            assert_eq!(
+                session
+                    .post_api_exec_job(&ExecJobSubmissionRequest {
+                        script_name: format!("queue-{index}.lua"),
+                        source: "return nil".to_owned(),
+                    })
+                    .status(),
+                StatusCode::CREATED,
+            );
+        }
+        assert_eq!(
+            session
+                .post_api_exec_job(&ExecJobSubmissionRequest {
+                    script_name: "queue-overflow.lua".to_owned(),
+                    source: "return nil".to_owned(),
+                })
+                .status(),
+            StatusCode::TOO_MANY_REQUESTS,
+        );
+    });
+}
+
+fn submit_exec_job(session: &TestServeSession, script_name: &str, source: &str) -> ExecJobResponse {
+    let response = session.post_api_exec_job(&ExecJobSubmissionRequest {
+        script_name: script_name.to_owned(),
+        source: source.to_owned(),
+    });
+    assert_eq!(response.status(), StatusCode::CREATED);
+    decode_response(response)
+}
+
+fn decode_response<T: DeserializeOwned>(response: reqwest::blocking::Response) -> T {
+    let body = response.bytes().expect("Failed to read response body");
+    deserialize_msgpack(&body).expect("Server returned malformed response")
+}
 
 #[test]
 fn rejects_dns_rebinding_requests() {