import { cacheFetch , cacheHash , cachePut } from "../cache/backfill" ; import { Config } from "../types/Config" ; import { getDepsForTarget } from "../task/getDepsForTarget" ; import { getPackageAndTask } from "../task/taskId" ; import { getWorkspace } from "../workspace/getWorkspace" ; import { WorkerQueue , WorkerQueueJob } from "../task/WorkerQueue" ; import { logger } from "../logger" ; import { Pipeline , START_TARGET_ID } from "../task/Pipeline" ; import { PipelineTarget } from "../types/PipelineDefinition" ; import { TaskLogger } from "../logger/TaskLogger" ; import pLimit from "p-limit" ; /** Arbitrary parallel limit for network downloads of caches */ const NetworkParallelLimit = 15 ; /** Worker command */ export async function worker ( cwd , config ) { logger . verbose ( "worker started" ) ; const workspace = getWorkspace ( cwd , config ) ; const pipeline = new Pipeline ( workspace , config ) ; const networkLimit = pLimit ( NetworkParallelLimit ) ; /** the worker queue processor */ const processor = async ( job ) => { const id = job . name ; const { packageName , task } = getPackageAndTask ( id ) ; if ( ! pipeline . targets ! . has ( id ) ) { logger . info ( `pipeline doesn't have  ${ id }` ) ; return ; } const target = pipeline . targets . get ( id ) ! ; const deps = getDepsForTarget ( id , pipeline . dependencies ) ; const taskLogger = new TaskLogger ( packageName || "[GLOBAL]" , task ) ; taskLogger . info ( `[${ job . id }] started` ) ; await Promise . all ( deps . filter ( ( d ) => d !== START_TARGET_ID ) . map ( ( depTargetId ) => { return networkLimit ( ( ) => getCache ( pipeline . targets . get ( depTargetId ) ! , workspace . root , config ) ) ; } ) ) ; taskLogger . info ( `[${ job . id }] restored all dependent cache` ) ; const cacheResult = await getCache ( target , workspace . root , config ) ; if ( cacheResult . cacheHit ) { taskLogger . info ( `[${ job . id }] skipped` ) ; return { hash : cacheResult . hash , id : target . id , cwd : target . cwd , outputGlob : target . outputGlob } ; } try { let result ; if ( target . run ) { if ( target . packageName ) { result = target . run ( { packageName : target . packageName , config , cwd : target . cwd , options : target . options , taskName : getPackageAndTask ( target . id ) . task , logger : taskLogger , } ) ; } else { result = target . run ( { config , cwd : target . cwd , options : target . options , logger : taskLogger , } ) ; } } if ( result && typeof result [ "then" ] === "function" ) { await result ; } await saveCache ( cacheResult . hash , target , config ) ; taskLogger . info ( `[${ job . id }] done` ) ; return { hash : cacheResult . hash , id : target . id , cwd : target . cwd , outputGlob : target . outputGlob } ; } catch ( e ) { taskLogger . error ( `[${ job . id }] failed` ) ; throw new Error ( taskLogger . getLogs ( ) . map ( ( e ) => e . msg ) . join ( "\n" ) ) ; } } ; const workerQueue = new WorkerQueue ( config ) ; await workerQueue . initializeAsWorker ( processor ) ; logger . verbose ( "worker queue initialized" ) ; } /********************************************
 * Utilities to support caching of the worker
 *********************************************/ param target param config returns function getCacheOptions ( target , config ) { return { ... config . cacheOptions , outputGlob : target . outputGlob || config . cacheOptions . outputGlob , } ; } async function getCache ( target , root , config ) { let hash = null ; let cacheHit = false ; const { id , cwd } = target ; const cacheOptions = getCacheOptions ( target , config ) ; hash = await cacheHash ( id , root , cwd , cacheOptions , config . args ) ; if ( hash && ! config . resetCache ) { cacheHit = await cacheFetch ( hash , id , cwd , config . cacheOptions ) ; } return { hash , cacheHit } ; } async function saveCache ( hash , target , config ) { const cacheOptions = getCacheOptions ( target , config ) ; await cachePut ( hash , target . cwd , cacheOptions ) ; }