opened
adrian-gierakowski
username_0
The following test deadlocks (`toCallback` never gets called):

```js
/* eslint-disable no-unused-vars, no-shadow, no-redeclare */
var _, EventEmitter = require('events').EventEmitter,
    through = require('through'),
    // sinon = require('sinon'),
    Stream = require('stream'),
    streamify = require('stream-array'),
    concat = require('concat-stream'),
    // Promise = RSVP.Promise,
    transducers = require('transducers-js'),
    bluebird = require('bluebird'),
    runTask = require('orchestrator/lib/runTask'),
    fl = require('fantasy-land'),
    bufferFrom = require('buffer-from');

if (global.highland != null) {
    _ = global.highland;
}
else {
    _ = require('../lib/index');
}

// Use bluebird cancellation. We want to test against it.
bluebird.config({
    cancellation: true,
    longStackTraces : true
});


var MongoClient = require('mongodb').MongoClient

exports['batch - mongodb'] = function (test) {
    test.expect(1);

    MongoClient.connect('mongodb://localhost:27017')
    .then(async (client) => {

        var db = client.db("test")
        var col = db.collection("test")
        const docs = Array.from(Array(22).keys()).map( _id => { return { _id } })
        await col.removeMany({})
        await col.insertMany(docs)

        cursor = col.find()
        cursor.batchSize(2)
        // Changing batch size to 3 causes the test to pass.
        // cursor.batchSize(3)

        _(cursor.stream())
        .batch(11)
        // Changing batch size to 12 causes the test to pass.
        // .batch(12)
        .flatMap( x => {
            console.log("x:", x)
            return _(x)
        })
        // If flatMap above is replaced with .map, the test passes.
        // .map( x => {
        //     console.log("x:", x)
        //     return x
        // })
        .reduce(
            (acc, x) => {
                console.log("conc", acc, x)
                return acc.concat(x)
            },
            []
        )
        .toCallback((e, xs) => {
            console.log("xs",xs)
            test.same(xs, docs);
            test.done();
        })
    })

};
```

You save save the above in `tests/batch_deadlock.js`, install mongo drives with `npm i mongodb` and finally run it with `npx nodeunit tests/batch_deadlock.js`. It should deadlock when running with node version >= 10, but passes when ran with node 8.x.

