diff --git a/apps/sim/app/workspace/[workspaceId]/home/components/message-content/components/special-tags/special-tags.tsx b/apps/sim/app/workspace/[workspaceId]/home/components/message-content/components/special-tags/special-tags.tsx index 8c361df0f07..366d0c2c192 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/components/message-content/components/special-tags/special-tags.tsx +++ b/apps/sim/app/workspace/[workspaceId]/home/components/message-content/components/special-tags/special-tags.tsx @@ -3478,8 +3478,12 @@ function JobRunBudgetDisplay({ message }: JobRunBudgetDisplayProps) { ) } +interface PlanUsageLimitDisplayProps { + data: UsageUpgradeTagData +} + /** The payer's plan limit: the card links to where the plan or the limit can change. */ -function PlanUsageLimitDisplay({ data }: { data: UsageUpgradeTagData }) { +function PlanUsageLimitDisplay({ data }: PlanUsageLimitDisplayProps) { const { data: session } = useSession() const hostContext = useOptionalWorkspaceHostContext() const organizationContext = useOptionalOrganizationContext() diff --git a/apps/sim/lib/mothership/jobs/brief.ts b/apps/sim/lib/mothership/jobs/brief.ts index 90abda091bf..bad6861c24e 100644 --- a/apps/sim/lib/mothership/jobs/brief.ts +++ b/apps/sim/lib/mothership/jobs/brief.ts @@ -9,7 +9,7 @@ const BRIEF_SUMMARY_CHARS = 400 type JobBriefSchedule = Pick< JobRun, - 'scheduleId' | 'jobTitle' | 'prompt' | 'successCondition' | 'lifecycle' | 'maxRuns' | 'runCount' + 'scheduleId' | 'jobTitle' | 'prompt' | 'successCondition' | 'lifecycle' | 'maxRuns' | 'runNumber' > /** @@ -20,7 +20,7 @@ type JobBriefSchedule = Pick< export async function buildJobBrief( schedule: JobBriefSchedule, occurrenceAt: Date, - runBudgetUsd: number + runBudgetUsd: number | null ): Promise { const recent = await db .select({ @@ -40,8 +40,8 @@ export async function buildJobBrief( .limit(BRIEF_RECENT_RUNS) const run = schedule.maxRuns - ? `Run ${schedule.runCount} of ${schedule.maxRuns}` - : `Run ${schedule.runCount}` + ? `Run ${schedule.runNumber} of ${schedule.maxRuns}` + : `Run ${schedule.runNumber}` return [ `Job: ${schedule.jobTitle ?? 'Untitled job'} (${run}, scheduled for ${occurrenceAt.toISOString()})`, `Task:\n${schedule.prompt ?? ''}`, @@ -59,7 +59,7 @@ export async function buildJobBrief( ) .join('\n')}` : 'This is the first run.', - `This run may spend up to $${runBudgetUsd.toFixed(2)}.`, + runBudgetUsd === null ? '' : `This run may spend up to $${runBudgetUsd.toFixed(2)}.`, ] .filter((part) => part.length > 0) .join('\n\n') diff --git a/apps/sim/lib/mothership/jobs/claim.ts b/apps/sim/lib/mothership/jobs/claim.ts index 39e8011b48a..e183b1a1108 100644 --- a/apps/sim/lib/mothership/jobs/claim.ts +++ b/apps/sim/lib/mothership/jobs/claim.ts @@ -84,6 +84,8 @@ export async function claimDueJobOccurrences( status: 'queued', trigger: JOB_RUN_TRIGGER, startedAt: now, + /** Fixed at claim: the schedule's count moves on as later occurrences are claimed. */ + executionData: { runNumber: runCount }, }) .onConflictDoNothing() await tx diff --git a/apps/sim/lib/mothership/jobs/dispatch.ts b/apps/sim/lib/mothership/jobs/dispatch.ts index cb4683ad478..bbbc344697f 100644 --- a/apps/sim/lib/mothership/jobs/dispatch.ts +++ b/apps/sim/lib/mothership/jobs/dispatch.ts @@ -1,10 +1,12 @@ import { createLogger } from '@sim/logger' -import { getErrorMessage } from '@sim/utils/errors' +import { getErrorMessage, toError } from '@sim/utils/errors' +import { generateShortId } from '@sim/utils/id' import { checkAttributedUsageLimits, resolveBillingAttribution, } from '@/lib/billing/core/attribution' import { isHosted } from '@/lib/core/config/env-flags' +import { acquireLock, releaseLock } from '@/lib/core/config/redis' import { asOrchestrationError } from '@/lib/core/orchestration/types' import { buildJobBrief } from '@/lib/mothership/jobs/brief' import { jobRunBudgetUsd } from '@/lib/mothership/jobs/budget' @@ -26,6 +28,8 @@ const logger = createLogger('MothershipJobDispatch') /** A queued occurrence whose start failed this many times ends as failed instead of retrying. */ const MAX_JOB_DISPATCH_ATTEMPTS = 5 +/** Outlasts one dispatch (its reads and the turn's admission), so an attempt never overlaps another. */ +const JOB_DISPATCH_LOCK_SECONDS = 120 type JobDispatchOutcome = | 'started' @@ -67,6 +71,27 @@ async function block(run: JobRun, summary: string): Promise * skipped rather than queued behind it. */ export async function dispatchJobOccurrence(executionId: string): Promise { + /** + * One attempt at a time per occurrence (a tick's and the reconciler's can meet): a skip or + * block decided while another attempt admits the turn would close the record of a live run. + */ + const lock = `mothership:job-dispatch:${executionId}` + const holder = generateShortId() + if (!(await acquireLock(lock, holder, JOB_DISPATCH_LOCK_SECONDS, { reclaimOnFailure: true }))) + return 'already_started' + try { + return await dispatchQueuedOccurrence(executionId) + } finally { + await releaseLock(lock, holder).catch((error) => + logger.warn('Job dispatch lock release failed; it expires', { + executionId, + error: toError(error).message, + }) + ) + } +} + +async function dispatchQueuedOccurrence(executionId: string): Promise { const run = await loadJobRun(executionId) if (!run || run.status !== 'queued' || !run.occurrenceAt) return 'not_queued' const { userId, workspaceId, chatId } = run @@ -85,7 +110,9 @@ export async function dispatchJobOccurrence(executionId: string): Promise= ${MAX_CONSECUTIVE_JOB_FAILURES} AND ${workflowSchedule.status} = 'active' THEN 'disabled' ELSE ${workflowSchedule.status} END`, updatedAt: new Date(), }) .where(eq(workflowSchedule.id, scheduleId)) - .returning({ status: workflowSchedule.status, failedCount: workflowSchedule.failedCount }) - return schedule?.status === 'disabled' && schedule.failedCount === MAX_CONSECUTIVE_JOB_FAILURES + .returning({ failedCount: workflowSchedule.failedCount }) + if (!schedule || schedule.failedCount < MAX_CONSECUTIVE_JOB_FAILURES) return false + /** Only a job still on is turned off here; one a person already turned off is not reported. */ + const disabled = await tx + .update(workflowSchedule) + .set({ status: 'disabled', updatedAt: new Date() }) + .where(and(eq(workflowSchedule.id, scheduleId), eq(workflowSchedule.status, 'active'))) + .returning({ id: workflowSchedule.id }) + return disabled.length > 0 } /** The turn's last assistant text, as the summary of a run that reported no outcome. */ diff --git a/apps/sim/lib/mothership/jobs/lane.integration.ts b/apps/sim/lib/mothership/jobs/lane.integration.ts index 2811722f461..5428c4b5b72 100644 --- a/apps/sim/lib/mothership/jobs/lane.integration.ts +++ b/apps/sim/lib/mothership/jobs/lane.integration.ts @@ -108,6 +108,10 @@ vi.mock('@/lib/mothership/jobs/reconcile', async (importOriginal) => { return { ...actual, reconcileJobRuns: vi.fn(actual.reconcileJobRuns) } }) vi.mock('@/lib/core/config/feature-flags', () => featureFlagsMock) +/** Spied so a test can tell an automatic turn-off, which emails the job's people, from none. */ +vi.mock('@/lib/workflows/schedules/disable-notifications', () => ({ + notifyScheduleAutoDisabled: vi.fn(async () => {}), +})) const { controllers } = vi.hoisted(() => ({ controllers: new Map() })) vi.mock('@/lib/mothership/request/session/abort', async (importOriginal) => { @@ -136,7 +140,12 @@ import { sleep } from '@sim/utils/helpers' import { generateId } from '@sim/utils/id' import { eq, inArray, sql } from 'drizzle-orm' import { resolveBillingAttribution } from '@/lib/billing/core/attribution' -import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis' +import { + acquireLock, + closeRedisConnection, + getRedisClient, + releaseLock, +} from '@/lib/core/config/redis' import { ABANDONED_RUN_CURSOR_KEY } from '@/lib/mothership/async-runs/orphaned-runs' import { chatPubSub } from '@/lib/mothership/chat-status' import { recordJobRunSpend } from '@/lib/mothership/jobs/budget' @@ -152,6 +161,7 @@ import { chatStreamLockKey, StreamControllerSupersededError, } from '@/lib/mothership/request/session/controller-lease' +import { notifyScheduleAutoDisabled } from '@/lib/workflows/schedules/disable-notifications' interface QueryPlan { 'Node Type': string @@ -542,6 +552,88 @@ describe.runIf(Boolean(redisUrl))('the scheduled job lane', () => { worker.open.delete(occurrenceId) }, 60_000) + it('delivers a time-limit Stop an earlier pass recorded but died delivering, before finalizing', async () => { + const { scheduleId, occurrenceId } = await seedJob() + worker.legs.set(occurrenceId, [{ text: 'Still triaging. ', mode: 'hang' }]) + await runJobLane(new Date()) + await waitFor(async () => worker.open.has(occurrenceId), 'the first leg') + /** What a pass leaves when it dies after recording why the run stops, before delivering it. */ + await db + .update(jobExecutionLogs) + .set({ executionData: sql`jsonb_build_object('deadlineStopRequestedAt', now())` }) + .where(eq(jobExecutionLogs.executionId, occurrenceId)) + + await reconcileUntilSettled(occurrenceId) + + expect( + worker.requests.filter( + (request) => + request.path === '/api/streams/explicit-abort' && request.body.messageId === occurrenceId + ) + ).toHaveLength(1) + expect(await runsOf(scheduleId)).toEqual([ + expect.objectContaining({ + outcome: 'interrupted', + summary: 'Stopped at the run time limit.', + }), + ]) + worker.open.get(occurrenceId)?.response.end() + worker.open.delete(occurrenceId) + }, 60_000) + + it('does not report a job a person turned off as turned off by its failures', async () => { + vi.mocked(notifyScheduleAutoDisabled).mockClear() + const { scheduleId, occurrenceId } = await seedJob() + worker.legs.set(occurrenceId, [{ text: 'Still triaging. ', mode: 'hang' }]) + await runJobLane(new Date()) + await waitFor(async () => worker.open.has(occurrenceId), 'the first leg') + /** Off by hand with two failures behind it, while this run is still live. */ + await db + .update(workflowSchedule) + .set({ status: 'disabled', failedCount: 2 }) + .where(eq(workflowSchedule.id, scheduleId)) + + await reconcileJobRuns(new Date(Date.now() + 6 * 60 * 60_000 + 60_000)) + await reconcileUntilSettled(occurrenceId) + + expect(await schedule(scheduleId)).toMatchObject({ status: 'disabled', failedCount: 3 }) + expect(notifyScheduleAutoDisabled).not.toHaveBeenCalled() + worker.open.get(occurrenceId)?.response.end() + worker.open.delete(occurrenceId) + }, 60_000) + + it('briefs a retried occurrence with its own run number, not the job count a later claim moved', async () => { + const { scheduleId, occurrenceId } = await seedJob() + await claimDueJobOccurrences(new Date(), 10) + const later = await dueAgain(scheduleId) + await claimDueJobOccurrences(new Date(), 10) + expect(await schedule(scheduleId)).toMatchObject({ runCount: 2 }) + + for (const id of [occurrenceId, later]) { + await dispatchJobOccurrence(id) + await reconcileUntilSettled(id) + } + + expect(legsFor(occurrenceId)[0]?.body.message).toContain('(Run 1, scheduled for') + expect(legsFor(later)[0]?.body.message).toContain('(Run 2, scheduled for') + }) + + it('leaves an occurrence another attempt is dispatching to that attempt', async () => { + const { scheduleId, occurrenceId } = await seedJob() + await claimDueJobOccurrences(new Date(), 10) + const lock = `mothership:job-dispatch:${occurrenceId}` + expect(await acquireLock(lock, 'another-attempt', 60)).toBe(true) + try { + expect(await dispatchJobOccurrence(occurrenceId)).toBe('already_started') + expect(legsFor(occurrenceId)).toHaveLength(0) + expect(await runsOf(scheduleId)).toEqual([expect.objectContaining({ status: 'queued' })]) + } finally { + await releaseLock(lock, 'another-attempt') + } + expect(await dispatchJobOccurrence(occurrenceId)).toBe('started') + await reconcileUntilSettled(occurrenceId) + }) + it('runs a one-shot job once and completes it', async () => { const { scheduleId, occurrenceId } = await seedJob({ cronExpression: null }) await runJobLane(new Date()) @@ -682,6 +774,7 @@ describe.runIf(Boolean(redisUrl))('the scheduled job lane', () => { }) it('turns a job off after three failed runs in a row, and a verdict clears the streak', async () => { + vi.mocked(notifyScheduleAutoDisabled).mockClear() const { scheduleId, occurrenceId } = await seedJob() const fail = (id: string) => worker.legs.set( @@ -710,6 +803,10 @@ describe.runIf(Boolean(redisUrl))('the scheduled job lane', () => { } const outcomes = (await runsOf(scheduleId)).map((run) => run.outcome).sort() expect(outcomes).toEqual(['failed', 'failed', 'failed', 'failed', 'not_met']) + expect(notifyScheduleAutoDisabled).toHaveBeenCalledExactlyOnceWith({ + scheduleId, + reason: 'consecutive_failures', + }) await dueAgain(scheduleId) await runJobLane(new Date()) diff --git a/apps/sim/lib/mothership/jobs/reconcile.ts b/apps/sim/lib/mothership/jobs/reconcile.ts index 8274fbc2e47..41b86c5530a 100644 --- a/apps/sim/lib/mothership/jobs/reconcile.ts +++ b/apps/sim/lib/mothership/jobs/reconcile.ts @@ -46,8 +46,8 @@ const admittedBillingRequestId = sql< /** * Open records that need this pass, oldest first. Every queued record does; a running one only - * once its turn ended or its run row is gone, its time limit passed without a Stop, or its - * spend-cap link is missing, so a backlog of healthy long runs never hides the ones that finished. + * once its turn ended or its run row is gone, its time limit passed without a Stop, its Stop's + * delivery never finished, or its spend-cap link is missing, so a backlog of healthy long runs never hides the ones that finished. */ function openJobRuns(status: OpenJobRunStatus, now: Date) { const needsAction = @@ -60,6 +60,7 @@ function openJobRuns(status: OpenJobRunStatus, now: Date) { lte(jobExecutionLogs.startedAt, new Date(now.getTime() - JOB_RUN_TIME_LIMIT_MS)), sql`${jobExecutionLogs.executionData}->>'deadlineStopRequestedAt' IS NULL` ), + sql`${jobExecutionLogs.executionData}->>'deadlineStopRequestedAt' IS NOT NULL AND ${jobExecutionLogs.executionData}->>'deadlineStopDeliveredAt' IS NULL`, and(isNull(jobExecutionLogs.billingRequestId), isNotNull(admittedBillingRequestId)) ) return db @@ -70,6 +71,9 @@ function openJobRuns(status: OpenJobRunStatus, now: Date) { deadlineStopRequestedAt: sql< string | null >`${jobExecutionLogs.executionData}->>'deadlineStopRequestedAt'`, + deadlineStopDeliveredAt: sql< + string | null + >`${jobExecutionLogs.executionData}->>'deadlineStopDeliveredAt'`, billingRequestId: jobExecutionLogs.billingRequestId, admittedBillingRequestId, scheduleId: jobExecutionLogs.scheduleId, @@ -89,8 +93,9 @@ function openJobRuns(status: OpenJobRunStatus, now: Date) { * Drives every open occurrence to an end, so no crash leaves one queued or running forever: * a queued record whose dispatch was lost is dispatched again under the same id (the turn is * admitted at most once, so a retry of one that did start only finds it); a record whose turn - * ended is finalized; one past the time limit is stopped and finalized once it settles; one - * whose job was deleted before it started is ended. A run whose spend-cap link was lost (its + * ended is finalized; one past the time limit is stopped and finalized once it settles, its + * Stop delivered again first if an earlier pass died delivering it; one whose job was deleted + * before it started is ended. A run whose spend-cap link was lost (its * process died between recording the billing admission and the link) gets it back. */ export async function reconcileJobRuns(now: Date): Promise { @@ -121,15 +126,15 @@ export async function reconcileJobRuns(now: Date): Promise { if (!row.billingRequestId && row.admittedBillingRequestId) { await recordJobRunBillingRequest(row.executionId, row.admittedBillingRequestId) } - if (TERMINAL_RUN_STATUSES.includes(row.runStatus)) { - await finalizeJobRun(row.executionId) - } else if ( + const ended = TERMINAL_RUN_STATUSES.includes(row.runStatus) + /** A Stop whose delivery never finished is delivered before the run is finalized. */ + const undelivered = + row.deadlineStopRequestedAt !== null && row.deadlineStopDeliveredAt === null + const overdue = + !ended && !row.deadlineStopRequestedAt && - row.userId && - row.workspaceId && - row.chatId && now.getTime() - row.startedAt.getTime() >= JOB_RUN_TIME_LIMIT_MS - ) { + if ((undelivered || overdue) && row.userId && row.workspaceId && row.chatId) { await stopAtTimeLimit({ executionId: row.executionId, userId: row.userId, @@ -137,6 +142,7 @@ export async function reconcileJobRuns(now: Date): Promise { chatId: row.chatId, }) } + if (ended) await finalizeJobRun(row.executionId) } catch (error) { logger.warn('Job run could not be reconciled; the next pass retries', { executionId: row.executionId, @@ -161,13 +167,12 @@ async function stopAtTimeLimit(run: { }) /** A run outside the job's scope records no Stop, so nothing is delivered to it. */ if (!stopped || stopped.workspaceId !== run.workspaceId || stopped.chatId !== run.chatId) return - /** Marked once the Stop is durable, so a pass that fails before it stops the run again. */ - await db - .update(jobExecutionLogs) - .set({ - executionData: sql`jsonb_set(${jobExecutionLogs.executionData}, '{deadlineStopRequestedAt}', to_jsonb(now()))`, - }) - .where(eq(jobExecutionLogs.executionId, run.executionId)) + /** + * Why the run stopped is recorded before delivery, so its summary survives a crash; delivery + * is marked separately once done, and a pass that finds it unmarked delivers again (requesting + * and delivering a Stop are idempotent), so the run's client workflows and sandbox work end. + */ + await markDeadlineStop(run.executionId, 'deadlineStopRequestedAt') await deliverRunStop( { id: stopped.id, @@ -178,5 +183,19 @@ async function stopAtTimeLimit(run: { }, readStoppedRunWork(run.executionId, run.userId) ) + await markDeadlineStop(run.executionId, 'deadlineStopDeliveredAt') logger.info('Stopped a job run at its time limit', { executionId: run.executionId }) } + +/** Records a step of the time-limit Stop once; a repeated pass keeps the first time. */ +async function markDeadlineStop( + executionId: string, + step: 'deadlineStopRequestedAt' | 'deadlineStopDeliveredAt' +): Promise { + await db + .update(jobExecutionLogs) + .set({ + executionData: sql`jsonb_set(${jobExecutionLogs.executionData}, ${`{${step}}`}::text[], COALESCE(${jobExecutionLogs.executionData}->${step}, to_jsonb(now())))`, + }) + .where(eq(jobExecutionLogs.executionId, executionId)) +} diff --git a/apps/sim/lib/mothership/jobs/runs.ts b/apps/sim/lib/mothership/jobs/runs.ts index 53f4917ac72..93359b848b5 100644 --- a/apps/sim/lib/mothership/jobs/runs.ts +++ b/apps/sim/lib/mothership/jobs/runs.ts @@ -40,7 +40,8 @@ export async function loadJobRun(executionId: string) { successCondition: workflowSchedule.successCondition, lifecycle: workflowSchedule.lifecycle, maxRuns: workflowSchedule.maxRuns, - runCount: workflowSchedule.runCount, + /** Records claimed before the run number was stored fall back to the schedule's count. */ + runNumber: sql`COALESCE((${jobExecutionLogs.executionData}->>'runNumber')::int, ${workflowSchedule.runCount})`, runBudgetUsd: workflowSchedule.runBudgetUsd, userId: workflowSchedule.sourceUserId, workspaceId: workflowSchedule.sourceWorkspaceId, diff --git a/packages/db/migrations/0437_scheduled_job_occurrences.sql b/packages/db/migrations/0437_scheduled_job_occurrences.sql index cc047db1815..0c480351727 100644 --- a/packages/db/migrations/0437_scheduled_job_occurrences.sql +++ b/packages/db/migrations/0437_scheduled_job_occurrences.sql @@ -5,9 +5,37 @@ ALTER TABLE "job_execution_logs" ADD COLUMN IF NOT EXISTS "outcome" text;--> sta ALTER TABLE "job_execution_logs" ADD COLUMN IF NOT EXISTS "summary" text;--> statement-breakpoint ALTER TABLE "job_execution_logs" ADD COLUMN IF NOT EXISTS "billing_request_id" text;--> statement-breakpoint ALTER TABLE "workflow_schedule" ADD COLUMN IF NOT EXISTS "run_budget_usd" numeric(12, 4);--> statement-breakpoint +-- A failed concurrent build leaves an invalid index that IF NOT EXISTS would keep: each replay +-- moves one aside to be dropped, never a valid index or one on another table. +DO $$ +DECLARE + index_name text; +BEGIN + FOREACH index_name IN ARRAY ARRAY[ + 'job_execution_logs_schedule_occurrence_unique', + 'job_execution_logs_billing_request_unique', + 'job_execution_logs_open_idx' + ] LOOP + IF EXISTS ( + SELECT 1 FROM pg_index WHERE indexrelid = to_regclass(index_name || '_failed') + AND (indisvalid OR indrelid <> to_regclass('job_execution_logs')) + ) THEN + RAISE EXCEPTION 'Refusing to drop unexpected recovery index %_failed', index_name; + END IF; + IF EXISTS ( + SELECT 1 FROM pg_index WHERE indexrelid = to_regclass(index_name) + AND indrelid = to_regclass('job_execution_logs') AND NOT indisvalid + ) THEN + EXECUTE format('ALTER INDEX %I RENAME TO %I', index_name, index_name || '_failed'); + END IF; + END LOOP; +END $$;--> statement-breakpoint COMMIT;--> statement-breakpoint SET lock_timeout = 0;--> statement-breakpoint +DROP INDEX CONCURRENTLY IF EXISTS "job_execution_logs_schedule_occurrence_unique_failed";--> statement-breakpoint CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "job_execution_logs_schedule_occurrence_unique" ON "job_execution_logs" USING btree ("schedule_id","occurrence_at");--> statement-breakpoint +DROP INDEX CONCURRENTLY IF EXISTS "job_execution_logs_billing_request_unique_failed";--> statement-breakpoint CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "job_execution_logs_billing_request_unique" ON "job_execution_logs" USING btree ("billing_request_id") WHERE "job_execution_logs"."billing_request_id" IS NOT NULL;--> statement-breakpoint +DROP INDEX CONCURRENTLY IF EXISTS "job_execution_logs_open_idx_failed";--> statement-breakpoint CREATE INDEX CONCURRENTLY IF NOT EXISTS "job_execution_logs_open_idx" ON "job_execution_logs" USING btree ("started_at") WHERE "job_execution_logs"."status" IN ('queued', 'running');--> statement-breakpoint SET lock_timeout = '5s';