@@ -776,4 +776,184 @@ describe("BatchQueue", () => {
776776 }
777777 ) ;
778778 } ) ;
779+
780+ describe ( "skipRetries on failed items" , ( ) => {
781+ function createBatchQueueWithRetry (
782+ redisContainer : { getHost : ( ) => string ; getPort : ( ) => number } ,
783+ maxAttempts : number
784+ ) {
785+ return new BatchQueue ( {
786+ redis : {
787+ host : redisContainer . getHost ( ) ,
788+ port : redisContainer . getPort ( ) ,
789+ keyPrefix : "test:" ,
790+ } ,
791+ drr : { quantum : 5 , maxDeficit : 50 } ,
792+ consumerCount : 1 ,
793+ consumerIntervalMs : 50 ,
794+ startConsumers : true ,
795+ retry : {
796+ maxAttempts,
797+ // Keep the ladder tiny so a regression (items retrying N times)
798+ // still finishes inside the waitFor timeout and the test surfaces
799+ // the problem as a failed attempt-count assertion rather than a
800+ // timeout.
801+ minTimeoutInMs : 20 ,
802+ maxTimeoutInMs : 100 ,
803+ factor : 2 ,
804+ randomize : false ,
805+ } ,
806+ } ) ;
807+ }
808+
809+ redisTest (
810+ "should not retry when callback returns skipRetries: true" ,
811+ async ( { redisContainer } ) => {
812+ const queue = createBatchQueueWithRetry ( redisContainer , 6 ) ;
813+ const itemAttempts = new Map < number , number > ( ) ;
814+ let completionResult : CompleteBatchResult | null = null ;
815+
816+ try {
817+ queue . onProcessItem ( async ( { itemIndex } ) => {
818+ itemAttempts . set ( itemIndex , ( itemAttempts . get ( itemIndex ) ?? 0 ) + 1 ) ;
819+ return {
820+ success : false as const ,
821+ error : "Queue at maximum size" ,
822+ errorCode : "QUEUE_SIZE_LIMIT_EXCEEDED" ,
823+ skipRetries : true ,
824+ } ;
825+ } ) ;
826+
827+ queue . onBatchComplete ( async ( result ) => {
828+ completionResult = result ;
829+ } ) ;
830+
831+ await queue . initializeBatch ( createInitOptions ( "batch1" , "env1" , 3 ) ) ;
832+ await enqueueItems ( queue , "batch1" , "env1" , createBatchItems ( 3 ) ) ;
833+
834+ await vi . waitFor (
835+ ( ) => {
836+ expect ( completionResult ) . not . toBeNull ( ) ;
837+ } ,
838+ { timeout : 5000 }
839+ ) ;
840+
841+ // Every item should have been called exactly once — skipRetries
842+ // must bypass the 6-attempt retry ladder on the very first attempt.
843+ expect ( itemAttempts . get ( 0 ) ) . toBe ( 1 ) ;
844+ expect ( itemAttempts . get ( 1 ) ) . toBe ( 1 ) ;
845+ expect ( itemAttempts . get ( 2 ) ) . toBe ( 1 ) ;
846+
847+ expect ( completionResult ! . successfulRunCount ) . toBe ( 0 ) ;
848+ expect ( completionResult ! . failedRunCount ) . toBe ( 3 ) ;
849+ expect ( completionResult ! . failures ) . toHaveLength ( 3 ) ;
850+ for ( const failure of completionResult ! . failures ) {
851+ expect ( failure . errorCode ) . toBe ( "QUEUE_SIZE_LIMIT_EXCEEDED" ) ;
852+ }
853+ } finally {
854+ await queue . close ( ) ;
855+ }
856+ }
857+ ) ;
858+
859+ redisTest (
860+ "should still retry up to maxAttempts when skipRetries is not set (regression guard)" ,
861+ async ( { redisContainer } ) => {
862+ const maxAttempts = 3 ;
863+ const queue = createBatchQueueWithRetry ( redisContainer , maxAttempts ) ;
864+ const itemAttempts = new Map < number , number > ( ) ;
865+ let completionResult : CompleteBatchResult | null = null ;
866+
867+ try {
868+ queue . onProcessItem ( async ( { itemIndex } ) => {
869+ itemAttempts . set ( itemIndex , ( itemAttempts . get ( itemIndex ) ?? 0 ) + 1 ) ;
870+ return {
871+ success : false as const ,
872+ error : "Transient error" ,
873+ errorCode : "TRIGGER_ERROR" ,
874+ // Intentionally NOT setting skipRetries — the existing
875+ // exponential-backoff retry path should still be honored.
876+ } ;
877+ } ) ;
878+
879+ queue . onBatchComplete ( async ( result ) => {
880+ completionResult = result ;
881+ } ) ;
882+
883+ await queue . initializeBatch ( createInitOptions ( "batch1" , "env1" , 2 ) ) ;
884+ await enqueueItems ( queue , "batch1" , "env1" , createBatchItems ( 2 ) ) ;
885+
886+ await vi . waitFor (
887+ ( ) => {
888+ expect ( completionResult ) . not . toBeNull ( ) ;
889+ } ,
890+ { timeout : 5000 }
891+ ) ;
892+
893+ expect ( itemAttempts . get ( 0 ) ) . toBe ( maxAttempts ) ;
894+ expect ( itemAttempts . get ( 1 ) ) . toBe ( maxAttempts ) ;
895+
896+ expect ( completionResult ! . failedRunCount ) . toBe ( 2 ) ;
897+ } finally {
898+ await queue . close ( ) ;
899+ }
900+ }
901+ ) ;
902+
903+ redisTest (
904+ "should honor skipRetries on a per-item basis within the same batch" ,
905+ async ( { redisContainer } ) => {
906+ const maxAttempts = 4 ;
907+ const queue = createBatchQueueWithRetry ( redisContainer , maxAttempts ) ;
908+ const itemAttempts = new Map < number , number > ( ) ;
909+ let completionResult : CompleteBatchResult | null = null ;
910+
911+ try {
912+ queue . onProcessItem ( async ( { itemIndex } ) => {
913+ itemAttempts . set ( itemIndex , ( itemAttempts . get ( itemIndex ) ?? 0 ) + 1 ) ;
914+ // Even items fast-fail (queue-size-limit style),
915+ // odd items retry the full ladder.
916+ if ( itemIndex % 2 === 0 ) {
917+ return {
918+ success : false as const ,
919+ error : "Queue at maximum size" ,
920+ errorCode : "QUEUE_SIZE_LIMIT_EXCEEDED" ,
921+ skipRetries : true ,
922+ } ;
923+ }
924+ return {
925+ success : false as const ,
926+ error : "Transient error" ,
927+ errorCode : "TRIGGER_ERROR" ,
928+ } ;
929+ } ) ;
930+
931+ queue . onBatchComplete ( async ( result ) => {
932+ completionResult = result ;
933+ } ) ;
934+
935+ await queue . initializeBatch ( createInitOptions ( "batch1" , "env1" , 4 ) ) ;
936+ await enqueueItems ( queue , "batch1" , "env1" , createBatchItems ( 4 ) ) ;
937+
938+ await vi . waitFor (
939+ ( ) => {
940+ expect ( completionResult ) . not . toBeNull ( ) ;
941+ } ,
942+ { timeout : 5000 }
943+ ) ;
944+
945+ // Even-indexed items should fast-fail (1 attempt each)
946+ expect ( itemAttempts . get ( 0 ) ) . toBe ( 1 ) ;
947+ expect ( itemAttempts . get ( 2 ) ) . toBe ( 1 ) ;
948+ // Odd-indexed items should exhaust the retry ladder
949+ expect ( itemAttempts . get ( 1 ) ) . toBe ( maxAttempts ) ;
950+ expect ( itemAttempts . get ( 3 ) ) . toBe ( maxAttempts ) ;
951+
952+ expect ( completionResult ! . failedRunCount ) . toBe ( 4 ) ;
953+ } finally {
954+ await queue . close ( ) ;
955+ }
956+ }
957+ ) ;
958+ } ) ;
779959} ) ;
0 commit comments