I've also left a few comments in the code which show some tweaks that make the deadlock go away.
deadlock when using `batch` with `flatMap` and mongo stream as source (node >= 10)
issue
created
szabolcs-szilagyi
username_1
Created a small github repo with Travis on it with a similar test case:
https://github.com/username_1/highland-node-12
https://travis-ci.org/username_1/highland-node-12
[![Build Status](https://travis-ci.org/username_1/highland-node-12.svg?branch=master)](https://travis-ci.org/username_1/highland-node-12)

Setup Travis to build it daily, so the above image should go green once its fixed.
comment
created
szabolcs-szilagyi
username_1
interestingly if you use the `ratelimit` function before calling the batch on the stream with some settings it can finish. Think it might be a timing issue on setting some internal state. E.g. if in the test project I put:
```js
        hl(mongoStream)
          .ratelimit(110, 0)
          .batch(300)
```
it will actually finish, while if I put 120 for the amount it will hang.
comment
created
szabolcs-szilagyi
username_1
what I've found sofar:
it is able to pull data from the mongo stream 4 times, but then it forgets to resume/pull from it again. Tried to run it with the `--inspect` flag to see - after lots of console.logs - but of course like all timing bugs it runs perfectly when you start it in debug mode. :/
Tried to reproduce the bug with file streams, but those ran okay. Another case that _mitght_ be related is when I stream http requests using the `got` library, it also hangs. (node@12, got@10).

I'm tempted to give up at this point and just start on a big project of replacing all highland usages with some other solutions. :(
comment
created
eccentric-j
username_2
I took a quick look at your test repo and saw you are using highland 3 which if I recall, is missing some work around when resources are destroyed which may be affecting this. What happens if you use the stable Highland 2.13.5? https://www.npmjs.com/package/highland

Also what happens if you use `.toPromise(Promise)` instead of `.toCallback()`? 

Lastly, looking at the source:

https://github.com/caolan/highland/blob/3.0.0/lib/index.js#L2785-L2792

It has a clause to close the stream when the source stream closes. That suggests it may be the mongo node stream that is not triggering the completion because it's also batching and will not close until it's satisfied.

Could you try adding an event listener to the mongo stream to see if it dispatches a `finish` or `close` event?
comment
created
szabolcs-szilagyi
username_1
Hmm with version 2.13.5 it finishes nicely with out the issue. That is cool, now just need to see why did we started to use 3. :D

In regards to the `finish` and `close` events, I've added them, but they don't get triggered. (updated the repo with the event listeners too.
comment
created
eccentric-j
username_2
Ok that's at least progress. I see your updated test repo and the use of toPromise. You're right that `close` and `finish` may not fire but is `end` firing?

Also, based on how you're using batch and flatMap here, would `.take(11)` not suffice?
comment
created
szabolcs-szilagyi
username_1
Indeed progress, thank you for the support :)

For when the test runs on the non-batched stream it triggers [all three event handlers](https://travis-ci.org/username_1/highland-node-12/jobs/657010982#L1708-L1710), but on the second run with batched stream, [none of them gets triggered](https://travis-ci.org/username_1/highland-node-12/jobs/657010982#L2914-L2915).

How would the `.take(n)` method help? It gives a new stream with the first _n_ items, but don't see how would we be able to iterate through all the items in the original stream.
comment
created
eccentric-j
username_2
Thanks for that info! I'm going to dig in today and see if I can pinpoint if it's a highland issue specifically or something with the batched mongo stream.

As for `take`, I wasn't sure what the intention was looking at the example. Based on the use of `.toCallback` and the fact that this example batches it into an array then flattens it out again, I was wondering if the goal was to get a finite stream out of an infinite stream.
comment
created
eccentric-j
username_2
Made some progress.

```js
async function main () {
  const client = await MongoClient.connect(url, opts);
  const db = client.db("test");
  const col = db.collection("test");
  await col.removeMany();
  await col.insertMany(docs);

  // This function will explode in six seconds
  timeout(6000)
    .done(() => {
      throw new Error("APPLICATION TIMED OUT");
      client.close();
      process.exit(1);
    });

  return _.of(col)
    .flatMap(fetchDocs)
    .batch(11)
    .consume(logStream)
    .flatMap(_)
    .reduce(append, [])
    .toCallback((err, xs) => {
      client.close();

      if (err) {
        console.error(err);
        process.exit(1);
      }

      assert.deepEqual(xs, docs, `DOCS WERE NOT RETUREND. INSTEAD FOUND "${xs}"`);

      console.log("Completed!", xs);
      process.exit(0);
    });
}
```

That works, the main different being `_(cursor.stream())` -> `_.of(x).flatMap(() => _(cursor.stream))`.

My next step is to create a new Node Readable stream that kinda simulates the cursor.stream. That should confirm whether this is a Highland or Mongo DB API issue.

My progress is at https://github.com/username_2/highland-batch-repro which includes a docker-compose file to make it easier to run.
comment
created
szabolcs-szilagyi
username_1
wow interesting, tried to use
```js
  const mongoStream = hl.of(collection.find().stream()).flatMap(hl);
```
instead of the 
```js
  const mongoStream = collection.find().stream();
```
in my tests and indeed the streaming works if wrapped like this.
comment
created
eccentric-j
username_2
Tried `_(mockMongoStream(docs))` with a custom subclass of `Stream.Readable` and it worked without issue https://travis-ci.com/username_2/highland-batch-repro/builds/151668807. I also tried `_(_(docs).toNodeReadable({ objectMode: true }))` and it also worked without issue https://travis-ci.com/username_2/highland-batch-repro/builds/151663566.

This has me leaning towards the issue residing in mongodb's `cursor.stream()` method. The next step is to use the mongo db stream and try consuming it in a batch Node Transform stream to see if I can reproduce the error without using Highland.
comment
created
adrian-gierakowski
username_0
has anyone managed to make any further progress on this?
comment
created
szabolcs-szilagyi
username_1
@username_0 unfortunately no, personally I started to move away from using highland utilizing `async-await`, `for await of` and nodejs' built in [`pipeline`](stream_stream_pipeline_source_transforms_destination_callback) method.
There is a helper lib that a collegue of mine made - called [`aih`](https://github.com/jeanbaptiste-brasselet/aih) - have a look into the `lib` folder you can see some good examples on how you can use `await` with streams.
comment