refactor: update shutdown process

packages/auth/src/app.ts

@@ -52,7 +52,6 @@ import { InteractionRoutes } from './interaction/routes'
 
 export interface AppContextData extends DefaultContext {
   logger: Logger
-  closeEmitter: EventEmitter
   container: AppContainer
   // Set by @koa/router
   params: { [key: string]: string }
@@ -88,7 +87,6 @@ export interface DatabaseCleanupRule {
 export interface AppServices {
   logger: Promise<Logger>
   knex: Promise<Knex>
-  closeEmitter: Promise<EventEmitter>
   config: Promise<IAppConfig>
   clientService: Promise<ClientService>
   grantService: Promise<GrantService>
@@ -106,7 +104,6 @@ export class App {
   private introspectionServer!: Server
   private adminServer!: Server
   public apolloServer!: ApolloServer
-  private closeEmitter!: EventEmitter
   private logger!: Logger
   private config!: IAppConfig
   private databaseCleanupRules!: {
@@ -124,7 +121,6 @@ export class App {
    */
   public async boot(): Promise<void> {
     this.config = await this.container.use('config')
-    this.closeEmitter = await this.container.use('closeEmitter')
     this.logger = await this.container.use('logger')
 
     this.databaseCleanupRules = {
@@ -374,7 +370,6 @@ export class App {
     const koa = new Koa<DefaultState, AppContext>()
 
     koa.context.container = this.container
-    koa.context.closeEmitter = await this.container.use('closeEmitter')
     koa.context.logger = await this.container.use('logger')
 
     koa.use(
@@ -402,7 +397,6 @@ export class App {
   public async shutdown(): Promise<void> {
     return new Promise((resolve): void => {
       this.isShuttingDown = true
-      this.closeEmitter.emit('shutdown')
 
       if (this.adminServer) {
         this.adminServer.close((): void => {

packages/auth/src/index.ts

@@ -73,8 +73,6 @@ export function initIocContainer(
     return db
   })
 
-  container.singleton('closeEmitter', async () => new EventEmitter())
-
   container.singleton('openPaymentsClient', async (deps) => {
     const logger = await deps.use('logger')
     return createOpenPaymentsClient({ logger })

packages/auth/src/tests/context.ts

@@ -36,7 +36,6 @@ export function createContext<T extends AppContext>(
   ctx.params = params
   ctx.query = reqOpts.query || {}
   ctx.session = { ...req.session }
-  ctx.closeEmitter = new EventEmitter()
   ctx.container = container
   return ctx as T
 }

packages/backend/src/app.ts

@@ -1,7 +1,7 @@
 import { join } from 'path'
 import http, { Server } from 'http'
-import { EventEmitter } from 'events'
 import { ParsedUrlQuery } from 'querystring'
+import { Client as TigerbeetleClient } from 'tigerbeetle-node'
 
 import { IocContract } from '@adonisjs/fold'
 import { Knex } from 'knex'
@@ -74,10 +74,10 @@ import { CombinedPaymentService } from './open_payments/payment/combined/service
 import { FeeService } from './fee/service'
 import { AutoPeeringService } from './auto-peering/service'
 import { AutoPeeringRoutes } from './auto-peering/routes'
+import { Rafiki as ConnectorApp } from './connector/core'
 
 export interface AppContextData {
   logger: Logger
-  closeEmitter: EventEmitter
   container: AppContainer
   // Set by @koa/router.
   params: { [key: string]: string }
@@ -177,7 +177,6 @@ const PAYMENT_POINTER_PATH = '/:paymentPointerPath+'
 export interface AppServices {
   logger: Promise<Logger>
   knex: Promise<Knex>
-  closeEmitter: Promise<EventEmitter>
   config: Promise<IAppConfig>
   httpTokenService: Promise<HttpTokenService>
   assetService: Promise<AssetService>
@@ -207,21 +206,21 @@ export interface AppServices {
   feeService: Promise<FeeService>
   autoPeeringService: Promise<AutoPeeringService>
   autoPeeringRoutes: Promise<AutoPeeringRoutes>
+  connectorApp: Promise<ConnectorApp>
+  tigerbeetle: Promise<TigerbeetleClient>
 }
 
 export type AppContainer = IocContract<AppServices>
 
 export class App {
   private openPaymentsServer!: Server
+  private ilpConnectorService!: Server
   private adminServer!: Server
   private autoPeeringServer!: Server
   public apolloServer!: ApolloServer
-  public closeEmitter!: EventEmitter
   public isShuttingDown = false
   private logger!: Logger
   private config!: IAppConfig
-  private outgoingPaymentTimer!: NodeJS.Timer
-  private deactivateInvoiceTimer!: NodeJS.Timer
 
   public constructor(private container: IocContract<AppServices>) {}
 
@@ -233,7 +232,6 @@ export class App {
    */
   public async boot(): Promise<void> {
     this.config = await this.container.use('config')
-    this.closeEmitter = await this.container.use('closeEmitter')
     this.logger = await this.container.use('logger')
 
     // Workers are in the way during tests.
@@ -253,7 +251,7 @@ export class App {
     }
   }
 
-  public async startAdminServer(port: number | string): Promise<void> {
+  public async startAdminServer(port: number): Promise<void> {
     const koa = await this.createKoaServer()
     const httpServer = http.createServer(koa.callback())
 
@@ -321,7 +319,7 @@ export class App {
     this.adminServer = httpServer.listen(port)
   }
 
-  public async startOpenPaymentsServer(port: number | string): Promise<void> {
+  public async startOpenPaymentsServer(port: number): Promise<void> {
     const koa = await this.createKoaServer()
 
     const router = new Router<DefaultState, AppContext>()
@@ -482,7 +480,7 @@ export class App {
     this.openPaymentsServer = koa.listen(port)
   }
 
-  public async startAutoPeeringServer(port: number | string): Promise<void> {
+  public async startAutoPeeringServer(port: number): Promise<void> {
     const koa = await this.createKoaServer()
 
     const autoPeeringRoutes = await this.container.use('autoPeeringRoutes')
@@ -497,28 +495,37 @@ export class App {
     this.autoPeeringServer = koa.listen(port)
   }
 
+  public async startIlpConnectorServer(port: number): Promise<void> {
+    const ilpConnectorService = await this.container.use('connectorApp')
+    this.ilpConnectorService = ilpConnectorService.listenPublic(port)
+  }
+
   public async shutdown(): Promise<void> {
-    return new Promise((resolve): void => {
-      this.isShuttingDown = true
-      this.closeEmitter.emit('shutdown')
+    this.isShuttingDown = true
 
-      if (this.openPaymentsServer) {
-        this.openPaymentsServer.close((): void => {
-          resolve()
-        })
-      }
+    if (this.openPaymentsServer) {
+      await this.stopServer(this.openPaymentsServer)
+    }
+    if (this.adminServer) {
+      await this.stopServer(this.adminServer)
+    }
+    if (this.ilpConnectorService) {
+      await this.stopServer(this.ilpConnectorService)
+    }
+    if (this.autoPeeringServer) {
+      await this.stopServer(this.autoPeeringServer)
+    }
+  }
 
-      if (this.adminServer) {
-        this.adminServer.close((): void => {
-          resolve()
-        })
-      }
+  private async stopServer(server: Server): Promise<void> {
+    return new Promise((resolve, reject) => {
+      server.close((err) => {
+        if (err) {
+          reject(err)
+        }
 
-      if (this.autoPeeringServer) {
-        this.autoPeeringServer.close((): void => {
-          resolve()
-        })
-      }
+        resolve()
+      })
     })
   }
 
@@ -583,7 +590,6 @@ export class App {
     const koa = new Koa<DefaultState, AppContext>()
 
     koa.context.container = this.container
-    koa.context.closeEmitter = await this.container.use('closeEmitter')
     koa.context.logger = await this.container.use('logger')
 
     koa.use(

packages/backend/src/index.ts

@@ -1,5 +1,3 @@
-import { EventEmitter } from 'events'
-import { Server } from 'http'
 import path from 'path'
 import createLogger from 'pino'
 import { knex } from 'knex'
@@ -55,7 +53,6 @@ BigInt.prototype.toJSON = function () {
 
 const container = initIocContainer(Config)
 const app = new App(container)
-let connectorServer: Server
 
 export function initIocContainer(
   config: typeof Config
@@ -106,7 +103,6 @@ export function initIocContainer(
     )
     return db
   })
-  container.singleton('closeEmitter', async () => new EventEmitter())
   container.singleton('redis', async (deps): Promise<Redis> => {
     const config = await deps.use('config')
     return new Redis(config.redisUrl, {
@@ -445,19 +441,12 @@ export const gracefulShutdown = async (
   const logger = await container.use('logger')
   logger.info('shutting down.')
   await app.shutdown()
-  if (connectorServer) {
-    await new Promise((resolve, reject) => {
-      connectorServer.close((err?: Error) =>
-        err ? reject(err) : resolve(null)
-      )
-    })
-  }
   const knex = await container.use('knex')
   await knex.destroy()
   const tigerbeetle = await container.use('tigerbeetle')
-  await tigerbeetle.destroy()
+  tigerbeetle.destroy()
   const redis = await container.use('redis')
-  await redis.disconnect()
+  redis.disconnect()
 }
 
 export const start = async (
@@ -493,6 +482,15 @@ export const start = async (
     logger.info('received SIGTERM attempting graceful shutdown')
 
     try {
+      if (shuttingDown) {
+        logger.warn(
+          'received second SIGTERM during graceful shutdown, exiting forcefully.'
+        )
+        process.exit(1)
+      }
+
+      shuttingDown = true
+
       // Graceful shutdown
       await gracefulShutdown(container, app)
       logger.info('completed graceful shutdown.')
@@ -518,12 +516,12 @@ export const start = async (
   const config = await container.use('config')
   await app.boot()
   await app.startAdminServer(config.adminPort)
-  await app.startOpenPaymentsServer(config.openPaymentsPort)
   logger.info(`Admin listening on ${app.getAdminPort()}`)
+
+  await app.startOpenPaymentsServer(config.openPaymentsPort)
   logger.info(`Open Payments listening on ${app.getOpenPaymentsPort()}`)
 
-  const connectorApp = await container.use('connectorApp')
-  connectorServer = connectorApp.listenPublic(config.connectorPort)
+  await app.startIlpConnectorServer(config.connectorPort)
   logger.info(`Connector listening on ${config.connectorPort}`)
   logger.info('🐒 has 🚀. Get ready for 🍌🍌🍌🍌🍌')
 

packages/backend/src/shared/ilp_plugin.ts

@@ -1,4 +1,4 @@
-import { LiquidityAccount } from '../accounting/service'
+import { ConnectorAccount } from '../connector/core'
 
 // Maybe @interledger/pay should export this interface.
 export interface IlpPlugin {
@@ -37,6 +37,6 @@ export class OutgoingIlpPlugin implements IlpPlugin {
 }
 
 export interface IlpPluginOptions {
-  sourceAccount: LiquidityAccount
+  sourceAccount: ConnectorAccount
   unfulfillable?: boolean
 }

packages/backend/src/tests/context.ts

@@ -1,4 +1,3 @@
-import EventEmitter from 'events'
 import * as httpMocks from 'node-mocks-http'
 import Koa from 'koa'
 import { AppContext, AppContextData, AppRequest } from '../app'
@@ -18,6 +17,5 @@ export function createContext<T extends AppContext>(
   if (reqOpts.body !== undefined) {
     ctx.request.body = reqOpts.body
   }
-  ctx.closeEmitter = new EventEmitter()
   return ctx as T
 }