Обработка ошибок потоковой передачи/конвейера Node.js (изменение статуса ответа при ошибке)

У меня есть миллионы строк в моей базе данных Cassandra, которые я хочу передать клиенту в виде zip-файла (не хочу потенциально огромного zip-файла в памяти). Я использую функцию stream() из драйвера Cassandra-Node, направляя ее в Transformer, который извлекает одно поле из каждой строки, которая мне нужна, и добавляет новую строку, а также передает в архив, какие каналы направляются в объект экспресс-ответа. Кажется, это работает нормально, но я не могу понять, как правильно обрабатывать ошибки во время потоковой передачи. Я должен установить соответствующие заголовки/статус перед потоковой передачей для клиента, но если во время потоковой передачи возникает ошибка, например, в dbStream, я хочу очистить все каналы и сбросить статус ответа, чтобы он был примерно таким, как 404. Но если я попытаюсь сбросить статус после установки заголовков и запуска стриминга, я получаю Can't set headers after they are sent. Я просмотрел все и не могу найти, как правильно обрабатывать ошибки в Node при передаче/потоке в объект Response. Как клиент может определить, действительно ли были переданы действительные данные, если я не могу отправить правильный код ответа при ошибке? Кто-нибудь может помочь?

function streamNamesToWriteStream(query, res, options) {
  return new Promise((resolve, reject) => {

    let success = true;

    const dbStream = db.client.stream(query);
    const rowTransformer = new Transform({
      objectMode: true,
      transform(row, encoding, callback) {
        try {
          const vote = row.name + '\n';
          callback(null, vote);
        } catch (err) {
          callback(null, err.message  + '\n');
        }
      }
    });

    // Handle res events
    res.on('error', (err) => {
      logger.error(`res ${res} error`);
      return reject(err);
    });

    dbStream.on('error', function(err) {
      res.status(404).send() // Can't set headers after they are sent.
      logger.debug(`dbStream error: ${err}`);
      success = false;
      //res.end();
      //return reject(err);
    });

    res.writeHead(200, {
      'Content-Type': 'application/zip',
      'Content-disposition': 'attachment; filename=myFile.zip'
    });

    const archive = archiver.create('zip');
    archive.on('error', function(err) { throw err; });
    archive.on('end', function(err) {
      logger.debug(`Archive done`);
      //res.status(404).end()
    });

    archive.pipe(res, {
      //end:false
    });
    archive.append(dbStream.pipe(rowTransformer), { name: 'file1.txt' });
    archive.append(dbStream.pipe(rowTransformer), { name: 'file1.txt' });
    archive.finalize();
  });
}