From 3e3719f33cb9d68dd178a416e7f145491eb5bb93 Mon Sep 17 00:00:00 2001 From: Nico <47644445+nicotsx@users.noreply.github.com> Date: Tue, 30 Jun 2026 13:03:12 +0200 Subject: [PATCH] fix(bacups): reschedule immediately warning backups (#1016) * fix(bacups): reschedule immediately warning backups Backups that didn't end properly due to a container restart should be re-scheduled immediately * chore: pr feedback --- app/server/modules/backups/backups.service.ts | 78 +++++++ .../lifecycle/__tests__/startup.test.ts | 200 +++++++++++++++++- app/server/modules/lifecycle/startup.ts | 22 +- app/server/modules/tasks/tasks.store.ts | 2 + 4 files changed, 283 insertions(+), 19 deletions(-) diff --git a/app/server/modules/backups/backups.service.ts b/app/server/modules/backups/backups.service.ts index 79d4fbf2..e9181e3b 100644 --- a/app/server/modules/backups/backups.service.ts +++ b/app/server/modules/backups/backups.service.ts @@ -32,8 +32,10 @@ import { runEffectPromise, toMessage } from "../../utils/errors"; import { Effect } from "effect"; import { taskStore } from "../tasks/tasks.store"; import { createTaskProgressBuffer } from "../tasks/progress-buffer"; +import type { ParsedTask } from "../tasks/tasks.schemas"; const BACKUP_TASK_RESOURCE_TYPE = "backup_schedule"; +const RESTART_BACKUP_ERROR = "Zerobyte was restarted during the last scheduled backup"; const tryCancelTask = ( taskId: string, @@ -564,6 +566,81 @@ const getSchedulesToExecute = async () => { return scheduleQueries.findExecutable(organizationId); }; +const getInterruptedBackupScheduleIds = (staleTasks: ParsedTask[]) => { + const backupTaskScheduleIds = new Set(); + const retryableScheduledTaskScheduleIds = new Set(); + + for (const task of staleTasks) { + if (task.kind !== "backup" || task.resourceType !== BACKUP_TASK_RESOURCE_TYPE) { + continue; + } + + const scheduleId = Number(task.resourceId); + if (!Number.isInteger(scheduleId)) { + continue; + } + + backupTaskScheduleIds.add(scheduleId); + if (task.input.kind === "backup" && !task.input.manual && !task.cancellationRequested) { + retryableScheduledTaskScheduleIds.add(scheduleId); + } + } + + return { backupTaskScheduleIds, retryableScheduledTaskScheduleIds }; +}; + +const recoverInterruptedBackups = async (staleTasks: ParsedTask[]) => { + const { backupTaskScheduleIds, retryableScheduledTaskScheduleIds } = getInterruptedBackupScheduleIds(staleTasks); + const now = Date.now(); + + db.transaction((tx) => { + const recoveredSchedules = tx + .update(backupSchedulesTable) + .set({ + lastBackupStatus: "warning", + lastBackupError: RESTART_BACKUP_ERROR, + updatedAt: now, + }) + .where(eq(backupSchedulesTable.lastBackupStatus, "in_progress")) + .returning() + .all(); + + const recoveredScheduleIds = new Set(); + const retryScheduleIds = new Set(); + + for (const schedule of recoveredSchedules) { + recoveredScheduleIds.add(schedule.id); + + if (schedule.enabled && schedule.cronExpression && !backupTaskScheduleIds.has(schedule.id)) { + retryScheduleIds.add(schedule.id); + } + } + + for (const scheduleId of retryableScheduledTaskScheduleIds) { + if (recoveredScheduleIds.has(scheduleId)) { + retryScheduleIds.add(scheduleId); + } + } + + const retryScheduleIdList = [...retryScheduleIds]; + + if (retryScheduleIdList.length > 0) { + tx.update(backupSchedulesTable) + .set({ + nextBackupAt: null, + updatedAt: now, + }) + .where( + and( + inArray(backupSchedulesTable.id, retryScheduleIdList), + eq(backupSchedulesTable.lastBackupStatus, "warning"), + ), + ) + .run(); + } + }); +}; + const stopBackup = async (scheduleId: number) => { const organizationId = getOrganizationId(); const schedule = await scheduleQueries.findById(scheduleId, organizationId); @@ -694,6 +771,7 @@ export const backupsService = { validateBackupExecution, executeBackup, getSchedulesToExecute, + recoverInterruptedBackups, stopBackup, runForget, copyToMirrors, diff --git a/app/server/modules/lifecycle/__tests__/startup.test.ts b/app/server/modules/lifecycle/__tests__/startup.test.ts index ee0a73f4..4b6d98bd 100644 --- a/app/server/modules/lifecycle/__tests__/startup.test.ts +++ b/app/server/modules/lifecycle/__tests__/startup.test.ts @@ -8,7 +8,8 @@ import { repositoriesService } from "~/server/modules/repositories/repositories. import { notificationsService } from "~/server/modules/notifications/notifications.service"; import { volumeService } from "~/server/modules/volumes/volume.service"; import * as provisioningModule from "~/server/modules/provisioning/provisioning"; -import { taskStore } from "~/server/modules/tasks/tasks.store"; +import { RESTART_TASK_ERROR, taskStore } from "~/server/modules/tasks/tasks.store"; +import { withContext } from "~/server/core/request-context"; import { createTestBackupSchedule } from "~/test/helpers/backup"; import { createTestRepository } from "~/test/helpers/repository"; import { createTestVolume } from "~/test/helpers/volume"; @@ -41,15 +42,17 @@ afterEach(() => { vi.restoreAllMocks(); }); -test("marks active tasks stale and keeps stuck backup schedule recovery silent", async () => { +test("marks active scheduled backup tasks stale and makes them executable again", async () => { const emitSpy = vi.spyOn(serverEvents, "emit"); const notificationSpy = vi.spyOn(notificationsService, "sendBackupNotification").mockResolvedValue(); const volume = await createTestVolume(); const repository = await createTestRepository(); + const nextBackupAt = Date.now() + 24 * 60 * 60 * 1000; const schedule = await createTestBackupSchedule({ volumeId: volume.id, repositoryId: repository.id, lastBackupStatus: "in_progress", + nextBackupAt, }); const task = taskStore.create({ id: "task-startup-active", @@ -73,10 +76,201 @@ test("marks active tasks stale and keeps stuck backup schedule recovery silent", const updatedTask = await db.query.tasksTable.findFirst({ where: { id: task.id } }); const updatedSchedule = await db.query.backupSchedulesTable.findFirst({ where: { id: schedule.id } }); expect(updatedTask?.status).toBe("stale"); - expect(updatedTask?.error).toBe("Zerobyte was restarted before this task completed"); + expect(updatedTask?.error).toBe(RESTART_TASK_ERROR); expect(updatedTask?.finishedAt).toEqual(expect.any(Number)); expect(updatedSchedule?.lastBackupStatus).toBe("warning"); expect(updatedSchedule?.lastBackupError).toBe("Zerobyte was restarted during the last scheduled backup"); + expect(updatedSchedule?.nextBackupAt).toBeNull(); + await withContext({ organizationId: TEST_ORG_ID }, async () => { + expect(await backupsService.getSchedulesToExecute()).toContain(schedule.id); + }); expect(emitSpy).not.toHaveBeenCalledWith("backup:completed", expect.anything()); expect(notificationSpy).not.toHaveBeenCalled(); }); + +test("marks active manual backup tasks stale without making the schedule executable", async () => { + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const nextBackupAt = Date.now() + 24 * 60 * 60 * 1000; + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + lastBackupStatus: "in_progress", + nextBackupAt, + }); + const task = taskStore.create({ + id: "task-startup-manual", + organizationId: TEST_ORG_ID, + resourceType: "backup_schedule", + resourceId: String(schedule.id), + targetAgentId: "local", + input: { + kind: "backup", + scheduleId: schedule.id, + scheduleShortId: schedule.shortId, + manual: true, + }, + }); + taskStore.markRunning(task.id); + + const { startup } = await loadStartupModule(); + + await startup(); + + const updatedSchedule = await db.query.backupSchedulesTable.findFirst({ where: { id: schedule.id } }); + expect(updatedSchedule?.lastBackupStatus).toBe("warning"); + expect(updatedSchedule?.lastBackupError).toBe("Zerobyte was restarted during the last scheduled backup"); + expect(updatedSchedule?.nextBackupAt).toBe(nextBackupAt); +}); + +test("does not immediately retry cancellation-requested scheduled backups", async () => { + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const nextBackupAt = Date.now() + 24 * 60 * 60 * 1000; + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + lastBackupStatus: "in_progress", + nextBackupAt, + }); + const task = taskStore.create({ + id: "task-startup-cancelling", + organizationId: TEST_ORG_ID, + resourceType: "backup_schedule", + resourceId: String(schedule.id), + targetAgentId: "local", + input: { + kind: "backup", + scheduleId: schedule.id, + scheduleShortId: schedule.shortId, + manual: false, + }, + }); + taskStore.markRunning(task.id); + taskStore.requestCancel(task.id); + + const { startup } = await loadStartupModule(); + + await startup(); + + const updatedSchedule = await db.query.backupSchedulesTable.findFirst({ where: { id: schedule.id } }); + expect(updatedSchedule?.lastBackupStatus).toBe("warning"); + expect(updatedSchedule?.lastBackupError).toBe("Zerobyte was restarted during the last scheduled backup"); + expect(updatedSchedule?.nextBackupAt).toBe(nextBackupAt); +}); + +test("makes in-progress scheduled backups without task rows executable again", async () => { + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const nextBackupAt = Date.now() + 24 * 60 * 60 * 1000; + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + lastBackupStatus: "in_progress", + nextBackupAt, + }); + + const { startup } = await loadStartupModule(); + + await startup(); + + const updatedSchedule = await db.query.backupSchedulesTable.findFirst({ where: { id: schedule.id } }); + expect(updatedSchedule?.lastBackupStatus).toBe("warning"); + expect(updatedSchedule?.lastBackupError).toBe("Zerobyte was restarted during the last scheduled backup"); + expect(updatedSchedule?.nextBackupAt).toBeNull(); +}); + +test("ignores previously stale scheduled tasks when the current interrupted task is manual", async () => { + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const nextBackupAt = Date.now() + 24 * 60 * 60 * 1000; + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + lastBackupStatus: "in_progress", + nextBackupAt, + }); + taskStore.create({ + id: "task-a-old-scheduled", + organizationId: TEST_ORG_ID, + resourceType: "backup_schedule", + resourceId: String(schedule.id), + targetAgentId: "local", + input: { + kind: "backup", + scheduleId: schedule.id, + scheduleShortId: schedule.shortId, + manual: false, + }, + }); + taskStore.markActiveStale({ + organizationId: TEST_ORG_ID, + kind: "backup", + resourceType: "backup_schedule", + resourceId: String(schedule.id), + error: RESTART_TASK_ERROR, + }); + const latestTask = taskStore.create({ + id: "task-z-new-manual", + organizationId: TEST_ORG_ID, + resourceType: "backup_schedule", + resourceId: String(schedule.id), + targetAgentId: "local", + input: { + kind: "backup", + scheduleId: schedule.id, + scheduleShortId: schedule.shortId, + manual: true, + }, + }); + taskStore.markRunning(latestTask.id); + + const { startup } = await loadStartupModule(); + + await startup(); + + const updatedSchedule = await db.query.backupSchedulesTable.findFirst({ where: { id: schedule.id } }); + expect(updatedSchedule?.lastBackupStatus).toBe("warning"); + expect(updatedSchedule?.lastBackupError).toBe("Zerobyte was restarted during the last scheduled backup"); + expect(updatedSchedule?.nextBackupAt).toBe(nextBackupAt); +}); + +test("does not use previously stale scheduled tasks to retry immediately", async () => { + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const nextBackupAt = Date.now() + 24 * 60 * 60 * 1000; + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + lastBackupStatus: "warning", + lastBackupError: "Previous interrupted scheduled backup warning", + nextBackupAt, + }); + taskStore.create({ + id: "task-existing-restart-warning", + organizationId: TEST_ORG_ID, + resourceType: "backup_schedule", + resourceId: String(schedule.id), + targetAgentId: "local", + input: { + kind: "backup", + scheduleId: schedule.id, + scheduleShortId: schedule.shortId, + manual: false, + }, + }); + taskStore.markActiveStale({ + organizationId: TEST_ORG_ID, + kind: "backup", + resourceType: "backup_schedule", + resourceId: String(schedule.id), + error: RESTART_TASK_ERROR, + }); + + const { startup } = await loadStartupModule(); + + await startup(); + + const updatedSchedule = await db.query.backupSchedulesTable.findFirst({ where: { id: schedule.id } }); + expect(updatedSchedule?.nextBackupAt).toBe(nextBackupAt); +}); diff --git a/app/server/modules/lifecycle/startup.ts b/app/server/modules/lifecycle/startup.ts index 421024f6..bd5cc11d 100644 --- a/app/server/modules/lifecycle/startup.ts +++ b/app/server/modules/lifecycle/startup.ts @@ -1,7 +1,5 @@ import { Scheduler } from "../../core/scheduler"; -import { eq } from "drizzle-orm"; import { db } from "../../db/db"; -import { backupSchedulesTable } from "../../db/schema"; import { logger } from "@zerobyte/core/node"; import { volumeService } from "../volumes/volume.service"; import { CleanupDanglingMountsJob } from "../../jobs/cleanup-dangling"; @@ -18,7 +16,7 @@ import { config } from "~/server/core/config"; import { syncProvisionedResources } from "../provisioning/provisioning"; import { toMessage } from "~/server/utils/errors"; import { LOCAL_AGENT_ID } from "../agents/constants"; -import { taskStore } from "../tasks/tasks.store"; +import { RESTART_TASK_ERROR, taskStore } from "../tasks/tasks.store"; const ensureLatestConfigurationSchema = async () => { const volumes = await db.query.volumesTable.findMany({}); @@ -99,9 +97,9 @@ export const startup = async () => { } } + let staleTasks: ReturnType = []; try { - const staleTasks = taskStore.markActiveStale({ error: "Zerobyte was restarted before this task completed" }); - + staleTasks = taskStore.markActiveStale({ error: RESTART_TASK_ERROR }); if (staleTasks.length > 0) { logger.warn(`Marked ${staleTasks.length} active task(s) stale during startup`); } @@ -109,17 +107,9 @@ export const startup = async () => { logger.error(`Failed to mark stale tasks on startup: ${toMessage(err)}`); } - await db - .update(backupSchedulesTable) - .set({ - lastBackupStatus: "warning", - lastBackupError: "Zerobyte was restarted during the last scheduled backup", - updatedAt: Date.now(), - }) - .where(eq(backupSchedulesTable.lastBackupStatus, "in_progress")) - .catch((err) => { - logger.error(`Failed to update stuck backup schedules on startup: ${err.message}`); - }); + await backupsService.recoverInterruptedBackups(staleTasks).catch((err) => { + logger.error(`Failed to recover interrupted backup schedules on startup: ${err.message}`); + }); if (!config.flags.enableLocalAgent) { Scheduler.build(CleanupDanglingMountsJob).schedule("0 * * * *"); diff --git a/app/server/modules/tasks/tasks.store.ts b/app/server/modules/tasks/tasks.store.ts index f7bd3b73..f97594bf 100644 --- a/app/server/modules/tasks/tasks.store.ts +++ b/app/server/modules/tasks/tasks.store.ts @@ -32,6 +32,8 @@ type CreateTaskParams = { type MarkActiveStaleParams = Partial & { error?: string }; +export const RESTART_TASK_ERROR = "Zerobyte was restarted before this task completed"; + const parseTask = (row: unknown): ParsedTask => taskSchema.parse(row); const activeStatusCondition = () => inArray(tasksTable.status, activeTaskStatuses);