feat: durable tasks (#927)

app/drizzle/20260530131042_wakeful_santa_claus/migration.sql

@@ -0,0 +1,22 @@
+CREATE TABLE `tasks` (
+	`id` text PRIMARY KEY,
+	`organization_id` text NOT NULL,
+	`kind` text NOT NULL,
+	`status` text NOT NULL,
+	`resource_type` text NOT NULL,
+	`resource_id` text NOT NULL,
+	`target_agent_id` text,
+	`input` text NOT NULL,
+	`progress` text,
+	`result` text,
+	`error` text,
+	`cancellation_requested` integer DEFAULT false NOT NULL,
+	`created_at` integer DEFAULT (unixepoch() * 1000) NOT NULL,
+	`started_at` integer,
+	`updated_at` integer DEFAULT (unixepoch() * 1000) NOT NULL,
+	`finished_at` integer,
+	CONSTRAINT `fk_tasks_organization_id_organization_id_fk` FOREIGN KEY (`organization_id`) REFERENCES `organization`(`id`) ON DELETE CASCADE
+);
+--> statement-breakpoint
+CREATE INDEX `tasks_org_kind_resource_status_idx` ON `tasks` (`organization_id`,`kind`,`resource_type`,`resource_id`,`status`);--> statement-breakpoint
+CREATE INDEX `tasks_org_status_updated_at_idx` ON `tasks` (`organization_id`,`status`,`updated_at`);

app/server/db/schema.ts

@@ -352,6 +352,48 @@ export const repositoryLockWaitersTable = sqliteTable(
 );
 export type RepositoryLockWaiter = typeof repositoryLockWaitersTable.$inferSelect;
 
+type TaskJson = Record<string, unknown>;
+
+export const tasksTable = sqliteTable(
+	"tasks",
+	{
+		id: text("id").primaryKey(),
+		organizationId: text("organization_id")
+			.notNull()
+			.references(() => organization.id, { onDelete: "cascade" }),
+		kind: text("kind").notNull(),
+		status: text("status").notNull(),
+		resourceType: text("resource_type").notNull(),
+		resourceId: text("resource_id").notNull(),
+		targetAgentId: text("target_agent_id"),
+		input: text("input", { mode: "json" }).$type<TaskJson>().notNull(),
+		progress: text("progress", { mode: "json" }).$type<TaskJson | null>(),
+		result: text("result", { mode: "json" }).$type<TaskJson | null>(),
+		error: text("error"),
+		cancellationRequested: int("cancellation_requested", { mode: "boolean" }).notNull().default(false),
+		createdAt: int("created_at", { mode: "number" })
+			.notNull()
+			.default(sql`(unixepoch() * 1000)`),
+		startedAt: int("started_at", { mode: "number" }),
+		updatedAt: int("updated_at", { mode: "number" })
+			.notNull()
+			.default(sql`(unixepoch() * 1000)`),
+		finishedAt: int("finished_at", { mode: "number" }),
+	},
+	(table) => [
+		index("tasks_org_kind_resource_status_idx").on(
+			table.organizationId,
+			table.kind,
+			table.resourceType,
+			table.resourceId,
+			table.status,
+		),
+		index("tasks_org_status_updated_at_idx").on(table.organizationId, table.status, table.updatedAt),
+	],
+);
+export type Task = typeof tasksTable.$inferSelect;
+export type TaskInsert = typeof tasksTable.$inferInsert;
+
 /**
  * Backup Schedules Table
  */

app/server/modules/backups/__tests__/backups.service.execution.test.ts

@@ -25,6 +25,7 @@ import { volumeService } from "~/server/modules/volumes/volume.service";
 import { db } from "~/server/db/db";
 import { config } from "~/server/core/config";
 import { Effect } from "effect";
+import { taskStore } from "~/server/modules/tasks/tasks.store";
 
 const setup = () => {
 	const resticBackupMock = vi.fn((_: SafeSpawnParams) =>
@@ -90,6 +91,18 @@ const setup = () => {
 	};
 };
 
+const getBackupTaskForSchedule = (scheduleId: number) =>
+	db.query.tasksTable.findFirst({
+		where: {
+			AND: [
+				{ organizationId: TEST_ORG_ID },
+				{ kind: "backup" },
+				{ resourceType: "backup_schedule" },
+				{ resourceId: String(scheduleId) },
+			],
+		},
+	});
+
 afterEach(() => {
 	vi.restoreAllMocks();
 	config.flags.enableLocalAgent = true;
@@ -233,6 +246,102 @@ describe("backup execution - validation failures", () => {
 		).toBe(false);
 	});
 
+	test("creates a backup task and uses the task id as the agent job id", async () => {
+		const { runBackupMock } = setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+		});
+
+		await backupsService.executeBackup(schedule.id);
+
+		const task = await getBackupTaskForSchedule(schedule.id);
+		expect(task).toBeDefined();
+		expect(task).toMatchObject({
+			organizationId: TEST_ORG_ID,
+			kind: "backup",
+			status: "succeeded",
+			resourceType: "backup_schedule",
+			resourceId: String(schedule.id),
+			targetAgentId: "local",
+			input: {
+				kind: "backup",
+				scheduleId: schedule.id,
+				scheduleShortId: schedule.shortId,
+				manual: false,
+			},
+			result: expect.objectContaining({
+				kind: "backup",
+				exitCode: 0,
+				warningDetails: null,
+			}),
+		});
+		expect(runBackupMock.mock.calls[0]?.[1].payload.jobId).toBe(task!.id);
+	});
+
+	test("does not leave a queued task behind when backup startup state fails", async () => {
+		const { runBackupMock } = setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+		});
+
+		vi.spyOn(scheduleQueries, "updateStatus").mockRejectedValueOnce(new Error("status update failed"));
+
+		await expect(backupsService.executeBackup(schedule.id)).rejects.toThrow("status update failed");
+
+		expect(await getBackupTaskForSchedule(schedule.id)).toBeUndefined();
+		expect(runBackupMock).not.toHaveBeenCalled();
+	});
+
+	test("persists latest backup progress while preserving execution", async () => {
+		const { runBackupMock } = setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+		});
+
+		runBackupMock.mockImplementationOnce(async (_agentId, request) => {
+			request.onProgress({
+				message_type: "status",
+				seconds_elapsed: 1,
+				seconds_remaining: 9,
+				percent_done: 0.25,
+				total_files: 100,
+				files_done: 25,
+				total_bytes: 1000,
+				bytes_done: 250,
+				current_files: ["file.txt"],
+			});
+
+			return {
+				status: "completed",
+				exitCode: 0,
+				result: JSON.parse(generateBackupOutput()),
+				warningDetails: null,
+			};
+		});
+
+		await backupsService.executeBackup(schedule.id);
+
+		const task = await getBackupTaskForSchedule(schedule.id);
+		expect(task?.status).toBe("succeeded");
+		expect(task?.progress).toMatchObject({
+			kind: "backup",
+			progress: {
+				percent_done: 0.25,
+				bytes_done: 250,
+				current_files: ["file.txt"],
+			},
+		});
+	});
+
 	test("passes configured backup webhooks to the backup agent", async () => {
 		const { runBackupMock } = setup();
 		const volume = await createTestVolume();
@@ -374,8 +483,11 @@ describe("backup execution - validation failures", () => {
 		await backupsService.executeBackup(schedule.id);
 
 		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		const task = await getBackupTaskForSchedule(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("error");
 		expect(updatedSchedule.lastBackupError).toBe("Local backup agent is not connected");
+		expect(task?.status).toBe("failed");
+		expect(task?.error).toBe("Local backup agent is not connected");
 	});
 
 	test("removes stale locks and retries once when the local backup fallback hits a restic lock", async () => {
@@ -491,8 +603,15 @@ describe("stop backup", () => {
 		await backupsService.executeBackup(schedule.id);
 
 		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		const task = await getBackupTaskForSchedule(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		expect(updatedSchedule.lastBackupError).toBe("error: open /mnt/data/private.db: permission denied");
+		expect(task?.status).toBe("succeeded");
+		expect(task?.result).toMatchObject({
+			kind: "backup",
+			exitCode: 3,
+			warningDetails: "error: open /mnt/data/private.db: permission denied",
+		});
 		expect(notificationSpy).toHaveBeenLastCalledWith(
 			schedule.id,
 			"warning",
@@ -645,6 +764,48 @@ describe("stop backup", () => {
 		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
 	});
 
+	test("treats a task that finishes during stop as no longer running", async () => {
+		setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+			lastBackupStatus: "in_progress",
+		});
+		const task = taskStore.create({
+			id: "task-stop-race",
+			organizationId: TEST_ORG_ID,
+			resourceType: "backup_schedule",
+			resourceId: String(schedule.id),
+			targetAgentId: "local",
+			input: {
+				kind: "backup",
+				scheduleId: schedule.id,
+				scheduleShortId: schedule.shortId,
+				manual: false,
+			},
+		});
+		const requestCancel = taskStore.requestCancel;
+		vi.spyOn(taskStore, "requestCancel").mockImplementation((taskId) => {
+			taskStore.complete(taskId, {
+				kind: "backup",
+				exitCode: 0,
+				result: JSON.parse(generateBackupOutput()),
+				warningDetails: null,
+			});
+
+			return requestCancel(taskId);
+		});
+
+		await expect(backupsService.stopBackup(schedule.id)).rejects.toThrow(
+			"No backup is currently running for this schedule",
+		);
+
+		const updatedTask = await getBackupTaskForSchedule(schedule.id);
+		expect(updatedTask).toMatchObject({ id: task.id, status: "succeeded" });
+	});
+
 	test("should stop a running backup when the cancel command cannot be delivered", async () => {
 		const { resticBackupMock, cancelBackupMock } = setup();
 		const volume = await createTestVolume();
@@ -688,6 +849,8 @@ describe("stop backup", () => {
 			await waitForExpect(async () => {
 				const queuedSchedule = await getScheduleByIdOrShortId(schedule.id);
 				expect(queuedSchedule.lastBackupStatus).toBe("in_progress");
+				const task = await getBackupTaskForSchedule(schedule.id);
+				expect(task?.status).toBe("queued");
 			});
 
 			expect(resticBackupMock).not.toHaveBeenCalled();
@@ -700,8 +863,11 @@ describe("stop backup", () => {
 		await executePromise;
 
 		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		const task = await getBackupTaskForSchedule(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
+		expect(task?.status).toBe("cancelled");
+		expect(task?.cancellationRequested).toBe(true);
 		expect(resticBackupMock).not.toHaveBeenCalled();
 	});
 
@@ -802,6 +968,41 @@ describe("stop backup", () => {
 		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
 	});
 
+	test("marks an active task stale when stopping a stuck schedule without a live executor", async () => {
+		setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+			lastBackupStatus: "in_progress",
+		});
+		taskStore.create({
+			id: "task-stuck-backup",
+			organizationId: TEST_ORG_ID,
+			resourceType: "backup_schedule",
+			resourceId: String(schedule.id),
+			targetAgentId: "local",
+			input: {
+				kind: "backup",
+				scheduleId: schedule.id,
+				scheduleShortId: schedule.shortId,
+				manual: false,
+			},
+		});
+
+		await expect(backupsService.stopBackup(schedule.id)).rejects.toThrow(
+			"No backup is currently running for this schedule",
+		);
+
+		const task = await getBackupTaskForSchedule(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		expect(task?.status).toBe("stale");
+		expect(task?.error).toBe("No live backup execution was found for this schedule");
+		expect(updatedSchedule.lastBackupStatus).toBe("warning");
+		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
+	});
+
 	test("should throw NotFoundError when schedule does not exist", async () => {
 		setup();
 		// act & assert

app/server/modules/backups/backup-executor.ts

@@ -15,6 +15,7 @@ import { BadRequestError } from "http-errors-enhanced";
 const FUSE_VOLUME_BACKENDS = new Set<Volume["type"]>(["rclone", "sftp", "webdav"]);
 const IGNORE_INODE_FLAG = "--ignore-inode";
 type BackupExecutionRequest = {
+	jobId: string;
 	scheduleId: number;
 	schedule: BackupSchedule;
 	volume: Volume;
@@ -42,7 +43,7 @@ const createBackupRunPayload = async ({
 	volume,
 	repository,
 	organizationId,
-}: BackupExecutionRequest & { jobId: string }): Promise<BackupRunPayload> => {
+}: BackupExecutionRequest): Promise<BackupRunPayload> => {
 	const agentVolume = { ...volume, config: await decryptVolumeConfig(volume.config) };
 	const customResticParams = schedule.customResticParams ?? [];
 
@@ -119,7 +120,7 @@ export const backupExecutor = {
 			activeControllersByScheduleId.delete(scheduleId);
 		}
 	},
-	execute: async (request: Omit<BackupExecutionRequest, "jobId">) => {
+	execute: async (request: BackupExecutionRequest) => {
 		const trackedExecution = activeControllersByScheduleId.get(request.scheduleId);
 		if (!trackedExecution || trackedExecution.abortController.signal !== request.signal) {
 			throw new Error(`Backup execution for schedule ${request.scheduleId} was not tracked`);
@@ -129,9 +130,7 @@ export const backupExecutor = {
 			throw request.signal.reason || new Error("Operation aborted");
 		}
 
-		const jobId = Bun.randomUUIDv7();
-
-		const payload = await createBackupRunPayload({ ...request, jobId });
+		const payload = await createBackupRunPayload(request);
 
 		if (request.signal.aborted) {
 			throw request.signal.reason || new Error("Operation aborted");

app/server/modules/backups/backups.service.ts

@@ -30,6 +30,27 @@ import { restic } from "../../core/restic";
 import { mirrorQueries } from "./backups.queries";
 import { runEffectPromise, toMessage } from "../../utils/errors";
 import { Effect } from "effect";
+import { taskStore } from "../tasks/tasks.store";
+
+const BACKUP_TASK_RESOURCE_TYPE = "backup_schedule";
+
+const tryCancelTask = (
+	taskId: string,
+	activeTaskResource: { organizationId: string; kind: "backup"; resourceType: string; resourceId: string },
+) => {
+	try {
+		taskStore.requestCancel(taskId);
+		return true;
+	} catch (error) {
+		const currentTask = taskStore.findActiveByResource(activeTaskResource);
+		if (!currentTask || currentTask.id !== taskId) {
+			return false;
+		}
+
+		throw error;
+	}
+};
+
 const listSchedules = async () => {
 	const organizationId = getOrganizationId();
 	const schedules = await db.query.backupSchedulesTable.findMany({
@@ -418,7 +439,16 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 		...(ctx.schedule.cronExpression ? { nextBackupAt: calculateNextRun(ctx.schedule.cronExpression) } : {}),
 	});
 
+	const task = taskStore.create({
+		organizationId: ctx.organizationId,
+		resourceType: BACKUP_TASK_RESOURCE_TYPE,
+		resourceId: String(scheduleId),
+		targetAgentId: ctx.volume.agentId,
+		input: { kind: "backup", scheduleId, scheduleShortId: ctx.schedule.shortId, manual },
+	});
+
 	const abortController = backupExecutor.track(scheduleId);
+	let domainHandlerCompleted = false;
 
 	try {
 		const releaseLock = await repoMutex.acquireShared(
@@ -428,7 +458,10 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 		);
 
 		try {
+			taskStore.markRunning(task.id);
+
 			const executionResult = await backupExecutor.execute({
+				jobId: task.id,
 				scheduleId,
 				schedule: ctx.schedule,
 				volume: ctx.volume,
@@ -437,33 +470,63 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 				signal: abortController.signal,
 				onProgress: (progress) => {
 					updateBackupProgress(ctx, progress);
+					try {
+						taskStore.updateProgress(task.id, { kind: "backup", progress });
+					} catch (error) {
+						logger.error(`Failed to persist backup task progress for ${task.id}: ${toMessage(error)}`);
+					}
 				},
 			});
 
 			switch (executionResult.status) {
-				case "unavailable":
-					return handleBackupFailure(scheduleId, ctx.organizationId, executionResult.error, manual, ctx);
+				case "unavailable": {
+					await handleBackupFailure(scheduleId, ctx.organizationId, executionResult.error, manual, ctx);
+					domainHandlerCompleted = true;
+					taskStore.fail(task.id, toMessage(executionResult.error));
+					return;
+				}
 				case "completed":
-					return finalizeSuccessfulBackup(
+					await finalizeSuccessfulBackup(
 						ctx,
 						executionResult.exitCode,
 						executionResult.result,
 						executionResult.warningDetails,
 					);
-				case "failed":
-					return handleBackupFailure(scheduleId, ctx.organizationId, executionResult.error, manual, ctx);
+					domainHandlerCompleted = true;
+					taskStore.complete(task.id, {
+						kind: "backup",
+						exitCode: executionResult.exitCode,
+						result: executionResult.result,
+						warningDetails: executionResult.warningDetails,
+					});
+					return;
+				case "failed": {
+					await handleBackupFailure(scheduleId, ctx.organizationId, executionResult.error, manual, ctx);
+					domainHandlerCompleted = true;
+					taskStore.fail(task.id, toMessage(executionResult.error));
+					return;
+				}
 				case "cancelled":
-					return handleBackupCancellation(scheduleId, ctx.organizationId, executionResult.message);
+					await handleBackupCancellation(scheduleId, ctx.organizationId, executionResult.message);
+					domainHandlerCompleted = true;
+					taskStore.cancel(task.id, executionResult.message ?? "Backup was stopped by the user");
+					return;
 			}
 		} finally {
 			releaseLock();
 		}
 	} catch (error) {
 		if (abortController.signal.aborted) {
+			taskStore.cancel(task.id, "Backup was stopped by the user");
 			return;
 		}
 
-		return handleBackupFailure(scheduleId, ctx.organizationId, error, manual, ctx);
+		if (domainHandlerCompleted) {
+			throw error;
+		}
+
+		await handleBackupFailure(scheduleId, ctx.organizationId, error, manual, ctx);
+		taskStore.fail(task.id, toMessage(error));
 	} finally {
 		backupExecutor.untrack(scheduleId, abortController);
 		cache.del(cacheKeys.backup.progress(scheduleId));
@@ -483,8 +546,26 @@ const stopBackup = async (scheduleId: number) => {
 		throw new NotFoundError("Backup schedule not found");
 	}
 
+	const activeTaskResource = {
+		organizationId,
+		kind: "backup",
+		resourceType: BACKUP_TASK_RESOURCE_TYPE,
+		resourceId: String(scheduleId),
+	} as const;
+	const activeTask = taskStore.findActiveByResource(activeTaskResource);
+	let shouldMarkActiveTaskStale = false;
+	if (activeTask) {
+		shouldMarkActiveTaskStale = tryCancelTask(activeTask.id, activeTaskResource);
+	}
+
 	try {
 		if (!(await backupExecutor.cancel(scheduleId))) {
+			if (shouldMarkActiveTaskStale) {
+				taskStore.markActiveStale({
+					...activeTaskResource,
+					error: "No live backup execution was found for this schedule",
+				});
+			}
 			throw new ConflictError("No backup is currently running for this schedule");
 		}
 

app/server/modules/lifecycle/__tests__/startup.test.ts

@@ -0,0 +1,82 @@
+import { afterEach, beforeEach, expect, test, vi } from "vitest";
+import { Scheduler } from "~/server/core/scheduler";
+import { serverEvents } from "~/server/core/events";
+import { config } from "~/server/core/config";
+import { db } from "~/server/db/db";
+import { backupsService } from "~/server/modules/backups/backups.service";
+import { repositoriesService } from "~/server/modules/repositories/repositories.service";
+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 { createTestBackupSchedule } from "~/test/helpers/backup";
+import { createTestRepository } from "~/test/helpers/repository";
+import { createTestVolume } from "~/test/helpers/volume";
+import { TEST_ORG_ID } from "~/test/helpers/organization";
+
+const loadStartupModule = async () => {
+	const moduleUrl = new URL("../startup.ts", import.meta.url);
+	moduleUrl.searchParams.set("test", crypto.randomUUID());
+	return import(moduleUrl.href);
+};
+
+let originalEnableLocalAgent: boolean;
+
+beforeEach(() => {
+	originalEnableLocalAgent = config.flags.enableLocalAgent;
+	config.flags.enableLocalAgent = true;
+
+	vi.spyOn(Scheduler, "start").mockResolvedValue();
+	vi.spyOn(Scheduler, "clear").mockResolvedValue();
+	vi.spyOn(Scheduler, "build").mockImplementation(() => ({ schedule: vi.fn() }));
+	vi.spyOn(provisioningModule, "syncProvisionedResources").mockResolvedValue();
+	vi.spyOn(backupsService, "cleanupOrphanedSchedules").mockResolvedValue({ deletedSchedules: 0 });
+	vi.spyOn(volumeService, "updateVolume").mockResolvedValue(undefined as never);
+	vi.spyOn(repositoriesService, "updateRepository").mockResolvedValue(undefined as never);
+	vi.spyOn(notificationsService, "updateDestination").mockResolvedValue(undefined as never);
+});
+
+afterEach(() => {
+	config.flags.enableLocalAgent = originalEnableLocalAgent;
+	vi.restoreAllMocks();
+});
+
+test("marks active tasks stale and keeps stuck backup schedule recovery silent", async () => {
+	const emitSpy = vi.spyOn(serverEvents, "emit");
+	const notificationSpy = vi.spyOn(notificationsService, "sendBackupNotification").mockResolvedValue();
+	const volume = await createTestVolume();
+	const repository = await createTestRepository();
+	const schedule = await createTestBackupSchedule({
+		volumeId: volume.id,
+		repositoryId: repository.id,
+		lastBackupStatus: "in_progress",
+	});
+	const task = taskStore.create({
+		id: "task-startup-active",
+		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);
+
+	const { startup } = await loadStartupModule();
+
+	await startup();
+
+	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?.finishedAt).toEqual(expect.any(Number));
+	expect(updatedSchedule?.lastBackupStatus).toBe("warning");
+	expect(updatedSchedule?.lastBackupError).toBe("Zerobyte was restarted during the last scheduled backup");
+	expect(emitSpy).not.toHaveBeenCalledWith("backup:completed", expect.anything());
+	expect(notificationSpy).not.toHaveBeenCalled();
+});

app/server/modules/lifecycle/startup.ts

@@ -18,6 +18,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";
 
 const ensureLatestConfigurationSchema = async () => {
 	const volumes = await db.query.volumesTable.findMany({});
@@ -98,6 +99,16 @@ export const startup = async () => {
 		}
 	}
 
+	try {
+		const staleTasks = taskStore.markActiveStale({ error: "Zerobyte was restarted before this task completed" });
+
+		if (staleTasks.length > 0) {
+			logger.warn(`Marked ${staleTasks.length} active task(s) stale during startup`);
+		}
+	} catch (err) {
+		logger.error(`Failed to mark stale tasks on startup: ${toMessage(err)}`);
+	}
+
 	await db
 		.update(backupSchedulesTable)
 		.set({

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

@@ -0,0 +1,195 @@
+import { beforeEach, expect, test } from "vitest";
+import { db } from "~/server/db/db";
+import { tasksTable } from "~/server/db/schema";
+import { ensureTestOrganization, TEST_ORG_ID } from "~/test/helpers/organization";
+import { generateBackupOutput } from "~/test/helpers/restic";
+import { taskStore } from "../tasks.store";
+import type { TaskInput, TaskProgress, TaskResult } from "../tasks.schemas";
+
+const backupInput = (scheduleId = 1): TaskInput => ({
+	kind: "backup",
+	scheduleId,
+	scheduleShortId: `schedule-${scheduleId}`,
+	manual: false,
+});
+
+const backupProgress = (percentDone = 0.5): TaskProgress => ({
+	kind: "backup",
+	progress: {
+		message_type: "status",
+		seconds_elapsed: 1,
+		seconds_remaining: 1,
+		percent_done: percentDone,
+		total_files: 10,
+		files_done: 5,
+		total_bytes: 100,
+		bytes_done: 50,
+		current_files: [],
+	},
+});
+
+const backupResult = (): TaskResult => ({
+	kind: "backup",
+	exitCode: 0,
+	result: JSON.parse(generateBackupOutput()),
+	warningDetails: null,
+});
+
+const createBackupTask = (overrides: Partial<Parameters<typeof taskStore.create>[0]> = {}) =>
+	taskStore.create({
+		organizationId: TEST_ORG_ID,
+		resourceType: "backup_schedule",
+		resourceId: "1",
+		targetAgentId: "local",
+		input: backupInput(),
+		...overrides,
+	});
+
+beforeEach(async () => {
+	await ensureTestOrganization();
+	await db.delete(tasksTable);
+});
+
+test("creates queued backup tasks with parsed input and durable metadata only", () => {
+	const task = createBackupTask({
+		id: "task-create",
+		input: { ...backupInput(12), manual: true },
+		resourceId: "12",
+	});
+
+	expect(task).toMatchObject({
+		id: "task-create",
+		organizationId: TEST_ORG_ID,
+		kind: "backup",
+		status: "queued",
+		resourceType: "backup_schedule",
+		resourceId: "12",
+		targetAgentId: "local",
+		input: {
+			kind: "backup",
+			scheduleId: 12,
+			scheduleShortId: "schedule-12",
+			manual: true,
+		},
+		progress: null,
+		result: null,
+		error: null,
+		cancellationRequested: false,
+		startedAt: null,
+		finishedAt: null,
+	});
+	expect(Object.keys(task.input)).toEqual(["kind", "scheduleId", "scheduleShortId", "manual"]);
+});
+
+test("moves an active task through running, progress, cancellation request, and success", () => {
+	const task = createBackupTask({ id: "task-success" });
+
+	const running = taskStore.markRunning(task.id);
+	expect(running.status).toBe("running");
+	expect(running.startedAt).toEqual(expect.any(Number));
+
+	const progressed = taskStore.updateProgress(task.id, backupProgress(0.7));
+	expect(progressed.progress?.progress.percent_done).toBe(0.7);
+
+	const cancelling = taskStore.requestCancel(task.id);
+	expect(cancelling.status).toBe("cancelling");
+	expect(cancelling.cancellationRequested).toBe(true);
+
+	const completed = taskStore.complete(task.id, backupResult());
+	expect(completed.status).toBe("succeeded");
+	expect(completed.result?.result?.snapshot_id).toBe("abcd1234");
+	expect(completed.finishedAt).toEqual(expect.any(Number));
+	expect(completed.cancellationRequested).toBe(true);
+});
+
+test("records failed and cancelled terminal task states", () => {
+	const failedTask = createBackupTask({ id: "task-failed", resourceId: "failed" });
+	const failed = taskStore.fail(failedTask.id, "restic failed");
+	expect(failed.status).toBe("failed");
+	expect(failed.error).toBe("restic failed");
+	expect(failed.finishedAt).toEqual(expect.any(Number));
+
+	const cancelledTask = createBackupTask({ id: "task-cancelled", resourceId: "cancelled" });
+	taskStore.requestCancel(cancelledTask.id);
+	const cancelled = taskStore.cancel(cancelledTask.id, "Backup was stopped by the user");
+	expect(cancelled.status).toBe("cancelled");
+	expect(cancelled.error).toBe("Backup was stopped by the user");
+	expect(cancelled.cancellationRequested).toBe(true);
+});
+
+test("finds the newest active task for a resource and marks only matching active tasks stale", async () => {
+	createBackupTask({ id: "task-a-old", resourceId: "shared" });
+	const newest = createBackupTask({ id: "task-z-new", resourceId: "shared" });
+	const otherResource = createBackupTask({ id: "task-other", resourceId: "other" });
+	const terminal = taskStore.complete(
+		createBackupTask({ id: "task-terminal", resourceId: "shared" }).id,
+		backupResult(),
+	);
+
+	const active = taskStore.findActiveByResource({
+		organizationId: TEST_ORG_ID,
+		kind: "backup",
+		resourceType: "backup_schedule",
+		resourceId: "shared",
+	});
+	expect(active?.id).toBe(newest.id);
+
+	const staleTasks = taskStore.markActiveStale({
+		organizationId: TEST_ORG_ID,
+		kind: "backup",
+		resourceType: "backup_schedule",
+		resourceId: "shared",
+		error: "process restarted",
+	});
+	expect(staleTasks.map((task) => task.id).sort()).toEqual(["task-a-old", "task-z-new"]);
+
+	const other = await db.query.tasksTable.findFirst({ where: { id: otherResource.id } });
+	const completed = await db.query.tasksTable.findFirst({ where: { id: terminal.id } });
+	expect(other?.status).toBe("queued");
+	expect(completed?.status).toBe("succeeded");
+});
+
+test("parses task JSON on reads and rejects invalid persisted shapes", () => {
+	db.insert(tasksTable)
+		.values({
+			id: "task-invalid-json",
+			organizationId: TEST_ORG_ID,
+			kind: "backup",
+			status: "queued",
+			resourceType: "backup_schedule",
+			resourceId: "invalid",
+			input: {
+				kind: "backup",
+				scheduleId: "not-a-number",
+				scheduleShortId: "schedule-invalid",
+				manual: false,
+			},
+			cancellationRequested: false,
+			createdAt: Date.now(),
+			updatedAt: Date.now(),
+		})
+		.run();
+
+	expect(() =>
+		taskStore.findActiveByResource({
+			organizationId: TEST_ORG_ID,
+			kind: "backup",
+			resourceType: "backup_schedule",
+			resourceId: "invalid",
+		}),
+	).toThrow();
+});
+
+test("terminal updates do not mutate unrelated task rows", async () => {
+	const task = createBackupTask({ id: "task-target", resourceId: "target" });
+	const unrelated = createBackupTask({ id: "task-unrelated", resourceId: "unrelated" });
+
+	taskStore.complete(task.id, backupResult());
+
+	const targetRow = await db.query.tasksTable.findFirst({ where: { id: task.id } });
+	const unrelatedRow = await db.query.tasksTable.findFirst({ where: { id: unrelated.id } });
+
+	expect(targetRow?.status).toBe("succeeded");
+	expect(unrelatedRow?.status).toBe("queued");
+	expect(unrelatedRow?.finishedAt).toBeNull();
+});

app/server/modules/tasks/tasks.schemas.ts

@@ -0,0 +1,87 @@
+import { resticBackupOutputSchema, resticBackupProgressSchema } from "@zerobyte/core/restic";
+import { z } from "zod";
+
+export const taskStatuses = ["queued", "running", "cancelling", "cancelled", "succeeded", "failed", "stale"] as const;
+export const activeTaskStatuses = ["queued", "running", "cancelling"] as const;
+
+export const taskStatusSchema = z.enum(taskStatuses);
+export const activeTaskStatusSchema = z.enum(activeTaskStatuses);
+export const taskKindSchema = z.enum(["backup"]);
+
+export const taskInputSchema = z.discriminatedUnion("kind", [
+	z.object({
+		kind: z.literal("backup"),
+		scheduleId: z.number(),
+		scheduleShortId: z.string(),
+		manual: z.boolean(),
+	}),
+]);
+
+export const taskProgressSchema = z.discriminatedUnion("kind", [
+	z.object({
+		kind: z.literal("backup"),
+		progress: resticBackupProgressSchema,
+	}),
+]);
+
+export const taskResultSchema = z.discriminatedUnion("kind", [
+	z.object({
+		kind: z.literal("backup"),
+		exitCode: z.number(),
+		result: resticBackupOutputSchema.nullable(),
+		warningDetails: z.string().nullable(),
+	}),
+]);
+
+export const taskSchema = z
+	.object({
+		id: z.string(),
+		organizationId: z.string(),
+		kind: taskKindSchema,
+		status: taskStatusSchema,
+		resourceType: z.string(),
+		resourceId: z.string(),
+		targetAgentId: z.string().nullable(),
+		input: taskInputSchema,
+		progress: taskProgressSchema.nullable(),
+		result: taskResultSchema.nullable(),
+		error: z.string().nullable(),
+		cancellationRequested: z.boolean(),
+		createdAt: z.number(),
+		startedAt: z.number().nullable(),
+		updatedAt: z.number(),
+		finishedAt: z.number().nullable(),
+	})
+	.superRefine((task, ctx) => {
+		if (task.kind !== task.input.kind) {
+			ctx.addIssue({
+				code: "custom",
+				path: ["input", "kind"],
+				message: "Task input kind must match task kind",
+			});
+		}
+
+		if (task.progress && task.kind !== task.progress.kind) {
+			ctx.addIssue({
+				code: "custom",
+				path: ["progress", "kind"],
+				message: "Task progress kind must match task kind",
+			});
+		}
+
+		if (task.result && task.kind !== task.result.kind) {
+			ctx.addIssue({
+				code: "custom",
+				path: ["result", "kind"],
+				message: "Task result kind must match task kind",
+			});
+		}
+	});
+
+export type TaskStatus = z.infer<typeof taskStatusSchema>;
+export type ActiveTaskStatus = z.infer<typeof activeTaskStatusSchema>;
+export type TaskKind = z.infer<typeof taskKindSchema>;
+export type TaskInput = z.infer<typeof taskInputSchema>;
+export type TaskProgress = z.infer<typeof taskProgressSchema>;
+export type TaskResult = z.infer<typeof taskResultSchema>;
+export type ParsedTask = z.infer<typeof taskSchema>;

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

@@ -0,0 +1,207 @@
+import { and, desc, eq, inArray, type SQL } from "drizzle-orm";
+import { db } from "~/server/db/db";
+import { tasksTable } from "~/server/db/schema";
+import {
+	activeTaskStatuses,
+	taskInputSchema,
+	taskProgressSchema,
+	taskResultSchema,
+	taskSchema,
+	type ParsedTask,
+	type TaskInput,
+	type TaskKind,
+	type TaskProgress,
+	type TaskResult,
+} from "./tasks.schemas";
+
+type TaskResource = {
+	organizationId: string;
+	kind: TaskKind;
+	resourceType: string;
+	resourceId: string;
+};
+
+type CreateTaskParams = {
+	id?: string;
+	organizationId: string;
+	resourceType: string;
+	resourceId: string;
+	targetAgentId?: string | null;
+	input: TaskInput;
+};
+
+type MarkActiveStaleParams = Partial<TaskResource> & {
+	error?: string;
+};
+
+const parseTask = (row: unknown): ParsedTask => taskSchema.parse(row);
+
+const activeStatusCondition = () => inArray(tasksTable.status, activeTaskStatuses);
+
+const byIdCondition = (id: string) => eq(tasksTable.id, id);
+
+const buildActiveConditions = (params: Partial<TaskResource> = {}) => {
+	const conditions: SQL[] = [activeStatusCondition()];
+
+	if (params.organizationId) conditions.push(eq(tasksTable.organizationId, params.organizationId));
+	if (params.kind) conditions.push(eq(tasksTable.kind, params.kind));
+	if (params.resourceType) conditions.push(eq(tasksTable.resourceType, params.resourceType));
+	if (params.resourceId) conditions.push(eq(tasksTable.resourceId, params.resourceId));
+
+	return conditions;
+};
+
+const getUpdatedTask = (row: unknown, taskId: string, operation: string) => {
+	if (!row) {
+		throw new Error(`Task ${taskId} was not ${operation}`);
+	}
+
+	return parseTask(row);
+};
+
+export const taskStore = {
+	create: (params: CreateTaskParams): ParsedTask => {
+		const input = taskInputSchema.parse(params.input);
+		const now = Date.now();
+		const row = db
+			.insert(tasksTable)
+			.values({
+				id: params.id ?? Bun.randomUUIDv7(),
+				organizationId: params.organizationId,
+				kind: input.kind,
+				status: "queued",
+				resourceType: params.resourceType,
+				resourceId: params.resourceId,
+				targetAgentId: params.targetAgentId ?? null,
+				input,
+				progress: null,
+				result: null,
+				error: null,
+				cancellationRequested: false,
+				createdAt: now,
+				updatedAt: now,
+			})
+			.returning()
+			.get();
+
+		return parseTask(row);
+	},
+
+	markRunning: (taskId: string): ParsedTask => {
+		const now = Date.now();
+		const row = db
+			.update(tasksTable)
+			.set({ status: "running", startedAt: now, updatedAt: now })
+			.where(and(byIdCondition(taskId), activeStatusCondition()))
+			.returning()
+			.get();
+
+		return getUpdatedTask(row, taskId, "marked running");
+	},
+
+	updateProgress: (taskId: string, progress: TaskProgress): ParsedTask => {
+		const parsedProgress = taskProgressSchema.parse(progress);
+		const row = db
+			.update(tasksTable)
+			.set({ progress: parsedProgress, updatedAt: Date.now() })
+			.where(and(byIdCondition(taskId), activeStatusCondition()))
+			.returning()
+			.get();
+
+		return getUpdatedTask(row, taskId, "updated with progress");
+	},
+
+	requestCancel: (taskId: string): ParsedTask => {
+		const row = db
+			.update(tasksTable)
+			.set({ status: "cancelling", cancellationRequested: true, updatedAt: Date.now() })
+			.where(and(byIdCondition(taskId), activeStatusCondition()))
+			.returning()
+			.get();
+
+		return getUpdatedTask(row, taskId, "marked cancelling");
+	},
+
+	complete: (taskId: string, result: TaskResult): ParsedTask => {
+		const parsedResult = taskResultSchema.parse(result);
+		const now = Date.now();
+		const row = db
+			.update(tasksTable)
+			.set({
+				status: "succeeded",
+				result: parsedResult,
+				error: null,
+				updatedAt: now,
+				finishedAt: now,
+			})
+			.where(and(byIdCondition(taskId), activeStatusCondition()))
+			.returning()
+			.get();
+
+		return getUpdatedTask(row, taskId, "completed");
+	},
+
+	fail: (taskId: string, error: string): ParsedTask => {
+		const now = Date.now();
+		const row = db
+			.update(tasksTable)
+			.set({
+				status: "failed",
+				error,
+				updatedAt: now,
+				finishedAt: now,
+			})
+			.where(and(byIdCondition(taskId), activeStatusCondition()))
+			.returning()
+			.get();
+
+		return getUpdatedTask(row, taskId, "failed");
+	},
+
+	cancel: (taskId: string, error: string | null = null): ParsedTask => {
+		const now = Date.now();
+		const row = db
+			.update(tasksTable)
+			.set({
+				status: "cancelled",
+				error,
+				updatedAt: now,
+				finishedAt: now,
+			})
+			.where(and(byIdCondition(taskId), activeStatusCondition()))
+			.returning()
+			.get();
+
+		return getUpdatedTask(row, taskId, "cancelled");
+	},
+
+	findActiveByResource: (params: TaskResource): ParsedTask | null => {
+		const rows = db
+			.select()
+			.from(tasksTable)
+			.where(and(...buildActiveConditions(params)))
+			.orderBy(desc(tasksTable.createdAt), desc(tasksTable.id))
+			.limit(1)
+			.all();
+
+		const [row] = rows;
+		return row ? parseTask(row) : null;
+	},
+
+	markActiveStale: (params: MarkActiveStaleParams = {}): ParsedTask[] => {
+		const now = Date.now();
+		const rows = db
+			.update(tasksTable)
+			.set({
+				status: "stale",
+				error: params.error ?? "Task was interrupted before it completed",
+				updatedAt: now,
+				finishedAt: now,
+			})
+			.where(and(...buildActiveConditions(params)))
+			.returning()
+			.all();
+
+		return rows.map(parseTask);
+	},
+};

app/test/helpers/agent-mock.ts

@@ -1,7 +1,7 @@
 import { vi } from "vitest";
 import { fromAny, fromPartial } from "@total-typescript/shoehorn";
 import type { SafeSpawnParams } from "@zerobyte/core/node";
-import type { BackupExecutionResult } from "~/server/modules/agents/agents-manager";
+import type { BackupExecutionProgress, BackupExecutionResult } from "~/server/modules/agents/agents-manager";
 
 export const createAgentBackupMocks = (
 	resticBackupMock: (params: SafeSpawnParams) => Promise<{
@@ -14,7 +14,15 @@ export const createAgentBackupMocks = (
 	const runningBackups = new Map<number, { resolve: (result: BackupExecutionResult) => void; cancelled: boolean }>();
 
 	const runBackupMock = vi.fn(
-		async (_agentId: string, request: { scheduleId: number; payload: { jobId: string }; signal: AbortSignal }) => {
+		async (
+			_agentId: string,
+			request: {
+				scheduleId: number;
+				payload: { jobId: string };
+				signal: AbortSignal;
+				onProgress: (progress: BackupExecutionProgress) => void;
+			},
+		) => {
 			return new Promise<BackupExecutionResult>((resolve) => {
 				runningBackups.set(request.scheduleId, { resolve, cancelled: false });