Read routine transitions and schedules under the owner cap lock

CHANGELOG.md

@@ -8,6 +8,12 @@ Newest first. `Unreleased` is what is on `main` and not yet tagged.
 
 ## Unreleased
 
+### Concurrent routine edits keep their limit and saved schedule
+
+An update now reads the current routine under the same owner lock used to count and write it. An
+enable queued behind a pause and another create cannot take a person past 20 enabled routines.
+Concurrent cron and timezone edits also calculate the next firing from the schedule actually saved.
+
 **Before upgrading.** Sixteen things change for an existing deployment:
 - Bots now share one computer per person instead of each having its own. Migration
   `0069_isolated_bots` adds `agent_profiles.isolated`, off for every existing Bot. With the

server/src/routines/store.ts

@@ -362,12 +362,13 @@ export function createRoutineStore(database: Database): RoutineStore {
     ownerUserId: string,
     agentId: string,
     channelId?: string,
+    handle: Database | Transaction = database,
   ): Promise<string> {
     if (channelId !== undefined) {
       // One query for all four conditions: the channel exists, it is not deleted, the owner is a
       // member, and this agent is in it. Any miss is the same refusal, because telling them apart
       // would tell a caller which channel ids exist.
-      const rows = await database
+      const rows = await handle
         .select({ id: channels.id })
         .from(channels)
         .innerJoin(
@@ -393,7 +394,7 @@ export function createRoutineStore(database: Database): RoutineStore {
 
     // Six rows is enough to answer the question: one resolves, more than one refuses, and the
     // sentence names at most five before it says "and others".
-    const candidates = await database
+    const candidates = await handle
       .select({ id: channels.id, name: channels.name })
       .from(channels)
       .innerJoin(
@@ -472,8 +473,9 @@ export function createRoutineStore(database: Database): RoutineStore {
   async function loadOwned(
     ownerUserId: string,
     id: string,
+    handle: Database | Transaction = database,
   ): Promise<RoutineRow> {
-    const [row] = await database
+    const [row] = await handle
       .select()
       .from(routines)
       .where(and(eq(routines.id, id), eq(routines.ownerUserId, ownerUserId)))
@@ -488,44 +490,45 @@ export function createRoutineStore(database: Database): RoutineStore {
     id: string,
     patch: RoutinePatch,
   ): Promise<Routine> {
-    const existing = await loadOwned(ownerUserId, id);
+    const [row] = await withEnabledCapLock(ownerUserId, async (transaction) => {
+      // A waiting update must derive its transition and schedule from the row that now holds the lock.
+      const existing = await loadOwned(ownerUserId, id, transaction);
+
+      const values: Partial<typeof routines.$inferInsert> = {
+        updatedAt: new Date(),
+      };
+      if (patch.instruction !== undefined) {
+        values.instruction = validInstruction(patch.instruction);
+      }
+      if (patch.channelId !== undefined) {
+        values.channelId = await resolveChannel(
+          ownerUserId,
+          existing.agentId,
+          patch.channelId,
+          transaction,
+        );
+      }
+      if (patch.enabled !== undefined) values.enabled = patch.enabled;
 
-    const values: Partial<typeof routines.$inferInsert> = {
-      updatedAt: new Date(),
-    };
-    if (patch.instruction !== undefined) {
-      values.instruction = validInstruction(patch.instruction);
-    }
-    if (patch.channelId !== undefined) {
-      values.channelId = await resolveChannel(
-        ownerUserId,
-        existing.agentId,
-        patch.channelId,
-      );
-    }
-    if (patch.enabled !== undefined) values.enabled = patch.enabled;
-
-    const enabling = patch.enabled === true && !existing.enabled;
-
-    const cron = patch.cron ?? existing.cron;
-    const timezone = patch.timezone ?? existing.timezone;
-    if (patch.cron !== undefined) values.cron = patch.cron;
-    if (patch.timezone !== undefined) values.timezone = timezone;
-    /*
-     * Recomputed for a new cron, a new zone, and for switching back on. That last one is the subtle
-     * case: a routine switched off in June still holds June's `next_run_at`, and enabling it without
-     * recomputing hands the sweep a firing that was due months ago.
-     */
-    if (patch.cron !== undefined || patch.timezone !== undefined || enabling) {
-      values.nextRunAt = nextRunFor(cron, timezone, new Date());
-    }
+      const enabling = patch.enabled === true && !existing.enabled;
+
+      const cron = patch.cron ?? existing.cron;
+      const timezone = patch.timezone ?? existing.timezone;
+      if (patch.cron !== undefined) values.cron = patch.cron;
+      if (patch.timezone !== undefined) values.timezone = timezone;
+      /*
+       * Recomputed for a new cron, a new zone, and for switching back on. That last one is the subtle
+       * case: a routine switched off in June still holds June's `next_run_at`, and enabling it without
+       * recomputing hands the sweep a firing that was due months ago.
+       */
+      if (
+        patch.cron !== undefined ||
+        patch.timezone !== undefined ||
+        enabling
+      ) {
+        values.nextRunAt = nextRunFor(cron, timezone, new Date());
+      }
 
-    /*
-     * The cap is re-counted inside the lock, in the same transaction as the write. Counting outside
-     * and writing inside would keep the very race the lock removes: two enables that both counted 19
-     * before either committed.
-     */
-    const [row] = await withEnabledCapLock(ownerUserId, async (transaction) => {
       if (
         enabling &&
         (await countEnabled(transaction, ownerUserId)) >= MAX_ENABLED_ROUTINES

server/tests/routines-store.integration.test.ts

@@ -1,6 +1,6 @@
 import { afterAll, afterEach, describe, expect, test } from "bun:test";
 import { randomUUID } from "node:crypto";
-import { eq } from "drizzle-orm";
+import { eq, sql } from "drizzle-orm";
 import { createAgentProfileStore } from "../src/agents/profile-store";
 import type { AgentActor } from "../src/agents/profile-types";
 import { createChannelStore } from "../src/channels/routes";
@@ -109,6 +109,19 @@ async function setUp() {
   return { owner, agentId, channel };
 }
 
+async function waitForDatabaseLock(name: string) {
+  const deadline = Date.now() + 5000;
+  while (Date.now() < deadline) {
+    const rows = await database.execute(
+      sql`select wait_event_type from pg_stat_activity where application_name = ${name} and datname = current_database()`,
+    );
+    if (rows.some((row) => row.wait_event_type === "Lock")) return;
+  }
+  throw new Error(
+    `The database did not report the expected lock wait for ${name}`,
+  );
+}
+
 /**
  * A routine is one person's standing instruction, so every method takes the owner and every
  * statement is guarded by it. A routine belonging to somebody else has to be indistinguishable from
@@ -237,6 +250,82 @@ describe("what the schedule has to be", () => {
  * talked into creating a hundred. It counts what is switched ON, so switching one off makes room.
  */
 describe("how many a person may have switched on", () => {
+  test("an enable queued behind a pause rechecks the cap after another create takes the slot", async () => {
+    const { owner, agentId, channel } = await setUp();
+    const input = {
+      ownerUserId: owner.id,
+      agentId,
+      channelId: channel.id,
+      instruction: "Summarise receipts.",
+      cron: DAILY,
+    };
+    const target = await store.create(input);
+    for (let index = 1; index < MAX_ENABLED_ROUTINES; index += 1)
+      await store.create(input);
+    const blocker = createDatabase(databaseUrl, { max: 1 });
+    const pausing = createDatabase(databaseUrl, { max: 1 });
+    const creating = createDatabase(databaseUrl, { max: 1 });
+    const enabling = createDatabase(databaseUrl, { max: 1 });
+    const names = ["pausing", "creating", "enabling"].map(
+      (name) => `${name}_${randomUUID()}`,
+    );
+    await pausing.execute(
+      sql`select set_config('application_name', ${names[0]!}, false)`,
+    );
+    await creating.execute(
+      sql`select set_config('application_name', ${names[1]!}, false)`,
+    );
+    await enabling.execute(
+      sql`select set_config('application_name', ${names[2]!}, false)`,
+    );
+    const locked = Promise.withResolvers<void>();
+    const release = Promise.withResolvers<void>();
+    const gate = blocker.transaction(async (tx) => {
+      await tx.execute(
+        sql`select pg_advisory_xact_lock(hashtext(${`routine-cap-${owner.id}`}))`,
+      );
+      locked.resolve();
+      await release.promise;
+    });
+    void gate.catch(locked.reject);
+    const operations: Promise<unknown>[] = [];
+
+    try {
+      await locked.promise;
+      operations.push(
+        createRoutineStore(pausing).setEnabled(owner.id, target.id, false),
+      );
+      await waitForDatabaseLock(names[0]!);
+      operations.push(createRoutineStore(creating).create(input));
+      await waitForDatabaseLock(names[1]!);
+      operations.push(
+        createRoutineStore(enabling).setEnabled(owner.id, target.id, true),
+      );
+      await waitForDatabaseLock(names[2]!);
+      release.resolve();
+      await gate;
+      const outcomes = await Promise.allSettled(operations);
+      expect(
+        (await store.listFor(owner.id)).filter((routine) => routine.enabled),
+      ).toHaveLength(MAX_ENABLED_ROUTINES);
+      expect(outcomes[0]?.status).toBe("fulfilled");
+      expect(outcomes[1]?.status).toBe("fulfilled");
+      const enabled = outcomes[2];
+      expect(enabled?.status).toBe("rejected");
+      if (enabled?.status === "rejected")
+        expect(enabled.reason).toBeInstanceOf(RoutineRefusedError);
+    } finally {
+      release.resolve();
+      await Promise.allSettled([gate, ...operations]);
+      await Promise.all([
+        blocker.$client.end(),
+        pausing.$client.end(),
+        creating.$client.end(),
+        enabling.$client.end(),
+      ]);
+    }
+  }, 15000);
+
   test("refuses the one past the cap, and accepts it once another is off", async () => {
     const { owner, agentId, channel } = await setUp();
     const created: string[] = [];
@@ -597,6 +686,80 @@ describe("reading a person's routines", () => {
 });
 
 describe("changing a routine", () => {
+  test("queued cron and timezone edits derive one coherent next firing", async () => {
+    const { owner, agentId, channel } = await setUp();
+    const routine = await store.create({
+      ownerUserId: owner.id,
+      agentId,
+      channelId: channel.id,
+      instruction: "Summarise receipts.",
+      cron: DAILY,
+      timezone: "UTC",
+    });
+    const blocker = createDatabase(databaseUrl, { max: 1 });
+    const changingCron = createDatabase(databaseUrl, { max: 1 });
+    const changingZone = createDatabase(databaseUrl, { max: 1 });
+    const cronName = `cron_${randomUUID()}`;
+    const zoneName = `zone_${randomUUID()}`;
+    await changingCron.execute(
+      sql`select set_config('application_name', ${cronName}, false)`,
+    );
+    await changingZone.execute(
+      sql`select set_config('application_name', ${zoneName}, false)`,
+    );
+    const locked = Promise.withResolvers<void>();
+    const release = Promise.withResolvers<void>();
+    const gate = blocker.transaction(async (tx) => {
+      await tx.execute(
+        sql`select pg_advisory_xact_lock(hashtext(${`routine-cap-${owner.id}`}))`,
+      );
+      locked.resolve();
+      await release.promise;
+    });
+    void gate.catch(locked.reject);
+    const operations: Promise<unknown>[] = [];
+    try {
+      await locked.promise;
+      const cron = createRoutineStore(changingCron).update(
+        owner.id,
+        routine.id,
+        { cron: "0 17 * * *", channelId: channel.id },
+      );
+      operations.push(cron);
+      await waitForDatabaseLock(cronName);
+      const zone = createRoutineStore(changingZone).update(
+        owner.id,
+        routine.id,
+        { timezone: "America/New_York" },
+      );
+      operations.push(zone);
+      await waitForDatabaseLock(zoneName);
+      release.resolve();
+      await gate;
+      await cron;
+      const result = await zone;
+      expect(result.cron).toBe("0 17 * * *");
+      expect(result.timezone).toBe("America/New_York");
+      expect(result.channelId).toBe(channel.id);
+      // The stored clock must mean 17:00 in its stored zone, independently checked with Intl.
+      expect(
+        new Intl.DateTimeFormat("en", {
+          timeZone: result.timezone,
+          hour: "2-digit",
+          hourCycle: "h23",
+        }).format(result.nextRunAt),
+      ).toBe("17");
+    } finally {
+      release.resolve();
+      await Promise.allSettled([gate, ...operations]);
+      await Promise.all([
+        blocker.$client.end(),
+        changingCron.$client.end(),
+        changingZone.$client.end(),
+      ]);
+    }
+  }, 15000);
+
   test("recomputes the next run when the cron changes", async () => {
     const { owner, agentId, channel } = await setUp();
     const routine = await store.create({