//! Worker-backed Runtime REST process wrapper. //! //! This binary starts a Runtime command API with a real worker execution backend. //! A REST Runtime process that cannot spawn Workers is not a valid Runtime for the //! Workspace Browser. use std::collections::VecDeque; use std::env; use std::error::Error; use std::fmt; use std::net::SocketAddr; use std::path::PathBuf; use std::process::ExitCode; use std::sync::Arc; use worker_runtime::error::RuntimeError; use worker_runtime::fs_store::FsRuntimeStoreOptions; use worker_runtime::http_server::{ RuntimeHttpServerConfig, RuntimeHttpServerError, RuntimeHttpStoreSelection, }; use worker_runtime::identity::RuntimeId; use worker_runtime::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeExecutionBackend}; use worker_runtime::{Runtime, RuntimeOptions}; fn main() -> ExitCode { match run() { Ok(()) => ExitCode::SUCCESS, Err(error) => { eprintln!("worker-runtime-rest-server: {error}"); if let ProcessError::Usage(_) = error { eprintln!(); eprintln!("{}", usage()); ExitCode::from(2) } else { ExitCode::FAILURE } } } } fn run() -> Result<(), ProcessError> { let Some(config) = parse_args(env::args().skip(1))? else { println!("{}", usage()); return Ok(()); }; let runtime = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; runtime.block_on(async move { let listener = tokio::net::TcpListener::bind(config.http.bind_addr).await?; let local_addr = listener.local_addr()?; let worker_runtime = build_runtime(&config)?; eprintln!( "worker-runtime REST server listening on {local_addr}; intended client is a trusted backend/proxy, not a browser" ); worker_runtime::http_server::serve_runtime_http( worker_runtime, listener, config.http.local_token, ) .await .map_err(ProcessError::from) })?; Ok(()) } fn build_runtime(config: &ProcessConfig) -> Result { let mut factory = ProfileRuntimeWorkerFactory::new(config.workspace_root.clone()) .with_cwd(config.cwd.clone()); if let Some(store_dir) = config.worker_store_dir.clone() { factory = factory.with_store_dir(store_dir); } if let Some(metadata_dir) = config.worker_metadata_dir.clone() { factory = factory.with_worker_metadata_dir(metadata_dir); } if let Some(runtime_base_dir) = config.worker_runtime_base_dir.clone() { factory = factory.with_runtime_base_dir(runtime_base_dir); } if let Some(profile) = config.profile.clone() { factory = factory.with_profile(profile); } let backend = Arc::new(WorkerRuntimeExecutionBackend::new(factory).map_err(ProcessError::WorkerAdapter)?); match &config.http.store { RuntimeHttpStoreSelection::Memory => { Runtime::with_execution_backend(runtime_options_from_http(&config.http), backend) .map_err(ProcessError::Runtime) } RuntimeHttpStoreSelection::Fs { root } => { let mut options = FsRuntimeStoreOptions::new(root.clone()); options.runtime_id = config.http.runtime_id.clone(); options.display_name = config.http.display_name.clone(); options.limits = config.http.limits.clone(); Runtime::with_fs_store_and_execution_backend(options, backend) .map_err(ProcessError::Runtime) } _ => Err(ProcessError::usage( "unsupported Runtime catalog store selection".to_string(), )), } } fn runtime_options_from_http(config: &RuntimeHttpServerConfig) -> RuntimeOptions { RuntimeOptions { runtime_id: config.runtime_id.clone(), display_name: config.display_name.clone(), limits: config.limits.clone(), } } fn parse_args(args: I) -> Result, ProcessError> where I: IntoIterator, S: Into, { let mut config = ProcessConfig::default()?; let mut store = StoreArg::Memory; let mut args = args.into_iter().map(Into::into).collect::>(); while let Some(arg) = args.pop_front() { let (flag, inline_value) = split_flag_value(arg)?; match flag.as_str() { "--help" | "-h" => return Ok(None), "--bind" => { let value = take_value(&flag, inline_value, &mut args)?; config.http.bind_addr = value.parse::().map_err(|error| { ProcessError::usage(format!("invalid --bind socket address `{value}`: {error}")) })?; } "--runtime-id" => { let value = take_value(&flag, inline_value, &mut args)?; config.http.runtime_id = Some(RuntimeId::new(value).ok_or_else(|| { ProcessError::usage("--runtime-id must not be empty".to_string()) })?); } "--display-name" => { config.http.display_name = Some(take_value(&flag, inline_value, &mut args)?); } "--workspace" => { config.workspace_root = PathBuf::from(take_value(&flag, inline_value, &mut args)?); } "--cwd" => { config.cwd = PathBuf::from(take_value(&flag, inline_value, &mut args)?); } "--worker-store-dir" => { config.worker_store_dir = Some(PathBuf::from(take_value(&flag, inline_value, &mut args)?)); } "--worker-metadata-dir" => { config.worker_metadata_dir = Some(PathBuf::from(take_value(&flag, inline_value, &mut args)?)); } "--worker-runtime-base-dir" => { config.worker_runtime_base_dir = Some(PathBuf::from(take_value(&flag, inline_value, &mut args)?)); } "--profile" => { config.profile = Some(take_value(&flag, inline_value, &mut args)?); } "--store" => { let value = take_value(&flag, inline_value, &mut args)?; store = match value.as_str() { "memory" => StoreArg::Memory, "fs" | "fs-store" => StoreArg::Fs { root: None }, _ => { return Err(ProcessError::usage(format!( "unsupported --store `{value}`; expected `memory` or `fs`" ))); } }; } "--fs-root" => { let value = take_value(&flag, inline_value, &mut args)?; store = StoreArg::Fs { root: Some(PathBuf::from(value)), }; } "--local-token" => { let value = take_value(&flag, inline_value, &mut args)?; if value.is_empty() { return Err(ProcessError::usage( "--local-token must not be empty when provided".to_string(), )); } config.http.local_token = Some(value); } "--local-token-env" => { let name = take_value(&flag, inline_value, &mut args)?; let value = env::var(&name).map_err(|error| { ProcessError::usage(format!( "failed to read --local-token-env `{name}`: {error}" )) })?; if value.is_empty() { return Err(ProcessError::usage(format!( "--local-token-env `{name}` resolved to an empty value" ))); } config.http.local_token = Some(value); } "--max-transcript-projection-items" => { config.http.limits.max_transcript_projection_items = parse_usize_flag(&flag, take_value(&flag, inline_value, &mut args)?)?; } "--max-event-batch-items" => { config.http.limits.max_event_batch_items = parse_usize_flag(&flag, take_value(&flag, inline_value, &mut args)?)?; } _ => { return Err(ProcessError::usage(format!("unknown argument `{flag}`"))); } } } apply_store_selection(&mut config.http, store)?; config.workspace_root = normalize_workspace_path(config.workspace_root)?; if config.cwd.as_os_str().is_empty() { config.cwd = config.workspace_root.clone(); } else { config.cwd = normalize_child_path(&config.workspace_root, config.cwd); } Ok(Some(config)) } fn normalize_workspace_path(path: PathBuf) -> Result { let absolute = if path.is_absolute() { path } else { env::current_dir()?.join(path) }; Ok(absolute.canonicalize().unwrap_or(absolute)) } fn normalize_child_path(base: &std::path::Path, path: PathBuf) -> PathBuf { if path.is_absolute() { path.canonicalize().unwrap_or(path) } else { let absolute = base.join(path); absolute.canonicalize().unwrap_or(absolute) } } fn split_flag_value(arg: String) -> Result<(String, Option), ProcessError> { if !arg.starts_with('-') { return Err(ProcessError::usage(format!( "unexpected positional argument `{arg}`" ))); } if let Some((flag, value)) = arg.split_once('=') { Ok((flag.to_string(), Some(value.to_string()))) } else { Ok((arg, None)) } } fn take_value( flag: &str, inline_value: Option, args: &mut VecDeque, ) -> Result { if let Some(value) = inline_value { return Ok(value); } args.pop_front() .ok_or_else(|| ProcessError::usage(format!("{flag} requires a value"))) } fn parse_usize_flag(flag: &str, value: String) -> Result { value .parse::() .map_err(|error| ProcessError::usage(format!("invalid {flag} value `{value}`: {error}"))) } fn apply_store_selection( config: &mut RuntimeHttpServerConfig, store: StoreArg, ) -> Result<(), ProcessError> { match store { StoreArg::Memory => { config.store = RuntimeHttpStoreSelection::Memory; Ok(()) } StoreArg::Fs { root } => { let root = root.ok_or_else(|| { ProcessError::usage("--store fs requires --fs-root ".to_string()) })?; config.store = RuntimeHttpStoreSelection::Fs { root }; Ok(()) } } } #[derive(Clone, Debug, PartialEq, Eq)] struct ProcessConfig { http: RuntimeHttpServerConfig, workspace_root: PathBuf, cwd: PathBuf, worker_store_dir: Option, worker_metadata_dir: Option, worker_runtime_base_dir: Option, profile: Option, } impl ProcessConfig { fn default() -> Result { let workspace_root = env::current_dir()?; Ok(Self { http: RuntimeHttpServerConfig::default(), cwd: workspace_root.clone(), workspace_root, worker_store_dir: None, worker_metadata_dir: None, worker_runtime_base_dir: None, profile: None, }) } } #[derive(Clone, Debug, PartialEq, Eq)] enum StoreArg { Memory, Fs { root: Option }, } #[derive(Debug)] enum ProcessError { Usage(String), Server(RuntimeHttpServerError), Runtime(RuntimeError), WorkerAdapter(String), Io(std::io::Error), } impl ProcessError { fn usage(message: String) -> Self { Self::Usage(message) } } impl fmt::Display for ProcessError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::Usage(message) => message.fmt(f), Self::Server(error) => error.fmt(f), Self::Runtime(error) => error.fmt(f), Self::WorkerAdapter(error) => error.fmt(f), Self::Io(error) => error.fmt(f), } } } impl Error for ProcessError { fn source(&self) -> Option<&(dyn Error + 'static)> { match self { Self::Usage(_) | Self::WorkerAdapter(_) => None, Self::Server(error) => Some(error), Self::Runtime(error) => Some(error), Self::Io(error) => Some(error), } } } impl From for ProcessError { fn from(error: RuntimeHttpServerError) -> Self { Self::Server(error) } } impl From for ProcessError { fn from(error: std::io::Error) -> Self { Self::Io(error) } } fn usage() -> &'static str { "Usage: worker-runtime-rest-server [OPTIONS]\n\n\ Starts a worker-backed Runtime REST command API for a trusted backend/proxy.\n\ Browsers must not connect to this Runtime process directly.\n\n\ Options:\n\ --bind Bind socket address (default: 127.0.0.1:38800)\n\ --runtime-id Runtime authority id (default: generated)\n\ --display-name Runtime display name\n\ --workspace Workspace root used for spawned Workers (default: cwd)\n\ --cwd Process cwd used for spawned Workers (default: workspace)\n\ --profile Force spawned Workers to use a Profile selector\n\ --worker-store-dir Worker session store directory\n\ --worker-metadata-dir Worker metadata directory\n\ --worker-runtime-base-dir Worker controller runtime directory\n\ --store Runtime catalog store selection (default: memory)\n\ --fs-root Runtime catalog filesystem store root\n\ --local-token Minimal local bearer token placeholder\n\ --local-token-env Read local bearer token placeholder from env\n\ --max-transcript-projection-items Override transcript projection limit\n\ --max-event-batch-items Override event batch limit\n\ -h, --help Show this help" } #[cfg(test)] mod tests { use super::*; #[test] fn help_does_not_start_server() { assert!(parse_args(["--help"]).unwrap().is_none()); } #[test] fn default_bind_addr_is_stable_for_local_dev() { let config = parse_args([] as [&str; 0]).unwrap().unwrap(); assert_eq!( config.http.bind_addr, "127.0.0.1:38800".parse::().unwrap() ); } #[test] fn normalizes_relative_workspace_for_worker_spawn() { let current_dir = env::current_dir().unwrap().canonicalize().unwrap(); let config = parse_args(["--workspace", ".", "--cwd", "."]) .unwrap() .unwrap(); assert_eq!(config.workspace_root, current_dir); assert_eq!(config.cwd, current_dir); } #[test] fn parses_memory_runtime_process_config() { let config = parse_args([ "--bind", "127.0.0.1:0", "--runtime-id=runtime-alpha", "--display-name", "Runtime Alpha", "--workspace", "/tmp/workspace", "--cwd", "/tmp/workspace/subdir", "--profile", "builtin:coder", "--local-token-env", "TEST_RUNTIME_TOKEN", ]); assert!(config.is_err()); unsafe { env::set_var("TEST_RUNTIME_TOKEN", "secret-token"); } let config = parse_args([ "--bind", "127.0.0.1:0", "--runtime-id=runtime-alpha", "--display-name", "Runtime Alpha", "--workspace", "/tmp/workspace", "--cwd", "/tmp/workspace/subdir", "--profile", "builtin:coder", "--local-token-env", "TEST_RUNTIME_TOKEN", ]) .unwrap() .unwrap(); unsafe { env::remove_var("TEST_RUNTIME_TOKEN"); } assert_eq!(config.http.bind_addr, "127.0.0.1:0".parse().unwrap()); assert_eq!( config.http.runtime_id.as_ref().map(ToString::to_string), Some("runtime-alpha".to_string()) ); assert_eq!(config.http.display_name.as_deref(), Some("Runtime Alpha")); assert_eq!(config.workspace_root, PathBuf::from("/tmp/workspace")); assert_eq!(config.cwd, PathBuf::from("/tmp/workspace/subdir")); assert_eq!(config.profile.as_deref(), Some("builtin:coder")); assert_eq!(config.http.local_token.as_deref(), Some("secret-token")); } #[test] fn parses_fs_store_runtime_process_config() { let config = parse_args(["--store", "fs", "--fs-root", "/tmp/runtime-store"]) .unwrap() .unwrap(); assert!(matches!( config.http.store, RuntimeHttpStoreSelection::Fs { ref root } if root == &PathBuf::from("/tmp/runtime-store") )); } }