import { filter , take } from 'rxjs/operators' ; import { pipe } from 'fp-ts/lib/pipeable' ; import { Option , map as mapOptional , getOrElse , isSome } from 'fp-ts/lib/Option' ; import uuid from 'uuid' ; import { pick } from 'lodash' ; import { merge , Subject } from 'rxjs' ; import agent from 'elastic-apm-node' ; import { Logger } from '../../../../src/core/server' ; import { asOk , either , map , mapErr , promiseResult , isErr } from './lib/result_type' ; import { isTaskRunEvent , isTaskClaimEvent , isTaskRunRequestEvent , RanTask , ErroredTask , OkResultOf , ErrResultOf , ClaimTaskErr , TaskClaimErrorType , } from './task_events' ; import { Middleware } from './lib/middleware' ; import { ConcreteTaskInstance , TaskInstanceWithId , TaskInstanceWithDeprecatedFields , TaskLifecycle , TaskLifecycleResult , TaskStatus , EphemeralTask , } from './task' ; import { TaskStore } from './task_store' ; import { ensureDeprecatedFieldsAreCorrected } from './lib/correct_deprecated_fields' ; import { TaskLifecycleEvent , TaskPollingLifecycle } from './polling_lifecycle' ; import { TaskTypeDictionary } from './task_type_dictionary' ; import { EphemeralTaskLifecycle } from './ephemeral_task_lifecycle' ; import { EphemeralTaskRejectedDueToCapacityError } from './task_running' ; const VERSION_CONFLICT_STATUS = 409 ; export interface TaskSchedulingOpts { logger : Logger ; taskStore : TaskStore ; taskPollingLifecycle : TaskPollingLifecycle ; ephemeralTaskLifecycle : EphemeralTaskLifecycle ; middleware : Middleware ; definitions : TaskTypeDictionary ; taskManagerId : string ; } export interface RunNowResult { id : ConcreteTaskInstance [ 'id' ] ; state ? : ConcreteTaskInstance [ 'state' ] ; } export class TaskScheduling { private store : TaskStore ; private taskPollingLifecycle : TaskPollingLifecycle ; private ephemeralTaskLifecycle : EphemeralTaskLifecycle ; private logger : Logger ; private middleware : Middleware ; private definitions : TaskTypeDictionary ; private taskManagerId : string ; /**
     * Initializes the task manager, preventing any further addition of middleware,
     * enabling the task manipulation methods, and beginning the background polling
     * mechanism.
     */ constructor ( opts ) { this . logger = opts . logger ; this . middleware = opts . middleware ; this . taskPollingLifecycle = opts . taskPollingLifecycle ; this . ephemeralTaskLifecycle = opts . ephemeralTaskLifecycle ; this . store = opts . taskStore ; this . definitions = opts . definitions ; this . taskManagerId = opts . taskManagerId ; } param task returns { Promise < ConcreteTaskInstance > } public async schedule ( taskInstance , options ? ) { const { taskInstance : modifiedTask } = await this . middleware . beforeSave ( { ... options , taskInstance : ensureDeprecatedFieldsAreCorrected ( taskInstance , this . logger ) , } ) ; const traceparent = agent . currentTransaction && agent . currentTransaction . type !== 'request' ? agent . currentTraceparent : '' ; return await this . store . schedule ( { ... modifiedTask , traceparent : traceparent || '' , } ) ; } param taskId returns { Promise < ConcreteTaskInstance > } public async runNow ( taskId ) { return new Promise ( async ( resolve , reject ) => { try { this . awaitTaskRunResult ( taskId ) . then ( ( { id } ) => resolve ( { id , } ) ) . catch ( reject ) ; this . taskPollingLifecycle . attemptToRun ( taskId ) ; } catch ( error ) { reject ( error ) ; } } ) ; } param task returns { Promise < ConcreteTaskInstance > } public async ephemeralRunNow ( task , options ? ) { const id = uuid . v4 ( ) ; const { taskInstance : modifiedTask } = await this . middleware . beforeSave ( { ... options , taskInstance : task , } ) ; return new Promise ( async ( resolve , reject ) => { try { const { cancel , resolveOnCancel } = cancellablePromise ( ) ; this . awaitTaskRunResult ( id , resolveOnCancel ) . then ( ( arg ) => { resolve ( arg ) ; } ) . catch ( ( err ) => { reject ( err ) ; } ) ; const attemptToRunResult = this . ephemeralTaskLifecycle . attemptToRun ( { id , scheduledAt : new Date ( ) , runAt : new Date ( ) , status : TaskStatus . Idle , ownerId : this . taskManagerId , ... modifiedTask , } ) ; if ( isErr ( attemptToRunResult ) ) { cancel ( ) ; reject ( new EphemeralTaskRejectedDueToCapacityError ( `Ephemeral Task of type ${ task . taskType } was rejected` , task ) ) ; } } catch ( error ) { reject ( error ) ; } } ) ; } param task returns { Promise < TaskInstanceWithId > } public async ensureScheduled ( taskInstance , options ? ) { try { return await this . schedule ( taskInstance , options ) ; } catch ( err ) { if ( err . statusCode === VERSION_CONFLICT_STATUS ) { return taskInstance ; } throw err ; } } private awaitTaskRunResult ( taskId , cancel ? ) { return new Promise ( ( resolve , reject ) => { const subscription = merge ( this . taskPollingLifecycle . events , this . ephemeralTaskLifecycle . events ) . pipe ( filter ( ( { id } ) => id === taskId ) ) . subscribe ( ( taskEvent ) => { if ( isTaskClaimEvent ( taskEvent ) ) { mapErr ( async ( error ) => { subscription . unsubscribe ( ) ; if ( isSome ( error . task ) && error . errorType === TaskClaimErrorType . CLAIMED_BY_ID_OUT_OF_CAPACITY ) { const task = error . task . value ; const definition = this . definitions . get ( task . taskType ) ; return reject ( new Error ( `Failed to run task "${ taskId }" as we would exceed the max concurrency of "${ definition ?. title ?? task . taskType }" which is ${ definition ?. maxConcurrency }. Rescheduled the task to ensure it is picked up as soon as possible.` ) ) ; } else { return reject ( await this . identifyTaskFailureReason ( taskId , error . task ) ) ; } } , taskEvent . event ) ; } else { either < OkResultOf < TaskLifecycleEvent > , ErrResultOf < TaskLifecycleEvent > > ( taskEvent . event , ( taskInstance ) => { if ( isTaskRunEvent ( taskEvent ) ) { subscription . unsubscribe ( ) ; resolve ( pick ( ( taskInstance as RanTask ) . task , [ 'id' , 'state' ] ) ) ; } } , async ( errorResult ) => { subscription . unsubscribe ( ) ; return reject ( new Error ( `Failed to run task "${ taskId }": ${ isTaskRunRequestEvent ( taskEvent ) ? `Task Manager is at capacity, please try again later` : isTaskRunEvent ( taskEvent ) ? `${ ( errorResult as ErroredTask ) . error }` : `${ errorResult }` }` ) ) ; } ) ; } } ) ; if ( cancel ) { cancel . then ( ( ) => { subscription . unsubscribe ( ) ; } ) ; } } ) ; } private async identifyTaskFailureReason ( taskId , error ) { return map ( await pipe ( error , mapOptional ( async ( taskReturnedBySweep ) => asOk ( taskReturnedBySweep . status ) ) , getOrElse ( ( ) => promiseResult < TaskLifecycle , Error > ( this . store . getLifecycle ( taskId ) ) ) ) , ( taskLifecycleStatus ) => { if ( taskLifecycleStatus === TaskLifecycleResult . NotFound ) { return new Error ( `Failed to run task "${ taskId }" as it does not exist` ) ; } else if ( taskLifecycleStatus === TaskStatus . Running || taskLifecycleStatus === TaskStatus . Claiming ) { return new Error ( `Failed to run task "${ taskId }" as it is currently running` ) ; } return new Error ( `Failed to run task "${ taskId }" for unknown reason (Current Task Lifecycle is "${ taskLifecycleStatus }")` ) ; } , ( getLifecycleError ) => new Error ( `Failed to run task "${ taskId }" and failed to get current Status:${ getLifecycleError }` ) ) ; } } const cancellablePromise = ( ) => { const boolStream = new Subject < boolean > ( ) ; return { cancel : ( ) => boolStream . next ( true ) , resolveOnCancel : boolStream . pipe ( take ( 1 ) ) . toPromise ( ) . then ( ( ) => { } ) , } ; } ;