stream() bug from 0.2.25 to 0.2.26 with postgresql

[IricBing]

Issue Description

I use selectQueryBuilder.stream() method to get a readable stream from postgresql 13. It worked very well before! Today, I create a new project and use the latest package, It block all database operation when invoke selectQueryBuilder.stream() more than 10 times! Then I use sql select query from pg_stat_activity; to get database connection, I found sql with stream not release. I'm sure it work well before, so I try to upgrade form 0.2.23 to latest version, then I found it not work from 0.2.25 to 0.2.26.

Expected Behavior

selectQueryBuilder.stream() method should release connection after stream close

Actual Behavior

selectQueryBuilder.stream() method doesn't release connection

Steps to Reproduce

1. use version 0.2.26 and invoke selectQueryBuilder.stream() method more than 10 times, or use sql select query from pg_stat_activity; to get pg connections
2. use version 0.2.25 and try again.

My Environment

Dependency	Version
Operating System	Ubuntu 20.04.1
Node.js version	v14.15.0
Typescript version	v3.7.4
TypeORM version	v0.2.26

Additional Context

Relevant Database Driver(s)

• aurora-data-api
• aurora-data-api-pg
• better-sqlite3
• cockroachdb
• cordova
• expo
• mongodb
• mysql
• nativescript
• oracle
• postgres
• react-native
• sap
• sqlite
• sqlite-abstract
• sqljs
• sqlserver

Are you willing to resolve this issue by submitting a Pull Request?

• Yes, I have the time, and I know how to start.
• Yes, I have the time, but I don't know how to start. I would need guidance.
• No, I don't have the time, although I believe I could do it if I had the time...
• No, I don't have the time and I wouldn't even know how to start.

[alvisetrevisan]

We have the same problem on postgres 12. Worked fine up until 0.2.25

[stvvt]

Here is a workaround I found, tested with typeorm@0.2.28 with Postgres 9.6

Before calling selectQueryBuilder.stream(), I created a queryRunner myself like so:

const queryRunner = selectQueryBuilder.connection.createQueryRunner("slave");
selectQueryBuilder.setQueryRunner(queryRunner);

const stream  = await selectQueryBuilder.stream();
stream.on('error' => queryRunner.release());
stream.on('end' => queryRunner.release());
stream.on('data' => (rawRow) => { /* ... */ } );
...

I think this worked (at least for me) because when SelectQueryBuilder doesn't have an already created queryRunner, it creates one itself (using the same connection.createQueryRunner("slave") factory method invocation), but in addition to that it automatically releases it. However here

typeorm/src/query-builder/SelectQueryBuilder.ts

Lines 1264 to 1267 in c4a36da

	} finally {
	if (queryRunner !== this.queryRunner) // means we created our own query runner
	await queryRunner.release();
	}

it calls queryRunner.release() too early - the queryRunner might not be connected yet, its connection flow finishes after the release() call and so it remains connected forever.

With pre-created queryRunner, SelectQueryBuilder never tries to release it (but you have to release it youself!), so the workaround works.

[IricBing]

@stvvt I didn't use queryRunner, This is my code

@Injectable()
export class PositionService {
  constructor(@InjectRepository(PositionEntity) private readonly positionRepository: Repository<PositionEntity>) {}

  async findStreamWithEmployee(): Promise<ReadStream> {
    return this.positionRepository
      .createQueryBuilder('positions')
      .innerJoinAndSelect('positions.employee', 'employees')
      .innerJoinAndSelect('employees.card', 'cards')
      .where('positions.deleted = false')
      .stream();
  }
}

And deal with stream elsewhere.

[stvvt]

@IricBing This should work for you:

  async findStreamWithEmployee(): Promise<ReadStream> {
    const queryBuilder = this.positionRepository
      .createQueryBuilder('positions')
      .innerJoinAndSelect('positions.employee', 'employees')
      .innerJoinAndSelect('employees.card', 'cards')
      .where('positions.deleted = false');
    const queryRunner = queryBuilder.connection.createQueryRunner("slave");
    queryBuilder.setQueryRunner(queryRunner);
    const stream = queryBuilder.stream();
    stream.on('error' => queryRunner.release());
    stream.on('end' => queryRunner.release());

    return stream;
  }

[IricBing]

@stvvt Thanks，maybe I should upgrade my code format, although my code format is more brief.
Thanks very much and mud in your eye.

[alvisetrevisan]

@stvvt Thanks! Your workaround is a bit problematic to implement in our current code. I don't mind making a PR to fix it properly, but I'm not familiar with the TypeORM codebase. Do you have any clue of what went wrong between 0.2.25 and 0.2.26?
I looked here at the obvious places (like QueryRunner and PostgresQueryRunner), but nothing suspicious there Diff between 0.2.25 and 0.2.26

[stvvt]

@alvisetrevisan I guess this change exposed the bug: ae3cf0e#diff-b924b9a6d4ab4328f4ad59d4d96b794f6a1975fae8d1d3708909e31c151730b9R106-R108

The change itself looks good. However, before it, the second (which is if fact the proper) call to queryRunner.release() succeeded. After the change, the second call does nothing, because the release() method has already been called in the finally block I mentioned in my earlier comment. (which call I think is the real bug here)

[alvisetrevisan]

@stvvt thanks for the pointer. I created a PR with what I think is a fix. I did not want to mess with the other drivers, so I opted for something limited to the postgres driver.

[vegerot]

Here is a workaround I found, tested with typeorm@0.2.28 with Postgres 9.6

Before calling selectQueryBuilder.stream(), I created a queryRunner myself like so:

const queryRunner = selectQueryBuilder.connection.createQueryRunner("slave");
selectQueryBuilder.setQueryRunner(queryRunner);

const stream  = await selectQueryBuilder.stream();
stream.on('error' => queryRunner.release());
stream.on('end' => queryRunner.release());
stream.on('data' => (rawRow) => { /* ... */ } );
...

@stvvt Using the latest version of typeorm (

❯ npm ls typeorm
├─┬ @nestjs/typeorm@7.1.5
│ └── typeorm@0.2.31 deduped
└── typeorm@0.2.31

) and postgres (psql (PostgreSQL) 13.2) I still experience this issue after @alvisetrevisan 's 9abe007.

Your comment was very helpful, though. This bug occurred in a migration, and I had to go from

    public async up(queryRunner: QueryRunner): Promise<void> {
        const stream = await queryRunner.stream(`SELECT "duration", "missionId", "landTime", "barAmount", "takeOffTime", "wingName" FROM "foos"`);
        for await (const foo of stream) {
            const barCount: { count: number }[] = await queryRunner.query(
              `SELECT COUNT(*) FROM "bars" WHERE "fooId" = $1;`,
              [foo.fooId],
            );
            for (let i = foo.barAmount - barCount[0].count; i > 0; --i) {
          
            }
        }

to

  public async up(selectQueryRunner: QueryRunner): Promise<void> {
    const queryRunner = selectQueryRunner.connection.createQueryRunner('slave')
    const stream = await queryRunner.stream(`SELECT "duration", "missionId", "landTime", "barAmount", "takeOffTime", "wingName" FROM "foos"`, undefined, () => queryRunner.release(), () => queryRunner.release());
    stream.on('error', () => queryRunner.release());
    stream.on('error', () => queryRunner.release());

    stream.on('data', async (foo: any) => {
      const barCount: { count: number }[] = await queryRunner.query(
        `SELECT COUNT(*) FROM "bars" WHERE "fooId" = $1;`,
        [foo.fooId],
      );
      for (let i = foo.barAmount - barCount[0].count; i > 0; i--) {
      }
    })

And that mostly worked.