import { BackoffRetryStrategy , CappedDelayRateLimitStrategy , JobStore , RateLimitState , Step , StepPrioritizer , StepProcessor , StepResult } from "../../../src/backfill/looper/api" ; import { Looper } from "../../../src/backfill/looper/looper" ; import { getLogger } from "../../../src/config/logger" ; const logger = getLogger ( "looper.test.ts" ) ; describe ( "Looper" , ( ) => { let now = new Date ( 2021 , 10 , 10 , 10 , 10 , 10 ) ; const nowFunction = ( ) => now ; const aMinuteFromNow = new Date ( 2021 , 10 , 10 , 10 , 11 , 10 ) ; const DATABASE = { rateLimitState : { budgetLeft : 5 , refreshDate : aMinuteFromNow } , jobState : { cursor : 0 } , failedAttemptsCount : 0 } ; type MyJobId = string ; type MyJobState = { cursor : number ; } ; class MyProcessor implements StepProcessor < MyJobState > { private readonly failingCursors : number [ ] = [ ] ; itemsProcessed : number [ ] = [ ] ; constructor ( failingCursors ) { this . failingCursors = failingCursors ; } process ( jobState , rateLimitState ) { rateLimitState . budgetLeft -- ; if ( this . failingCursors . includes ( jobState . cursor ) ) { return { success : false , rateLimit : rateLimitState , error : { isRetryable : true , message : "retryable error" , isFatal : false } , jobState } ; } this . itemsProcessed . push ( jobState . cursor ) ; jobState . cursor ++ ; return { success : true , rateLimit : rateLimitState , jobState } ; } } class MyPrioritizer implements StepPrioritizer < MyJobId , MyJobState > { private readonly processor : StepProcessor < MyJobState > ; constructor ( processor ) { this . processor = processor ; } getStepProcessor ( _ , jobState , __ ? ) { if ( jobState . cursor < 10 ) { return this . processor ; } else { return undefined ; } } skip ( _ , jobState , __ ? ) { if ( jobState . cursor < 10 ) { jobState . cursor ++ ; } return jobState ; } } class MyJobStore implements JobStore < MyJobId , MyJobState > { getFailedAttemptsCount ( _ ) { return DATABASE . failedAttemptsCount ; } getJobState ( _ ) { return DATABASE . jobState ; } getRateLimitState ( _ ) { return DATABASE . rateLimitState ; } setFailedAttemptsCount ( _ , count ) { DATABASE . failedAttemptsCount = count ; } updateRateLimitState ( _ , rateLimitState ) { DATABASE . rateLimitState = rateLimitState ; } setJobState ( _ , jobState ) { DATABASE . jobState = jobState ; } } const jobStore = new MyJobStore ( ) ; const processor = new MyProcessor ( [ 3 , 7 ] ) ; const prioritizer = new MyPrioritizer ( processor ) ; const looper = new Looper < MyJobId , MyJobState > ( prioritizer , jobStore , new CappedDelayRateLimitStrategy ( 15 * 60 , nowFunction ) , new BackoffRetryStrategy ( 3 , 2 , 2 ) , logger ) ; it ( "completes a loop through a job" , async ( ) => { let nextAction = looper . processStep ( { jobId : "job1" } ) ; let i = 1 ; let retryCount = 0 ; let ratelimitedCount = 0 ; while ( nextAction . scheduleNextStep ) { logger . info ( `STATE AFTER STEP ${ i }: ${ JSON . stringify ( DATABASE , null , 2 ) }` ) ; if ( nextAction . delay ) { logger . info ( `DELAYING BY ${ nextAction . delay . seconds } seconds due to ${ nextAction . delay . reason }!` ) ; now = new Date ( now . getTime ( ) + nextAction . delay . seconds * 1000 ) ; if ( nextAction . delay . reason == "retry" ) { retryCount ++ ; } if ( nextAction . delay . reason == "rate-limit" ) { ratelimitedCount ++ ; logger . info ( "REFRESHING RATE LIMIT" ) ; DATABASE . rateLimitState . refreshDate = new Date ( now . getTime ( ) + 60 * 1000 ) ; DATABASE . rateLimitState . budgetLeft = 5 ; } } nextAction = looper . processStep ( { jobId : "job1" } ) ; i ++ ; } expect ( DATABASE . jobState . cursor ) . toBe ( 10 ) ; expect ( processor . itemsProcessed ) . toHaveLength ( 8 ) ; expect ( processor . itemsProcessed ) . toEqual ( expect . arrayContaining ( [ 1 , 2 , 4 , 5 , 6 , 8 , 9 ] ) ) ; expect ( processor . itemsProcessed ) . not . toContain ( 3 ) ; expect ( processor . itemsProcessed ) . not . toContain ( 7 ) ; expect ( retryCount ) . toBe ( 6 ) ; expect ( ratelimitedCount ) . toBe ( 3 ) ; } ) ; } ) ;