node.js handling stream back pressure when using setTimeout

This is a follow on question for a further issue I've encountered from this earlier question:

nodejs: read from file and store to db, limit maximum concurrent db operations

Problem:

I want to condtionally reschedule some operations for a later time, however this is breaking my method for handling back-pressure.

Detail:

I have a CSV file that I am reading in as a stream, and using transforms to convert to JSON and then asynchronously store each line to a DB.

As lines are processed by the transform, they are placed onto an async queue which is responsible for issuing the DB operations.

E.g.

parser._transform = function(data, encoding, done) {

    var tick = this._parseRow(data);

    dbQueue.push(tick, function(err, result) {
      if (typeof(err) != 'undefined') { console.log(err) }
    });

    this.push(tick);
    done();
}

Back pressure is handled by pausing and resuming the parser when the queue is saturated/empty:

dbQueue.saturated = function() {
  parser.pause();
}

dbQueue.empty = function() {
  parser.resume();
}

The change I have been trying to make is that when an item is pulled off the queue, it is conditionally rescheduled for some time (100ms) in future:

var dbQueue = async.queue(function(data, callback) {
  if (condition) {
    // re-schedule operation by adding back to queue 100ms later
    setTimeout(function(data, callback) {
        dbQueue.push(data, function(err, result){
      });
    }, 100, data, callback);
  } else {
    //execute the db store
     ... ...
  }
}

I believe my problem is that now many operations will spend most their time in setTimeout, so the dbQueue will be empty, and the back-pressure on the transform stream is not being handled as desired.

I have tried a few attempts at using counters such as max_ops and running_ops to ensure the stream is paused/resumed, but unsuccessfully.

Is there a more idiomatic way of handling this in node.js?

Since this looks like it's an external condition and not something related to what dbQueue is doing, instead of re-inserting the data in to the queue when the condition occurs, I would pause simply pause dbQueue. For example, lets say your condition is that the database disconnected for some reason and there's an event you can listen to for that. In that case you can just do something similar to what you're doing when dbQueue is saturated/empty:

db.on('disconnect', function() {
    dbQueue.pause();
});

db.on('connect', function() {
    dbQueue.resume();
});

This is usually a better approach than waiting for some pre-determined timeout. That being said, sometimes waiting for a timeout is the only option. In that case you could do something similar but, instead of waiting for a separate event to trigger the resume(), simply use setTimeout():

db.on('disconnect', function() {
    dbQueue.pause();
    setTimeout(function() {
        dbQueue.resume();
    });
});

Note: If we are really talking about db disconnects here, then you might also want to pause/resume dbQueue if there's a db error in the case that 100ms isn't enough time for the db to re-connect

If you have a more specific condition you're looking for, and you're willing to share what that is, I may be able to give you a better example :)