Add updated flush handling on sigterm

packages/next/src/cli/next-dev.ts

@@ -60,9 +60,10 @@ const handleSessionStop = async () => {
         durationMilliseconds: Date.now() - sessionStarted,
         pagesDir,
         appDir,
-      })
+      }),
+      true
     )
-    await telemetry.flush()
+    telemetry.flushDetached('dev', dir)
   } catch (err) {
     console.error(err)
   }

packages/next/src/telemetry/deteched-flush.ts

@@ -0,0 +1,45 @@
+import fs from 'fs'
+import path from 'path'
+import { Telemetry, TelemetryEvent } from './storage'
+import loadConfig from '../server/config'
+import { getProjectDir } from '../lib/get-project-dir'
+import { PHASE_DEVELOPMENT_SERVER } from '../shared/lib/constants'
+
+// this process should be started with following arg order
+// 1. mode e.g. dev, export, start
+// 2. project dir
+;(async () => {
+  const args = [...process.argv]
+  let dir = args.pop()
+  const mode = args.pop()
+
+  if (!dir || mode !== 'dev') {
+    throw new Error(
+      `Invalid flags should be run as node detached-flush dev ./path-to/project`
+    )
+  }
+  dir = getProjectDir(dir)
+
+  const config = await loadConfig(PHASE_DEVELOPMENT_SERVER, dir)
+  const distDir = path.join(dir, config.distDir || '.next')
+  const eventsPath = path.join(distDir, '_events.json')
+
+  let events: TelemetryEvent[]
+  try {
+    events = JSON.parse(fs.readFileSync(eventsPath, 'utf8'))
+  } catch (err: any) {
+    if (err.code === 'ENOENT') {
+      // no events to process we can exit now
+      process.exit(0)
+    }
+    throw err
+  }
+
+  const telemetry = new Telemetry({ distDir })
+  await telemetry.record(events)
+  await telemetry.flush()
+
+  // finished flushing events clean-up/exit
+  fs.unlinkSync(eventsPath)
+  process.exit(0)
+})()

packages/next/src/telemetry/post-payload.ts

@@ -1,7 +1,7 @@
 import retry from 'next/dist/compiled/async-retry'
 import fetch from 'next/dist/compiled/node-fetch'
 
