Bad state: Cannot add event after closing when reading large table

[davidmartos96]

When doing a SELECT * of a large table the library throws with:

Unhandled exception:
Bad state: Cannot add event after closing
#0      _MultiStreamController.addErrorSync (dart:async/stream_impl.dart:1131:24)
#1      _readMessages.<anonymous closure>.<anonymous closure>.handleChunk (package:postgres/src/v3/protocol.dart:77:20)
<asynchronous suspension>

It seems to happen from version 3.4.0 onwards.
One interesting thing I noticed while testing different versions is that version 3.4.0 is almost twice as slow than 3.3.0 when batching inserting many rows, necessary to trigger this issue.

Here is the reproduction code:

import 'package:postgres/postgres.dart';

Future<void> createTableAndPopulate(Connection conn) async {
  final sw = Stopwatch()..start();

  await conn.execute('''
    CREATE TABLE IF NOT EXISTS large_table (
      id SERIAL PRIMARY KEY,
      c1 INTEGER NOT NULL,
      c2 INTEGER NOT NULL,
      c3 TEXT NOT NULL,
      c4 TEXT NOT NULL,
      c5 TEXT NOT NULL,
      c6 TEXT NOT NULL,
      c7 TEXT NOT NULL,
      c8 TEXT NOT NULL,
      c9 TEXT NOT NULL,
      c10 TEXT NOT NULL
    )
  ''');

  final numBatches = 20;
  final batchSize = 5000;

  for (var i = 0; i < numBatches; i++) {
    print("Batch $i of $numBatches");
    final values = List.generate(
        batchSize,
        (i) => [
              i,
              i * 2,
              'value $i',
              'value $i',
              'value $i',

              'value $i',
              'value $i',
              'value $i',
              'value $i',
              'value $i',
            ]);

    final allArgs = values.expand((e) => e).toList();
    final valuesPart = List.generate(
            batchSize,
            (i) =>
                '(${List.generate(10, (j) => '\$${i * 10 + j + 1}').join(', ')})')
        .join(', ');

    final stmt =
        'INSERT INTO large_table (c1, c2, c3, c4, c5, c6, c7, c8, c9, c10) VALUES $valuesPart';
    await conn.execute(
      stmt,
      parameters: allArgs,
    );
  }

  print('Inserted ${numBatches * batchSize} rows in ${sw.elapsed}');
}

void main() async {
  final conn = await Connection.open(
    Endpoint(
      host: 'localhost',
      database: 'large',
      username: 'postgres',
      password: 'changeme',
      port: 5434,
    ),
    settings: ConnectionSettings(sslMode: SslMode.disable),
  );
  print('has connection!');

  await conn.execute('DROP SCHEMA IF EXISTS public CASCADE');
  await conn.execute('CREATE SCHEMA public');
  await createTableAndPopulate(conn);

  final rows = await conn.execute('SELECT * FROM large_table');
  print("SELECTED ROWS ${rows.length}");

  await conn.close();
}

[davidmartos96]

Regarding the speed difference, I did a git bisect and it looks like the cause is the following commit de56b3f

[isoos]

@davidmartos96: Thanks for the detailed case, I've added 99% of it in #399, however, neither local nor the CI test failed. What is your postgres version here?

[davidmartos96]

@isoos Weirdly enough, I'm not able to replicate on a test environment like your test, but I can reproduce it on a freshly created docker container and running dart bin/sample.dart

Version is PostgreSQL 17.2 (Debian 17.2-1.pgdg120+1) on x86_64-pc-linux-gnu, compiled by gcc (Debian 12.2.0-14) 12.2.0, 64-bit

name: test  

services:
  postgres:
    container_name: postgrestest
    image: postgres
    environment:
      POSTGRES_USER: ${POSTGRES_USER:-postgres}
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-changeme}
      PGDATA: /data/postgres
    volumes:
       - postgres:/data/postgres
    ports:
      - "5435:5432"
    restart: unless-stopped

volumes:
    postgres:

[isoos]

One small progress: if I use server.newConnection(sslMode: SslMode.disable) it will reach the error, otherwise it is fine. At least something to debug there...

[isoos]

I've released 3.4.5 with a fix for this issue. Huge thanks for the test, I am keeping it for both disabled and required SSL mode, and we may expand on it for further benchmarking the library to maybe debug this message parsing part if it is efficient or if there is a better way to do it. Closing this for now.

[davidmartos96]

@isoos Great, thank you!

[davidmartos96]

@isoos Is this fix suppose to fix also the speed difference caused by de56b3f ? Or is that what you meant by inefficient message parsing?

[isoos]

@davidmartos96: I wanted to fix the issue and restore correctness first, publishing it as a new version. Will take a look into message parsing later, possibly tomorrow or next week.