@@ -401,6 +401,36 @@ async function createWorkerTasks(
401401 * row's bounds and override state — dropping it from listing and name
402402 * resolution so the queue-row fallback resolves — and removes its engine keys.
403403 */
404+ /**
405+ * Best-effort engine sync of a limit row's bounds (pause-aware): both writes
406+ * settle, and any rejection is surfaced in the logs rather than swallowed — a
407+ * silent failure here is a limit the dashboard reports but the engine does not
408+ * enforce.
409+ */
410+ async function syncLimitRowEngineStateBestEffort (
411+ environment : AuthenticatedEnvironment ,
412+ row : Pick < TaskQueue , "name" | "paused" | "concurrencyLimit" | "totalConcurrencyLimit" >
413+ ) : Promise < void > {
414+ const results = await Promise . allSettled ( [
415+ row . paused
416+ ? updateQueueConcurrencyLimits ( environment , row . name , 0 )
417+ : typeof row . concurrencyLimit === "number"
418+ ? updateQueueConcurrencyLimits ( environment , row . name , row . concurrencyLimit )
419+ : removeQueueConcurrencyLimits ( environment , row . name ) ,
420+ typeof row . totalConcurrencyLimit === "number"
421+ ? updateQueueTotalConcurrencyLimits ( environment , row . name , row . totalConcurrencyLimit )
422+ : removeQueueTotalConcurrencyLimits ( environment , row . name ) ,
423+ ] ) ;
424+ const failures = results . filter ( ( result ) => result . status === "rejected" ) ;
425+ if ( failures . length > 0 ) {
426+ logger . error ( "syncLimitRowEngineStateBestEffort: engine sync failed" , {
427+ environmentId : environment . id ,
428+ queueName : row . name ,
429+ errors : failures . map ( ( failure ) => String ( ( failure as PromiseRejectedResult ) . reason ) ) ,
430+ } ) ;
431+ }
432+ }
433+
404434async function retireStaleAnonymousConcurrencyLimitRows (
405435 tasks : TaskResource [ ] ,
406436 environment : AuthenticatedEnvironment ,
@@ -420,7 +450,14 @@ async function retireStaleAnonymousConcurrencyLimitRows(
420450 name : { in : boundedIn ( candidateNames ) } ,
421451 OR : [ { concurrencyLimit : { not : null } } , { totalConcurrencyLimit : { not : null } } ] ,
422452 } ,
423- select : { id : true , name : true , updatedAt : true } ,
453+ select : {
454+ id : true ,
455+ name : true ,
456+ updatedAt : true ,
457+ paused : true ,
458+ concurrencyLimit : true ,
459+ totalConcurrencyLimit : true ,
460+ } ,
424461 } ) ;
425462 if ( staleRows . length === 0 ) {
426463 return ;
@@ -442,13 +479,14 @@ async function retireStaleAnonymousConcurrencyLimitRows(
442479 ] ) ;
443480 } catch ( error ) {
444481 logger . error (
445- "retireStaleAnonymousConcurrencyLimitRows: engine cleanup failed, retrying next deploy" ,
482+ "retireStaleAnonymousConcurrencyLimitRows: engine cleanup failed, healing and retrying next deploy" ,
446483 {
447484 environmentId : environment . id ,
448485 queueName : row . name ,
449486 error,
450487 }
451488 ) ;
489+ await syncLimitRowEngineStateBestEffort ( environment , row ) ;
452490 continue ;
453491 }
454492 const retired = await prisma . taskQueue . updateMany ( {
@@ -474,20 +512,7 @@ async function retireStaleAnonymousConcurrencyLimitRows(
474512 * race converges. */
475513 const fresh = await prisma . taskQueue . findFirst ( { where : { id : row . id } } ) ;
476514 if ( fresh ) {
477- await Promise . allSettled ( [
478- fresh . paused
479- ? updateQueueConcurrencyLimits ( environment , fresh . name , 0 )
480- : typeof fresh . concurrencyLimit === "number"
481- ? updateQueueConcurrencyLimits ( environment , fresh . name , fresh . concurrencyLimit )
482- : removeQueueConcurrencyLimits ( environment , fresh . name ) ,
483- typeof fresh . totalConcurrencyLimit === "number"
484- ? updateQueueTotalConcurrencyLimits (
485- environment ,
486- fresh . name ,
487- fresh . totalConcurrencyLimit
488- )
489- : removeQueueTotalConcurrencyLimits ( environment , fresh . name ) ,
490- ] ) ;
515+ await syncLimitRowEngineStateBestEffort ( environment , fresh ) ;
491516 }
492517 }
493518 }
0 commit comments