feat(routines): suppress scheduled ticks while project is paused (TON-2139) (#7502)
Fixes #7501 ## Thinking Path Issue checkout already rejects paused projects (`issues.ts` → 409), but that fires only *after* the routine execution issue exists. Per internal TON-1102, the pause must be honored earlier — at the scheduler tick — so no execution issue is ever created while a project is paused. The fix gates dispatch in `tickScheduledTriggers` on the due routine's `projects.pausedAt`, while preserving normal cron advancement so resume does not backfill missed firings. ## What Changed - `server/src/services/routines.ts`: - `tickScheduledTriggers` LEFT JOINs `projects` and derives `projectPaused` from `projects.pausedAt`. Routines with no project are never suppressed. - When paused: the tick is still claimed and `routineTriggers.nextRunAt` advances by a single cron step (catch-up backfill bypassed while paused — no replay on resume); `recordSuppressedScheduleRun` runs instead of `dispatchRoutineRun`. - New `recordSuppressedScheduleRun` inserts one `routine_runs` row (`source: schedule`, `status: skipped`, `failureReason: paused`, `linkedIssueId: null`, `completedAt` set), updates routine/trigger touched state, and logs a `routine.run_skipped` activity entry. - `nextResultText` gains a `skipped_paused` branch for pause-specific audit text (does not overload the live-issue `skipped` text). - `server/src/__tests__/routines-service.test.ts`: focused test proving all four acceptance criteria (no issue created while paused; one `skipped`/`paused`/null-link run; `nextRunAt` advanced; normal resume on a later unpaused tick). ## Verification ```bash pnpm --filter @paperclipai/server exec tsc --noEmit npx vitest run server/src/__tests__/routines-service.test.ts ``` - The focused test in `routines-service.test.ts` proves all four acceptance criteria: no execution issue is created while the project is paused; exactly one `routine_runs` row is written with `source: schedule` / `status: skipped` / `failureReason: paused` / `linkedIssueId: null`; the trigger's `nextRunAt` advances by a single cron step (missed firings are not backfilled); and a later tick after the project is unpaused dispatches normally. ## Risks - Catch-up backfill is intentionally bypassed while paused — missed firings are **not** replayed on resume (specified no-backfill behavior). - Scope is deliberately narrow: manual runs, webhook/API triggers, routine-level paused status, workspace runtime start/stop, `concurrencyPolicy`, and `catchUpPolicy` (beyond no-backfill) are untouched. ## Model Used claude-opus-4-8 (Paperclip CTO heartbeat) Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -1514,4 +1514,88 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => {
|
|||||||
expect(run.source).toBe("webhook");
|
expect(run.source).toBe("webhook");
|
||||||
expect(run.status).toBe("issue_created");
|
expect(run.status).toBe("issue_created");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("suppresses scheduled ticks while the routine project is paused, then resumes when unpaused", async () => {
|
||||||
|
const { companyId, projectId, routine, svc } = await seedFixture();
|
||||||
|
const { trigger } = await svc.createTrigger(
|
||||||
|
routine.id,
|
||||||
|
{
|
||||||
|
kind: "schedule",
|
||||||
|
label: "daily",
|
||||||
|
cronExpression: "0 0 * * *",
|
||||||
|
timezone: "UTC",
|
||||||
|
},
|
||||||
|
{},
|
||||||
|
);
|
||||||
|
|
||||||
|
const pastDue = new Date("2020-01-01T00:00:00.000Z");
|
||||||
|
|
||||||
|
// Pause the project and make the schedule trigger due.
|
||||||
|
await db
|
||||||
|
.update(projects)
|
||||||
|
.set({ pausedAt: new Date(), pauseReason: "manual pause" })
|
||||||
|
.where(eq(projects.id, projectId));
|
||||||
|
await db
|
||||||
|
.update(routineTriggers)
|
||||||
|
.set({ nextRunAt: pastDue })
|
||||||
|
.where(eq(routineTriggers.id, trigger.id));
|
||||||
|
|
||||||
|
const pausedResult = await svc.tickScheduledTriggers(new Date());
|
||||||
|
expect(pausedResult.triggered).toBe(0);
|
||||||
|
|
||||||
|
// No execution issue should be created while paused.
|
||||||
|
const issuesWhilePaused = await db
|
||||||
|
.select()
|
||||||
|
.from(issues)
|
||||||
|
.where(eq(issues.companyId, companyId));
|
||||||
|
expect(issuesWhilePaused).toHaveLength(0);
|
||||||
|
|
||||||
|
// One skipped routine run with pause-specific reason and no linked issue.
|
||||||
|
const skippedRuns = await db
|
||||||
|
.select()
|
||||||
|
.from(routineRuns)
|
||||||
|
.where(eq(routineRuns.routineId, routine.id));
|
||||||
|
expect(skippedRuns).toHaveLength(1);
|
||||||
|
expect(skippedRuns[0]?.status).toBe("skipped");
|
||||||
|
expect(skippedRuns[0]?.source).toBe("schedule");
|
||||||
|
expect(skippedRuns[0]?.failureReason).toBe("paused");
|
||||||
|
expect(skippedRuns[0]?.linkedIssueId).toBeNull();
|
||||||
|
expect(skippedRuns[0]?.completedAt).not.toBeNull();
|
||||||
|
|
||||||
|
// Trigger advanced past the paused firing and audit reflects the pause skip.
|
||||||
|
const pausedTrigger = await db
|
||||||
|
.select()
|
||||||
|
.from(routineTriggers)
|
||||||
|
.where(eq(routineTriggers.id, trigger.id))
|
||||||
|
.then((rows) => rows[0]);
|
||||||
|
expect(pausedTrigger?.nextRunAt).not.toBeNull();
|
||||||
|
expect(pausedTrigger!.nextRunAt!.getTime()).toBeGreaterThan(pastDue.getTime());
|
||||||
|
expect(pausedTrigger?.lastResult).toMatch(/paused/i);
|
||||||
|
|
||||||
|
// Unpause and make the trigger due again; a normal tick now creates an issue.
|
||||||
|
await db
|
||||||
|
.update(projects)
|
||||||
|
.set({ pausedAt: null, pauseReason: null })
|
||||||
|
.where(eq(projects.id, projectId));
|
||||||
|
await db
|
||||||
|
.update(routineTriggers)
|
||||||
|
.set({ nextRunAt: pastDue })
|
||||||
|
.where(eq(routineTriggers.id, trigger.id));
|
||||||
|
|
||||||
|
const resumedResult = await svc.tickScheduledTriggers(new Date());
|
||||||
|
expect(resumedResult.triggered).toBe(1);
|
||||||
|
|
||||||
|
const issuesAfterResume = await db
|
||||||
|
.select()
|
||||||
|
.from(issues)
|
||||||
|
.where(eq(issues.companyId, companyId));
|
||||||
|
expect(issuesAfterResume).toHaveLength(1);
|
||||||
|
|
||||||
|
const runsAfterResume = await db
|
||||||
|
.select()
|
||||||
|
.from(routineRuns)
|
||||||
|
.where(eq(routineRuns.routineId, routine.id));
|
||||||
|
expect(runsAfterResume).toHaveLength(2);
|
||||||
|
expect(runsAfterResume.some((run) => run.status === "issue_created")).toBe(true);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -163,6 +163,7 @@ function nextCronTickInTimeZone(expression: string, timeZone: string, after: Dat
|
|||||||
function nextResultText(status: string, issueId?: string | null) {
|
function nextResultText(status: string, issueId?: string | null) {
|
||||||
if (status === "issue_created" && issueId) return `Created execution issue ${issueId}`;
|
if (status === "issue_created" && issueId) return `Created execution issue ${issueId}`;
|
||||||
if (status === "coalesced") return "Coalesced into an existing live execution issue";
|
if (status === "coalesced") return "Coalesced into an existing live execution issue";
|
||||||
|
if (status === "skipped_paused") return "Skipped because the project is paused";
|
||||||
if (status === "skipped") return "Skipped because a live execution issue already exists";
|
if (status === "skipped") return "Skipped because a live execution issue already exists";
|
||||||
if (status === "completed") return "Execution issue completed";
|
if (status === "completed") return "Execution issue completed";
|
||||||
if (status === "failed") return "Execution failed";
|
if (status === "failed") return "Execution failed";
|
||||||
@@ -877,6 +878,66 @@ export function routineService(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Records a skipped scheduled firing without creating an execution issue. Used when the
|
||||||
|
// routine's project is paused: the tick is still claimed/advanced upstream (no backfill),
|
||||||
|
// and run history + trigger audit reflect the pause-specific skip.
|
||||||
|
async function recordSuppressedScheduleRun(input: {
|
||||||
|
routine: typeof routines.$inferSelect;
|
||||||
|
trigger: typeof routineTriggers.$inferSelect;
|
||||||
|
reason: string;
|
||||||
|
nextRunAt: Date | null;
|
||||||
|
}) {
|
||||||
|
const triggeredAt = new Date();
|
||||||
|
const run = await db.transaction(async (tx) => {
|
||||||
|
const txDb = tx as unknown as Db;
|
||||||
|
const [createdRun] = await txDb
|
||||||
|
.insert(routineRuns)
|
||||||
|
.values({
|
||||||
|
companyId: input.routine.companyId,
|
||||||
|
routineId: input.routine.id,
|
||||||
|
triggerId: input.trigger.id,
|
||||||
|
source: "schedule",
|
||||||
|
status: "skipped",
|
||||||
|
triggeredAt,
|
||||||
|
failureReason: input.reason,
|
||||||
|
completedAt: triggeredAt,
|
||||||
|
linkedIssueId: null,
|
||||||
|
routineRevisionId: input.routine.latestRevisionId,
|
||||||
|
})
|
||||||
|
.returning();
|
||||||
|
await updateRoutineTouchedState({
|
||||||
|
routineId: input.routine.id,
|
||||||
|
triggerId: input.trigger.id,
|
||||||
|
triggeredAt,
|
||||||
|
status: "skipped_paused",
|
||||||
|
nextRunAt: input.nextRunAt,
|
||||||
|
}, txDb);
|
||||||
|
return createdRun;
|
||||||
|
});
|
||||||
|
|
||||||
|
try {
|
||||||
|
await logActivity(db, {
|
||||||
|
companyId: input.routine.companyId,
|
||||||
|
actorType: "system",
|
||||||
|
actorId: "routine-scheduler",
|
||||||
|
action: "routine.run_skipped",
|
||||||
|
entityType: "routine_run",
|
||||||
|
entityId: run.id,
|
||||||
|
details: {
|
||||||
|
routineId: input.routine.id,
|
||||||
|
triggerId: input.trigger.id,
|
||||||
|
source: "schedule",
|
||||||
|
status: "skipped",
|
||||||
|
reason: input.reason,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
} catch (err) {
|
||||||
|
logger.warn({ err, routineId: input.routine.id, runId: run.id }, "failed to log skipped routine run");
|
||||||
|
}
|
||||||
|
|
||||||
|
return run;
|
||||||
|
}
|
||||||
|
|
||||||
function routineExecutionFingerprintCondition(dispatchFingerprint?: string | null) {
|
function routineExecutionFingerprintCondition(dispatchFingerprint?: string | null) {
|
||||||
if (!dispatchFingerprint) return null;
|
if (!dispatchFingerprint) return null;
|
||||||
// The "default" arm preserves coalescing against pre-migration open issues.
|
// The "default" arm preserves coalescing against pre-migration open issues.
|
||||||
@@ -2315,9 +2376,11 @@ export function routineService(
|
|||||||
.select({
|
.select({
|
||||||
trigger: routineTriggers,
|
trigger: routineTriggers,
|
||||||
routine: routines,
|
routine: routines,
|
||||||
|
projectPausedAt: projects.pausedAt,
|
||||||
})
|
})
|
||||||
.from(routineTriggers)
|
.from(routineTriggers)
|
||||||
.innerJoin(routines, eq(routineTriggers.routineId, routines.id))
|
.innerJoin(routines, eq(routineTriggers.routineId, routines.id))
|
||||||
|
.leftJoin(projects, eq(routines.projectId, projects.id))
|
||||||
.where(
|
.where(
|
||||||
and(
|
and(
|
||||||
eq(routineTriggers.kind, "schedule"),
|
eq(routineTriggers.kind, "schedule"),
|
||||||
@@ -2333,10 +2396,16 @@ export function routineService(
|
|||||||
for (const row of due) {
|
for (const row of due) {
|
||||||
if (!row.trigger.nextRunAt || !row.trigger.cronExpression || !row.trigger.timezone) continue;
|
if (!row.trigger.nextRunAt || !row.trigger.cronExpression || !row.trigger.timezone) continue;
|
||||||
|
|
||||||
|
// Suppress scheduled firings while the routine's project is paused. The tick is still
|
||||||
|
// claimed and advanced to the next single cron tick (no backfill), so resume continues
|
||||||
|
// at the next cron boundary instead of replaying missed firings. Routines with no
|
||||||
|
// project are never suppressed here.
|
||||||
|
const projectPaused = !!(row.routine.projectId && row.projectPausedAt);
|
||||||
|
|
||||||
let runCount = 1;
|
let runCount = 1;
|
||||||
let claimedNextRunAt = nextCronTickInTimeZone(row.trigger.cronExpression, row.trigger.timezone, now);
|
let claimedNextRunAt = nextCronTickInTimeZone(row.trigger.cronExpression, row.trigger.timezone, now);
|
||||||
|
|
||||||
if (row.routine.catchUpPolicy === "enqueue_missed_with_cap") {
|
if (!projectPaused && row.routine.catchUpPolicy === "enqueue_missed_with_cap") {
|
||||||
let cursor: Date | null = row.trigger.nextRunAt;
|
let cursor: Date | null = row.trigger.nextRunAt;
|
||||||
runCount = 0;
|
runCount = 0;
|
||||||
while (cursor && cursor <= now && runCount < MAX_CATCH_UP_RUNS) {
|
while (cursor && cursor <= now && runCount < MAX_CATCH_UP_RUNS) {
|
||||||
@@ -2363,6 +2432,16 @@ export function routineService(
|
|||||||
.then((rows) => rows[0] ?? null);
|
.then((rows) => rows[0] ?? null);
|
||||||
if (!claimed) continue;
|
if (!claimed) continue;
|
||||||
|
|
||||||
|
if (projectPaused) {
|
||||||
|
await recordSuppressedScheduleRun({
|
||||||
|
routine: row.routine,
|
||||||
|
trigger: row.trigger,
|
||||||
|
reason: "paused",
|
||||||
|
nextRunAt: claimedNextRunAt,
|
||||||
|
});
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
for (let i = 0; i < runCount; i += 1) {
|
for (let i = 0; i < runCount; i += 1) {
|
||||||
await dispatchRoutineRun({
|
await dispatchRoutineRun({
|
||||||
routine: row.routine,
|
routine: row.routine,
|
||||||
|
|||||||
Reference in New Issue
Block a user