import * as assert from "assert" ; import { Pool } from "pg" ; import { cronItemMatches } from "./cronMatcher" ; import { parseCrontab } from "./crontab" ; import defer from "./deferred" ; import getCronItems from "./getCronItems" ; import { Cron , CronJob , JobAndCronIdentifier , KnownCrontab , ParsedCronItem , RunnerOptions , TimestampDigest , WorkerEvents , } from "./interfaces" ; import { processSharedOptions , Releasers } from "./lib" ; interface CronRequirements { pgPool : Pool ; events : WorkerEvents ; } /**
 * This function looks through all the cron items we have (e.g. from our
 * crontab file) and compares them to the items we already know about. If the
 * item is not previously know, we add it to the list `unknownIdentifiers` so
 * that it can be recorded in the database (i.e. it will be "known" from now
 * on). If the item was previously known, we add an entry to
 * `backfillItemsAndDates` indicating the `item` and earliest time
 * (`notBefore`) that a backfill should operate from. This is later compared
 * to the configuration to see how much backfilling to do.
 */ function getBackfillAndUnknownItems ( parsedCronItems , knownCrontabs ) { const backfillItemsAndDates = [ ] ; const unknownIdentifiers = [ ] ; for ( const item of parsedCronItems ) { const known = knownCrontabs . find ( ( record ) => record . identifier === item . identifier ) ; if ( known ) { const notBefore = known . last_execution || known . known_since ; backfillItemsAndDates . push ( { item , notBefore , } ) ; } else { unknownIdentifiers . push ( item . identifier ) ; } } return { backfillItemsAndDates , unknownIdentifiers } ; } /**
 * Rounds the incoming date to the nearest minute (either rounding up or down).
 * Tagged "unsafe" because it mutates the argument, this is desired for
 * performance but may be unexpected.
 */ function unsafeRoundToMinute ( ts , roundUp = false ) { if ( ts . getUTCSeconds ( ) > 0 || ts . getUTCMilliseconds ( ) > 0 ) { ts . setUTCSeconds ( 0 ) ; ts . setUTCMilliseconds ( 0 ) ; if ( roundUp ) { ts . setUTCMinutes ( ts . getUTCMinutes ( ) + 1 ) ; } } return ts ; } function makeJobForItem ( item , ts , backfilled = false ) { return { task : item . task , payload : { ... item . payload , _cron : { ts , backfilled , } , } , queueName : item . options . queueName , runAt : ts , maxAttempts : item . options . maxAttempts , priority : item . options . priority , } ; } /**
 * Schedules a list of cron jobs all due at the same timestamp. Jobs that were
 * already scheduled (e.g. via a different Worker instance) will be skipped
 * automatically.
 */ async function scheduleCronJobs ( pgPool , escapedWorkerSchema , jobsAndIdentifiers , ts , useNodeTime ) { await pgPool . query ( `
      with specs as (
        select
          index,
          (json->>'identifier')::text as identifier,
          ((json->'job')->>'task')::text as task,
          ((json->'job')->'payload')::json as payload,
          ((json->'job')->>'queueName')::text as queue_name,
          ((json->'job')->>'runAt')::timestamptz as run_at,
          ((json->'job')->>'maxAttempts')::int as max_attempts,
          ((json->'job')->>'priority')::int as priority
        from json_array_elements($1::json) with ordinality AS entries (json, index)
      ),
      locks as (
        insert into ${ escapedWorkerSchema }.known_crontabs (identifier, known_since, last_execution)
        select
          specs.identifier,
          $2 as known_since,
          $2 as last_execution
        from specs
        on conflict (identifier)
        do update set last_execution = excluded.last_execution
        where (known_crontabs.last_execution is null or known_crontabs.last_execution < excluded.last_execution)
        returning known_crontabs.identifier
      )
      select
        ${ escapedWorkerSchema }.add_job(
          specs.task,
          specs.payload,
          specs.queue_name,
          coalesce(specs.run_at, $3::timestamptz, now()),
          specs.max_attempts,
          null, -- job key
          specs.priority
        )
      from specs
      inner join locks on (locks.identifier = specs.identifier)
      order by specs.index asc
    ` , [ JSON . stringify ( jobsAndIdentifiers ) , ts , useNodeTime ? new Date ( ) . toISOString ( ) : null , ] ) ; } /**
 * Marks any previously unknown crontab identifiers as now being known. Then
 * performs backfilling on any crontab tasks that need it.
 */ async function registerAndBackfillItems ( { pgPool , events , cron } , escapedWorkerSchema , parsedCronItems , startTime , useNodeTime ) { const { rows } = await pgPool . query < KnownCrontab > ( `SELECT * FROM ${ escapedWorkerSchema }.known_crontabs` ) ; const { backfillItemsAndDates , unknownIdentifiers , } = getBackfillAndUnknownItems ( parsedCronItems , rows ) ; if ( unknownIdentifiers . length ) { await pgPool . query ( `
      INSERT INTO ${ escapedWorkerSchema }.known_crontabs (identifier, known_since)
      SELECT identifier, $2
      FROM unnest($1::text[]) AS unnest (identifier)
      ON CONFLICT DO NOTHING
      ` , [ unknownIdentifiers , startTime . toISOString ( ) ] ) ; } const largestBackfill = parsedCronItems . reduce ( ( largest , item ) => Math . max ( item . options . backfillPeriod , largest ) , 0 ) ; if ( largestBackfill > 0 ) { const unsafeTs = new Date ( + startTime - largestBackfill ) ; unsafeRoundToMinute ( unsafeTs , true ) ; while ( unsafeTs < startTime ) { const timeAgo = + startTime - + unsafeTs ; const ts = unsafeTs . toISOString ( ) ; const digest = digestTimestamp ( unsafeTs ) ; const itemsToBackfill = [ ] ; for ( const { item , notBefore } of backfillItemsAndDates ) { if ( item . options . backfillPeriod >= timeAgo && unsafeTs >= notBefore && cronItemMatches ( item , digest ) ) { itemsToBackfill . push ( { identifier : item . identifier , job : makeJobForItem ( item , ts , true ) , } ) ; } } if ( itemsToBackfill . length ) { events . emit ( "cron:backfill" , { cron , itemsToBackfill , timestamp : ts , } ) ; await scheduleCronJobs ( pgPool , escapedWorkerSchema , itemsToBackfill , ts , useNodeTime ) ; } unsafeTs . setUTCMinutes ( unsafeTs . getUTCMinutes ( ) + 1 ) ; } } } /** One minute in milliseconds */ const ONE_MINUTE = 60 * 1000 ; internal param options param parsedCronItems param requirements export const runCron = ( options , parsedCronItems , requirements ) : Cron => { const { pgPool } = requirements ; const { logger , escapedWorkerSchema , events , useNodeTime , } = processSharedOptions ( options ) ; const promise = defer ( ) ; let released = false ; let timeout = null ; let stopCalled = false ; function stop ( e ? ) { if ( ! stopCalled ) { stopCalled = true ; if ( timeout ) { clearTimeout ( timeout ) ; timeout = null ; } if ( e ) { promise . reject ( e ) ; } else { promise . resolve ( ) ; } } else { logger . error ( "Graphile Worker internal bug in src/cron.ts: calling `stop()` more than once shouldn't be possible. Please report this." ) ; } } async function cronMain ( ) { if ( released ) { return stop ( ) ; } const start = new Date ( ) ; events . emit ( "cron:starting" , { cron : this , start } ) ; await registerAndBackfillItems ( { pgPool , events , cron : this } , escapedWorkerSchema , parsedCronItems , new Date ( + start ) , useNodeTime ) ; events . emit ( "cron:started" , { cron : this , start } ) ; if ( released ) { return stop ( ) ; } let nextTimestamp = unsafeRoundToMinute ( new Date ( + start ) , true ) ; const scheduleNextLoop = ( ) => { if ( released ) { return stop ( ) ; } timeout = setTimeout ( ( ) => { timeout = null ; loop ( ) ; } , Math . max ( + nextTimestamp - Date . now ( ) + 1 , 1 ) ) ; } ; async function loop ( ) { try { if ( released ) { return stop ( ) ; } const digest = digestTimestamp ( nextTimestamp ) ; const ts = nextTimestamp . toISOString ( ) ; const expectedTimestamp = + nextTimestamp ; const currentTimestamp = Date . now ( ) ; const roundedCurrentTimestamp = currentTimestamp - ( currentTimestamp % ONE_MINUTE ) ; if ( roundedCurrentTimestamp < expectedTimestamp ) { logger . debug ( `Graphile Worker Cron fired ${ ( ( expectedTimestamp - currentTimestamp ) / 1000 ) . toFixed ( 3 ) }s too early (clock skew?); rescheduling` , { expectedTimestamp , currentTimestamp , } ) ; events . emit ( "cron:prematureTimer" , { cron : this , currentTimestamp , expectedTimestamp , } ) ; scheduleNextLoop ( ) ; return ; } else if ( roundedCurrentTimestamp > expectedTimestamp ) { logger . debug ( `Graphile Worker Cron fired too late; catching up (${ Math . floor ( ( currentTimestamp - expectedTimestamp ) / ONE_MINUTE ) }m${ Math . floor ( ( ( currentTimestamp - expectedTimestamp ) % ONE_MINUTE ) / 1000 ) }s behind)` ) ; events . emit ( "cron:overdueTimer" , { cron : this , currentTimestamp , expectedTimestamp , } ) ; } const jobsAndIdentifiers = [ ] ; for ( const item of parsedCronItems ) { if ( cronItemMatches ( item , digest ) ) { jobsAndIdentifiers . push ( { identifier : item . identifier , job : makeJobForItem ( item , ts ) , } ) ; } } if ( jobsAndIdentifiers . length ) { events . emit ( "cron:schedule" , { cron : this , timestamp : expectedTimestamp , jobsAndIdentifiers , } ) ; await scheduleCronJobs ( pgPool , escapedWorkerSchema , jobsAndIdentifiers , ts , useNodeTime ) ; events . emit ( "cron:scheduled" , { cron : this , timestamp : expectedTimestamp , jobsAndIdentifiers , } ) ; if ( released ) { return stop ( ) ; } } nextTimestamp . setUTCMinutes ( nextTimestamp . getUTCMinutes ( ) + 1 ) ; scheduleNextLoop ( ) ; } catch ( e ) { return stop ( e ) ; } } scheduleNextLoop ( ) ; } cronMain ( ) . catch ( stop ) ; return { release ( ) { if ( ! released ) { released = true ; if ( timeout ) { stop ( ) ; } } return promise ; } , promise , } ; } ; export async function getParsedCronItemsFromOptions ( options , releasers ) { const { crontabFile , parsedCronItems , crontab } = options ; if ( ! crontabFile && ! parsedCronItems && ! crontab ) { return [ ] ; } if ( crontab ) { assert ( ! crontabFile , "`crontab` and `crontabFile` must not be set at the same time." ) ; assert ( ! parsedCronItems , "`crontab` and `parsedCronItems` must not be set at the same time." ) ; return parseCrontab ( crontab ) ; } else if ( crontabFile ) { assert ( ! parsedCronItems , "`crontabFile` and `parsedCronItems` must not be set at the same time." ) ; const watchedCronItems = await getCronItems ( options , crontabFile , false ) ; releasers . push ( ( ) => watchedCronItems . release ( ) ) ; return watchedCronItems . items ; } else { assert ( parsedCronItems != null , "Expected `parsedCronItems` to be set." ) ; assert ( Array . isArray ( parsedCronItems ) , "Expected `parsedCronItems` to be an array; you must use a helper e.g. `parseCrontab()` or `parseCronItems()` to produce this value." ) ; const firstItem = parsedCronItems [ 0 ] ; if ( firstItem ) { if ( ! Array . isArray ( firstItem . minutes ) || ! Array . isArray ( firstItem . hours ) || ! Array . isArray ( firstItem . dates ) || ! Array . isArray ( firstItem . months ) || ! Array . isArray ( firstItem . dows ) ) { throw new Error ( "Invalid `parsedCronItems`; you must use a helper e.g. `parseCrontab()` or `parseCronItems()` to produce this value." ) ; } } return parsedCronItems ; } } /**
 * Digests a timestamp into its min/hour/date/month/dow components that are
 * needed for cron matching.
 *
 * WARNING: the timestamp passed into this function might be mutated later, so
 * **do not memoize** using the value itself. If memoization is necessary, it
 * could be done using `+ts` as the key.
 */ function digestTimestamp ( ts ) { const min = ts . getUTCMinutes ( ) ; const hour = ts . getUTCHours ( ) ; const date = ts . getUTCDate ( ) ; const month = ts . getUTCMonth ( ) + 1 ; const dow = ts . getUTCDay ( ) ; return { min , hour , date , month , dow } ; }