cron

Author: jaredLunde

openapi/v1.json

@@ -2188,4 +2188,4 @@
       "description": "Dry-run schedule expressions"
     }
   ]
-}
+}

sdk/ts/cron/ARCHITECTURE.md

@@ -106,7 +106,7 @@ Only `__cron_*` topic targets are touched. Schedules targeting queues, user-defi
 
 ### Management client
 
-`createCronClient()` / the lazy `cron` singleton provide direct schedule management without a handler or HTTP listener. All methods follow the same `cmd()` → `wrap()` → `camelize` pipeline as `@beyond.dev/queue` and `@beyond.dev/events`. `schedules.sync()` is a client-side operation: upsert all provided specs in parallel, then list all schedules and delete those not in the desired set. There is no server-side sync endpoint.
+`createCronClient()` / the lazy `cron` singleton provide direct schedule management without a handler or HTTP listener. All methods follow the same `cmd()` → `wrap()` → `camelize` pipeline as `@beyond.dev/queue` and `@beyond.dev/events`.
 
 ## State Machine
 

sdk/ts/cron/__tests__/retry.test.ts

@@ -0,0 +1,77 @@
+import { createServer } from "node:http";
+import { describe, expect, it } from "vitest";
+import { createCronClient } from "../src/client.js";
+import { findFreePort } from "./harness.js";
+
+type MockResponse = { status: number; body: string };
+
+async function withMockServer(
+  responses: MockResponse[],
+  fn: (url: string, requestCount: () => number) => Promise<void>,
+) {
+  const port = await findFreePort();
+  let count = 0;
+  const server = createServer((_, res) => {
+    const r = responses[Math.min(count++, responses.length - 1)]!;
+    res.writeHead(r.status, { "content-type": "application/json" });
+    res.end(r.body);
+  });
+  await new Promise<void>((r) => server.listen(port, "127.0.0.1", r));
+  try {
+    await fn(`http://127.0.0.1:${port}`, () => count);
+  } finally {
+    await new Promise<void>((r) => server.close(() => r()));
+  }
+}
+
+describe("buildFetch — retry behavior", () => {
+  it("retries on 5xx and returns the successful response", async () => {
+    await withMockServer(
+      [
+        {
+          status: 503,
+          body: "{\"error\":{\"code\":\"unavailable\",\"message\":\"down\"}}",
+        },
+        { status: 200, body: "{\"name\":\"t\",\"status\":\"active\"}" },
+      ],
+      async (url, requestCount) => {
+        const client = createCronClient({ url, retries: 2 });
+        const { data, error } = await client.schedules.get("t");
+        expect(requestCount()).toBe(2);
+        expect(error).toBeUndefined();
+        expect(data?.name).toBe("t");
+      },
+    );
+  });
+
+  it("returns error after exhausting retries", async () => {
+    await withMockServer(
+      [{
+        status: 503,
+        body: "{\"error\":{\"code\":\"unavailable\",\"message\":\"down\"}}",
+      }],
+      async (url, requestCount) => {
+        const client = createCronClient({ url, retries: 2 });
+        const { error } = await client.schedules.get("t");
+        expect(requestCount()).toBe(3); // 1 initial + 2 retries
+        expect(error?.status).toBe(503);
+        expect(error?.code).toBe("unavailable");
+      },
+    );
+  });
+
+  it("does not retry on 4xx errors", async () => {
+    await withMockServer(
+      [{
+        status: 404,
+        body: "{\"error\":{\"code\":\"not_found\",\"message\":\"missing\"}}",
+      }],
+      async (url, requestCount) => {
+        const client = createCronClient({ url, retries: 2 });
+        const { error } = await client.schedules.get("t");
+        expect(requestCount()).toBe(1);
+        expect(error?.status).toBe(404);
+      },
+    );
+  });
+});

sdk/ts/cron/__tests__/schedules.test.ts

@@ -179,55 +179,20 @@ describe("schedules — preview", () => {
     expect(error).toBeUndefined();
     expect(typeof data?.humanReadable).toBe("string");
   });
