feat: pass output flusher to callChild (#5677)

* feat: pass output flusher to plugins

* refactor: move file to TypeScript

Author: eduardoboucas

packages/build/src/log/output_flusher.ts

@@ -1,5 +1,8 @@
+import { stdout, stderr } from 'process'
 import { Transform } from 'stream'
 
+import type { StandardStreams } from './stream.js'
+
 const flusherSymbol = Symbol.for('@netlify/output-gate')
 
 /**
@@ -46,3 +49,18 @@ export class OutputFlusherTransform extends Transform {
     callback()
   }
 }
+
+export const getStandardStreams = (outputFlusher?: OutputFlusher): StandardStreams => {
+  if (!outputFlusher) {
+    return {
+      stdout,
+      stderr,
+    }
+  }
+
+  return {
+    outputFlusher,
+    stdout: new OutputFlusherTransform(outputFlusher).pipe(stdout),
+    stderr: new OutputFlusherTransform(outputFlusher).pipe(stderr),
+  }
+}

packages/build/src/log/stream.js

@@ -0,0 +1,120 @@
+import { promisify } from 'util'
+
+import type { ChildProcess } from '../plugins/spawn.js'
+
+import { BufferedLogs, logsAreBuffered, Logs } from './logger.js'
+import type { OutputFlusher } from './output_flusher.js'
+
+export type StandardStreams = {
+  stderr: NodeJS.WriteStream
+  stdout: NodeJS.WriteStream
+  outputFlusher?: OutputFlusher
+}
+
+type LogsListener = (logs: string[], outputFlusher: OutputFlusher | undefined, chunk: Buffer) => void
+type LogsListeners = { stderrListener: LogsListener; stdoutListener: LogsListener }
+
+// TODO: replace with `timers/promises` after dropping Node < 15.0.0
+const pSetTimeout = promisify(setTimeout)
+
+// We try to use `stdio: inherit` because it keeps `stdout/stderr` as `TTY`,
+// which solves many problems. However we can only do it in build.command.
+// Plugins have several events, so need to be switch on and off instead.
+// In buffer mode, `pipe` is necessary.
+export const getBuildCommandStdio = function (logs: Logs) {
+  if (logsAreBuffered(logs)) {
+    return 'pipe'
+  }
+
+  return 'inherit'
+}
+
+// Add build command output
+export const handleBuildCommandOutput = function (
+  { stdout: commandStdout, stderr: commandStderr }: { stdout: string; stderr: string },
+  logs: Logs,
+) {
+  if (!logsAreBuffered(logs)) {
+    return
+  }
+
+  pushBuildCommandOutput(commandStdout, logs.stdout)
+  pushBuildCommandOutput(commandStderr, logs.stderr)
+}
+
+const pushBuildCommandOutput = function (output: string, logsArray: string[]) {
+  if (output === '') {
+    return
+  }
+
+  logsArray.push(output)
+}
+
+// Start plugin step output
+export const pipePluginOutput = function (childProcess: ChildProcess, logs: Logs, standardStreams: StandardStreams) {
+  if (!logsAreBuffered(logs)) {
+    return streamOutput(childProcess, standardStreams)
+  }
+
+  return pushOutputToLogs(childProcess, logs, standardStreams.outputFlusher)
+}
+
+// Stop streaming/buffering plugin step output
+export const unpipePluginOutput = async function (
+  childProcess: ChildProcess,
+  logs: Logs,
+  listeners: LogsListeners,
+  standardStreams: StandardStreams,
+) {
+  // Let `childProcess` `stdout` and `stderr` flush before stopping redirecting
+  await pSetTimeout(0)
+
+  if (!logsAreBuffered(logs)) {
+    return unstreamOutput(childProcess, standardStreams)
+  }
+
+  unpushOutputToLogs(childProcess, listeners.stdoutListener, listeners.stderrListener)
+}
+
+// Usually, we stream stdout/stderr because it is more efficient
+const streamOutput = function (childProcess: ChildProcess, standardStreams: StandardStreams) {
+  childProcess.stdout?.pipe(standardStreams.stdout)
+  childProcess.stderr?.pipe(standardStreams.stderr)
+}
+
+const unstreamOutput = function (childProcess: ChildProcess, standardStreams: StandardStreams) {
+  childProcess.stdout?.unpipe(standardStreams.stdout)
+  childProcess.stderr?.unpipe(standardStreams.stderr)
+}
+
+// In tests, we push to the `logs` array instead
+const pushOutputToLogs = function (
+  childProcess: ChildProcess,
+  logs: BufferedLogs,
+  outputFlusher?: OutputFlusher,
+): LogsListeners {
+  const stdoutListener = logsListener.bind(null, logs.stdout, outputFlusher)
+  const stderrListener = logsListener.bind(null, logs.stderr, outputFlusher)
+
+  childProcess.stdout?.on('data', stdoutListener)
+  childProcess.stderr?.on('data', stderrListener)
+
+  return { stdoutListener, stderrListener }
+}
+
+const logsListener: LogsListener = function (logs, outputFlusher, chunk) {
+  if (outputFlusher) {
+    outputFlusher.flush()
+  }
+
+  logs.push(chunk.toString().trimEnd())
+}
+
+const unpushOutputToLogs = function (
+  childProcess: ChildProcess,
+  stdoutListener: LogsListener,
+  stderrListener: LogsListener,
+) {
+  childProcess.stdout?.removeListener('data', stdoutListener)
+  childProcess.stderr?.removeListener('data', stderrListener)
+}

packages/build/src/log/stream.ts

@@ -22,6 +22,8 @@ import { callChild, getEventFromChild } from './ipc.js'
 import { PluginsOptions } from './node_version.js'
 import { getSpawnInfo } from './options.js'
 
+export type ChildProcess = ExecaChildProcess<string>
+
 const CHILD_MAIN_FILE = fileURLToPath(new URL('child/main.js', import.meta.url))
 
 const require = createRequire(import.meta.url)

packages/build/src/plugins/spawn.ts

@@ -3,6 +3,7 @@ import { context, propagation } from '@opentelemetry/api'
 import { addErrorInfo } from '../error/info.js'
 import { addOutputFlusher } from '../log/logger.js'
 import { logStepCompleted } from '../log/messages/ipc.js'
+import { getStandardStreams } from '../log/output_flusher.js'
 import { pipePluginOutput, unpipePluginOutput } from '../log/stream.js'
 import { callChild } from '../plugins/ipc.js'
 import { getSuccessStatus } from '../status/success.js'
@@ -38,12 +39,13 @@ export const firePluginStep = async function ({
   debug,
   verbose,
 }) {
-  const listeners = pipePluginOutput(childProcess, logs, outputFlusher)
+  const standardStreams = getStandardStreams(outputFlusher)
+  const listeners = pipePluginOutput(childProcess, logs, standardStreams)
 
   const otelCarrier = {}
   propagation.inject(context.active(), otelCarrier)
 
-  const logsA = addOutputFlusher(logs, outputFlusher)
+  const logsA = outputFlusher ? addOutputFlusher(logs, outputFlusher) : logs
 
   try {
     const configSideFiles = await listConfigSideFiles([headersPath, redirectsPath])
@@ -63,7 +65,7 @@ export const firePluginStep = async function ({
         constants,
         otelCarrier,
       },
-      logs,
+      logs: logsA,
       verbose,
     })
     const {
@@ -104,7 +106,7 @@ export const firePluginStep = async function ({
     })
     return { newError }
   } finally {
-    await unpipePluginOutput(childProcess, logs, listeners)
+    await unpipePluginOutput(childProcess, logs, listeners, standardStreams)
     logStepCompleted(logs, verbose)
   }
 }