import { getRandStr , sleep } from '@cromwell/core' ; import { serverMessages } from '@cromwell/core-backend/dist/helpers/constants' ; import { getLogger } from '@cromwell/core-backend/dist/helpers/logger' ; import { getServerBuildPath } from '@cromwell/core-backend/dist/helpers/paths' ; import { ChildProcess , fork } from 'child_process' ; import fs from 'fs-extra' ; import tcpPortUsed from 'tcp-port-used' ; import { restartMessage } from './constants' ; import { loadEnv } from './settings' ; const logger = getLogger ( ) ; /**
 * Proxy manager
 */ type ServerInfo = { id ? : string ; port ? : number ; childInst ? : ChildProcess ; } ; const activeServer = { } ; let isRestarting = false ; let isPendingRestart = false ; const madeServers = { } ; export const getServerPort = ( ) => activeServer . port ; /** Used only at Proxy startup. Should not be called from any other place */ export const launchServerManager = async ( proxyPort ? , init ? ) => { activeServer . proxyPort = proxyPort ; if ( activeServer . port ) { logger . warn ( 'Proxy manager: called launch, but Server was already launched!' ) ; } try { const info = await makeServer ( init ) ; updateActiveServer ( info ) ; } catch ( error ) { logger . error ( 'Proxy manager could not launch API server' , error ) ; } } ; const makeServer = async ( init ? ) : Promise < ServerInfo > => { logger . info ( 'Proxy manager: Launching new API server...' ) ; const info = { } ; const env = loadEnv ( ) ; const serverId = getRandStr ( 8 ) ; info . id = serverId ; const buildPath = getServerBuildPath ( ) ; if ( ! buildPath || ! fs . existsSync ( buildPath ) ) { const msg = 'Proxy manager: could not find server build at: ' + buildPath ; logger . error ( msg ) ; throw new Error ( msg ) ; } const serverProc = fork ( buildPath , [ env . scriptName , activeServer . proxyPort ? `proxy-port=${ activeServer . proxyPort }` : '' , init ? '--init' : '' , ] , { stdio : 'inherit' , cwd : process . cwd ( ) } ) ; parentRegisterChild ( serverProc , info ) ; info . childInst = serverProc ; let hasReported = false ; await new Promise ( done => { setTimeout ( ( ) => { if ( ! hasReported ) done ( false ) ; } , 15000 ) ; serverProc . on ( 'message' , ( message ) => { let msg ; try { msg = JSON . parse ( String ( message ) ) ; } catch ( error ) { logger . error ( error ) ; } if ( msg . message === restartMessage ) { restartServer ( ) ; } if ( msg . message === serverMessages . onStartMessage ) { if ( process . send ) process . send ( serverMessages . onStartMessage ) ; info . port = msg . port ; hasReported = true ; done ( true ) ; } if ( msg . message === serverMessages . onStartErrorMessage ) { if ( process . send ) process . send ( serverMessages . onStartErrorMessage ) ; hasReported = true ; done ( false ) ; } } ) ; } ) ; if ( ! info . port ) throw new Error ( 'Proxy manager: Failed to start API server' ) ; try { await tcpPortUsed . waitUntilUsed ( info . port , 500 , 8000 ) ; } catch ( error ) { logger . error ( error ) ; } madeServers [ serverId ] = info ; return info ; } ; const closeServer = async ( info ) => { logger . info ( `Proxy manager: killing API server at port: ${ info . port }...` ) ; info . childInst ?. kill ( ) ; } ; /** Safely restarts server instance */ const restartServer = async ( ) => { if ( isRestarting ) { isPendingRestart = true ; return ; } logger . info ( 'Proxy manager: restarting API server...' ) ; isRestarting = true ; let newServer ; const oldServer = { ... activeServer } ; try { newServer = await makeServer ( ) ; } catch ( error ) { logger . error ( 'Proxy manager: Failed to launch new API server' , error ) ; isRestarting = false ; if ( isPendingRestart ) { isPendingRestart = false ; await restartServer ( ) ; } return ; } await sleep ( 0.5 ) ; updateActiveServer ( newServer ) ; await sleep ( 4 ) ; try { await closeServer ( oldServer ) ; await tcpPortUsed . waitUntilFree ( oldServer . port , 500 , 5000 ) ; } catch ( error ) { logger . error ( 'Proxy manager: Failed to kill old API server at ' + oldServer . port , error ) ; } isRestarting = false ; if ( isPendingRestart ) { isPendingRestart = false ; await restartServer ( ) ; } } ; const updateActiveServer = ( info ) => { Object . keys ( info ) . forEach ( key => { activeServer [ key ] = info [ key ] ; } ) ; } ; export const serverAliveWatcher = async ( ) => { await sleep ( 30 ) ; let isAlive = true ; if ( ! activeServer . port ) { isAlive = false ; } try { if ( activeServer . port ) { isAlive = await tcpPortUsed . check ( activeServer . port , '127.0.0.1' ) ; } } catch ( error ) { } if ( ! isAlive ) { logger . error ( 'Proxy manager watcher: API Server is not alive. Restarting...' ) ; await restartServer ( ) ; } for ( const info of Object . values ( madeServers ) ) { if ( info ?. port && info . port !== activeServer . port && info . childInst ) { if ( await checkServerAlive ( info ) ) { setTimeout ( async ( ) => { try { if ( info ?. port && info . port !== activeServer . port && await tcpPortUsed . check ( info . port , '127.0.0.1' ) ) { await closeServer ( info ) ; info . childInst = undefined ; info . port = undefined ; } } catch ( error ) { logger . error ( error ) ; } } , 50000 ) ; } } } serverAliveWatcher ( ) ; } ; const checkServerAlive = async ( info ) => { if ( info ?. port ) { try { if ( await tcpPortUsed . check ( info . port , '127.0.0.1' ) ) { return true ; } } catch ( error ) { logger . error ( error ) ; } } return false ; } ; const onMessageCallbacks = [ ] ; export const childRegister = ( port ) => { updateActiveServer ( { port } ) ; if ( process . send ) process . send ( JSON . stringify ( { message : serverMessages . onStartMessage , port : port , } ) ) ; process . on ( 'message' , ( m ) => { onMessageCallbacks . forEach ( cb => cb ?. ( m ) ) ; } ) ; } ; type IPCMessageType = 'make-new' | 'apply-new' | 'restart-me' | 'kill-me' | 'success' | 'failed' ; type IPCMessage = { id : string ; message : IPCMessageType ; payload ? : any ; port ? : number ; } ; export const childSendMessage = async ( message , payload ? ) : Promise < IPCMessage > => { const messageId = getRandStr ( 8 ) ; let responseResolver ; const responsePromise = new Promise < IPCMessage > ( res => responseResolver = res ) ; const cb = ( message ) => { message = JSON . parse ( message ) ; if ( message . id === messageId ) { responseResolver ( message ) ; } } ; onMessageCallbacks . push ( cb ) ; if ( process . send ) process . send ( JSON . stringify ( { id : messageId , port : activeServer . port , message , payload , } as IPCMessage ) ) ; const resp = await responsePromise ; delete onMessageCallbacks [ onMessageCallbacks . indexOf ( cb ) ] ; return resp ; } ; const parentRegisterChild = ( child , childInfo ) => { child . on ( 'message' , async ( msg ) => { try { const message = JSON . parse ( msg ) ; if ( message . message === 'make-new' ) { try { const info = await makeServer ( ) ; if ( ! info . port ) throw new Error ( '!info.port' ) ; if ( ! ( await checkServerAlive ( info ) ) ) throw new Error ( 'new server is not alive' ) ; if ( await checkServerAlive ( childInfo ) ) { child . send ( JSON . stringify ( { id : message . id , message : 'success' , payload : info . id , } as IPCMessage ) ) ; } return ; } catch ( error ) { if ( await checkServerAlive ( childInfo ) ) { child . send ( JSON . stringify ( { id : message . id , message : 'failed' , } as IPCMessage ) ) ; } logger . error ( error ) ; return ; } } if ( message . message === 'apply-new' && message . payload ) { const port = madeServers [ message . payload ] ?. port ; if ( port ) { const isAlive = await checkServerAlive ( { port } ) ; if ( ! isAlive ) { if ( await checkServerAlive ( childInfo ) ) { child . send ( JSON . stringify ( { id : message . id , message : 'failed' , } as IPCMessage ) ) ; } return ; } else { updateActiveServer ( madeServers [ message . payload ] ) ; await sleep ( 1 ) ; if ( await checkServerAlive ( childInfo ) ) { child . send ( JSON . stringify ( { id : message . id , message : 'success' , } as IPCMessage ) ) ; } return ; } } } if ( message . message === 'kill-me' ) { await closeServer ( { childInst : child } ) ; return ; } if ( message . message === 'restart-me' ) { await restartServer ( ) ; return ; } if ( message . id ) { if ( await checkServerAlive ( childInfo ) ) { child . send ( JSON . stringify ( { id : message . id , message : 'failed' , } as IPCMessage ) ) ; } } } catch ( error ) { logger . error ( error ) ; } } ) ; } ; export const closeAllServers = ( ) => { Object . values ( madeServers ) . forEach ( closeServer ) ; } ;