1065 lines
46 KiB
Rust
1065 lines
46 KiB
Rust
//! 唯一串行执行 VM 的 API worker。保留活跃运行时，另建 PUT 候选，并在候选完整成功后切换发布。
|
|
//! 输入、视口刷新与候选布局可以跨 worker 轮次挂起；中途只公布完整文档，不能让新输入进入未结束的只读调用。
|
|
//! 事件身份同时校验窗口、session、layout_version 和 Mount 能力。失败、替换与关闭分别处理，避免旧回调或旧命令泄漏到新 VM。
|
|
//!
|
|
//! Serial VM ownership, incremental requests, publication and shutdown.
|
|
use super::*;
|
|
|
|
type HttpResponse = Response<std::io::Cursor<Vec<u8>>>;
|
|
// 外层 thread::Result 区分 Rust panic，内层 Result 区分可诊断编译/运行错误；正常结果另带待发布窗口命令。
|
|
type InputOutcome =
|
|
std::thread::Result<Result<(endor_host::ReplResult, Vec<crate::window::Command>)>>;
|
|
|
|
type CandidateOutcome = std::thread::Result<
|
|
Result<(
|
|
endor_host::ReplSession,
|
|
crate::document::RenderDocument,
|
|
Vec<crate::window::Command>,
|
|
)>,
|
|
>;
|
|
|
|
/// 已执行业务部分但仍在等待展示刷新的输入元数据，避免恢复时重放回调。
|
|
pub(super) struct PendingInput {
|
|
// 输入开始时所属 submission，与 id（VM 会话）分开；新 PUT 接受时可保留正在执行的原 VM 至输入收敛。
|
|
epoch: u64,
|
|
id: u64,
|
|
event: bool,
|
|
result: endor_host::ReplResult,
|
|
}
|
|
|
|
/// 挂起 HTTP 请求的阶段：等旧刷新、继续输入或继续隔离候选。
|
|
pub(super) enum PendingRequest {
|
|
// 仅等待旧只读刷新结束，业务输入尚未执行；恢复时可重新走入口，而 Input 阶段不能重放业务副作用。
|
|
BeforeInput,
|
|
Input(Box<PendingInput>),
|
|
Candidate(Box<endor_host::ReplSession>),
|
|
}
|
|
|
|
/// 一次 worker 推进的结果：继续保留请求或交付最终响应。
|
|
pub(super) enum RequestProgress {
|
|
Pending(PendingRequest),
|
|
Complete(HttpResponse),
|
|
}
|
|
|
|
/// 只提取仍属于当前 session 的窗口尺寸，避免旧原生窗口给新候选提供错误约束。
|
|
fn current_viewports(
|
|
current: &NamespaceState,
|
|
) -> std::collections::BTreeMap<u64, crate::layout::Size> {
|
|
let mut viewports: std::collections::BTreeMap<_, _> = current
|
|
.window_viewports
|
|
.iter()
|
|
.filter(|(id, _)| {
|
|
**id == crate::window::MAIN_WINDOW
|
|
|| (current.runtime_session != 0
|
|
&& current
|
|
.document
|
|
.windows
|
|
.get(id)
|
|
.is_some_and(|view| view.frame.session == current.runtime_session))
|
|
})
|
|
.map(|(id, size)| (*id, *size))
|
|
.collect();
|
|
// 主窗口逻辑视口优先覆盖 map 同名值，主壳可以跨会话沿用；副窗口必须仍属于当前发布世代。
|
|
if let Some(viewport) = current.viewport {
|
|
viewports.insert(crate::window::MAIN_WINDOW, viewport);
|
|
}
|
|
viewports
|
|
}
|
|
|
|
#[derive(Default)]
|
|
/// 唯一活跃 VM 的串行拥有者，另行管理候选和下一 session 身份。
|
|
pub(super) struct Worker {
|
|
pub(super) repl: Option<(u64, endor_host::ReplSession)>,
|
|
pub(super) next_session: u64,
|
|
}
|
|
impl Worker {
|
|
/// 撤销活跃运行时并有界清理窗口、Mount 和任务；故障域不再执行新的业务回调。
|
|
pub(super) fn shutdown_runtime(&mut self) -> ShutdownOutcome {
|
|
let Some((_, mut runtime)) = self.repl.take() else {
|
|
return ShutdownOutcome {
|
|
summary: "Tessera shutdown: no live runtime".into(),
|
|
error: None,
|
|
};
|
|
};
|
|
// 正常域允许有界关闭回调，故障域只做宿主清理；catch_unwind 防止清理 panic 跳过原生退出通知。
|
|
let callbacks = !runtime.is_faulted();
|
|
let result = catch_unwind(AssertUnwindSafe(|| runtime.shutdown()));
|
|
match result {
|
|
Ok(Ok((windows, mounts, tasks))) => ShutdownOutcome {
|
|
summary: format!(
|
|
"Tessera shutdown: {windows} windows, {mounts} mounts, {tasks} task entries; callbacks={callbacks}"
|
|
),
|
|
error: None,
|
|
},
|
|
result => {
|
|
let error = match result {
|
|
Ok(Err(error)) => format!("shutdown cleanup failed: {error:#}"),
|
|
_ => "shutdown cleanup failed after an internal panic".into(),
|
|
};
|
|
let error: String = error.chars().take(4096).collect();
|
|
ShutdownOutcome {
|
|
summary: error.clone(),
|
|
error: Some(error),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 处理当前无法继续正常交互的窗口关闭，撤销原生资源与相应运行时记录。
|
|
pub(super) fn close_inactive_window(
|
|
&mut self,
|
|
input: &crate::events::Input,
|
|
epoch: u64,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
) -> Response<std::io::Cursor<Vec<u8>>> {
|
|
// 故障/失活资源的专用关闭只处理副窗口，主壳由 OS 关闭控制走全应用退出流程。
|
|
if input.window == crate::window::MAIN_WINDOW {
|
|
return error_response(
|
|
StatusCode(409),
|
|
"close the primary shell with its OS window control",
|
|
);
|
|
}
|
|
let owns_runtime = self.repl.as_ref().is_some_and(|(owner, runtime)| {
|
|
*owner == epoch && runtime.id == input.session && input.session != 0
|
|
});
|
|
let mut discarded = false;
|
|
if owns_runtime {
|
|
let result = catch_unwind(AssertUnwindSafe(|| {
|
|
self.repl
|
|
.as_mut()
|
|
.expect("owned runtime")
|
|
.1
|
|
.retire_window(input.window)
|
|
}));
|
|
// 退役失败不能继续借用半清理 VM，放弃整个活域并在下方清其他帧的会话与命令资格。
|
|
if !matches!(result, Ok(Ok(()))) {
|
|
// Never keep a half-cleaned VM live or borrow another generation.
|
|
self.repl = None;
|
|
discarded = true;
|
|
}
|
|
}
|
|
let Ok(mut current) = namespace.write() else {
|
|
return error_response(StatusCode(500), "namespace lock poisoned");
|
|
};
|
|
if current.submission != epoch {
|
|
return superseded(epoch);
|
|
}
|
|
let Some(view) = current.document.windows.get(&input.window) else {
|
|
return error_response(StatusCode(409), "window is no longer active");
|
|
};
|
|
// 即便副窗口数值 ID 相同，也不能让旧会话的关闭请求删除新会话投影。
|
|
if view.frame.session != input.session {
|
|
return error_response(
|
|
StatusCode(409),
|
|
"window belongs to another runtime generation",
|
|
);
|
|
}
|
|
let mut document = current.document.clone();
|
|
document.windows.remove(&input.window);
|
|
if discarded {
|
|
document.frame.session = 0;
|
|
document.frame.focus = None;
|
|
for view in document.windows.values_mut() {
|
|
view.frame.session = 0;
|
|
view.frame.focus = None;
|
|
}
|
|
current.runtime_session = 0;
|
|
current.window_commands.clear();
|
|
}
|
|
match current.commit(epoch, document) {
|
|
Ok(Some(_)) => {}
|
|
Ok(None) => return superseded(epoch),
|
|
Err(error) => return error_response(StatusCode(500), error.to_string()),
|
|
}
|
|
current.window_viewports.remove(&input.window);
|
|
drop(current);
|
|
notify();
|
|
json(
|
|
StatusCode(200),
|
|
serde_json::json!({"ok":true,"execution":"window-retire","window":input.window,"callbacks":false,"runtime_discarded":discarded}),
|
|
)
|
|
}
|
|
|
|
/// 在没有前台输入的 worker 轮次推进尺寸和任务引起的展示刷新。
|
|
pub(super) fn refresh_viewport(
|
|
&mut self,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
) {
|
|
self.refresh_viewport_mode(namespace, notify, false);
|
|
}
|
|
|
|
/// 在输入之间推进视口/任务刷新；提交序号或 session 已变时拒绝旧结果发布。
|
|
pub(super) fn refresh_viewport_mode(
|
|
&mut self,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
before_input: bool,
|
|
) {
|
|
// 在读锁内取得一个一致的提交身份和尺寸快照；耗时 VM 工作不占用共享状态写锁。
|
|
let (epoch, viewports) = match namespace.read() {
|
|
Ok(current) => {
|
|
let viewports = current_viewports(¤t);
|
|
(current.submission, viewports)
|
|
}
|
|
Err(_) => return,
|
|
};
|
|
let Some((owner, runtime)) = &mut self.repl else {
|
|
return;
|
|
};
|
|
// submission 变化先尝试保留仍权威的活域所有者，不立即拿旧 epoch 继续发布后台结果。
|
|
if *owner != epoch {
|
|
if let Ok(current) = namespace.read() {
|
|
self.preserve_live_owner(¤t);
|
|
}
|
|
return;
|
|
}
|
|
let id = runtime.id;
|
|
let result = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
|
|
if runtime.is_faulted() {
|
|
runtime.poll_tasks()?;
|
|
return Ok((None, Vec::new()));
|
|
}
|
|
let resized = runtime.resize_windows_incremental(&viewports, false)?;
|
|
if runtime.has_pending_refresh() {
|
|
return Ok((None, Vec::new()));
|
|
}
|
|
// An accepted foreground input waits only for the existing
|
|
// refresh; do not keep placing new callbacks ahead of it.
|
|
// 已有前台请求只等当前刷新，不不断把新任务回调插到它前面，避免忙后台工作饿死用户输入。
|
|
let completed = if before_input {
|
|
None
|
|
} else {
|
|
runtime.poll_tasks_incremental(&viewports)?
|
|
};
|
|
// 任务后续刷新未收敛时可以发布较早已完整的 resize 文档，但暂扣任务产生的命令，防止半完成副作用外泄。
|
|
if runtime.has_pending_refresh() {
|
|
// A complete earlier resize may publish; task effects and their
|
|
// native commands wait until the task's refresh has settled.
|
|
return Ok((resized, Vec::new()));
|
|
}
|
|
Ok((completed.or(resized), runtime.take_window_commands()?))
|
|
}));
|
|
// 计算结束后重新核对归属：期间若发生 PUT/DELETE，不能把旧域草稿发布出去。
|
|
let mut current = match namespace.write() {
|
|
Ok(current) => current,
|
|
Err(_) => return,
|
|
};
|
|
// 耗时 VM 工作期间 HTTP 线程可能接受 PUT/DELETE，发布写锁内再核对 submission 与 session 双身份。
|
|
if current.submission != epoch || current.runtime_session != id {
|
|
return;
|
|
}
|
|
let (document, commands) = match result {
|
|
Ok(Ok((document, commands))) => {
|
|
if let Some(document) = document {
|
|
(document, Some(commands))
|
|
} else {
|
|
// 无新文档也可能有标题/焦点等原生命令；只在实际有投影变化时唤醒 GUI，避免空闲无效重绘。
|
|
let projected = !commands.is_empty();
|
|
current.publish_window_commands(id, commands);
|
|
drop(current);
|
|
if projected {
|
|
notify();
|
|
}
|
|
return;
|
|
}
|
|
}
|
|
failure => {
|
|
let mut document = current.document.clone();
|
|
document.frame.fault = Some(match failure {
|
|
Ok(Err(error)) => format!("Runtime background update failed: {error:#}")
|
|
.chars()
|
|
.take(1024)
|
|
.collect(),
|
|
// 内部 panic 与普通背景错误不同，必须丢弃 VM 并撤销所有原生命令，不能继续运行损坏的域。
|
|
_ => {
|
|
self.repl = None;
|
|
current.runtime_session = 0;
|
|
current.window_commands.clear();
|
|
"Runtime discarded after an internal background failure".into()
|
|
}
|
|
});
|
|
for view in document.windows.values_mut() {
|
|
view.frame.fault = document.frame.fault.clone();
|
|
}
|
|
(document, None)
|
|
}
|
|
};
|
|
let changed = matches!(current.commit(epoch, document), Ok(Some(_)));
|
|
if changed && let Some(commands) = commands {
|
|
current.publish_window_commands(id, commands);
|
|
}
|
|
drop(current);
|
|
if changed {
|
|
notify();
|
|
}
|
|
}
|
|
|
|
/// 检查请求种类和归属，必要时先等已有刷新结束，再建立输入或候选任务。
|
|
pub(super) fn begin_request(
|
|
&mut self,
|
|
job: &BuildRequest,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
) -> RequestProgress {
|
|
// 完整 PUT 独立候选，表达式隔离求值；只有 REPL/事件会进入当前持久域并等待已有刷新。
|
|
if job.kind == RequestKind::Code {
|
|
return self.begin_program_request(job, namespace, notify);
|
|
}
|
|
if job.kind == RequestKind::Expression {
|
|
return RequestProgress::Complete(build(job, namespace, notify));
|
|
}
|
|
// Advance one existing refresh quantum, then park the HTTP request.
|
|
// No input statement executes inside the read-only scope; shutdown and
|
|
// newer viewport targets can be observed at every worker turn.
|
|
self.refresh_viewport_mode(namespace, notify, true);
|
|
let epoch = match namespace.read() {
|
|
Ok(current) => current.submission,
|
|
Err(_) => {
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(500),
|
|
"namespace lock is poisoned",
|
|
));
|
|
}
|
|
};
|
|
if epoch != job.submission {
|
|
return RequestProgress::Complete(superseded(job.submission));
|
|
}
|
|
// 重置丢弃内存会话但保留显示文档，清帧 session/focus 使旧画面不再接受旧对象事件。
|
|
if job.kind == RequestKind::ResetRepl {
|
|
self.repl = None;
|
|
if let Ok(mut current) = namespace.write() {
|
|
current.document.frame.session = 0;
|
|
current.document.frame.focus = None;
|
|
for document in current.document.windows.values_mut() {
|
|
document.frame.session = 0;
|
|
document.frame.focus = None;
|
|
}
|
|
current.runtime_session = 0;
|
|
current.window_commands.clear();
|
|
}
|
|
notify();
|
|
return RequestProgress::Complete(json(
|
|
StatusCode(200),
|
|
serde_json::json!({"ok":true,"execution":"repl-reset","persistent":true,"session_reset":true,"message":"Console memory cleared; displayed page preserved"}),
|
|
));
|
|
}
|
|
if self
|
|
.repl
|
|
.as_ref()
|
|
.is_some_and(|(_, runtime)| runtime.has_pending_refresh())
|
|
{
|
|
return RequestProgress::Pending(PendingRequest::BeforeInput);
|
|
}
|
|
// 事件先做 JSON 结构、kind 白名单和 Input 字段校验，再用权威发布帧核对具体资源能力。
|
|
let event = if job.kind == RequestKind::Event {
|
|
let mut input: crate::events::Input = if let Some(input) = &job.native {
|
|
input.clone()
|
|
} else {
|
|
match serde_json::from_str(&job.source) {
|
|
Ok(input) => input,
|
|
Err(error) => {
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(400),
|
|
error.to_string(),
|
|
));
|
|
}
|
|
}
|
|
};
|
|
if !input.kind.starts_with("developer-open:")
|
|
&& !matches!(
|
|
input.kind.as_str(),
|
|
"click"
|
|
| "pointermove"
|
|
| "pointerdown"
|
|
| "pointerup"
|
|
| "pointerleave"
|
|
| "pointercancel"
|
|
| "focus"
|
|
| "scroll"
|
|
| "page-pan"
|
|
| "page-zoom"
|
|
| "viewport-pan"
|
|
| "viewport-zoom"
|
|
| "scene-orbit"
|
|
| "scene-pan"
|
|
| "scene-zoom"
|
|
| "tab-activate"
|
|
| "tab-close"
|
|
| "tab-detach"
|
|
| "tab-drop"
|
|
| "window-close"
|
|
)
|
|
{
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(400),
|
|
"unsupported event kind",
|
|
));
|
|
}
|
|
if let Err(message) = input.validate() {
|
|
return RequestProgress::Complete(error_response(StatusCode(400), message));
|
|
}
|
|
let current = match namespace.read() {
|
|
Ok(current) => current,
|
|
Err(_) => {
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(500),
|
|
"namespace lock poisoned",
|
|
));
|
|
}
|
|
};
|
|
// Rebase trusted captured native gestures only after BeforeInput
|
|
// has drained the prior refresh. External HTTP remains strict.
|
|
if job.native.is_some()
|
|
&& !crate::events::refresh_continuous_input(&mut input, ¤t)
|
|
{
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(409),
|
|
"native target was replaced or detached",
|
|
));
|
|
}
|
|
let frame = if input.window == crate::window::MAIN_WINDOW {
|
|
¤t.document.frame
|
|
} else {
|
|
let Some(document) = current.document.windows.get(&input.window) else {
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(409),
|
|
"window is no longer active",
|
|
));
|
|
};
|
|
&document.frame
|
|
};
|
|
// 窗口关闭允许处理已失活/故障副窗口，不像点击一样要求可运行 VM；由专用宿主退役路径处理。
|
|
if input.kind == "window-close"
|
|
&& frame.window == input.window
|
|
&& frame.session == input.session
|
|
{
|
|
let owns = self.repl.as_ref().is_some_and(|(owner, runtime)| {
|
|
*owner == epoch && runtime.id == input.session && input.session != 0
|
|
});
|
|
let stopped = frame.fault.is_some()
|
|
|| !owns
|
|
|| self
|
|
.repl
|
|
.as_ref()
|
|
.is_some_and(|(_, runtime)| runtime.is_faulted());
|
|
if stopped {
|
|
drop(current);
|
|
return RequestProgress::Complete(
|
|
self.close_inactive_window(&input, epoch, namespace, notify),
|
|
);
|
|
}
|
|
}
|
|
// 标签拖放同时验证目标窗口的会话、版本、无故障及插入位置，源窗口校验不能替代目标校验。
|
|
if let Some(target) = &input.tab_drop {
|
|
let target_frame = if target.window == crate::window::MAIN_WINDOW {
|
|
Some(¤t.document.frame)
|
|
} else {
|
|
current
|
|
.document
|
|
.windows
|
|
.get(&target.window)
|
|
.map(|document| &document.frame)
|
|
};
|
|
if target_frame.is_none_or(|frame| {
|
|
frame.window != target.window
|
|
|| frame.session != input.session
|
|
|| frame.layout_version != target.layout_version
|
|
|| frame.fault.is_some()
|
|
|| target.index > frame.tabs.len()
|
|
}) || self
|
|
.repl
|
|
.as_ref()
|
|
.is_none_or(|(_, runtime)| !runtime.accepts_tab_drop(&input))
|
|
{
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(409),
|
|
"drop targets an obsolete or inactive frame",
|
|
));
|
|
}
|
|
}
|
|
// 统一事件归属门：窗口/session/版本及操作对应资源必须都合法；window-close 特别允许旧布局版本。
|
|
if frame.window != input.window
|
|
|| frame.session != input.session
|
|
|| (input.kind != "window-close" && frame.layout_version != input.layout_version)
|
|
|| frame.fault.is_some()
|
|
|| if input.kind == "window-close"
|
|
|| input.kind.starts_with("developer-open:")
|
|
|| matches!(input.kind.as_str(), "page-pan" | "page-zoom")
|
|
{
|
|
input.mount != 0
|
|
} else if matches!(
|
|
input.kind.as_str(),
|
|
"tab-activate" | "tab-close" | "tab-detach" | "tab-drop"
|
|
) {
|
|
// 标签操作的 mount 字段承载 Tab ID，而普通输入承载 Mount ID；两套身份不能混着查询。
|
|
!frame.tabs.iter().any(|tab| tab.id == input.mount)
|
|
} else {
|
|
!frame.mounts.iter().any(|mount| {
|
|
mount.id == input.mount
|
|
&& if input.kind.starts_with("viewport-") {
|
|
mount.canvas_viewport
|
|
} else if input.kind.starts_with("scene-") {
|
|
// 3D 手势还要求显式控制器能力，scroll 要有滚动投影，点击/焦点则要求 interactive。
|
|
mount.scene3d_viewport && mount.camera_controls
|
|
} else if input.kind == "scroll" {
|
|
mount.scroll.is_some()
|
|
} else {
|
|
mount.interactive
|
|
}
|
|
})
|
|
}
|
|
|| self.repl.as_ref().is_none_or(|(context, runtime)| {
|
|
*context != epoch
|
|
|| runtime.id != input.session
|
|
|| (input.kind != "window-close" && !runtime.accepts_window(input.window))
|
|
})
|
|
{
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(409),
|
|
"event targets an obsolete or inactive frame",
|
|
));
|
|
}
|
|
// 兼容页面仅 Canvas 可缩放，Window 模式不接受 pan；结构合法不等于当前页面模式支持操作。
|
|
if input.kind == "page-zoom" && frame.surface.mode != crate::page::Mode::Canvas
|
|
|| input.kind == "page-pan" && frame.surface.mode == crate::page::Mode::Window
|
|
{
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(400),
|
|
"camera input is unavailable in this Page mode",
|
|
));
|
|
}
|
|
// 缩放锚点须经过面板反投影与内容 clip 检查，不能落在视口边框或被遮挡面板上仍缩放。
|
|
if matches!(input.kind.as_str(), "viewport-zoom" | "scene-zoom")
|
|
&& !frame.mounts.iter().any(|mount| {
|
|
mount.id == input.mount
|
|
&& input.point.is_some_and(|point| {
|
|
let point = crate::geometry::Point::new(point[0], point[1]);
|
|
mount.contains(point)
|
|
&& mount.viewport_clip.is_some_and(|clip| {
|
|
mount.view_point(point, true).is_some_and(|point| {
|
|
point.x >= clip[0]
|
|
&& point.y >= clip[1]
|
|
&& point.x < clip[0] + clip[2]
|
|
&& point.y < clip[1] + clip[3]
|
|
})
|
|
})
|
|
})
|
|
})
|
|
{
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(400),
|
|
"zoom anchor is outside content viewport",
|
|
));
|
|
}
|
|
// 世界 UI 的非线性位移变换需要位置锚点；普通二维视口的线性位移可以省略 point。
|
|
if (input.kind.starts_with("viewport-") || input.kind.starts_with("scene-"))
|
|
&& input.point.is_none()
|
|
&& frame
|
|
.mounts
|
|
.iter()
|
|
.any(|mount| mount.id == input.mount && mount.panel.is_some())
|
|
{
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(400),
|
|
"world-space UI camera input requires a pointer",
|
|
));
|
|
}
|
|
Some(input)
|
|
} else {
|
|
None
|
|
};
|
|
if self
|
|
.repl
|
|
.as_ref()
|
|
.is_none_or(|(current, _)| *current != epoch)
|
|
{
|
|
// 没有当前 epoch 会话时惰性建立空 REPL，每次分配新会话 ID 并检查溢出，禁止复用旧输入身份。
|
|
self.next_session = match self.next_session.checked_add(1) {
|
|
Some(id) => id,
|
|
None => {
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(500),
|
|
"session identities exhausted",
|
|
));
|
|
}
|
|
};
|
|
let viewport = namespace.read().ok().and_then(|current| current.viewport);
|
|
match endor_host::ReplSession::new_with_viewport(self.next_session, viewport) {
|
|
Ok(session) => {
|
|
if let Ok(mut current) = namespace.write()
|
|
&& current.submission == epoch
|
|
{
|
|
current.publish_window_commands(session.id, Vec::new());
|
|
}
|
|
self.repl = Some((epoch, session));
|
|
}
|
|
Err(error) => {
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(500),
|
|
error.to_string(),
|
|
));
|
|
}
|
|
}
|
|
}
|
|
let viewports = match namespace.read() {
|
|
Ok(current) => current_viewports(¤t),
|
|
Err(_) => {
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(500),
|
|
"namespace lock is poisoned",
|
|
));
|
|
}
|
|
};
|
|
let session = &mut self.repl.as_mut().expect("initialized REPL").1;
|
|
let id = session.id;
|
|
let result = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
|
|
let result = if let Some(event) = &event {
|
|
session.dispatch_incremental(event, &viewports)?
|
|
} else {
|
|
session.execute_incremental(&job.source, &job.namespace, &viewports)?
|
|
};
|
|
// 业务执行可能已改 VM，但展示未完成时暂不取窗口命令；PendingInput 保存结果并只恢复展示阶段。
|
|
let commands = if session.has_pending_refresh() {
|
|
Vec::new()
|
|
} else {
|
|
session.take_window_commands()?
|
|
};
|
|
Ok((result, commands))
|
|
}));
|
|
let faulted = session.is_faulted();
|
|
if session.has_pending_refresh() {
|
|
if let Ok(Ok((result, _))) = result {
|
|
return RequestProgress::Pending(PendingRequest::Input(Box::new(PendingInput {
|
|
epoch,
|
|
id,
|
|
event: event.is_some(),
|
|
result,
|
|
})));
|
|
}
|
|
// 启动失败且残留挂起刷新时直接丢弃域，不能保留一个无法安全推进的 HTTP 请求。
|
|
// A failed start must never leave a request waiting on a broken VM.
|
|
self.repl = None;
|
|
}
|
|
RequestProgress::Complete(self.finish_input(
|
|
job,
|
|
namespace,
|
|
notify,
|
|
epoch,
|
|
id,
|
|
event.is_some(),
|
|
faulted,
|
|
result,
|
|
))
|
|
}
|
|
|
|
/// 恢复已挂起的请求，保持原 HTTP 响应和运行时身份直到最终完成。
|
|
pub(super) fn resume_request(
|
|
&mut self,
|
|
job: &BuildRequest,
|
|
pending: PendingRequest,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
) -> RequestProgress {
|
|
match pending {
|
|
PendingRequest::BeforeInput => self.begin_request(job, namespace, notify),
|
|
PendingRequest::Input(input) => self.resume_input(job, input, namespace, notify),
|
|
PendingRequest::Candidate(runtime) => {
|
|
self.resume_candidate(job, runtime, namespace, notify)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 继续同一 Console/事件输入的展示刷新，不再次执行其业务回调。
|
|
fn resume_input(
|
|
&mut self,
|
|
job: &BuildRequest,
|
|
mut input: Box<PendingInput>,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
) -> RequestProgress {
|
|
let viewports = match namespace.read() {
|
|
// 正在执行输入只要仍属当前提交或仍拥有权威 VM 就可收敛；两项皆失效才彻底废弃。
|
|
Ok(current)
|
|
if current.submission == input.epoch || current.runtime_session == input.id =>
|
|
{
|
|
current_viewports(¤t)
|
|
}
|
|
Ok(_) => {
|
|
self.repl = None;
|
|
return RequestProgress::Complete(superseded(job.submission));
|
|
}
|
|
Err(_) => {
|
|
self.repl = None;
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(500),
|
|
"namespace lock is poisoned",
|
|
));
|
|
}
|
|
};
|
|
let Some((epoch, session)) = &mut self.repl else {
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(409),
|
|
"input runtime was discarded",
|
|
));
|
|
};
|
|
if *epoch != input.epoch || session.id != input.id {
|
|
return RequestProgress::Complete(superseded(job.submission));
|
|
}
|
|
let resumed = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
|
|
// 这里只恢复 pending 展示/尺寸工作，不再调用 execute/dispatch，所以前一个量子的源副作用不会重放。
|
|
let document = session.resize_windows_incremental(&viewports, false)?;
|
|
if session.has_pending_refresh() {
|
|
return Ok(None);
|
|
}
|
|
Ok(Some((document, session.take_window_commands()?)))
|
|
}));
|
|
let faulted = session.is_faulted();
|
|
let result = match resumed {
|
|
Ok(Ok(None)) => return RequestProgress::Pending(PendingRequest::Input(input)),
|
|
Ok(Ok(Some((document, commands)))) => {
|
|
input.result.document = document;
|
|
Ok(Ok((input.result, commands)))
|
|
}
|
|
Ok(Err(error)) => Ok(Err(error)),
|
|
Err(panic) => Err(panic),
|
|
};
|
|
RequestProgress::Complete(self.finish_input(
|
|
job,
|
|
namespace,
|
|
notify,
|
|
input.epoch,
|
|
input.id,
|
|
input.event,
|
|
faulted,
|
|
result,
|
|
))
|
|
}
|
|
|
|
#[allow(clippy::too_many_arguments)]
|
|
/// 校验输入仍属于当前域，再一次性发布结果、元数据和原生窗口命令。
|
|
fn finish_input(
|
|
&mut self,
|
|
job: &BuildRequest,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
epoch: u64,
|
|
id: u64,
|
|
event: bool,
|
|
faulted: bool,
|
|
result: InputOutcome,
|
|
) -> HttpResponse {
|
|
let Ok(mut current) = namespace.write() else {
|
|
return error_response(StatusCode(500), "namespace lock is poisoned");
|
|
};
|
|
// 接受新 PUT 会使排队旧输入失效，但已执行的活域输入需先收敛，才能在候选失败时留下内部一致的旧程序。
|
|
let epoch = if current.submission != epoch {
|
|
if current.runtime_session != id {
|
|
drop(current);
|
|
self.repl = None;
|
|
return superseded(job.submission);
|
|
}
|
|
// Accepted PUTs invalidate queued inputs, not a mutation already
|
|
// executing on this owner. Settle it before building the candidate
|
|
// so a failed replacement can preserve a consistent live heap.
|
|
if let Some((owner, runtime)) = &mut self.repl
|
|
&& runtime.id == id
|
|
{
|
|
*owner = current.submission;
|
|
}
|
|
current.submission
|
|
} else {
|
|
epoch
|
|
};
|
|
match result {
|
|
Ok(Ok((result, commands))) => {
|
|
let projected = !commands.is_empty();
|
|
// rendered 只表示此次交付带新文档，成功执行但页面未变仍可返回 value/session；不因此虚增 revision。
|
|
let rendered = result.document.is_some();
|
|
if let Some(document) = result.document {
|
|
match current.commit(epoch, document) {
|
|
Ok(Some(_)) => {}
|
|
Ok(None) => return superseded(job.submission),
|
|
Err(error) => return error_response(StatusCode(500), error.to_string()),
|
|
}
|
|
}
|
|
current.publish_window_commands(id, commands);
|
|
let revision = current.revision;
|
|
drop(current);
|
|
if rendered || projected {
|
|
notify();
|
|
}
|
|
json(
|
|
StatusCode(200),
|
|
serde_json::json!({"ok":true,"execution":if event { "event" } else { "repl" },"persistent":true,"session":id,"input":result.input,"value":result.value,"bindings":result.bindings,"rendered":rendered,"revision":revision}),
|
|
)
|
|
}
|
|
Ok(Err(error)) => {
|
|
if event || faulted {
|
|
let mut document = current.document.clone();
|
|
// Preserve the first failure of this VM; subsequent
|
|
// rejected Console input must not hide the callback cause.
|
|
// A preserved picture from another generation is unrelated.
|
|
// 同 VM 保留最初故障原因，后续“域已停止”错误不能覆盖真正回调故障；其他世代旧图不沿用故障。
|
|
let fault = (document.frame.session == id)
|
|
.then(|| document.frame.fault.clone())
|
|
.flatten()
|
|
.unwrap_or_else(|| format!("{error:#}").chars().take(1024).collect());
|
|
document.frame.fault = Some(fault.clone());
|
|
for view in document.windows.values_mut() {
|
|
view.frame.fault = Some(fault.clone());
|
|
}
|
|
let _ = current.commit(epoch, document);
|
|
drop(current);
|
|
notify();
|
|
}
|
|
compile_error(error)
|
|
}
|
|
Err(_) => {
|
|
let mut document = current.document.clone();
|
|
document.frame.fault =
|
|
Some("Runtime discarded after an internal execution failure".into());
|
|
for view in document.windows.values_mut() {
|
|
view.frame.fault = document.frame.fault.clone();
|
|
}
|
|
// 输入 panic 保留最近画面但撤销运行时/命令并发布故障条，再将 self.repl 丢弃。
|
|
current.runtime_session = 0;
|
|
current.window_commands.clear();
|
|
let _ = current.commit(epoch, document);
|
|
drop(current);
|
|
self.repl = None;
|
|
notify();
|
|
json(
|
|
StatusCode(500),
|
|
serde_json::json!({"ok":false,"session_reset":true,"error":"Console execution failed internally; session discarded, displayed page preserved"}),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// An accepted PUT changes the submission, but does not revoke the old
|
|
/// runtime. DELETE clears runtime_session and does revoke it.
|
|
/// 接受较新提交不等于立即销毁旧 VM；候选失败时仍需保留当前可用运行时。
|
|
pub(super) fn preserve_live_owner(&mut self, current: &NamespaceState) {
|
|
if let Some((epoch, runtime)) = &mut self.repl {
|
|
if current.runtime_session == runtime.id {
|
|
*epoch = current.submission;
|
|
} else {
|
|
self.repl = None;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 为完整 PUT 建立隔离候选，保留旧运行时直到新候选完整成功。
|
|
fn begin_program_request(
|
|
&mut self,
|
|
job: &BuildRequest,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
) -> RequestProgress {
|
|
let viewport = match namespace.read() {
|
|
Ok(current) if current.submission == job.submission => current.viewport,
|
|
Ok(current) => {
|
|
self.preserve_live_owner(¤t);
|
|
return RequestProgress::Complete(superseded(job.submission));
|
|
}
|
|
Err(_) => {
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(500),
|
|
"namespace lock poisoned",
|
|
));
|
|
}
|
|
};
|
|
self.next_session = match self.next_session.checked_add(1) {
|
|
Some(id) => id,
|
|
None => {
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(500),
|
|
"session identities exhausted",
|
|
));
|
|
}
|
|
};
|
|
let id = self.next_session;
|
|
// 候选装载运行在独立 ReplSession，未成功前不修改 self.repl，旧程序仍是权威展示来源。
|
|
let started = catch_unwind(AssertUnwindSafe(|| {
|
|
endor_host::ReplSession::load_incremental(
|
|
&job.source,
|
|
&job.namespace,
|
|
job.submission,
|
|
id,
|
|
viewport,
|
|
)
|
|
}));
|
|
let result = match started {
|
|
// 候选业务装载成功但布局未完成，整个候选移进 PendingRequest；不能把 None 当成空文档提交。
|
|
Ok(Ok((runtime, None))) => {
|
|
return RequestProgress::Pending(PendingRequest::Candidate(Box::new(runtime)));
|
|
}
|
|
Ok(Ok((runtime, Some(document)))) => match runtime.take_window_commands() {
|
|
Ok(commands) => Ok(Ok((runtime, document, commands))),
|
|
Err(error) => Ok(Err(error)),
|
|
},
|
|
Ok(Err(error)) => Ok(Err(error)),
|
|
Err(panic) => Err(panic),
|
|
};
|
|
RequestProgress::Complete(self.finish_candidate(job, namespace, notify, id, result))
|
|
}
|
|
|
|
/// 推进候选的布局与视口收敛，检查其是否已被新提交撤销。
|
|
fn resume_candidate(
|
|
&mut self,
|
|
job: &BuildRequest,
|
|
mut runtime: Box<endor_host::ReplSession>,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
) -> RequestProgress {
|
|
let viewports = match namespace.read() {
|
|
Ok(current) if current.submission == job.submission => {
|
|
// Secondary native resources still belong to the old VM. Only
|
|
// the primary shell's current viewport applies to a candidate.
|
|
// 候选不沿用旧副窗口的视口；副窗口资源归旧 VM，只有主壳当前尺寸可作为新候选分配条件。
|
|
current
|
|
.viewport
|
|
.map(|size| (crate::window::MAIN_WINDOW, size))
|
|
.into_iter()
|
|
.collect()
|
|
}
|
|
Ok(current) => {
|
|
self.preserve_live_owner(¤t);
|
|
return RequestProgress::Complete(superseded(job.submission));
|
|
}
|
|
Err(_) => {
|
|
return RequestProgress::Complete(error_response(
|
|
StatusCode(500),
|
|
"namespace lock poisoned",
|
|
));
|
|
}
|
|
};
|
|
let id = runtime.id;
|
|
let resumed = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
|
|
let document = runtime.resize_windows_incremental(&viewports, false)?;
|
|
if runtime.has_pending_refresh() {
|
|
return Ok(None);
|
|
}
|
|
// 所有刷新结束仍没有 Page 文档时明确失败，不用旧显示值冒充候选自己的输出。
|
|
let document =
|
|
document.ok_or_else(|| anyhow::anyhow!("program did not publish a Page"))?;
|
|
Ok(Some((document, runtime.take_window_commands()?)))
|
|
}));
|
|
let result = match resumed {
|
|
Ok(Ok(None)) => return RequestProgress::Pending(PendingRequest::Candidate(runtime)),
|
|
Ok(Ok(Some((document, commands)))) => Ok(Ok((*runtime, document, commands))),
|
|
Ok(Err(error)) => Ok(Err(error)),
|
|
Err(panic) => Err(panic),
|
|
};
|
|
RequestProgress::Complete(self.finish_candidate(job, namespace, notify, id, result))
|
|
}
|
|
|
|
/// 候选成功才切换活跃 VM 和文档；失败保留先前已发布程序。
|
|
fn finish_candidate(
|
|
&mut self,
|
|
job: &BuildRequest,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
id: u64,
|
|
result: CandidateOutcome,
|
|
) -> HttpResponse {
|
|
let Ok(mut current) = namespace.write() else {
|
|
return error_response(StatusCode(500), "namespace lock poisoned");
|
|
};
|
|
if current.submission != job.submission {
|
|
self.preserve_live_owner(¤t);
|
|
return superseded(job.submission);
|
|
}
|
|
match result {
|
|
Ok(Ok((runtime, document, commands))) => match current.commit(job.submission, document)
|
|
{
|
|
// 写锁内先原子提交文档，再切换命令世代与活域；解锁后唤醒 GUI，新 VM 至此才取代旧 VM。
|
|
Ok(Some(revision)) => {
|
|
current.publish_window_commands(id, commands);
|
|
self.repl = Some((job.submission, runtime));
|
|
drop(current);
|
|
notify();
|
|
json(
|
|
StatusCode(200),
|
|
serde_json::json!({"ok":true,"namespace":job.namespace,"submission":job.submission,"revision":revision,"diagnostics":[],"session":id,"persistent":true}),
|
|
)
|
|
}
|
|
Ok(None) => superseded(job.submission),
|
|
Err(error) => error_response(StatusCode(500), error.to_string()),
|
|
},
|
|
failed => {
|
|
// The candidate never mutated the displayed program's heap.
|
|
// Restore that program's event epoch when a replacement fails.
|
|
// 候选失败只更新旧活域 epoch，保留其堆与页面；不存在旧域时也不会借用其他会话伪造成功。
|
|
self.preserve_live_owner(¤t);
|
|
drop(current);
|
|
match failed {
|
|
Ok(Err(error)) => compile_error(error),
|
|
_ => error_response(
|
|
StatusCode(500),
|
|
"candidate failed internally; previous runtime preserved",
|
|
),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 处理隔离表达式或一次性文档编译；编译前后核对 submission，只发布仍有效的完整快照。
|
|
fn build(
|
|
job: &BuildRequest,
|
|
namespace: &Arc<RwLock<NamespaceState>>,
|
|
notify: &Notify,
|
|
) -> Response<std::io::Cursor<Vec<u8>>> {
|
|
// 隔离表达式不读写持久 REPL 堆，也不发布文档，响应明确 persistent=false。
|
|
if job.kind == RequestKind::Expression {
|
|
return match catch_unwind(AssertUnwindSafe(|| endor_host::evaluate(&job.source))) {
|
|
Ok(Ok(value)) => json(
|
|
StatusCode(200),
|
|
serde_json::json!({"ok":true,"value":value,"execution":"isolated-expression","persistent":false}),
|
|
),
|
|
Ok(Err(error)) => compile_error(error),
|
|
Err(_) => error_response(StatusCode(500), "expression execution failed internally"),
|
|
};
|
|
}
|
|
match namespace.read() {
|
|
Ok(current) if current.submission != job.submission => return superseded(job.submission),
|
|
Err(_) => return error_response(StatusCode(500), "namespace lock is poisoned"),
|
|
_ => {}
|
|
}
|
|
// Only an independent native snapshot crosses into the published state.
|
|
// A fresh VM is discarded on success, error and unwind alike.
|
|
// 一次性构建同样用独立 VM，计算锁外执行，返回后再次核对 submission 才能发布原生快照。
|
|
let result = catch_unwind(AssertUnwindSafe(|| {
|
|
endor_host::compile(&job.source, &job.namespace, job.submission)
|
|
}));
|
|
let Ok(mut current) = namespace.write() else {
|
|
return error_response(StatusCode(500), "namespace lock is poisoned");
|
|
};
|
|
if current.submission != job.submission {
|
|
return superseded(job.submission);
|
|
}
|
|
match result {
|
|
Ok(Ok(document)) => match current.commit(job.submission, document) {
|
|
Ok(Some(revision)) => {
|
|
drop(current);
|
|
notify();
|
|
json(
|
|
StatusCode(200),
|
|
serde_json::json!({ "ok": true, "namespace": job.namespace,
|
|
"submission": job.submission, "revision": revision, "diagnostics": [] }),
|
|
)
|
|
}
|
|
Ok(None) => superseded(job.submission),
|
|
Err(error) => error_response(StatusCode(500), error.to_string()),
|
|
},
|
|
Ok(Err(error)) => compile_error(error),
|
|
Err(_) => error_response(
|
|
StatusCode(500),
|
|
"candidate execution failed internally; previous document preserved",
|
|
),
|
|
}
|
|
}
|