Cloud Storage not persisting on error

[techniq]

I've ran into an issue that is only present in production using the Google Cloud Storage API. I have an importer that loads 100 entries at a time. When developing locally (with require('fs')) and an error occurs on say page 5, the previous pages worth of entries (400 total) are still persisted to disk. When using GCS, this is not the case (the file doesn't even appear in the bucket).

Here is a snippet of the problematic part of the pipeline.

    stream = importer.streamPages({perPage: perPage, importDate: importDate}, maxPages);

    let writeStream = null;
    if (process.env.NODE_ENV === 'production') {
      writeStream = bucket.file(importFileName).createWriteStream();
    } else {
      writeStream = fs.createWriteStream(`data/${importFileName}`);
    } 

    stream
      .on('error', err => this.logError(err))
      .pipe(ndjson.stringify())
      .pipe(writeStream);

[stephenplusplus]

The API can receive a file in two ways; a single request that contains the entire contents of the file (simple), or a series of requests (resumable). By default, createWriteStream uses resumable uploads, so that if an error occurs during the upload, you can start the transfer over, and only send data starting from where the last upload quit.

With a simple upload (you can pass resumable: false to createWriteStream), if an error occurs, the API never sees a completed HTTP request, so no file is ever created (@jgeewax, is this right?).

[techniq]

@stephenplusplus hmm, I'm using the default resumable stream. If an error occurs, is there a way I can resume and just "commit" the current progress.

