import { Redis } from "ioredis" ; import { v4 as uuidv4 } from "uuid" ; /**
 *
 * The generic idea of the Deduplicator(tm) is very simple: while executing the job, Deduplicator raises a flag (in redis)
 * to inform other job runners not to run the same job ("same"-ness is defined by a key). At the end of the execution it
 * removes the previously raised flag.
 *
 * As all "very simple" ideas, it has several challenges, namely:
 *
 * THE FIRST CHALLENGE is: what if the worker dies (e.g. the EC2 node is rotated) during the execution? This might
 * have a flag raised but not being deleted.
 *
 * The implemented solution: save the timestamp of the flag and keep updating it in the job runner while it is doing its
 * job. If it dies, the flag will be invalidated in some reasonable time.
 *
 * THE SECOND CHALLENGE is: let's say a job runner observes the flag, what should it do? Reschedule the job or drop it?
 *
 * Always rescheduling is bad: if we have the issue with duplicates, this will keep queue increasing, not good for Redis'
 * CPU.
 *
 * Always dropping is dangerous: what if the flag belongs to a "dead" worker, however it wasn't invalidated yet?
 *
 * The implemented solution: once the worker observes the flag of a concurrent worker, it will watch for it for
 * some period of time and if it is moving (timestamp is being updated, which means the worker is doing "something"),
 * we are safe to discard the job as a detected duplicate; however if it is not moving it is dangerous to discard it
 * and should be rescheduled instead.
 *
 */ type InProgressStorage = { param jobKey param workerId setInProgressFlag : ( jobKey : string , jobRunnerId : string ) => Promise < void > ; param jobKey removeInProgressFlag : ( jobKey : string ) => Promise < void > ; param jobKey param invalidatingTimestamp return hasInProgressFlag : ( jobKey : string , invalidatingTimestamp : number ) => Promise < string | null > ; param jobKey param jobRunnerId param jobRunnerFlagUpdateTimeoutMsecs return isJobRunnerLive : ( jobKey : string , jobRunnerId : string , jobRunnerFlagUpdateTimeoutMsecs : number ) => Promise < boolean > ; } ; function sleep ( ms ) { return new Promise ( resolve => setTimeout ( resolve , ms ) ) ; } type Flag = { jobRunnerId : string ; timestamp : number ; } ; export class RedisInProgressStorageWithTimeout implements InProgressStorage { private redis : Redis ; constructor ( redis ) { this . redis = redis ; } hasInProgressFlag ( jobKey , invalidatingTimestamp ) { return this . redis . get ( jobKey ) . then ( json => { if ( ! json ) { return null ; } const flag = JSON . parse ( json ) as Flag ; const isStaled = ( new Date ( ) . getTime ( ) - flag . timestamp ) >= invalidatingTimestamp ; if ( isStaled ) { return null ; } return flag . jobRunnerId ; } ) ; } async removeInProgressFlag ( jobKey ) { await this . redis . unlink ( jobKey ) ; } async setInProgressFlag ( jobKey , jobRunnerId ) { const flag = { jobRunnerId : jobRunnerId , timestamp : new Date ( ) . getTime ( ) } ; const REDIS_CLEANUP_TIMEOUT = 24 * 3600 * 1000 ; await this . redis . set ( jobKey , JSON . stringify ( flag ) , 'px' , REDIS_CLEANUP_TIMEOUT ) ; } async isJobRunnerLive ( jobKey , jobRunnerId , jobRunnerFlagUpdateTimeoutMsecs ) { const flagOld = await this . redis . get ( jobKey ) ; if ( ! flagOld ) { return false ; } const flagOldParsed = JSON . parse ( flagOld ) as Flag ; if ( flagOldParsed . jobRunnerId !== jobRunnerId ) { return false ; } if ( jobRunnerFlagUpdateTimeoutMsecs > 5000 ) { throw new Error ( "One shouldn't block the execution for too long!" ) ; } await sleep ( jobRunnerFlagUpdateTimeoutMsecs * 2 ) ; const flagNew = await this . redis . get ( jobKey ) ; if ( ! flagNew ) { return false ; } const flagNewParsed = JSON . parse ( flagNew ) as Flag ; return flagNewParsed . jobRunnerId === jobRunnerId && flagOldParsed . timestamp < flagNewParsed . timestamp ; } } export enum DeduplicatorResult { E_OK , E_OTHER_WORKER_DOING_THIS_JOB , E_NOT_SURE_TRY_AGAIN_LATER } export class Deduplicator { private inProgressStorage : InProgressStorage ; private jobRunnerFlagUpdateTimeoutMsecs : number ; constructor ( inProgressStorage , jobRunnerFlagUpdateTimeoutMsecs ) { this . inProgressStorage = inProgressStorage ; this . jobRunnerFlagUpdateTimeoutMsecs = Math . min ( jobRunnerFlagUpdateTimeoutMsecs , 5000 ) ; } /**
     * Executes the job if and only if it can determine that there's no other worker running the job (by checking the
     * flag in the storage). If it can't tell and there's a flag, tries to understand if that other worker is live or
     * this is some weird state.
     *
     * During the execution, refreshes the flag each jobRunnerFlagUpdateTimeoutMsecs to keep other workers informed.
     *
     * Based on the outcome of the call, you might decide to do the following:
     * 	E_OK - the job has been processed, the flag was dropped, ANOTHER JOB with the same key can be scheduled and
     * 				 it WILL BE PROCESSED NEXT
     *  E_OTHER_WORKER_DOING_THIS_JOB - we have determined that the current task has been processed right now,
     *  																you might want to DISCARD the job to avoid duplication
     *  E_UNSURE_TRY_AGAIN_LATER - we are not sure if the other process is processing the job right now or not; you might like to
     *  					  						   RESCHEDULE the task for good (remember there's a timeout for the flag)
     *  <an exception is thrown> - same as E_OK with the only difference that the underlying job threw an exception; the
     *  													 flag was removed.
     */ async executeWithDeduplication ( jobKey , jobRunner ) { const jobRunnerId = 'jobRunnerId-' + uuidv4 ( ) ; const inProgressFlagWorkerId = await this . inProgressStorage . hasInProgressFlag ( jobKey , this . jobRunnerFlagUpdateTimeoutMsecs * 10 ) ; if ( inProgressFlagWorkerId ) { if ( await this . inProgressStorage . isJobRunnerLive ( jobKey , inProgressFlagWorkerId , this . jobRunnerFlagUpdateTimeoutMsecs ) ) { return DeduplicatorResult . E_OTHER_WORKER_DOING_THIS_JOB ; } return DeduplicatorResult . E_NOT_SURE_TRY_AGAIN_LATER ; } await this . inProgressStorage . setInProgressFlag ( jobKey , jobRunnerId ) ; const intervalIdx = setInterval ( async ( ) => { await this . inProgressStorage . setInProgressFlag ( jobKey , jobRunnerId ) ; } , this . jobRunnerFlagUpdateTimeoutMsecs ) ; try { await jobRunner ( ) ; return DeduplicatorResult . E_OK ; } finally { clearInterval ( intervalIdx ) ; await this . inProgressStorage . removeInProgressFlag ( jobKey ) ; } } }