Skip to content

Commit 16862f6

Browse files
committed
fix(webapp): the limits surface preserves pauses, percent sources, and never resolves a stale row
Per-key engine writes from the limits surface preserve a queue's pause (the pause is the engine value 0, so an override or reset touching only total no longer silently resumes a paused queue), and a perKey write clears the percent override source so the percent recalculation cannot later clobber or resurrect it. Deploys retire a task's anonymous LIMIT row when its inline limit no longer compiles there, clearing bounds and engine keys so the row cannot shadow the live task/<id> queue row; name resolution and listing skip boundless LIMIT rows accordingly. Boundless V2 queues resolve by name again so an operator can cap an undeclared task through the surface the V1 endpoints point at, while staying out of the list.
1 parent 63b0d9f commit 16862f6

3 files changed

Lines changed: 211 additions & 18 deletions

File tree

‎apps/webapp/app/v3/services/concurrencyLimitsSystem.server.ts‎

Lines changed: 41 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -126,6 +126,7 @@ export class ConcurrencyLimitsSystem {
126126
alreadySynced: {
127127
perKey: row.concurrencyLimit,
128128
total: row.totalConcurrencyLimit,
129+
paused: row.paused,
129130
},
130131
})
131132
.orElse(() => okAsync(undefined))
@@ -192,14 +193,17 @@ function concurrencyLimitNameFromRow(row: Pick<TaskQueue, "name">): string {
192193
* match: their limit is queue surface, managed through the queues API.
193194
*/
194195
function limitRowsWhere(environment: AuthenticatedEnvironment) {
196+
const hasBounds = {
197+
OR: [{ concurrencyLimit: { not: null } }, { totalConcurrencyLimit: { not: null } }],
198+
};
195199
return {
196200
runtimeEnvironmentId: environment.id,
197201
OR: [
198-
{ role: "LIMIT" as const },
202+
{ role: "LIMIT" as const, ...hasBounds },
199203
{
200204
role: "QUEUE" as const,
201205
concurrencyVersion: "V2" as const,
202-
OR: [{ concurrencyLimit: { not: null } }, { totalConcurrencyLimit: { not: null } }],
206+
...hasBounds,
203207
},
204208
],
205209
};
@@ -220,6 +224,7 @@ function findLimitByName(
220224
runtimeEnvironmentId: environment.id,
221225
name: `${LIMIT_QUEUE_PREFIX}${name}`,
222226
role: "LIMIT",
227+
OR: [{ concurrencyLimit: { not: null } }, { totalConcurrencyLimit: { not: null } }],
223228
},
224229
}),
225230
(error) => ({ type: "other" as const, cause: error })
@@ -237,7 +242,6 @@ function findLimitByName(
237242
name,
238243
role: "QUEUE",
239244
concurrencyVersion: "V2",
240-
OR: [{ concurrencyLimit: { not: null } }, { totalConcurrencyLimit: { not: null } }],
241245
},
242246
}),
243247
(error) => ({ type: "other" as const, cause: error })
@@ -322,6 +326,7 @@ function applyLimitOverride(
322326
: (row.concurrencyLimit ?? null);
323327
data.concurrencyLimitOverriddenAt = now;
324328
data.concurrencyLimitOverriddenBy = overriddenBy?.id ?? null;
329+
data.concurrencyLimitOverridePercent = null;
325330
}
326331

