Is there a way to tell PG not to send messages after the one I acknowledged last ?

[lorenzofox3]

I understand that acknowledge is more an instruction for PG to clean the related WALs, but I was wondering if there is a way to tell postgres to stop sending me messages until I have acknowledged the one I am currently processing ?

[jpdsteno]

I don't see a way. I think this package needs to expose the pg connection in the data event handler args. Without this, I don't see a way to control back pressure. With it, you can pause the connection's stream and flush the current message before taking on the next. This work would need to be done within the wal_sender_timeout to avoid a connection close. I wonder if we could also just make this lib pause the stream until acknowledge is called.

[lorenzofox3]

I ended up stopping the service and going back to listening mode on demand. It may not be the most efficient, but after conducting some tests, it does the job.

I wrote a blog post about my experiments

[WeismannS]

This
This is a bit of critical issue since the library does not handle async handlers and keeps adding messages to the buffer
I'm increasinly concerned of going OOM if my pipeline can't handle the endless streams
I'll be forced to service.stop() on demend too in this case but i don't see that as a very good solution
@kibae pinging you here, since this is a bit of a big deal and hope we find a solution to the library in the future, thank you for all your work

[kibae]

Hi, @lorenzofox3 @jpdsteno @lorenzofox3 @WeismannS

I encountered this issue years ago myself, but hadn't addressed it when implementing the library. In version 1.x, before the TypeScript rewrite, accessing the internal stream of the pg connection was challenging.

Thanks to your feedback, I've implemented a solution with a new flowControl option that controls the pg connection stream to enable backpressure support.

How it works:

• When flowControl.enabled is true, the stream is paused while processing each message
• Messages are queued and processed sequentially using EventEmitter2.emitAsync() to properly await async handlers
• The stream resumes only after the handler completes
• This prevents memory overflow (OOM) when handlers are slower than the incoming message rate

Usage:

const service = new LogicalReplicationService(clientConfig, {
  flowControl: { enabled: true }
});

service.on('data', async (lsn, log) => {
  await someSlowAsyncOperation(log);  // Safe: next message waits for this to complete
});

The default is enabled: false for backward compatibility.

See PR #134

Once the tests are complete, I'll publish version 2.3.0 to npm.

Thank you again for bringing this up. Your feedback is always welcome! 👍

[WeismannS]

Oh wow that was pretty fast...
I've finished reading the commit, it's exactly what i needed -- thank you very much!