681 lines
22 KiB
Rust
681 lines
22 KiB
Rust
mod guide;
|
|
mod worker;
|
|
|
|
use guide::agent_guide;
|
|
use worker::{PendingRequest, RequestProgress, Worker};
|
|
|
|
use std::{
|
|
io::Read,
|
|
ops::{Deref, DerefMut},
|
|
panic::{AssertUnwindSafe, catch_unwind},
|
|
sync::{
|
|
Arc, RwLock,
|
|
mpsc::{Receiver, SyncSender, TrySendError, sync_channel},
|
|
},
|
|
sync::{
|
|
Condvar, Mutex,
|
|
atomic::{AtomicBool, Ordering},
|
|
},
|
|
thread,
|
|
};
|
|
|
|
use anyhow::{Context as _, Result};
|
|
use tiny_http::{Header, Method, Request, Response, Server, StatusCode};
|
|
use winit::event_loop::EventLoopProxy;
|
|
|
|
use crate::{
|
|
activity::{Activity, ActivityLog},
|
|
app::AppEvent,
|
|
document::NamespaceState,
|
|
endor_host,
|
|
};
|
|
|
|
pub const API_ADDRESS: &str = "127.0.0.1:4737";
|
|
const MAX_CODE_SIZE: u64 = endor_host::MAX_SOURCE_BYTES as u64;
|
|
type Notify = Arc<dyn Fn() + Send + Sync>;
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub struct ShutdownOutcome {
|
|
pub summary: String,
|
|
pub error: Option<String>,
|
|
}
|
|
type ShutdownCompletion = Arc<(Mutex<Option<ShutdownOutcome>>, Condvar)>;
|
|
#[derive(Clone)]
|
|
pub struct RuntimeControl {
|
|
sender: SyncSender<()>,
|
|
requested: Arc<AtomicBool>,
|
|
completion: ShutdownCompletion,
|
|
}
|
|
impl RuntimeControl {
|
|
pub fn request_shutdown(&self) -> bool {
|
|
if self.requested.swap(true, Ordering::AcqRel) {
|
|
return true;
|
|
}
|
|
if self.sender.try_send(()).is_ok() {
|
|
true
|
|
} else {
|
|
self.requested.store(false, Ordering::Release);
|
|
false
|
|
}
|
|
}
|
|
pub fn is_exiting(&self) -> bool {
|
|
self.requested.load(Ordering::Acquire)
|
|
}
|
|
pub fn wait_shutdown(&self, timeout: std::time::Duration) -> Option<ShutdownOutcome> {
|
|
let (lock, ready) = &*self.completion;
|
|
let result = lock.lock().ok()?;
|
|
let (result, _) = ready
|
|
.wait_timeout_while(result, timeout, |result| result.is_none())
|
|
.ok()?;
|
|
result.clone()
|
|
}
|
|
}
|
|
struct Shutdown {
|
|
receiver: Receiver<()>,
|
|
control: RuntimeControl,
|
|
proxy: EventLoopProxy<AppEvent>,
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
|
enum RequestKind {
|
|
Event,
|
|
Code,
|
|
Expression,
|
|
Repl,
|
|
ResetRepl,
|
|
}
|
|
|
|
struct BuildRequest {
|
|
request: ObservedRequest,
|
|
kind: RequestKind,
|
|
source: String,
|
|
namespace: String,
|
|
submission: u64,
|
|
}
|
|
|
|
pub fn start(
|
|
namespace: Arc<RwLock<NamespaceState>>,
|
|
proxy: EventLoopProxy<AppEvent>,
|
|
) -> Result<(Activity, RuntimeControl)> {
|
|
let (sender, receiver) = sync_channel(1);
|
|
let control = RuntimeControl {
|
|
sender,
|
|
requested: Arc::new(AtomicBool::new(false)),
|
|
completion: Arc::new((Mutex::new(None), Condvar::new())),
|
|
};
|
|
let shutdown = Shutdown {
|
|
receiver,
|
|
control: control.clone(),
|
|
proxy: proxy.clone(),
|
|
};
|
|
let server = Server::http(API_ADDRESS)
|
|
.map_err(|error| anyhow::anyhow!(error))
|
|
.with_context(|| format!("bind Tessera API to {API_ADDRESS}"))?;
|
|
let notify: Notify = Arc::new(move || {
|
|
let _ = proxy.send_event(AppEvent::DocumentUpdated);
|
|
});
|
|
let activity = Arc::new(Mutex::new(ActivityLog::default()));
|
|
start_server_observed(
|
|
Arc::new(server),
|
|
namespace,
|
|
notify.clone(),
|
|
activity.clone(),
|
|
notify,
|
|
Some(shutdown),
|
|
)?;
|
|
Ok((activity, control))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
fn start_server(
|
|
server: Arc<Server>,
|
|
namespace: Arc<RwLock<NamespaceState>>,
|
|
notify: Notify,
|
|
) -> Result<thread::JoinHandle<()>> {
|
|
start_server_observed(
|
|
server,
|
|
namespace,
|
|
notify,
|
|
Arc::new(Mutex::new(ActivityLog::default())),
|
|
Arc::new(|| {}),
|
|
None,
|
|
)
|
|
}
|
|
|
|
fn start_server_observed(
|
|
server: Arc<Server>,
|
|
namespace: Arc<RwLock<NamespaceState>>,
|
|
notify: Notify,
|
|
activity: Activity,
|
|
observer_notify: Notify,
|
|
shutdown: Option<Shutdown>,
|
|
) -> Result<thread::JoinHandle<()>> {
|
|
let (sender, receiver) = sync_channel::<BuildRequest>(8);
|
|
let worker_namespace = namespace.clone();
|
|
let worker_notify = notify.clone();
|
|
let exiting = shutdown
|
|
.as_ref()
|
|
.map(|shutdown| shutdown.control.requested.clone());
|
|
thread::Builder::new()
|
|
.name("tessera-endor-worker".into())
|
|
.spawn(move || {
|
|
let mut worker = Worker::default();
|
|
let mut pending: Option<(BuildRequest, PendingRequest)> = None;
|
|
loop {
|
|
if let Some(shutdown) = &shutdown
|
|
&& shutdown.receiver.try_recv().is_ok()
|
|
{
|
|
if let Some((job, input)) = pending.take() {
|
|
drop(input);
|
|
let _ = job
|
|
.request
|
|
.respond(error_response(StatusCode(503), "application is exiting"));
|
|
}
|
|
let outcome = worker.shutdown_runtime();
|
|
while let Ok(job) = receiver.try_recv() {
|
|
let _ = job
|
|
.request
|
|
.respond(error_response(StatusCode(503), "application is exiting"));
|
|
}
|
|
if let Ok(mut current) = worker_namespace.write() {
|
|
current.runtime_session = 0;
|
|
current.window_commands.clear();
|
|
current.document.frame.session = 0;
|
|
current.document.frame.focus = None;
|
|
for view in current.document.windows.values_mut() {
|
|
view.frame.session = 0;
|
|
view.frame.focus = None;
|
|
}
|
|
}
|
|
eprintln!("{}", outcome.summary);
|
|
let (lock, ready) = &*shutdown.control.completion;
|
|
if let Ok(mut result) = lock.lock() {
|
|
*result = Some(outcome.clone());
|
|
ready.notify_all();
|
|
}
|
|
let _ = shutdown.proxy.send_event(AppEvent::ShutdownComplete {
|
|
error: outcome.error,
|
|
});
|
|
break;
|
|
}
|
|
if let Some((job, input)) = pending.take() {
|
|
match worker.resume_request(&job, input, &worker_namespace, &worker_notify) {
|
|
RequestProgress::Pending(input) => {
|
|
pending = Some((job, input));
|
|
// A soft VM quantum is a control boundary, not a
|
|
// minimum-duration time slice. Yield to other ready
|
|
// threads without charging a timer delay per quantum.
|
|
thread::yield_now();
|
|
}
|
|
RequestProgress::Complete(response) => {
|
|
let _ = job.request.respond(response);
|
|
}
|
|
}
|
|
continue;
|
|
}
|
|
worker.refresh_viewport(&worker_namespace, &worker_notify);
|
|
let wait = if worker.repl.as_ref().is_some_and(|(_, runtime)| {
|
|
runtime.has_pending_refresh() || runtime.has_cooperative_tasks()
|
|
}) {
|
|
std::time::Duration::ZERO
|
|
} else {
|
|
std::time::Duration::from_millis(100)
|
|
};
|
|
match receiver.recv_timeout(wait) {
|
|
Ok(job) => {
|
|
match worker.begin_request(&job, &worker_namespace, &worker_notify) {
|
|
RequestProgress::Pending(input) => {
|
|
if matches!(&input, PendingRequest::BeforeInput) {
|
|
eprintln!(
|
|
"Tessera worker: submission {} waits for active refresh",
|
|
job.submission
|
|
);
|
|
}
|
|
pending = Some((job, input));
|
|
}
|
|
RequestProgress::Complete(response) => {
|
|
let _ = job.request.respond(response);
|
|
}
|
|
}
|
|
}
|
|
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
|
|
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
|
|
if worker.repl.as_ref().is_some_and(|(epoch, runtime)| {
|
|
worker_namespace.read().is_ok_and(|current| {
|
|
current.submission != *epoch
|
|
&& current.runtime_session != runtime.id
|
|
})
|
|
}) {
|
|
worker.repl = None;
|
|
}
|
|
if wait.is_zero() {
|
|
thread::yield_now();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
})
|
|
.context("start Aster execution worker")?;
|
|
thread::Builder::new()
|
|
.name("tessera-agent-api".to_owned())
|
|
.spawn(move || {
|
|
for request in server.incoming_requests() {
|
|
if exiting
|
|
.as_ref()
|
|
.is_some_and(|flag| flag.load(Ordering::Acquire))
|
|
&& !matches!(request.method(), Method::Get | Method::Options)
|
|
{
|
|
let _ =
|
|
request.respond(error_response(StatusCode(503), "application is exiting"));
|
|
continue;
|
|
}
|
|
handle(
|
|
request,
|
|
&namespace,
|
|
¬ify,
|
|
&sender,
|
|
&activity,
|
|
&observer_notify,
|
|
);
|
|
}
|
|
})
|
|
.context("start Tessera agent API thread")
|
|
}
|
|
|
|
// Capture only allowlisted headers: credentials and cookies never enter the inspector.
|
|
struct ObservedRequest {
|
|
inner: Request,
|
|
activity: Activity,
|
|
id: u64,
|
|
notify: Notify,
|
|
}
|
|
impl ObservedRequest {
|
|
fn new(inner: Request, activity: Activity, notify: Notify) -> Self {
|
|
let headers = inner
|
|
.headers()
|
|
.iter()
|
|
.filter(|h| h.field.equiv("Content-Type") || h.field.equiv("Content-Length"))
|
|
.map(ToString::to_string)
|
|
.collect::<Vec<_>>()
|
|
.join("\n");
|
|
let origin = if inner
|
|
.headers()
|
|
.iter()
|
|
.any(|h| h.field.equiv("X-Tessera-Origin") && h.value.as_str() == "console")
|
|
{
|
|
"Console"
|
|
} else {
|
|
"HTTP"
|
|
}
|
|
.to_owned();
|
|
let id = activity.lock().unwrap().begin(
|
|
inner.method().to_string(),
|
|
inner.url().to_owned(),
|
|
origin,
|
|
headers,
|
|
);
|
|
notify();
|
|
Self {
|
|
inner,
|
|
activity,
|
|
id,
|
|
notify,
|
|
}
|
|
}
|
|
fn record_body(&self, body: &str) {
|
|
self.activity.lock().unwrap().body(self.id, body);
|
|
(self.notify)();
|
|
}
|
|
fn respond(self, response: Response<std::io::Cursor<Vec<u8>>>) -> std::io::Result<()> {
|
|
let status = response.status_code();
|
|
let headers = response.headers().to_vec();
|
|
let length = response.data_length();
|
|
let reader = response.into_reader();
|
|
self.activity.lock().unwrap().finish(
|
|
self.id,
|
|
status.0,
|
|
&String::from_utf8_lossy(reader.get_ref()),
|
|
headers
|
|
.iter()
|
|
.map(ToString::to_string)
|
|
.collect::<Vec<_>>()
|
|
.join("\n"),
|
|
);
|
|
(self.notify)();
|
|
self.inner
|
|
.respond(Response::new(status, headers, reader, length, None))
|
|
}
|
|
}
|
|
impl Deref for ObservedRequest {
|
|
type Target = Request;
|
|
fn deref(&self) -> &Request {
|
|
&self.inner
|
|
}
|
|
}
|
|
impl DerefMut for ObservedRequest {
|
|
fn deref_mut(&mut self) -> &mut Request {
|
|
&mut self.inner
|
|
}
|
|
}
|
|
|
|
fn handle(
|
|
request: Request,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
sender: &SyncSender<BuildRequest>,
|
|
activity: &Activity,
|
|
observer_notify: &Notify,
|
|
) {
|
|
let request = ObservedRequest::new(request, activity.clone(), observer_notify.clone());
|
|
let method = request.method().clone();
|
|
let path = request
|
|
.url()
|
|
.split('?')
|
|
.next()
|
|
.unwrap_or(request.url())
|
|
.to_owned();
|
|
|
|
let evaluation = namespace_from_route(&path, "/evaluate");
|
|
let repl = namespace_from_route(&path, "/repl");
|
|
let event = namespace_from_route(&path, "/events");
|
|
let route = if method == Method::Put {
|
|
namespace_from_path(&path).map(|name| (name, RequestKind::Code))
|
|
} else if method == Method::Post {
|
|
event
|
|
.map(|name| (name, RequestKind::Event))
|
|
.or_else(|| repl.map(|name| (name, RequestKind::Repl)))
|
|
.or(evaluation.map(|name| (name, RequestKind::Expression)))
|
|
} else if method == Method::Delete {
|
|
repl.map(|name| (name, RequestKind::ResetRepl))
|
|
} else {
|
|
None
|
|
};
|
|
if let Some((addressed, kind)) = route {
|
|
submit(request, addressed, namespace, sender, kind);
|
|
return;
|
|
}
|
|
let response = if method == Method::Get && path == "/agent" {
|
|
match namespace.read() {
|
|
Ok(namespace) => markdown(StatusCode(200), agent_guide(&namespace)),
|
|
Err(_) => error_response(StatusCode(500), "namespace lock is poisoned"),
|
|
}
|
|
} else if let Some(addressed_namespace) = namespace_from_path(&path).map(str::to_owned) {
|
|
code_request(method, &addressed_namespace, namespace, notify)
|
|
} else if method == Method::Options {
|
|
text(StatusCode(204), "", "text/plain")
|
|
} else {
|
|
error_response(StatusCode(404), "route not found; start with GET /agent")
|
|
};
|
|
|
|
let _ = request.respond(response);
|
|
}
|
|
|
|
fn code_request(
|
|
method: Method,
|
|
addressed_namespace: &str,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
) -> Response<std::io::Cursor<Vec<u8>>> {
|
|
let Ok(current) = namespace.read() else {
|
|
return error_response(StatusCode(500), "namespace lock is poisoned");
|
|
};
|
|
if addressed_namespace != current.name {
|
|
return error_response(
|
|
StatusCode(404),
|
|
format!(
|
|
"namespace `{addressed_namespace}` is not active; current namespace is `{}`",
|
|
current.name
|
|
),
|
|
);
|
|
}
|
|
|
|
if method == Method::Get {
|
|
let mut response = text(
|
|
StatusCode(200),
|
|
¤t.document.source,
|
|
"text/plain; charset=utf-8",
|
|
);
|
|
add_header(
|
|
&mut response,
|
|
"X-Tessera-Revision",
|
|
¤t.revision.to_string(),
|
|
);
|
|
add_header(
|
|
&mut response,
|
|
"X-Tessera-Session",
|
|
¤t.document.frame.session.to_string(),
|
|
);
|
|
add_header(
|
|
&mut response,
|
|
"X-Tessera-Window",
|
|
¤t.document.frame.window.to_string(),
|
|
);
|
|
add_header(
|
|
&mut response,
|
|
"X-Tessera-Layout-Version",
|
|
¤t.document.frame.layout_version.to_string(),
|
|
);
|
|
if let Some(viewport) = current.viewport {
|
|
add_header(
|
|
&mut response,
|
|
"X-Tessera-Viewport-Width",
|
|
&viewport.width.to_string(),
|
|
);
|
|
add_header(
|
|
&mut response,
|
|
"X-Tessera-Viewport-Height",
|
|
&viewport.height.to_string(),
|
|
);
|
|
}
|
|
add_header(
|
|
&mut response,
|
|
"X-Tessera-Source-Kind",
|
|
if current
|
|
.document
|
|
.root
|
|
.as_ref()
|
|
.is_some_and(|root| root.kind == "repl-session")
|
|
{
|
|
"repl-transcript"
|
|
} else {
|
|
"source-snapshot"
|
|
},
|
|
);
|
|
return response;
|
|
}
|
|
drop(current);
|
|
|
|
match method {
|
|
Method::Delete => match namespace.write() {
|
|
Ok(mut current) => {
|
|
let revision = match current.clear() {
|
|
Ok(revision) => revision,
|
|
Err(error) => return error_response(StatusCode(500), error.to_string()),
|
|
};
|
|
drop(current);
|
|
notify();
|
|
json(
|
|
StatusCode(200),
|
|
serde_json::json!({
|
|
"ok": true,
|
|
"namespace": addressed_namespace,
|
|
"revision": revision
|
|
}),
|
|
)
|
|
}
|
|
Err(_) => error_response(StatusCode(500), "namespace lock is poisoned"),
|
|
},
|
|
Method::Options => text(StatusCode(204), "", "text/plain"),
|
|
_ => error_response(StatusCode(405), "code supports GET, PUT, and DELETE"),
|
|
}
|
|
}
|
|
|
|
fn submit(
|
|
mut request: ObservedRequest,
|
|
addressed: &str,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
sender: &SyncSender<BuildRequest>,
|
|
kind: RequestKind,
|
|
) {
|
|
let source = match read_code(&mut request) {
|
|
Ok(source) => source,
|
|
Err(error) => {
|
|
let _ = request.respond(error_response(StatusCode(400), error.to_string()));
|
|
return;
|
|
}
|
|
};
|
|
request.record_body(&source);
|
|
let Ok(mut current) = namespace.write() else {
|
|
let _ = request.respond(error_response(
|
|
StatusCode(500),
|
|
"namespace lock is poisoned",
|
|
));
|
|
return;
|
|
};
|
|
if current.name != addressed {
|
|
let _ = request.respond(error_response(StatusCode(404), "namespace is not active"));
|
|
return;
|
|
}
|
|
let submission = match if kind == RequestKind::Code {
|
|
current.next_submission()
|
|
} else {
|
|
Ok(current.submission)
|
|
} {
|
|
Ok(id) => id,
|
|
Err(error) => {
|
|
let _ = request.respond(error_response(StatusCode(500), error.to_string()));
|
|
return;
|
|
}
|
|
};
|
|
let job = BuildRequest {
|
|
request,
|
|
kind,
|
|
source,
|
|
namespace: addressed.into(),
|
|
submission,
|
|
};
|
|
match sender.try_send(job) {
|
|
// A rejected queue entry never invalidates accepted work. The worker
|
|
// uses this same lock when committing, so publication is atomic.
|
|
Ok(()) => {
|
|
if kind == RequestKind::Code {
|
|
current.submission = submission;
|
|
}
|
|
}
|
|
Err(TrySendError::Full(job)) => {
|
|
let _ = job
|
|
.request
|
|
.respond(error_response(StatusCode(503), "build queue is full"));
|
|
}
|
|
Err(TrySendError::Disconnected(job)) => {
|
|
let _ = job.request.respond(error_response(
|
|
StatusCode(503),
|
|
"execution worker is unavailable",
|
|
));
|
|
}
|
|
}
|
|
}
|
|
|
|
fn superseded(submission: u64) -> Response<std::io::Cursor<Vec<u8>>> {
|
|
json(
|
|
StatusCode(409),
|
|
serde_json::json!({ "ok": false, "submission": submission,
|
|
"error": "submission superseded by a newer PUT or DELETE" }),
|
|
)
|
|
}
|
|
|
|
fn namespace_from_route<'a>(path: &'a str, suffix: &str) -> Option<&'a str> {
|
|
path.strip_prefix("/v1/namespaces/")?
|
|
.strip_suffix(suffix)
|
|
.filter(|namespace| !namespace.is_empty() && !namespace.contains('/'))
|
|
}
|
|
fn namespace_from_path(path: &str) -> Option<&str> {
|
|
namespace_from_route(path, "/code")
|
|
}
|
|
|
|
fn read_code(request: &mut Request) -> Result<String> {
|
|
if request
|
|
.body_length()
|
|
.is_some_and(|length| length as u64 > MAX_CODE_SIZE)
|
|
{
|
|
anyhow::bail!("code exceeds the 1 MiB limit");
|
|
}
|
|
let mut code = String::new();
|
|
request
|
|
.as_reader()
|
|
.take(MAX_CODE_SIZE + 1)
|
|
.read_to_string(&mut code)
|
|
.context("read code")?;
|
|
if code.len() as u64 > MAX_CODE_SIZE {
|
|
anyhow::bail!("code exceeds the 1 MiB limit");
|
|
}
|
|
Ok(code)
|
|
}
|
|
|
|
fn compile_error(error: anyhow::Error) -> Response<std::io::Cursor<Vec<u8>>> {
|
|
let mut diagnostic =
|
|
serde_json::json!({ "severity": "error", "message": format!("{error:#}") });
|
|
if let Some(source) = error.downcast_ref::<frontend::SourceDiagnostic>() {
|
|
diagnostic["line"] = source.location.line.into();
|
|
diagnostic["column"] = source.location.column.into();
|
|
diagnostic["start_byte"] = source.diagnostic.span.start.into();
|
|
diagnostic["end_byte"] = source.diagnostic.span.end.into();
|
|
}
|
|
json(
|
|
StatusCode(422),
|
|
serde_json::json!({
|
|
"ok": false,
|
|
"diagnostics": [diagnostic]
|
|
}),
|
|
)
|
|
}
|
|
|
|
fn error_response(
|
|
status: StatusCode,
|
|
message: impl Into<String>,
|
|
) -> Response<std::io::Cursor<Vec<u8>>> {
|
|
json(status, serde_json::json!({ "error": message.into() }))
|
|
}
|
|
|
|
fn json(status: StatusCode, value: serde_json::Value) -> Response<std::io::Cursor<Vec<u8>>> {
|
|
text(
|
|
status,
|
|
&value.to_string(),
|
|
"application/json; charset=utf-8",
|
|
)
|
|
}
|
|
|
|
fn markdown(status: StatusCode, value: String) -> Response<std::io::Cursor<Vec<u8>>> {
|
|
text(status, &value, "text/markdown; charset=utf-8")
|
|
}
|
|
|
|
fn text(status: StatusCode, value: &str, content_type: &str) -> Response<std::io::Cursor<Vec<u8>>> {
|
|
let mut response = Response::from_string(value).with_status_code(status);
|
|
add_header(&mut response, "Content-Type", content_type);
|
|
add_header(
|
|
&mut response,
|
|
"Access-Control-Allow-Origin",
|
|
"http://127.0.0.1",
|
|
);
|
|
add_header(
|
|
&mut response,
|
|
"Access-Control-Allow-Methods",
|
|
"GET, PUT, POST, DELETE, OPTIONS",
|
|
);
|
|
add_header(
|
|
&mut response,
|
|
"Access-Control-Allow-Headers",
|
|
"Content-Type, X-Tessera-Origin",
|
|
);
|
|
response
|
|
}
|
|
|
|
fn add_header(response: &mut Response<std::io::Cursor<Vec<u8>>>, name: &str, value: &str) {
|
|
if let Ok(header) = Header::from_bytes(name, value) {
|
|
response.add_header(header);
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests;
|