Producer/consumer connect corrupts libuv async queue

[trevorr]

We recently switched from KafkaJS to this library, which has mostly been great, but we've been hitting an intermittent hang. I don't have a shareable reproducible test case yet, but I can explain where and why the issue arises. It affects any OS and Node version.

It was a regression introduced in 34e4bd3 by moving ActivateDispatchers from HandleOKCallback to Execute in ProducerConnect (here) and KafkaConsumerConnect (here). That method results in a call to uv_async_init, which is not thread safe and must be called from the event loop thread. HandleOKCallback and HandleErrorCallback are called from that thread, but Execute is called from a worker thread.

To demonstrate the issue, I added an assertion that dumps a stack trace whenever uv_async_init is called from a different thread. It ends up failing here:

0   confluent-kafka-javascript.node     0x000000011a0186f8 _ZN9NodeKafka18assert_main_threadEv + 228
1   confluent-kafka-javascript.node     0x000000011a0047a4 _ZN9NodeKafka9Callbacks10Dispatcher8ActivateEv + 64
2   confluent-kafka-javascript.node     0x000000011a029180 _ZN9NodeKafka4Conf6listenEv + 64
3   confluent-kafka-javascript.node     0x000000011a02e228 _ZN9NodeKafka13KafkaConsumer19ActivateDispatchersEv + 32
4   confluent-kafka-javascript.node     0x000000011a040010 _ZN9NodeKafka7Workers20KafkaConsumerConnect7ExecuteEv + 40
5   confluent-kafka-javascript.node     0x000000011a02c8f8 _ZN3Nan12AsyncExecuteEP9uv_work_s + 44
6   node                                0x00000001033616d4 worker + 248
7   libsystem_pthread.dylib             0x00000001867372e4 _pthread_start + 136
8   libsystem_pthread.dylib             0x00000001867320fc thread_start + 8
Assertion failed: (false), function assert_main_thread, file common.cc, line 46.

I believe those ActivateDispatchers calls should be moved back to HandleOKCallback. If that's too late for the scenario mentioned in the comment, they could possibly be moved to the worker constructor. I'm happy to submit a PR doing one of those things, but wanted to get the discussion started first.

Background