-export function _postPayload(endpoint: string, body: object) {
+export function _postPayload(endpoint: string, body: object, signal?: any) {
   return (
     retry(
       () =>
@@ -10,6 +10,7 @@ export function _postPayload(endpoint: string, body: object) {
           body: JSON.stringify(body),
           headers: { 'content-type': 'application/json' },
           timeout: 5000,
+          signal,
         }).then((res) => {
           if (!res.ok) {
             const err = new Error(res.statusText)

packages/next/src/telemetry/storage.ts

@@ -8,6 +8,9 @@ import { getAnonymousMeta } from './anonymous-meta'
 import * as ciEnvironment from './ci-info'
 import { _postPayload } from './post-payload'
 import { getRawProjectId } from './project-id'
+import { AbortController } from 'next/dist/compiled/@edge-runtime/primitives/abort-controller'
+import fs from 'fs'
+import spawn from 'next/dist/compiled/cross-spawn'
 
 // This is the key that stores whether or not telemetry is enabled or disabled.
 const TELEMETRY_KEY_ENABLED = 'telemetry.enabled'
@@ -26,7 +29,7 @@ const TELEMETRY_KEY_ID = `telemetry.anonymousId`
 // See the `oneWayHash` function.
 const TELEMETRY_KEY_SALT = `telemetry.salt`
 
-type TelemetryEvent = { eventName: string; payload: object }
+export type TelemetryEvent = { eventName: string; payload: object }
 type EventContext = {
   anonymousId: string
   projectId: string
@@ -57,6 +60,7 @@ function getStorageDirectory(distDir: string): string | undefined {
 
 export class Telemetry {
   private conf: Conf<any> | null
+  private distDir: string
   private sessionId: string
   private rawProjectId: string
   private NEXT_TELEMETRY_DISABLED: any
@@ -69,6 +73,7 @@ export class Telemetry {
     const { NEXT_TELEMETRY_DISABLED, NEXT_TELEMETRY_DEBUG } = process.env
     this.NEXT_TELEMETRY_DISABLED = NEXT_TELEMETRY_DISABLED
     this.NEXT_TELEMETRY_DEBUG = NEXT_TELEMETRY_DEBUG
+    this.distDir = distDir
     const storageDirectory = getStorageDirectory(distDir)
 
     try {
@@ -177,15 +182,23 @@ export class Telemetry {
   }
 
   record = (
-    _events: TelemetryEvent | TelemetryEvent[]
+    _events: TelemetryEvent | TelemetryEvent[],
+    deferred?: boolean
   ): Promise<RecordObject> => {
-    const _this = this
-    // pseudo try-catch
-    async function wrapper() {
-      return await _this.submitRecord(_events)
-    }
-
-    const prom = wrapper()
+    const prom = (
+      deferred
+        ? // if we know we are going to immediately call
+          // flushDetached we can skip starting the initial
+          // submitRecord which will then be cancelled
+          new Promise((resolve) =>
+            resolve({
+              isFulfilled: true,
+              isRejected: false,
+              value: _events,
+            })
+          )
+        : this.submitRecord(_events)
+    )
       .then((value) => ({
         isFulfilled: true,
         isRejected: false,
@@ -203,6 +216,8 @@ export class Telemetry {
         return res
       })
 
+    ;(prom as any)._events = Array.isArray(_events) ? _events : [_events]
+    ;(prom as any)._controller = (prom as any)._controller
     // Track this `Promise` so we can flush pending events
     this.queue.add(prom)
 
@@ -211,6 +226,35 @@ export class Telemetry {
 
   flush = async () => Promise.all(this.queue).catch(() => null)
 
+  // writes current events to disk and spawns separate
+  // detached process to submit the records without blocking
+  // the main process from exiting
+  flushDetached = (mode: 'dev', dir: string) => {
+    const allEvents: TelemetryEvent[] = []
+
+    this.queue.forEach((item: any) => {
+      try {
+        item._controller?.abort()
+        allEvents.push(...item._events)
+      } catch (_) {
+        // if we fail to abort ignore this event
+      }
+    })
+    fs.writeFileSync(
+      path.join(this.distDir, '_events.json'),
+      JSON.stringify(allEvents)
+    )
+
+    spawn('node', [require.resolve('./deteched-flush'), mode, dir], {
+      detached: !this.NEXT_TELEMETRY_DEBUG,
+      ...(this.NEXT_TELEMETRY_DEBUG
+        ? {
+            stdio: 'inherit',
+          }
+        : {}),
+    })
+  }
+
   private submitRecord = (
     _events: TelemetryEvent | TelemetryEvent[]
   ): Promise<any> => {
@@ -248,13 +292,20 @@ export class Telemetry {
       sessionId: this.sessionId,
     }
     const meta: EventMeta = getAnonymousMeta()
-    return _postPayload(`https://telemetry.nextjs.org/api/v1/record`, {
-      context,
-      meta,
-      events: events.map(({ eventName, payload }) => ({
-        eventName,
-        fields: payload,
-      })) as Array<EventBatchShape>,
-    })
+    const postController = new AbortController()
+    const res = _postPayload(
+      `https://telemetry.nextjs.org/api/v1/record`,
+      {
+        context,
+        meta,
+        events: events.map(({ eventName, payload }) => ({
+          eventName,
+          fields: payload,
+        })) as Array<EventBatchShape>,
+      },
+      postController.signal
+    )
+    res._controller = postController
+    return res
   }
 }

test/integration/telemetry/test/index.test.js

@@ -480,6 +480,10 @@ describe('Telemetry CLI', () => {
 
       expect(event1).toMatch(/"pagesDir": true/)
       expect(event1).toMatch(/"turboFlag": true/)
+
+      expect(await fs.pathExists(path.join(appDir, '.next/_events.json'))).toBe(
+        false
+      )
     } finally {
       await teardown()
     }
@@ -516,6 +520,10 @@ describe('Telemetry CLI', () => {
       expect(event1).toMatch(/"turboFlag": false/)
       expect(event1).toMatch(/"pagesDir": true/)
       expect(event1).toMatch(/"appDir": true/)
+
+      expect(await fs.pathExists(path.join(appDir, '.next/_events.json'))).toBe(
+        false
+      )
     } finally {
       await teardown()
     }