-});
-
-describe("schedules — sync", () => {
-  const client = cronClient();
-
-  it("sync upserts all specs and removes others", async () => {
-    const keep = uniqueName("keep");
-    const remove = uniqueName("remove");
 
-    // Seed a schedule that should be removed
-    await client.schedules.upsert({
-      name: remove,
-      every: "1h",
-      target: { queue: "test-q", message: {} },
+  it("preview returns humanReadable for fireAt one-shot", async () => {
+    const { data, error } = await client.schedules.preview({
+      fireAt: "2099-01-01T00:00:00Z",
     });
-
-    const { data, error } = await client.schedules.sync([
-      {
-        name: keep,
-        every: "1h",
-        target: { queue: "test-q", message: {} },
-      },
-    ]);
     expect(error).toBeUndefined();
-    expect(data?.upserted).toBe(1);
-    expect(data?.removed).toBeGreaterThanOrEqual(1);
-
-    const { data: all } = await client.schedules.list();
-    expect(all?.some((s) => s.name === keep)).toBe(true);
-    expect(all?.some((s) => s.name === remove)).toBe(false);
-
-    // Cleanup
-    await client.schedules.delete(keep);
+    expect(typeof data?.humanReadable).toBe("string");
   });
 });
 
 describe("schedules — observability hooks", () => {
   it("fires onRequest and onResponse for each call", async () => {
     const requests: string[] = [];
     const responses: string[] = [];
-    const c = cronClient();
-    // Wrap with hooks via a new client instance
-    const hooked = {
-      ...c,
-      schedules: {
-        ...c.schedules,
-      },
-    };
-    void hooked; // just verifying createCronClient accepts hooks
     const name = uniqueName();
     const hClient = (await import("../src/client.js")).createCronClient({
       url: process.env["QUEUE_TEST_URL"]!,
@@ -241,6 +206,8 @@ describe("schedules — observability hooks", () => {
     });
     await hClient.schedules.delete(name);
     expect(requests).toContain("schedules.upsert");
+    expect(requests).toContain("schedules.delete");
     expect(responses).toContain("schedules.upsert");
+    expect(responses).toContain("schedules.delete");
   });
 });

sdk/ts/cron/__tests__/start.test.ts

@@ -1,7 +1,53 @@
 import { afterEach, beforeEach, describe, expect, it } from "vitest";
-import { start } from "../src/client.js";
+import { schedule, start } from "../src/client.js";
 import { cronClient, findFreePort, uniqueName } from "./harness.js";
 
+describe("start() — input validation", () => {
+  const serverUrl = process.env["QUEUE_TEST_URL"]!;
+
+  it("throws for an invalid job name", async () => {
+    const job = schedule({
+      name: "Bad Name",
+      every: "1h",
+      run: async () => {},
+    });
+    await expect(start([job], { url: serverUrl })).rejects.toThrow(
+      /Invalid schedule name/,
+    );
+  });
+
+  it("throws for duplicate job names", async () => {
+    const job = schedule({ name: "my-job", every: "1h", run: async () => {} });
+    await expect(start([job, job], { url: serverUrl })).rejects.toThrow(
+      /Duplicate schedule name/,
+    );
+  });
+
+  it("throws when BEYOND_QUEUE_URL is missing and no url option provided", async () => {
+    const job = schedule({ name: "my-job", every: "1h", run: async () => {} });
+    const saved = process.env["BEYOND_QUEUE_URL"];
+    delete process.env["BEYOND_QUEUE_URL"];
+    try {
+      await expect(start([job])).rejects.toThrow(/BEYOND_QUEUE_URL/);
+    } finally {
+      if (saved !== undefined) process.env["BEYOND_QUEUE_URL"] = saved;
+    }
+  });
+
+  it("throws when BEYOND_INTERNAL_URL is missing", async () => {
+    const job = schedule({ name: "my-job", every: "1h", run: async () => {} });
+    const saved = process.env["BEYOND_INTERNAL_URL"];
+    delete process.env["BEYOND_INTERNAL_URL"];
+    try {
+      await expect(start([job], { url: serverUrl })).rejects.toThrow(
+        /BEYOND_INTERNAL_URL/,
+      );
+    } finally {
+      if (saved !== undefined) process.env["BEYOND_INTERNAL_URL"] = saved;
+    }
+  });
+});
+
 describe("start() — worker lifecycle", () => {
   const client = cronClient();
   const serverUrl = process.env["QUEUE_TEST_URL"]!;
@@ -250,6 +296,105 @@ describe("start() — worker lifecycle", () => {
     }
   });
 
+  it("passes CronContext with correct name, ISO-8601 scheduledFor, and boolean outOfBand", async () => {
+    const name = uniqueName("ctx");
+    cleanupNames.push(name);
+    const port = await findFreePort();
+    const ac = new AbortController();
+    let capturedCtx:
+      | { name: string; scheduledFor: string; outOfBand: boolean }
+      | undefined;
+
+    const done = start(
+      [{
+        name,
+        spec: { name, every: "1h" },
+        handler: async (ctx) => {
+          capturedCtx = ctx;
+        },
+      }],
+      { url: serverUrl, port, signal: ac.signal },
+    );
+
+    await new Promise<void>((r) => setTimeout(r, 200));
+
+    try {
+      const res = await fetch(`http://127.0.0.1:${port}/__cron/${name}`, {
+        method: "POST",
+        body: "{}",
+        headers: { "content-type": "application/json" },
+      });
+      expect(res.status).toBe(200);
+      expect(capturedCtx?.name).toBe(name);
+      expect(capturedCtx?.scheduledFor).toMatch(
+        /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}/,
+      );
+      expect(typeof capturedCtx?.outOfBand).toBe("boolean");
+    } finally {
+      ac.abort();
+      await done;
+    }
+  });
+
+  it("returns 405 for non-POST requests", async () => {
+    const name = uniqueName("405");
+    cleanupNames.push(name);
+    const port = await findFreePort();
+    const ac = new AbortController();
+
+    const done = start(
+      [{ name, spec: { name, every: "1h" }, handler: async () => {} }],
+      { url: serverUrl, port, signal: ac.signal },
+    );
+
+    await new Promise<void>((r) => setTimeout(r, 200));
+
+    try {
+      const res = await fetch(`http://127.0.0.1:${port}/__cron/${name}`, {
+        method: "GET",
+      });
+      expect(res.status).toBe(405);
+    } finally {
+      ac.abort();
+      await done;
+    }
+  });
+
+  it("resolves cleanly when aborted during an in-flight handler", async () => {
+    const name = uniqueName("inflight");
+    cleanupNames.push(name);
+    const port = await findFreePort();
+    const ac = new AbortController();
+    let handlerStarted = false;
+
+    const done = start(
+      [{
+        name,
+        spec: { name, every: "1h" },
+        handler: async () => {
+          handlerStarted = true;
+          await new Promise<void>((r) => setTimeout(r, 500));
+        },
+      }],
+      { url: serverUrl, port, signal: ac.signal },
+    );
+
+    await new Promise<void>((r) => setTimeout(r, 200));
+
+    // Fire the handler without awaiting the response
+    fetch(`http://127.0.0.1:${port}/__cron/${name}`, {
+      method: "POST",
+      body: "{}",
+    }).catch(() => {});
+
+    // Let the handler begin, then abort
+    await new Promise<void>((r) => setTimeout(r, 50));
+    expect(handlerStarted).toBe(true);
+
+    ac.abort();
+    await expect(done).resolves.toBeUndefined();
+  });
+
   it("shuts down cleanly on signal abort", async () => {
     const name = uniqueName("shut");
     cleanupNames.push(name);

sdk/ts/cron/src/client.ts

@@ -136,14 +136,6 @@ export interface CronClient {
     /** Create or replace a schedule (PUT semantics — idempotent). */
     upsert(spec: ScheduleSpec): CronResult<Schedule>;
 
-    /**
-     * Upsert all provided specs and delete any schedules not in the list.
-     * Always declarative — the provided list is the desired state.
-     */
-    sync(
-      specs: ScheduleSpec[],
-    ): CronResult<{ upserted: number; removed: number }>;
-
     /** Dry-run a schedule expression. No schedule is created. */
     preview(input: PreviewSpec): CronResult<Preview>;
 
@@ -552,48 +544,6 @@ export function createCronClient(opts?: CronClientOptions): CronClient {
           }),
         )),
 
-      sync: cmd("schedules.sync", async (specs) => {
-        const desired = new Map(specs.map((s) => [s.name, s]));
-
-        // Upsert all
-        await Promise.all(
-          specs.map((spec) =>
-            client.PUT("/v1/schedules/{name}", {
-              params: { path: { name: spec.name } },
-              body: snakenize(spec) as components["schemas"]["ScheduleSpec"],
-            })
-          ),
-        );
-
-        // List all and delete those not in desired set
-        const { data: all, error, response } = await client.GET(
-          "/v1/schedules",
-          {},
-        );
-        if (error) {
-          return {
-            data: undefined,
-            error: toCronError(error, response),
-            response,
-          };
-        }
-
-        const toDelete = (all ?? []).filter((s) => !desired.has(s.name));
-        await Promise.all(
-          toDelete.map((s) =>
-            client.DELETE("/v1/schedules/{name}", {
-              params: { path: { name: s.name } },
-            })
-          ),
-        );
-
-        return {
-          data: { upserted: specs.length, removed: toDelete.length },
-          error: undefined,
-          response,
-        };
-      }),
-
       preview: cmd(
         "schedules.preview",
         (input) =>