[BIGTABLE] Insert callback called multiple times if err

[harscoet]

Environment details

• OS: macOS Sierra 10.12.11
• Node.js version: 6.9.1
• npm version: 3.10.8
• google-cloud-node version: "@google-cloud/bigtable": "^0.6.0"

Steps to reproduce

I use "insert" function to insert an array of 2 entries, for the example I force error with "wrong" timestamp (yes timestamp defined in milliseconds is invalid...)

let i = 0;
const INSTANCE_NAME = 'myInstance';

require('@google-cloud/bigtable')().instance(INSTANCE_NAME).table('test').insert([{
  key: 'alincoln',
  data: {
    follows: {
      gwashington: {
        value: 1,
        timestamp: 1480517318949 // Invalid timestamp (note 1480517318 in seconds is OK, 1480517318949000 in microseconds is OK)
      }
    }
  }
},{
  key: 'alincoln',
  data: {
    follows: {
      helloworld: {
        value: 1,
        timestamp: 1480517318949 // Invalid timestamp (note 1480517318 in seconds is OK, 1480517318949000 in microseconds is OK)
      }
    }
  }
}], err => {
  i++;
  console.log('--');
  console.log(`CALLBACK ${i}`);
  if (err) console.error(err.message);
  console.log('--');
});

/*
--
CALLBACK 1
--
--
CALLBACK 2
Error in field 'entries' : Error in element #0 : Error in field 'Mutation list' : Error in element #0 : Timestamp granularity mismatch. Expected a multiple of 1000 (millisecond granularity), but got 1480517318949.
--
*/

In source file "table.js" I just add some logs to find source of error, callback is called multiple times, first time with end event, and last with error event
If I use promise syntax I just never see that an error occurred because promise is resolved with end event without error

// table.js - line 868
this.requestStream(grpcOpts, reqOpts)
  .on('error', (err) => {
    console.log('event:error', err); // LAST_LOG: event:error { Error: Error in field 'entries' : Error in element #0 : Error in field 'Mutation list' : Error in element #0 : Timestamp granularity mismatch. Expected a multiple of 1000 (millisecond granularity), but got 1480517318949.
    callback(err);
  })
  .on('data', function (obj) {
    obj.entries.forEach(function (entry) {
      // Mutation was successful.
      if (entry.status.code === 0) {
        return;
      }

      var status = common.GrpcService.decorateStatus_(entry.status);
      status.entry = entries[entry.index];

      mutationErrors.push(status);
    });
  })
  .on('end', function () {
    var err = null;

    if (mutationErrors.length > 0) {
      err = new common.util.PartialFailureError({
        errors: mutationErrors
      });
    }

    console.log('event:end', err); // FIRST_LOG: event:end null
    callback(err);
  });

[stephenplusplus]

Thanks for reporting, we'll put out a patch soon.

[stephenplusplus]

Fix sent in #1847.

[stephenplusplus]

The fix turned out to be incorrect and the issue persists.

[stephenplusplus]

It looks like the gRPC stream emits both metadata and error, which triggers two event handlers that retry-request is listening for to decide if it should retry the request: https://github.com/stephenplusplus/retry-request/blob/85ee18dfc48e3bff174a4711440f55ed4748dc0f/index.js#L83-L84

metadata is what we've been using to say "nothing obvious went wrong with the request", but it looks like we're going to need to beef up that logic, since an error can still emerge.

A solution is to wait after receiving the metadata event before giving retry-request the all clear, but it would introduce an undesirable delay at the start of each request. That would look like:

  var waitForErrorTimeout;

  var retryOpts = {
    retries: this.maxRetries,
    objectMode: objectMode,
    shouldRetryFn: GrpcService.shouldRetryRequest_,

    request: function() {
      return service[protoOpts.method](reqOpts, self.grpcMetadata, grpcOpts)
        .on('metadata', function() {
          // retry-request requires a server response before it starts emitting
          // data. The closest mechanism grpc provides is a metadata event, but
          // this does not provide any kind of response status. So we're faking
          // it here with code `0` which translates to HTTP 200.
          //
          // https://github.com/GoogleCloudPlatform/google-cloud-node/pull/1444#discussion_r71812636
          var self = this;
          waitForErrorTimeout = setTimeout(function() {
            var grcpStatus = GrpcService.decorateStatus_({ code: 0 });
            self.emit('response', grcpStatus);
          }, 1000);
        });
    }
  };

  return retryRequest(null, retryOpts)
    .on('error', function(err) {
      clearTimeout(waitForErrorTimeout);

      var grpcError = GrpcService.decorateError_(err);
      stream.destroy(grpcError || err);
    })
    .pipe(stream);

@callmehiphop any other ideas on how we can handle this?

[callmehiphop]

@stephenplusplus I'm not too sure, maybe some one on the gRPC side of things would have insight to help us.

/cc @murgatroid99

[murgatroid99]

I think it's worth clarifying what those emitted events mean. The "metadata" event is emitted unconditionally at the beginning of any call that the server accepts and starts handling, whether or not the call eventually succeeds. "error" is emitted if the call fails in any way.

There is also the "status" event, which is emitted unconditionally when the call finishes, with a code indicating success or failure. You could do something like

call.on('status', function(status) {
  if (status.code == grpc.status.OK) {
    self.emit(response);
  }
});

Receiving an OK status code should be mutually exclusive with seeing an "error" event.

[stephenplusplus]

@murgatroid99 thank you for the explanation. After digging in further, it looks like the gRPC stream will emit both an error and an end event. While the stream is technically "ended" after an error occurs-- maybe more accurately, "closed"-- the end event should not be emitted.

require('fs')
  .createReadStream('non-existent-file')
  .on('error', function(err) { console.log('emitted') })
  .on('end', function() { console.log('not emitted') })
  .on('data', function() {}) // to drain the data

require('request')
  .get('http://www.non-exitent-url.com')
  .on('error', function(err) { console.log('emitted') })
  .on('end', function() { console.log('not emitted') })
  .on('data', function() {}) // to drain the data

This is the root of our problem when trying to use retry-request. It's expecting the stream to either emit error or end, but not both. The stream will emit end first, then error.

I haven't looked deeply into the implementation in gRPC, but making a generalization; if this is the process:

• allow the stream the user is holding to complete its lifecycle
• do some post-processing
• emit status
• if there was an error, emit error

Consider using an intermediary stream, such as a through stream or a Transform stream, then registering a prefinish event handler to cork() the stream while you do the post processing. If after determining there was no error, you can call uncork() on the through stream, which will allow the end event to fire.

[stephenplusplus]

Posted to a new issue in the gRPC repo: grpc/grpc#8954