Make scheduleEvery() idempotent (#1052)

* fix: make scheduleEvery() idempotent on callback name

scheduleEvery() now deduplicates by callback name: calling it multiple
times with the same callback returns the existing schedule instead of
creating a duplicate. If the interval or payload changed, the existing
schedule is updated in place.

This fixes the common pattern of calling scheduleEvery() inside
onStart(), which runs on every Durable Object wake. Previously each
wake created a new interval schedule, leading to duplicate executions.

keepAlive() opts out of idempotency via an internal _idempotent: false
flag so multiple concurrent keepAlive calls still get independent
schedules with independent disposers.

Closes #1049

* Use (callback, interval, payload) as idempotency key for scheduleEvery()

Instead of updating the existing row when interval/payload differs,
treat a different interval or payload as a distinct schedule that
gets its own row. The idempotency key is now the full tuple of
(callback, intervalSeconds, payload) rather than just callback.

Uses SQL IS instead of = for payload comparison to correctly handle
NULL payload values (NULL IS NULL is true in SQLite).

---------

Co-authored-by: ask-bonk[bot] <ask-bonk[bot]@users.noreply.github.com>

Author: ask-bonk[bot]

.changeset/idempotent-schedule-every.md

@@ -0,0 +1,9 @@
+---
+"agents": patch
+---
+
+Make `scheduleEvery()` idempotent
+
+`scheduleEvery()` now deduplicates by the combination of callback name, interval, and payload: calling it multiple times with the same arguments returns the existing schedule instead of creating a duplicate. A different interval or payload creates a separate, independent schedule.
+
+This fixes the common pattern of calling `scheduleEvery()` inside `onStart()`, which runs on every Durable Object wake. Previously each wake created a new interval schedule, leading to a thundering herd of duplicate executions.

docs/scheduling.md

@@ -172,6 +172,46 @@ await this.scheduleEvery(45, "healthCheck", {});
 await this.scheduleEvery(90, "syncData", { destination: "warehouse" });
 ```
 
+**Idempotency:**
+
+`scheduleEvery()` is idempotent on the combination of callback name, interval, and payload — calling it multiple times with the same arguments does not create duplicate schedules. This makes it safe to call in `onStart()`, which runs on every Durable Object wake:
+
+```typescript
+class MyAgent extends Agent {
+  async onStart() {
+    // Safe: only one schedule is created, no matter how many times the DO wakes
+    await this.scheduleEvery(30, "tick");
+  }
+
+  async tick() {
+    console.log("tick", new Date().toISOString());
+  }
+}
+```
+
+Calling `scheduleEvery()` with a different interval or payload creates a separate schedule, even for the same callback:
+
+```typescript
+// First call creates one schedule
+await this.scheduleEvery(30, "poll");
+
+// Second call with a different interval creates a second schedule
+await this.scheduleEvery(60, "poll");
+// Two "poll" schedules exist: one every 30s and one every 60s
+
+// Third call with the same arguments as the first is a no-op
+await this.scheduleEvery(30, "poll");
+// Still two schedules
+```
+
+Different callbacks also get their own independent schedules:
+
+```typescript
+// These create two separate schedules (different callbacks)
+await this.scheduleEvery(30, "poll");
+await this.scheduleEvery(30, "healthCheck");
+```
+
 **Key differences from cron:**
 
 | Feature             | Cron                           | Interval               |
@@ -757,6 +797,7 @@ Schedule a task to run repeatedly at a fixed interval.
 
 **Behavior:**
 
+- **Idempotent on (callback, interval, payload)** — calling with the same callback, interval, and payload returns the existing schedule instead of creating a duplicate. A different interval or payload creates a new, independent schedule.
 - First execution occurs after `intervalSeconds` (not immediately)
 - If callback is still running when next execution is due, it's skipped (overlap prevention)
 - If callback throws an error, the interval continues

packages/agents/src/index.ts

@@ -2174,7 +2174,23 @@ export class Agent<
   }
 
   /**
-   * Schedule a task to run repeatedly at a fixed interval
+   * Schedule a task to run repeatedly at a fixed interval.
+   *
+   * This method is **idempotent** — calling it multiple times with the same
+   * `callback`, `intervalSeconds`, and `payload` returns the existing schedule
+   * instead of creating a duplicate. A different interval or payload is
+   * treated as a distinct schedule and creates a new row.
+   *
+   * This makes it safe to call in `onStart()`, which runs on every Durable
+   * Object wake:
+   *
+   * ```ts
+   * async onStart() {
+   *   // Only one schedule is created, no matter how many times the DO wakes
+   *   await this.scheduleEvery(30, "tick");
+   * }
+   * ```
+   *
    * @template T Type of the payload data
    * @param intervalSeconds Number of seconds between executions
    * @param callback Name of the method to call
@@ -2187,7 +2203,7 @@ export class Agent<
     intervalSeconds: number,
     callback: keyof this,
     payload?: T,
-    options?: { retry?: RetryOptions }
+    options?: { retry?: RetryOptions; _idempotent?: boolean }
   ): Promise<Schedule<T>> {
     // DO alarms have a max schedule time of 30 days
     const MAX_INTERVAL_SECONDS = 30 * 24 * 60 * 60; // 30 days in seconds
@@ -2214,6 +2230,46 @@ export class Agent<
       validateRetryOptions(options.retry, this._resolvedOptions.retry);
     }
 
+    const idempotent = options?._idempotent !== false;
+    const payloadJson = JSON.stringify(payload);
+
+    // Idempotency: check for an existing interval schedule with the same
+    // callback, interval, and payload. A different interval or payload is
+    // treated as a distinct schedule and gets its own row.
+    if (idempotent) {
+      const existing = this.sql<{
+        id: string;
+        callback: string;
+        payload: string;
+        type: string;
+        time: number;
+        intervalSeconds: number;
+        retry_options: string | null;
+      }>`
+        SELECT * FROM cf_agents_schedules
+        WHERE type = 'interval'
+          AND callback = ${callback}
+          AND intervalSeconds = ${intervalSeconds}
+          AND payload IS ${payloadJson}
+        LIMIT 1
+      `;
+
+      if (existing.length > 0) {
+        const row = existing[0];
+
+        // Exact match — return existing schedule as-is (no-op)
+        return {
+          callback: row.callback,
+          id: row.id,
+          intervalSeconds: row.intervalSeconds,
+          payload: JSON.parse(row.payload) as T,
+          retry: parseRetryOptions(row as unknown as Record<string, unknown>),
+          time: row.time,
+          type: "interval"
+        };
+      }
+    }
+
     const id = nanoid(9);
     const time = new Date(Date.now() + intervalSeconds * 1000);
     const timestamp = Math.floor(time.getTime() / 1000);
@@ -2222,7 +2278,7 @@ export class Agent<
 
     this.sql`
       INSERT OR REPLACE INTO cf_agents_schedules (id, callback, payload, type, intervalSeconds, time, running, retry_options)
-      VALUES (${id}, ${callback}, ${JSON.stringify(payload)}, 'interval', ${intervalSeconds}, ${timestamp}, 0, ${retryJson})
+      VALUES (${id}, ${callback}, ${payloadJson}, 'interval', ${intervalSeconds}, ${timestamp}, 0, ${retryJson})
     `;
 
     await this._scheduleNextAlarm();
@@ -2357,7 +2413,9 @@ export class Agent<
     const heartbeatSeconds = Math.ceil(KEEP_ALIVE_INTERVAL_MS / 1000);
     const schedule = await this.scheduleEvery(
       heartbeatSeconds,
-      "_cf_keepAliveHeartbeat" as keyof this
+      "_cf_keepAliveHeartbeat" as keyof this,
+      undefined,
+      { _idempotent: false }
     );
 
     let disposed = false;

packages/agents/src/tests/agents/schedule.ts

@@ -222,4 +222,52 @@ export class TestScheduleAgent extends Agent<Record<string, unknown>> {
 
     return schedule.id;
   }
+
+  // --- Idempotency test helpers ---
+
+  // A second callback for testing that idempotency is per-callback
+  secondIntervalCallback() {
+    // Intentionally empty
+  }
+
+  @callable()
+  async createIntervalScheduleWithPayload(
+    intervalSeconds: number,
+    payload: string
+  ): Promise<string> {
+    const schedule = await this.scheduleEvery(
+      intervalSeconds,
+      "intervalCallback",
+      payload
+    );
+    return schedule.id;
+  }
+
+  @callable()
+  async createSecondIntervalSchedule(intervalSeconds: number): Promise<string> {
+    const schedule = await this.scheduleEvery(
+      intervalSeconds,
+      "secondIntervalCallback"
+    );
+    return schedule.id;
+  }
+
+  @callable()
+  async countIntervalSchedules(): Promise<number> {
+    const result = this.sql<{ count: number }>`
+      SELECT COUNT(*) as count FROM cf_agents_schedules WHERE type = 'interval'
+    `;
+    return result[0].count;
+  }
+
+  @callable()
+  async countIntervalSchedulesForCallback(
+    callbackName: string
+  ): Promise<number> {
+    const result = this.sql<{ count: number }>`
+      SELECT COUNT(*) as count FROM cf_agents_schedules
+      WHERE type = 'interval' AND callback = ${callbackName}
+    `;
+    return result[0].count;
+  }
 }

packages/agents/src/tests/schedule.test.ts

@@ -257,4 +257,167 @@ describe("schedule operations", () => {
       await agentStub.cancelScheduleById(scheduleId);
     });
   });
+
+  describe("scheduleEvery idempotency", () => {
+    it("should return existing schedule when called with same callback and interval", async () => {
+      const agentStub = await getAgentByName(
+        env.TestScheduleAgent,
+        "idempotent-same-args-test"
+      );
+
+      // Create an interval schedule
+      const firstId = await agentStub.createIntervalSchedule(30);
+
+      // Call again with the same callback and interval
+      const secondId = await agentStub.createIntervalSchedule(30);
+
+      // Both calls should return the same schedule ID
+      expect(secondId).toBe(firstId);
+
+      // Only one schedule should exist
+      const count =
+        await agentStub.countIntervalSchedulesForCallback("intervalCallback");
+      expect(count).toBe(1);
+    });
+
+    it("should return existing schedule when called with same callback, interval, and payload", async () => {
+      const agentStub = await getAgentByName(
+        env.TestScheduleAgent,
+        "idempotent-same-payload-test"
+      );
+
+      // Create with payload
+      const firstId = await agentStub.createIntervalScheduleWithPayload(
+        30,
+        "hello"
+      );
+
+      // Call again with the same arguments
+      const secondId = await agentStub.createIntervalScheduleWithPayload(
+        30,
+        "hello"
+      );
+
+      // Same schedule returned
+      expect(secondId).toBe(firstId);
+
+      const count =
+        await agentStub.countIntervalSchedulesForCallback("intervalCallback");
+      expect(count).toBe(1);
+    });
+
+    it("should create a new row when interval changes for same callback", async () => {
+      const agentStub = await getAgentByName(
+        env.TestScheduleAgent,
+        "idempotent-interval-change-test"
+      );
+
+      // Create with 30s interval
+      const firstId = await agentStub.createIntervalSchedule(30);
+
+      // Call again with different interval
+      const secondId = await agentStub.createIntervalSchedule(60);
+
+      // Different interval means a different schedule
+      expect(secondId).not.toBe(firstId);
+
+      // Two schedules should exist for this callback
+      const count =
+        await agentStub.countIntervalSchedulesForCallback("intervalCallback");
+      expect(count).toBe(2);
+
+      // The new schedule should have the new interval
+      const schedule = await agentStub.getScheduleById(secondId);
+      expect(schedule).toBeDefined();
+      if (schedule?.type === "interval") {
+        expect(schedule.intervalSeconds).toBe(60);
+      }
+
+      // The original schedule should still have the old interval
+      const original = await agentStub.getScheduleById(firstId);
+      expect(original).toBeDefined();
+      if (original?.type === "interval") {
+        expect(original.intervalSeconds).toBe(30);
+      }
+    });
+
+    it("should create a new row when payload changes for same callback", async () => {
+      const agentStub = await getAgentByName(
+        env.TestScheduleAgent,
+        "idempotent-payload-change-test"
+      );
+
+      // Create with payload "foo"
+      const firstId = await agentStub.createIntervalScheduleWithPayload(
+        30,
+        "foo"
+      );
+
+      // Call again with different payload
+      const secondId = await agentStub.createIntervalScheduleWithPayload(
+        30,
+        "bar"
+      );
+
+      // Different payload means a different schedule
+      expect(secondId).not.toBe(firstId);
+
+      // Two schedules should exist for this callback
+      const count =
+        await agentStub.countIntervalSchedulesForCallback("intervalCallback");
+      expect(count).toBe(2);
+
+      // Each schedule should have its own payload
+      const first = await agentStub.getScheduleById(firstId);
+      expect(first).toBeDefined();
+      expect(first?.payload).toBe("foo");
+
+      const second = await agentStub.getScheduleById(secondId);
+      expect(second).toBeDefined();
+      expect(second?.payload).toBe("bar");
+    });
+
+    it("should allow different callbacks to have their own interval schedules", async () => {
+      const agentStub = await getAgentByName(
+        env.TestScheduleAgent,
+        "idempotent-different-callbacks-test"
+      );
+
+      // Create interval for callback A
+      const firstId = await agentStub.createIntervalSchedule(30);
+
+      // Create interval for callback B
+      const secondId = await agentStub.createSecondIntervalSchedule(30);
+
+      // Different callbacks should create different schedules
+      expect(secondId).not.toBe(firstId);
+
+      // Two interval schedules should exist total
+      const count = await agentStub.countIntervalSchedules();
+      expect(count).toBe(2);
+    });
+
+    it("should not create duplicates when called many times (simulating repeated onStart)", async () => {
+      const agentStub = await getAgentByName(
+        env.TestScheduleAgent,
+        "idempotent-repeated-calls-test"
+      );
+
+      // Simulate calling scheduleEvery in onStart many times
+      const ids: string[] = [];
+      for (let i = 0; i < 5; i++) {
+        const id = await agentStub.createIntervalSchedule(30);
+        ids.push(id);
+      }
+
+      // All IDs should be the same
+      const uniqueIds = [...new Set(ids)];
+      expect(uniqueIds.length).toBe(1);
+
+      // Only one schedule should exist
+      const count =
+        await agentStub.countIntervalSchedulesForCallback("intervalCallback");
+      expect(count).toBe(1);
+    });
+  });
 });