Diffinery

codician/diffinerypublic

Open-source, agent-first GitHub and GitHub Actions alternative that runs on your own Cloudflare account: git hosting, pull requests, code review and drop-in GitHub Actions YAML on Workers, Durable Objects, R2 and Containers.

Scheduled workflows fire from a Durable Object alarm that catches up missed ticks (#65)

Schedules no longer depend on the cron trigger arriving: SchedulerDO ticks on its own alarm every 5 minutes, which survives deploys, and starts every schedule due since the last finished tick (up to an hour back), once each, 16 repositories at a time. The cron trigger keeps the housekeeping and re-arms the alarm. Site admins can read the scheduler's state at GET /api/v3/diffinery/admin/scheduler.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HYTSnBF9opF9G7y65UPafB
Canberk Demirci authored · GitHub committed signed1 parent 5412327commit 30ba0b370727fdec25269ada20c19ae0f8554cfeBrowse files
15 files changed+269−24
Files (15)
modifiedapps/diffinery/docs/actions/events.md+1−1View file
@@ -359,7 +359,7 @@
359359
360360 How schedules run:
361361
362-- **Granularity is 5 minutes.** Diffinery checks schedules every 5 minutes and starts every schedule that was due at any minute since the last check. A schedule for `2 * * * *` starts a few minutes after :02, and one for `* * * * *` runs once per check, not every minute. GitHub's minimum interval is also 5 minutes.
362+- **Granularity is 5 minutes.** Diffinery checks schedules every 5 minutes and starts every schedule that was due at any minute since the last check. A schedule for `2 * * * *` starts a few minutes after :02, and one for `* * * * *` runs once per check, not every minute. GitHub's minimum interval is also 5 minutes. A check that comes late is caught up by the next one (up to an hour back), and a schedule starts once however many of its minutes it covers.
363363 - **Only the default branch counts.** Schedules are read from the workflow files on the default branch, and refreshed on every push to it. A `schedule` in a workflow on another branch does nothing.
364364 - Each run uses the workflow file on the default branch at that moment, and only if it still has that exact cron string.
365365 - Schedules of archived repositories don't run.
modifiedapps/diffinery/docs/self-hosting/architecture.md+9−4View file
@@ -20,7 +20,8 @@
2020 | Agent sandboxes | Containers (`SandboxContainer`, one container per sandbox) | `SANDBOX` |
2121 | GitHub import orchestration (durable, with paged and retried steps) | Workflows (`diffinery-import`, class `ImportWorkflow`) | `IMPORT_WORKFLOW` |
2222 | The Git transfer of a GitHub import | Containers (`ImportContainer`, one container per transfer) | `IMPORTER` |
23-| Scheduled workflows, token expiry reminders, email log pruning, registry cleanup, expiry of draft imports | Cron Triggers, every 5 minutes | `scheduled` handler |
23+| Scheduled workflows | Durable Object alarm, every 5 minutes (`SchedulerDO`, one instance) | `SCHEDULER` |
24+| Token expiry reminders, email log pruning, registry cleanup, expiry of draft imports, a check that the scheduler's alarm is set | Cron Triggers, every 5 minutes | `scheduled` handler |
2425 | Live run logs and pull request updates in the browser | Durable Object WebSockets with hibernation | Through `REPO` |
2526 | Outgoing email | Email Sending | `EMAIL` |
2627 | Email-based account recovery and first admin sign-in | Cloudflare Access, on `/auth/access` only | `ACCESS_TEAM_DOMAIN`, `ACCESS_AUD` |
@@ -50,7 +51,7 @@
5051
5152 ## The Worker
5253
53-`apps/diffinery/src/index.ts` exports the Worker's `fetch`, `queue` and `scheduled` handlers, and the classes Cloudflare instantiates: `RepoDO`, `RunnerContainer`, `SandboxContainer`, `RunWorkflow`, `ImportWorkflow` and `ImportContainer`.
54+`apps/diffinery/src/index.ts` exports the Worker's `fetch`, `queue` and `scheduled` handlers, and the classes Cloudflare instantiates: `RepoDO`, `SchedulerDO`, `RunnerContainer`, `SandboxContainer`, `RunWorkflow`, `ImportWorkflow` and `ImportContainer`.
5455
5556 Every request goes through one middleware before it reaches a route:
5657
@@ -179,12 +180,16 @@
179180
180181 The Durable Object decides what each request may read or write and keeps the records (`rt_caches` and `rt_artifacts`). It applies the 10 GB cache budget per repository and the artifact quota (`ARTIFACT_QUOTA_BYTES`), evicts caches and expires artifacts. Older runner agents use their own cache and artifact endpoints under `/runner/jobs/<id>/...`. Their artifacts are recorded in `rt_artifacts` (marked `legacy`, a `.tar.gz` at the older key) so the REST API, the quota and the 90-day expiry cover them. Their caches stay in the `caches` table, with their own 10 GB budget and the same 7-day idle eviction. Limits are listed in [Caches and artifacts](/docs/reference/limits#caches-and-artifacts).
181182
183+## Workflow schedules
184+
185+`SchedulerDO` starts scheduled workflow runs. Its one instance ticks on its own Durable Object alarm every 5 minutes (on the hour, :05, :10 and so on; `SCHEDULE_TICK_MINUTES` changes the interval, 1 to 60). Each tick sets the next alarm first, reads every workflow schedule from D1 (kept in sync from the default branch on each push) for repositories that aren't archived, and starts a run for each cron expression that matched any minute since the last tick that finished, at most 16 repositories at a time. A tick that comes late, or fails, is caught up by the next one, up to an hour back. Alarms survive deploys; a cron trigger can take minutes to apply after a deploy, which is why schedules don't depend on it. The alarm is set when a repository first registers a schedule, and the cron trigger checks it on every run. Site administrators can read the scheduler's state at `GET /api/v3/diffinery/admin/scheduler`: `tick_minutes`, `last_tick_at` (when the last tick finished), `covered_until` (the last minute it covered) and `next_tick_at`.
186+
182187 ## Cron trigger
183188
184-The `scheduled` handler runs every 5 minutes (`*/5 * * * *`, GitHub's minimum schedule interval) and:
189+The `scheduled` handler runs every 5 minutes (`*/5 * * * *`) and:
185190
186191 1. Cleans up the container registry: abandoned uploads and blobs no image uses.
187-2. Reads every workflow schedule from D1 (kept in sync from the default branch on each push) for repositories that aren't archived, and starts a run for each cron expression that matched any minute of the last 5.
192+2. Sets the workflow scheduler's alarm if it isn't set.
188193 3. Sends token expiry reminders and, once a day, prunes the email log.
189194 4. Cancels draft organization imports (started in the web form, repositories never chosen) that are more than 7 days old, with the error `draft expired`, and erases their GitHub token.
190195
modifiedapps/diffinery/docs/self-hosting/configuration.md+3−1View file
@@ -32,6 +32,7 @@
3232 | `IMPORT_GHE_HOST` | Not set | No | The host name of a GitHub Enterprise Server (like `ghe.example.com`) that imports may download release assets from. Without it, imports download release assets only from github.com and its asset hosts, and assets of an Enterprise import are skipped with a warning. |
3333 | `GITHUB_OAUTH_CLIENT_ID` | Not set | No | The client ID of a GitHub OAuth app, for linking GitHub accounts with a GitHub sign-in instead of a gist. Needs `GITHUB_OAUTH_CLIENT_SECRET` too. See [GitHub imports](#github-imports). |
3434 | `IMPORT_GIT_AGENT` | Not set in production; `external` in the `dev` environment | Never in production | With `external`, an import's Git transfer doesn't start a container: it waits for an import agent started by hand (see [The dev environment](#the-dev-environment)). |
35+| `SCHEDULE_TICK_MINUTES` | Not set | No | Minutes between the workflow scheduler's checks, a whole number from 1 to 60. Default `5`, GitHub's shortest schedule interval. Anything else counts as not set. The end-to-end tests use `1`. |
3536
3637 When either `ACCESS_TEAM_DOMAIN` or `ACCESS_AUD` is empty, email sign-in at `/auth/access` is off: the page answers "Account recovery by email is not set up on this instance. Ask an administrator for a sign-in link.", and the sign-in page doesn't offer **Recover your account by email**. Passkeys, invitations and admin sign-in links keep working, but the first administrator can't sign in, because email sign-in is how an account in `BOOTSTRAP_ADMINS` is created.
3738
@@ -91,13 +92,14 @@
9192 | `MAIL` | Queue producer, with a consumer on the same Worker | `diffinery-mail`, dead-letter queue `diffinery-mail-dlq` | Email jobs. Consumer: batches of up to 20, waiting up to 2 seconds, 8 retries. Optional: without it emails are dropped with a warning in the logs. |
9293 | `EMAIL` | `send_email` (Email Sending) | Your onboarded domain | Sending email. Optional: without it every email is logged as `skipped`. |
9394 | `REPO` | Durable Object namespace | Class `RepoDO` (SQLite) | One object per repository. |
95+| `SCHEDULER` | Durable Object namespace | Class `SchedulerDO` (SQLite) | One object that starts scheduled workflow runs on its own alarm. |
9496 | `RUNNER` | Durable Object namespace with a container | Class `RunnerContainer` (SQLite) | One container per hosted workflow job. |
9597 | `SANDBOX` | Durable Object namespace with a container | Class `SandboxContainer` (SQLite) | One container per agent sandbox. |
9698 | `RUN_WORKFLOW` | Workflow | `diffinery-run`, class `RunWorkflow` | One durable instance per workflow run. |
9799 | `IMPORTER` | Durable Object namespace with a container | Class `ImportContainer` (SQLite) | One container per Git transfer of a GitHub import. |
98100 | `IMPORT_WORKFLOW` | Workflow | `diffinery-import`, class `ImportWorkflow` | One durable instance per attempt of a GitHub import. |
99101
100-The Durable Object classes are declared in two migrations (`new_sqlite_classes`): tag `v1` for `RepoDO`, `RunnerContainer` and `SandboxContainer`, and tag `v2` for `ImportContainer`. `wrangler deploy` applies a migration tag the account doesn't have yet.
102+The Durable Object classes are declared in four migrations (`new_sqlite_classes`): tag `v1` for `RepoDO`, `RunnerContainer` and `SandboxContainer`, tag `v2` for `ImportContainer`, tag `v3` for `ProjectDO` and tag `v4` for `SchedulerDO`. `wrangler deploy` applies a migration tag the account doesn't have yet.
101103
102104 ## Other Worker settings
103105
modifiedapps/diffinery/docs/self-hosting/deploy.md+1−1View file
@@ -166,7 +166,7 @@
166166
167167 ### wrangler deploy
168168
169-`pnpm exec wrangler deploy -c wrangler.deploy.jsonc` (run in `apps/diffinery`) deploys the Worker with all its bindings, and builds the runner image from `containers/runner/Dockerfile` (Ubuntu 24.04 with Node.js 22, git, git-lfs and the runner agent) for all three container classes: `RunnerContainer` (hosted jobs), `SandboxContainer` (agent sandboxes) and `ImportContainer` (the Git transfers of GitHub imports). It also applies Durable Object migrations the account doesn't have yet: tag `v1` creates `RepoDO`, `RunnerContainer` and `SandboxContainer`, and tag `v2` creates `ImportContainer`.
169+`pnpm exec wrangler deploy -c wrangler.deploy.jsonc` (run in `apps/diffinery`) deploys the Worker with all its bindings, and builds the runner image from `containers/runner/Dockerfile` (Ubuntu 24.04 with Node.js 22, git, git-lfs and the runner agent) for all three container classes: `RunnerContainer` (hosted jobs), `SandboxContainer` (agent sandboxes) and `ImportContainer` (the Git transfers of GitHub imports). It also applies Durable Object migrations the account doesn't have yet: tag `v1` creates `RepoDO`, `RunnerContainer` and `SandboxContainer`, tag `v2` creates `ImportContainer`, tag `v3` creates `ProjectDO`, and tag `v4` creates `SchedulerDO`.
170170
171171 ### scripts/ensure-secrets.sh
172172
modifiedapps/diffinery/src/app.ts+7−15View file
@@ -1,7 +1,6 @@
11 import { Hono } from "hono";
22 import { getCookie } from "hono/cookie";
33 import type { DiffineryEvent } from "@diffinery/shared";
4-import { cronMatches } from "@diffinery/workflow-yaml";
54 import type { Env } from "./env.js";
65 import { credentialFrom, resolvePrincipal, type AppVars } from "./auth/principal.js";
76 import { Directory } from "./lib/directory.js";
@@ -25,6 +24,7 @@
2524 import { packagesApi } from "./registry/api.js";
2625 import { RegistryStore } from "./registry/store.js";
2726 import { mailCron, notifyEvent } from "./mail/notify.js";
27+import { schedulerStub } from "./lib/schedules.js";
2828 import { mailQueue } from "./mail/send.js";
2929 import { recordActivity } from "./agentic/today.js";
3030 import { modelProxy } from "./agentic/proxy.js";
@@ -212,28 +212,20 @@
212212 }
213213 }
214214
215-// ---------------------------------------------------------------- cron: workflow schedules
215+// ---------------------------------------------------------------- cron: housekeeping
216216
217217 export async function scheduled(controller: ScheduledController, env: Env, ctx?: ExecutionContext): Promise<void> {
218218 // Container registry housekeeping: abandoned uploads and bytes no image uses any more.
219219 const cleanup = new RegistryStore(env).cleanup().catch((e) => console.error("registry cleanup failed", e));
220220 if (ctx) ctx.waitUntil(cleanup);
221221 else await cleanup;
222- // The trigger runs every 5 minutes (GitHub's minimum schedule interval); fire anything due in the window.
223- const end = new Date(controller.scheduledTime);
224- end.setUTCSeconds(0, 0);
225- const minutes = Array.from({ length: 5 }, (_, i) => new Date(end.getTime() - i * 60_000));
226- const { results } = await env.DB.prepare("SELECT s.repo_id, s.workflow_path, s.cron FROM schedules s JOIN repos r ON r.id = s.repo_id WHERE r.archived = 0").all<{ repo_id: string; workflow_path: string; cron: string }>();
222+ const at = new Date(controller.scheduledTime);
223+ at.setUTCSeconds(0, 0);
227224 await Promise.all([
228- ...results.map(async (s) => {
229- try {
230- if (minutes.some((m) => cronMatches(s.cron, m))) await repoStub(env, s.repo_id).scheduleTick(s.workflow_path, s.cron);
231- } catch (e) {
232- console.error("schedule failed", s.repo_id, s.workflow_path, e);
233- }
234- }),
225+ // Workflow schedules fire from the scheduler's own alarm (src/scheduler.ts); this trigger only makes sure it is set.
226+ schedulerStub(env).ensure().catch((e) => console.error("scheduler alarm check failed", e)),
235227 // Token expiry reminders and email log pruning.
236- mailCron(env, new Directory(env), end).catch((e) => console.error("mail cron failed", e)),
228+ mailCron(env, new Directory(env), at).catch((e) => console.error("mail cron failed", e)),
237229 // Expire abandoned draft bulk imports after 7 days, and null their stored GitHub token (Low 5).
238230 env.DB.prepare("UPDATE imports SET status = 'cancelled', token_ciphertext = NULL, error = 'draft expired', finished_at = ?, updated_at = ? WHERE status = 'draft' AND created_at < ?")
239231 .bind(now(), now(), new Date(Date.now() - 7 * 86400_000).toISOString())
modifiedapps/diffinery/src/env.ts+5−0View file
@@ -1,5 +1,6 @@
11 import type { DiffineryEvent } from "@diffinery/shared";
22 import type { RepoDO } from "./repo/RepoDO.js";
3+import type { SchedulerDO } from "./scheduler.js";
34 import type { ImportContainer } from "./imports/container.js";
45 import type { RunnerContainer, SandboxContainer } from "./runner/containers.js";
56 import type { MailJob } from "./mail/types.js";
@@ -10,6 +11,10 @@
1011 BUCKET: R2Bucket;
1112 EVENTS: Queue<DiffineryEvent>;
1213 REPO: DurableObjectNamespace<RepoDO>;
14+ /** Fires workflow schedules (src/scheduler.ts). */
15+ SCHEDULER: DurableObjectNamespace<SchedulerDO>;
16+ /** Minutes between scheduler ticks, 1 to 60 (default 5, GitHub's shortest schedule interval). */
17+ SCHEDULE_TICK_MINUTES?: string;
1318 RUNNER: DurableObjectNamespace<RunnerContainer>;
1419 SANDBOX: DurableObjectNamespace<SandboxContainer>;
1520 /** One per repository: its agent tasks (agentic/ProjectDO.ts). */
modifiedapps/diffinery/src/http/api.ts+6−0View file
@@ -11,6 +11,7 @@
1111 import type { CommitInfo } from "../repo/RepoDO.js";
1212 import type { QueueEntry } from "../repo/queue.js";
1313 import { imports } from "./imports.js";
14+import { schedulerStub } from "../lib/schedules.js";
1415 import { agenticApi } from "../agentic/api.js";
1516 import { auditRepo, loadRepo, refuseAgent, repoStub } from "./common.js";
1617 import { repoCreated } from "../mail/notify.js";
@@ -62,6 +63,11 @@
6263
6364 api.get("/", (c) => c.json({ name: "Diffinery", api: "v3", docs: `${c.env.PUBLIC_URL}/docs/api`, mcp: `${c.env.PUBLIC_URL}/mcp` }));
6465 api.get("/meta", (c) => c.json({ name: "Diffinery", oidc: !!c.env.OIDC_PRIVATE_KEY }));
66+/** Site admin: the workflow scheduler's state, to tell a scheduler that stopped ticking from a run that failed to start. */
67+api.get("/diffinery/admin/scheduler", async (c) => {
68+ if (!requireAccount(c).site_admin) throw new HttpError(403, "site administrators only");
69+ return c.json(await schedulerStub(c.env).status());
70+});
6571 api.get("/user", (c) => {
6672 const a = requireAccount(c);
6773 return c.json({ ...userJson(c, a), email: a.email, site_admin: !!a.site_admin });
modifiedapps/diffinery/src/index.ts+1−0View file
@@ -6,6 +6,7 @@
66 import { web } from "./web/routes.js";
77
88 export { RepoDO } from "./repo/RepoDO.js";
9+export { SchedulerDO } from "./scheduler.js";
910 export { RunnerContainer, SandboxContainer, SandboxPoolDO } from "./runner/containers.js";
1011 export { RunWorkflow } from "./runner/workflow.js";
1112 export { ImportWorkflow } from "./imports/workflow.js";
addedapps/diffinery/src/lib/schedules.ts+62−0View file
@@ -1,0 +1,62 @@
1+import { cronMatches } from "@diffinery/workflow-yaml";
2+import type { Env } from "../env.js";
3+
4+/**
5+ * Timing of workflow `schedule` triggers, kept free of platform imports so it can be unit tested.
6+ * The scheduler (src/scheduler.ts) ticks every few minutes; each tick fires the schedules due in
7+ * the minutes since the last tick that finished, so a late or missed tick is caught up by the next.
8+ */
9+
10+/** The one scheduler object (src/scheduler.ts). */
11+export const schedulerStub = (env: Env) => env.SCHEDULER.get(env.SCHEDULER.idFromName("global"));
12+
13+/** GitHub's shortest schedule interval, and the default tick. */
14+export const DEFAULT_TICK_MINUTES = 5;
15+/** A tick looks back at most this far: schedules missed for longer are dropped, as on GitHub. */
16+export const CATCH_UP_MINUTES = 60;
17+
18+const MINUTE = 60_000;
19+
20+/** Minutes between ticks: SCHEDULE_TICK_MINUTES if it's a whole number from 1 to 60, else the default. */
21+export function tickMinutes(value: string | undefined): number {
22+ const n = Number(value);
23+ return Number.isInteger(n) && n >= 1 && n <= 60 ? n : DEFAULT_TICK_MINUTES;
24+}
25+
26+/** When the tick after `now` is due: the next multiple of the tick (on the hour, :05, :10 ... for 5). */
27+export function nextTick(now: number, tick: number): number {
28+ const step = tick * MINUTE;
29+ return (Math.floor(now / step) + 1) * step;
30+}
31+
32+/**
33+ * The whole minutes (UTC) a tick at `now` covers: those after `last` (the previous finished tick)
34+ * up to now's minute, at most CATCH_UP_MINUTES of them. Without a previous tick, the tick's own window.
35+ */
36+export function tickWindow(now: number, last: number | undefined, tick: number): Date[] {
37+ const end = Math.floor(now / MINUTE) * MINUTE;
38+ const from = Math.max(last !== undefined ? Math.floor(last / MINUTE) * MINUTE + MINUTE : end - (tick - 1) * MINUTE, end - (CATCH_UP_MINUTES - 1) * MINUTE);
39+ const minutes: Date[] = [];
40+ for (let t = from; t <= end; t += MINUTE) minutes.push(new Date(t));
41+ return minutes;
42+}
43+
44+/** The schedules due in any of `minutes`, each once however many of them match. Invalid crons are skipped. */
45+export function dueSchedules<T extends { cron: string }>(rows: T[], minutes: Date[]): T[] {
46+ return rows.filter((s) => {
47+ try {
48+ return minutes.some((m) => cronMatches(s.cron, m));
49+ } catch {
50+ return false;
51+ }
52+ });
53+}
54+
55+/** Runs `fn` over `items`, at most `limit` at a time. Each call's failure is its own (`fn` should catch). */
56+export async function eachLimited<T>(items: T[], limit: number, fn: (item: T) => Promise<void>): Promise<void> {
57+ let next = 0;
58+ const worker = async () => {
59+ while (next < items.length) await fn(items[next++]!);
60+ };
61+ await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker));
62+}
modifiedapps/diffinery/src/repo/actions.ts+3−0View file
@@ -30,6 +30,7 @@
3030 import { isJobCaller } from "./token-policy.js";
3131 import { ReusableError, expandReusable, localKey, parseCallRef, type Expanded, type Source } from "./reusable.js";
3232 import { ancestors, combineResults, jobStatus, runnerLabels, type JobNode } from "./job-graph.js";
33+import { schedulerStub } from "../lib/schedules.js";
3334
3435 /** How long a sandbox token works after the sandbox starts. */
3536 export const SANDBOX_TOKEN_TTL_MS = 24 * 3600_000;
@@ -379,6 +380,8 @@
379380 const stmts = [this.env.DB.prepare("DELETE FROM schedules WHERE repo_id = ?").bind(repoId)];
380381 for (const w of workflows) for (const s of w.workflow?.on.schedule ?? []) stmts.push(this.env.DB.prepare("INSERT OR IGNORE INTO schedules (repo_id, workflow_path, cron) VALUES (?, ?, ?)").bind(repoId, w.path, s.cron));
381382 await this.env.DB.batch(stmts);
383+ // The scheduler's alarm starts with the first schedule, without waiting for the cron trigger.
384+ if (stmts.length > 1) await schedulerStub(this.env).ensure().catch((e) => console.error("scheduler alarm check failed", e));
382385 }
383386
384387 // ---------------------------------------------------------------- runs
addedapps/diffinery/src/scheduler.ts+60−0View file
@@ -1,0 +1,60 @@
1+import { DurableObject } from "cloudflare:workers";
2+import type { Env } from "./env.js";
3+import { dueSchedules, eachLimited, nextTick, tickMinutes, tickWindow } from "./lib/schedules.js";
4+
5+/** Repository objects asked to start scheduled runs at the same time. */
6+const FAN_OUT = 16;
7+
8+/**
9+ * Fires workflow `schedule` triggers. One instance ("global") ticks on its own alarm, which
10+ * survives deploys and restarts, instead of on the Worker's cron trigger: Cloudflare can take
11+ * minutes to apply a cron trigger after a deploy (every deploy re-sends it), and an invocation
12+ * that never comes, or fails, would drop that window's schedules. Each tick fires what is due
13+ * since the last tick that finished, so a late or failed tick is caught up by the next one.
14+ * The cron trigger and every repository that registers schedules only make sure the alarm is set.
15+ */
16+export class SchedulerDO extends DurableObject<Env> {
17+ /** Sets the alarm when none is due by the next tick (a first start, or a lost alarm). */
18+ async ensure(): Promise<void> {
19+ const next = nextTick(Date.now(), this.tick());
20+ const at = await this.ctx.storage.getAlarm();
21+ if (at === null || at > next) await this.ctx.storage.setAlarm(next);
22+ }
23+
24+ /** When the last tick finished, the minutes it covered up to, and when the next is due (ISO times, or null). */
25+ async status(): Promise<{ tick_minutes: number; last_tick_at: string | null; covered_until: string | null; next_tick_at: string | null }> {
26+ const [finished, last, next] = await Promise.all([this.ctx.storage.get<number>("finished"), this.ctx.storage.get<number>("last"), this.ctx.storage.getAlarm()]);
27+ const iso = (t: number | null | undefined) => (typeof t === "number" ? new Date(t).toISOString() : null);
28+ return { tick_minutes: this.tick(), last_tick_at: iso(finished), covered_until: iso(last), next_tick_at: iso(next) };
29+ }
30+
31+ override async alarm(): Promise<void> {
32+ const now = Date.now();
33+ // The next tick first, so nothing below can end the chain.
34+ await this.ctx.storage.setAlarm(nextTick(now, this.tick()));
35+ const minutes = tickWindow(now, await this.ctx.storage.get<number>("last"), this.tick());
36+ if (!minutes.length) return;
37+ try {
38+ await fireSchedules(this.env, minutes);
39+ await this.ctx.storage.put({ last: minutes[minutes.length - 1]!.getTime(), finished: Date.now() });
40+ } catch (e) {
41+ console.error("schedule tick failed; the next tick catches it up", e);
42+ }
43+ }
44+
45+ private tick(): number {
46+ return tickMinutes(this.env.SCHEDULE_TICK_MINUTES);
47+ }
48+}
49+
50+/** Starts a run for every schedule due in `minutes`. Throws only when the schedules can't be read. */
51+export async function fireSchedules(env: Env, minutes: Date[]): Promise<void> {
52+ const { results } = await env.DB.prepare("SELECT s.repo_id, s.workflow_path, s.cron FROM schedules s JOIN repos r ON r.id = s.repo_id WHERE r.archived = 0").all<{ repo_id: string; workflow_path: string; cron: string }>();
53+ await eachLimited(dueSchedules(results, minutes), FAN_OUT, async (s) => {
54+ try {
55+ await env.REPO.get(env.REPO.idFromName(s.repo_id)).scheduleTick(s.workflow_path, s.cron);
56+ } catch (e) {
57+ console.error("schedule failed", s.repo_id, s.workflow_path, e);
58+ }
59+ });
60+}
modifiedapps/diffinery/test/e2e/e2e.test.mjs+42−1View file
@@ -138,7 +138,7 @@
138138 execFileSync("pnpm", ["exec", "wrangler", "d1", "migrations", "apply", "diffinery-dev", "--local", "-e", "dev", "--persist-to", persist], { cwd: APP, stdio: "pipe", env: { ...process.env, CI: "1" } });
139139 // Until the web UI exists, run the same app through the API-only entry.
140140 const entry = existsSync(join(APP, "src/web/routes.tsx")) ? [] : ["test/e2e/entry.ts"];
141- dev = spawn("pnpm", ["exec", "wrangler", "dev", ...entry, "-e", "dev", "--ip", "127.0.0.1", "--port", String(port), "--inspector-port", String(await freePort()), "--persist-to", persist, "--var", `PUBLIC_URL:${base}`, "--var", `MASTER_KEY:${randomBytes(32).toString("base64")}`, "--show-interactive-dev-session=false"], {
141+ dev = spawn("pnpm", ["exec", "wrangler", "dev", ...entry, "-e", "dev", "--ip", "127.0.0.1", "--port", String(port), "--inspector-port", String(await freePort()), "--persist-to", persist, "--var", `PUBLIC_URL:${base}`, "--var", `MASTER_KEY:${randomBytes(32).toString("base64")}`, "--var", "SCHEDULE_TICK_MINUTES:1", "--show-interactive-dev-session=false"], {
142142 cwd: APP,
143143 env: { ...process.env, CI: "1", WRANGLER_SEND_METRICS: "false" },
144144 stdio: ["ignore", "pipe", "pipe"],
@@ -168,6 +168,23 @@
168168 if (process.env.E2E_KEEP !== "1") rmSync(work, { recursive: true, force: true });
169169 });
170170
171+// Scheduled workflows: the dev environment has no cron trigger, like a deploy whose cron trigger
172+// hasn't been applied yet, so schedules must fire from the scheduler's own alarm. The schedule is
173+// registered first and checked by the last test, so its ticks (1 minute here) come while the rest runs.
174+let tickRepo;
175+test("scheduled workflows: a push to the default branch registers the schedule", async () => {
176+ const me = await ok("GET", "/user");
177+ tickRepo = `${me.login}/schedules`;
178+ await ok("POST", "/user/repos", { name: "schedules" });
179+ const src = join(work, "schedules");
180+ mkdirSync(join(src, ".github/workflows"), { recursive: true });
181+ writeFileSync(join(src, ".github/workflows/tick.yml"), `on:\n schedule:\n - cron: "* * * * *"\njobs:\n tick:\n runs-on: [self-hosted, linux]\n steps:\n - run: echo tick\n`);
182+ git(src, ["init", "-q", "-b", "main"]);
183+ git(src, ["add", "-A"]);
184+ git(src, ["commit", "-q", "-m", "schedule"]);
185+ git(src, ["push", "-q", `${base}/${tickRepo}.git`, "main"]);
186+});
187+
171188 test("git hosting, Actions, protected PR flow and MCP", async () => {
172189 const me = await ok("GET", "/user");
173190 const owner = me.login;
@@ -2584,3 +2601,27 @@
25842601 assert.match(await (await fetch(`${base}/${owner}/imported-app/issues/3`)).text(), /alice/);
25852602 await gh.close();
25862603 });
2604+
2605+test("scheduled workflows: the schedule fires without a cron trigger", async () => {
2606+ assert.ok(tickRepo, "the schedule was registered");
2607+ const runs = await until("a scheduled run", async () => {
2608+ const r = await ok("GET", `/repos/${tickRepo}/actions/runs?event=schedule`);
2609+ return r.total_count > 0 && r.workflow_runs;
2610+ }, 150_000);
2611+ assert.equal(runs[0].event, "schedule");
2612+ assert.equal(runs[0].path, ".github/workflows/tick.yml");
2613+ assert.equal(runs[0].actor.login, "diffinery-scheduler");
2614+ // One run per tick, never two for the same minute.
2615+ const minutes = runs.map((r) => r.created_at.slice(0, 16));
2616+ assert.equal(new Set(minutes).size, minutes.length, `two runs in one tick: ${minutes.join(", ")}`);
2617+ // Site admins can see the scheduler's state (the live test reports it when no run comes).
2618+ const state = await ok("GET", "/diffinery/admin/scheduler");
2619+ assert.equal(state.tick_minutes, 1);
2620+ assert.ok(state.last_tick_at && state.covered_until, JSON.stringify(state));
2621+ assert.ok(Date.parse(state.next_tick_at) > Date.parse(state.last_tick_at), JSON.stringify(state));
2622+ // A push without the schedule removes it.
2623+ const src = join(work, "schedules");
2624+ rmSync(join(src, ".github/workflows/tick.yml"));
2625+ git(src, ["commit", "-q", "-am", "no schedule"]);
2626+ git(src, ["push", "-q", `${base}/${tickRepo}.git`, "main"]);
2627+});
modifiedapps/diffinery/test/e2e/entry.ts+1−0View file
@@ -5,6 +5,7 @@
55 import { app, queue, scheduled } from "../../src/app.js";
66
77 export { RepoDO } from "../../src/repo/RepoDO.js";
8+export { SchedulerDO } from "../../src/scheduler.js";
89 export { RunnerContainer, SandboxContainer } from "../../src/runner/containers.js";
910 export { RunWorkflow } from "../../src/runner/workflow.js";
1011 export { ImportWorkflow } from "../../src/imports/workflow.js";
addedapps/diffinery/test/unit/schedules.test.ts+62−0View file
@@ -1,0 +1,62 @@
1+import { describe, expect, it } from "vitest";
2+import { CATCH_UP_MINUTES, dueSchedules, eachLimited, nextTick, tickMinutes, tickWindow } from "../../src/lib/schedules.js";
3+
4+const t = (iso: string) => Date.parse(iso);
5+const iso = (ds: Date[]) => ds.map((d) => d.toISOString().slice(11, 16));
6+
7+describe("workflow schedules", () => {
8+ it("ticks on multiples of the tick length", () => {
9+ expect(new Date(nextTick(t("2026-10-13T16:05:45Z"), 5)).toISOString()).toBe("2026-10-13T16:10:00.000Z");
10+ expect(new Date(nextTick(t("2026-10-13T16:10:00Z"), 5)).toISOString()).toBe("2026-10-13T16:15:00.000Z");
11+ expect(new Date(nextTick(t("2026-10-13T16:10:30Z"), 1)).toISOString()).toBe("2026-10-13T16:11:00.000Z");
12+ expect(tickMinutes(undefined)).toBe(5);
13+ expect(tickMinutes("1")).toBe(1);
14+ expect(tickMinutes("0")).toBe(5);
15+ expect(tickMinutes("soon")).toBe(5);
16+ });
17+
18+ it("a first tick covers its own window", () => {
19+ expect(iso(tickWindow(t("2026-10-13T16:10:00.250Z"), undefined, 5))).toEqual(["16:06", "16:07", "16:08", "16:09", "16:10"]);
20+ });
21+
22+ it("a tick covers the minutes since the last finished tick", () => {
23+ expect(iso(tickWindow(t("2026-10-13T16:15:00Z"), t("2026-10-13T16:10:00Z"), 5))).toEqual(["16:11", "16:12", "16:13", "16:14", "16:15"]);
24+ // An early alarm leaves nothing behind: the next tick starts where this one stopped.
25+ expect(iso(tickWindow(t("2026-10-13T16:14:59.900Z"), t("2026-10-13T16:10:00Z"), 5))).toEqual(["16:11", "16:12", "16:13", "16:14"]);
26+ expect(tickWindow(t("2026-10-13T16:10:00Z"), t("2026-10-13T16:10:00Z"), 5)).toEqual([]);
27+ });
28+
29+ it("catches up ticks that never came (the 2026-10-13 live test: none from 16:10 to 16:20)", () => {
30+ const rows = [{ repo: "live", cron: "*/5 * * * *" }, { repo: "nightly", cron: "0 3 * * *" }];
31+ // The 16:25 tick is the first since 16:05: it covers every minute since then, and the run starts once.
32+ const minutes = tickWindow(t("2026-10-13T16:25:00Z"), t("2026-10-13T16:05:00Z"), 5);
33+ expect(minutes).toHaveLength(20);
34+ expect(dueSchedules(rows, minutes)).toEqual([rows[0]]);
35+ });
36+
37+ it("looks back at most an hour", () => {
38+ const minutes = tickWindow(t("2026-10-13T18:00:00Z"), t("2026-10-13T10:00:00Z"), 5);
39+ expect(minutes).toHaveLength(CATCH_UP_MINUTES);
40+ expect(iso(minutes).slice(0, 1)).toEqual(["17:01"]);
41+ });
42+
43+ it("skips crons that don't parse instead of failing the tick", () => {
44+ const rows = [{ cron: "not a cron" }, { cron: "* * * * *" }];
45+ expect(dueSchedules(rows, [new Date(t("2026-10-13T16:10:00Z"))])).toEqual([rows[1]]);
46+ });
47+
48+ it("starts runs a few at a time", async () => {
49+ let running = 0;
50+ let peak = 0;
51+ const done: number[] = [];
52+ await eachLimited([1, 2, 3, 4, 5, 6, 7], 3, async (n) => {
53+ running++;
54+ peak = Math.max(peak, running);
55+ await new Promise((r) => setTimeout(r, 5));
56+ done.push(n);
57+ running--;
58+ });
59+ expect(peak).toBe(3);
60+ expect(done.sort()).toEqual([1, 2, 3, 4, 5, 6, 7]);
61+ });
62+});
modifiedapps/diffinery/wrangler.jsonc+6−1View file
@@ -43,6 +43,7 @@
4343 "durable_objects": {
4444 "bindings": [
4545 { "name": "REPO", "class_name": "RepoDO" },
46+ { "name": "SCHEDULER", "class_name": "SchedulerDO" },
4647 { "name": "RUNNER", "class_name": "RunnerContainer" },
4748 { "name": "SANDBOX", "class_name": "SandboxContainer" },
4849 { "name": "IMPORTER", "class_name": "ImportContainer" },
@@ -53,7 +54,9 @@
5354 { "tag": "v1", "new_sqlite_classes": ["RepoDO", "RunnerContainer", "SandboxContainer"] },
5455 { "tag": "v2", "new_sqlite_classes": ["ImportContainer"] },
5556 // Agent tasks, one Durable Object per repository (agentic/ProjectDO.ts).
56- { "tag": "v3", "new_sqlite_classes": ["ProjectDO"] }
57+ { "tag": "v3", "new_sqlite_classes": ["ProjectDO"] },
58+ // Workflow schedules (scheduler.ts).
59+ { "tag": "v4", "new_sqlite_classes": ["SchedulerDO"] }
5760 ],
5861 "workflows": [
5962 { "name": "diffinery-run", "binding": "RUN_WORKFLOW", "class_name": "RunWorkflow" },
@@ -65,6 +68,7 @@
6568 // GitHub imports: a mirror clone of a large repository needs disk and memory.
6669 { "class_name": "ImportContainer", "image": "../../containers/runner/Dockerfile", "image_build_context": "../../containers/runner", "instance_type": "standard-4", "max_instances": 20, "rollout_active_grace_period": 3600 }
6770 ],
71+ // Housekeeping, and a check that the workflow scheduler's alarm is set (src/scheduler.ts).
6872 "triggers": { "crons": ["*/5 * * * *"] },
6973
7074 "env": {
@@ -99,6 +103,7 @@
99103 "durable_objects": {
100104 "bindings": [
101105 { "name": "REPO", "class_name": "RepoDO" },
106+ { "name": "SCHEDULER", "class_name": "SchedulerDO" },
102107 { "name": "RUNNER", "class_name": "RunnerContainer" },
103108 // Sandboxes run in an `agent.mjs --sandbox-pool` process (SANDBOX_POOL_URL), not containers.
104109 { "name": "SANDBOX", "class_name": "SandboxPoolDO" },