import stats from 'stats-lite' ; import { isNumber , random } from 'lodash' ; import { merge , of , Observable , combineLatest , ReplaySubject } from 'rxjs' ; import { filter , map } from 'rxjs/operators' ; import { Option , none , some , isSome , Some } from 'fp-ts/lib/Option' ; import { isOk } from '../lib/result_type' ; import { ManagedConfiguration } from '../lib/create_managed_configuration' ; import { TaskLifecycleEvent } from '../polling_lifecycle' ; import { isTaskPollingCycleEvent } from '../task_events' ; import { ClaimAndFillPoolResult } from '../lib/fill_pool' ; import { createRunningAveragedStat } from '../monitoring/task_run_calcultors' ; /**
 * Emits a delay amount in ms to apply to polling whenever the task store exceeds a threshold of claim claimClashes
 */ export function delayOnClaimConflicts ( maxWorkersConfiguration$ , pollIntervalConfiguration$ , taskLifecycleEvents$ , claimClashesPercentageThreshold , runningAverageWindowSize ) { const claimConflictQueue = createRunningAveragedStat < number > ( runningAverageWindowSize ) ; const multiCastDelays$ = new ReplaySubject < number > ( 1 ) ; merge ( of ( 0 ) , combineLatest ( [ maxWorkersConfiguration$ , pollIntervalConfiguration$ , taskLifecycleEvents$ . pipe ( map < TaskLifecycleEvent , Option < number > > ( ( taskEvent ) => isTaskPollingCycleEvent ( taskEvent ) && isOk < ClaimAndFillPoolResult , unknown > ( taskEvent . event ) && isNumber ( taskEvent . event . value . stats ?. tasksConflicted ) ? some ( taskEvent . event . value . stats ! . tasksConflicted ) : none ) , filter < Option < number > > ( ( claimClashes ) => isSome ( claimClashes ) ) , map ( ( claimClashes ) => ( claimClashes as Some < number > ) . value ) ) , ] ) . pipe ( map ( ( [ maxWorkers , pollInterval , latestClaimConflicts ] ) => { claimConflictQueue ( latestClaimConflicts ) ; const emitWhenExceeds = ( claimClashesPercentageThreshold * maxWorkers ) / 100 ; if ( latestClaimConflicts >= emitWhenExceeds && stats . percentile ( claimConflictQueue ( ) , 0.5 ) >= emitWhenExceeds ) { return some ( pollInterval ) ; } return none ; } ) , filter < Option < number > > ( ( pollInterval ) => isSome ( pollInterval ) ) , map < Option < number > , number > ( ( maybePollInterval ) => { const pollInterval = ( maybePollInterval as Some < number > ) . value ; return random ( pollInterval * 0.25 , pollInterval * 0.75 , false ) ; } ) ) ) . subscribe ( ( delay ) => { multiCastDelays$ . next ( delay ) ; } ) ; return multiCastDelays$ ; }