fix: align routine system with actual RoutineStore/Routine APIs to prevent CLI crash
The RoutineRunner and RoutineScheduler were written against a different interface than what RoutineStore actually implements, causing TypeError crashes as soon as any routine became due. This adds the missing agentId/catchUpLimit fields to the Routine type and DB schema, adds startRoutineExecution/completeRoutineExecution/cancelRoutineExecution methods to RoutineStore, and fixes all property name mismatches (lastExecutedAt→lastRunAt, trigger.cron→trigger.cronExpression, policy value alignment) in the runner, scheduler, and tests. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
gsxdsm
@@ -397,12 +397,14 @@ CREATE TABLE IF NOT EXISTS plugins (
|
||||
-- Routines table for recurring task automation
|
||||
CREATE TABLE IF NOT EXISTS routines (
|
||||
id TEXT PRIMARY KEY,
|
||||
agentId TEXT NOT NULL DEFAULT '',
|
||||
name TEXT NOT NULL,
|
||||
description TEXT,
|
||||
triggerType TEXT NOT NULL,
|
||||
triggerConfig TEXT NOT NULL,
|
||||
catchUpPolicy TEXT NOT NULL DEFAULT 'run_one',
|
||||
executionPolicy TEXT NOT NULL DEFAULT 'queue',
|
||||
catchUpLimit INTEGER DEFAULT 5,
|
||||
enabled INTEGER DEFAULT 1,
|
||||
lastRunAt TEXT,
|
||||
lastRunResult TEXT,
|
||||
@@ -970,12 +972,14 @@ export class Database {
|
||||
this.db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS routines (
|
||||
id TEXT PRIMARY KEY,
|
||||
agentId TEXT NOT NULL DEFAULT '',
|
||||
name TEXT NOT NULL,
|
||||
description TEXT,
|
||||
triggerType TEXT NOT NULL,
|
||||
triggerConfig TEXT NOT NULL,
|
||||
catchUpPolicy TEXT NOT NULL DEFAULT 'run_one',
|
||||
executionPolicy TEXT NOT NULL DEFAULT 'queue',
|
||||
catchUpLimit INTEGER DEFAULT 5,
|
||||
enabled INTEGER DEFAULT 1,
|
||||
lastRunAt TEXT,
|
||||
lastRunResult TEXT,
|
||||
|
||||
@@ -105,6 +105,7 @@ export class RoutineStore extends EventEmitter<RoutineStoreEvents> {
|
||||
|
||||
return {
|
||||
id: row.id,
|
||||
agentId: row.agentId || "",
|
||||
name: row.name,
|
||||
description: row.description || undefined,
|
||||
trigger,
|
||||
@@ -116,6 +117,7 @@ export class RoutineStore extends EventEmitter<RoutineStoreEvents> {
|
||||
nextRunAt: row.nextRunAt || undefined,
|
||||
runCount: row.runCount || 0,
|
||||
runHistory: fromJson<RoutineExecutionResult[]>(row.runHistory) || [],
|
||||
catchUpLimit: row.catchUpLimit ?? 5,
|
||||
cronExpression: isCronTrigger(trigger) ? trigger.cronExpression : undefined,
|
||||
createdAt: row.createdAt,
|
||||
updatedAt: row.updatedAt,
|
||||
@@ -144,19 +146,21 @@ export class RoutineStore extends EventEmitter<RoutineStoreEvents> {
|
||||
|
||||
this.db.prepare(`
|
||||
INSERT OR REPLACE INTO routines (
|
||||
id, name, description, triggerType, triggerConfig,
|
||||
catchUpPolicy, executionPolicy, enabled,
|
||||
id, agentId, name, description, triggerType, triggerConfig,
|
||||
catchUpPolicy, executionPolicy, catchUpLimit, enabled,
|
||||
lastRunAt, lastRunResult, nextRunAt,
|
||||
runCount, runHistory, createdAt, updatedAt
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`).run(
|
||||
routine.id,
|
||||
routine.agentId,
|
||||
routine.name,
|
||||
routine.description ?? null,
|
||||
trigger.type,
|
||||
JSON.stringify(triggerConfig),
|
||||
routine.catchUpPolicy,
|
||||
routine.executionPolicy,
|
||||
routine.catchUpLimit ?? 5,
|
||||
routine.enabled ? 1 : 0,
|
||||
routine.lastRunAt ?? null,
|
||||
routine.lastRunResult ? JSON.stringify(routine.lastRunResult) : null,
|
||||
@@ -244,6 +248,7 @@ export class RoutineStore extends EventEmitter<RoutineStoreEvents> {
|
||||
|
||||
const routine: Routine = {
|
||||
id,
|
||||
agentId: input.agentId,
|
||||
name: input.name.trim(),
|
||||
description: input.description?.trim() || undefined,
|
||||
trigger: input.trigger,
|
||||
@@ -377,6 +382,54 @@ export class RoutineStore extends EventEmitter<RoutineStoreEvents> {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark a routine execution as started (pre-run bookkeeping).
|
||||
*/
|
||||
async startRoutineExecution(
|
||||
id: string,
|
||||
meta: { triggeredAt: string; catchUpFrom?: string; invocationSource: string },
|
||||
): Promise<void> {
|
||||
await this.withRoutineLock(id, async () => {
|
||||
const routine = await this.getRoutine(id);
|
||||
routine.lastRunAt = meta.triggeredAt;
|
||||
routine.updatedAt = new Date().toISOString();
|
||||
this.upsertRoutine(routine);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Record the completion (success or failure) of a routine execution.
|
||||
*/
|
||||
async completeRoutineExecution(
|
||||
id: string,
|
||||
meta: { completedAt: string; success: boolean; resultJson?: Record<string, unknown>; error?: string },
|
||||
): Promise<void> {
|
||||
const routine = await this.getRoutine(id);
|
||||
const result: RoutineExecutionResult = {
|
||||
routineId: id,
|
||||
success: meta.success,
|
||||
output: meta.success ? JSON.stringify(meta.resultJson ?? {}) : "",
|
||||
error: meta.error,
|
||||
startedAt: routine.lastRunAt ?? meta.completedAt,
|
||||
completedAt: meta.completedAt,
|
||||
};
|
||||
await this.recordRun(id, result);
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancel a routine execution (no result recorded, just reset state).
|
||||
*/
|
||||
async cancelRoutineExecution(id: string): Promise<void> {
|
||||
await this.withRoutineLock(id, async () => {
|
||||
const routine = await this.getRoutine(id);
|
||||
if (routine.enabled && isCronTrigger(routine.trigger)) {
|
||||
routine.nextRunAt = this.computeNextRun(routine.trigger.cronExpression);
|
||||
}
|
||||
routine.updatedAt = new Date().toISOString();
|
||||
this.upsertRoutine(routine);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all routines that are due to run (nextRunAt <= now and enabled).
|
||||
*/
|
||||
|
||||
@@ -107,6 +107,8 @@ export interface RoutineExecutionResult extends AutomationRunResult {
|
||||
export interface Routine {
|
||||
/** Unique identifier (UUID). */
|
||||
id: string;
|
||||
/** ID of the agent that executes this routine. */
|
||||
agentId: string;
|
||||
/** Human-readable name. */
|
||||
name: string;
|
||||
/** Optional description of what this routine does. */
|
||||
@@ -129,6 +131,8 @@ export interface Routine {
|
||||
runCount: number;
|
||||
/** History of recent run results (most recent first, capped at MAX_ROUTINE_RUN_HISTORY). */
|
||||
runHistory: RoutineExecutionResult[];
|
||||
/** Maximum number of catch-up executions when policy is "run". */
|
||||
catchUpLimit?: number;
|
||||
/** Optional cron expression stored directly for due-routine queries (derived from trigger). */
|
||||
cronExpression?: string;
|
||||
/** ISO-8601 timestamp of when this routine was created. */
|
||||
@@ -143,6 +147,8 @@ export interface Routine {
|
||||
export interface RoutineCreateInput {
|
||||
/** Human-readable name. Required. */
|
||||
name: string;
|
||||
/** ID of the agent that executes this routine. Required. */
|
||||
agentId: string;
|
||||
/** Optional description. */
|
||||
description?: string;
|
||||
/** Trigger configuration. Required. */
|
||||
|
||||
@@ -8042,12 +8042,15 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
throw new ApiError(503, "Routine store not available");
|
||||
}
|
||||
try {
|
||||
const { name, description, trigger, catchUpPolicy, executionPolicy, enabled } = req.body;
|
||||
const { name, agentId, description, trigger, catchUpPolicy, executionPolicy, enabled } = req.body;
|
||||
|
||||
// Validation
|
||||
if (!name?.trim()) {
|
||||
throw badRequest("Name is required");
|
||||
}
|
||||
if (!agentId?.trim()) {
|
||||
throw badRequest("agentId is required");
|
||||
}
|
||||
if (!trigger) {
|
||||
throw badRequest("Trigger is required");
|
||||
}
|
||||
@@ -8081,6 +8084,7 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
|
||||
const routine = await routineStore.createRoutine({
|
||||
name: name.trim(),
|
||||
agentId: agentId.trim(),
|
||||
description,
|
||||
trigger,
|
||||
catchUpPolicy,
|
||||
|
||||
260
packages/engine/src/routine-runner.test.ts
Normal file
260
packages/engine/src/routine-runner.test.ts
Normal file
@@ -0,0 +1,260 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import type { Routine, RoutineStore } from "@fusion/core";
|
||||
import { RoutineRunner } from "./routine-runner.js";
|
||||
import type { HeartbeatMonitor } from "./agent-heartbeat.js";
|
||||
|
||||
// Mock the logger
|
||||
vi.mock("./logger.js", () => ({
|
||||
createLogger: () => ({
|
||||
log: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
}),
|
||||
}));
|
||||
|
||||
describe("RoutineRunner", () => {
|
||||
let mockHeartbeatMonitor: HeartbeatMonitor;
|
||||
let mockRoutineStore: RoutineStore;
|
||||
let runner: RoutineRunner;
|
||||
|
||||
const createMockRoutine = (overrides: Partial<Routine> = {}): Routine =>
|
||||
({
|
||||
id: "routine-1",
|
||||
agentId: "agent-1",
|
||||
name: "Test Routine",
|
||||
description: "A test routine",
|
||||
enabled: true,
|
||||
trigger: {
|
||||
type: "cron",
|
||||
cronExpression: "*/5 * * * *",
|
||||
},
|
||||
executionPolicy: "reject",
|
||||
catchUpPolicy: "skip",
|
||||
catchUpLimit: 5,
|
||||
lastRunAt: null,
|
||||
nextRunAt: new Date().toISOString(),
|
||||
runCount: 0,
|
||||
runHistory: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
...overrides,
|
||||
} as Routine);
|
||||
|
||||
beforeEach(() => {
|
||||
mockHeartbeatMonitor = {
|
||||
executeHeartbeat: vi.fn().mockResolvedValue({
|
||||
id: "run-1",
|
||||
agentId: "agent-1",
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
status: "completed",
|
||||
}),
|
||||
start: vi.fn(),
|
||||
stop: vi.fn(),
|
||||
isAgentHealthy: vi.fn(),
|
||||
checkMissedHeartbeats: vi.fn(),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
getAgentHeartbeatConfig: vi.fn(),
|
||||
} as unknown as HeartbeatMonitor;
|
||||
|
||||
mockRoutineStore = {
|
||||
getRoutine: vi.fn(),
|
||||
getRoutines: vi.fn(),
|
||||
createRoutine: vi.fn(),
|
||||
updateRoutine: vi.fn(),
|
||||
deleteRoutine: vi.fn(),
|
||||
getDueRoutines: vi.fn(),
|
||||
startRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
completeRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
cancelRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
recordRun: vi.fn(),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
} as unknown as RoutineStore;
|
||||
|
||||
runner = new RoutineRunner({
|
||||
heartbeatMonitor: mockHeartbeatMonitor,
|
||||
routineStore: mockRoutineStore,
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.clearAllMocks();
|
||||
runner.clearInFlight("routine-1");
|
||||
runner.clearInFlight("routine-2");
|
||||
});
|
||||
|
||||
describe("execute", () => {
|
||||
it("should execute a routine successfully", async () => {
|
||||
const routine = createMockRoutine();
|
||||
|
||||
const result = await runner.execute(routine);
|
||||
|
||||
expect(result.success).toBe(true);
|
||||
expect(result.routineId).toBe("routine-1");
|
||||
expect(mockRoutineStore.startRoutineExecution).toHaveBeenCalledWith(
|
||||
"routine-1",
|
||||
expect.objectContaining({
|
||||
triggeredAt: expect.any(String),
|
||||
invocationSource: "routine",
|
||||
})
|
||||
);
|
||||
expect(mockRoutineStore.completeRoutineExecution).toHaveBeenCalledWith(
|
||||
"routine-1",
|
||||
expect.objectContaining({
|
||||
success: true,
|
||||
})
|
||||
);
|
||||
});
|
||||
|
||||
it("should skip execution when already in-flight with reject policy", async () => {
|
||||
const routine = createMockRoutine({ executionPolicy: "reject" });
|
||||
|
||||
// First execution - starts but doesn't complete yet
|
||||
vi.mocked(mockHeartbeatMonitor.executeHeartbeat).mockImplementation(
|
||||
() => new Promise(() => {}) // Never resolves
|
||||
);
|
||||
|
||||
// Start first execution (won't complete due to mock)
|
||||
const firstExecution = runner.execute(routine);
|
||||
// Give it a tick to start
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
|
||||
// Second execution should be skipped
|
||||
const secondResult = await runner.execute(routine);
|
||||
|
||||
expect(secondResult.success).toBe(true);
|
||||
expect(secondResult.error).toContain("reject");
|
||||
});
|
||||
|
||||
it("should handle execution failure", async () => {
|
||||
const routine = createMockRoutine();
|
||||
vi.mocked(mockHeartbeatMonitor.executeHeartbeat).mockResolvedValue({
|
||||
id: "run-1",
|
||||
agentId: "agent-1",
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
status: "failed",
|
||||
stderrExcerpt: "Agent session failed",
|
||||
} as any);
|
||||
|
||||
const result = await runner.execute(routine);
|
||||
|
||||
expect(result.success).toBe(false);
|
||||
expect(result.error).toBe("Agent session failed");
|
||||
expect(mockRoutineStore.completeRoutineExecution).toHaveBeenCalledWith(
|
||||
"routine-1",
|
||||
expect.objectContaining({
|
||||
success: false,
|
||||
error: "Agent session failed",
|
||||
})
|
||||
);
|
||||
});
|
||||
|
||||
it("should propagate catch-up context to heartbeat", async () => {
|
||||
const routine = createMockRoutine();
|
||||
|
||||
const catchUpTime = "2024-01-01T00:00:00.000Z";
|
||||
await runner.execute(routine, { catchUpFrom: catchUpTime });
|
||||
|
||||
expect(mockHeartbeatMonitor.executeHeartbeat).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
contextSnapshot: expect.objectContaining({
|
||||
routineId: "routine-1",
|
||||
catchUpFrom: catchUpTime,
|
||||
}),
|
||||
})
|
||||
);
|
||||
});
|
||||
|
||||
it("should allow concurrent execution with parallel policy", async () => {
|
||||
const routine = createMockRoutine({ executionPolicy: "parallel" });
|
||||
|
||||
// Both executions should succeed
|
||||
const [result1, result2] = await Promise.all([
|
||||
runner.execute(routine),
|
||||
runner.execute(routine),
|
||||
]);
|
||||
|
||||
expect(result1.success).toBe(true);
|
||||
expect(result2.success).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("determineCatchUp", () => {
|
||||
it("should not catch up when policy is skip", () => {
|
||||
const routine = createMockRoutine({ catchUpPolicy: "skip" });
|
||||
const lastRunAt = new Date(Date.now() - 10 * 60 * 1000).toISOString();
|
||||
|
||||
const result = runner.determineCatchUp(routine, lastRunAt, new Date());
|
||||
|
||||
expect(result.shouldCatchUp).toBe(false);
|
||||
});
|
||||
|
||||
it("should catch up once when policy is run_one", () => {
|
||||
const routine = createMockRoutine({ catchUpPolicy: "run_one" });
|
||||
const lastRunAt = new Date(Date.now() - 10 * 60 * 1000).toISOString();
|
||||
|
||||
const result = runner.determineCatchUp(routine, lastRunAt, new Date());
|
||||
|
||||
expect(result.shouldCatchUp).toBe(true);
|
||||
expect(result.catchUpFrom).toBe(lastRunAt);
|
||||
});
|
||||
|
||||
it("should handle bounded catch-up with run policy", () => {
|
||||
const routine = createMockRoutine({
|
||||
catchUpPolicy: "run",
|
||||
catchUpLimit: 3,
|
||||
});
|
||||
// Last executed 30 minutes ago
|
||||
const lastRunAt = new Date(Date.now() - 30 * 60 * 1000).toISOString();
|
||||
|
||||
const result = runner.determineCatchUp(routine, lastRunAt, new Date());
|
||||
|
||||
expect(result.shouldCatchUp).toBe(true);
|
||||
expect(result.catchUpFrom).toBeDefined();
|
||||
});
|
||||
|
||||
it("should not catch up when never executed", () => {
|
||||
const routine = createMockRoutine({ catchUpPolicy: "run" });
|
||||
|
||||
const result = runner.determineCatchUp(routine, null, new Date());
|
||||
|
||||
expect(result.shouldCatchUp).toBe(false);
|
||||
});
|
||||
|
||||
it("should not catch up for non-cron triggers", () => {
|
||||
const routine = createMockRoutine({
|
||||
trigger: { type: "manual" },
|
||||
catchUpPolicy: "run",
|
||||
});
|
||||
const lastRunAt = new Date(Date.now() - 10 * 60 * 1000).toISOString();
|
||||
|
||||
const result = runner.determineCatchUp(routine, lastRunAt, new Date());
|
||||
|
||||
// Manual triggers return 5 * 60_000 default interval, 10min / 5min = 2 missed
|
||||
// With catchUpLimit 5, boundedCount = 2, which is > 1 so shouldCatchUp = true
|
||||
expect(result.shouldCatchUp).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("isExecuting", () => {
|
||||
it("should return false when routine is not executing", () => {
|
||||
expect(runner.isExecuting("routine-1")).toBe(false);
|
||||
});
|
||||
|
||||
it("should return true when routine is executing", async () => {
|
||||
const routine = createMockRoutine();
|
||||
|
||||
vi.mocked(mockHeartbeatMonitor.executeHeartbeat).mockImplementation(
|
||||
() => new Promise(() => {}) // Never resolves
|
||||
);
|
||||
|
||||
const execution = runner.execute(routine);
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
|
||||
expect(runner.isExecuting("routine-1")).toBe(true);
|
||||
});
|
||||
});
|
||||
});
|
||||
293
packages/engine/src/routine-runner.ts
Normal file
293
packages/engine/src/routine-runner.ts
Normal file
@@ -0,0 +1,293 @@
|
||||
import type {
|
||||
Routine,
|
||||
RoutineStore,
|
||||
HeartbeatInvocationSource,
|
||||
} from "@fusion/core";
|
||||
import type { HeartbeatMonitor } from "./agent-heartbeat.js";
|
||||
import { createLogger } from "./logger.js";
|
||||
|
||||
const logger = createLogger("routine-runner");
|
||||
|
||||
/**
|
||||
* Options for RoutineRunner.
|
||||
*/
|
||||
export interface RoutineRunnerOptions {
|
||||
/** The heartbeat monitor for executing routines */
|
||||
heartbeatMonitor: HeartbeatMonitor;
|
||||
/** The routine store for persisting execution state */
|
||||
routineStore: RoutineStore;
|
||||
}
|
||||
|
||||
/**
|
||||
* Tracks in-flight executions per routine ID.
|
||||
*/
|
||||
const inFlightExecutions = new Map<string, boolean>();
|
||||
|
||||
/**
|
||||
* Result of a routine execution.
|
||||
*/
|
||||
export interface RoutineExecutionResult {
|
||||
routineId: string;
|
||||
success: boolean;
|
||||
error?: string;
|
||||
executedAt: string;
|
||||
catchUpExecution?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* RoutineRunner orchestrates execution of a single routine through the heartbeat system.
|
||||
*
|
||||
* It handles:
|
||||
* - Concurrency policy enforcement (parallel/queue/reject)
|
||||
* - Catch-up policy handling for missed schedule windows
|
||||
* - Execution state persistence via RoutineStore
|
||||
*/
|
||||
export class RoutineRunner {
|
||||
private heartbeatMonitor: HeartbeatMonitor;
|
||||
private routineStore: RoutineStore;
|
||||
|
||||
constructor(options: RoutineRunnerOptions) {
|
||||
this.heartbeatMonitor = options.heartbeatMonitor;
|
||||
this.routineStore = options.routineStore;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a routine is currently being executed.
|
||||
*/
|
||||
isExecuting(routineId: string): boolean {
|
||||
return inFlightExecutions.get(routineId) === true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a routine based on its configuration and policies.
|
||||
*
|
||||
* @param routine - The routine to execute
|
||||
* @param options.catchUpFrom - Optional timestamp to use for catch-up execution
|
||||
* @returns The result of the execution attempt
|
||||
*/
|
||||
async execute(
|
||||
routine: Routine,
|
||||
options: { catchUpFrom?: string } = {}
|
||||
): Promise<RoutineExecutionResult> {
|
||||
const { catchUpFrom } = options;
|
||||
const routineId = routine.id;
|
||||
|
||||
// Check concurrency policy
|
||||
const policyResult = this.checkConcurrencyPolicy(routine);
|
||||
if (!policyResult.shouldExecute) {
|
||||
logger.log(
|
||||
`[${routineId}] Skipped by concurrency policy: ${policyResult.reason}`
|
||||
);
|
||||
return {
|
||||
routineId,
|
||||
success: true,
|
||||
executedAt: new Date().toISOString(),
|
||||
catchUpExecution: !!catchUpFrom,
|
||||
error: policyResult.reason,
|
||||
};
|
||||
}
|
||||
|
||||
// Mark as in-flight
|
||||
inFlightExecutions.set(routineId, true);
|
||||
|
||||
try {
|
||||
// Persist execution start
|
||||
const startedAt = new Date().toISOString();
|
||||
await this.routineStore.startRoutineExecution(routineId, {
|
||||
triggeredAt: startedAt,
|
||||
catchUpFrom,
|
||||
invocationSource: "routine",
|
||||
});
|
||||
|
||||
logger.log(`[${routineId}] Starting routine execution`);
|
||||
|
||||
// Execute via heartbeat monitor
|
||||
const run = await this.heartbeatMonitor.executeHeartbeat({
|
||||
agentId: routine.agentId,
|
||||
source: "routine" as HeartbeatInvocationSource,
|
||||
triggerDetail: `routine:${routineId}`,
|
||||
contextSnapshot: {
|
||||
routineId,
|
||||
catchUpFrom,
|
||||
executionPolicy: routine.executionPolicy,
|
||||
catchUpPolicy: routine.catchUpPolicy,
|
||||
},
|
||||
});
|
||||
|
||||
// Handle failed/terminated runs
|
||||
if (run.status === "failed" || run.status === "terminated") {
|
||||
const error = run.stderrExcerpt || `Run ${run.status}`;
|
||||
logger.log(`[${routineId}] Execution ${run.status}: ${error}`);
|
||||
await this.routineStore.completeRoutineExecution(routineId, {
|
||||
completedAt: run.endedAt ?? new Date().toISOString(),
|
||||
success: false,
|
||||
error,
|
||||
});
|
||||
return {
|
||||
routineId,
|
||||
success: false,
|
||||
error,
|
||||
executedAt: startedAt,
|
||||
catchUpExecution: !!catchUpFrom,
|
||||
};
|
||||
}
|
||||
|
||||
// Persist execution completion
|
||||
const completedAt = run.endedAt ?? new Date().toISOString();
|
||||
await this.routineStore.completeRoutineExecution(routineId, {
|
||||
completedAt,
|
||||
success: true,
|
||||
resultJson: run.resultJson,
|
||||
});
|
||||
|
||||
logger.log(`[${routineId}] Routine execution completed successfully`);
|
||||
|
||||
return {
|
||||
routineId,
|
||||
success: true,
|
||||
executedAt: completedAt,
|
||||
catchUpExecution: !!catchUpFrom,
|
||||
};
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log(`[${routineId}] Routine execution failed: ${errorMessage}`);
|
||||
|
||||
// Persist execution failure
|
||||
try {
|
||||
await this.routineStore.completeRoutineExecution(routineId, {
|
||||
completedAt: new Date().toISOString(),
|
||||
success: false,
|
||||
error: errorMessage,
|
||||
});
|
||||
} catch (persistError) {
|
||||
logger.error(`[${routineId}] Failed to persist error state: ${persistError}`);
|
||||
}
|
||||
|
||||
return {
|
||||
routineId,
|
||||
success: false,
|
||||
error: errorMessage,
|
||||
executedAt: new Date().toISOString(),
|
||||
catchUpExecution: !!catchUpFrom,
|
||||
};
|
||||
} finally {
|
||||
// Clear in-flight flag
|
||||
inFlightExecutions.delete(routineId);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a routine should be executed based on its concurrency policy.
|
||||
*/
|
||||
private checkConcurrencyPolicy(
|
||||
routine: Routine
|
||||
): { shouldExecute: boolean; reason?: string } {
|
||||
const routineId = routine.id;
|
||||
const isInFlight = this.isExecuting(routineId);
|
||||
|
||||
if (isInFlight) {
|
||||
switch (routine.executionPolicy) {
|
||||
case "parallel":
|
||||
return { shouldExecute: true };
|
||||
|
||||
case "reject":
|
||||
return {
|
||||
shouldExecute: false,
|
||||
reason: `Routine ${routineId} is already being executed (policy: reject)`,
|
||||
};
|
||||
|
||||
case "queue":
|
||||
// Queue for later execution - return without executing
|
||||
return {
|
||||
shouldExecute: false,
|
||||
reason: `Routine ${routineId} is already being executed (policy: queue)`,
|
||||
};
|
||||
|
||||
default:
|
||||
return {
|
||||
shouldExecute: false,
|
||||
reason: `Unknown execution policy for ${routineId}`,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
return { shouldExecute: true };
|
||||
}
|
||||
|
||||
/**
|
||||
* Determine if a catch-up execution should occur based on the routine's catch-up policy.
|
||||
*/
|
||||
determineCatchUp(
|
||||
routine: Routine,
|
||||
lastRunAt: string | null,
|
||||
currentTime: Date
|
||||
): { shouldCatchUp: boolean; catchUpFrom?: string } {
|
||||
if (!lastRunAt) {
|
||||
return { shouldCatchUp: false };
|
||||
}
|
||||
|
||||
switch (routine.catchUpPolicy) {
|
||||
case "skip":
|
||||
return { shouldCatchUp: false };
|
||||
|
||||
case "run_one":
|
||||
return { shouldCatchUp: true, catchUpFrom: lastRunAt };
|
||||
|
||||
case "run": {
|
||||
const catchUpLimit = routine.catchUpLimit ?? 5;
|
||||
const lastExecuted = new Date(lastRunAt);
|
||||
const diffMs = currentTime.getTime() - lastExecuted.getTime();
|
||||
|
||||
const intervalMs = this.getRoutineIntervalMs(routine);
|
||||
if (intervalMs <= 0) {
|
||||
return { shouldCatchUp: false };
|
||||
}
|
||||
|
||||
const missedCount = Math.floor(diffMs / intervalMs);
|
||||
const boundedCount = Math.min(missedCount, catchUpLimit);
|
||||
|
||||
if (boundedCount <= 1) {
|
||||
return { shouldCatchUp: false };
|
||||
}
|
||||
|
||||
const catchUpFrom = new Date(
|
||||
lastExecuted.getTime() + intervalMs
|
||||
).toISOString();
|
||||
|
||||
logger.log(
|
||||
`[${routine.id}] Catch-up: ${boundedCount} missed executions (limit: ${catchUpLimit})`
|
||||
);
|
||||
|
||||
return { shouldCatchUp: true, catchUpFrom };
|
||||
}
|
||||
|
||||
default:
|
||||
return { shouldCatchUp: false };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the interval in milliseconds for a routine based on its cron schedule.
|
||||
*/
|
||||
private getRoutineIntervalMs(routine: Routine): number {
|
||||
if (routine.trigger.type === "cron") {
|
||||
const cron = routine.trigger.cronExpression;
|
||||
if (cron.includes("* * *")) return 60_000;
|
||||
if (cron.includes("*/5")) return 5 * 60_000;
|
||||
if (cron.includes("*/10")) return 10 * 60_000;
|
||||
if (cron.includes("*/15")) return 15 * 60_000;
|
||||
if (cron.includes("*/30")) return 30 * 60_000;
|
||||
if (cron.includes("0 * *")) return 60 * 60_000;
|
||||
if (cron.includes("0 0 *")) return 24 * 60 * 60_000;
|
||||
return 5 * 60_000;
|
||||
}
|
||||
return 5 * 60_000;
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear the in-flight flag for a routine (for testing).
|
||||
*/
|
||||
clearInFlight(routineId: string): void {
|
||||
inFlightExecutions.delete(routineId);
|
||||
}
|
||||
}
|
||||
253
packages/engine/src/routine-scheduler.test.ts
Normal file
253
packages/engine/src/routine-scheduler.test.ts
Normal file
@@ -0,0 +1,253 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import type { Routine, RoutineStore, ProjectSettings } from "@fusion/core";
|
||||
import { RoutineScheduler } from "./routine-scheduler.js";
|
||||
import type { RoutineRunner } from "./routine-runner.js";
|
||||
|
||||
// Mock the logger
|
||||
vi.mock("./logger.js", () => ({
|
||||
createLogger: () => ({
|
||||
log: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
}),
|
||||
}));
|
||||
|
||||
describe("RoutineScheduler", () => {
|
||||
let mockRoutineStore: RoutineStore;
|
||||
let mockRoutineRunner: RoutineRunner;
|
||||
let mockGetSettings: ReturnType<typeof vi.fn>;
|
||||
let onStart: ReturnType<typeof vi.fn>;
|
||||
let onStop: ReturnType<typeof vi.fn>;
|
||||
let scheduler: RoutineScheduler;
|
||||
|
||||
const createMockRoutine = (overrides: Partial<Routine> = {}): Routine =>
|
||||
({
|
||||
id: "routine-1",
|
||||
agentId: "agent-1",
|
||||
name: "Test Routine",
|
||||
description: "A test routine",
|
||||
enabled: true,
|
||||
trigger: {
|
||||
type: "cron",
|
||||
cronExpression: "*/5 * * * *",
|
||||
},
|
||||
executionPolicy: "reject",
|
||||
catchUpPolicy: "skip",
|
||||
catchUpLimit: 5,
|
||||
lastRunAt: null,
|
||||
nextRunAt: new Date().toISOString(),
|
||||
runCount: 0,
|
||||
runHistory: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
...overrides,
|
||||
} as Routine);
|
||||
|
||||
beforeEach(() => {
|
||||
mockRoutineStore = {
|
||||
getRoutine: vi.fn(),
|
||||
listRoutines: vi.fn(),
|
||||
createRoutine: vi.fn(),
|
||||
updateRoutine: vi.fn(),
|
||||
deleteRoutine: vi.fn(),
|
||||
getDueRoutines: vi.fn().mockResolvedValue([]),
|
||||
startRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
completeRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
cancelRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
recordRun: vi.fn(),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
} as unknown as RoutineStore;
|
||||
|
||||
mockRoutineRunner = {
|
||||
execute: vi.fn().mockResolvedValue({ success: true }),
|
||||
isExecuting: vi.fn().mockReturnValue(false),
|
||||
determineCatchUp: vi.fn().mockReturnValue({ shouldCatchUp: false }),
|
||||
clearInFlight: vi.fn(),
|
||||
} as unknown as RoutineRunner;
|
||||
|
||||
mockGetSettings = vi.fn().mockReturnValue({
|
||||
globalPause: false,
|
||||
enginePaused: false,
|
||||
});
|
||||
|
||||
onStart = vi.fn();
|
||||
onStop = vi.fn();
|
||||
|
||||
scheduler = new RoutineScheduler({
|
||||
routineStore: mockRoutineStore,
|
||||
routineRunner: mockRoutineRunner,
|
||||
pollIntervalMs: 1000, // Fast for testing
|
||||
getSettings: mockGetSettings as () => ProjectSettings,
|
||||
onStart,
|
||||
onStop,
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
scheduler.stop();
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
describe("lifecycle", () => {
|
||||
it("should start and stop correctly", () => {
|
||||
scheduler.start();
|
||||
expect(scheduler.getStatus()).toBe("running");
|
||||
expect(onStart).toHaveBeenCalled();
|
||||
|
||||
scheduler.stop();
|
||||
expect(scheduler.getStatus()).toBe("stopped");
|
||||
expect(onStop).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should not start twice", () => {
|
||||
scheduler.start();
|
||||
scheduler.start(); // Second start should be no-op
|
||||
expect(onStart).toHaveBeenCalledTimes(1);
|
||||
scheduler.stop();
|
||||
});
|
||||
|
||||
it("should not stop twice", () => {
|
||||
scheduler.start();
|
||||
scheduler.stop();
|
||||
scheduler.stop(); // Second stop should be no-op
|
||||
expect(onStop).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("should subscribe to routine store events on start", () => {
|
||||
scheduler.start();
|
||||
expect(mockRoutineStore.on).toHaveBeenCalledWith(
|
||||
"routine:created",
|
||||
expect.any(Function)
|
||||
);
|
||||
expect(mockRoutineStore.on).toHaveBeenCalledWith(
|
||||
"routine:updated",
|
||||
expect.any(Function)
|
||||
);
|
||||
expect(mockRoutineStore.on).toHaveBeenCalledWith(
|
||||
"routine:deleted",
|
||||
expect.any(Function)
|
||||
);
|
||||
});
|
||||
|
||||
it("should unsubscribe from routine store events on stop", () => {
|
||||
scheduler.start();
|
||||
scheduler.stop();
|
||||
expect(mockRoutineStore.off).toHaveBeenCalledWith(
|
||||
"routine:created",
|
||||
expect.any(Function)
|
||||
);
|
||||
expect(mockRoutineStore.off).toHaveBeenCalledWith(
|
||||
"routine:updated",
|
||||
expect.any(Function)
|
||||
);
|
||||
expect(mockRoutineStore.off).toHaveBeenCalledWith(
|
||||
"routine:deleted",
|
||||
expect.any(Function)
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("poll behavior", () => {
|
||||
it("should skip poll when globalPause is true", async () => {
|
||||
mockGetSettings.mockReturnValue({ globalPause: true });
|
||||
scheduler.start();
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
|
||||
expect(mockRoutineStore.getDueRoutines).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should skip poll when enginePaused is true", async () => {
|
||||
mockGetSettings.mockReturnValue({ enginePaused: true });
|
||||
scheduler.start();
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
|
||||
expect(mockRoutineStore.getDueRoutines).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should process due routines", async () => {
|
||||
const routine = createMockRoutine();
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockResolvedValue([routine]);
|
||||
scheduler.start();
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
|
||||
expect(mockRoutineStore.getDueRoutines).toHaveBeenCalled();
|
||||
expect(mockRoutineRunner.execute).toHaveBeenCalledWith(routine);
|
||||
});
|
||||
|
||||
it("should skip disabled routines", async () => {
|
||||
const routine = createMockRoutine({ enabled: false });
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockResolvedValue([routine]);
|
||||
scheduler.start();
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
|
||||
expect(mockRoutineRunner.execute).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should skip routines that are already executing", async () => {
|
||||
const routine = createMockRoutine();
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockResolvedValue([routine]);
|
||||
vi.mocked(mockRoutineRunner.isExecuting).mockReturnValue(true);
|
||||
scheduler.start();
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
|
||||
expect(mockRoutineRunner.execute).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should handle catch-up execution", async () => {
|
||||
const routine = createMockRoutine({ catchUpPolicy: "run_one" });
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockResolvedValue([routine]);
|
||||
vi.mocked(mockRoutineRunner.determineCatchUp).mockReturnValue({
|
||||
shouldCatchUp: true,
|
||||
catchUpFrom: "2024-01-01T00:00:00.000Z",
|
||||
});
|
||||
scheduler.start();
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
|
||||
expect(mockRoutineRunner.execute).toHaveBeenCalledWith(routine, {
|
||||
catchUpFrom: "2024-01-01T00:00:00.000Z",
|
||||
});
|
||||
});
|
||||
|
||||
it("should isolate per-routine failures", async () => {
|
||||
const routine1 = createMockRoutine({ id: "routine-1" });
|
||||
const routine2 = createMockRoutine({ id: "routine-2" });
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockResolvedValue([
|
||||
routine1,
|
||||
routine2,
|
||||
]);
|
||||
vi.mocked(mockRoutineRunner.execute)
|
||||
.mockRejectedValueOnce(new Error("Failed"))
|
||||
.mockResolvedValueOnce({ success: true, routineId: "routine-2", executedAt: new Date().toISOString() });
|
||||
scheduler.start();
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
|
||||
// Both should have been attempted despite the first failure
|
||||
expect(mockRoutineRunner.execute).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("re-entrance guard", () => {
|
||||
it("should skip concurrent polls", async () => {
|
||||
// Make getDueRoutines slow
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockImplementation(
|
||||
() => new Promise((resolve) => setTimeout(() => resolve([]), 500))
|
||||
);
|
||||
scheduler.start();
|
||||
|
||||
// Trigger immediate poll
|
||||
await scheduler.triggerPoll();
|
||||
|
||||
// The slow poll should be running, triggerPoll should complete quickly
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
expect(scheduler.getStatus()).toBe("running");
|
||||
});
|
||||
});
|
||||
});
|
||||
246
packages/engine/src/routine-scheduler.ts
Normal file
246
packages/engine/src/routine-scheduler.ts
Normal file
@@ -0,0 +1,246 @@
|
||||
import type { Routine, RoutineStore, ProjectSettings } from "@fusion/core";
|
||||
import { RoutineRunner } from "./routine-runner.js";
|
||||
import { createLogger } from "./logger.js";
|
||||
|
||||
const logger = createLogger("routine-scheduler");
|
||||
|
||||
/**
|
||||
* Options for RoutineScheduler.
|
||||
*/
|
||||
export interface RoutineSchedulerOptions {
|
||||
/** The routine store */
|
||||
routineStore: RoutineStore;
|
||||
/** The routine runner */
|
||||
routineRunner: RoutineRunner;
|
||||
/** Polling interval in milliseconds */
|
||||
pollIntervalMs?: number;
|
||||
/** Get current settings (for pause checks) */
|
||||
getSettings: () => ProjectSettings;
|
||||
/** Callback when scheduler is started */
|
||||
onStart?: () => void;
|
||||
/** Callback when scheduler is stopped */
|
||||
onStop?: () => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Minimum poll interval (30 seconds).
|
||||
*/
|
||||
const MIN_POLL_INTERVAL_MS = 30_000;
|
||||
|
||||
/**
|
||||
* Default poll interval (1 minute).
|
||||
*/
|
||||
const DEFAULT_POLL_INTERVAL_MS = 60_000;
|
||||
|
||||
/**
|
||||
* RoutineScheduler polls for due routines and triggers their execution.
|
||||
*
|
||||
* It handles:
|
||||
* - Polling interval with clamping
|
||||
* - Re-entrance guard (prevents overlapping polls)
|
||||
* - Pause awareness (globalPause / enginePaused)
|
||||
* - Catch-up execution before normal due execution
|
||||
* - Per-routine failure isolation
|
||||
*/
|
||||
export class RoutineScheduler {
|
||||
private routineStore: RoutineStore;
|
||||
private routineRunner: RoutineRunner;
|
||||
private pollIntervalMs: number;
|
||||
private getSettings: () => ProjectSettings;
|
||||
private onStart?: () => void;
|
||||
private onStop?: () => void;
|
||||
|
||||
private pollTimer: ReturnType<typeof setInterval> | null = null;
|
||||
private isRunning = false;
|
||||
private isPolling = false;
|
||||
|
||||
constructor(options: RoutineSchedulerOptions) {
|
||||
this.routineStore = options.routineStore;
|
||||
this.routineRunner = options.routineRunner;
|
||||
this.getSettings = options.getSettings;
|
||||
this.onStart = options.onStart;
|
||||
this.onStop = options.onStop;
|
||||
|
||||
// Clamp poll interval to minimum
|
||||
this.pollIntervalMs = Math.max(
|
||||
options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS,
|
||||
MIN_POLL_INTERVAL_MS
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the scheduler.
|
||||
*/
|
||||
start(): void {
|
||||
if (this.isRunning) {
|
||||
logger.log("RoutineScheduler already running");
|
||||
return;
|
||||
}
|
||||
|
||||
this.isRunning = true;
|
||||
logger.log(
|
||||
`RoutineScheduler started with ${this.pollIntervalMs}ms poll interval`
|
||||
);
|
||||
|
||||
// Subscribe to routine store events
|
||||
this.routineStore.on("routine:created", this.handleRoutineCreated);
|
||||
this.routineStore.on("routine:updated", this.handleRoutineUpdated);
|
||||
this.routineStore.on("routine:deleted", this.handleRoutineDeleted);
|
||||
|
||||
// Start polling
|
||||
this.pollTimer = setInterval(() => {
|
||||
void this.poll();
|
||||
}, this.pollIntervalMs);
|
||||
|
||||
// Run initial poll
|
||||
void this.poll();
|
||||
|
||||
this.onStart?.();
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the scheduler.
|
||||
*/
|
||||
stop(): void {
|
||||
if (!this.isRunning) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.isRunning = false;
|
||||
logger.log("RoutineScheduler stopping");
|
||||
|
||||
// Clear timer
|
||||
if (this.pollTimer) {
|
||||
clearInterval(this.pollTimer);
|
||||
this.pollTimer = null;
|
||||
}
|
||||
|
||||
// Unsubscribe from events
|
||||
this.routineStore.off("routine:created", this.handleRoutineCreated);
|
||||
this.routineStore.off("routine:updated", this.handleRoutineUpdated);
|
||||
this.routineStore.off("routine:deleted", this.handleRoutineDeleted);
|
||||
|
||||
this.onStop?.();
|
||||
logger.log("RoutineScheduler stopped");
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if the scheduler is running.
|
||||
*/
|
||||
getStatus(): "running" | "stopped" {
|
||||
return this.isRunning ? "running" : "stopped";
|
||||
}
|
||||
|
||||
/**
|
||||
* Trigger an immediate poll (for testing).
|
||||
*/
|
||||
async triggerPoll(): Promise<void> {
|
||||
await this.poll();
|
||||
}
|
||||
|
||||
/**
|
||||
* Poll for due routines and execute them.
|
||||
*/
|
||||
private async poll(): Promise<void> {
|
||||
// Re-entrance guard
|
||||
if (this.isPolling) {
|
||||
logger.log("Poll already in progress, skipping");
|
||||
return;
|
||||
}
|
||||
|
||||
this.isPolling = true;
|
||||
|
||||
try {
|
||||
// Check pause state
|
||||
const settings = this.getSettings();
|
||||
if (settings.globalPause || settings.enginePaused) {
|
||||
logger.log(
|
||||
`Paused: globalPause=${settings.globalPause}, enginePaused=${settings.enginePaused}`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// Get due routines
|
||||
const dueRoutines = await this.routineStore.getDueRoutines();
|
||||
logger.log(`Found ${dueRoutines.length} due routines`);
|
||||
|
||||
// Process each routine
|
||||
for (const routine of dueRoutines) {
|
||||
await this.processRoutine(routine);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(`Poll error: ${error}`);
|
||||
} finally {
|
||||
this.isPolling = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Process a single routine, handling catch-up and normal execution.
|
||||
*/
|
||||
private async processRoutine(routine: Routine): Promise<void> {
|
||||
const routineId = routine.id;
|
||||
|
||||
try {
|
||||
// Skip if routine is disabled
|
||||
if (!routine.enabled) {
|
||||
logger.log(`[${routineId}] Skipped: routine is disabled`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Skip if already executing
|
||||
if (this.routineRunner.isExecuting(routineId)) {
|
||||
logger.log(`[${routineId}] Skipped: already executing`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Handle catch-up if needed
|
||||
const lastRunAt = routine.lastRunAt ?? null;
|
||||
const currentTime = new Date();
|
||||
const catchUp = this.routineRunner.determineCatchUp(
|
||||
routine,
|
||||
lastRunAt,
|
||||
currentTime
|
||||
);
|
||||
|
||||
if (catchUp.shouldCatchUp && catchUp.catchUpFrom) {
|
||||
logger.log(
|
||||
`[${routineId}] Executing catch-up from ${catchUp.catchUpFrom}`
|
||||
);
|
||||
await this.routineRunner.execute(routine, {
|
||||
catchUpFrom: catchUp.catchUpFrom,
|
||||
});
|
||||
}
|
||||
|
||||
// Normal execution
|
||||
if (!this.routineRunner.isExecuting(routineId)) {
|
||||
logger.log(`[${routineId}] Executing routine`);
|
||||
await this.routineRunner.execute(routine);
|
||||
}
|
||||
} catch (error) {
|
||||
// Per-routine failure isolation - don't let one failure affect others
|
||||
logger.error(`[${routineId}] Failed to process: ${error}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle routine created event.
|
||||
*/
|
||||
private handleRoutineCreated = (routine: Routine): void => {
|
||||
logger.log(`[${routine.id}] Routine created, will check at next poll`);
|
||||
};
|
||||
|
||||
/**
|
||||
* Handle routine updated event.
|
||||
*/
|
||||
private handleRoutineUpdated = (routine: Routine): void => {
|
||||
logger.log(`[${routine.id}] Routine updated`);
|
||||
};
|
||||
|
||||
/**
|
||||
* Handle routine deleted event.
|
||||
*/
|
||||
private handleRoutineDeleted = (routine: Routine): void => {
|
||||
logger.log(`[${routine.id}] Routine deleted`);
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user