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

@@ -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<number>();
+	const retryableScheduledTaskScheduleIds = new Set<number>();
+
+	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<number>();
+		const retryScheduleIds = new Set<number>();
+
+		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,

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);
+});

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<typeof taskStore.markActiveStale> = [];
 	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 * * * *");

app/server/modules/tasks/tasks.store.ts

@@ -32,6 +32,8 @@ type CreateTaskParams = {
 
 type MarkActiveStaleParams = Partial<TaskResource> & { 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);