915 lines
44 KiB
Rust
915 lines
44 KiB
Rust
//! 本机 HTTP API 入口：路由请求、限制输入大小、观察活动，并将执行工作送入串行 worker。
|
||
//! HTTP 线程不操作 VM；共享 NamespaceState 只保存最近完整发布文档和资源命令。提交序号用来拒绝已被较新 PUT/DELETE 取代的结果。
|
||
//! 关闭流程先请求运行时有界清理，再通过完成通知让原生事件循环退出。
|
||
//!
|
||
//! main::main 调用 start，shell 持有 RuntimeControl；HTTP 请求经 handle/submit 入队，原生输入经 dispatch_native 入同一队列。
|
||
//! worker 子模块拥有实际 ReplSession；本文件的线程循环只决定继续哪个请求、何时等待、如何响应和退出。
|
||
//! 本地端点、观察器和 EventLoopProxy 都是 Tessera 宿主适配，不能把整个文件直接作为无 GUI 的 Nexus 核心。
|
||
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, OnceLock, 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,
|
||
};
|
||
|
||
// 只监听 IPv4 回环固定端口，供本机 agent/Console HTTP 调用；原生输入现在直接走同一 worker 队列。
|
||
pub const API_ADDRESS: &str = "127.0.0.1:4737";
|
||
// 请求流读取与 endor_host 的源预算一致，Content-Length 检查不能替代流上的硬上限。
|
||
const MAX_CODE_SIZE: u64 = endor_host::MAX_SOURCE_BYTES as u64;
|
||
// GUI 更新回调只投递原生通知，不携带 VM/托管引用，也不在调用线程直接绘制窗口。
|
||
type Notify = Arc<dyn Fn() + Send + Sync>;
|
||
|
||
#[derive(Clone, Debug)]
|
||
/// 运行时清理的摘要与可选错误，供最终原生退出结果报告。
|
||
pub struct ShutdownOutcome {
|
||
// worker 生成的清理数量/故障摘要，供退出日志与 shell 最终报告使用。
|
||
pub summary: String,
|
||
// 清理失败时保留文本，不能因为退出事件已经发出就把失败说成正常关闭。
|
||
pub error: Option<String>,
|
||
}
|
||
// 完成槽和条件变量共享同一生命周期，退出发起者可有限等待或由 GUI ShutdownComplete 事件唤醒。
|
||
type ShutdownCompletion = Arc<(Mutex<Option<ShutdownOutcome>>, Condvar)>;
|
||
#[derive(Clone)]
|
||
/// 跨线程的退出控制句柄，使用有界通知和完成条件变量，不携带 VM。
|
||
pub struct RuntimeControl {
|
||
// 独立容量一的关闭通道，即使正常输入队列满了也能通知 worker 优先清理。
|
||
sender: SyncSender<()>,
|
||
// 合并重复退出请求，并让 HTTP/原生输入尽早拒绝新执行工作。
|
||
requested: Arc<AtomicBool>,
|
||
// worker 在退出收尾后填写结果，wait_shutdown 不在 GUI 内自行执行清理。
|
||
completion: ShutdownCompletion,
|
||
// start_server_observed 一次性装入共享执行队列，不能让每个原生窗口自行创建另一个 VM worker。
|
||
native_sender: Arc<OnceLock<SyncSender<BuildRequest>>>,
|
||
// 只读当前 namespace/submission 以给原生 job 盖身份，不包含 VM 对象图。
|
||
namespace: Arc<RwLock<NamespaceState>>,
|
||
}
|
||
impl RuntimeControl {
|
||
/// events::Client 的后台 relay 调用，向共享 worker 队列发送原生信封并等回执；GUI 仍只非阻塞发给 Client。
|
||
pub(crate) fn dispatch_native(&self, input: crate::events::Input) -> Result<()> {
|
||
// 检查退出门闩后才创建回执/工作，不让已关闭域继续排入源回调。
|
||
anyhow::ensure!(!self.is_exiting(), "application is exiting");
|
||
// 一个输入一个回执，relay 等待不会阻塞 GUI；容量一足以让 worker 写回后继续自身循环。
|
||
let (done, completed) = sync_channel(1);
|
||
// 短读锁只复制当前接受身份，发送/等待都在锁外，worker 发布时仍可取得写锁。
|
||
let job = {
|
||
// 原生来源由构造类型决定，HTTP 头或 JSON 不能伪造 RequestReply::Native。
|
||
let state = self
|
||
.namespace
|
||
.read()
|
||
.map_err(|_| anyhow::anyhow!("namespace lock is poisoned"))?;
|
||
BuildRequest {
|
||
request: RequestReply::Native(done),
|
||
kind: RequestKind::Event,
|
||
source: String::new(),
|
||
namespace: state.name.clone(),
|
||
submission: state.submission,
|
||
native: Some(input),
|
||
}
|
||
};
|
||
self.native_sender
|
||
.get()
|
||
.ok_or_else(|| anyhow::anyhow!("native input queue is unavailable"))?
|
||
.send(job)
|
||
.map_err(|_| anyhow::anyhow!("native input worker stopped"))?;
|
||
completed
|
||
.recv()
|
||
.map_err(|_| anyhow::anyhow!("native input acknowledgement stopped"))?
|
||
.map_err(anyhow::Error::msg)
|
||
}
|
||
|
||
/// shell 关闭流程调用，只通知一次，由 owning worker 清理 VM；true 表示已排入或此前已请求，不等于已清理成功。
|
||
pub fn request_shutdown(&self) -> bool {
|
||
// 原子标志合并重复退出请求；已请求直接返回 true，不在 GUI 线程重复排队清理。
|
||
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
|
||
}
|
||
}
|
||
/// shell/原生 relay 查询退出请求状态，不读取完成槽，也不能由此推断资源已全部清理。
|
||
pub fn is_exiting(&self) -> bool {
|
||
self.requested.load(Ordering::Acquire)
|
||
}
|
||
/// shell 最终退出收尾调用，有限等待 worker 保存的 ShutdownOutcome；None 为未取得结果，不是成功。
|
||
pub fn wait_shutdown(&self, timeout: std::time::Duration) -> Option<ShutdownOutcome> {
|
||
// 同一锁保护可选结果与等待谓词，避免“检查为空后错过通知”的竞争。
|
||
let (lock, ready) = &*self.completion;
|
||
// 锁异常直接返回 None，不在未知完成状态下继续阻塞。
|
||
let result = lock.lock().ok()?;
|
||
// wait_timeout_while 反复检查 Option，返回前重新拿锁，克隆纯文本结果交还 shell。
|
||
let (result, _) = ready
|
||
// 条件变量处理虚假唤醒，直到完成槽非空或超时；返回 None 也可能来自锁中毒，不能当作清理成功。
|
||
.wait_timeout_while(result, timeout, |result| result.is_none())
|
||
.ok()?;
|
||
result.clone()
|
||
}
|
||
}
|
||
/// worker 持有的关闭接收端、控制状态和 GUI 完成通知代理。
|
||
struct Shutdown {
|
||
// worker 每个控制边界 try_recv，关闭信号优先于下一片 VM 刷新或下一条普通输入。
|
||
receiver: Receiver<()>,
|
||
// 保存完成槽及已请求门闩，与 shell 手中的句柄是同一共享状态。
|
||
control: RuntimeControl,
|
||
// 完成后向原生事件循环发消息，worker 不直接调用窗口退出/绘制 API。
|
||
proxy: EventLoopProxy<AppEvent>,
|
||
}
|
||
|
||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||
/// 送入执行队列的请求类别，区分完整源码、持久输入、事件与重置。
|
||
enum RequestKind {
|
||
// 原生或 HTTP 输入，均由已有运行时执行，不启动新的程序堆。
|
||
Event,
|
||
// 完整程序 PUT：候选成功后才替换旧域和发布文档。
|
||
Code,
|
||
// 隔离表达式求值，不写活程序或 Console 的 globals。
|
||
Expression,
|
||
// 在真实持久 globals 上执行一轮交互，运行错误不回滚此前成功语句。
|
||
Repl,
|
||
// 显式清理/替换 Console 上下文，与完整 DELETE/code 的文档清空不同。
|
||
ResetRepl,
|
||
}
|
||
|
||
/// submit/dispatch_native 产生的纯数据工作信封，保留回应端和接受身份，worker 完成后才回应或确认过期。
|
||
struct BuildRequest {
|
||
// 原 HTTP 请求的观察包装或原生一次性回执，不能 Clone 成多次响应。
|
||
request: RequestReply,
|
||
// 原生来源只由进程内构造，HTTP 输入的事件 JSON 仍保存在 source 中由 worker 解码。
|
||
native: Option<crate::events::Input>,
|
||
// 执行语义标签，不从请求体猜是替换程序还是追加 Console。
|
||
kind: RequestKind,
|
||
// 已限制字节数并验证 UTF-8 的正文；原生事件无需字符串往返因此为空。
|
||
source: String,
|
||
// 路由所指 namespace，执行/发布时须仍匹配当前共享状态。
|
||
namespace: String,
|
||
// 完整提交的代际号，worker 核验后才允许替换发布值，旧结果不能覆盖新 PUT/DELETE。
|
||
submission: u64,
|
||
}
|
||
|
||
/// main::main 初始化时调用，绑定本机 API 并启动两个线程，返回 shell 使用的活动记录和 RuntimeControl。
|
||
pub fn start(
|
||
namespace: Arc<RwLock<NamespaceState>>,
|
||
proxy: EventLoopProxy<AppEvent>,
|
||
) -> Result<(Activity, RuntimeControl)> {
|
||
// 退出控制仅需一个消息槽，正常执行请求另走容量8队列，避免工作排满后无法通知关闭。
|
||
let (sender, receiver) = sync_channel(1);
|
||
// HTTP、shell 和 worker 共享请求/完成门闩，原生队列指针由实际线程启动函数一次性填入。
|
||
let control = RuntimeControl {
|
||
sender,
|
||
requested: Arc::new(AtomicBool::new(false)),
|
||
completion: Arc::new((Mutex::new(None), Condvar::new())),
|
||
native_sender: Arc::new(OnceLock::new()),
|
||
namespace: namespace.clone(),
|
||
};
|
||
// worker 专用接收端与通知代理；不复制 Receiver 给多个执行线程抢同一退出消息。
|
||
let shutdown = Shutdown {
|
||
receiver,
|
||
control: control.clone(),
|
||
proxy: proxy.clone(),
|
||
};
|
||
// 绑定失败在启动阶段反馈，不能假装本机 API 可用；后台通知仅发送 GUI 事件，不直接画窗口。
|
||
let server = Server::http(API_ADDRESS)
|
||
.map_err(|error| anyhow::anyhow!(error))
|
||
.with_context(|| format!("bind Tessera API to {API_ADDRESS}"))?;
|
||
// 无 VM/文档 payload 的更新通知，GUI 收到后自行读最近完整 NamespaceState。
|
||
let notify: Notify = Arc::new(move || {
|
||
// 事件循环已退出时通知失败无需再触发 VM 故障，发布值本身仍由共享状态保存。
|
||
let _ = proxy.send_event(AppEvent::DocumentUpdated);
|
||
});
|
||
// Network/Console 观察器共享纯数据有界日志，和 VM 的 globals/堆状态分开。
|
||
let activity = Arc::new(Mutex::new(ActivityLog::default()));
|
||
start_server_observed(
|
||
Arc::new(server),
|
||
namespace,
|
||
notify.clone(),
|
||
activity.clone(),
|
||
notify,
|
||
Some(shutdown),
|
||
Some(control.native_sender.clone()),
|
||
)?;
|
||
Ok((activity, control))
|
||
}
|
||
|
||
#[cfg(test)]
|
||
/// API 测试调用，使用测试自己绑定的 Server，不安装原生退出/输入端口，复用真实路由与 worker。
|
||
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,
|
||
None,
|
||
)
|
||
}
|
||
|
||
/// start 与 start_server 共用线程启动器，建立唯一执行队列、VM worker 和只负责路由的 HTTP 接收线程。
|
||
fn start_server_observed(
|
||
server: Arc<Server>,
|
||
namespace: Arc<RwLock<NamespaceState>>,
|
||
notify: Notify,
|
||
activity: Activity,
|
||
observer_notify: Notify,
|
||
shutdown: Option<Shutdown>,
|
||
native_port: Option<Arc<OnceLock<SyncSender<BuildRequest>>>>,
|
||
) -> Result<thread::JoinHandle<()>> {
|
||
// HTTP 接收与 VM 执行分离，最多8个排队请求；源代码只在单个 endor-worker 中操作 VM。
|
||
let (sender, receiver) = sync_channel::<BuildRequest>(8);
|
||
if let Some(native_port) = native_port {
|
||
// 原生输入接入同一个 sender，不能有绕过串行执行/刷新约束的第二条解释器路径。
|
||
native_port
|
||
.set(sender.clone())
|
||
.map_err(|_| anyhow::anyhow!("native input queue already attached"))?;
|
||
}
|
||
// worker 线程与 HTTP 线程各持发布状态 Arc，任何一方都不把读写锁贯穿源代码执行。
|
||
let worker_namespace = namespace.clone();
|
||
// worker 的发布通知与 HTTP 观察通知用途不同，均只投递原生事件。
|
||
let worker_notify = notify.clone();
|
||
// HTTP 线程仅借门闩判断是否拒绝新执行请求，不持 Shutdown 的独占接收端。
|
||
let exiting = shutdown
|
||
.as_ref()
|
||
.map(|shutdown| shutdown.control.requested.clone());
|
||
thread::Builder::new()
|
||
.name("tessera-endor-worker".into())
|
||
.spawn(move || {
|
||
// 所有实际 VM 会话只在这个线程构建/操作，Default 此时尚未创建任何用户程序。
|
||
let mut worker = Worker::default();
|
||
// 分片请求跨循环保留原 HTTP 请求和继续句柄；同一时刻只推进一个有未完成 VM 执行的输入。
|
||
let mut pending: Option<(BuildRequest, PendingRequest)> = None;
|
||
loop {
|
||
if let Some(shutdown) = &shutdown
|
||
&& shutdown.receiver.try_recv().is_ok()
|
||
{
|
||
// 退出优先于下一量子：先撤去待响应请求，再由 shutdown_runtime 清理 VM 续体与生命周期资源。
|
||
if let Some((job, input)) = pending.take() {
|
||
drop(input);
|
||
// 客户端可能已断开，尽力回应后仍继续清理，不能因 socket 失败阻止应用退出。
|
||
let _ = job
|
||
.request
|
||
.respond(error_response(StatusCode(503), "application is exiting"));
|
||
}
|
||
// 清理完成文本含可能的错误，不因收到了退出请求就直接向 GUI 报成功。
|
||
let outcome = worker.shutdown_runtime();
|
||
// 排队请求全部明确回应503，不能留着客户端无期限等待已经停止的执行线程。
|
||
while let Ok(job) = receiver.try_recv() {
|
||
// 已接受但未执行的请求明确发 503；发送失败仅表示接收者已离开，不再重新入队。
|
||
let _ = job
|
||
.request
|
||
.respond(error_response(StatusCode(503), "application is exiting"));
|
||
}
|
||
if let Ok(mut current) = worker_namespace.write() {
|
||
// 撤销所有帧的会话/焦点与原生命令，让 GUI 不再把旧资源当成仍可执行的活域。
|
||
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);
|
||
// 共享完成槽同时服务有限等待和 GUI 退出事件，先存 outcome 再通知。
|
||
let (lock, ready) = &*shutdown.control.completion;
|
||
if let Ok(mut result) = lock.lock() {
|
||
*result = Some(outcome.clone());
|
||
// 先在锁内保存清理结果再唤醒等待者，同时用 AppEvent 通知原生事件循环退出，两条路径复用同一 outcome。
|
||
ready.notify_all();
|
||
}
|
||
// 完成槽已先保存 outcome，GUI 事件投递失败不抹掉可由 wait_shutdown 取得的结果。
|
||
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));
|
||
// 软量子只提供控制边界，不强制睡固定毫秒，避免慢布局每片额外累计等待延迟。
|
||
thread::yield_now();
|
||
}
|
||
RequestProgress::Complete(response) => {
|
||
// 一次性回应消费 job；客户端断开不重执行已完成业务输入，否则会重复副作用。
|
||
let _ = job.request.respond(response);
|
||
}
|
||
}
|
||
continue;
|
||
}
|
||
// 空闲输入边界先推进视口刷新与任务，待执行刷新/协作任务时零等待，否则最多100ms检查一次。
|
||
worker.refresh_viewport(&worker_namespace, &worker_notify);
|
||
// 活跃视觉续体/可运行协程应立即再推进；只有未来计时/I/O 等待时才最多阻塞 100ms。
|
||
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) => {
|
||
// begin_request 既可能立即完成，也可能要求等待旧刷新或保存新的候选续体。
|
||
match worker.begin_request(&job, &worker_namespace, &worker_notify) {
|
||
RequestProgress::Pending(input) => {
|
||
// 请求与继续句柄绑定留到下轮，不能先回应成功后再执行它的 UI 刷新。
|
||
if job.native.is_none()
|
||
&& matches!(&input, PendingRequest::BeforeInput)
|
||
{
|
||
eprintln!(
|
||
"Tessera worker: submission {} waits for active refresh",
|
||
job.submission
|
||
);
|
||
}
|
||
pending = Some((job, input));
|
||
}
|
||
RequestProgress::Complete(response) => {
|
||
// 同步完成也只尝试一次回执，发送失败不回滚 VM 已完成写入或发布文档。
|
||
let _ = job.request.respond(response);
|
||
}
|
||
}
|
||
}
|
||
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
|
||
// timeout 不是 worker 已停止，仅是检查拥有资格/继续下一轮空闲推进的边界。
|
||
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
|
||
})
|
||
}) {
|
||
// 当前持久域同时失去 submission 和 runtime_session 归属时退役；不能仅因源码请求排队就误删仍权威的旧域。
|
||
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))
|
||
// 退出后拒绝写入/执行路由，GET 与 OPTIONS 仍可读取最近发布值或探测接口。
|
||
&& !matches!(request.method(), Method::Get | Method::Options)
|
||
{
|
||
// 写入/执行在退出门闩后拒绝，GET/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")
|
||
}
|
||
|
||
/// BuildRequest 的一次性回应端，HTTP 与进程内原生工作共享执行/刷新路径但保持不同来源身份。
|
||
/// HTTP 头或正文不能创建 Native 变体，也不能让任意远程报文走原生连续输入的宽松版本追随。
|
||
enum RequestReply {
|
||
Http(ObservedRequest),
|
||
Native(SyncSender<std::result::Result<(), String>>),
|
||
}
|
||
impl RequestReply {
|
||
/// worker 线程在请求终结/退出时调用，HTTP 发响应，Native 只写回一次确认或有限错误文本。
|
||
fn respond(self, response: Response<std::io::Cursor<Vec<u8>>>) -> std::io::Result<()> {
|
||
match self {
|
||
Self::Http(request) => request.respond(response),
|
||
Self::Native(done) => {
|
||
// 连续原生输入过期可能正常返回 409，按已消费处理，不把每条旧采样都显示成运行时故障。
|
||
let status = response.status_code().0;
|
||
// 原生侧不解析响应 JSON；非 200/409 取至多 2048 字符错误预览交 GUI 展示。
|
||
let result = if status == 200 || status == 409 {
|
||
Ok(())
|
||
} else {
|
||
Err(String::from_utf8_lossy(response.into_reader().get_ref())
|
||
.chars()
|
||
.take(2048)
|
||
.collect())
|
||
};
|
||
// relay 可能已因窗口退出丢弃接收端，worker 不应为此把已处理输入重新执行或停止。
|
||
let _ = done.send(result);
|
||
Ok(())
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// handle 包装 tiny_http 请求，持有直到 worker 最终响应；只允许特定头部进入 Network 观察日志。
|
||
struct ObservedRequest {
|
||
// 真正 socket 请求所有权，respond 消费它，Deref 只让路由/正文读取复用 tiny_http 接口。
|
||
inner: Request,
|
||
// 共享有界观察日志，不持 VM 强引用或业务对象身份。
|
||
activity: Activity,
|
||
// begin 返回的日志记录身份，异步执行后仍能对应正确请求项。
|
||
id: u64,
|
||
// 日志正文/响应更新后通知 GUI，而非只在实际文档发布时刷新工具。
|
||
notify: Notify,
|
||
}
|
||
impl ObservedRequest {
|
||
/// handle 在路由前调用，提取白名单头和来源显示标签，不把凭据/Cookie 等复制到观察日志。
|
||
fn new(inner: Request, activity: Activity, notify: Notify) -> Self {
|
||
// 活动观察只保留 Content-Type/Length；X-Tessera-Origin 仅判来源标签，不把任意认证头复制进工具日志。
|
||
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");
|
||
// console 标记只是显示来源，不提供权限，也不改变 VM 执行语义。
|
||
let origin = if inner
|
||
.headers()
|
||
.iter()
|
||
.any(|h| h.field.equiv("X-Tessera-Origin") && h.value.as_str() == "console")
|
||
{
|
||
"Console"
|
||
} else {
|
||
"HTTP"
|
||
}
|
||
.to_owned();
|
||
// 创建日志项后只留 id，worker 可以稍后再填响应，不借用锁中的 ActivityLog 记录。
|
||
let id = activity.lock().unwrap().begin(
|
||
inner.method().to_string(),
|
||
inner.url().to_owned(),
|
||
origin,
|
||
headers,
|
||
);
|
||
notify();
|
||
Self {
|
||
inner,
|
||
activity,
|
||
id,
|
||
notify,
|
||
}
|
||
}
|
||
/// submit 成功读取有界 UTF-8 正文后调用，写观察预览并通知工具，不执行源码或创建候选 VM。
|
||
fn record_body(&self, body: &str) {
|
||
self.activity.lock().unwrap().body(self.id, body);
|
||
(self.notify)();
|
||
}
|
||
/// RequestReply::respond/handle 错误路径调用，先记预构造响应，再消费真实 tiny_http 请求发回客户端。
|
||
fn respond(self, response: Response<std::io::Cursor<Vec<u8>>>) -> std::io::Result<()> {
|
||
// 拆开元信息和内存正文 Cursor，记录预览后按原内容重建，避免观察器吃掉实际响应。
|
||
let status = response.status_code();
|
||
// 克隆头列表供日志和原响应同时使用，观察器不修改协议状态或内容类型。
|
||
let headers = response.headers().to_vec();
|
||
// 保留数据长度给重建响应，不按 Unicode 字符数重算 Content-Length。
|
||
let length = response.data_length();
|
||
// 响应正文本已在内存 Cursor，记录预览后用原 status/headers/reader/length 重建响应，不消费正文内容。
|
||
let reader = response.into_reader();
|
||
// 记录的是已构造响应，即使后面 socket 发送失败也已结束活动记录，不能据此声称客户端成功接收。
|
||
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;
|
||
/// Rust 在路由/头部访问时隐式调用，借用底层 Request；包装日志所有权不转移。
|
||
fn deref(&self) -> &Request {
|
||
&self.inner
|
||
}
|
||
}
|
||
impl DerefMut for ObservedRequest {
|
||
/// submit/read_code 请求正文读取时隐式调用，访问底层流，不绕过有界读取检查。
|
||
fn deref_mut(&mut self) -> &mut Request {
|
||
&mut self.inner
|
||
}
|
||
}
|
||
|
||
/// HTTP 接收线程逐个请求调用，区分执行入队和纯发布值读取；执行请求由 worker 持有回应，不在此等 VM。
|
||
fn handle(
|
||
request: Request,
|
||
namespace: &Arc<RwLock<NamespaceState>>,
|
||
notify: &Notify,
|
||
sender: &SyncSender<BuildRequest>,
|
||
activity: &Activity,
|
||
observer_notify: &Notify,
|
||
) {
|
||
// 先建观察记录，即使路由非法也能在 Network 看到实际响应，不只记录成功执行请求。
|
||
let request = ObservedRequest::new(request, activity.clone(), observer_notify.clone());
|
||
// method 克隆为纯枚举，后面 submit 移动 request 所有权不会留下 Request 借用。
|
||
let method = request.method().clone();
|
||
// 路由匹配去掉查询串，namespace 从固定路径提取；不把 query 当作命名空间或源代码。
|
||
let path = request
|
||
.url()
|
||
.split('?')
|
||
.next()
|
||
.unwrap_or(request.url())
|
||
.to_owned();
|
||
|
||
// 三个 POST 后缀分别对应隔离表达式、持久 Console 和当前程序输入，不能因正文像源码就互换。
|
||
let evaluation = namespace_from_route(&path, "/evaluate");
|
||
// DELETE 同一个 /repl 地址表示重置该持久会话。
|
||
let repl = namespace_from_route(&path, "/repl");
|
||
// /events 接受 JSON 事件，后续仍由 worker 校验会话/窗口/布局身份。
|
||
let event = namespace_from_route(&path, "/events");
|
||
// PUT/code 完整替换，POST 在 events/repl/evaluate 中分类，DELETE/repl 是持久会话重置；其他交给资源读取/清空路径。
|
||
let route = if method == Method::Put {
|
||
// 完整替换只接受 /code，不让 PUT /repl 混成持久追加操作。
|
||
namespace_from_path(&path).map(|name| (name, RequestKind::Code))
|
||
} else if method == Method::Post {
|
||
// 先找事件，再 Console，再表达式；后缀严格匹配，命名空间不能含额外路径段。
|
||
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 {
|
||
// 这里只识别需 worker 清理的 /repl，DELETE /code 的发布状态撤销留给 code_request。
|
||
repl.map(|name| (name, RequestKind::ResetRepl))
|
||
} else {
|
||
// GET/OPTIONS 与不支持方法继续同步读取/错误响应，不创建执行 job。
|
||
None
|
||
};
|
||
if let Some((addressed, kind)) = route {
|
||
// submit 消费请求，队列/输入错误也由它负责响应；handle 不再重复发送第二个响应。
|
||
submit(request, addressed, namespace, sender, kind);
|
||
return;
|
||
}
|
||
// 这些只读/清空资源路径只读写 NamespaceState，不读取活 VM globals 或执行源函数。
|
||
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 的 GET/DELETE/OPTIONS 统一走资源处理，路径匹配不等于 namespace 已被授权存在。
|
||
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")
|
||
};
|
||
|
||
// 同步路由只回应一次；连接已断开也不应再次处理路由或重复 DELETE 的清空副作用。
|
||
let _ = request.respond(response);
|
||
}
|
||
|
||
/// handle 对 /code 的同步资源路径调用，GET 返回最近发布源码，DELETE 撤销发布/执行归属并通知 GUI。
|
||
fn code_request(
|
||
method: Method,
|
||
addressed_namespace: &str,
|
||
namespace: &Arc<RwLock<NamespaceState>>,
|
||
notify: &Notify,
|
||
) -> Response<std::io::Cursor<Vec<u8>>> {
|
||
// 先短读检查 namespace 和 GET 元信息；DELETE 需要下方先 drop 读锁再取得写锁。
|
||
let Ok(current) = namespace.read() else {
|
||
return error_response(StatusCode(500), "namespace lock is poisoned");
|
||
};
|
||
if addressed_namespace != current.name {
|
||
// 单活 namespace 模型不自动创建另一个域，拼错名字直接 404 而非向当前程序投递。
|
||
return error_response(
|
||
StatusCode(404),
|
||
format!(
|
||
"namespace `{addressed_namespace}` is not active; current namespace is `{}`",
|
||
current.name
|
||
),
|
||
);
|
||
}
|
||
|
||
// GET 返回最近成功发布的 source 而非正在排队的输入，并通过头部暴露 revision/session/window/layout 及逻辑视口。
|
||
if method == Method::Get {
|
||
// 响应正文来自成功发布的文档 source，排队中的候选/未稳定 Console 草稿不可提前读取。
|
||
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 {
|
||
// 尺寸头是逻辑可用 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")
|
||
{
|
||
// transcript 含多个已接受输入，调用者不应把它误当可直接替换的单个源快照。
|
||
"repl-transcript"
|
||
} else {
|
||
// 完整程序原文快照与独立 Console transcript 的返回含义明确区分。
|
||
"source-snapshot"
|
||
},
|
||
);
|
||
return response;
|
||
}
|
||
// 释放读锁后才申请写锁，避免同一线程持读锁升级为写锁死锁；修改结束也先解锁再发送 GUI 通知。
|
||
drop(current);
|
||
|
||
match method {
|
||
Method::Delete => match namespace.write() {
|
||
Ok(mut current) => {
|
||
// clear 同时推进撤销身份；worker 后续检查归属时不会再提交先前候选的晚到结果。
|
||
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"),
|
||
}
|
||
}
|
||
|
||
/// handle 的执行路由调用，先读有界 UTF-8 正文，再持发布锁完成入队和接受身份更新，不执行 VM。
|
||
fn submit(
|
||
mut request: ObservedRequest,
|
||
addressed: &str,
|
||
namespace: &Arc<RwLock<NamespaceState>>,
|
||
sender: &SyncSender<BuildRequest>,
|
||
kind: RequestKind,
|
||
) {
|
||
// 正文读取失败直接 400，不推进 submission 或创建一项只能在执行阶段再失败的队列工作。
|
||
let source = match read_code(&mut request) {
|
||
Ok(source) => source,
|
||
Err(error) => {
|
||
// 正文错误在入队前响应，失败连接不影响其他请求，也不分配新的提交身份。
|
||
let _ = request.respond(error_response(StatusCode(400), error.to_string()));
|
||
return;
|
||
}
|
||
};
|
||
// 读取成功后才保存受限活动正文预览；超长或非法 UTF-8 请求不会进入执行队列。
|
||
request.record_body(&source);
|
||
// 写锁把“接受新提交”与 worker 的“发布旧提交”互斥，不能先递增身份再入队失败。
|
||
let Ok(mut current) = namespace.write() else {
|
||
// 无法可靠读取发布身份时返回服务端错误，不能绕过锁以猜测 submission 接受新代码。
|
||
let _ = request.respond(error_response(
|
||
StatusCode(500),
|
||
"namespace lock is poisoned",
|
||
));
|
||
return;
|
||
};
|
||
if current.name != addressed {
|
||
// 路由名字只是请求指向，真正执行归属要匹配当前共享 namespace。
|
||
let _ = request.respond(error_response(StatusCode(404), "namespace is not active"));
|
||
return;
|
||
}
|
||
// 仅完整 Code 请求分配新 submission，事件/REPL/表达式继承当前身份，保留共享域输入顺序。
|
||
let submission = match if kind == RequestKind::Code {
|
||
current.next_submission()
|
||
} else {
|
||
Ok(current.submission)
|
||
} {
|
||
Ok(id) => id,
|
||
Err(error) => {
|
||
// 身份计数耗尽等错误尚未入队，保留当前有效 submission，不接受无法唯一标识的工作。
|
||
let _ = request.respond(error_response(StatusCode(500), error.to_string()));
|
||
return;
|
||
}
|
||
};
|
||
// HTTP 信封明确 native=None，正文即使声明某个字段也不能成为原生后台回执输入。
|
||
let job = BuildRequest {
|
||
request: RequestReply::Http(request),
|
||
native: None,
|
||
kind,
|
||
source,
|
||
namespace: addressed.into(),
|
||
submission,
|
||
};
|
||
// 持 namespace 写锁完成入队和接受身份更新；worker 提交使用同一锁，保证接受与发布之间没有竞态窗口。
|
||
match sender.try_send(job) {
|
||
Ok(()) => {
|
||
// 只有成功入队的完整 PUT 才把 submission 改成新值，未接受请求不会废掉已有工作。
|
||
if kind == RequestKind::Code {
|
||
current.submission = submission;
|
||
}
|
||
}
|
||
// 队列拒绝时不推进 submission，所以已经接受的旧工作不会因为本次失败请求而失去发布资格。
|
||
Err(TrySendError::Full(job)) => {
|
||
// try_send 把请求所有权退回，立即回应暂不可用；不等待空位占住 HTTP 接收线程。
|
||
let _ = job
|
||
.request
|
||
.respond(error_response(StatusCode(503), "build queue is full"));
|
||
}
|
||
Err(TrySendError::Disconnected(job)) => {
|
||
// worker 已消失与容量暂满分别报告，原 HTTP 请求仍在返回 job 中，可明确回应而非挂起。
|
||
let _ = job.request.respond(error_response(
|
||
StatusCode(503),
|
||
"execution worker is unavailable",
|
||
));
|
||
}
|
||
}
|
||
}
|
||
|
||
/// worker 在执行/发布边界发现提交已被新 PUT/DELETE 取代时调用，返回 409，不覆盖当前文档。
|
||
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" }),
|
||
)
|
||
}
|
||
|
||
/// handle 的各 API 路由解析调用，借用严格前后缀之间的单段 namespace，失败返回 None 不做模糊回退。
|
||
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('/'))
|
||
}
|
||
/// handle 为 /code 资源调用的特化解析，复用 namespace_from_route 的单段/非空约束。
|
||
fn namespace_from_path(path: &str) -> Option<&str> {
|
||
namespace_from_route(path, "/code")
|
||
}
|
||
|
||
/// submit 在入队之前调用，已知长度预拒绝、真实流再硬限长，任何非 UTF-8 都不能进入 source 字符串。
|
||
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");
|
||
}
|
||
// 已知 Content-Length 先预拒绝，再在流上最多读上限+1；防止缺失/不实长度绕过容量限制。
|
||
// read_to_string 要求 UTF-8，不用有损替换改变用户源码。
|
||
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)
|
||
}
|
||
|
||
/// worker 执行/加载失败时调用，将 anyhow 错误链转成 422；有前端来源才附精确行列，不执行 VM 求值。
|
||
fn compile_error(error: anyhow::Error) -> Response<std::io::Cursor<Vec<u8>>> {
|
||
// 先保存完整可读错误链，只有确有 SourceDiagnostic 才追加源位置字段。
|
||
let mut diagnostic =
|
||
serde_json::json!({ "severity": "error", "message": format!("{error:#}") });
|
||
// 有 SourceDiagnostic 才附行/列和半开字节范围；其他运行时错误仍保留完整错误链消息，不编造源码位置。
|
||
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]
|
||
}),
|
||
)
|
||
}
|
||
|
||
/// 路由、入队、退出及 worker 的协议/运行状态错误共用助手，保留调用者选择的 HTTP 状态码。
|
||
fn error_response(
|
||
status: StatusCode,
|
||
message: impl Into<String>,
|
||
) -> Response<std::io::Cursor<Vec<u8>>> {
|
||
json(status, serde_json::json!({ "error": message.into() }))
|
||
}
|
||
|
||
/// 成功/错误 API 分支共用 JSON 编码入口，正文交 text 一次性包装成内存字节 Cursor。
|
||
fn json(status: StatusCode, value: serde_json::Value) -> Response<std::io::Cursor<Vec<u8>>> {
|
||
text(
|
||
status,
|
||
&value.to_string(),
|
||
"application/json; charset=utf-8",
|
||
)
|
||
}
|
||
|
||
/// handle 的 GET /agent 调用，输出当前发布状态派生的 Markdown 指南，不把它标成 JSON。
|
||
fn markdown(status: StatusCode, value: String) -> Response<std::io::Cursor<Vec<u8>>> {
|
||
text(status, &value, "text/markdown; charset=utf-8")
|
||
}
|
||
|
||
/// json/markdown 和源码 GET/OPTIONS 共用响应助手，统一状态/内容类型和固定本机 CORS 头。
|
||
fn text(status: StatusCode, value: &str, content_type: &str) -> Response<std::io::Cursor<Vec<u8>>> {
|
||
// 公共响应头只允许固定本机 Origin/方法/头名，非通配任意来源；这些头不替代回环绑定与路由校验。
|
||
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
|
||
}
|
||
|
||
/// text/code_request 为内容类型、CORS 和版本信息调用，字段格式交 tiny_http 验证，非法头不直接拼接发送。
|
||
fn add_header(response: &mut Response<std::io::Cursor<Vec<u8>>>, name: &str, value: &str) {
|
||
// 由 HTTP 库验证字段格式，非法头被忽略，不把未校验字符拼进原始响应报文。
|
||
if let Ok(header) = Header::from_bytes(name, value) {
|
||
response.add_header(header);
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests;
|