From 5def6bc7562973f79a4a44c7c5265b947de05e3a Mon Sep 17 00:00:00 2001 From: Rudy Celekli Date: Sat, 31 Oct 2026 02:01:51 -0400 Subject: [PATCH] Read routine transitions and schedules under the owner cap lock Implemented and locally validated with OpenAI Codex assistance. --- CHANGELOG.md | 6 + server/src/routines/store.ts | 81 ++++----- .../tests/routines-store.integration.test.ts | 165 +++++++++++++++++- 3 files changed, 212 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f9d3f16f4..3366ba01f3 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/server/src/routines/store.ts b/server/src/routines/store.ts index 1c8bf9da0f..3eda3515fa 100644 --- a/server/src/routines/store.ts +++ b/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 { 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 { - 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 { - 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 = { + 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 = { - 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 diff --git a/server/tests/routines-store.integration.test.ts b/server/tests/routines-store.integration.test.ts index 489b5f8964..ea2e9cbc6e 100644 --- a/server/tests/routines-store.integration.test.ts +++ b/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(); + const release = Promise.withResolvers(); + 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[] = []; + + 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(); + const release = Promise.withResolvers(); + 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[] = []; + 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({