In my example, importer.streamPages() walks a paginated api, emitting a perPage amount of objects (it's an object stream) on each data event. It would be nice if I could do something like this.

stream
      .on('error', err => { this.logError(err); this.done(); )

[stephenplusplus]

That sounds like a cool idea. I'll see if I can find a way to do that 👍

[stephenplusplus]

I think what you can do is:

stream
      .on('error', err => { this.logError(err); this.end(); })

When a readable stream gracefully ends, it's saying "And you, next stream in the pipe, have received all I have to give you." Our write stream will continue normally, and write that data to the remote object. Try it out and let me know!

[techniq]

Thanks, I'll give that a shot and let you know. In my example, I'll need to not sure an arrow function so this is the stream and not the closure, so I figure it will be more like...

let logError = this.logError;
stream
      .on('error', function(err) { logError(err); this.end(); })

I currently don't have a test for this, but will try to make one (against Cloud Storage) and see how it goes, as otherwise I'll have to wait for the failure to occur again (import.streamPages retries a call 5 times before failing, which a gateway time-out caused the issue before)

[stephenplusplus]

Cool, thanks for sharing details about your use case. It always helps to hear real-world scenarios.

RE: the this stuff, good catch, I didn't think of that. Here's another option:

stream
      .on('error', err => { this.logError(err); stream.end(); })

And just for reference, here is the script I used to test:

'use strict';

var through = require('through2')
var gcloud = require('gcloud')({ projectId: '...' })
var gcs = gcloud.storage()
var bucket = gcs.bucket('...')
var file = bucket.file('new-file-' + Date.now())

var srcStream = through.obj(function (chunk, enc, next) {
  next(null, JSON.stringify(chunk))
})
var writeStream = file.createWriteStream()

srcStream.on('error', function () {
  this.end()
})

srcStream.pipe(writeStream)

srcStream.write({ a: 'b' })
srcStream.write({ a: 'b' })
srcStream.write({ a: 'b' })
srcStream.destroy(new Error('error'))

[techniq]

@stephenplusplus thanks for your help. I've made some progress on this, but my limited knowledge of Node streams has me hitting a wall.

I found some of my issue to be in my retryStream function, that uses a Passthrough stream/transform that I was previously just logging "Max attempts reached" when the streamFunc reached it's maximum number of retries.

Now, calling .end() on the PassThrough stream fixes the problem I was having with the file not showing up in GCS (after adding an arbitrary 1000ms setTimeout() to my test before calling done() so the node/mocha process didn't terminate before gcloude-node had time to write the file up to GCS. I tried using setImmediate() but it didn't work).

Anyways, the problem I have is I really need to bubble an error out of retryStream through streamPage and streamPages so I can add an 'error' event listener at the end of the pipeline so I can properly log/email the problem before writing what I have grabbed so far to GCS (and a importStream which also processes the records and persists them to Cloud SQL) .

If I remove the error handler from BaseImporter#streamPage, then I get Uncaught Error: Uncaught, unspecified "error" event. (Max attempts reached), but if I leave it on, the error gets swallowed and the test times out because end or error never makes it.

If I attempt to "re-emit" the error (.on('error', function(err) { console.error('streamPage error', err); this.emit('error', err) })) I get a Uncaught RangeError: Maximum call stack size exceeded

This is quite likely out of the scope of this issue, but thought I'd at least present my problem in case you had any suggestions.

Here are snippets of the related functions. Thanks for any help you have on this.

retry-stream.js

'use strict';
const PassThrough = require('stream').PassThrough;

module.exports = function retryStream(streamFunc, retries) {
  var resultStream = new PassThrough();
  let attempt = 1;
  retries = retries || 3;

  function attemptStream() {
    let stream = streamFunc()
      .on('error', (err) => {
        console.warn(`Attempt (${attempt} of ${retries}) failed`, err);
        if (attempt < retries) {
          console.log('Retrying');
          attempt++;
          attemptStream();
        } else {
          console.error('Max attempts reached');
          // resultStream.end(); // <-- Works
          resultStream.emit('error', 'Max attempts reached'); // <-- Would like to work
        }
      })
      .on('response', () => {
        // console.log('success');
        stream.pipe(resultStream);
      })
  };
  attemptStream();

  return resultStream;
}

base.js

const got = require('got');
const MultiStream = require('multistream');
const through2 = require('through2');
const JSONStream = require('JSONStream');
const JSONPath = require('jsonpath');
...

class BaseImporter {
...
  streamPage(query, options) {
    const pageOptions = this._getPageOptions(query, false);
    const queryString = Object.keys(pageOptions.query).map(key => `${key}=${pageOptions.query[key]}`).join('&');
    // console.log('Fetching url', `${this.config.baseUrl}?${queryString}`);

    if (options && options.retries) {
      return retryStream(() => got.stream(this.config.baseUrl, pageOptions), options.retries)
        .on('error', err => console.error('streamPage error', err))
        .pipe(JSONStream.parse(this.config.jsonPath))
    } else {
      return got.stream(this.config.baseUrl, pageOptions)
        .on('error', err => console.error(err))
        .pipe(JSONStream.parse(this.config.jsonPath))
    }
  }

  streamPages(query, maxPages) {
    query.page = query.page || 1;
    let totalRaceCount = 0;
    let isDone = false;

    let factory = (streamCallback) => {
      if (isDone) {
        // console.log('Done');
        streamCallback(null);
        return;
      } else if (maxPages && query.page > maxPages) {
        console.log('Maximum pages reached.');
        streamCallback(null);
        return;
      } else {
        // console.log('Next page')
      }

      let raceCount = 0;
      let stream = this.streamPage(query, { retries: 5 })
        //.on('error', (err) => console.error('streamPages error', err))
        .on('end', () => {
          // console.error('streamPage end')
          console.log(`Fetched ${raceCount} races`);
          if (raceCount < query.perPage) {
            // console.log('Last page');
            isDone = true;
          }

          streamCallback(null, stream);
        })
        .pipe(through2({ objectMode: true, highWaterMark: query.perPage },
          function(chunk, enc, transformCallback) {
            // console.log('transform');
            // console.log('chunk', chunk);

            raceCount += 1;
            totalRaceCount += 1;

            this.push(chunk)
            transformCallback()
          })
        )

      query.page += 1;
    }

    return MultiStream(factory, { objectMode: true, highWaterMark: query.perPage })
      .on('error', (err) => console.error('MultiStream error', err))
      .on('close', () => console.log(`Total races: ${totalRaceCount}`))
  }
...
}

test.js

describe.only('BaseImporter GCS', function() {

  beforeEach(function() {
    nock('http://localhost')
      .get('/')
        .query({page: 1, perPage: 5})
        .reply(200, (uri, requestBody) => {
          // console.log('returning page1, 200')
          return [{}, {}, {}, {}, {}]
        })
      .get('/')
        .query({page: 2, perPage: 5})
        .reply(200, (uri, requestBody) => {
          // console.log('returning page2, 200')
          return [{}, {}, {}, {}, {}]
        })
      .get('/')
        .query({page: 3, perPage: 5})
        .times(10)
        .reply(500, () => {
          // console.log('returning 500')
        })
  })

  afterEach(function() {
    nock.cleanAll();
  })

  describe('#streamPages()', function(done) {
    it('should still write contents to GCS for previous pages after exhausting retries for later page', function(done) {
      this.timeout(5000);

      const importer = new BaseImporter({ baseUrl: 'http://localhost', jsonPath: '*' });

      const gcloud = require('gcloud');
      const ndjson = require('ndjson');
      const gcsConfig = require('../../../app/config').get('storage:googleCloudStorage');
      const gcs = gcloud.storage({ projectId: gcsConfig.projectId });
      const bucket = gcs.bucket(gcsConfig.bucket);
      const importFileName = `test/${new Date().toISOString()}`;
      const writeStream = bucket.file(importFileName).createWriteStream();
      // const writeStream = require('fs').createWriteStream('test.txt');

      const stream = importer.streamPages({perPage: 5}, 3)
        .on(
        .on('end', () => {
          // Give GCS a second to upload file before ending test (and killing process)
          setTimeout(done, 1000);
        })
        .pipe(ndjson.stringify())
        .pipe(writeStream)

        ;
    });
  });

});

[stephenplusplus]

No problem, I'm happy to take a look! I'll let you know if I can come up with anything

[techniq]

@stephenplusplus you are crazy responsive :)

[stephenplusplus]

Also just repo-dropping this because I also go through some headaches when trying to deal with streams, so I wrote up some guides for under/non-documented use cases: https://github.com/stream-faqs - I haven't dug into your code yet, but it might have a helpful section or two!

[techniq]

You mean https://github.com/stephenplusplus/stream-faqs and I've been there - https://github.com/techniq/wiki/wiki/Node.js :)

[stephenplusplus]

Haha, I just make sure to always keep a GH tab open:

And oh, nice! :)

[techniq]

As well a nice game. You have a daughter I image (or really like Frozen :) )

[techniq]

btw, I'm on Hangouts/Slack/etc if you want to chat outside of the issue to keep the noise down. I'm leaving the office in about 10 min to work my second shift (Chief Family Officer) but will be back on ~9:00pm EST.