docs: explain worker publication and HTTP integration regressions

This commit is contained in:
2026-11-12 17:27:39 +08:00
parent e7f387ed9f
commit ed1a9d2956
4 changed files with 393 additions and 48 deletions
+129 -40
View File
@@ -2,14 +2,18 @@
//! 输入、视口刷新与候选布局可以跨 worker 轮次挂起;中途只公布完整文档,不能让新输入进入未结束的只读调用。
//! 事件身份同时校验窗口、session、layout_version 和 Mount 能力。失败、替换与关闭分别处理,避免旧回调或旧命令泄漏到新 VM。
//!
//! Serial VM ownership, incremental requests, publication and shutdown.
//! api::start_server_observed 的唯一执行线程创建 Worker,每次循环调用 begin_request/resume_request 或 refresh_viewport。
//! HTTP/原生请求仍由外层持有,PendingRequest 只记录继续哪一步;本文件不重新读取请求流或回放已执行输入。
//! NamespaceState 是对外权威值,repl 是实际 VM 所有者;submission 是接受顺序,runtime_session 才表示画面仍属哪个 VM。
use super::*;
// 响应已编码成内存字节,外层 RequestReply 可发 HTTP 或转换成原生回执,不需要读取 VM 返回值。
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>)>>;
// 候选成功必须成套交回持久 VM、完整文档及命令;不能只返回快照后丢掉事件需要的真实运行时。
type CandidateOutcome = std::thread::Result<
Result<(
endor_host::ReplSession,
@@ -22,8 +26,11 @@ type CandidateOutcome = std::thread::Result<
pub(super) struct PendingInput {
// 输入开始时所属 submission,与 id(VM 会话)分开;新 PUT 接受时可保留正在执行的原 VM 至输入收敛。
epoch: u64,
// 实际运行时代际,和 epoch 一起核验;接受新 PUT 不代表已经切换到新 VM。
id: u64,
// 决定响应 execution 和错误停域规则,恢复展示时不再重新猜原始输入类别。
event: bool,
// 已经执行得到的值/统计,后续只补 document,不重复 callback 或 execute 取相同结果。
result: endor_host::ReplResult,
}
@@ -31,20 +38,25 @@ pub(super) struct PendingInput {
pub(super) enum PendingRequest {
// 仅等待旧只读刷新结束,业务输入尚未执行;恢复时可重新走入口,而 Input 阶段不能重放业务副作用。
BeforeInput,
// 输入业务部分已执行,保留活 repl 并只恢复该输入的视觉刷新。
Input(Box<PendingInput>),
// 新 PUT 的隔离候选还未发布,活 repl 仍保留旧程序,不能提前替换。
Candidate(Box<endor_host::ReplSession>),
}
/// 一次 worker 推进的结果:继续保留请求或交付最终响应。
pub(super) enum RequestProgress {
// 外层将原 job 和继续状态一起保存,下一量子进入 resume_request。
Pending(PendingRequest),
// 最终响应值,外层消费原 RequestReply 一次性回应,不再继续这个请求。
Complete(HttpResponse),
}
/// 只提取仍属于当前 session 的窗口尺寸,避免旧原生窗口给新候选提供错误约束。
/// 空闲刷新/输入/恢复路径在发布状态读锁内调用,筛出当前运行时合法尺寸,不执行 VM 或修改窗口表。
fn current_viewports(
current: &NamespaceState,
) -> std::collections::BTreeMap<u64, crate::layout::Size> {
// 主壳可跨 VM 使用;副窗必须仍在当前 document 且属于 runtime_session,旧数字 ID 本身不算归属。
let mut viewports: std::collections::BTreeMap<_, _> = current
.window_viewports
.iter()
@@ -69,12 +81,15 @@ fn current_viewports(
#[derive(Default)]
/// 唯一活跃 VM 的串行拥有者,另行管理候选和下一 session 身份。
pub(super) struct Worker {
// 元组第一项是当前接受提交 epoch,第二项是真实 VM;新 PUT 接受时旧 VM 仍可能保持权威。
pub(super) repl: Option<(u64, endor_host::ReplSession)>,
// 单调 session 分配高水位,即使候选失败也不复用已经分配的代际身份。
pub(super) next_session: u64,
}
impl Worker {
/// 撤销活跃运行时并有界清理窗口、Mount 和任务;故障域不再执行新的业务回调。
/// api 线程循环收到退出信号时调用,移出活运行时并生成纯文本 ShutdownOutcome,供原生退出报告。
pub(super) fn shutdown_runtime(&mut self) -> ShutdownOutcome {
// take 先撤 worker 的活域拥有关系,没有域时仍返回明确成功摘要而非伪造清理数量。
let Some((_, mut runtime)) = self.repl.take() else {
return ShutdownOutcome {
summary: "Tessera shutdown: no live runtime".into(),
@@ -83,6 +98,7 @@ impl Worker {
};
// 正常域允许有界关闭回调,故障域只做宿主清理;catch_unwind 防止清理 panic 跳过原生退出通知。
let callbacks = !runtime.is_faulted();
// 普通清理错误与 Rust panic 都转换成结果,不能让执行线程提前消失而使 GUI 永远等不到完成通知。
let result = catch_unwind(AssertUnwindSafe(|| runtime.shutdown()));
match result {
Ok(Ok((windows, mounts, tasks))) => ShutdownOutcome {
@@ -92,10 +108,12 @@ impl Worker {
error: None,
},
result => {
// 内部 panic 没有可安全继续执行的源错误对象,单独说明;正常 Err 则保留错误链。
let error = match result {
Ok(Err(error)) => format!("shutdown cleanup failed: {error:#}"),
_ => "shutdown cleanup failed after an internal panic".into(),
};
// 按 Unicode 字符限长,不切坏 UTF-8,也不把无限错误链塞入退出消息。
let error: String = error.chars().take(4096).collect();
ShutdownOutcome {
summary: error.clone(),
@@ -105,7 +123,7 @@ impl Worker {
}
}
/// 处理当前无法继续正常交互的窗口关闭,撤销原生资源与相应运行时记录。
/// begin_request 的故障/失活副窗 window-close 分支调用,做纯宿主退役,不运行任何源关闭/生命周期回调。
pub(super) fn close_inactive_window(
&mut self,
input: &crate::events::Input,
@@ -120,11 +138,14 @@ impl Worker {
"close the primary shell with its OS window control",
);
}
// 只有 epoch/session 都匹配才清该 VM 内部资源;相同窗口 ID 不授权操作另一个会话。
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 {
// 接住退役 panic,保证后续仍能从共享文档删除原生副窗投影,而不保留半清理 VM。
let result = catch_unwind(AssertUnwindSafe(|| {
self.repl
.as_mut()
@@ -134,17 +155,19 @@ impl Worker {
}));
// 退役失败不能继续借用半清理 VM,放弃整个活域并在下方清其他帧的会话与命令资格。
if !matches!(result, Ok(Ok(()))) {
// Never keep a half-cleaned VM live or borrow another generation.
// 不借用其他代际补清理;放弃整个 repl 后下方只操作最近发布的纯数据文档。
self.repl = None;
discarded = true;
}
}
// VM 清理在锁外执行完才申请发布写锁,重新验证期间是否有 PUT/DELETE 接受。
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");
};
@@ -155,9 +178,11 @@ impl Worker {
"window belongs to another runtime generation",
);
}
// 在完整已发布文档副本上删目标,保留其他窗口画面;commit 再验证整份跨线程值。
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() {
@@ -181,7 +206,7 @@ impl Worker {
)
}
/// 在没有前台输入的 worker 轮次推进尺寸和任务引起的展示刷新。
/// api 的空闲 worker 轮调用,允许推进尺寸和任务回调;有前台输入等待时使用 before_input=true 的变体。
pub(super) fn refresh_viewport(
&mut self,
namespace: &Arc<RwLock<NamespaceState>>,
@@ -190,7 +215,7 @@ impl Worker {
self.refresh_viewport_mode(namespace, notify, false);
}
/// 在输入之间推进视口/任务刷新;提交序号或 session 已变时拒绝旧结果发布。
/// 空闲循环与 begin_request 共用推进器,锁外执行 VM、写锁内核对 epoch/session 后再发布,防止晚到覆盖。
pub(super) fn refresh_viewport_mode(
&mut self,
namespace: &Arc<RwLock<NamespaceState>>,
@@ -200,11 +225,13 @@ impl Worker {
// 在读锁内取得一个一致的提交身份和尺寸快照;耗时 VM 工作不占用共享状态写锁。
let (epoch, viewports) = match namespace.read() {
Ok(current) => {
// 同一读锁下复制身份和合法尺寸,避免使用来自不同发布世代的两个快照。
let viewports = current_viewports(&current);
(current.submission, viewports)
}
Err(_) => return,
};
// 无持久运行时就没有任务/尺寸可推进;GET 的旧静态文档本身不创造 VM。
let Some((owner, runtime)) = &mut self.repl else {
return;
};
@@ -215,28 +242,31 @@ impl Worker {
}
return;
}
// 保存执行前真实 VM 身份,发布时重新检查,不能只核对新的 submission。
let id = runtime.id;
// 耗时刷新不持 namespace 锁,HTTP 可接受新请求;普通错误/panic 留到写锁阶段区分处理。
let result = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
if runtime.is_faulted() {
// 故障域 poll_tasks 仅取消/回收旧任务,不再启动新的用户回调。
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.
// resized 已完整可以返回,任务新产生窗口命令仍扣留至其完整刷新结束。
return Ok((resized, Vec::new()));
}
Ok((completed.or(resized), runtime.take_window_commands()?))
@@ -250,6 +280,7 @@ impl Worker {
if current.submission != epoch || current.runtime_session != id {
return;
}
// success 可仅有命令而无新帧;错误发布故障条,panic 还要丢弃 VM 与命令世代。
let (document, commands) = match result {
Ok(Ok((document, commands))) => {
if let Some(document) = document {
@@ -266,6 +297,7 @@ impl Worker {
}
}
failure => {
// 故障画面基于上次完整发布文档,不把此次失败的半成品几何泄漏出去。
let mut document = current.document.clone();
document.frame.fault = Some(match failure {
Ok(Err(error)) => format!("Runtime background update failed: {error:#}")
@@ -286,6 +318,7 @@ impl Worker {
(document, None)
}
};
// 只有提交成功才发布同世代原生命令和通知 GUI,错误/过期结果不能算一次有效刷新。
let changed = matches!(current.commit(epoch, document), Ok(Some(_)));
if changed && let Some(commands) = commands {
current.publish_window_commands(id, commands);
@@ -296,7 +329,7 @@ impl Worker {
}
}
/// 检查请求种类和归属,必要时先等已有刷新结束,再建立输入或候选任务。
/// api 线程循环收到 job 或恢复 BeforeInput 时调用;业务未执行前可等待旧刷新,执行后切换 PendingInput。
pub(super) fn begin_request(
&mut self,
job: &BuildRequest,
@@ -305,15 +338,16 @@ impl Worker {
) -> RequestProgress {
// 完整 PUT 独立候选,表达式隔离求值;只有 REPL/事件会进入当前持久域并等待已有刷新。
if job.kind == RequestKind::Code {
// PUT 创建隔离候选,不向当前 VM 追加语句;失败可保留旧程序的实际对象图。
return self.begin_program_request(job, namespace, notify);
}
if job.kind == RequestKind::Expression {
// evaluate 不访问活 VM/显示文档,无需等待当前只读刷新完成。
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.
// 先只推进已有刷新一片,关闭和最新尺寸每个 worker 控制边界仍可见;未空闲不运行新语句。
self.refresh_viewport_mode(namespace, notify, true);
// 推进可能跨发布边界,重新读取当前接受序号,不沿用 job 作为最新状态。
let epoch = match namespace.read() {
Ok(current) => current.submission,
Err(_) => {
@@ -324,10 +358,12 @@ impl Worker {
}
};
if epoch != job.submission {
// 尚未执行的旧输入被新提交撤销,不能借“仍有旧 VM”在新接受顺序后插入旧队列动作。
return RequestProgress::Complete(superseded(job.submission));
}
// 重置丢弃内存会话但保留显示文档,清帧 session/focus 使旧画面不再接受旧对象事件。
if job.kind == RequestKind::ResetRepl {
// 丢真正 VM 的内存状态,画面只作为不可交互静态值保留,不声称用户对象仍可响应。
self.repl = None;
if let Ok(mut current) = namespace.write() {
current.document.frame.session = 0;
@@ -350,13 +386,17 @@ impl Worker {
.as_ref()
.is_some_and(|(_, runtime)| runtime.has_pending_refresh())
{
// BeforeInput 表示尚无业务副作用,恢复可重新走 begin_request;不能把已执行输入也放此状态。
return RequestProgress::Pending(PendingRequest::BeforeInput);
}
// 事件先做 JSON 结构、kind 白名单和 Input 字段校验,再用权威发布帧核对具体资源能力。
let event = if job.kind == RequestKind::Event {
// 可信原生来源取自进程内 job.native;HTTP 正文只能被严格反序列化成普通事件。
let mut input: crate::events::Input = if let Some(input) = &job.native {
// 原生 Input 仍需下面全部字段/能力验证,可信来源只影响连续输入能否追随新布局版本。
input.clone()
} else {
// deny_unknown_fields 和类型检查由 Input 的 serde 声明执行,不按任意 JSON 字段猜动作。
match serde_json::from_str(&job.source) {
Ok(input) => input,
Err(error) => {
@@ -392,14 +432,17 @@ impl Worker {
| "window-close"
)
{
// 开发工具名字由后续工具白名单再查;其余只接受明确列出的原生输入类别。
return RequestProgress::Complete(error_response(
StatusCode(400),
"unsupported event kind",
));
}
if let Err(message) = input.validate() {
// 结构错属于 400,在碰任何活 VM 对象之前回应,不能当作过期帧冲突或源运行错误。
return RequestProgress::Complete(error_response(StatusCode(400), message));
}
// 以下窗口、session、版本与几何能力都从同一个权威发布状态读锁中核对。
let current = match namespace.read() {
Ok(current) => current,
Err(_) => {
@@ -409,8 +452,7 @@ impl Worker {
));
}
};
// Rebase trusted captured native gestures only after BeforeInput
// has drained the prior refresh. External HTTP remains strict.
// 原生 relay 排队期间可能又完成了 BeforeInput 刷新;在此再追随一次同目标新版本,HTTP 保持严格。
if job.native.is_some()
&& !crate::events::refresh_continuous_input(&mut input, &current)
{
@@ -419,9 +461,11 @@ impl Worker {
"native target was replaced or detached",
));
}
// 必须选输入声明的窗口帧,不能因为副窗缺失就回退到主窗口同数值 Mount。
let frame = if input.window == crate::window::MAIN_WINDOW {
&current.document.frame
} else {
// 副窗只来自 current.document.windows,GUI 残留 OS 窗口不自动代表活域资源。
let Some(document) = current.document.windows.get(&input.window) else {
return RequestProgress::Complete(error_response(
StatusCode(409),
@@ -435,9 +479,11 @@ impl Worker {
&& frame.window == input.window
&& frame.session == input.session
{
// 有效资源身份但 VM 已停的关闭可进入纯宿主退役,其他输入仍必须拒绝。
let owns = self.repl.as_ref().is_some_and(|(owner, runtime)| {
*owner == epoch && runtime.id == input.session && input.session != 0
});
// 画面故障、无对应活 VM 或 VM 内已 faulted 都说明不能调用可覆盖的源 Window.close。
let stopped = frame.fault.is_some()
|| !owns
|| self
@@ -445,6 +491,7 @@ impl Worker {
.as_ref()
.is_some_and(|(_, runtime)| runtime.is_faulted());
if stopped {
// close_inactive_window 需要发布写锁,调用前必须释放这里持有的读锁。
drop(current);
return RequestProgress::Complete(
self.close_inactive_window(&input, epoch, namespace, notify),
@@ -453,6 +500,7 @@ impl Worker {
}
// 标签拖放同时验证目标窗口的会话、版本、无故障及插入位置,源窗口校验不能替代目标校验。
if let Some(target) = &input.tab_drop {
// 拖放目标同样要找到原窗口权威帧,不能只验证源标签仍存在就信任目标下标。
let target_frame = if target.window == crate::window::MAIN_WINDOW {
Some(&current.document.frame)
} else {
@@ -463,6 +511,7 @@ impl Worker {
.map(|document| &document.frame)
};
if target_frame.is_none_or(|frame| {
// 列表插入缝允许等于 tabs.len,超过长度、换代、换几何或故障都拒绝。
frame.window != target.window
|| frame.session != input.session
|| frame.layout_version != target.layout_version
@@ -473,6 +522,7 @@ impl Worker {
.as_ref()
.is_none_or(|(_, runtime)| !runtime.accepts_tab_drop(&input))
{
// 错误是本次目标已过期,不让源 moveToWindow 方法运行到一半才发现无法重排。
return RequestProgress::Complete(error_response(
StatusCode(409),
"drop targets an obsolete or inactive frame",
@@ -488,6 +538,7 @@ impl Worker {
|| input.kind.starts_with("developer-open:")
|| matches!(input.kind.as_str(), "page-pan" | "page-zoom")
{
// 全窗关闭、工具打开与整页相机不指向某个 Mount,必须携带零目标,防止身份多义。
input.mount != 0
} else if matches!(
input.kind.as_str(),
@@ -496,26 +547,32 @@ impl Worker {
// 标签操作的 mount 字段承载 Tab ID,而普通输入承载 Mount ID;两套身份不能混着查询。
!frame.tabs.iter().any(|tab| tab.id == input.mount)
} else {
// 普通 Mount 输入查出现表再按 kind 校验对应能力,不能只验证存在同号几何。
!frame.mounts.iter().any(|mount| {
mount.id == input.mount
&& if input.kind.starts_with("viewport-") {
// 二维局部相机仅允许 CanvasViewport,普通可点击矩形不能变成相机目标。
mount.canvas_viewport
} else if input.kind.starts_with("scene-") {
// 3D 手势还要求显式控制器能力,scroll 要有滚动投影,点击/焦点则要求 interactive。
mount.scene3d_viewport && mount.camera_controls
} else if input.kind == "scroll" {
// 滚动投影存在才有滚动语义,不凭 Block 名义类型或外框可点击推测。
mount.scroll.is_some()
} else {
// 点击/焦点等至少要求交互资格;具体源 hitTest/指针订阅仍在 bridge 中再查。
mount.interactive
}
})
}
|| self.repl.as_ref().is_none_or(|(context, runtime)| {
// 发布帧与实际活域必须匹配;静态保留画面不能用作对不存在 VM 的调用凭据。
*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",
@@ -535,6 +592,7 @@ impl Worker {
&& !frame.mounts.iter().any(|mount| {
mount.id == input.mount
&& input.point.is_some_and(|point| {
// 先检查出现外框,再用面板/视图逆映射检测真实内容 clip,边框不作缩放锚点。
let point = crate::geometry::Point::new(point[0], point[1]);
mount.contains(point)
&& mount.viewport_clip.is_some_and(|clip| {
@@ -548,6 +606,7 @@ impl Worker {
})
})
{
// 锚点必须在实际内容区域内,外框/面板反投影验证失败就先拒绝,不运行相机控制器。
return RequestProgress::Complete(error_response(
StatusCode(400),
"zoom anchor is outside content viewport",
@@ -561,6 +620,7 @@ impl Worker {
.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",
@@ -568,6 +628,7 @@ impl Worker {
}
Some(input)
} else {
// Console 不是事件,不制造一个空 Input;下面按 source 执行同一持久会话的交互语句。
None
};
if self
@@ -585,9 +646,11 @@ impl Worker {
));
}
};
// 空 Console 初建只沿用主壳尺寸,不从静态旧帧复制可变业务对象或副窗资源。
let viewport = namespace.read().ok().and_then(|current| current.viewport);
match endor_host::ReplSession::new_with_viewport(self.next_session, viewport) {
Ok(session) => {
// 仅共享状态仍是当前 epoch 才记录新命令世代;真正画面仍等待源程序打开页面。
if let Ok(mut current) = namespace.write()
&& current.submission == epoch
{
@@ -603,6 +666,7 @@ impl Worker {
}
}
}
// 业务执行前复制合法视口集合,之后整个 VM 调用不持 namespace 锁。
let viewports = match namespace.read() {
Ok(current) => current_viewports(&current),
Err(_) => {
@@ -612,9 +676,13 @@ impl Worker {
));
}
};
// 惰性创建已保证 repl 存在,借用这个实际会话而非重放历史或建立每请求临时 VM。
let session = &mut self.repl.as_mut().expect("initialized REPL").1;
// 保存实际会话号,执行后可能收到新 PUT,finish_input 会再次核验归属。
let id = session.id;
// 统一收集可诊断错误与内部 panic,后者必须丢弃域,不能继续借损坏对象图交互。
let result = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
// 两种输入只执行一次业务,挂起时 PendingInput 将保存这个返回值,只继续视觉阶段。
let result = if let Some(event) = &event {
session.dispatch_incremental(event, &viewports)?
} else {
@@ -628,6 +696,7 @@ impl Worker {
};
Ok((result, commands))
}));
// 是否需对可见帧加故障条取决于 VM 的实际执行状态,普通编译失败不等于整个域已损坏。
let faulted = session.is_faulted();
if session.has_pending_refresh() {
if let Ok(Ok((result, _))) = result {
@@ -639,7 +708,7 @@ impl Worker {
})));
}
// 启动失败且残留挂起刷新时直接丢弃域,不能保留一个无法安全推进的 HTTP 请求。
// A failed start must never leave a request waiting on a broken VM.
// 不给无法恢复的执行错误留下“等待下一片可能成功”的请求,保留完整旧画面另作响应。
self.repl = None;
}
RequestProgress::Complete(self.finish_input(
@@ -654,7 +723,7 @@ impl Worker {
))
}
/// 恢复已挂起的请求,保持原 HTTP 响应和运行时身份直到最终完成。
/// api 外层循环保存原 job 后每量子调用,按继续状态分流,只有 BeforeInput 能重新进入业务入口。
pub(super) fn resume_request(
&mut self,
job: &BuildRequest,
@@ -663,6 +732,7 @@ impl Worker {
notify: &Notify,
) -> RequestProgress {
match pending {
// 业务尚未执行,可重新核验提交和等待条件;Input/Candidate 都只继续原执行域。
PendingRequest::BeforeInput => self.begin_request(job, namespace, notify),
PendingRequest::Input(input) => self.resume_input(job, input, namespace, notify),
PendingRequest::Candidate(runtime) => {
@@ -671,7 +741,7 @@ impl Worker {
}
}
/// 继续同一 Console/事件输入的展示刷新,不再次执行其业务回调。
/// resume_request 的 Input 分支调用,核验原会话并继续原视觉工作,保留先前返回值/统计不重执行语句。
fn resume_input(
&mut self,
job: &BuildRequest,
@@ -679,6 +749,7 @@ impl Worker {
namespace: &Arc<RwLock<NamespaceState>>,
notify: &Notify,
) -> RequestProgress {
// 可继续条件和最新合法尺寸在同一读锁中取得,执行时不长期持有发布锁。
let viewports = match namespace.read() {
// 正在执行输入只要仍属当前提交或仍拥有权威 VM 就可收敛;两项皆失效才彻底废弃。
Ok(current)
@@ -687,6 +758,7 @@ impl Worker {
current_viewports(&current)
}
Ok(_) => {
// 提交和权威 VM 都已撤销,不能把旧输入结果重新提交到另一个活动程序上。
self.repl = None;
return RequestProgress::Complete(superseded(job.submission));
}
@@ -698,6 +770,7 @@ impl Worker {
));
}
};
// PendingInput 不自己强持另一个运行时,必须与 Worker 当前拥有的真实 VM 一致。
let Some((epoch, session)) = &mut self.repl else {
return RequestProgress::Complete(error_response(
StatusCode(409),
@@ -707,6 +780,7 @@ impl Worker {
if *epoch != input.epoch || session.id != input.id {
return RequestProgress::Complete(superseded(job.submission));
}
// 只推进 resize/刷新,catch_unwind 保证源视觉方法 panic 后可撤销整个损坏域。
let resumed = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
// 这里只恢复 pending 展示/尺寸工作,不再调用 execute/dispatch,所以前一个量子的源副作用不会重放。
let document = session.resize_windows_incremental(&viewports, false)?;
@@ -715,10 +789,13 @@ impl Worker {
}
Ok(Some((document, session.take_window_commands()?)))
}));
// 恢复失败是否停域由实际 runtime 状态决定,交 finish_input 统一决定画面故障报告。
let faulted = session.is_faulted();
// Pending 原样保留输入元数据,Complete 才把新文档补入第一次业务执行的结果。
let result = match resumed {
Ok(Ok(None)) => return RequestProgress::Pending(PendingRequest::Input(input)),
Ok(Ok(Some((document, commands)))) => {
// 不覆盖 value/input/bindings,避免一次挂起输入变成第二次 Console 记录。
input.result.document = document;
Ok(Ok((input.result, commands)))
}
@@ -738,7 +815,7 @@ impl Worker {
}
#[allow(clippy::too_many_arguments)]
/// 校验输入仍属于当前域,再一次性发布结果、元数据和原生窗口命令。
/// begin_request/resume_input 终结输入时调用,写锁内核验接受身份,提交完整文档与同 VM 命令后回应。
fn finish_input(
&mut self,
job: &BuildRequest,
@@ -750,30 +827,33 @@ impl Worker {
faulted: bool,
result: InputOutcome,
) -> HttpResponse {
// 锁覆盖提交检查/文档/命令世代更新,HTTP 新 PUT 不能插在验证与提交之间制造竞态。
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 {
// 已失去实际画面归属才彻底丢旧 VM,避免旧输入在另一代程序上产生回执或发布。
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.
// 新 PUT 撤销未执行旧队列项,但旧 VM 已开始的变更要先收敛,候选失败才有一致的旧堆可保留。
if let Some((owner, runtime)) = &mut self.repl
&& runtime.id == id
{
// 更新活 VM 的接受 epoch,而不改变其 session;不是把它冒充为新候选或复制对象图。
*owner = current.submission;
}
current.submission
} else {
// 接受序号没变,直接沿用原输入 epoch,不制造额外 revision 或 session。
epoch
};
match result {
Ok(Ok((result, commands))) => {
// 有标题/焦点等命令即使无新文档也需通知 GUI,和 rendered 的含义分别记录。
let projected = !commands.is_empty();
// rendered 只表示此次交付带新文档,成功执行但页面未变仍可返回 value/session;不因此虚增 revision。
let rendered = result.document.is_some();
@@ -785,6 +865,7 @@ impl Worker {
}
}
current.publish_window_commands(id, commands);
// 在释放锁前读取这次确定的 revision,响应不再为字符串化借用共享状态。
let revision = current.revision;
drop(current);
if rendered || projected {
@@ -797,10 +878,8 @@ impl Worker {
}
Ok(Err(error)) => {
if event || faulted {
// 事件/渲染回调故障会影响整个共享 VM;单纯 Console 编译错误可只返回 422 而不覆盖画面。
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())
@@ -810,6 +889,7 @@ impl Worker {
for view in document.windows.values_mut() {
view.frame.fault = Some(fault.clone());
}
// 尽力提交故障条,主响应仍保留原始执行错误;不能因二次故障覆盖真正原因。
let _ = current.commit(epoch, document);
drop(current);
notify();
@@ -817,6 +897,7 @@ impl Worker {
compile_error(error)
}
Err(_) => {
// Rust panic 不作为普通语言错误继续执行,清运行时/命令资格但保留上次完整静态画面。
let mut document = current.document.clone();
document.frame.fault =
Some("Runtime discarded after an internal execution failure".into());
@@ -826,6 +907,7 @@ impl Worker {
// 输入 panic 保留最近画面但撤销运行时/命令并发布故障条,再将 self.repl 丢弃。
current.runtime_session = 0;
current.window_commands.clear();
// 优先保留原始 panic 的响应,故障画面提交尽力而为;不因二次提交失败重新使用已损坏 VM。
let _ = current.commit(epoch, document);
drop(current);
self.repl = None;
@@ -838,29 +920,31 @@ impl Worker {
}
}
/// An accepted PUT changes the submission, but does not revoke the old
/// runtime. DELETE clears runtime_session and does revoke it.
/// 接受较新提交不等于立即销毁旧 VM;候选失败时仍需保留当前可用运行时。
/// 新提交/候选过期或失败时调用,若旧 VM 仍拥有发布画面则只跟进 epoch;DELETE 已撤权威则丢弃它。
pub(super) fn preserve_live_owner(&mut self, current: &NamespaceState) {
if let Some((epoch, runtime)) = &mut self.repl {
if current.runtime_session == runtime.id {
// 旧程序还是真正交互域,接受次序更新但其实际对象和 session 身份都不改变。
*epoch = current.submission;
} else {
// 权威已经撤销或切到另一代,不能仅因还持有 Rust ReplSession 就继续执行。
self.repl = None;
}
}
}
/// 为完整 PUT 建立隔离候选,保留旧运行时直到新候选完整成功。
/// begin_request 的 Code 分支调用,先核验接受身份,再分配新 session 创建隔离候选,旧 self.repl 保持不动。
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) => {
// 排队期间已有更新 PUT/DELETE,当前候选还没执行就应拒绝,不浪费新 VM 构建。
self.preserve_live_owner(&current);
return RequestProgress::Complete(superseded(job.submission));
}
@@ -880,6 +964,7 @@ impl Worker {
));
}
};
// 分配过的 session 即使候选失败也不回收,避免排队旧输入命中下一次新域。
let id = self.next_session;
// 候选装载运行在独立 ReplSession,未成功前不修改 self.repl,旧程序仍是权威展示来源。
let started = catch_unwind(AssertUnwindSafe(|| {
@@ -891,6 +976,7 @@ impl Worker {
viewport,
)
}));
// 两层 Result 分别区分 panic 和语言/布局错误;None 文档表示保留完整候选继续视觉阶段。
let result = match started {
// 候选业务装载成功但布局未完成,整个候选移进 PendingRequest;不能把 None 当成空文档提交。
Ok(Ok((runtime, None))) => {
@@ -906,7 +992,7 @@ impl Worker {
RequestProgress::Complete(self.finish_candidate(job, namespace, notify, id, result))
}
/// 推进候选的布局与视口收敛,检查其是否已被新提交撤销。
/// resume_request 的 Candidate 分支调用,用同一个隔离 VM 继续布局/绘制,期间持续检查提交是否已过期。
fn resume_candidate(
&mut self,
job: &BuildRequest,
@@ -914,10 +1000,9 @@ impl Worker {
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
@@ -936,8 +1021,11 @@ impl Worker {
));
}
};
// 保留候选实际 session,不能在每片恢复时重新分配 ID 或重建 UI 类型域。
let id = runtime.id;
// 续跑视觉状态,不重执行顶层语句或 main;错误/panic 都交 finish_candidate 保留旧活程序。
let resumed = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
// 与活域一样,resize 可重定向外部目标或继续原 Work,但始终留在此候选 owning VM 上。
let document = runtime.resize_windows_incremental(&viewports, false)?;
if runtime.has_pending_refresh() {
return Ok(None);
@@ -947,6 +1035,7 @@ impl Worker {
document.ok_or_else(|| anyhow::anyhow!("program did not publish a Page"))?;
Ok(Some((document, runtime.take_window_commands()?)))
}));
// Complete 才把持久 runtime、完整文档和命令作为一组交给切换发布,Pending 继续原 Box。
let result = match resumed {
Ok(Ok(None)) => return RequestProgress::Pending(PendingRequest::Candidate(runtime)),
Ok(Ok(Some((document, commands)))) => Ok(Ok((*runtime, document, commands))),
@@ -956,7 +1045,7 @@ impl Worker {
RequestProgress::Complete(self.finish_candidate(job, namespace, notify, id, result))
}
/// 候选成功才切换活跃 VM 和文档;失败保留先前已发布程序。
/// 完整加载/恢复候选终结时调用,写锁内二次核验 submission,只有完整成功才切换文档、命令与活 VM。
fn finish_candidate(
&mut self,
job: &BuildRequest,
@@ -965,6 +1054,7 @@ impl Worker {
id: u64,
result: CandidateOutcome,
) -> HttpResponse {
// 字节码/源视觉执行不持锁,只有最终这个切换边界持锁原子更新外部权威。
let Ok(mut current) = namespace.write() else {
return error_response(StatusCode(500), "namespace lock poisoned");
};
@@ -990,8 +1080,6 @@ impl Worker {
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(&current);
drop(current);
@@ -1007,7 +1095,7 @@ impl Worker {
}
}
/// 处理隔离表达式或一次性文档编译;编译前后核对 submission,只发布仍有效的完整快照。
/// begin_request 的 Expression 分支和 API 测试一次性构建路径调用;无持久 VM 输出,源码构建前后核验接受身份。
fn build(
job: &BuildRequest,
namespace: &Arc<RwLock<NamespaceState>>,
@@ -1015,6 +1103,7 @@ fn build(
) -> Response<std::io::Cursor<Vec<u8>>> {
// 隔离表达式不读写持久 REPL 堆,也不发布文档,响应明确 persistent=false。
if job.kind == RequestKind::Expression {
// 表达式只回文本,不能从已有画面反取 globals,成功也明确 persistent=false。
return match catch_unwind(AssertUnwindSafe(|| endor_host::evaluate(&job.source))) {
Ok(Ok(value)) => json(
StatusCode(200),
@@ -1029,12 +1118,12 @@ fn build(
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 才能发布原生快照。
// 包住一次性 VM 执行,panic 也不会让 HTTP 客户端一直等待失去响应的 worker。
let result = catch_unwind(AssertUnwindSafe(|| {
endor_host::compile(&job.source, &job.namespace, job.submission)
}));
// 重新取发布写锁再检查 submission,候选计算期间新请求可接受但旧结果无权覆盖。
let Ok(mut current) = namespace.write() else {
return error_response(StatusCode(500), "namespace lock is poisoned");
};