Как отловить UnhandledPromiseRejectionWarning для GCS WriteStream

Question

наблюдаемое поведение приложения

Я получаю UnhandledPromiseRejectionWarning: Error: Upload failed при использовании @google-cloud/storage в node.js.

Эти ошибки появляются при обработке тысяч запросов. Это небольшой процент, вызывающий ошибки, но из-за отсутствия способности обрабатывать ошибки и отсутствия надлежащего контекста из сообщения об ошибке очень трудно определить, КОТОРЫЕ файлы терпят неудачу.

Я знаю, что обещания должны иметь .catch или быть окружены блоком try / catch. Но в этом случае я использую поток записи. Я немного сбит с толку относительно того, где на самом деле находится обещание, которое отклоняется, и как я могу его перехватить. Трассировка стека бесполезна, поскольку содержит только код библиотеки:

UnhandledPromiseRejectionWarning: Error: Upload failed
    at Request.requestStream.on.resp (.../node_modules/gcs-resumable-upload/build/src/index.js:163:34)
    at emitTwo (events.js:131:20)
    at Request.emit (events.js:214:7)
    at Request.<anonymous> (.../node_modules/request/request.js:1161:10)
    at emitOne (events.js:121:20)
    at Request.emit (events.js:211:7)
    at IncomingMessage.<anonymous> (.../node_modules/request/request.js:1083:12)
    at Object.onceWrapper (events.js:313:30)
    at emitNone (events.js:111:20)
    at IncomingMessage.emit (events.js:208:7)

Мой код

Код, который создает writeStream, выглядит следующим образом:

const {join} = require('path')
const {Storage} = require('@google-cloud/storage')

module.exports = (config) => {
  const storage = new Storage({
    projectId: config.gcloud.project,
    keyFilename: config.gcloud.auth_file
  })

  return {
    getBucketWS(path, contentType) {
      const {bucket, path_prefix} = config.gcloud

      // add path_prefix if we have one
      if (path_prefix) {
        path = join(path_prefix, path)
      }

      let setup = storage.bucket(bucket).file(path)
      let opts = {}
      if (contentType) {
        opts = {
          contentType,
          metadata: {contentType}
        }
      }
      const stream = setup.createWriteStream(opts)
      stream._bucket = bucket
      stream._path = path
      return stream
    }
  }
}

И код потребления выглядит так:

const gcs = require('./gcs-helper.js')

module.exports = ({writePath, contentType, item}, done) => {
  let ws = gcs.getBucketWS(writePath, contentType)
  ws.on('error', (err) => {
    err.message = `Could not open gs://${ws._bucket}/${ws._path}: ${err.message}`
    done(err)
  })
  ws.on('finish', () => {
    done(null, {
      path: writePath,
      item
    })
  })
  ws.write(item)
  ws.end()
}

Учитывая, что я уже слушаю событие error в потоке, я не вижу, что еще я могу здесь сделать. Там не обещание происходит на уровне @google-cloud/storage, который я потребляю.

Копание в библиотеку @ google-cloud / storage

Первая строка трассировки стека приводит нас к блоку кода в модуле узла gcs-resumable-upload, который выглядит следующим образом:

requestStream.on('complete', resp => {
    if (resp.statusCode < 200 || resp.statusCode > 299) {
        this.destroy(new Error('Upload failed'));
        return;
    }
    this.emit('metadata', resp.body);
    this.deleteConfig();
    this.uncork();
});

Это передает ошибку методу destroy в потоке. Поток создается утилитным модулем @ google-cloud / common проекта , и для создания потока используется модуль узла duplexify . Метод destroy определен в потоке duplexify и может быть найден в документации README.

Чтение дуплексного кода , я вижу, что сначала проверяется this._ondrain, прежде чем выдать ошибку. Может быть, я могу предоставить обратный вызов, чтобы избежать обработки этой ошибки?

Я попробовал ws.write(item, null, cb) и все еще получил то же предупреждение UnhandledPromiseRejection. Я попытался ws.end(item, null, cb) и даже обернул вызов .end в try catch, и в итоге получил эту ошибку, которая полностью завершила процесс:

events.js:183
      throw er; // Unhandled 'error' event
      ^

Error: The uploaded data did not match the data from the server. As a precaution, the file has been deleted. To be sure the content is the same, you should try uploading the file again.
    at delete (.../node_modules/@google-cloud/storage/build/src/file.js:1295:35)
    at Util.handleResp (.../node_modules/@google-cloud/common/build/src/util.js:123:9)
    at retryRequest (.../node_modules/@google-cloud/common/build/src/util.js:404:22)
    at onResponse (.../node_modules/retry-request/index.js:200:7)
    at .../node_modules/teeny-request/build/src/index.js:208:17
    at <anonymous>
    at process._tickCallback (internal/process/next_tick.js:189:7)

Мой окончательный код выглядит примерно так:

let ws = gcs.getBucketWS(writePath, contentType)
const handleErr = (err) => {
  if (err) err.message = `Could not open gs://${ws._bucket}/${ws._path}: ${err.message}`
  done(err)
}
ws.on('error', handleErr)
// trying to do everything we can to handle these errors
// for some reason we still get UnhandledPromiseRejectionWarning
try {
  ws.write(item, null, err => {
    handleErr(err)
  })
  ws.end()
} catch (e) {
  handleErr(e)
}

Заключение

Для меня до сих пор загадка, как пользователь библиотеки @google-cloud/storage, или, если на то пошло, duplexify, должен правильно обрабатывать ошибки. Комментарии от сопровождающих библиотек любого проекта будут оценены. Спасибо!