+30
-112
@@ -18,118 +18,12 @@ import {
|
||||
selectReleaseAsset,
|
||||
validateHttpsUrl,
|
||||
RELEASE_NOTES_MAX_BYTES,
|
||||
downloadReleaseAsset,
|
||||
extractSafeArchive,
|
||||
normalizeReleasePermissions,
|
||||
applicationUpdateRuntimeHash,
|
||||
type ReleaseAsset,
|
||||
type ReleaseMetadata,
|
||||
} from "./update.js";
|
||||
import type { UpdateJobStatus } from "../shared/contracts.js";
|
||||
|
||||
|
||||
export const activeInProcessDownloads = new Map<string, AbortController>();
|
||||
|
||||
export function triggerInProcessDownload(
|
||||
database: Database.Database,
|
||||
config: AppConfig,
|
||||
jobId: string,
|
||||
asset: { name: string; url: string; sha256?: string },
|
||||
expectedSha256?: string,
|
||||
): void {
|
||||
setImmediate(async () => {
|
||||
try {
|
||||
const row = database.prepare("SELECT id, status, operation FROM update_jobs WHERE id=?").get(jobId) as { id: string; status: string; operation: string } | undefined;
|
||||
if (!row || row.status !== "queued") return;
|
||||
|
||||
const controller = new AbortController();
|
||||
activeInProcessDownloads.set(jobId, controller);
|
||||
|
||||
const workspace = path.join(path.resolve(config.stagingDir), `update-${jobId}`);
|
||||
const archivePath = path.join(workspace, asset.name.endsWith(".gz") || asset.name.endsWith(".zip") ? asset.name : `${asset.name}.tar.gz`);
|
||||
|
||||
await mkdir(workspace, { recursive: true, mode: 0o700 });
|
||||
const now = Date.now();
|
||||
database.prepare("UPDATE update_jobs SET status='downloading', download_started_at=?, started_at=?, download_path=?, updated_at=? WHERE id=? AND status='queued'").run(now, now, path.basename(archivePath), now, jobId);
|
||||
|
||||
const progressStartedAt = Date.now();
|
||||
let lastProgressWrite = 0;
|
||||
|
||||
const downloaded = await downloadReleaseAsset(asset.url, archivePath, {
|
||||
allowedHosts: config.updateAllowedHosts,
|
||||
maxBytes: config.updateMaxBytes,
|
||||
fetchImpl: (input, init) => fetch(input, { ...init, signal: controller.signal }),
|
||||
onProgress: (downloadedBytes, totalBytes) => {
|
||||
const cur = Date.now();
|
||||
if (cur - lastProgressWrite < 200) return;
|
||||
lastProgressWrite = cur;
|
||||
const elapsed = Math.max(1, cur - progressStartedAt);
|
||||
const speedBps = Math.round(downloadedBytes * 1000 / elapsed);
|
||||
try {
|
||||
database.prepare("UPDATE update_jobs SET downloaded_bytes=?, size_bytes=COALESCE(?, size_bytes), download_speed_bps=?, updated_at=? WHERE id=? AND status='downloading'").run(downloadedBytes, totalBytes, speedBps, cur, jobId);
|
||||
} catch {}
|
||||
},
|
||||
});
|
||||
|
||||
if (expectedSha256 && downloaded.sha256.toLowerCase() !== expectedSha256.toLowerCase()) {
|
||||
throw new Error("更新文件 SHA-256 校验失败");
|
||||
}
|
||||
|
||||
database.prepare("UPDATE update_jobs SET status='verifying', actual_sha256=?, size_bytes=?, downloaded_bytes=?, updated_at=? WHERE id=? AND status='downloading'").run(downloaded.sha256, downloaded.size, downloaded.size, Date.now(), jobId);
|
||||
|
||||
const stagedDir = path.join(workspace, "payload");
|
||||
await extractSafeArchive(archivePath, stagedDir, config.updateMaxBytes === undefined ? {} : { maxBytes: config.updateMaxBytes });
|
||||
|
||||
if (applicationUpdateRuntimeHash(asset.name)) {
|
||||
try {
|
||||
const currentRelease = realpathSync(config.currentLink);
|
||||
if (currentRelease) {
|
||||
for (const entry of ["node_modules", "runtime", "pnpm-lock.yaml"] as const) {
|
||||
const source = path.join(currentRelease, entry);
|
||||
const target = path.join(stagedDir, entry);
|
||||
let copied = false;
|
||||
try {
|
||||
const { execFile } = await import("node:child_process");
|
||||
const { promisify } = await import("node:util");
|
||||
await promisify(execFile)("cp", ["-a", source, target]);
|
||||
copied = true;
|
||||
} catch {}
|
||||
if (!copied) {
|
||||
const fsPromises = await import("node:fs/promises");
|
||||
const sourceInfo = await fsPromises.lstat(source).catch(() => null);
|
||||
if (sourceInfo && !sourceInfo.isSymbolicLink()) {
|
||||
await fsPromises.cp(source, target, { recursive: sourceInfo.isDirectory(), errorOnExist: true, force: false }).catch(() => {});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch {}
|
||||
}
|
||||
|
||||
await normalizeReleasePermissions(stagedDir).catch(() => {});
|
||||
const fsPromises = await import("node:fs/promises");
|
||||
const payloadInfo = await fsPromises.lstat(path.join(stagedDir, "dist")).catch(() => null);
|
||||
if (!payloadInfo?.isDirectory() || payloadInfo.isSymbolicLink()) {
|
||||
throw new Error("发布包缺少 dist 目录");
|
||||
}
|
||||
|
||||
database.prepare("UPDATE update_jobs SET status='staged', actual_sha256=?, size_bytes=?, download_path=?, updated_at=? WHERE id=? AND status IN ('verifying', 'downloading')").run(downloaded.sha256, downloaded.size, workspace, Date.now(), jobId);
|
||||
} catch (error) {
|
||||
const controller = activeInProcessDownloads.get(jobId);
|
||||
if (controller?.signal.aborted) return;
|
||||
const rawMsg = error instanceof Error ? error.message : "更新文件下载失败";
|
||||
try {
|
||||
database.prepare("UPDATE update_jobs SET status='failed', error_message=?, updated_at=? WHERE id=? AND status NOT IN ('completed', 'staged', 'cancelled')").run(rawMsg, Date.now(), jobId);
|
||||
} catch {}
|
||||
const workspace = path.join(path.resolve(config.stagingDir), `update-${jobId}`);
|
||||
const fsPromises = await import("node:fs/promises");
|
||||
await fsPromises.rm(workspace, { recursive: true, force: true }).catch(() => {});
|
||||
} finally {
|
||||
activeInProcessDownloads.delete(jobId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
export const UPDATE_CACHE_KEY = "update.release.v1";
|
||||
export const ACTIVE_UPDATE_STATUSES: readonly UpdateJobStatus[] = [
|
||||
"queued",
|
||||
@@ -574,6 +468,36 @@ export function reconcileOrphanedUpdateJobs(database: Database.Database, config:
|
||||
let reconciled = 0;
|
||||
const reconciledIds = new Set<string>();
|
||||
for (const row of rows) {
|
||||
// A request that never gets claimed by the root runner must not remain in
|
||||
// the UI as an endless "queued" task. Once the short hand-off window has
|
||||
// elapsed and no recovery marker exists, release the queue explicitly;
|
||||
// a fresh state marker proves that the runner has already claimed it.
|
||||
if (row.status === "queued" && typeof row.updatedAt === "number" && !stateFresh && now - row.updatedAt >= QUEUED_UPDATE_TIMEOUT_MS) {
|
||||
const changed = database.transaction(() => {
|
||||
const result = database.prepare(`
|
||||
UPDATE update_jobs
|
||||
SET status='failed', error_message=?, completed_at=?, updated_at=?
|
||||
WHERE id=? AND status='queued' AND updated_at=?
|
||||
`).run("更新服务未在规定时间内接管任务", now, now, row.id, row.updatedAt);
|
||||
if (result.changes !== 1) return false;
|
||||
writeAudit(database, {
|
||||
requestId: row.requestId || randomUUID(),
|
||||
actorAdminId: row.adminId,
|
||||
action: "update.reconciled",
|
||||
targetType: "update",
|
||||
targetId: row.id,
|
||||
outcome: "failure",
|
||||
before: { status: row.status, version: row.version },
|
||||
after: { status: "failed", version: row.version, reason: "runner_claim_timeout" },
|
||||
});
|
||||
return true;
|
||||
})();
|
||||
if (changed) {
|
||||
reconciled += 1;
|
||||
reconciledIds.add(row.id);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (typeof row.updatedAt !== "number" || now - row.updatedAt < ORPHANED_UPDATE_TIMEOUT_MS) continue;
|
||||
// The runner refreshes the state marker while a download is in flight.
|
||||
// A stale request/state marker therefore no longer protects an orphaned
|
||||
@@ -668,12 +592,6 @@ export function cancelUpdateJob(
|
||||
if (!job) return { cancelled: false, message: "当前没有处于等待调度或下载中的更新任务" };
|
||||
if (job.status !== "queued" && job.status !== "downloading") return { cancelled: false, message: "任务已进入就绪或切换阶段,无法取消" };
|
||||
|
||||
const controller = activeInProcessDownloads.get(job.id);
|
||||
if (controller) {
|
||||
controller.abort();
|
||||
activeInProcessDownloads.delete(job.id);
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
const changed = database.transaction(() => {
|
||||
const result = database.prepare("UPDATE update_jobs SET status='cancelled', error_message='已手动取消更新', completed_at=?, updated_at=? WHERE id=? AND status IN ('queued', 'downloading')").run(now, now, job.id);
|
||||
|
||||
Reference in New Issue
Block a user