When I attached a debugger to our hung Node process, it was in an infinite loop here in libuv:

  while (!uv__queue_empty(&queue)) {
    q = uv__queue_head(&queue);
    h = uv__queue_data(q, uv_async_t, queue);

    uv__queue_remove(q);
    uv__queue_insert_tail(&loop->async_handles, q);

    /* Atomically fetch and clear pending flag */
    pending = (_Atomic int*) &h->pending;
    if (atomic_exchange(pending, 0) == 0)
      continue;

That loop is supposed to move all current async handles to a temporary queue, then put them back in the async handles queue one at a time, calling the async callback if it is defined and pending is set. However, it ends up with the same handle each time:

* thread #1, queue = 'com.apple.main-thread', stop reason = step over
    frame #0: 0x0000000105154cc8 node`uv__async_io(loop=0x00000001087a0c90, w=<unavailable>, events=<unavailable>) at async.c:165:5 [opt]
   162 	    q = uv__queue_head(&queue);
   163 	    h = uv__queue_data(q, uv_async_t, queue);
   164 	
-> 165 	    uv__queue_remove(q);
   166 	    uv__queue_insert_tail(&loop->async_handles, q);
   167 	
   168 	    /* Atomically fetch and clear pending flag */
(llnode) p queue
(uv__queue) {
  next = 0x0000000134807748
  prev = 0x000000012381f790
}
(llnode) n
Process 22379 stopped
* thread #1, queue = 'com.apple.main-thread', stop reason = step over
    frame #0: 0x0000000105154cd8 node`uv__async_io(loop=0x00000001087a0c90, w=<unavailable>, events=<unavailable>) at async.c:166:5 [opt]
   163 	    h = uv__queue_data(q, uv_async_t, queue);
   164 	
   165 	    uv__queue_remove(q);
-> 166 	    uv__queue_insert_tail(&loop->async_handles, q);
   167 	
   168 	    /* Atomically fetch and clear pending flag */
   169 	    pending = (_Atomic int*) &h->pending;
(llnode) p queue
(uv__queue) {
  next = 0x00000001087a0ee8
  prev = 0x000000012381f790
}
(llnode) n
Process 22379 stopped
* thread #1, queue = 'com.apple.main-thread', stop reason = step over
    frame #0: 0x0000000105154cec node`uv__async_io(loop=0x00000001087a0c90, w=<unavailable>, events=<unavailable>) at async.c:169:34 [opt]
   166 	    uv__queue_insert_tail(&loop->async_handles, q);
   167 	
   168 	    /* Atomically fetch and clear pending flag */
-> 169 	    pending = (_Atomic int*) &h->pending;
   170 	    if (atomic_exchange(pending, 0) == 0)
   171 	      continue;
   172 	
(llnode) p queue
(uv__queue) {
  next = 0x0000000134807748 // it's back in queue?????????????????
  prev = 0x000000012381f790
}

This should only be possible if both queues share nodes, which they're not supposed to. However, queue->next == loop->async_handles->prev:

(llnode) p queue // temp queue
(uv__queue) {
  next = 0x0000000134807748 // !!!
  prev = 0x000000012381f790
}
(llnode) p loop->async_handles // event loop async queue
(uv__queue) {
  next = 0x00000001087a0dc0
  prev = 0x0000000134807748 // !!!
}

I believe this is caused by calling uv_async_init from the wrong thread. Our app creates several Kafka producers and consumers at startup, which makes this more likely to hit. A standalone reproduction would likely involve connecting and disconnecting producers and/or consumers in a loop until the process hangs.

[trevorr]

@milindl I was going to add an e2e test to reproduce this, but I can't figure out how to run them.

$ make e2e
+ ./node_modules/.bin/mocha --exit --timeout 120000 --ui exports e2e/admin.spec.js e2e/both.spec.js e2e/consumer.spec.js e2e/groups.spec.js e2e/oauthbearer_cb.spec.js e2e/producer-transaction.spec.js e2e/producer.spec.js

 Exception during run: ReferenceError: describe is not defined
    at Object.<anonymous> (confluent-kafka-javascript/e2e/admin.spec.js:72:1)

Any idea what I'm missing? I also tried running the Jest tests with make promisified_test, but that ran for about 15 minutes before dumping a huge amount of console logs and dying with segfault 11 (OOM?). During that time, a number of tests passed and a few failed. I'm still trying to figure out the project structure.

[milindl]

Thank you for the really great report @trevorr . I was able to reproduce it (including the location of where the node process is stuck) with some test code rather consistently.

I believe those ActivateDispatchers calls should be moved back to HandleOKCallback. If that's too late for the scenario mentioned in the comment, they could possibly be moved to the worker constructor.

I tried this out: by removing the ActivateDispatchers from Execute, and putting it just before the creation of the relevant AsyncQueueWorker, this particular issue seems to be resolved. I haven't run all the tests yet, or checked if I broke some other functionality, but the hung node process problem seems to go away. I'm going to let it run for a couple hours.

Regarding the crash that you might be seeing, we merged a fix to it very recently to master (#151), so if you're behind that, it's likely to be the same issue. Otherwise, the promisified test suite is pretty heavy and there are a few flaky tests - I'm not sure what might've caused the crash.

For running the tests, I'm using

npx mocha  --timeout 120000 e2e/
npx jest --force-exit --verbose test/promisified/

The reason for both jest and mocha tests is that in an effort to make migration easier, we ported some jest-based tests from KafkaJS. Some come from node-rdkafka, which uses mocha. I'll update the Makefile for the tests.

Here's the code I used to reproduce it, ran it with npx jest --force-exit --verbose test/promisified/preactivate_dispatcher.spec.js

jest.setTimeout(350000); // we don't want to timeout.

const {
    sleep,
    createProducer,
} = require('./testhelpers');

describe('Hang', () => {
    it('Creates producers again and again', async () => {
        let promises = [];
        let iters = 0;
        while (true) {
            const createConnectAndDisconnectProducer = async () => {
                const client = createProducer({}, {'log_level': 0});
                await client.connect();
                await sleep(5);
                await client.disconnect();
            };
            promises.push(createConnectAndDisconnectProducer());
            if (promises.length === 100) {
                await Promise.all(promises);
                console.log(`Completed ${++iters} iterations`);
                promises = [];
            }
        }
    });
});

I can pick this up to verify the fix idea soonish, after addressing some other stuff, this is definitely something we want to fix.

[trevorr]

Awesome, thanks for verifying it so quickly! Do you have a guess on when the fix might make it into a release? This is causing us daily pain by making our CI runs flaky (and they must pass to merge), so we're trying to decide if we need to make a temporary fork or wait it out.

[trevorr]

For running the tests, I'm using

npx mocha  --timeout 120000 e2e/
npx jest --force-exit --verbose test/promisified/

Thanks, these commands work for me. However, the e2e tests always segfault in groups.spec.js (with or without the fix for this issue). It appears to be a null pointer issue in the consumer. I opened #164 to track it.

[trevorr]

@milindl Is there anything I can do to help with this, such as submitting a PR? This is causing us daily pain, so anything you can do to help us get a working build would be greatly appreciated. 🙏🏻

[milindl]

Closing this, should be addressed in 1.0.0