chore: flush pending GQL executor jobs before graceful shutdown

packages/gql-executor/gqlExecutor.ts

@@ -6,9 +6,9 @@ import executeGraphQL from '../server/graphql/executeGraphQL'
 import '../server/initSentry'
 import '../server/monkeyPatchFetch'
 import {GQLRequest} from '../server/types/custom'
+import {Logger} from '../server/utils/Logger'
 import RedisInstance from '../server/utils/RedisInstance'
 import RedisStream from './RedisStream'
-import {Logger} from '../server/utils/Logger'
 
 tracer.init({
   service: `gql`,
@@ -29,26 +29,43 @@ const run = async () => {
   const publisher = new RedisInstance('gql_pub')
   const subscriber = new RedisInstance('gql_sub')
   const executorChannel = GQLExecutorChannelId.join(SERVER_ID!)
+  let activeJobCount = 0
 
-  // on shutdown, remove consumer from the group
+  // on shutdown, remove consumer from the group wait for current jobs to complete
   process.on('SIGTERM', async (signal) => {
-    Logger.log(`Server ID: ${SERVER_ID}. Kill signal received: ${signal}, starting graceful shutdown.`)
+    const MAX_SHUTDOWN_TIME = 40000
+    const SHUTDOWN_CHECK_INTERVAL = 1000
+    let start = Date.now()
+    Logger.log(
+      `Server ID: ${SERVER_ID}. Kill signal received: ${signal}, starting graceful shutdown.`
+    )
+
     await publisher.xgroup(
       'DELCONSUMER',
       ServerChannel.GQL_EXECUTOR_STREAM,
       ServerChannel.GQL_EXECUTOR_CONSUMER_GROUP,
       executorChannel
     )
-    Logger.log(`Server ID: ${SERVER_ID}. Graceful shutdown complete, exiting.`)
-    process.exit()
+
+    setInterval(() => {
+      if (Date.now() - start >= MAX_SHUTDOWN_TIME) {
+        Logger.log(`Server ID: ${SERVER_ID}. Graceful shutdown timed out, exiting.`)
+        process.exit()
+      } else if (activeJobCount <= 0) {
+        Logger.log(`Server ID: ${SERVER_ID}. Graceful shutdown complete, exiting.`)
+        process.exit()
+      }
+    }, SHUTDOWN_CHECK_INTERVAL)
   })
 
   // subscribe to direct messages
   const onMessage = async (_channel: string, message: string) => {
     const {jobId, socketServerId, request} = JSON.parse(message) as PubSubPromiseMessage
+    activeJobCount++
     const response = await executeGraphQL(request)
     const channel = SocketServerChannelId.join(socketServerId)
     publisher.publish(channel, JSON.stringify({response, jobId}))
+    activeJobCount--
   }
 
   subscriber.on('message', onMessage)

packages/server/utils/Logger.ts

@@ -21,6 +21,10 @@ function trace(level: LogLevel, message: any, ...optionalParameters: any[]) {
 
   if (span) {
     tracer.inject(span.context(), formats.LOG, record)
+    const tags = optionalParameters.find((param) => param.tags) as Record<string, any> | undefined
+    if (tags && typeof tags === 'object') {
+      span.addTags(tags)
+    }
   }
 
   LogFun[level](JSON.stringify(record))

packages/server/utils/publishInternalGQL.ts

@@ -30,7 +30,9 @@ const publishInternalGQL = async <NarrowResponse>(options: Options) => {
         userId: getUserId(authToken),
         tags: {
           authToken: JSON.stringify(authToken),
-          query: query || ''
+          query: query || '',
+          variables: JSON.stringify(variables),
+          socketServerId: socketId
         }
       })
     } else {

packages/server/utils/sendToSentry.ts

@@ -14,8 +14,8 @@ export interface SentryOptions {
 
 // Even though this is a promise we'll never need to await it, so we'll never need to worry about catching an error
 const sendToSentry = async (error: Error, options: SentryOptions = {}) => {
-  Logger.log('SEND TO SENTRY', error || JSON.stringify(error))
   const {sampleRate, tags, extras, userId, ip} = options
+  Logger.log('SEND TO SENTRY', error || JSON.stringify(error), {tags})
   if (sampleRate && Math.random() > sampleRate) return
   const fullUser = userId ? await getUserById(userId) : null
   const user = fullUser ? {id: fullUser.id, email: fullUser.email} : null