mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
fix(update): run updates in background and retain release notes (#188)
Detach online updates from HTTP request lifetimes and recover their persisted status in the admin UI. Retain current release notes when no upgrade is available while preserving upgrade restrictions. Fixes #185 Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
@@ -71,6 +71,7 @@ pub struct SystemOperationState {
|
||||
pub struct SystemUpdateStatus {
|
||||
pub previous_version: Option<String>,
|
||||
pub current_version: Option<String>,
|
||||
pub need_restart: bool,
|
||||
pub operation: SystemOperationState,
|
||||
}
|
||||
|
||||
@@ -103,7 +104,6 @@ pub enum SystemOperationAccepted {
|
||||
operation_id: String,
|
||||
deployment_mode: String,
|
||||
message: String,
|
||||
need_restart: bool,
|
||||
target_version: String,
|
||||
},
|
||||
Rollback {
|
||||
|
||||
@@ -60,13 +60,13 @@ impl SystemOperations for RecordingSystemOperations {
|
||||
operation_id: "operation-update".to_owned(),
|
||||
deployment_mode: "source".to_owned(),
|
||||
message: "accepted".to_owned(),
|
||||
need_restart: true,
|
||||
target_version: target_version.unwrap_or_else(|| "latest".to_owned()),
|
||||
})
|
||||
}
|
||||
|
||||
async fn update_status(&self) -> Result<SystemUpdateStatus, SystemOperationError> {
|
||||
Ok(SystemUpdateStatus {
|
||||
need_restart: false,
|
||||
previous_version: None,
|
||||
current_version: Some("1.0.0".to_owned()),
|
||||
operation: SystemOperationState {
|
||||
|
||||
@@ -163,6 +163,7 @@ impl From<SystemOperationState> for SystemOperationStateView {
|
||||
struct SystemUpdateStatusView {
|
||||
previous_version: Option<String>,
|
||||
current_version: Option<String>,
|
||||
need_restart: bool,
|
||||
operation: SystemOperationStateView,
|
||||
}
|
||||
|
||||
@@ -171,6 +172,7 @@ impl From<SystemUpdateStatus> for SystemUpdateStatusView {
|
||||
Self {
|
||||
previous_version: status.previous_version,
|
||||
current_version: status.current_version,
|
||||
need_restart: status.need_restart,
|
||||
operation: status.operation.into(),
|
||||
}
|
||||
}
|
||||
@@ -182,7 +184,6 @@ struct UpdateAcceptedView {
|
||||
operation_id: String,
|
||||
deployment_mode: String,
|
||||
message: String,
|
||||
need_restart: bool,
|
||||
target_version: String,
|
||||
}
|
||||
|
||||
@@ -308,19 +309,17 @@ where
|
||||
operation_id,
|
||||
deployment_mode,
|
||||
message,
|
||||
need_restart,
|
||||
target_version,
|
||||
} = result
|
||||
else {
|
||||
return Err(AdminError::internal());
|
||||
};
|
||||
Ok(AdminResponse::new(
|
||||
StatusCode::OK,
|
||||
StatusCode::ACCEPTED,
|
||||
AdminEnvelope::ok(UpdateAcceptedView {
|
||||
operation_id,
|
||||
deployment_mode,
|
||||
message,
|
||||
need_restart,
|
||||
target_version,
|
||||
}),
|
||||
))
|
||||
|
||||
@@ -4,14 +4,15 @@ use async_trait::async_trait;
|
||||
use axum::{
|
||||
Router,
|
||||
body::{Body, to_bytes},
|
||||
http::{Request, header},
|
||||
http::{Request, StatusCode, header},
|
||||
};
|
||||
use chrono::Utc;
|
||||
use futures::stream;
|
||||
use gateway_admin::{
|
||||
model::system::{
|
||||
SystemOperationAccepted, SystemUpdateDetail, SystemUpdateEvent, SystemUpdateEventLevel,
|
||||
SystemUpdateStatus, SystemVersion,
|
||||
SystemOperationAccepted, SystemOperationKind, SystemOperationState, SystemOperationStatus,
|
||||
SystemUpdateDetail, SystemUpdateEvent, SystemUpdateEventLevel, SystemUpdateStatus,
|
||||
SystemVersion,
|
||||
},
|
||||
ports::system::{SystemOperationError, SystemOperations, SystemUpdateEventStream},
|
||||
};
|
||||
@@ -67,6 +68,51 @@ async fn update_event_stream_should_preserve_event_id_and_download_progress_perc
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_should_return_accepted_and_report_restart_only_through_status() {
|
||||
let fixture = AdminTestFixture::with_system(Arc::new(ProgressSystem)).await;
|
||||
fixture.auth.insert_session("valid-session");
|
||||
let router = app(fixture.state());
|
||||
let response = router
|
||||
.clone()
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.method("POST")
|
||||
.uri("/api/admin/system/update")
|
||||
.header(header::COOKIE, "cpr_session=valid-session")
|
||||
.header("x-request-id", "req_system_update")
|
||||
.header(header::CONTENT_TYPE, "application/json")
|
||||
.body(Body::from(r#"{"targetVersion":"0.2.0"}"#))
|
||||
.expect("request"),
|
||||
)
|
||||
.await
|
||||
.expect("response");
|
||||
assert_eq!(response.status(), StatusCode::ACCEPTED);
|
||||
let body: serde_json::Value =
|
||||
serde_json::from_slice(&to_bytes(response.into_body(), 4096).await.expect("body"))
|
||||
.expect("JSON");
|
||||
assert_eq!(body["data"]["operationId"], "update-1");
|
||||
assert_eq!(body["data"]["targetVersion"], "0.2.0");
|
||||
assert!(body["data"].get("needRestart").is_none());
|
||||
let response = router
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.uri("/api/admin/system/update/status")
|
||||
.header(header::COOKIE, "cpr_session=valid-session")
|
||||
.header("x-request-id", "req_system_update")
|
||||
.body(Body::empty())
|
||||
.expect("request"),
|
||||
)
|
||||
.await
|
||||
.expect("response");
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let body: serde_json::Value =
|
||||
serde_json::from_slice(&to_bytes(response.into_body(), 4096).await.expect("body"))
|
||||
.expect("JSON");
|
||||
assert_eq!(body["data"]["operation"]["status"], "succeeded");
|
||||
assert_eq!(body["data"]["needRestart"], true);
|
||||
}
|
||||
|
||||
fn app(state: AdminTestState) -> Router {
|
||||
system::router::<AdminTestState>().with_state(state)
|
||||
}
|
||||
@@ -100,13 +146,32 @@ impl SystemOperations for ProgressSystem {
|
||||
|
||||
async fn perform_update(
|
||||
&self,
|
||||
_: Option<String>,
|
||||
target: Option<String>,
|
||||
) -> Result<SystemOperationAccepted, SystemOperationError> {
|
||||
Err(unavailable_system())
|
||||
Ok(SystemOperationAccepted::Update {
|
||||
operation_id: "update-1".to_owned(),
|
||||
deployment_mode: "docker".to_owned(),
|
||||
message: "更新已开始".to_owned(),
|
||||
target_version: target.expect("target"),
|
||||
})
|
||||
}
|
||||
|
||||
async fn update_status(&self) -> Result<SystemUpdateStatus, SystemOperationError> {
|
||||
Err(unavailable_system())
|
||||
Ok(SystemUpdateStatus {
|
||||
previous_version: Some("0.1.0".to_owned()),
|
||||
current_version: Some("0.2.0".to_owned()),
|
||||
need_restart: true,
|
||||
operation: SystemOperationState {
|
||||
operation_id: Some("update-1".to_owned()),
|
||||
kind: Some(SystemOperationKind::Update),
|
||||
status: SystemOperationStatus::Succeeded,
|
||||
target_version: Some("0.2.0".to_owned()),
|
||||
message: Some("done".to_owned()),
|
||||
error: None,
|
||||
started_at: Some(Utc::now()),
|
||||
finished_at: Some(Utc::now()),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
async fn rollback(&self) -> Result<SystemOperationAccepted, SystemOperationError> {
|
||||
|
||||
@@ -10,6 +10,7 @@ mod swap;
|
||||
use std::env;
|
||||
use std::fs;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
@@ -36,7 +37,8 @@ use self::release::{
|
||||
select_archive, validate_update_target, version_channel,
|
||||
};
|
||||
use self::state::{
|
||||
OperationFileLock, UpdateTempDir, finish, operation_id, read_status, set_running,
|
||||
OperationFileLock, UpdateOperation, UpdateTempDir, finish, operation_id, read_status,
|
||||
recover_interrupted, set_running,
|
||||
};
|
||||
use self::swap::{replace_release_files, rollback_release};
|
||||
|
||||
@@ -212,12 +214,13 @@ impl SystemUpdateConfig {
|
||||
}
|
||||
|
||||
/// gateway-admin 消费的真实进程/文件系统操作实现。
|
||||
#[derive(Clone)]
|
||||
pub struct ProcessSystemOperations {
|
||||
cancellation: CancellationToken,
|
||||
events: UpdateEvents,
|
||||
config: SystemUpdateConfig,
|
||||
operation_lock: AsyncMutex<()>,
|
||||
release_cache: ReleaseCache,
|
||||
events: Arc<UpdateEvents>,
|
||||
config: Arc<SystemUpdateConfig>,
|
||||
operation_lock: Arc<AsyncMutex<()>>,
|
||||
release_cache: Arc<ReleaseCache>,
|
||||
}
|
||||
|
||||
impl ProcessSystemOperations {
|
||||
@@ -225,21 +228,23 @@ impl ProcessSystemOperations {
|
||||
pub fn new(cancellation: CancellationToken, config: SystemUpdateConfig) -> Self {
|
||||
Self {
|
||||
cancellation,
|
||||
events: UpdateEvents::default(),
|
||||
config,
|
||||
operation_lock: AsyncMutex::const_new(()),
|
||||
release_cache: ReleaseCache::default(),
|
||||
events: Arc::new(UpdateEvents::default()),
|
||||
config: Arc::new(config),
|
||||
operation_lock: Arc::new(AsyncMutex::new(())),
|
||||
release_cache: Arc::new(ReleaseCache::default()),
|
||||
}
|
||||
}
|
||||
|
||||
async fn perform_update_inner(
|
||||
fn start_update(
|
||||
&self,
|
||||
target_version: Option<String>,
|
||||
) -> Result<SystemOperationAccepted, OperationError> {
|
||||
let _operation = self
|
||||
.operation_lock
|
||||
.try_lock()
|
||||
let operation_lock = Arc::clone(&self.operation_lock)
|
||||
.try_lock_owned()
|
||||
.map_err(|_| conflict("system operation is already running"))?;
|
||||
if self.cancellation.is_cancelled() {
|
||||
return Err(conflict("服务正在关闭"));
|
||||
}
|
||||
if let Some(reason) = self.config.update_support_error() {
|
||||
self.events
|
||||
.error_terminal(None, Some("preflight"), reason.clone());
|
||||
@@ -251,89 +256,89 @@ impl ProcessSystemOperations {
|
||||
.error_terminal(None, Some("preflight"), error.to_string());
|
||||
return Err(error);
|
||||
}
|
||||
let file_lock = OperationFileLock::acquire(&self.config.update_lock_file)?;
|
||||
let status = read_status(&self.config.update_state_file, &self.config.version)?;
|
||||
if status.need_restart {
|
||||
return Err(conflict("更新已完成,请先重启服务"));
|
||||
}
|
||||
let operation_id = operation_id("update");
|
||||
let mut operation = UpdateOperation::start(
|
||||
&self.config.update_state_file,
|
||||
&operation_id,
|
||||
&target,
|
||||
&self.config.version,
|
||||
file_lock,
|
||||
)?;
|
||||
let accepted = SystemOperationAccepted::Update {
|
||||
operation_id: operation_id.clone(),
|
||||
deployment_mode: self.config.deployment_mode.clone(),
|
||||
message: "更新已开始".to_owned(),
|
||||
target_version: target.clone(),
|
||||
};
|
||||
let service = self.clone();
|
||||
// 任务持有互斥锁和落盘守卫,HTTP 断开不会取消更新;Host 关闭仍会收敛终态。
|
||||
drop(tokio::spawn(async move {
|
||||
let _operation_lock = operation_lock;
|
||||
let result = tokio::select! {
|
||||
biased;
|
||||
() = service.cancellation.cancelled() => Err(conflict("服务关闭,更新已中断")),
|
||||
result = service.perform_update_inner(&target, &operation_id) => result,
|
||||
};
|
||||
if let Err(error) = operation.complete(&result) {
|
||||
tracing::warn!(error = %error, "系统更新终态落盘失败");
|
||||
return;
|
||||
}
|
||||
match result {
|
||||
Ok(()) => service.events.success_terminal(
|
||||
Some(&operation_id),
|
||||
Some("done"),
|
||||
"更新文件已替换,等待服务重启生效",
|
||||
),
|
||||
Err(error) => service.events.error_terminal(
|
||||
Some(&operation_id),
|
||||
Some("failed"),
|
||||
error.to_string(),
|
||||
),
|
||||
}
|
||||
}));
|
||||
Ok(accepted)
|
||||
}
|
||||
|
||||
async fn perform_update_inner(
|
||||
&self,
|
||||
target: &str,
|
||||
operation_id: &str,
|
||||
) -> Result<(), OperationError> {
|
||||
let repository = self
|
||||
.config
|
||||
.update_repository
|
||||
.as_deref()
|
||||
.ok_or_else(|| conflict("update repository is not configured"))?;
|
||||
self.events
|
||||
.info(None, Some("release"), "正在获取最新 Release 信息");
|
||||
let release = match fetch_latest(
|
||||
self.events.info(
|
||||
Some(operation_id),
|
||||
Some("release"),
|
||||
"正在获取最新 Release 信息",
|
||||
);
|
||||
let release = fetch_latest(
|
||||
&self.config.github_api_base,
|
||||
repository,
|
||||
&self.config.version,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(Some(release)) => release,
|
||||
Ok(None) => {
|
||||
let reason = "当前发行通道没有可用更新";
|
||||
self.events.warning_terminal(None, Some("release"), reason);
|
||||
return Err(conflict(reason));
|
||||
}
|
||||
Err(error) => {
|
||||
self.events
|
||||
.error_terminal(None, Some("release"), error.to_string());
|
||||
return Err(error);
|
||||
}
|
||||
};
|
||||
.await?
|
||||
.ok_or_else(|| conflict("当前发行通道没有可用更新"))?;
|
||||
let detail = detail_from_release(&self.config, &release);
|
||||
if detail.latest_version != target {
|
||||
self.events.warning_terminal(
|
||||
None,
|
||||
Some("release"),
|
||||
"remote latest version changed; confirm again",
|
||||
);
|
||||
return Err(conflict("remote latest version changed"));
|
||||
return Err(conflict("远端最新版本已变化,请重新确认"));
|
||||
}
|
||||
if !detail.has_update {
|
||||
self.events
|
||||
.warning_terminal(None, Some("release"), "already up to date");
|
||||
return Err(conflict("already up to date"));
|
||||
return Err(conflict("当前没有可用更新"));
|
||||
}
|
||||
|
||||
let operation_id = operation_id("update");
|
||||
let file_lock = OperationFileLock::acquire(&self.config.update_lock_file)?;
|
||||
set_running(
|
||||
&self.config.update_state_file,
|
||||
&operation_id,
|
||||
SystemOperationKind::Update,
|
||||
Some(&target),
|
||||
&self.config.version,
|
||||
)?;
|
||||
self.events.info(
|
||||
Some(&operation_id),
|
||||
Some(operation_id),
|
||||
Some("prepare"),
|
||||
format!("准备更新到 v{target}"),
|
||||
);
|
||||
let result = self.install_release(&release, &target, &operation_id).await;
|
||||
match &result {
|
||||
Ok(()) => self.events.success_terminal(
|
||||
Some(&operation_id),
|
||||
Some("done"),
|
||||
"更新文件已替换,等待服务重启生效",
|
||||
),
|
||||
Err(error) => {
|
||||
self.events
|
||||
.error_terminal(Some(&operation_id), Some("failed"), error.to_string())
|
||||
}
|
||||
}
|
||||
finish(
|
||||
&self.config.update_state_file,
|
||||
&operation_id,
|
||||
SystemOperationKind::Update,
|
||||
result.as_ref().ok().map(|()| target.clone()),
|
||||
result.as_ref().err().map(ToString::to_string),
|
||||
);
|
||||
drop(file_lock);
|
||||
result?;
|
||||
Ok(SystemOperationAccepted::Update {
|
||||
operation_id,
|
||||
deployment_mode: self.config.deployment_mode.clone(),
|
||||
message: "更新完成,请重启服务。".to_owned(),
|
||||
need_restart: true,
|
||||
target_version: target,
|
||||
})
|
||||
self.install_release(&release, target, operation_id).await
|
||||
}
|
||||
|
||||
async fn install_release(
|
||||
@@ -474,11 +479,17 @@ impl SystemOperations for ProcessSystemOperations {
|
||||
&self,
|
||||
target_version: Option<String>,
|
||||
) -> Result<SystemOperationAccepted, OperationError> {
|
||||
self.perform_update_inner(target_version).await
|
||||
self.start_update(target_version)
|
||||
}
|
||||
|
||||
async fn update_status(&self) -> Result<SystemUpdateStatus, OperationError> {
|
||||
read_status(&self.config.update_state_file)
|
||||
if let Ok(_operation) = self.operation_lock.try_lock() {
|
||||
recover_interrupted(
|
||||
&self.config.update_state_file,
|
||||
&self.config.update_lock_file,
|
||||
)?;
|
||||
}
|
||||
read_status(&self.config.update_state_file, &self.config.version)
|
||||
}
|
||||
|
||||
async fn rollback(&self) -> Result<SystemOperationAccepted, OperationError> {
|
||||
@@ -499,6 +510,13 @@ impl SystemOperations for ProcessSystemOperations {
|
||||
&self.config.version,
|
||||
)?;
|
||||
let result = rollback_release(&self.config);
|
||||
finish(
|
||||
&self.config.update_state_file,
|
||||
&operation_id,
|
||||
SystemOperationKind::Rollback,
|
||||
None,
|
||||
result.as_ref().err().map(ToString::to_string),
|
||||
)?;
|
||||
match &result {
|
||||
Ok(()) => self.events.success_terminal(
|
||||
Some(&operation_id),
|
||||
@@ -510,13 +528,6 @@ impl SystemOperations for ProcessSystemOperations {
|
||||
.error_terminal(Some(&operation_id), Some("failed"), error.to_string())
|
||||
}
|
||||
}
|
||||
finish(
|
||||
&self.config.update_state_file,
|
||||
&operation_id,
|
||||
SystemOperationKind::Rollback,
|
||||
None,
|
||||
result.as_ref().err().map(ToString::to_string),
|
||||
);
|
||||
drop(file_lock);
|
||||
result?;
|
||||
Ok(SystemOperationAccepted::Rollback {
|
||||
@@ -617,22 +628,6 @@ impl UpdateEvents {
|
||||
);
|
||||
}
|
||||
|
||||
fn warning_terminal(
|
||||
&self,
|
||||
operation_id: Option<&str>,
|
||||
step: Option<&str>,
|
||||
message: impl Into<String>,
|
||||
) {
|
||||
self.emit(
|
||||
SystemUpdateEventLevel::Warning,
|
||||
operation_id,
|
||||
step,
|
||||
message,
|
||||
None,
|
||||
true,
|
||||
);
|
||||
}
|
||||
|
||||
fn success_terminal(
|
||||
&self,
|
||||
operation_id: Option<&str>,
|
||||
|
||||
@@ -115,6 +115,7 @@ impl ReleaseCache {
|
||||
}
|
||||
}
|
||||
|
||||
/// 优先返回可升级的最高版本,没有候选时保留当前版本的发布信息。
|
||||
pub(crate) async fn fetch_latest(
|
||||
api_base: &str,
|
||||
repository: &str,
|
||||
@@ -166,7 +167,7 @@ fn eligible_release_version(
|
||||
let version = semver::Version::parse(&normalize_version(&release.tag_name)).ok()?;
|
||||
(!release.draft
|
||||
&& release.prerelease != version.pre.is_empty()
|
||||
&& update_target_allowed(current, &version))
|
||||
&& (version == *current || update_target_allowed(current, &version)))
|
||||
.then_some(version)
|
||||
}
|
||||
|
||||
@@ -198,15 +199,17 @@ pub(crate) fn detail_from_release(
|
||||
release: &GitHubRelease,
|
||||
) -> SystemUpdateDetail {
|
||||
let unsupported_reason = config.update_support_error();
|
||||
let latest_version = normalize_version(&release.tag_name);
|
||||
let has_update = validate_update_target(&config.version, &latest_version).is_ok();
|
||||
if unsupported_reason.is_some()
|
||||
|| validate_update_target(&config.version, &release.tag_name).is_err()
|
||||
|| (!has_update && latest_version != normalize_version(&config.version))
|
||||
{
|
||||
return super::base_update_detail(config, unsupported_reason, None);
|
||||
}
|
||||
SystemUpdateDetail {
|
||||
current_version: config.version.clone(),
|
||||
latest_version: normalize_version(&release.tag_name),
|
||||
has_update: true,
|
||||
latest_version,
|
||||
has_update,
|
||||
deployment_mode: config.deployment_mode.clone(),
|
||||
build_type: config.build_type.clone(),
|
||||
release_url: release.html_url.clone(),
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
use std::fs;
|
||||
use std::io::{self, Write};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
@@ -154,8 +155,101 @@ impl Drop for UpdateTempDir {
|
||||
}
|
||||
}
|
||||
|
||||
/// 更新任务拥有持久化终态的责任,Future 在取消或 panic 时析构也必须收尾。
|
||||
pub(crate) struct UpdateOperation {
|
||||
path: PathBuf,
|
||||
operation_id: String,
|
||||
target: String,
|
||||
completed: bool,
|
||||
_file_lock: OperationFileLock,
|
||||
}
|
||||
|
||||
impl UpdateOperation {
|
||||
pub(crate) fn start(
|
||||
path: &Path,
|
||||
operation_id: &str,
|
||||
target: &str,
|
||||
current_version: &str,
|
||||
file_lock: OperationFileLock,
|
||||
) -> Result<Self, OperationError> {
|
||||
set_running(
|
||||
path,
|
||||
operation_id,
|
||||
SystemOperationKind::Update,
|
||||
Some(target),
|
||||
current_version,
|
||||
)?;
|
||||
Ok(Self {
|
||||
path: path.to_owned(),
|
||||
operation_id: operation_id.to_owned(),
|
||||
target: target.to_owned(),
|
||||
completed: false,
|
||||
_file_lock: file_lock,
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn complete(
|
||||
&mut self,
|
||||
result: &Result<(), OperationError>,
|
||||
) -> Result<(), OperationError> {
|
||||
finish(
|
||||
&self.path,
|
||||
&self.operation_id,
|
||||
SystemOperationKind::Update,
|
||||
result.as_ref().ok().map(|()| self.target.clone()),
|
||||
result.as_ref().err().map(ToString::to_string),
|
||||
)?;
|
||||
self.completed = true;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for UpdateOperation {
|
||||
fn drop(&mut self) {
|
||||
if !self.completed
|
||||
&& let Err(error) = finish(
|
||||
&self.path,
|
||||
&self.operation_id,
|
||||
SystemOperationKind::Update,
|
||||
None,
|
||||
Some("更新任务已中断,请重新发起更新".to_owned()),
|
||||
)
|
||||
{
|
||||
tracing::warn!(error = %error, "收敛中断的系统更新状态失败");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 只在拿到进程内锁后调用,旧版本断连遗留且已无执行者的 running 不能永久保留。
|
||||
pub(crate) fn recover_interrupted(path: &Path, lock_path: &Path) -> Result<(), OperationError> {
|
||||
let state = read_persisted(path)?;
|
||||
if !matches!(state.operation.status, PersistedStatus::Running) {
|
||||
return Ok(());
|
||||
}
|
||||
let _lock = match OperationFileLock::acquire(lock_path) {
|
||||
Ok(lock) => lock,
|
||||
Err(error) if error.kind() == super::SystemOperationErrorKind::Conflict => return Ok(()),
|
||||
Err(error) => return Err(error),
|
||||
};
|
||||
if let (Some(operation_id), Some(kind)) = (state.operation.operation_id, state.operation.kind) {
|
||||
finish(
|
||||
path,
|
||||
&operation_id,
|
||||
kind.into(),
|
||||
None,
|
||||
Some("更新任务已中断,请重新发起更新".to_owned()),
|
||||
)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn operation_id(kind: &str) -> String {
|
||||
format!("sysop-{kind}-{}", Utc::now().timestamp_millis())
|
||||
static SEQUENCE: AtomicU64 = AtomicU64::new(0);
|
||||
format!(
|
||||
"sysop-{kind}-{}-{}",
|
||||
Utc::now().timestamp_millis(),
|
||||
SEQUENCE.fetch_add(1, Ordering::Relaxed)
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) fn set_running(
|
||||
@@ -188,16 +282,10 @@ pub(crate) fn finish(
|
||||
kind: SystemOperationKind,
|
||||
version: Option<String>,
|
||||
error: Option<String>,
|
||||
) {
|
||||
let mut state = match read_persisted(path) {
|
||||
Ok(state) => state,
|
||||
Err(error) => {
|
||||
tracing::warn!(error = %error, "读取系统更新状态失败");
|
||||
return;
|
||||
}
|
||||
};
|
||||
) -> Result<(), OperationError> {
|
||||
let mut state = read_persisted(path)?;
|
||||
if state.operation.operation_id.as_deref() != Some(operation_id) {
|
||||
return;
|
||||
return Ok(());
|
||||
}
|
||||
if let Some(error) = error {
|
||||
state.operation.status = PersistedStatus::Failed;
|
||||
@@ -222,15 +310,20 @@ pub(crate) fn finish(
|
||||
state.operation.target_version = version;
|
||||
}
|
||||
state.operation.finished_at = Some(Utc::now().to_rfc3339());
|
||||
if let Err(error) = write_persisted(path, &state) {
|
||||
tracing::warn!(error = %error, "写入系统更新状态失败");
|
||||
}
|
||||
write_persisted(path, &state)
|
||||
}
|
||||
|
||||
pub(crate) fn read_status(path: &Path) -> Result<SystemUpdateStatus, OperationError> {
|
||||
pub(crate) fn read_status(
|
||||
path: &Path,
|
||||
running_version: &str,
|
||||
) -> Result<SystemUpdateStatus, OperationError> {
|
||||
let state = read_persisted(path)?;
|
||||
let operation = state.operation;
|
||||
Ok(SystemUpdateStatus {
|
||||
need_restart: matches!(operation.status, PersistedStatus::Succeeded)
|
||||
&& state.current_version.as_deref().is_some_and(|version| {
|
||||
version.trim_start_matches('v') != running_version.trim_start_matches('v')
|
||||
}),
|
||||
previous_version: state.previous_version,
|
||||
current_version: state.current_version,
|
||||
operation: SystemOperationState {
|
||||
|
||||
@@ -7,7 +7,8 @@ use filetime::FileTime;
|
||||
use flate2::{Compression, write::GzEncoder};
|
||||
use futures::StreamExt as _;
|
||||
use gateway_admin::model::system::{
|
||||
SystemOperationKind, SystemOperationStatus, SystemUpdateEventLevel,
|
||||
SystemOperationAccepted, SystemOperationKind, SystemOperationStatus, SystemUpdateEventLevel,
|
||||
SystemUpdateStatus,
|
||||
};
|
||||
use gateway_admin::ports::system::{SystemOperationErrorKind, SystemOperations};
|
||||
use gateway_core::lifecycle::CancellationToken;
|
||||
@@ -90,7 +91,7 @@ async fn restart_should_conflict_while_another_system_operation_is_running() {
|
||||
.expect("release listener");
|
||||
let api_base = format!("http://{}/repos", listener.local_addr().expect("addr"));
|
||||
let (connected_tx, connected_rx) = tokio::sync::oneshot::channel();
|
||||
tokio::spawn(async move {
|
||||
let upstream = tokio::spawn(async move {
|
||||
let connection = listener.accept().await;
|
||||
let _ = connected_tx.send(());
|
||||
tokio::time::sleep(Duration::from_secs(30)).await;
|
||||
@@ -118,8 +119,13 @@ async fn restart_should_conflict_while_another_system_operation_is_running() {
|
||||
assert_eq!(error.kind(), SystemOperationErrorKind::Conflict);
|
||||
assert!(!shutdown.is_cancelled());
|
||||
|
||||
update.abort();
|
||||
let _ = update.await;
|
||||
update.await.expect("request").expect("accepted");
|
||||
assert!(fixture.lock().exists());
|
||||
upstream.abort();
|
||||
assert_eq!(
|
||||
wait_for_update(&service).await.operation.status,
|
||||
SystemOperationStatus::Failed
|
||||
);
|
||||
service.restart().await.expect("restart after lock release");
|
||||
}
|
||||
|
||||
@@ -136,10 +142,13 @@ async fn rollback_should_restore_binary_web_and_version_state() {
|
||||
)
|
||||
.await;
|
||||
let service = fixture.service(&server);
|
||||
service
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
.await
|
||||
.expect("update");
|
||||
assert_eq!(
|
||||
complete_update(&service, TARGET_VERSION)
|
||||
.await
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Succeeded
|
||||
);
|
||||
service.rollback().await.expect("rollback");
|
||||
|
||||
assert_eq!(
|
||||
@@ -353,10 +362,13 @@ async fn experimental_update_should_install_same_channel_release() {
|
||||
config.build_type = "experimental".to_owned();
|
||||
let service = ProcessSystemOperations::new(CancellationToken::new(), config);
|
||||
|
||||
service
|
||||
.perform_update(Some("3.10.0-exp.3".to_owned()))
|
||||
.await
|
||||
.expect("same-channel update");
|
||||
assert_eq!(
|
||||
complete_update(&service, "3.10.0-exp.3")
|
||||
.await
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Succeeded
|
||||
);
|
||||
assert_eq!(
|
||||
fs::read(fixture.executable()).expect("binary"),
|
||||
b"new-binary"
|
||||
@@ -468,6 +480,164 @@ async fn update_checks_should_follow_stage_promotion_and_experimental_isolation(
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_detail_should_keep_current_release_notes_without_allowing_reinstallation() {
|
||||
for current in [
|
||||
"3.12.1",
|
||||
"3.12.1-alpha.1",
|
||||
"3.12.1-beta.1",
|
||||
"3.12.1-rc.1",
|
||||
"3.12.1-exp.1",
|
||||
"3.12.1+build.1",
|
||||
] {
|
||||
let server = MockServer::start().await;
|
||||
let fixture = Fixture::new();
|
||||
let release_url = format!("https://github.com/owner/repository/releases/tag/v{current}");
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/repos/owner/repository/releases"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([
|
||||
{
|
||||
"tag_name": format!("v{current}"),
|
||||
"prerelease": !semver::Version::parse(current).expect("version").pre.is_empty(),
|
||||
"body": "## 当前版本\n\n- 修复更新状态",
|
||||
"html_url": release_url,
|
||||
}
|
||||
])))
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
let mut config = fixture.config(&format!("{}/repos", server.uri()));
|
||||
config.version = current.to_owned();
|
||||
if current.contains("-exp.") {
|
||||
config.build_type = "experimental".to_owned();
|
||||
}
|
||||
let service = ProcessSystemOperations::new(CancellationToken::new(), config);
|
||||
|
||||
let detail = service.update_detail(true).await.expect("current release");
|
||||
assert_eq!(detail.latest_version, current);
|
||||
assert!(!detail.has_update);
|
||||
assert!(detail.update_supported);
|
||||
assert_eq!(
|
||||
detail.notes.as_deref(),
|
||||
Some("## 当前版本\n\n- 修复更新状态")
|
||||
);
|
||||
assert_eq!(detail.release_url.as_deref(), Some(release_url.as_str()));
|
||||
let cached = service.update_detail(false).await.expect("cached release");
|
||||
assert!(cached.cached);
|
||||
assert_eq!(cached.notes, detail.notes);
|
||||
assert_eq!(cached.release_url, detail.release_url);
|
||||
assert!(!service.version().await.expect("version").has_update);
|
||||
|
||||
let error = service
|
||||
.perform_update(Some(current.to_owned()))
|
||||
.await
|
||||
.expect_err("current release must not be installed again");
|
||||
assert_eq!(error.kind(), SystemOperationErrorKind::Conflict);
|
||||
assert!(!fixture.state().exists());
|
||||
assert_eq!(
|
||||
fs::read(fixture.executable()).expect("binary"),
|
||||
b"old-binary"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_detail_should_prefer_available_update_notes_over_current_release() {
|
||||
for current_first in [true, false] {
|
||||
let server = MockServer::start().await;
|
||||
let fixture = Fixture::new();
|
||||
let mut releases = vec![
|
||||
serde_json::json!({
|
||||
"tag_name": "v1.0.0", "prerelease": false, "body": "current notes"
|
||||
}),
|
||||
serde_json::json!({
|
||||
"tag_name": "v1.9.9", "prerelease": false, "body": "update notes",
|
||||
"html_url": "https://github.com/owner/repository/releases/tag/v1.9.9"
|
||||
}),
|
||||
];
|
||||
if !current_first {
|
||||
releases.reverse();
|
||||
}
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/repos/owner/repository/releases"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(releases))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let detail = fixture
|
||||
.service(&server)
|
||||
.update_detail(true)
|
||||
.await
|
||||
.expect("detail");
|
||||
assert!(detail.has_update);
|
||||
assert_eq!(detail.latest_version, TARGET_VERSION);
|
||||
assert_eq!(detail.notes.as_deref(), Some("update notes"));
|
||||
assert_eq!(
|
||||
detail.release_url.as_deref(),
|
||||
Some("https://github.com/owner/repository/releases/tag/v1.9.9")
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_detail_should_find_current_release_notes_across_pages() {
|
||||
let server = MockServer::start().await;
|
||||
let fixture = Fixture::new();
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/repos/owner/repository/releases"))
|
||||
.and(query_param("page", "1"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(vec![
|
||||
serde_json::json!({ "tag_name": "v2.0.0", "prerelease": false });
|
||||
100
|
||||
]))
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/repos/owner/repository/releases"))
|
||||
.and(query_param("page", "2"))
|
||||
.respond_with(release_response("1.0.0", Vec::new()))
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let detail = fixture
|
||||
.service(&server)
|
||||
.update_detail(true)
|
||||
.await
|
||||
.expect("detail");
|
||||
assert!(!detail.has_update);
|
||||
assert_eq!(detail.latest_version, "1.0.0");
|
||||
assert_eq!(detail.notes.as_deref(), Some("notes"));
|
||||
assert!(detail.release_url.is_some());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_detail_should_not_use_draft_or_mislabeled_current_release_notes() {
|
||||
for (current, prerelease) in [("1.0.0", false), ("1.0.0-beta.1", true)] {
|
||||
let server = MockServer::start().await;
|
||||
let fixture = Fixture::new();
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/repos/owner/repository/releases"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([
|
||||
{ "tag_name": format!("v{current}"), "prerelease": prerelease, "draft": true, "body": "draft notes" },
|
||||
{ "tag_name": format!("v{current}"), "prerelease": !prerelease, "body": "mislabeled notes" },
|
||||
{ "tag_name": "v0.9.0", "prerelease": false, "body": "older notes" },
|
||||
])))
|
||||
.mount(&server)
|
||||
.await;
|
||||
let mut config = fixture.config(&format!("{}/repos", server.uri()));
|
||||
config.version = current.to_owned();
|
||||
let service = ProcessSystemOperations::new(CancellationToken::new(), config);
|
||||
|
||||
let detail = service.update_detail(true).await.expect("detail");
|
||||
assert!(!detail.has_update);
|
||||
assert_eq!(detail.latest_version, current);
|
||||
assert!(detail.notes.is_none());
|
||||
assert!(detail.release_url.is_none());
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_checks_should_ignore_other_cycles_downgrades_and_build_metadata() {
|
||||
for (current, target) in [
|
||||
@@ -475,7 +645,6 @@ async fn update_checks_should_ignore_other_cycles_downgrades_and_build_metadata(
|
||||
("3.12.0-beta.1", "3.12.1-rc.1"),
|
||||
("3.12.0-rc.1", "3.13.0"),
|
||||
("3.12.0-beta.2", "3.12.0-beta.1"),
|
||||
("3.12.0-exp.2", "3.12.0-exp.2"),
|
||||
("3.12.0", "3.12.0+new-build"),
|
||||
("3.12.0-beta.1+aaa", "3.12.0-beta.1+zzz"),
|
||||
("3.12.0", "3.11.0"),
|
||||
@@ -576,10 +745,10 @@ async fn prerelease_updates_should_install_later_stages_and_stable_release() {
|
||||
.expect("detail")
|
||||
.has_update
|
||||
);
|
||||
service
|
||||
.perform_update(Some(target.to_owned()))
|
||||
.await
|
||||
.expect("promoted update");
|
||||
assert_eq!(
|
||||
complete_update(&service, target).await.operation.status,
|
||||
SystemOperationStatus::Succeeded
|
||||
);
|
||||
assert_eq!(
|
||||
fs::read(fixture.executable()).expect("binary"),
|
||||
b"new-binary"
|
||||
@@ -694,10 +863,13 @@ async fn update_events_should_preserve_complete_release_stage_sequence() {
|
||||
.await;
|
||||
let service = fixture.service(&server);
|
||||
let events = service.update_events();
|
||||
service
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
.await
|
||||
.expect("update");
|
||||
assert_eq!(
|
||||
complete_update(&service, TARGET_VERSION)
|
||||
.await
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Succeeded
|
||||
);
|
||||
let steps = events
|
||||
.map(|event| event.step.expect("release update step"))
|
||||
.collect::<Vec<_>>()
|
||||
@@ -726,10 +898,13 @@ async fn update_events_should_report_download_bytes_and_percent() {
|
||||
.await;
|
||||
let service = fixture.service(&server);
|
||||
let events = service.update_events();
|
||||
service
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
.await
|
||||
.expect("update");
|
||||
assert_eq!(
|
||||
complete_update(&service, TARGET_VERSION)
|
||||
.await
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Succeeded
|
||||
);
|
||||
let progress = events
|
||||
.filter_map(
|
||||
|event| async move { event.progress_percent.map(|value| (value, event.message)) },
|
||||
@@ -755,12 +930,12 @@ async fn update_should_fail_when_release_checksum_is_missing() {
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
fixture
|
||||
.service(&server)
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
assert_eq!(
|
||||
complete_update(&fixture.service(&server), TARGET_VERSION)
|
||||
.await
|
||||
.is_err()
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Failed,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -777,12 +952,12 @@ async fn update_should_fail_when_release_checksum_mismatches() {
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
fixture
|
||||
.service(&server)
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
assert_eq!(
|
||||
complete_update(&fixture.service(&server), TARGET_VERSION)
|
||||
.await
|
||||
.is_err()
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Failed,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -794,12 +969,12 @@ async fn update_should_reject_insecure_release_archive() {
|
||||
.mount_custom_release(&server, "http://github.com/archive", None)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
fixture
|
||||
.service(&server)
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
assert_eq!(
|
||||
complete_update(&fixture.service(&server), TARGET_VERSION)
|
||||
.await
|
||||
.is_err()
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Failed,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -811,12 +986,12 @@ async fn update_should_reject_release_archive_from_untrusted_host() {
|
||||
.mount_custom_release(&server, "https://evil.example/archive", None)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
fixture
|
||||
.service(&server)
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
assert_eq!(
|
||||
complete_update(&fixture.service(&server), TARGET_VERSION)
|
||||
.await
|
||||
.is_err()
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Failed,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -844,12 +1019,12 @@ async fn update_should_reject_release_archive_with_unsafe_path() {
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
fixture
|
||||
.service(&server)
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
assert_eq!(
|
||||
complete_update(&fixture.service(&server), TARGET_VERSION)
|
||||
.await
|
||||
.is_err()
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Failed,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -876,12 +1051,12 @@ async fn update_should_reject_when_confirmed_target_differs_from_remote_latest()
|
||||
.mount_release(&server, "1.9.8", ArchiveKind::Safe, ChecksumKind::Valid)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
fixture
|
||||
.service(&server)
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
assert_eq!(
|
||||
complete_update(&fixture.service(&server), TARGET_VERSION)
|
||||
.await
|
||||
.is_err()
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Failed,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -900,12 +1075,12 @@ async fn update_should_remove_stale_file_lock_and_continue() {
|
||||
fs::write(fixture.lock(), "stale").expect("lock");
|
||||
filetime::set_file_mtime(fixture.lock(), FileTime::from_unix_time(1, 0)).expect("old mtime");
|
||||
|
||||
assert!(
|
||||
fixture
|
||||
.service(&server)
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
assert_eq!(
|
||||
complete_update(&fixture.service(&server), TARGET_VERSION)
|
||||
.await
|
||||
.is_ok()
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Succeeded,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -935,10 +1110,13 @@ async fn update_and_rollback_should_use_the_default_api_asset_directory() {
|
||||
host.resolve_and_validate(&fixture.root.path().join("deploy"), &fixture.web())
|
||||
.expect("inherit API asset directory");
|
||||
let service = ProcessSystemOperations::new(CancellationToken::new(), host.system_update);
|
||||
service
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
.await
|
||||
.expect("update");
|
||||
assert_eq!(
|
||||
complete_update(&service, TARGET_VERSION)
|
||||
.await
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Succeeded
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
fs::read(fixture.executable()).expect("binary"),
|
||||
@@ -981,10 +1159,13 @@ async fn update_should_replace_web_assets_across_filesystems() {
|
||||
fs::create_dir_all(&web_dist).expect("web dir");
|
||||
fs::write(web_dist.join("index.html"), "old-web").expect("web");
|
||||
let service = ProcessSystemOperations::new(CancellationToken::new(), config);
|
||||
service
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
.await
|
||||
.expect("update");
|
||||
assert_eq!(
|
||||
complete_update(&service, TARGET_VERSION)
|
||||
.await
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Succeeded
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
fs::read(web_dist.join("index.html")).expect("web"),
|
||||
@@ -1006,12 +1187,12 @@ async fn update_should_restore_web_assets_when_binary_backup_fails() {
|
||||
.await;
|
||||
fs::create_dir(fixture.executable().with_extension("backup")).expect("blocking backup dir");
|
||||
|
||||
assert!(
|
||||
fixture
|
||||
.service(&server)
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
assert_eq!(
|
||||
complete_update(&fixture.service(&server), TARGET_VERSION)
|
||||
.await
|
||||
.is_err()
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Failed,
|
||||
);
|
||||
assert_eq!(
|
||||
fs::read(fixture.web().join("index.html")).expect("web"),
|
||||
@@ -1059,6 +1240,262 @@ async fn version_should_return_backend_build_metadata() {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn accepted_update_should_survive_a_lost_http_response() {
|
||||
let upstream = MockServer::start().await;
|
||||
let fixture = Fixture::new();
|
||||
fixture
|
||||
.mount_release(
|
||||
&upstream,
|
||||
TARGET_VERSION,
|
||||
ArchiveKind::Safe,
|
||||
ChecksumKind::Valid,
|
||||
)
|
||||
.await;
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/archive"))
|
||||
.respond_with(
|
||||
ResponseTemplate::new(200)
|
||||
.set_body_bytes(release_archive(ArchiveKind::Safe))
|
||||
.set_delay(Duration::from_secs(2)),
|
||||
)
|
||||
.with_priority(1)
|
||||
.mount(&upstream)
|
||||
.await;
|
||||
let service = Arc::new(fixture.service(&upstream));
|
||||
let handler_service = Arc::clone(&service);
|
||||
let accepted = Arc::new(tokio::sync::Notify::new());
|
||||
let handler_accepted = Arc::clone(&accepted);
|
||||
let router = axum::Router::new().route(
|
||||
"/update",
|
||||
axum::routing::post(move || {
|
||||
let service = Arc::clone(&handler_service);
|
||||
let accepted = Arc::clone(&handler_accepted);
|
||||
async move {
|
||||
service
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
.await
|
||||
.expect("accepted");
|
||||
accepted.notify_one();
|
||||
// 模拟受理响应丢失,客户端断开会取消此 HTTP handler。
|
||||
std::future::pending::<String>().await
|
||||
}
|
||||
}),
|
||||
);
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
||||
.await
|
||||
.expect("listener");
|
||||
let address = listener.local_addr().expect("address");
|
||||
let server = tokio::spawn(async move {
|
||||
axum::serve(listener, router).await.expect("serve");
|
||||
});
|
||||
let client = tokio::spawn(async move {
|
||||
reqwest::Client::new()
|
||||
.post(format!("http://{address}/update"))
|
||||
.send()
|
||||
.await
|
||||
});
|
||||
tokio::time::timeout(Duration::from_secs(1), accepted.notified())
|
||||
.await
|
||||
.expect("accepted before slow download finishes");
|
||||
client.abort();
|
||||
let _ = client.await;
|
||||
assert_eq!(
|
||||
service
|
||||
.update_status()
|
||||
.await
|
||||
.expect("running status")
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Running
|
||||
);
|
||||
assert_eq!(
|
||||
service
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
.await
|
||||
.expect_err("duplicate update")
|
||||
.kind(),
|
||||
SystemOperationErrorKind::Conflict
|
||||
);
|
||||
assert_eq!(
|
||||
service
|
||||
.rollback()
|
||||
.await
|
||||
.expect_err("rollback during update")
|
||||
.kind(),
|
||||
SystemOperationErrorKind::Conflict
|
||||
);
|
||||
let status = wait_for_update(&service).await;
|
||||
assert_eq!(status.operation.status, SystemOperationStatus::Succeeded);
|
||||
assert!(status.need_restart);
|
||||
assert_eq!(
|
||||
fs::read(fixture.executable()).expect("binary"),
|
||||
b"new-binary"
|
||||
);
|
||||
assert!(!fixture.lock().exists());
|
||||
assert_eq!(
|
||||
service
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
.await
|
||||
.expect_err("restart required")
|
||||
.kind(),
|
||||
SystemOperationErrorKind::Conflict
|
||||
);
|
||||
let mut restarted_config = fixture.config(&format!("{}/repos", upstream.uri()));
|
||||
restarted_config.version = TARGET_VERSION.to_owned();
|
||||
let restarted = ProcessSystemOperations::new(CancellationToken::new(), restarted_config);
|
||||
assert!(
|
||||
!restarted
|
||||
.update_status()
|
||||
.await
|
||||
.expect("restarted status")
|
||||
.need_restart
|
||||
);
|
||||
server.abort();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn host_shutdown_should_finish_an_accepted_update_and_release_its_lock() {
|
||||
let upstream = MockServer::start().await;
|
||||
let fixture = Fixture::new();
|
||||
fixture
|
||||
.mount_release(
|
||||
&upstream,
|
||||
TARGET_VERSION,
|
||||
ArchiveKind::Safe,
|
||||
ChecksumKind::Valid,
|
||||
)
|
||||
.await;
|
||||
let cancellation = CancellationToken::new();
|
||||
let service = ProcessSystemOperations::new(
|
||||
cancellation.clone(),
|
||||
fixture.config(&format!("{}/repos", upstream.uri())),
|
||||
);
|
||||
// 在后台任务第一次 poll 之前触发关闭,覆盖受理后的生命周期交接。
|
||||
service
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
.await
|
||||
.expect("accepted");
|
||||
cancellation.cancel();
|
||||
let status = wait_for_update(&service).await;
|
||||
assert_eq!(status.operation.status, SystemOperationStatus::Failed);
|
||||
assert!(status.operation.error.expect("reason").contains("中断"));
|
||||
assert!(!fixture.lock().exists());
|
||||
assert_eq!(
|
||||
fs::read(fixture.executable()).expect("binary"),
|
||||
b"old-binary"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dropped_update_task_should_persist_failure_even_before_its_first_poll() {
|
||||
let fixture = Fixture::new();
|
||||
let service = fixture.service_for_url("http://127.0.0.1:1/repos");
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.expect("runtime");
|
||||
runtime
|
||||
.block_on(service.perform_update(Some(TARGET_VERSION.to_owned())))
|
||||
.expect("accepted");
|
||||
drop(runtime);
|
||||
let status: serde_json::Value =
|
||||
serde_json::from_slice(&fs::read(fixture.state()).expect("state")).expect("JSON");
|
||||
assert_eq!(status["operation"]["status"], "failed");
|
||||
assert!(status["operation"]["finishedAt"].is_string());
|
||||
assert!(!fixture.lock().exists());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn status_should_recover_abandoned_running_state_without_stealing_a_live_lock() {
|
||||
let fixture = Fixture::new();
|
||||
fs::write(fixture.state(), r#"{"currentVersion":"1.0.0","operation":{"operationId":"abandoned","kind":"update","status":"running","targetVersion":"1.9.9"}}"#).expect("state");
|
||||
fs::write(fixture.lock(), "another process").expect("live lock");
|
||||
let service = fixture.service_for_url("http://127.0.0.1:1/repos");
|
||||
assert_eq!(
|
||||
service
|
||||
.update_status()
|
||||
.await
|
||||
.expect("locked status")
|
||||
.operation
|
||||
.status,
|
||||
SystemOperationStatus::Running
|
||||
);
|
||||
fs::remove_file(fixture.lock()).expect("release abandoned lock");
|
||||
let status = service.update_status().await.expect("recovered status");
|
||||
assert_eq!(status.operation.status, SystemOperationStatus::Failed);
|
||||
assert!(status.operation.finished_at.is_some());
|
||||
assert_eq!(status.operation.operation_id.as_deref(), Some("abandoned"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn terminal_event_should_only_be_visible_after_status_is_persisted() {
|
||||
let upstream = MockServer::start().await;
|
||||
let fixture = Fixture::new();
|
||||
fixture
|
||||
.mount_release(
|
||||
&upstream,
|
||||
TARGET_VERSION,
|
||||
ArchiveKind::Safe,
|
||||
ChecksumKind::Mismatch,
|
||||
)
|
||||
.await;
|
||||
let service = fixture.service(&upstream);
|
||||
let mut events = service.update_events();
|
||||
let SystemOperationAccepted::Update { operation_id, .. } = service
|
||||
.perform_update(Some(TARGET_VERSION.to_owned()))
|
||||
.await
|
||||
.expect("accepted")
|
||||
else {
|
||||
panic!("update")
|
||||
};
|
||||
tokio::time::timeout(Duration::from_secs(5), async {
|
||||
while let Some(event) = events.next().await {
|
||||
assert_eq!(event.operation_id.as_deref(), Some(operation_id.as_str()));
|
||||
if event.terminal {
|
||||
let status = service.update_status().await.expect("terminal status");
|
||||
assert_eq!(status.operation.status, SystemOperationStatus::Failed);
|
||||
assert!(status.operation.finished_at.is_some());
|
||||
return;
|
||||
}
|
||||
}
|
||||
panic!("missing terminal event");
|
||||
})
|
||||
.await
|
||||
.expect("terminal event");
|
||||
}
|
||||
|
||||
async fn wait_for_update(service: &ProcessSystemOperations) -> SystemUpdateStatus {
|
||||
tokio::time::timeout(Duration::from_secs(5), async {
|
||||
loop {
|
||||
let status = service.update_status().await.expect("update status");
|
||||
if status.operation.status != SystemOperationStatus::Running {
|
||||
assert!(status.operation.finished_at.is_some(), "terminal timestamp");
|
||||
return status;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(5)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("update reaches terminal state")
|
||||
}
|
||||
|
||||
async fn complete_update(service: &ProcessSystemOperations, target: &str) -> SystemUpdateStatus {
|
||||
let SystemOperationAccepted::Update { operation_id, .. } = service
|
||||
.perform_update(Some(target.to_owned()))
|
||||
.await
|
||||
.expect("update accepted")
|
||||
else {
|
||||
panic!("expected update operation")
|
||||
};
|
||||
let status = wait_for_update(service).await;
|
||||
assert_eq!(
|
||||
status.operation.operation_id.as_deref(),
|
||||
Some(operation_id.as_str())
|
||||
);
|
||||
status
|
||||
}
|
||||
|
||||
struct Fixture {
|
||||
root: tempfile::TempDir,
|
||||
}
|
||||
@@ -1118,10 +1555,11 @@ impl Fixture {
|
||||
}
|
||||
|
||||
fn service(&self, server: &MockServer) -> ProcessSystemOperations {
|
||||
ProcessSystemOperations::new(
|
||||
CancellationToken::new(),
|
||||
self.config(&format!("{}/repos", server.uri())),
|
||||
)
|
||||
self.service_for_url(&format!("{}/repos", server.uri()))
|
||||
}
|
||||
|
||||
fn service_for_url(&self, api_base: &str) -> ProcessSystemOperations {
|
||||
ProcessSystemOperations::new(CancellationToken::new(), self.config(api_base))
|
||||
}
|
||||
|
||||
async fn mount_release(
|
||||
|
||||
+17
-4
@@ -1370,7 +1370,7 @@ Key 已删除或未关联时为 `null`,不影响记录返回,不包含密钥
|
||||
| `GET` | `/api/admin/system/version` | 无 | 当前构建、部署模式和可用更新 |
|
||||
| `GET` | `/api/admin/system/update/detail` | `refresh=true|false` | 读取或强制刷新 Release 详情 |
|
||||
| `GET` | `/api/admin/system/update/events` | 无 | SSE 更新事件流 |
|
||||
| `POST` | `/api/admin/system/update` | 可选 `{ targetVersion }` | 开始在线更新 |
|
||||
| `POST` | `/api/admin/system/update` | `{ targetVersion }` | 受理后台在线更新,返回 `202` |
|
||||
| `GET` | `/api/admin/system/update/status` | 无 | 查询当前更新或回滚状态 |
|
||||
| `POST` | `/api/admin/system/rollback` | 无 | 回滚到保留的上一版本 |
|
||||
| `POST` | `/api/admin/system/restart` | 无 | 请求进程重启 |
|
||||
@@ -1379,8 +1379,21 @@ Key 已删除或未关联时为 `null`,不影响记录返回,不包含密钥
|
||||
`updateChannel` 由当前版本推导,取值为 `stable`、`alpha`、`beta`、`rc`、`exp`,无法识别时为 `unknown`。
|
||||
检查与执行使用同一规则,禁止的通道转换、跨实验线、跨大版本、降级或同版本重装均以 `40901` 拒绝。
|
||||
`hasUpdate=true` 仅表示当前构建支持在线更新,且存在允许的更高版本;`latestVersion`、`releaseUrl` 和
|
||||
`notes` 对应这个候选。没有候选或当前构建不支持时,`hasUpdate=false`、`latestVersion` 为当前版本,
|
||||
`releaseUrl` 和 `notes` 为空;不支持原因通过 `updateSupported=false`、`unsupportedReason` 返回。
|
||||
`notes` 对应这个候选。没有可升级候选时,`hasUpdate=false`、`latestVersion` 为当前版本,
|
||||
`releaseUrl` 和 `notes` 保留当前版本的已发布 Release 信息;找不到匹配当前版本的 Release 时为空。
|
||||
当前构建不支持在线更新时不查询 Release,`hasUpdate=false`、`latestVersion` 为当前版本,
|
||||
`releaseUrl` 和 `notes` 为空,不支持原因通过 `updateSupported=false`、`unsupportedReason` 返回。
|
||||
强制检查失败时通过 `warning` 返回错误,`hasUpdate=false`,不以旧缓存或“没有更新”掩盖失败。
|
||||
普通查询可复用 20 分钟内的结果。下载时仍会校验目标资产、校验和及归档,在线更新成功后按返回值重启。
|
||||
普通查询可复用 20 分钟内的结果。下载时仍会校验目标资产、校验和及归档。
|
||||
|
||||
更新 POST 在本地校验目标版本并持久化任务后返回 `202`,数据包含 `operationId`、`targetVersion`、
|
||||
`deploymentMode` 和 `message`,只表示已受理。Release 查询、远端目标复核、下载、校验及文件替换在后台
|
||||
执行,结果通过 `/update/status` 的 `operation` 查询:`status` 为 `idle`、`running`、`succeeded` 或 `failed`,
|
||||
终态包含 `finishedAt`,失败原因在 `error` 中。SSE 的 `operationId` 用于关联进度;终态事件发出前状态已落盘。
|
||||
连接中断不取消已受理任务;响应丢失时先查询状态,不自动重复提交。打开更新页面时也会恢复最近一次任务。
|
||||
|
||||
状态响应的 `currentVersion` 表示已安装文件的版本,运行中的版本仍以 `/version` 为准。
|
||||
`needRestart=true` 表示成功安装的版本尚未在当前进程生效,此时应调用重启接口,不能重复发起更新。
|
||||
Host 关闭或任务取消会记录失败终态;状态查询会收敛无执行锁的遗留 `running`。
|
||||
异常退出留下的锁仍遵循 30 分钟过期规则,未过期前不会抢占其他进程的操作。
|
||||
实例升级和仓库发版见 [部署文档](../deploy/README.md#镜像升级与源码构建)。
|
||||
|
||||
@@ -600,6 +600,8 @@ HTTP Client 构造失败也不会阻断网关启动。外部解析在已认证
|
||||
备份设置等明确的敏感 Admin 合同返回。
|
||||
- OAuth pending flow 使用有期限、带 owner 的一次性 claim;事务成功后才消费,失败释放 claim。
|
||||
- 在线更新校验 Release host、大小、SHA-256 和归档路径,并只允许同一大版本内更新。
|
||||
Host 在受理时持久化任务并转交后台执行,任务持有操作锁与终态写入责任,不依赖 HTTP 请求的生命周期。
|
||||
状态文件是更新结果的权威来源,SSE 终态在状态落盘后发送;Host 关闭与任务析构都必须收敛状态。
|
||||
|
||||
PostgreSQL 备份恢复属于人工维护操作。当前没有部署级维护模式开关,需要先停止应用,
|
||||
离线处理快照中的非终态任务、计划游标和到期清理条件,再重新验证对象存储并恢复计划。
|
||||
|
||||
@@ -34,10 +34,25 @@ export interface SystemUpdateAccepted {
|
||||
operationId: string
|
||||
deploymentMode: string
|
||||
message: string
|
||||
needRestart: boolean
|
||||
targetVersion: string
|
||||
}
|
||||
|
||||
export interface SystemUpdateStatus {
|
||||
previousVersion: string | null
|
||||
currentVersion: string | null
|
||||
needRestart: boolean
|
||||
operation: {
|
||||
operationId: string | null
|
||||
kind: 'update' | 'rollback' | 'restart' | null
|
||||
status: 'idle' | 'running' | 'succeeded' | 'failed'
|
||||
targetVersion: string | null
|
||||
message: string | null
|
||||
error: string | null
|
||||
startedAt: string | null
|
||||
finishedAt: string | null
|
||||
}
|
||||
}
|
||||
|
||||
export interface SystemRestartAccepted {
|
||||
message: string
|
||||
operationId: string
|
||||
@@ -67,11 +82,12 @@ export function getSystemUpdateDetail(data: SystemUpdateDetailQuery) {
|
||||
})
|
||||
}
|
||||
|
||||
export function performSystemUpdate(data: SystemUpdateTarget) {
|
||||
export function performSystemUpdate(data: SystemUpdateTarget, options: RequestOptions = {}) {
|
||||
return request<SystemUpdateAccepted>({
|
||||
url: '/api/admin/system/update',
|
||||
method: 'POST',
|
||||
data,
|
||||
...options,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -82,3 +98,11 @@ export function restartSystem(options: RequestOptions = {}) {
|
||||
...options,
|
||||
})
|
||||
}
|
||||
|
||||
export function getSystemUpdateStatus(options: RequestOptions = {}) {
|
||||
return request<SystemUpdateStatus>({
|
||||
url: '/api/admin/system/update/status',
|
||||
method: 'GET',
|
||||
...options,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -39,7 +39,6 @@ const {
|
||||
updateError,
|
||||
updateSuccess,
|
||||
needRestart,
|
||||
loadedOnce,
|
||||
updateLogs,
|
||||
updateStreaming,
|
||||
updateStreamError,
|
||||
@@ -144,8 +143,8 @@ async function handleUpdateRequest() {
|
||||
async function runConfirmedUpdate(targetVersion: string) {
|
||||
try {
|
||||
const result = await updateNow(targetVersion)
|
||||
if (result?.needRestart) {
|
||||
toast.success('更新完成,请重启服务')
|
||||
if (result) {
|
||||
toast.success('更新已开始')
|
||||
}
|
||||
}
|
||||
catch {}
|
||||
@@ -171,7 +170,7 @@ async function handleRestart() {
|
||||
}
|
||||
|
||||
watch(open, (visible) => {
|
||||
if (visible && !loadedOnce.value) {
|
||||
if (visible) {
|
||||
void loadSystem(false).catch(() => undefined)
|
||||
}
|
||||
})
|
||||
@@ -195,7 +194,7 @@ watch(
|
||||
description="检查版本、查看发布说明并执行在线更新"
|
||||
tone="success"
|
||||
size="lg"
|
||||
:dismissible="!updating && !restarting"
|
||||
:dismissible="!restarting"
|
||||
>
|
||||
<template #icon>
|
||||
<ArrowUpCircle class="size-4.5 text-cp-success" />
|
||||
|
||||
@@ -1,10 +1,12 @@
|
||||
import { until, useEventSource } from '@vueuse/core'
|
||||
import type { SystemUpdateStatus } from '@/api'
|
||||
import { until, useEventSource, useTimeoutPoll } from '@vueuse/core'
|
||||
import { delay } from 'es-toolkit'
|
||||
import { defineStore } from 'pinia'
|
||||
|
||||
import { computed, ref, shallowRef, watch } from 'vue'
|
||||
import {
|
||||
getSystemUpdateDetail,
|
||||
getSystemUpdateStatus,
|
||||
getSystemVersion,
|
||||
performSystemUpdate,
|
||||
restartSystem,
|
||||
@@ -18,9 +20,12 @@ const updateEventReadyTimeoutMs = 3_000
|
||||
const restartReadyTimeoutMs = 60_000
|
||||
const restartProbeTimeoutMs = 2_000
|
||||
const restartReadyPollIntervalMs = 500
|
||||
const updateStatusPollIntervalMs = 1_000
|
||||
const updateStatusTimeoutMs = 5_000
|
||||
|
||||
interface SystemUpdateEvent {
|
||||
id: string
|
||||
operationId?: string | null
|
||||
level: string
|
||||
message: string
|
||||
at: string
|
||||
@@ -58,6 +63,7 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
|
||||
const updateStreaming = shallowRef(false)
|
||||
const updateStreamError = shallowRef('')
|
||||
const restartTargetVersion = shallowRef('')
|
||||
const updateStatus = shallowRef<SystemUpdateStatus | null>(null)
|
||||
const {
|
||||
data: updateEventMessage,
|
||||
status: updateEventStatus,
|
||||
@@ -76,6 +82,12 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
|
||||
|
||||
let loadVersionPromise: ReturnType<typeof getSystemVersion> | undefined
|
||||
let loadSystemPromise: Promise<void> | undefined
|
||||
let statusRequest: Promise<void> | undefined
|
||||
let statusGeneration = 0
|
||||
let submitting = false
|
||||
let activeOperationId: string | null = null
|
||||
let unconfirmedPreviousId: string | null | undefined
|
||||
const statusPoll = useTimeoutPoll(refreshUpdateStatus, updateStatusPollIntervalMs, { immediate: false })
|
||||
|
||||
const phaseKind = computed(() => phase.value.kind)
|
||||
const loading = computed(() => phaseKind.value === 'loading')
|
||||
@@ -88,6 +100,7 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
|
||||
() =>
|
||||
hasUpdate.value
|
||||
&& Boolean(updateInfo.value?.updateSupported)
|
||||
&& !needRestart.value
|
||||
&& !UPDATE_BUSY_PHASES.has(phaseKind.value),
|
||||
)
|
||||
|
||||
@@ -128,9 +141,11 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
|
||||
return
|
||||
try {
|
||||
const event = JSON.parse(message.raw) as SystemUpdateEvent
|
||||
if (activeOperationId && event.operationId !== activeOperationId)
|
||||
return
|
||||
appendUpdateLog(event)
|
||||
if (event.terminal)
|
||||
disconnectUpdateEvents()
|
||||
if (event.terminal && !submitting)
|
||||
void refreshUpdateStatus()
|
||||
}
|
||||
catch {
|
||||
updateStreamError.value = '更新日志解析失败'
|
||||
@@ -167,6 +182,80 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
|
||||
closeUpdateEventSource()
|
||||
}
|
||||
|
||||
function applyUpdateStatus(status: SystemUpdateStatus) {
|
||||
const operation = status.operation
|
||||
updateStatus.value = status
|
||||
if (unconfirmedPreviousId !== undefined && operation.operationId === unconfirmedPreviousId) {
|
||||
unconfirmedPreviousId = undefined
|
||||
updateError.value = '尚未确认更新任务,请检查状态后重试'
|
||||
setPhase({ kind: 'ready' })
|
||||
statusPoll.pause()
|
||||
disconnectUpdateEvents()
|
||||
return
|
||||
}
|
||||
unconfirmedPreviousId = undefined
|
||||
activeOperationId = operation.operationId
|
||||
updateError.value = ''
|
||||
updateSuccess.value = operation.status === 'succeeded' && status.needRestart
|
||||
needRestart.value = status.needRestart
|
||||
restartTargetVersion.value = status.needRestart ? normalizeSystemVersion(status.currentVersion) : ''
|
||||
if (operation.status === 'running') {
|
||||
setPhase({ kind: 'updating' })
|
||||
if (!statusPoll.isActive.value)
|
||||
statusPoll.resume()
|
||||
void connectUpdateEvents()
|
||||
return
|
||||
}
|
||||
statusPoll.pause()
|
||||
disconnectUpdateEvents()
|
||||
if (operation.status === 'failed') {
|
||||
updateError.value = operation.error || operation.message || '更新失败'
|
||||
setPhase({ kind: 'failed' })
|
||||
return
|
||||
}
|
||||
if (status.needRestart && updateInfo.value) {
|
||||
updateInfo.value = { ...updateInfo.value, hasUpdate: false }
|
||||
}
|
||||
setPhase(status.needRestart ? { kind: 'restart_required' } : { kind: 'ready' })
|
||||
}
|
||||
|
||||
async function refreshUpdateStatus() {
|
||||
if (submitting || restarting.value)
|
||||
return
|
||||
if (statusRequest)
|
||||
return statusRequest
|
||||
const generation = statusGeneration
|
||||
statusRequest = (async () => {
|
||||
try {
|
||||
const status = await getSystemUpdateStatus({ silent: true, timeout: updateStatusTimeoutMs })
|
||||
if (generation !== statusGeneration)
|
||||
return
|
||||
updateStreamError.value = ''
|
||||
applyUpdateStatus(status)
|
||||
}
|
||||
catch (error: unknown) {
|
||||
if (generation !== statusGeneration)
|
||||
return
|
||||
if (error instanceof ApiError && (error.status === 401 || error.status === 403)) {
|
||||
statusPoll.pause()
|
||||
disconnectUpdateEvents()
|
||||
updateError.value = errorMessage(error)
|
||||
setPhase({ kind: 'failed' })
|
||||
return
|
||||
}
|
||||
// 传输失败不能证明任务失败,保留忙碌态,恢复连接后重新核对持久化结果。
|
||||
updateStreamError.value = '暂时无法确认更新状态,正在重试'
|
||||
setPhase({ kind: 'updating' })
|
||||
if (!statusPoll.isActive.value)
|
||||
statusPoll.resume()
|
||||
}
|
||||
finally {
|
||||
statusRequest = undefined
|
||||
}
|
||||
})()
|
||||
return statusRequest
|
||||
}
|
||||
|
||||
async function loadSystem(refresh = false) {
|
||||
if (loadSystemPromise)
|
||||
return loadSystemPromise
|
||||
@@ -175,10 +264,17 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
|
||||
setPhase({ kind: 'loading' })
|
||||
}
|
||||
loadSystemPromise = (async () => {
|
||||
updateInfo.value = await getSystemUpdateDetail({ refresh })
|
||||
if (!version.value)
|
||||
version.value = await getSystemVersion()
|
||||
loadedOnce.value = true
|
||||
await Promise.all([
|
||||
refreshUpdateStatus(),
|
||||
(async () => {
|
||||
updateInfo.value = await getSystemUpdateDetail({ refresh })
|
||||
if (!version.value)
|
||||
version.value = await getSystemVersion()
|
||||
loadedOnce.value = true
|
||||
})(),
|
||||
])
|
||||
if (needRestart.value && updateInfo.value)
|
||||
updateInfo.value = { ...updateInfo.value, hasUpdate: false }
|
||||
})()
|
||||
|
||||
try {
|
||||
@@ -207,7 +303,7 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
|
||||
}
|
||||
|
||||
async function checkUpdates(refresh = true) {
|
||||
if (checking.value)
|
||||
if (checking.value || updating.value || restarting.value)
|
||||
return updateInfo.value
|
||||
|
||||
setPhase({ kind: 'checking' })
|
||||
@@ -217,6 +313,7 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
|
||||
if (!version.value)
|
||||
version.value = await getSystemVersion()
|
||||
loadedOnce.value = true
|
||||
await refreshUpdateStatus()
|
||||
return updateInfo.value
|
||||
}
|
||||
finally {
|
||||
@@ -226,40 +323,46 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
|
||||
}
|
||||
|
||||
async function updateNow(targetVersion: string) {
|
||||
const currentInfo = updateInfo.value
|
||||
const confirmedTargetVersion = targetVersion.trim()
|
||||
if (!canUpdate.value || !currentInfo || updating.value || !confirmedTargetVersion)
|
||||
const confirmedTargetVersion = normalizeSystemVersion(targetVersion)
|
||||
if (!canUpdate.value || !confirmedTargetVersion)
|
||||
return null
|
||||
|
||||
statusGeneration += 1
|
||||
submitting = true
|
||||
statusPoll.pause()
|
||||
activeOperationId = null
|
||||
unconfirmedPreviousId = undefined
|
||||
const previousId = updateStatus.value?.operation.operationId ?? null
|
||||
resetUpdateResult()
|
||||
clearUpdateLogs()
|
||||
await connectUpdateEvents(true)
|
||||
setPhase({ kind: 'updating' })
|
||||
updateError.value = ''
|
||||
updateSuccess.value = false
|
||||
try {
|
||||
const result = await performSystemUpdate({ targetVersion: confirmedTargetVersion })
|
||||
updateSuccess.value = true
|
||||
needRestart.value = result.needRestart
|
||||
restartTargetVersion.value = result.needRestart ? normalizeSystemVersion(result.targetVersion) : ''
|
||||
updateInfo.value = {
|
||||
...currentInfo,
|
||||
latestVersion: result.targetVersion,
|
||||
hasUpdate: false,
|
||||
}
|
||||
setPhase(result.needRestart ? { kind: 'restart_required' } : { kind: 'ready' })
|
||||
await connectUpdateEvents(true)
|
||||
const result = await performSystemUpdate({ targetVersion: confirmedTargetVersion }, { silent: true })
|
||||
activeOperationId = result.operationId
|
||||
updateLogs.value = updateLogs.value.filter(event => event.operationId === result.operationId)
|
||||
return result
|
||||
}
|
||||
catch (error: unknown) {
|
||||
if (error instanceof ApiError && (error.status === 0 || error.status >= 500 || error.status === 408)) {
|
||||
unconfirmedPreviousId = previousId
|
||||
updateStreamError.value = '正在确认更新是否已开始'
|
||||
return null
|
||||
}
|
||||
updateError.value = errorMessage(error)
|
||||
appendUpdateLog({
|
||||
id: `update-client-error-${Date.now()}`,
|
||||
level: 'error',
|
||||
message: updateError.value,
|
||||
at: new Date().toISOString(),
|
||||
})
|
||||
setPhase({ kind: 'failed' })
|
||||
disconnectUpdateEvents()
|
||||
throw error
|
||||
}
|
||||
finally {
|
||||
submitting = false
|
||||
if (updating.value) {
|
||||
statusPoll.resume()
|
||||
// 提交前的查询可能仍在返回,先丢弃旧代次结果,再读取本次任务。
|
||||
await statusRequest
|
||||
await refreshUpdateStatus()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForServiceAndReload() {
|
||||
@@ -299,6 +402,8 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
|
||||
throw error
|
||||
}
|
||||
|
||||
statusGeneration += 1
|
||||
statusPoll.pause()
|
||||
setPhase({ kind: 'restarting' })
|
||||
updateError.value = ''
|
||||
disconnectUpdateEvents()
|
||||
|
||||
Reference in New Issue
Block a user