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) {