fix(routines): coalesce sub-hourly catch-up runs (#9649)
## Thinking Path > - Paperclip is the open source control plane people use to run AI-agent companies. > - Scheduled routines support catch-up policies when the server resumes after missed cron ticks. > - The existing capped replay policy dispatched once per missed tick, which can flood the board after downtime for frequent schedules. > - Sub-hourly routines usually need one prompt catch-up execution rather than historical per-tick replay, while hourly-or-slower schedules may rely on the existing behavior. > - This pull request coalesces missed sub-hourly ticks into one execution and keeps the slower-schedule behavior unchanged. > - The benefit is bounded recovery work without changing the semantics of lower-frequency scheduled routines. ## Linked Issues or Issue Description ### What happened? When a scheduled routine using `enqueue_missed_with_cap` resumes after several missed sub-hourly cron ticks, Paperclip dispatches one catch-up execution for every missed tick. Those executions arrive in a same-second burst and can flood the board with duplicate-looking work. ### Expected behavior Sub-hourly schedules should advance past all missed ticks but dispatch exactly one catch-up execution. Hourly-or-slower schedules should retain capped per-tick replay. ### Steps to reproduce 1. Build Paperclip from `master` and create a routine with a sub-hourly cron schedule and `catchUpPolicy: enqueue_missed_with_cap`. 2. Set its persisted `nextRunAt` far enough in the past to cover several scheduled occurrences. 3. Run routine catch-up processing. 4. Observe multiple catch-up dispatches instead of one coalesced execution. ### Paperclip version or commit Reproduced on `master` before this PR. ### Deployment mode Built from source in local development with embedded PGlite. ## What Changed - Classify sub-hourly cadence from timezone-aware scheduled occurrences, avoiding daily multi-minute false positives while supporting schedules restricted to active days. - Coalesce all missed sub-hourly ticks into one catch-up dispatch while advancing `nextRunAt` to the next future occurrence. - Preserve capped per-tick replay for hourly-or-slower schedules. - Clarify the catch-up policy labels in both routine editing surfaces. - Add regression coverage for both the coalesced and preserved behaviors. ## Verification - `pnpm exec vitest run server/src/__tests__/routines-service.test.ts --testNamePattern='coalesces multiple missed sub-hourly ticks|continues replaying each missed hourly tick|continues replaying missed ticks for daily schedules with multiple minute values|coalesces sub-hourly schedules restricted to weekdays'` — 4 passed. - `pnpm check:token-gates` — all gates clean. - `git diff --check origin/master...HEAD` — clean. ## Risks - Low-to-moderate behavioral risk: sub-hourly routines using `enqueue_missed_with_cap` now intentionally receive one recovery execution instead of one per missed tick. - Hourly-or-slower schedules retain their previous capped replay behavior, limiting the compatibility surface. - No schema, migration, workflow, or lockfile changes. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex CLI with GPT-5.5, medium reasoning, code execution and repository tool use; the runtime did not expose a context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
parent
bd7c0d5f83
commit
8368fb30b0
|
|
@ -2059,6 +2059,100 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => {
|
|||
expect(newRuns).toMatchObject([{ status: "issue_created" }]);
|
||||
});
|
||||
|
||||
it("coalesces multiple missed sub-hourly ticks into one catch-up run", async () => {
|
||||
const { routine, svc } = await seedFixture();
|
||||
await db.update(routines).set({
|
||||
catchUpPolicy: "enqueue_missed_with_cap",
|
||||
}).where(eq(routines.id, routine.id));
|
||||
const { trigger } = await svc.createTrigger(routine.id, {
|
||||
kind: "schedule",
|
||||
cronExpression: "*/10 * * * *",
|
||||
timezone: "UTC",
|
||||
}, {});
|
||||
await db.update(routineTriggers).set({
|
||||
nextRunAt: new Date("2026-07-16T00:00:00.000Z"),
|
||||
}).where(eq(routineTriggers.id, trigger.id));
|
||||
|
||||
expect(await svc.tickScheduledTriggers(new Date("2026-07-16T01:05:00.000Z"))).toEqual({ triggered: 1 });
|
||||
|
||||
const runs = await db.select().from(routineRuns).where(eq(routineRuns.routineId, routine.id));
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0]?.status).toBe("issue_created");
|
||||
const updatedTrigger = await db.select().from(routineTriggers).where(eq(routineTriggers.id, trigger.id)).then((rows) => rows[0]);
|
||||
expect(updatedTrigger?.nextRunAt).toEqual(new Date("2026-07-16T01:10:00.000Z"));
|
||||
});
|
||||
|
||||
it("continues replaying each missed hourly tick", async () => {
|
||||
const { routine, svc } = await seedFixture();
|
||||
await db.update(routines).set({
|
||||
catchUpPolicy: "enqueue_missed_with_cap",
|
||||
}).where(eq(routines.id, routine.id));
|
||||
const { trigger } = await svc.createTrigger(routine.id, {
|
||||
kind: "schedule",
|
||||
cronExpression: "0 * * * *",
|
||||
timezone: "UTC",
|
||||
}, {});
|
||||
await db.update(routineTriggers).set({
|
||||
nextRunAt: new Date("2026-07-16T00:00:00.000Z"),
|
||||
}).where(eq(routineTriggers.id, trigger.id));
|
||||
|
||||
expect(await svc.tickScheduledTriggers(new Date("2026-07-16T02:30:00.000Z"))).toEqual({ triggered: 3 });
|
||||
|
||||
const runs = await db.select().from(routineRuns).where(eq(routineRuns.routineId, routine.id));
|
||||
expect(runs).toHaveLength(3);
|
||||
expect(runs.filter((run) => run.status === "issue_created")).toHaveLength(1);
|
||||
expect(runs.filter((run) => run.status === "coalesced")).toHaveLength(2);
|
||||
const updatedTrigger = await db.select().from(routineTriggers).where(eq(routineTriggers.id, trigger.id)).then((rows) => rows[0]);
|
||||
expect(updatedTrigger?.nextRunAt).toEqual(new Date("2026-07-16T03:00:00.000Z"));
|
||||
});
|
||||
|
||||
it("continues replaying missed ticks for daily schedules with multiple minute values", async () => {
|
||||
const { routine, svc } = await seedFixture();
|
||||
await db.update(routines).set({
|
||||
catchUpPolicy: "enqueue_missed_with_cap",
|
||||
}).where(eq(routines.id, routine.id));
|
||||
const { trigger } = await svc.createTrigger(routine.id, {
|
||||
kind: "schedule",
|
||||
cronExpression: "0,30 9 * * *",
|
||||
timezone: "UTC",
|
||||
}, {});
|
||||
await db.update(routineTriggers).set({
|
||||
nextRunAt: new Date("2026-07-14T09:00:00.000Z"),
|
||||
}).where(eq(routineTriggers.id, trigger.id));
|
||||
|
||||
expect(await svc.tickScheduledTriggers(new Date("2026-07-15T10:00:00.000Z"))).toEqual({ triggered: 4 });
|
||||
|
||||
const runs = await db.select().from(routineRuns).where(eq(routineRuns.routineId, routine.id));
|
||||
expect(runs).toHaveLength(4);
|
||||
expect(runs.filter((run) => run.status === "issue_created")).toHaveLength(1);
|
||||
expect(runs.filter((run) => run.status === "coalesced")).toHaveLength(3);
|
||||
const updatedTrigger = await db.select().from(routineTriggers).where(eq(routineTriggers.id, trigger.id)).then((rows) => rows[0]);
|
||||
expect(updatedTrigger?.nextRunAt).toEqual(new Date("2026-07-16T09:00:00.000Z"));
|
||||
});
|
||||
|
||||
it("coalesces sub-hourly schedules restricted to weekdays", async () => {
|
||||
const { routine, svc } = await seedFixture();
|
||||
await db.update(routines).set({
|
||||
catchUpPolicy: "enqueue_missed_with_cap",
|
||||
}).where(eq(routines.id, routine.id));
|
||||
const { trigger } = await svc.createTrigger(routine.id, {
|
||||
kind: "schedule",
|
||||
cronExpression: "*/10 * * * 1-5",
|
||||
timezone: "UTC",
|
||||
}, {});
|
||||
await db.update(routineTriggers).set({
|
||||
nextRunAt: new Date("2026-07-13T00:00:00.000Z"),
|
||||
}).where(eq(routineTriggers.id, trigger.id));
|
||||
|
||||
expect(await svc.tickScheduledTriggers(new Date("2026-07-13T01:05:00.000Z"))).toEqual({ triggered: 1 });
|
||||
|
||||
const runs = await db.select().from(routineRuns).where(eq(routineRuns.routineId, routine.id));
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0]?.status).toBe("issue_created");
|
||||
const updatedTrigger = await db.select().from(routineTriggers).where(eq(routineTriggers.id, trigger.id)).then((rows) => rows[0]);
|
||||
expect(updatedTrigger?.nextRunAt).toEqual(new Date("2026-07-13T01:10:00.000Z"));
|
||||
});
|
||||
|
||||
it("applies the armed cutoff to webhook dispatch but not manual API runs", async () => {
|
||||
const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "worktree-routines-test" };
|
||||
const { routine, svc } = await seedFixture({ runtimeEnv });
|
||||
|
|
|
|||
|
|
@ -239,6 +239,22 @@ export function nextCronTickInTimeZone(expression: string, timeZone: string, aft
|
|||
return null;
|
||||
}
|
||||
|
||||
function isSubHourlyCronExpression(expression: string, timeZone: string, after: Date) {
|
||||
const firstTick = nextCronTickInTimeZone(expression, timeZone, after);
|
||||
if (!firstTick) return false;
|
||||
|
||||
const windowEnd = firstTick.getTime() + 24 * 60 * 60 * 1000;
|
||||
let occurrenceCount = 1;
|
||||
let cursor = firstTick;
|
||||
while (occurrenceCount <= 24) {
|
||||
const nextTick = nextCronTickInTimeZone(expression, timeZone, cursor);
|
||||
if (!nextTick || nextTick.getTime() >= windowEnd) return false;
|
||||
occurrenceCount += 1;
|
||||
cursor = nextTick;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
function nextResultText(status: string, issueId?: string | null) {
|
||||
if (status === "issue_created" && issueId) return `Created execution issue ${issueId}`;
|
||||
if (status === "coalesced") return "Coalesced into an existing live execution issue";
|
||||
|
|
@ -1594,6 +1610,7 @@ export function routineService(
|
|||
executionWorkspacePreference?: string | null;
|
||||
executionWorkspaceSettings?: Record<string, unknown> | null;
|
||||
descriptionAppendix?: string | null;
|
||||
nextRunAtOverride?: Date | null;
|
||||
actor?: Actor;
|
||||
}) {
|
||||
const projectId = input.projectId ?? input.routine.projectId ?? null;
|
||||
|
|
@ -1718,9 +1735,11 @@ export function routineService(
|
|||
})
|
||||
.returning();
|
||||
|
||||
const nextRunAt = input.trigger?.kind === "schedule" && input.trigger.cronExpression && input.trigger.timezone
|
||||
? nextCronTickInTimeZone(input.trigger.cronExpression, input.trigger.timezone, triggeredAt)
|
||||
: undefined;
|
||||
const nextRunAt = input.nextRunAtOverride !== undefined
|
||||
? input.nextRunAtOverride
|
||||
: input.trigger?.kind === "schedule" && input.trigger.cronExpression && input.trigger.timezone
|
||||
? nextCronTickInTimeZone(input.trigger.cronExpression, input.trigger.timezone, triggeredAt)
|
||||
: undefined;
|
||||
|
||||
let createdIssue: Awaited<ReturnType<typeof issueSvc.create>> | null = null;
|
||||
try {
|
||||
|
|
@ -2936,12 +2955,16 @@ export function routineService(
|
|||
let claimedNextRunAt = nextCronTickInTimeZone(row.trigger.cronExpression, row.trigger.timezone, now);
|
||||
|
||||
if (!projectPaused && !worktreeSuppressed && row.routine.catchUpPolicy === "enqueue_missed_with_cap") {
|
||||
let cursor: Date | null = row.trigger.nextRunAt;
|
||||
runCount = 0;
|
||||
while (cursor && cursor <= now && runCount < MAX_CATCH_UP_RUNS) {
|
||||
runCount += 1;
|
||||
claimedNextRunAt = nextCronTickInTimeZone(row.trigger.cronExpression, row.trigger.timezone, cursor);
|
||||
cursor = claimedNextRunAt;
|
||||
if (isSubHourlyCronExpression(row.trigger.cronExpression, row.trigger.timezone, now)) {
|
||||
claimedNextRunAt = nextCronTickInTimeZone(row.trigger.cronExpression, row.trigger.timezone, now);
|
||||
} else {
|
||||
let cursor: Date | null = row.trigger.nextRunAt;
|
||||
runCount = 0;
|
||||
while (cursor && cursor <= now && runCount < MAX_CATCH_UP_RUNS) {
|
||||
runCount += 1;
|
||||
claimedNextRunAt = nextCronTickInTimeZone(row.trigger.cronExpression, row.trigger.timezone, cursor);
|
||||
cursor = claimedNextRunAt;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -2999,6 +3022,7 @@ export function routineService(
|
|||
routine: row.routine,
|
||||
trigger: row.trigger,
|
||||
source: "schedule",
|
||||
nextRunAtOverride: claimedNextRunAt,
|
||||
});
|
||||
triggered += 1;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ const catchUpPolicyOptions = [
|
|||
{
|
||||
value: "enqueue_missed_with_cap",
|
||||
title: "Enqueue missed with cap",
|
||||
description: "Catch up missed schedule windows in capped batches after recovery.",
|
||||
description: "Catch up missed schedule windows after recovery; sub-hourly schedules are combined into one catch-up run, slower schedules replay each missed window up to a cap.",
|
||||
},
|
||||
];
|
||||
|
||||
|
|
|
|||
|
|
@ -56,7 +56,7 @@ const concurrencyPolicyDescriptions: Record<string, string> = {
|
|||
};
|
||||
const catchUpPolicyDescriptions: Record<string, string> = {
|
||||
skip_missed: "Ignore windows that were missed while the scheduler or routine was paused.",
|
||||
enqueue_missed_with_cap: "Catch up missed schedule windows in capped batches after recovery.",
|
||||
enqueue_missed_with_cap: "Catch up missed schedule windows after recovery; sub-hourly schedules are combined into one catch-up run, slower schedules replay each missed window up to a cap.",
|
||||
};
|
||||
|
||||
function autoResizeTextarea(element: HTMLTextAreaElement | null) {
|
||||
|
|
|
|||
Loading…
Reference in New Issue