diff --git a/src/exec.rs b/src/exec.rs index 1139b97c..e0c558e5 100644 --- a/src/exec.rs +++ b/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, + traceback: Option, + logs: Option>, + ) -> Result { + self.complete_timeout_at(id, runtime_error, traceback, logs, Instant::now()) + } + pub fn get(&self, id: Uuid) -> Result { 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, + 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::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(); diff --git a/src/serve_session.rs b/src/serve_session.rs index 6cffb0d2..601464ae 100644 --- a/src/serve_session.rs +++ b/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>, + /// 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, @@ -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 } diff --git a/src/web/api.rs b/src/web/api.rs index 4fd21c0a..33d30803 100644 --- a/src/web/api.rs +++ b/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 { + 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, remote_addr: SocketAddr, @@ -36,6 +78,21 @@ pub async fn call( ) -> Response { 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> { + 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) -> Response { + 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 { + 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 { + 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) -> Response { + 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 { 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, Response> { + 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 { + 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 { + 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) -> Option> { + if logs.is_empty() { + None + } else { + Some(logs.into_iter().map(Into::into).collect()) + } +} + +fn no_store(mut response: Response) -> Response { + 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 { @@ -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"); + } + } +} diff --git a/src/web/interface.rs b/src/web/interface.rs index 8ec332ed..73b3af8b 100644 --- a/src/web/interface.rs +++ b/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 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 }, + Table { value: Vec }, +} + +impl From 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 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 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 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, + #[serde(skip_serializing_if = "Option::is_none")] + pub logs: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub traceback: Option, +} + +impl From 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 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, + #[serde(default)] + logs: Vec, + }, + CompileFailure { + error: String, + #[serde(default)] + traceback: Option, + #[serde(default)] + logs: Vec, + }, + RuntimeFailure { + error: String, + #[serde(default)] + traceback: Option, + #[serde(default)] + logs: Vec, + }, + Rejected { + error: String, + #[serde(default)] + traceback: Option, + #[serde(default)] + logs: Vec, + }, + Timeout { + error: String, + #[serde(default)] + traceback: Option, + #[serde(default)] + logs: Vec, + }, +} + /// 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>(details: S) -> Self { + Self { + kind: ErrorResponseKind::Conflict, + details: details.into(), + } + } + + pub fn payload_too_large>(details: S) -> Self { + Self { + kind: ErrorResponseKind::PayloadTooLarge, + details: details.into(), + } + } + + pub fn too_many_requests>(details: S) -> Self { + Self { + kind: ErrorResponseKind::TooManyRequests, + details: details.into(), + } + } + pub fn internal_error>(details: S) -> Self { Self { kind: ErrorResponseKind::InternalError, @@ -310,5 +552,8 @@ pub enum ErrorResponseKind { NotFound, BadRequest, Forbidden, + Conflict, + PayloadTooLarge, + TooManyRequests, InternalError, } diff --git a/tests/rojo_test/serve_util.rs b/tests/rojo_test/serve_util.rs index 458743e0..905687b5 100644 --- a/tests/rojo_test/serve_util.rs +++ b/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>, + ) -> 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(value: T) -> Result, rmp_serde::encode::Error> { diff --git a/tests/tests/serve.rs b/tests/tests/serve.rs index 0d6f75cc..688280de 100644 --- a/tests/tests/serve.rs +++ b/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::(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::(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::(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::(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::(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(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() {