327332
if (override.total !== undefined) {
@@ -341,6 +346,27 @@ function applyLimitOverride(
341346
* override markers clear, so an engine failure leaves the markers set and a retry
342347
* converges instead of being rejected while the overridden limit stays enforced.
343348
*/
349+
/**
350+
* A paused row's pause IS the engine per-key value 0 (the DB concurrencyLimit
351+
* column keeps the configured value), so every per-key engine write from this
352+
* surface must preserve it — otherwise an override or reset that only touched
353+
* `total` would silently resume a queue every other surface still reports as
354+
* paused. Named LIMIT rows are never paused (pausing a limit is an override to
355+
* `{ total: 0 }`), so they always take the target branch.
356+
*/
357+
function perKeyEngineWrite(
358+
environment: AuthenticatedEnvironment,
359+
row: Pick<TaskQueue, "name" | "paused">,
360+
target: number | null | undefined
361+
) {
362+
if (row.paused) {
363+
return updateQueueConcurrencyLimits(environment, row.name, 0);
364+
}
365+
return typeof target === "number"
366+
? updateQueueConcurrencyLimits(environment, row.name, target)
367+
: removeQueueConcurrencyLimits(environment, row.name);
368+
}
369+
344370
function syncResetToEngine(
345371
environment: AuthenticatedEnvironment,
346372
row: TaskQueue
@@ -359,10 +385,7 @@ function syncResetToEngine(
359385
? row.totalConcurrencyLimitBase
360386
: row.totalConcurrencyLimit;
361387

362-
const perKeySync =
363-
typeof perKeyTarget === "number"
364-
? updateQueueConcurrencyLimits(environment, row.name, perKeyTarget)
365-
: removeQueueConcurrencyLimits(environment, row.name);
388+
const perKeySync = perKeyEngineWrite(environment, row, perKeyTarget);
366389

367390
const totalSync =
368391
typeof totalTarget === "number"
@@ -383,6 +406,7 @@ function resetLimitOverrides(db: PrismaClientOrTransaction, row: TaskQueue) {
383406
data.concurrencyLimitBase = null;
384407
data.concurrencyLimitOverriddenAt = null;
385408
data.concurrencyLimitOverriddenBy = null;
409+
data.concurrencyLimitOverridePercent = null;
386410
}
387411

388412
if (row.totalConcurrencyLimitOverriddenAt !== null) {
@@ -440,7 +464,7 @@ function guardedLimitUpdate(
440464
);
441465
}
442466

443-
type SyncedLimitValues = { perKey: number | null; total: number | null };
467+
type SyncedLimitValues = { perKey: number | null; total: number | null; paused: boolean };
444468

445469
/**
446470
* Re-syncs the engine from fresh reads of the row until the enforced values stop
@@ -475,14 +499,13 @@ function compensateEngineFromFreshRow(
475499
!fresh ||
476500
(lastSynced !== null &&
477501
fresh.concurrencyLimit === lastSynced.perKey &&
478-
fresh.totalConcurrencyLimit === lastSynced.total)
502+
fresh.totalConcurrencyLimit === lastSynced.total &&
503+
fresh.paused === lastSynced.paused)
479504
) {
480505
return;
481506
}
482507
await settleBothEngineWrites(
483-
typeof fresh.concurrencyLimit === "number"
484-
? updateQueueConcurrencyLimits(environment, fresh.name, fresh.concurrencyLimit)
485-
: removeQueueConcurrencyLimits(environment, fresh.name),
508+
perKeyEngineWrite(environment, fresh, fresh.concurrencyLimit),
486509
typeof fresh.totalConcurrencyLimit === "number"
487510
? updateQueueTotalConcurrencyLimits(
488511
environment,
@@ -491,7 +514,11 @@ function compensateEngineFromFreshRow(
491514
)
492515
: removeQueueTotalConcurrencyLimits(environment, fresh.name)
493516
);
494-
lastSynced = { perKey: fresh.concurrencyLimit, total: fresh.totalConcurrencyLimit };
517+
lastSynced = {
518+
perKey: fresh.concurrencyLimit,
519+
total: fresh.totalConcurrencyLimit,
520+
paused: fresh.paused,
521+
};
495522
}
496523
})(),
497524
(error) => ({ type: "other" as const, cause: error })
@@ -504,10 +531,7 @@ function compensateEngineFromFreshRow(
504531
* both keys sync unconditionally, unlike queue rows.
505532
*/
506533
function syncLimitToEngine(environment: AuthenticatedEnvironment, row: TaskQueue) {
507-
const perKeySync =
508-
typeof row.concurrencyLimit === "number"
509-
? updateQueueConcurrencyLimits(environment, row.name, row.concurrencyLimit)
510-
: removeQueueConcurrencyLimits(environment, row.name);
534+
const perKeySync = perKeyEngineWrite(environment, row, row.concurrencyLimit);
511535

512536
const totalSync =
513537
typeof row.totalConcurrencyLimit === "number"

‎apps/webapp/app/v3/services/createBackgroundWorker.server.ts‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -387,9 +387,68 @@ async function createWorkerTasks(
387387
if (entry) entries.push(entry);
388388
}
389389
}
390+
391+
await retireStaleAnonymousConcurrencyLimitRows(metadata.tasks, environment, prisma);
392+
390393
return entries;
391394
}
392395

396+
/**
397+
* A task whose inline limit no longer compiles to its anonymous LIMIT row (the
398+
* limit moved onto the task's own default queue, or was removed) leaves that
399+
* row behind, where it would shadow the live task/<id> name on the
400+
* concurrency-limits surface and keep stale engine keys. Retiring clears the
401+
* row's bounds and override state — dropping it from listing and name
402+
* resolution so the queue-row fallback resolves — and removes its engine keys.
403+
*/
404+
async function retireStaleAnonymousConcurrencyLimitRows(
405+
tasks: TaskResource[],
406+
environment: AuthenticatedEnvironment,
407+
prisma: PrismaClientOrTransaction
408+
): Promise<void> {
409+
const candidateNames = tasks
410+
.filter((task) => !(task.concurrency?.inline && task.queue?.name))
411+
.map((task) => anonymousConcurrencyLimitQueueName(task.id));
412+
if (candidateNames.length === 0) {
413+
return;
414+
}
415+
416+
const staleRows = await prisma.taskQueue.findMany({
417+
where: {
418+
runtimeEnvironmentId: environment.id,
419+
role: "LIMIT",
420+
name: { in: candidateNames },
421+
OR: [{ concurrencyLimit: { not: null } }, { totalConcurrencyLimit: { not: null } }],
422+
},
423+
select: { id: true, name: true },
424+
});
425+
if (staleRows.length === 0) {
426+
return;
427+
}
428+
429+
await prisma.taskQueue.updateMany({
430+
where: { id: { in: staleRows.map((row) => row.id) } },
431+
data: {
432+
concurrencyLimit: null,
433+
concurrencyLimitBase: null,
434+
concurrencyLimitOverriddenAt: null,
435+
concurrencyLimitOverriddenBy: null,
436+
concurrencyLimitOverridePercent: null,
437+
totalConcurrencyLimit: null,
438+
totalConcurrencyLimitBase: null,
439+
totalConcurrencyLimitOverriddenAt: null,
440+
totalConcurrencyLimitOverriddenBy: null,
441+
},
442+
});
443+
444+
await Promise.allSettled(
445+
staleRows.flatMap((row) => [
446+
removeQueueConcurrencyLimits(environment, row.name),
447+
removeQueueTotalConcurrencyLimits(environment, row.name),
448+
])
449+
);
450+
}
451+
393452
async function createWorkerTask(
394453
task: TaskResource,
395454
queues: Array<TaskQueue>,

‎apps/webapp/test/concurrencyLimitsSystem.test.ts‎

Lines changed: 111 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -344,8 +344,14 @@ describe("ConcurrencyLimitsSystem", () => {
344344
const v1 = await system.limits.retrieve(authEnv, "task/legacy-task");
345345
expect(v1.isErr()).toBe(true);
346346

347+
/** Boundless V2 queues stay out of the list but resolve by name, so an
348+
* operator can still cap an undeclared task through this surface. */
347349
const boundless = await system.limits.retrieve(authEnv, "task/unbounded-task");
348-
expect(boundless.isErr()).toBe(true);
350+
expect(boundless.isOk()).toBe(true);
351+
if (boundless.isOk()) {
352+
expect(boundless.value.perKey.current).toBeNull();
353+
expect(boundless.value.total.current).toBeNull();
354+
}
349355

350356
const listed = await system.limits.list(authEnv, { page: 1, perPage: 50 });
351357
expect(listed.isOk()).toBe(true);
@@ -355,6 +361,110 @@ describe("ConcurrencyLimitsSystem", () => {
355361
}
356362
);
357363

364+
postgresTest(
365+
"a retired anonymous LIMIT row falls through to the live queue row",
366+
async ({ prisma }) => {
367+
const { authEnv, system, environment } = await seedEnvAndLimit(prisma, { total: 25 });
368+
369+
await prisma.taskQueue.create({
370+
data: {
371+
friendlyId: `queue_rl${environment.slug}`,
372+
name: "limit/task/send-email",
373+
orderableName: "send-email",
374+
projectId: environment.projectId,
375+
runtimeEnvironmentId: environment.id,
376+
role: "LIMIT",
377+
concurrencyVersion: "V2",
378+
},
379+
});
380+
await prisma.taskQueue.create({
381+
data: {
382+
friendlyId: `queue_ql${environment.slug}`,
383+
name: "task/send-email",
384+
orderableName: "send-email-q",
385+
projectId: environment.projectId,
386+
runtimeEnvironmentId: environment.id,
387+
role: "QUEUE",
388+
concurrencyVersion: "V2",
389+
concurrencyLimit: 1,
390+
totalConcurrencyLimit: 10,
391+
},
392+
});
393+
394+
const retrieved = await system.limits.retrieve(authEnv, "task/send-email");
395+
expect(retrieved.isOk()).toBe(true);
396+
if (retrieved.isOk()) {
397+
expect(retrieved.value.total).toMatchObject({ current: 10 });
398+
}
399+
400+
const listed = await system.limits.list(authEnv, { page: 1, perPage: 50 });
401+
expect(listed.isOk()).toBe(true);
402+
if (listed.isOk()) {
403+
expect(listed.value.filter((item) => item.name === "task/send-email")).toHaveLength(1);
404+
}
405+
}
406+
);
407+
408+
postgresTest("overrides and resets preserve a queue pause in the engine", async ({ prisma }) => {
409+
const { authEnv, system, environment } = await seedEnvAndLimit(prisma, { total: 25 });
410+
411+
await prisma.taskQueue.create({
412+
data: {
413+
friendlyId: `queue_p${environment.slug}`,
414+
name: "task/paused-task",
415+
orderableName: "paused-task",
416+
projectId: environment.projectId,
417+
runtimeEnvironmentId: environment.id,
418+
role: "QUEUE",
419+
concurrencyVersion: "V2",
420+
concurrencyLimit: 1,
421+
totalConcurrencyLimit: 10,
422+
paused: true,
423+
},
424+
});
425+
426+
const overridden = await system.limits.override(authEnv, "task/paused-task", { total: 20 });
427+
expect(overridden.isOk()).toBe(true);
428+
expect(totalSyncMock).toHaveBeenCalledWith(authEnv, "task/paused-task", 20);
429+
/** The pause IS the per-key engine value 0; the sync must rewrite 0, never
430+
* the configured limit and never a removal. */
431+
expect(perKeySyncMock).toHaveBeenCalledWith(authEnv, "task/paused-task", 0);
432+
expect(perKeySyncMock).not.toHaveBeenCalledWith(authEnv, "task/paused-task", 1);
433+
expect(perKeyRemoveMock).not.toHaveBeenCalledWith(authEnv, "task/paused-task");
434+
435+
perKeySyncMock.mockClear();
436+
const reset = await system.limits.reset(authEnv, "task/paused-task");
437+
expect(reset.isOk()).toBe(true);
438+
expect(perKeySyncMock).toHaveBeenCalledWith(authEnv, "task/paused-task", 0);
439+
expect(perKeySyncMock).not.toHaveBeenCalledWith(authEnv, "task/paused-task", 1);
440+
});
441+
442+
postgresTest("a perKey override clears a stale percent override source", async ({ prisma }) => {
443+
const { authEnv, system, environment } = await seedEnvAndLimit(prisma, { total: 25 });
444+
445+
const row = await prisma.taskQueue.create({
446+
data: {
447+
friendlyId: `queue_pc${environment.slug}`,
448+
name: "task/percent-task",
449+
orderableName: "percent-task",
450+
projectId: environment.projectId,
451+
runtimeEnvironmentId: environment.id,
452+
role: "QUEUE",
453+
concurrencyVersion: "V2",
454+
concurrencyLimit: 50,
455+
concurrencyLimitBase: 100,
456+
concurrencyLimitOverriddenAt: new Date(),
457+
concurrencyLimitOverridePercent: 50,
458+
},
459+
});
460+
461+
const overridden = await system.limits.override(authEnv, "task/percent-task", { perKey: 3 });
462+
expect(overridden.isOk()).toBe(true);
463+
const updated = await prisma.taskQueue.findFirstOrThrow({ where: { id: row.id } });
464+
expect(updated.concurrencyLimit).toBe(3);
465+
expect(updated.concurrencyLimitOverridePercent).toBeNull();
466+
});
467+
358468
postgresTest("retrieve misses queue-role rows and unknown names", async ({ prisma }) => {
359469
const { authEnv, system, environment } = await seedEnvAndLimit(prisma, { total: 25 });
360470

0 commit comments

Comments
 (0)