Question: close the connection of the streamed query before the results are fully read about pgwire HOT 7 CLOSED

True cancellation of queries is not implemented yet. In current implementation stream.destroy() just reads and ignores records until end of result. The problem with implementation is that postgres protocol allows to cancel current query only. Its implossible to gracefully cancel all queued queries as far as I know, so cancellation algorithm for pgwire is not clear for me now. I will think about it now.

Cancellation of replication stream uses different protocol and it is implemented.

You queryStream function is almost ok. You can avoid monkey patch and just subscribe to stream event by using builtin utility.

import { finished } from 'stream';

finished(stream, _ => conn.end());

This code will also handle successfull (non-destroy) case.

from pgwire.

upd
finished(stream, _ => conn.end()); will swallow error silenlty. Need to rethrow it somehow

from pgwire.

This is what I had to do actually:

finished(stream, err => {
  conn.destroy();
  if (err && err.code !== "ERR_STREAM_PREMATURE_CLOSE") {
    throw err;
  }
});

And this is still not perfect, because for logicalReplication stream I suspect it may eat the last ackImmediate().

What I try to achieve for logicalReplication stream is to: a) send it a signal "please terminate gracefully", b) wait until this signal is correctly processed (i.e. all acks are flushed), and c) then destroy the connection.

How do I do this? I suspect that there is no way currently?

from pgwire.

And this is still not perfect, because for logicalReplication stream I suspect it may eat the last ackImmediate().

replicationStream.destroy() stops replication stream gracefully. Call of replicationStream.ackImmediate() after replicationStream.destroy() is prohibited and exception will be thrown #9 #18

and after destroy of replicationStream you can use regular conn.end() as you wrote in the first comment.

from pgwire.

I just understood that pgwire.pool already has logicalReplication function that behaves exactly as you want.

  logicalReplication(...args) {
    const conn = new Connection({
      ...this._options,
      replication: 'database',
    });
    return conn.logicalReplication(...args).finally(_ => {
      conn.end();
    });
  }

I forgot that there is no need to use finished at all. Just call conn.end() after replicationStream is created and gracefull connection termination will be queued and executed automatically after replication stream destroy

from pgwire.

I'm trying to implement query cancellation for cases when there is only one pending query.

• if queueSize > 1 then return and procceed with current skip-based implementation
• block connection tx to prevent sending new queries while cancelling
• send CancelRequest and wait for cancellation socket terminated by backend
• unblock connection tx

from pgwire.

query cancellation is implemented

from pgwire.