Add rethinkdb-changefeed-reconnect

[zllovesuki]

I have an application running v2.1.0 and updated the dependency to v3.1.3, and the queue stops working after some period of time.

It works when I initially start up the queue, the handler can process incoming jobs. However, after some time the handler was no longer called (.process(function(job, done) {}))

Could you walk me through the changes? I've eliminated other dependency in the application.

[grantcarthew]

Hi @zllovesuki
As per the Change Log the only breaking change from v2 to v3 was the event signatures. Specifically the error event signatures.

Read more about the error event signatures here: https://github.com/grantcarthew/node-rethinkdb-job-queue/wiki/Event.error

So check your event callbacks.

[zllovesuki]

@grantcarthew I think it's the same error that if a changefeed becomes unavailable, the Queue just silently ignores it and no errors were thrown.

[grantcarthew]

Oh. Not seen that before. There is a package called rethinkdb-changefeed-reconnect that might do the job here. Was wondering if it was worth adding it to the queue dependencies.

[zllovesuki]

It's similar to #72 .

It would be great if you could implement it. :)

[zllovesuki]

Actually, now that I think about it, it may not seem so easy.

My application connects to a proxy node (same application on the same node connects to the proxy on the same node), and each proxy node then connects to the actual data nodes.

If the proxy node drops (which is connection closed), then the liveness probe will pick it up and nothing terrible will happen.

However, if the data node drops, the error is silently ignored. If I use changefeeds by itself (.changes()), RethinkDB actually complains something like "changefeed unavailable".

[zllovesuki]

Right now my workaround is to query server_status and check for changes in time_connected (so the application can tell that a data node was dropped and it should restart).

[disarticulate]

@grantcarthew I used the rethinkdb-changefeed-reconnect almost immediately since the connections can be flaky.

I'd consider it a necessary dependency.

[grantcarthew]

OK, I just had a closer look at the project and will add this to the To Do list. My head isn't in this space at the moment so it will not be done asap. Sorry. If anyone wants to contribute?

I imagine exposing the reconnect package options onto the existing rethinkdb-job-queue connection options would work well.

[disarticulate]

This is the code I had when I inquired about broadcasting queues:

var processChangefeed = require('rethinkdb-changefeed-reconnect')
...
const queueChangeFeedWatcher = (name, q, queueHandler, vm) => {
  // wait for the q ready
  q.on('ready', (queueId) => {
    var r = q.r
    var newTasksFeed = () => { return r.table(name).changes() } // get the changes
    var handleNewTask = ({new_val, old_val}) => { //handle updates to changes
      if ('data' in new_val
        && new_val.status === 'completed'
        && new_val.data.broadcast
        && new_val.data.broadcasters.length > 0) { // watch for a broadcast flag
        var [host,proc,cmd,count,uuid] = new_val.queueId.split(':')
        if (new_val.data.broadcasters.indexOf(hostID()) === -1
            && queueHandler[cmd]) { // check whether processor has already managed
          vm.log('[queueChangeFeedWatcher] > ', new_val.data.broadcasters, host, proc, cmd)
          queueHandler[cmd] (// run handler from handler objects
            new_val,
            (err, result) => {},
            (job, cb) => {}
          )
        }
      }
    }

    var handleError = (err) => {
      vm.log(err.stack)
    }
    processChangefeed(newTasksFeed, handleNewTask, handleError, { 
      changefeedName: name,
      attemptDelay: 30000,
      maxAttempts: 30,
      silent: false,
      logger: console
    })
  })
}

I roughed out what it looks like would take to just swap the reconnect driver.

master...disarticulate:patch-1

[grantcarthew]

Ah, that's where I saw it! From your issue @disarticulate. I couldn't remember.

That master compare is a good start thanks, however as I said before the reconnect options need to be exposed as part of the public API. Again I imagine either properties of the cxn object or a nested object on the cxn object. There should also be an options to disable the reconnect feature by ether a specific switch or lack of options.

For reference I did some research into where we are at with respect to reliable change feeds. Here is the main discussion as far as I can tell: rethinkdb/rethinkdb#3471

It doesn't look like it will be anytime soon so the rethinkdb-changefeed-reconnect module looks like the way forward for now.

[grantcarthew]

I just published a minor change updating dependencies. I'm not adding features at this time. If someone is interested in working on this feature, please do so. I see this feature as the most useful addition to the queue.