315 lines
10 KiB
Rust
315 lines
10 KiB
Rust
//! Background providers carry plain data and weak task identities, never a VM.
|
|
use endor_core::{VmError, WeakObjectRef};
|
|
use std::{
|
|
collections::HashMap,
|
|
fs::File,
|
|
io::{Read, Write},
|
|
process::{Child, Command, Stdio},
|
|
sync::{
|
|
Arc, Mutex, OnceLock,
|
|
atomic::{AtomicBool, AtomicUsize, Ordering},
|
|
mpsc::{SyncSender, sync_channel},
|
|
},
|
|
thread,
|
|
time::Duration,
|
|
};
|
|
|
|
const MAX_TEXT: usize = 1024 * 1024;
|
|
const MAX_ACTIVE: usize = 16;
|
|
const WORKERS: usize = 4;
|
|
const QUEUE: usize = 64;
|
|
type Completion = Result<String, String>;
|
|
|
|
#[derive(Debug, Default)]
|
|
struct Shared {
|
|
ready: Mutex<HashMap<WeakObjectRef, Completion>>,
|
|
active: AtomicUsize,
|
|
}
|
|
#[derive(Clone, Debug, Default)]
|
|
pub struct Provider {
|
|
shared: Arc<Shared>,
|
|
}
|
|
#[derive(Debug, Default)]
|
|
struct Control {
|
|
cancelled: AtomicBool,
|
|
process: Mutex<Option<Child>>,
|
|
}
|
|
impl Control {
|
|
fn cancel(&self) {
|
|
self.cancelled.store(true, Ordering::Release);
|
|
if let Ok(mut process) = self.process.lock()
|
|
&& let Some(child) = process.as_mut()
|
|
{
|
|
// Reaping belongs to the provider worker, never the VM thread.
|
|
let _ = child.kill();
|
|
}
|
|
}
|
|
}
|
|
#[derive(Debug)]
|
|
pub enum Request {
|
|
ReadText(String),
|
|
FetchText(String),
|
|
}
|
|
/// Only Task payloads own this lease. Provider threads hold plain control data.
|
|
#[derive(Debug)]
|
|
pub struct Lease {
|
|
shared: Arc<Shared>,
|
|
task: WeakObjectRef,
|
|
control: Arc<Control>,
|
|
}
|
|
impl Drop for Lease {
|
|
fn drop(&mut self) {
|
|
self.control.cancel();
|
|
if let Ok(mut ready) = self.shared.ready.lock() {
|
|
ready.remove(&self.task);
|
|
}
|
|
self.shared.active.fetch_sub(1, Ordering::AcqRel);
|
|
}
|
|
}
|
|
struct Job {
|
|
request: Request,
|
|
task: WeakObjectRef,
|
|
control: Arc<Control>,
|
|
shared: Arc<Shared>,
|
|
}
|
|
impl Provider {
|
|
pub fn submit(
|
|
&self,
|
|
task: WeakObjectRef,
|
|
request: Request,
|
|
replacing: bool,
|
|
) -> Result<Arc<Lease>, VmError> {
|
|
self.shared
|
|
.active
|
|
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |active| {
|
|
(active < MAX_ACTIVE + usize::from(replacing)).then_some(active + 1)
|
|
})
|
|
.map_err(|_| VmError::ExecutionLimit("16 I/O tasks per runtime"))?;
|
|
let lease = Arc::new(Lease {
|
|
shared: self.shared.clone(),
|
|
task,
|
|
control: Arc::new(Control::default()),
|
|
});
|
|
let sender = pool().map_err(|_| VmError::InvalidArgument("I/O provider unavailable"))?;
|
|
sender
|
|
.try_send(Job {
|
|
request,
|
|
task,
|
|
control: lease.control.clone(),
|
|
shared: self.shared.clone(),
|
|
})
|
|
.map_err(|_| VmError::ExecutionLimit("I/O provider queue full or unavailable"))?;
|
|
Ok(lease)
|
|
}
|
|
pub fn take_ready(&self) -> Result<Vec<(WeakObjectRef, Completion)>, VmError> {
|
|
Ok(self
|
|
.shared
|
|
.ready
|
|
.lock()
|
|
.map_err(|_| VmError::InvalidArgument("I/O mailbox poisoned"))?
|
|
.drain()
|
|
.collect())
|
|
}
|
|
}
|
|
fn pool() -> Result<&'static SyncSender<Job>, &'static str> {
|
|
static POOL: OnceLock<Result<SyncSender<Job>, String>> = OnceLock::new();
|
|
POOL.get_or_init(|| {
|
|
let (sender, receiver) = sync_channel::<Job>(QUEUE);
|
|
let receiver = Arc::new(Mutex::new(receiver));
|
|
for index in 0..WORKERS {
|
|
let receiver = receiver.clone();
|
|
thread::Builder::new()
|
|
.name(format!("tessera-io-{index}"))
|
|
.spawn(move || {
|
|
loop {
|
|
let job = match receiver.lock() {
|
|
Ok(receiver) => receiver.recv(),
|
|
Err(_) => break,
|
|
};
|
|
let Ok(job) = job else {
|
|
break;
|
|
};
|
|
if job.control.cancelled.load(Ordering::Acquire) {
|
|
continue;
|
|
}
|
|
let result = match &job.request {
|
|
Request::ReadText(path) => read_text(path),
|
|
Request::FetchText(url) => fetch_text(url, &job.control),
|
|
};
|
|
if let Ok(mut ready) = job.shared.ready.lock()
|
|
&& !job.control.cancelled.load(Ordering::Acquire)
|
|
{
|
|
ready.insert(job.task, result);
|
|
}
|
|
}
|
|
})
|
|
.map_err(|error| error.to_string())?;
|
|
}
|
|
Ok(sender)
|
|
})
|
|
.as_ref()
|
|
.map_err(|_| "I/O provider unavailable")
|
|
}
|
|
fn read_text(path: &str) -> Completion {
|
|
let metadata = std::fs::metadata(path).map_err(|error| error.to_string())?;
|
|
if !metadata.is_file() {
|
|
return Err("text source must be a regular file".into());
|
|
}
|
|
if metadata.len() > MAX_TEXT as u64 {
|
|
return Err("text source exceeds 1 MiB".into());
|
|
}
|
|
let file = File::open(path).map_err(|error| error.to_string())?;
|
|
if !file
|
|
.metadata()
|
|
.map_err(|error| error.to_string())?
|
|
.is_file()
|
|
{
|
|
return Err("text source must be a regular file".into());
|
|
}
|
|
let mut bytes = Vec::new();
|
|
file.take(MAX_TEXT as u64 + 1)
|
|
.read_to_end(&mut bytes)
|
|
.map_err(|error| error.to_string())?;
|
|
if bytes.len() > MAX_TEXT {
|
|
return Err("text source exceeds 1 MiB".into());
|
|
}
|
|
String::from_utf8(bytes).map_err(|_| "text source must be UTF-8".into())
|
|
}
|
|
|
|
/// No shell interpretation, curlrc, URL globbing, or non-HTTP redirects.
|
|
/// Authentication/custom headers are deliberately outside this text GET API.
|
|
pub fn valid_url(url: &str) -> bool {
|
|
if url.len() > 8192 || url.chars().any(char::is_control) {
|
|
return false;
|
|
}
|
|
let Some((scheme, rest)) = url.split_once("://") else {
|
|
return false;
|
|
};
|
|
if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
|
|
return false;
|
|
}
|
|
let authority = rest.split(['/', '?', '#']).next().unwrap_or("");
|
|
!authority.is_empty() && !authority.contains('@') && !authority.chars().any(char::is_whitespace)
|
|
}
|
|
|
|
/// Created only on an I/O worker. All early exits kill and reap the child.
|
|
struct Reaper<'a>(&'a Control);
|
|
impl Drop for Reaper<'_> {
|
|
fn drop(&mut self) {
|
|
let child = self.0.process.lock().ok().and_then(|mut slot| slot.take());
|
|
if let Some(mut child) = child {
|
|
let _ = child.kill();
|
|
let _ = child.wait();
|
|
}
|
|
}
|
|
}
|
|
fn fetch_text(url: &str, control: &Control) -> Completion {
|
|
if !valid_url(url) {
|
|
return Err("invalid HTTP(S) URL".into());
|
|
}
|
|
#[cfg(unix)]
|
|
let executable = "/usr/bin/curl";
|
|
#[cfg(not(unix))]
|
|
let executable = "curl.exe";
|
|
let mut child = Command::new(executable)
|
|
.args([
|
|
"--disable",
|
|
"--silent",
|
|
"--show-error",
|
|
"--globoff",
|
|
"--location",
|
|
"--max-redirs",
|
|
"5",
|
|
"--proto",
|
|
"=http,https",
|
|
"--proto-redir",
|
|
"=http,https",
|
|
"--max-time",
|
|
"30",
|
|
"--connect-timeout",
|
|
"10",
|
|
"--max-filesize",
|
|
"1048576",
|
|
"--compressed",
|
|
"--fail",
|
|
"--write-out",
|
|
"\n%{http_code}",
|
|
"--config",
|
|
"-",
|
|
])
|
|
.stdin(Stdio::piped())
|
|
.stdout(Stdio::piped())
|
|
.stderr(Stdio::null())
|
|
.spawn()
|
|
.map_err(|_| "HTTP transport unavailable: system curl is required".to_owned())?;
|
|
let mut input = child.stdin.take().expect("piped stdin");
|
|
let output = child.stdout.take().expect("piped stdout");
|
|
match control.process.lock() {
|
|
Ok(mut slot) => *slot = Some(child),
|
|
Err(_) => {
|
|
let _ = child.kill();
|
|
let _ = child.wait();
|
|
return Err("HTTP process lock poisoned".into());
|
|
}
|
|
}
|
|
let _reaper = Reaper(control);
|
|
if control.cancelled.load(Ordering::Acquire) {
|
|
return Err("request cancelled".into());
|
|
}
|
|
// URL stays out of process arguments. Reject controls before config escaping.
|
|
let escaped = url.replace('\\', "\\\\").replace('"', "\\\"");
|
|
writeln!(input, "url = \"{escaped}\"").map_err(|_| "HTTP transport input failed".to_owned())?;
|
|
drop(input);
|
|
let mut bytes = Vec::new();
|
|
output
|
|
.take(MAX_TEXT as u64 + 5)
|
|
.read_to_end(&mut bytes)
|
|
.map_err(|_| "HTTP response read failed".to_owned())?;
|
|
if bytes.len() > MAX_TEXT + 4 {
|
|
return Err("HTTP text exceeds 1 MiB".into());
|
|
}
|
|
let status = loop {
|
|
let mut slot = control
|
|
.process
|
|
.lock()
|
|
.map_err(|_| "HTTP process lock poisoned".to_owned())?;
|
|
let child = slot.as_mut().ok_or("HTTP process missing")?;
|
|
if let Some(status) = child
|
|
.try_wait()
|
|
.map_err(|_| "HTTP transport wait failed".to_owned())?
|
|
{
|
|
slot.take();
|
|
break status;
|
|
}
|
|
drop(slot);
|
|
thread::sleep(Duration::from_millis(2));
|
|
};
|
|
if control.cancelled.load(Ordering::Acquire) {
|
|
return Err("request cancelled".into());
|
|
}
|
|
let code = bytes
|
|
.len()
|
|
.checked_sub(4)
|
|
.and_then(|offset| (bytes[offset] == b'\n').then_some(&bytes[offset + 1..]))
|
|
.and_then(|digits| std::str::from_utf8(digits).ok())
|
|
.and_then(|digits| digits.parse::<u16>().ok());
|
|
if let Some(code) = code
|
|
&& code >= 300
|
|
{
|
|
return Err(format!("HTTP {code}"));
|
|
}
|
|
if !status.success() {
|
|
return Err(match status.code() {
|
|
Some(28) => "HTTP request timed out".into(),
|
|
Some(63) => "HTTP text exceeds 1 MiB".into(),
|
|
Some(code) => format!("HTTP transport failed (curl {code})"),
|
|
None => "HTTP transport terminated".into(),
|
|
});
|
|
}
|
|
if !code.is_some_and(|code| (200..300).contains(&code)) {
|
|
return Err("HTTP response status unavailable".into());
|
|
}
|
|
bytes.truncate(bytes.len() - 4);
|
|
String::from_utf8(bytes).map_err(|_| "HTTP text must be UTF-8".into())
|
|
}
|