import crypto from 'crypto' ; import { v4 as uuidv4 } from 'uuid' ; import { Datastore , Key } from '@google-cloud/datastore' ; import { logger } from 'gcf-utils' ; const DEFAULT_LOCK_EXPIRY = 20 * 1000 ; const MAX_LOCK_EXPIRY = 60 * 1000 ; const DEFAULT_LOCK_ACQUIRE_TIMEOUT = 120 * 1000 ; const BACKOFF_INITIAL_DELAY = 2 * 1000 ; const BACKOFF_MAX_DELAY = 10 * 1000 ; const DATASTORE_LOCK_ERROR_NAME = 'DatastoreLockError' ; enum AcquireResult { Success = 1 , Failure , LockActive } const sleep = ( ms ) => { return new Promise ( r => setTimeout ( r , ms ) ) ; } ; /**
 * A simple lock backed by Cloud Datastore
 */ export class DatastoreLock { private uniqueId : string ; private kind : string ; private target : string ; private datastore : Datastore ; private key : Key ; private lockExpiry : number ; private lockAcquireTimeout : number ; constructor ( lockId , target , lockExpiry = DEFAULT_LOCK_EXPIRY , lockAcquireTimeout = DEFAULT_LOCK_ACQUIRE_TIMEOUT ) { if ( lockExpiry > MAX_LOCK_EXPIRY ) { throw new Error ( `lockExpiry is too long, max is ${ MAX_LOCK_EXPIRY }, ` + `given ${ lockExpiry }` ) ; } this . datastore = new Datastore ( ) ; this . kind = `ds-lock-${ lockId }` ; this . target = target ; const hash = crypto . createHash ( 'sha1' ) ; hash . update ( this . target ) ; this . key = this . datastore . key ( [ this . kind , hash . digest ( 'hex' ) ] ) ; this . uniqueId = uuidv4 ( ) ; this . lockExpiry = lockExpiry ; this . lockAcquireTimeout = lockAcquireTimeout ; } /**
     * Acquire the lock.
     */ public async acquire ( ) { const startTime = Date . now ( ) ; let waitTime = BACKOFF_INITIAL_DELAY + Math . floor ( Math . random ( ) * 1000 ) ; const maxDelay = BACKOFF_MAX_DELAY + Math . floor ( Math . random ( ) * 1000 ) ; while ( Date . now ( ) - startTime < this . lockAcquireTimeout ) { const result = await this . _acquire ( ) ; if ( result === AcquireResult . Success ) { return true ; } if ( result === AcquireResult . Failure ) { logger . debug ( `Failed to acquire the lock, retrying after ${ waitTime } miliseconds` ) ; } if ( result === AcquireResult . LockActive ) { logger . debug ( `The lock is active, retrying after ${ waitTime } miliseconds` ) ; } await sleep ( waitTime ) ; waitTime = Math . min ( maxDelay , waitTime + Math . floor ( Math . random ( ) * 2000 ) ) ; } return false ; } private async _acquire ( ) { const transaction = this . datastore . transaction ( ) ; try { await transaction . run ( ) ; const [ entity ] = await transaction . get ( this . key ) ; logger . debug ( entity ) ; if ( entity === undefined || Date . now ( ) > entity . expiry ) { const entity = { key : this . key , data : { expiry : Date . now ( ) + this . lockExpiry , uuid : this . uniqueId , } , } ; transaction . save ( entity ) ; logger . debug ( entity ) ; await transaction . commit ( ) ; return AcquireResult . Success ; } else { await transaction . rollback ( ) ; return AcquireResult . LockActive ; } } catch ( e ) { const err = e as Error ; err . message = `Error acquiring a lock for ${ this . target }: ${ err . message }` ; logger . error ( err ) ; await transaction . rollback ( ) ; return AcquireResult . Failure ; } } /**
     * Release the lock.
     */ public async release ( ) { const transaction = this . datastore . transaction ( ) ; try { await transaction . run ( ) ; const [ entity ] = await transaction . get ( this . key ) ; if ( entity === undefined ) { await transaction . rollback ( ) ; return true ; } if ( entity . uuid !== this . uniqueId ) { const err = new Error ( `The lock for ${ this . target } was acquired by another process.` ) ; err . name = DATASTORE_LOCK_ERROR_NAME ; throw err ; } transaction . delete ( this . key ) ; await transaction . commit ( ) ; return true ; } catch ( e ) { const err = e as Error ; if ( err . name === DATASTORE_LOCK_ERROR_NAME ) { throw err ; } err . message = `Error releasing a lock for ${ this . target }: ${ err . message }` ; logger . error ( err ) ; await transaction . rollback ( ) ; return false ; } } }