From a1b4a63552f10f96ca4e9fd2e2dca156debc606f Mon Sep 17 00:00:00 2001 From: tianmo Date: Sat, 17 Oct 2026 15:08:21 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=8C=E6=AD=A5=E5=BC=82=E5=B8=B8=E5=85=9C?= =?UTF-8?q?=E5=BA=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/components/storage/SiteSyncCard.vue | 29 +++ app/composables/useBackup.ts | 7 +- app/types/backup.ts | 1 + i18n/locales/en.json | 3 + i18n/locales/zh-CN.json | 3 + server/api/admin/site-sync.post.ts | 16 +- server/api/site-sync/cleanup.post.ts | 20 ++ server/api/site-sync/complete.post.ts | 21 +++ server/api/site-sync/media.post.ts | 8 +- server/api/site-sync/object.post.ts | 14 ++ server/plugins/startup.ts | 34 ++++ server/utils/backup/cancel-state.ts | 64 +++++++ server/utils/backup/runner.ts | 20 +- server/utils/site-sync/abandon.ts | 66 +++++++ server/utils/site-sync/artifacts.ts | 172 +++++++++++++++++ server/utils/site-sync/runner.test.ts | 19 ++ server/utils/site-sync/runner.ts | 237 +++++++++++++++++++----- server/utils/storage/index.ts | 2 +- server/utils/storage/local.ts | 60 +++++- server/utils/storage/resolver.ts | 12 ++ 20 files changed, 754 insertions(+), 54 deletions(-) create mode 100644 server/api/site-sync/cleanup.post.ts create mode 100644 server/api/site-sync/complete.post.ts create mode 100644 server/utils/backup/cancel-state.ts create mode 100644 server/utils/site-sync/abandon.ts create mode 100644 server/utils/site-sync/artifacts.ts diff --git a/app/components/storage/SiteSyncCard.vue b/app/components/storage/SiteSyncCard.vue index 3e9c9ef..08dd711 100644 --- a/app/components/storage/SiteSyncCard.vue +++ b/app/components/storage/SiteSyncCard.vue @@ -57,6 +57,11 @@ const fileProgress = computed(() => { } }) const isPull = computed(() => direction.value === 'pull') +const canAbandon = computed(() => + !props.busy + && Boolean(props.resumableJob) + && (isPull.value || Boolean(token.value.trim())) +) const directionSubtitle = computed(() => ( isPull.value ? t('storage.backup.siteSyncPullSubtitle') @@ -103,6 +108,19 @@ function resumeSync() { submit(true) } +function abandonSync() { + if (!canAbandon.value || !props.resumableJob) return + if (!window.confirm(t('storage.backup.siteSyncAbandonConfirm'))) return + emit('sync', { + targetBaseUrl: targetBaseUrl.value.trim(), + token: token.value.trim(), + conflictMode: conflictMode.value, + direction: direction.value, + abandonJobId: props.resumableJob.id + }) + token.value = '' +} + function setConflictMode(mode: 'skip' | 'overwrite') { conflictMode.value = mode } @@ -171,6 +189,17 @@ function setDirection(value: SiteSyncDirection) { class="lg:col-start-3" @click="resumeSync" /> +
diff --git a/app/composables/useBackup.ts b/app/composables/useBackup.ts index 3ec3788..6a286a5 100644 --- a/app/composables/useBackup.ts +++ b/app/composables/useBackup.ts @@ -268,7 +268,12 @@ export function useBackup() { ) await refreshJobsFirstPage() startPolling() - toast.add({ title: t('storage.backup.siteSyncStarted'), color: 'success' }) + toast.add({ + title: body.abandonJobId + ? t('storage.backup.siteSyncAbandoned') + : t('storage.backup.siteSyncStarted'), + color: 'success' + }) return data.job } catch (error: unknown) { handleError(error, t('storage.backup.siteSyncFailed')) diff --git a/app/types/backup.ts b/app/types/backup.ts index 2ac71b6..c617419 100644 --- a/app/types/backup.ts +++ b/app/types/backup.ts @@ -59,4 +59,5 @@ export interface SiteSyncRequest { conflictMode: 'skip' | 'overwrite' direction: SiteSyncDirection resumeJobId?: string + abandonJobId?: string } diff --git a/i18n/locales/en.json b/i18n/locales/en.json index 761d4ff..cb2ce11 100644 --- a/i18n/locales/en.json +++ b/i18n/locales/en.json @@ -665,6 +665,9 @@ "siteSyncSkip": "Skip existing", "siteSyncOverwrite": "Overwrite existing", "siteSyncResume": "Resume last sync", + "siteSyncAbandon": "Abandon and clean up", + "siteSyncAbandonConfirm": "Abandon this sync and delete partial objects created by it that are not yet referenced by a media record?", + "siteSyncAbandoned": "Sync abandoned and partial objects cleaned up", "siteSyncScanProgress": "Media scan progress", "siteSyncFileProgress": "Synced file count", "siteSyncStarted": "Site-to-site sync started", diff --git a/i18n/locales/zh-CN.json b/i18n/locales/zh-CN.json index 4cb395b..977c717 100644 --- a/i18n/locales/zh-CN.json +++ b/i18n/locales/zh-CN.json @@ -665,6 +665,9 @@ "siteSyncSkip": "跳过已存在", "siteSyncOverwrite": "覆盖已存在", "siteSyncResume": "继续上次同步", + "siteSyncAbandon": "放弃并清理残留", + "siteSyncAbandonConfirm": "确定放弃这次同步吗？将删除本次同步新建但尚未归入媒体记录的残留对象。", + "siteSyncAbandoned": "同步已放弃，残留对象已清理", "siteSyncScanProgress": "媒体扫描进度", "siteSyncFileProgress": "已同步文件数量", "siteSyncStarted": "两站点同步任务已开始", diff --git a/server/api/admin/site-sync.post.ts b/server/api/admin/site-sync.post.ts index 669ddf7..ae5c114 100644 --- a/server/api/admin/site-sync.post.ts +++ b/server/api/admin/site-sync.post.ts @@ -5,6 +5,7 @@ import { resumeSiteSyncJob, startSiteSyncJob } from '../../utils/backup/runner' +import { abandonSiteSyncJob } from '../../utils/site-sync/abandon' interface SiteSyncBody { targetBaseUrl?: string @@ -12,6 +13,7 @@ interface SiteSyncBody { conflictMode?: 'skip' | 'overwrite' direction?: 'push' | 'pull' resumeJobId?: string + abandonJobId?: string } export default defineEventHandler(async (event) => { @@ -22,11 +24,15 @@ export default defineEventHandler(async (event) => { const conflictMode = body?.conflictMode === 'skip' ? 'skip' : 'overwrite' const direction = body?.direction === 'pull' ? 'pull' : 'push' - if (!token) { + if (!token && !body?.abandonJobId) { createApiError(event, 'INVALID_REQUEST', '请输入目标站点管理员 API Token', 400) } try { + if (body?.abandonJobId) { + const job = await abandonSiteSyncJob(body.abandonJobId, token, user.id) + return { job: serializeBackupJob(job) } + } let job if (body?.resumeJobId) { job = resumeSiteSyncJob(body.resumeJobId, token, user.id) @@ -54,6 +60,14 @@ export default defineEventHandler(async (event) => { if (error instanceof Error && error.message === 'SITE_SYNC_JOB_NOT_RESUMABLE') { createApiError(event, 'INVALID_REQUEST', '该同步任务无法继续', 400) } + if (body?.abandonJobId) { + createApiError( + event, + 'INVALID_REQUEST', + error instanceof Error ? error.message : '放弃同步失败', + 400 + ) + } throw error } }) diff --git a/server/api/site-sync/cleanup.post.ts b/server/api/site-sync/cleanup.post.ts new file mode 100644 index 0000000..10fd834 --- /dev/null +++ b/server/api/site-sync/cleanup.post.ts @@ -0,0 +1,20 @@ +import { requireSiteSyncAuth } from '../../utils/access' +import { createApiError } from '../../utils/api-error' +import { cleanupSiteSyncJobArtifacts } from '../../utils/site-sync/artifacts' + +interface Body { + jobId?: string +} + +export default defineEventHandler(async (event) => { + await requireSiteSyncAuth(event) + const body = await readBody(event).catch(() => null) + const jobId = body?.jobId?.trim() ?? '' + if (!jobId || jobId.length > 128) { + createApiError(event, 'INVALID_REQUEST', '缺少同步任务 ID', 400) + } + return { + success: true, + ...await cleanupSiteSyncJobArtifacts(jobId) + } +}) diff --git a/server/api/site-sync/complete.post.ts b/server/api/site-sync/complete.post.ts new file mode 100644 index 0000000..c2972ed --- /dev/null +++ b/server/api/site-sync/complete.post.ts @@ -0,0 +1,21 @@ +import { requireSiteSyncAuth } from '../../utils/access' +import { createApiError } from '../../utils/api-error' +import { validateStorageKey } from '../../utils/image-key' +import { completeSiteSyncMediaArtifacts } from '../../utils/site-sync/artifacts' + +interface Body { + jobId?: string + mediaKey?: string +} + +export default defineEventHandler(async (event) => { + await requireSiteSyncAuth(event) + const body = await readBody(event).catch(() => null) + const jobId = body?.jobId?.trim() ?? '' + const mediaKey = body?.mediaKey?.trim() ?? '' + if (!jobId || jobId.length > 128 || !validateStorageKey(mediaKey)) { + createApiError(event, 'INVALID_REQUEST', '同步完成标记参数无效', 400) + } + completeSiteSyncMediaArtifacts(jobId, mediaKey) + return { success: true } +}) diff --git a/server/api/site-sync/media.post.ts b/server/api/site-sync/media.post.ts index 6f5a115..16cf32b 100644 --- a/server/api/site-sync/media.post.ts +++ b/server/api/site-sync/media.post.ts @@ -2,11 +2,13 @@ import { requireSiteSyncAuth } from '../../utils/access' import { createApiError } from '../../utils/api-error' import { validateStorageKey } from '../../utils/image-key' import { applySiteSyncMedia } from '../../utils/site-sync/apply' +import { completeSiteSyncMediaArtifacts } from '../../utils/site-sync/artifacts' import type { SiteSyncMediaTransfer } from '../../utils/site-sync/protocol' interface Body { media?: SiteSyncMediaTransfer conflictMode?: 'skip' | 'overwrite' + jobId?: string } export default defineEventHandler(async (event) => { @@ -18,7 +20,11 @@ export default defineEventHandler(async (event) => { } const conflictMode = body?.conflictMode === 'skip' ? 'skip' : 'overwrite' try { - return await applySiteSyncMedia(media, conflictMode, admin.id) + const result = await applySiteSyncMedia(media, conflictMode, admin.id) + if (body?.jobId) { + completeSiteSyncMediaArtifacts(body.jobId, media.key) + } + return result } catch (error) { const message = error instanceof Error ? error.message : '媒体同步失败' createApiError(event, 'INVALID_REQUEST', message, 400) diff --git a/server/api/site-sync/object.post.ts b/server/api/site-sync/object.post.ts index f386d3d..9faa5e7 100644 --- a/server/api/site-sync/object.post.ts +++ b/server/api/site-sync/object.post.ts @@ -5,6 +5,7 @@ import { MAX_VIDEO_FILE_SIZE } from '../../utils/constants' import { findUserByUsername } from '../../utils/db' import { validateStorageKey } from '../../utils/image-key' import { getActiveBackend } from '../../utils/storage/resolver' +import { recordSiteSyncArtifact } from '../../utils/site-sync/artifacts' function field(form: Awaited>, name: string): string { const part = form?.find(item => item.name === name) @@ -30,6 +31,7 @@ export default defineEventHandler(async (event) => { if (conflictMode === 'skip' && await backend.head(key)) { return { success: true, skipped: true } } + const existedBefore = Boolean(await backend.head(key)) const ownerUsername = field(form, 'ownerUsername') const owner = ownerUsername ? findUserByUsername(ownerUsername) : null @@ -37,6 +39,18 @@ export default defineEventHandler(async (event) => { createApiError(event, 'INVALID_REQUEST', `目标站点缺少用户：${ownerUsername}`, 400) } + const jobId = field(form, 'jobId') + const mediaKey = field(form, 'mediaKey') + if (jobId && validateStorageKey(mediaKey)) { + recordSiteSyncArtifact({ + jobId, + mediaKey, + key, + backendId: backend.id, + created: !existedBefore + }) + } + await backend.put(key, new Uint8Array(file.data), { originalName: field(form, 'originalName') || file.filename || key.split('/').pop() || 'file', uploadedAt: field(form, 'uploadedAt') || new Date().toISOString(), diff --git a/server/plugins/startup.ts b/server/plugins/startup.ts index fe1f79b..72a99b7 100644 --- a/server/plugins/startup.ts +++ b/server/plugins/startup.ts @@ -5,6 +5,9 @@ import { join } from 'node:path' import { runAutoDeleteCleanup } from '../utils/auto-delete' import { migrateLocalImagesToIndex, purgeOrphanImageIndexRows, reconcileLegacyImageKeys, repairLocalImageIndexMetadata } from '../utils/image-index' import { ensureDefaultBackends } from '../utils/storage-backends' +import { cleanupLocalAtomicTempFiles } from '../utils/storage/local' +import { cleanupExpiredSiteSyncArtifacts } from '../utils/site-sync/artifacts' +import { recoverInterruptedBackupJobs } from '../utils/backup/jobs' import { cleanupTranscodedSources, killOrphanFfmpeg, @@ -26,6 +29,37 @@ export default defineNitroPlugin(() => { ensureDefaultBackends() + try { + const interrupted = recoverInterruptedBackupJobs() + if (interrupted > 0) { + logInfo('已将异常退出时未完成的任务标记为失败', { interrupted }) + } + } catch (error) { + logException('恢复异常退出的备份任务失败', error) + } + + void (async () => { + try { + const removed = await cleanupLocalAtomicTempFiles() + if (removed > 0) { + logInfo('清理过期的原子写入临时文件', { removed }) + } + } catch (error) { + logException('清理原子写入临时文件失败', error) + } + })() + + void (async () => { + try { + const result = await cleanupExpiredSiteSyncArtifacts() + if (result.jobs > 0) { + logInfo('清理过期站点同步残留', result) + } + } catch (error) { + logException('清理站点同步残留失败', error) + } + })() + /** 清理上次异常退出留下的转码临时目录 */ void (async () => { try { diff --git a/server/utils/backup/cancel-state.ts b/server/utils/backup/cancel-state.ts new file mode 100644 index 0000000..48cd602 --- /dev/null +++ b/server/utils/backup/cancel-state.ts @@ -0,0 +1,64 @@ +interface JobCancellationState { + requested: Set + controllers: Map +} + +const CANCELLED_CODE = 'JOB_CANCELLED' + +const state = (globalThis as typeof globalThis & { + __pichostJobCancellationState?: JobCancellationState +}).__pichostJobCancellationState ?? { + requested: new Set(), + controllers: new Map() +} +;(globalThis as typeof globalThis & { + __pichostJobCancellationState?: JobCancellationState +}).__pichostJobCancellationState = state + +export function createJobCancelledError(): Error { + const error = new Error('任务已取消') + error.name = CANCELLED_CODE + return error +} + +export function isJobCancelledError(error: unknown): boolean { + return error instanceof Error && error.name === CANCELLED_CODE +} + +export function beginJobCancellationScope(jobId: string): void { + state.requested.delete(jobId) + state.controllers.get(jobId)?.abort() + state.controllers.set(jobId, new AbortController()) +} + +export function endJobCancellationScope(jobId: string): void { + state.controllers.delete(jobId) + state.requested.delete(jobId) +} + +export function requestJobCancellation(jobId: string): void { + state.requested.add(jobId) + state.controllers.get(jobId)?.abort(createJobCancelledError()) +} + +export function isJobCancellationRequested(jobId: string): boolean { + return state.requested.has(jobId) +} + +export function getJobAbortSignal(jobId: string): AbortSignal { + return state.controllers.get(jobId)?.signal ?? new AbortController().signal +} + +export function getJobOperationSignal( + jobId: string, + timeoutMs: number +): AbortSignal { + return AbortSignal.any([ + getJobAbortSignal(jobId), + AbortSignal.timeout(timeoutMs) + ]) +} + +export function throwIfJobCancellationRequested(jobId: string): void { + if (isJobCancellationRequested(jobId)) throw createJobCancelledError() +} diff --git a/server/utils/backup/runner.ts b/server/utils/backup/runner.ts index 82573ae..f49d03c 100644 --- a/server/utils/backup/runner.ts +++ b/server/utils/backup/runner.ts @@ -10,6 +10,11 @@ import { recoverInterruptedBackupJobs, type BackupJob } from './jobs' +import { + beginJobCancellationScope, + endJobCancellationScope, + requestJobCancellation +} from './cancel-state' interface BackupRunnerState { running: boolean @@ -42,11 +47,8 @@ export function cancelBackupJob(jobId: string): boolean { if (!job || (job.status !== 'running' && job.status !== 'pending')) { return false } - markJobFailed(jobId, '任务已取消') - if (runnerState.currentJobId === jobId) { - runnerState.running = false - runnerState.currentJobId = null - } + requestJobCancellation(jobId) + markJobFailed(jobId, '任务已取消', '已请求取消，正在停止…') return true } @@ -59,14 +61,18 @@ export function scheduleBackupJob( } runnerState.running = true runnerState.currentJobId = job.id + beginJobCancellationScope(job.id) markJobRunning(job.id) setImmediate(() => { runner() .catch(() => {}) .finally(() => { - runnerState.running = false - runnerState.currentJobId = null + endJobCancellationScope(job.id) + if (runnerState.currentJobId === job.id) { + runnerState.running = false + runnerState.currentJobId = null + } }) }) } diff --git a/server/utils/site-sync/abandon.ts b/server/utils/site-sync/abandon.ts new file mode 100644 index 0000000..9a9ab38 --- /dev/null +++ b/server/utils/site-sync/abandon.ts @@ -0,0 +1,66 @@ +import { + getBackupJob, + updateBackupJob, + type BackupJob +} from '../backup/jobs' +import { cleanupSiteSyncJobArtifacts } from './artifacts' +import { fetchSiteSyncJson } from './runner' + +interface CleanupResult { + deleted: number + referenced: number + failed: number +} + +export async function abandonSiteSyncJob( + jobId: string, + token: string, + userId: number +): Promise { + const job = getBackupJob(jobId, userId) + if (!job || job.type !== 'site_sync') throw new Error('同步任务不存在') + if (job.status !== 'failed') throw new Error('只能放弃已停止的同步任务') + + const direction = job.payload.direction === 'pull' ? 'pull' : 'push' + const targetBaseUrl = typeof job.payload.targetBaseUrl === 'string' + ? job.payload.targetBaseUrl.replace(/\/+$/, '') + : '' + let cleanup: CleanupResult + if (direction === 'push') { + if (!targetBaseUrl || !token) { + throw new Error('放弃推送任务并清理目标站残留时，需要目标站 Token') + } + cleanup = await fetchSiteSyncJson( + `${targetBaseUrl}/api/site-sync/cleanup`, + token, + { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ jobId }) + } + ) + } else { + cleanup = await cleanupSiteSyncJobArtifacts(jobId) + } + + if (cleanup.failed > 0) { + throw new Error(`同步残留清理失败：${cleanup.failed} 个对象`) + } + + const message = `已放弃同步 · 清理 ${cleanup.deleted} 个残留对象` + updateBackupJob(jobId, { + status: 'failed', + message, + error: '已放弃同步', + finishedAt: new Date().toISOString(), + payload: { + ...job.payload, + resumable: false, + abandoned: true, + cancelled: false, + failedKeys: [], + cleanup + } + }) + return getBackupJob(jobId, userId)! +} diff --git a/server/utils/site-sync/artifacts.ts b/server/utils/site-sync/artifacts.ts new file mode 100644 index 0000000..683aad4 --- /dev/null +++ b/server/utils/site-sync/artifacts.ts @@ -0,0 +1,172 @@ +import { getDb } from '../db' +import { getImageIndexRow } from '../image-index' +import { getBackendById } from '../storage/resolver' +import { hlsSegmentNames } from '../transcode' + +interface ArtifactRow { + job_id: string + media_key: string + key: string + backend_id: string + created: number + updated_at: string +} + +function ensureArtifactSchema(): void { + getDb().exec(` + CREATE TABLE IF NOT EXISTS site_sync_artifacts ( + job_id TEXT NOT NULL, + media_key TEXT NOT NULL, + key TEXT NOT NULL, + backend_id TEXT NOT NULL, + created INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL, + PRIMARY KEY(job_id, key) + ); + CREATE INDEX IF NOT EXISTS idx_site_sync_artifacts_updated + ON site_sync_artifacts(updated_at); + CREATE INDEX IF NOT EXISTS idx_site_sync_artifacts_media + ON site_sync_artifacts(job_id, media_key); + `) +} + +export function recordSiteSyncArtifact(input: { + jobId: string + mediaKey: string + key: string + backendId: string + created: boolean +}): void { + ensureArtifactSchema() + getDb().prepare(` + INSERT INTO site_sync_artifacts + (job_id, media_key, key, backend_id, created, updated_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(job_id, key) DO UPDATE SET + media_key = excluded.media_key, + backend_id = excluded.backend_id, + created = MAX(site_sync_artifacts.created, excluded.created), + updated_at = excluded.updated_at + `).run( + input.jobId, + input.mediaKey, + input.key, + input.backendId, + input.created ? 1 : 0, + new Date().toISOString() + ) +} + +export function completeSiteSyncMediaArtifacts( + jobId: string, + mediaKey: string +): void { + ensureArtifactSchema() + getDb().prepare(` + DELETE FROM site_sync_artifacts + WHERE job_id = ? AND media_key = ? + `).run(jobId, mediaKey) +} + +async function streamToString(stream: NodeJS.ReadableStream): Promise { + const chunks: Buffer[] = [] + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)) + } + return Buffer.concat(chunks).toString('utf8') +} + +async function isReferencedByCommittedMedia( + mediaKey: string, + key: string +): Promise { + const row = getImageIndexRow(mediaKey) + if (!row) return false + if (key === row.key || key === row.thumb_key || key === row.hls_key) { + return true + } + if (!row.hls_key) return false + + const slash = row.hls_key.lastIndexOf('/') + if (slash < 0 || !key.startsWith(`${row.hls_key.slice(0, slash)}/`)) { + return false + } + + try { + const backend = await getBackendById(row.backend_id) + if (!backend) return true + const playlist = await streamToString(await backend.createStream(row.hls_key)) + const dir = row.hls_key.slice(0, slash) + return hlsSegmentNames(playlist).some(name => key === `${dir}/${name}`) + } catch { + // If the playlist cannot be read, keep the object rather than risk data loss. + return true + } +} + +export async function cleanupSiteSyncJobArtifacts(jobId: string): Promise<{ + deleted: number + referenced: number + failed: number +}> { + ensureArtifactSchema() + const rows = getDb().prepare(` + SELECT job_id, media_key, key, backend_id, created, updated_at + FROM site_sync_artifacts + WHERE job_id = ? + ORDER BY updated_at ASC + `).all(jobId) as unknown as ArtifactRow[] + + let deleted = 0 + let referenced = 0 + let failed = 0 + for (const row of rows) { + if (Number(row.created) !== 1) { + getDb().prepare(` + DELETE FROM site_sync_artifacts WHERE job_id = ? AND key = ? + `).run(jobId, row.key) + continue + } + + try { + if (await isReferencedByCommittedMedia(row.media_key, row.key)) { + referenced++ + } else { + const backend = await getBackendById(row.backend_id) + if (!backend) throw new Error('存储后端不存在') + if (await backend.head(row.key)) { + await backend.delete(row.key) + deleted++ + } + } + getDb().prepare(` + DELETE FROM site_sync_artifacts WHERE job_id = ? AND key = ? + `).run(jobId, row.key) + } catch { + failed++ + } + } + + return { deleted, referenced, failed } +} + +export async function cleanupExpiredSiteSyncArtifacts( + maxAgeMs = 24 * 60 * 60 * 1000 +): Promise<{ jobs: number, deleted: number, failed: number }> { + ensureArtifactSchema() + const cutoff = new Date(Date.now() - maxAgeMs).toISOString() + const rows = getDb().prepare(` + SELECT DISTINCT job_id + FROM site_sync_artifacts + WHERE updated_at < ? + `).all(cutoff) as Array<{ job_id: string }> + + let deleted = 0 + let failed = 0 + for (const row of rows) { + const result = await cleanupSiteSyncJobArtifacts(row.job_id) + deleted += result.deleted + failed += result.failed + } + return { jobs: rows.length, deleted, failed } +} diff --git a/server/utils/site-sync/runner.test.ts b/server/utils/site-sync/runner.test.ts index c7d59af..a7a2f21 100644 --- a/server/utils/site-sync/runner.test.ts +++ b/server/utils/site-sync/runner.test.ts @@ -1,5 +1,11 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { fetchSiteSyncJson, retrySiteSyncOperation } from './runner' +import { + beginJobCancellationScope, + endJobCancellationScope, + getJobAbortSignal, + requestJobCancellation +} from '../backup/cancel-state' afterEach(() => { vi.unstubAllGlobals() @@ -52,4 +58,17 @@ describe('fetchSiteSyncJson', () => { await expect(retrySiteSyncOperation(operation)).rejects.toThrow('database is busy') expect(operation).toHaveBeenCalledTimes(3) }) + + it('stops retrying immediately when the job is cancelled', async () => { + const jobId = 'cancel-test' + beginJobCancellationScope(jobId) + const operation = vi.fn().mockRejectedValue(new Error('connection reset')) + const result = retrySiteSyncOperation(operation, getJobAbortSignal(jobId)) + + setTimeout(() => requestJobCancellation(jobId), 25) + + await expect(result).rejects.toThrow('任务已取消') + expect(operation).toHaveBeenCalledTimes(1) + endJobCancellationScope(jobId) + }) }) diff --git a/server/utils/site-sync/runner.ts b/server/utils/site-sync/runner.ts index 407494f..4c90c41 100644 --- a/server/utils/site-sync/runner.ts +++ b/server/utils/site-sync/runner.ts @@ -1,4 +1,5 @@ import { basename } from 'node:path' +import type { Readable } from 'node:stream' import { contentTypeFromKey } from '../content-type' import { getImageIndexRow, @@ -22,7 +23,19 @@ import { markJobFailed, updateBackupJob } from '../backup/jobs' +import { + createJobCancelledError, + getJobAbortSignal, + getJobOperationSignal, + isJobCancellationRequested, + isJobCancelledError, + throwIfJobCancellationRequested +} from '../backup/cancel-state' import { applySiteSyncMedia } from './apply' +import { + completeSiteSyncMediaArtifacts, + recordSiteSyncArtifact +} from './artifacts' import { toSiteSyncMediaTransferWithTags } from './manifest' import type { SiteSyncDirection, @@ -49,9 +62,19 @@ export function registerSiteSyncToken(jobId: string, token: string): void { tokens.set(jobId, token) } -async function streamToBuffer(stream: NodeJS.ReadableStream): Promise { +const SITE_SYNC_JSON_TIMEOUT_MS = 2 * 60 * 1000 +const SITE_SYNC_OBJECT_TIMEOUT_MS = 15 * 60 * 1000 + +async function streamToBuffer( + stream: Readable, + signal?: AbortSignal +): Promise { const chunks: Buffer[] = [] for await (const chunk of stream) { + if (signal?.aborted) { + stream.destroy(createJobCancelledError()) + throw createJobCancelledError() + } chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)) } return Buffer.concat(chunks) @@ -71,21 +94,46 @@ async function responseError(response: Response): Promise { const SITE_SYNC_MAX_ATTEMPTS = 3 +function normalizeSiteSyncError(error: Error): Error { + if (error.name === 'TimeoutError') { + return new Error('目标站点连接超时或无响应') + } + if (error instanceof TypeError && /fetch failed|terminated/i.test(error.message)) { + return new Error('目标站点无法连接') + } + return error +} + export async function retrySiteSyncOperation( - operation: () => Promise + operation: () => Promise, + signal?: AbortSignal ): Promise { + if (signal?.aborted) throw createJobCancelledError() let lastError: Error | null = null for (let attempt = 1; attempt <= SITE_SYNC_MAX_ATTEMPTS; attempt++) { try { return await operation() } catch (error) { + if (signal?.aborted) throw createJobCancelledError() lastError = error instanceof Error ? error : new Error('两站点同步失败') if (attempt < SITE_SYNC_MAX_ATTEMPTS) { - await new Promise(resolve => setTimeout(resolve, attempt * 500)) + await new Promise((resolve, reject) => { + const onAbort = () => { + clearTimeout(timer) + reject(createJobCancelledError()) + } + const timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort) + resolve() + }, attempt * 500) + signal?.addEventListener('abort', onAbort, { once: true }) + }) } } } - throw lastError ?? new Error('两站点同步失败') + throw lastError + ? normalizeSiteSyncError(lastError) + : new Error('两站点同步失败') } async function fetchSiteSyncJsonOnce( @@ -107,14 +155,25 @@ async function fetchSiteSyncJsonOnce( export async function fetchSiteSyncJson( url: string, token: string, - init?: RequestInit + init?: RequestInit, + signal?: AbortSignal ): Promise { - return retrySiteSyncOperation(() => ( - fetchSiteSyncJsonOnce(url, token, init) - )) + return retrySiteSyncOperation( + () => fetchSiteSyncJsonOnce(url, token, { + ...init, + signal: signal + ? AbortSignal.any([ + signal, + AbortSignal.timeout(SITE_SYNC_JSON_TIMEOUT_MS) + ]) + : AbortSignal.timeout(SITE_SYNC_JSON_TIMEOUT_MS) + }), + signal + ) } async function uploadObject( + jobId: string, options: SiteSyncOptions, token: string, backend: StorageBackend, @@ -124,9 +183,15 @@ async function uploadObject( ): Promise { const stored = await backend.head(key) if (!stored) throw new Error(`源文件不存在：${key}`) - const bytes = await streamToBuffer(await backend.createStream(key)) + const bytes = await streamToBuffer( + await backend.createStream(key), + getJobAbortSignal(jobId) + ) + throwIfJobCancellationRequested(jobId) const form = new FormData() const isPrimary = key === row.key + form.append('jobId', jobId) + form.append('mediaKey', row.key) form.append('key', key) form.append('conflictMode', options.conflictMode) form.append('ownerUsername', ownerUsername ?? '') @@ -143,11 +208,16 @@ async function uploadObject( await fetchSiteSyncJsonOnce<{ success: boolean }>( targetUrl(options.targetBaseUrl, '/api/site-sync/object'), token, - { method: 'POST', body: form } + { + method: 'POST', + body: form, + signal: getJobOperationSignal(jobId, SITE_SYNC_OBJECT_TIMEOUT_MS) + } ) } async function downloadSiteSyncObject( + jobId: string, options: SiteSyncOptions, token: string, key: string @@ -160,7 +230,8 @@ async function downloadSiteSyncObject( { headers: { 'Auth-Token': token - } + }, + signal: getJobOperationSignal(jobId, SITE_SYNC_OBJECT_TIMEOUT_MS) } ) if (!response.ok) throw await responseError(response) @@ -173,6 +244,7 @@ async function runSiteSyncPull( token: string ): Promise { const job = getBackupJob(jobId) + const jobSignal = getJobAbortSignal(jobId) const checkpoint = job?.payload ?? {} let lastMediaKey = typeof checkpoint.lastMediaKey === 'string' ? checkpoint.lastMediaKey @@ -189,7 +261,7 @@ async function runSiteSyncPull( : [] ) - const saveCheckpoint = (message: string) => { + const saveCheckpoint = (message: string, cancelled = false) => { updateBackupJob(jobId, { progress: scanned, total: totalMedia, @@ -198,6 +270,7 @@ async function runSiteSyncPull( targetBaseUrl: options.targetBaseUrl, conflictMode: options.conflictMode, direction: 'pull', + cancelled, lastMediaKey, scanned, added, @@ -215,11 +288,19 @@ async function runSiteSyncPull( `扫描 ${scanned}/${totalMedia || scanned} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size}${suffix}` ) const failAndCheckpoint = (error: unknown) => { - const message = progressMessage(' · 已保存断点') - saveCheckpoint(message) + const cancelled = isJobCancellationRequested(jobId) + || isJobCancelledError(error) + const message = cancelled + ? progressMessage(' · 已取消，已保存断点') + : progressMessage(' · 已保存断点') + saveCheckpoint(message, cancelled) markJobFailed( jobId, - error instanceof Error ? error.message : '从目标站点拉取失败', + cancelled + ? '任务已取消' + : error instanceof Error + ? error.message + : '从目标站点拉取失败', message ) } @@ -228,21 +309,22 @@ async function runSiteSyncPull( const targetUsers = new Set( (await fetchSiteSyncJson( targetUrl(options.targetBaseUrl, '/api/site-sync/users'), - token + token, + undefined, + jobSignal )).users ) - const localUserMap = listUserIdUsernameMap() - const localUsers = new Set(localUserMap.values()) - const missingUsers = [...targetUsers].filter(name => !localUsers.has(name)) - if (missingUsers.length) { - throw new Error(`当前站点缺少同名用户：${missingUsers.join('、')}`) - } - const backend = await retrySiteSyncOperation(() => getActiveBackend()) + const backend = await retrySiteSyncOperation( + () => getActiveBackend(), + jobSignal + ) const actorUserId = job?.userId ?? getPrimaryAdminUserId() ?? 0 const readMediaPage = (path: string) => fetchSiteSyncJson( targetUrl(options.targetBaseUrl, path), - token + token, + undefined, + jobSignal ) const processBundleOnce = async ( @@ -250,11 +332,16 @@ async function runSiteSyncPull( isRetry: boolean ): Promise<'added' | 'skipped'> => { const { media, assets } = bundle + throwIfJobCancellationRequested(jobId) + if (media.ownerUsername && !targetUsers.has(media.ownerUsername)) { + throw new Error(`当前站点缺少同名用户：${media.ownerUsername}`) + } if ( !isRetry && options.conflictMode === 'skip' && getImageIndexRow(media.key) ) { + completeSiteSyncMediaArtifacts(jobId, media.key) return 'skipped' } @@ -262,23 +349,31 @@ async function runSiteSyncPull( ? findUserByUsername(media.ownerUsername) : null for (const asset of assets) { - if ( - options.conflictMode === 'skip' - && await backend.head(asset.key) - ) { + throwIfJobCancellationRequested(jobId) + const existedBefore = Boolean(await backend.head(asset.key)) + if (options.conflictMode === 'skip' && existedBefore) { continue } const bytes = await downloadSiteSyncObject( + jobId, options, token, asset.key ) + throwIfJobCancellationRequested(jobId) if (bytes.length !== asset.size) { throw new Error( `同步对象大小不一致：${asset.key}，预期 ${asset.size}，实际 ${bytes.length}` ) } const isPrimary = asset.key === media.key + recordSiteSyncArtifact({ + jobId, + mediaKey: media.key, + key: asset.key, + backendId: backend.id, + created: !existedBefore + }) await backend.put(asset.key, new Uint8Array(bytes), { originalName: isPrimary ? media.originalName : basename(asset.key), uploadedAt: media.uploadedAt, @@ -295,13 +390,18 @@ async function runSiteSyncPull( ? 'file' : 'image' }) + throwIfJobCancellationRequested(jobId) } await applySiteSyncMedia(media, 'overwrite', actorUserId) + completeSiteSyncMediaArtifacts(jobId, media.key) return 'added' } const processBundle = (bundle: SiteSyncMediaBundle, isRetry: boolean) => ( - retrySiteSyncOperation(() => processBundleOnce(bundle, isRetry)) + retrySiteSyncOperation( + () => processBundleOnce(bundle, isRetry), + jobSignal + ) ) const recordOutcome = (outcome: 'added' | 'skipped', fileCount: number) => { if (outcome === 'added') added += 1 @@ -382,6 +482,7 @@ async function runSiteSyncPull( return } + throwIfJobCancellationRequested(jobId) totalFiles = filesSynced markJobDone(jobId, { message: `扫描 ${scanned}/${totalMedia || scanned} · 新增 ${added} · 跳过 ${skipped} · 失败 0`, @@ -412,6 +513,7 @@ export async function runSiteSyncJob( ): Promise { const token = tokens.get(jobId) if (!token) throw new Error('同步 Token 不可用，请重新发起任务') + const jobSignal = getJobAbortSignal(jobId) try { if (options.direction === 'pull') { @@ -421,7 +523,9 @@ export async function runSiteSyncJob( const inventory = await fetchSiteSyncJson( targetUrl(options.targetBaseUrl, '/api/site-sync/inventory'), - token + token, + undefined, + jobSignal ) const targetUsers = new Set(inventory.users) const targetMedia = new Set(inventory.mediaKeys) @@ -466,6 +570,7 @@ export async function runSiteSyncJob( }) for (const row of rows) { + throwIfJobCancellationRequested(jobId) const backend = await getBackendForKey(row.key) let assetKeys: string[] try { @@ -487,7 +592,7 @@ export async function runSiteSyncJob( }) } - const saveCheckpoint = (message: string) => { + const saveCheckpoint = (message: string, cancelled = false) => { updateBackupJob(jobId, { progress: scanned, total: rows.length, @@ -496,6 +601,7 @@ export async function runSiteSyncJob( targetBaseUrl: options.targetBaseUrl, conflictMode: options.conflictMode, direction: 'push', + cancelled, lastMediaKey, scanned, added, @@ -520,11 +626,22 @@ export async function runSiteSyncJob( let lastCheckpointAt = 0 const processPlanOnce = async (plan: typeof plans[number], isRetry: boolean) => { const { row, ownerUsername, backend, keys } = plan + throwIfJobCancellationRequested(jobId) if ( !isRetry && options.conflictMode === 'skip' && targetMedia.has(row.key) ) { + await fetchSiteSyncJsonOnce<{ success: boolean }>( + targetUrl(options.targetBaseUrl, '/api/site-sync/complete'), + token, + { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ jobId, mediaKey: row.key }), + signal: getJobOperationSignal(jobId, SITE_SYNC_JSON_TIMEOUT_MS) + } + ) skipped += 1 filesSynced = Math.min(totalFiles, filesSynced + keys.length) failedKeys.delete(row.key) @@ -532,6 +649,7 @@ export async function runSiteSyncJob( } for (const key of keys) { + throwIfJobCancellationRequested(jobId) const stored = await backend.head(key) if (!stored) throw new Error(`源文件不存在：${key}`) if ( @@ -540,9 +658,18 @@ export async function runSiteSyncJob( ) { continue } - await uploadObject(options, token, backend, row, ownerUsername, key) + await uploadObject( + jobId, + options, + token, + backend, + row, + ownerUsername, + key + ) } + throwIfJobCancellationRequested(jobId) await fetchSiteSyncJsonOnce<{ success: boolean }>( targetUrl(options.targetBaseUrl, '/api/site-sync/media'), token, @@ -550,9 +677,11 @@ export async function runSiteSyncJob( method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ + jobId, media: toSiteSyncMediaTransferWithTags(row, ownerUsername), conflictMode: options.conflictMode - }) + }), + signal: getJobOperationSignal(jobId, SITE_SYNC_JSON_TIMEOUT_MS) } ) @@ -563,18 +692,29 @@ export async function runSiteSyncJob( const processPlan = ( plan: typeof plans[number], isRetry: boolean - ) => retrySiteSyncOperation(() => processPlanOnce(plan, isRetry)) + ) => retrySiteSyncOperation( + () => processPlanOnce(plan, isRetry), + jobSignal + ) const retryPlans = plans.filter(plan => failedKeys.has(plan.row.key)) for (const plan of retryPlans) { try { await processPlan(plan, true) lastMediaKey = plan.row.key - } catch { + } catch (error) { failedKeys.add(plan.row.key) - const message = `扫描 ${scanned}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size} · 已保存断点` - saveCheckpoint(message) - markJobFailed(jobId, '连续 3 次重试失败，已保存断点', message) + const cancelled = isJobCancellationRequested(jobId) + || isJobCancelledError(error) + const message = `扫描 ${scanned}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size} · ${cancelled ? '已取消，' : ''}已保存断点` + saveCheckpoint(message, cancelled) + markJobFailed( + jobId, + cancelled + ? '任务已取消' + : '连续 3 次重试失败，已保存断点', + message + ) return } } @@ -584,12 +724,20 @@ export async function runSiteSyncJob( scanned += 1 try { await processPlan(plan, false) - } catch { + } catch (error) { failedKeys.add(plan.row.key) lastMediaKey = plan.row.key - const message = `扫描 ${scanned}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size} · 已保存断点` - saveCheckpoint(message) - markJobFailed(jobId, '连续 3 次重试失败，已保存断点', message) + const cancelled = isJobCancellationRequested(jobId) + || isJobCancelledError(error) + const message = `扫描 ${scanned}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size} · ${cancelled ? '已取消，' : ''}已保存断点` + saveCheckpoint(message, cancelled) + markJobFailed( + jobId, + cancelled + ? '任务已取消' + : '连续 3 次重试失败，已保存断点', + message + ) return } lastMediaKey = plan.row.key @@ -614,6 +762,7 @@ export async function runSiteSyncJob( return } + throwIfJobCancellationRequested(jobId) markJobDone(jobId, { message: `扫描 ${rows.length}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 0`, payload: { @@ -632,6 +781,10 @@ export async function runSiteSyncJob( } }) } catch (error) { + if (isJobCancellationRequested(jobId) || isJobCancelledError(error)) { + markJobFailed(jobId, '任务已取消', '任务已取消') + return + } const message = error instanceof Error ? error.message : '两站点同步失败' markJobFailed(jobId, message) throw error diff --git a/server/utils/storage/index.ts b/server/utils/storage/index.ts index 5b13899..7ec7360 100644 --- a/server/utils/storage/index.ts +++ b/server/utils/storage/index.ts @@ -1,4 +1,4 @@ -export { getActiveBackend, getBackendForKey, getActiveBackendRow, getBackendRowForKey } from './resolver' +export { getActiveBackend, getBackendById, getBackendForKey, getActiveBackendRow, getBackendRowForKey } from './resolver' export { LocalStorageBackend } from './local' export { S3StorageBackend, buildObjectKey, parseS3Config, validateS3Config } from './s3' export { diff --git a/server/utils/storage/local.ts b/server/utils/storage/local.ts index e92fbe8..d3dec66 100644 --- a/server/utils/storage/local.ts +++ b/server/utils/storage/local.ts @@ -1,3 +1,4 @@ +import { randomUUID } from 'node:crypto' import { createReadStream, promises as fs } from 'node:fs' import type { ReadStream } from 'node:fs' import { dirname, join } from 'node:path' @@ -7,6 +8,7 @@ import { validateStorageKey } from '../image-key' import type { StoredImage, StoredImageMeta, StorageBackend } from './types' const META_SUFFIX = '.meta.json' +const ATOMIC_TEMP_DIR = '.pichost-tmp' function keyToFilePath(key: string): string { return join(getDataDir(), ...key.split('/')) @@ -39,8 +41,36 @@ export class LocalStorageBackend implements StorageBackend { async put(key: string, bytes: Uint8Array, _meta: StoredImageMeta): Promise { const filePath = keyToFilePath(key) + const tempDir = join(getDataDir(), ATOMIC_TEMP_DIR) + const tempPath = join(tempDir, `${randomUUID()}.part`) + await fs.mkdir(tempDir, { recursive: true }) await fs.mkdir(dirname(filePath), { recursive: true }) - await fs.writeFile(filePath, bytes) + + const handle = await fs.open(tempPath, 'w') + try { + await handle.writeFile(bytes) + await handle.sync() + } finally { + await handle.close() + } + + try { + await fs.rename(tempPath, filePath) + } catch (error) { + await fs.rm(tempPath, { force: true }).catch(() => {}) + throw error + } + + try { + const directory = await fs.open(dirname(filePath), 'r') + try { + await directory.sync() + } finally { + await directory.close() + } + } catch { + // The rename already completed; directory fsync is best-effort on macOS. + } } async head(key: string): Promise { @@ -93,3 +123,31 @@ export class LocalStorageBackend implements StorageBackend { return { count: 0, bytes: 0 } } } + +export async function cleanupLocalAtomicTempFiles( + maxAgeMs = 60 * 60 * 1000 +): Promise { + const tempDir = join(getDataDir(), ATOMIC_TEMP_DIR) + let entries + try { + entries = await fs.readdir(tempDir, { withFileTypes: true }) + } catch { + return 0 + } + + const cutoff = Date.now() - maxAgeMs + let removed = 0 + for (const entry of entries) { + if (!entry.isFile() || !entry.name.endsWith('.part')) continue + const path = join(tempDir, entry.name) + try { + const stat = await fs.stat(path) + if (stat.mtimeMs > cutoff) continue + await fs.rm(path, { force: true }) + removed++ + } catch { + // A later startup will retry stale temporary files. + } + } + return removed +} diff --git a/server/utils/storage/resolver.ts b/server/utils/storage/resolver.ts index 6859977..20cf622 100644 --- a/server/utils/storage/resolver.ts +++ b/server/utils/storage/resolver.ts @@ -53,6 +53,18 @@ export async function getBackendForKey(key: string): Promise { } } +export async function getBackendById( + backendId: string +): Promise { + const fromEnv = createBackendFromEnv() + if (fromEnv) return fromEnv.id === backendId ? fromEnv : null + + ensureDefaultBackends() + const row = getStorageBackendRow(backendId) + if (!row) return null + return createBackendInstance(row) +} + export function getActiveBackendRow(): StorageBackendRow | null { const fromEnv = createBackendFromEnv() if (fromEnv) {