Repository navigation
Add pause and resume for replication streams - #799
Conversation
Co-authored-by: José Valim <jose.valim@gmail.com>
| ]} | ||
| end | ||
|
|
||
| defp handle_data(copies, %{paused: true} = s), do: {:keep_state, %{s | pending: copies}} |
There was a problem hiding this comment.
Shouldn't we be concatenating this? or is it guarantee this is called just once?
There was a problem hiding this comment.
it is guaranteed by the code indeed, but can maybe be made clearer with a comment and a pattern match on []. However this doesn't exist in the other pr, so let's see which one you prefer first before we do that.
|
Thank you! I understand the issue and the fix, but I am not sure if this is the best solution. We are now introducing a queue and the fact that we need to respond to pings and timeouts still has complexity to the "pause" perception. Perhaps a simpler alternative is that pause only pauses the next message but any other callback or message returning something other than pause kicks it back into life. But I am not sure if that's better. |
|
That works as well for my case I believe. If the context helps, here is how I fixed it on the Replicant end of things olivermt/replicant@a361b9d |
|
Can you please send a separate PR with the other appraoch then so we can compare them side by side? |
|
There is |
|
Sorry, I got absolutely sidetracked by life and disease :D @greg-rychlewski that is to control a replication from inside postgres, this replication is consumed from elixir (in Replicant), so that wont work :) The whole problem is from postgrex reading changes faster than replicant could put them into clickhouse, so even if we mimiced that solution in replicant, it wouldn't solve the issue of replays and slot ownership etc. @josevalim other approach here #804 |
I'm not too sure I understand what point you are making sorry. You are only receiving messages in Elixir because there is a subscription. If you pause the subscription using the query I gave you then it stops sending you messages without interfering with anything else coming over the connection. I think Jose is happy with the alternative though so it's ok to me to go ahead. I'm just not particularly convinced this is needed or even better than the query Postgres provides. |
|
Oh I forgot you are saying it is a large bulk import. So I guess everything is already dumped by Postgres before you can pause the subscription. Both solutions still feel kind of odd to me like there should be something better that the consumer can do without blockin the entire genserver or restarting it when you don't necessarily want to. But since I don't have an alternative solution right now I can't object too much. |
|
@greg-rychlewski my understanding is that Postgrex is in the middle of two systems: Postgres and a consumer. When the consumer is flooded, it needs to tell the postgrex connection to stop sending data, which this patch does. The other option is for this connection to start queueing this data but then you have more data in memory. My thinking, therefore, is that stopping consumption consuming the WAL is indeed a valid approach. WDYT? |
|
That is exactly it. Stopping postgrex from consuming using the ALTER is fine, but we need to stop postgrex from producing (in a buffering way) and thats what I am trying to solve. In the other PR that responsibility is now mostly at the consumer side (replicant). I have been giving this a lot of thought and I have no idea how to tackle this in any other way tbf. Once we consume off the sub, it has to buffer somewhere in elixir 🤷♂️ |
|
@josevalim the main thing that I think is niggling at me is that this seems like it should be handled by the consumer side. with this workaround you get some unusual caveats like avoiding the replication worker timing out and killing your connection or having wal buildup too much and kill the connection. Or having non replication messages restart replication (correct me if I’m wrong but I think this is what the other or does?) this seems like it should be handled on the consumer side by having configurable buffer size, controlling write speed, having some kind of persistent store to save changes. I might be wrong but I think this is what is normally done (think Debezium) and why these changes feel a bit off to me |
|
@greg-rychlewski I think all of those are natural consequences of back-pressure. If you assume the system is overloaded and you pause, then the replication worker may time out or WAL may build up indeed! But pushing downstream can also lead to failures if it cannot write to disk fast enough or disk is full. If a system is overlodead for a long period of time, then something somewhere will crash. Should we be allowed to push this backpressure upstream (by pausing)? Or always handling it downstream is better? |
It seems better handled outside of the driver, to me at least. I believe the driver should just be considered and event source and whoever is listening to the messages would have the usual ways of implementing back pressure. I'm not the biggest fan of the pausing mechanism because then people have to worry about the footguns (timeouts/wal buildup) that are usually meant to prevent catastrophic problems when consumers crash rather than being part of a well-behaving system. I'm not so opposed to it though that I want to cause a headache for everyone. I just wanted to say something because it was bothering me a little bit. |
|
@olivermt have you run this branch for a while to see if it effectively addresses the issues you have seen instead of causing further complications? |
|
This one yes, but not the other. Which one do you want? |
|
This one certainly, we can close the other one! |
|
There is one other thing that popped into my head. Not sure if you want to document it or if this is handled and I’m not noticing. But let’s say as part of your message processing you have to run a query. Iirc we allow that by returning :query from the callback. Will the query result get stuck behind the pending replication messages that we have queued up? |
|
@josevalim you approved the other one, thats why I was confused :D |
|
@greg-rychlewski it is handled, see latest commit, clarified it in a comment in latest commit |
|
@olivermt CI seems to be unhappy :) |
138c903 to
af0b82e
Compare
|
That SCRAM thing is not related to my code as I first thought, so yanked that back with a force push :) |
|
Trying to figure out what is up with the timing that makes the tests fail, but a little time constrained |
|
The last failure seems to be unrelated to this PR. |
|
There was a real failure I managed to reproduce locally where the keepalive didn't come soon enough. Seems to be all green now! |
Hi!
Discovered a bug in Replicant that required an addition to Postgrex to handle.
Basically a bulk import flooded WAL faster than CDC could process it. Postgrex kept reading with no pause mechanism, so Replicant’s buffer limit was exceeded and replication stalled. Replicant could detect overload but couldn’t stop Postgrex from reading more WAL. The patch adds pause/resume so CDC can drain its queue without disconnecting or losing replication-slot ownership.
Summary of code changes
Adds backpressure support to
Postgrex.ReplicationConnection.handle_data/2can return{:pause, state}or{:pause, replies, state}to stop socket delivery without disconnecting.handle_info/2andhandle_call/3can resume delivery with equivalent{:resume, ...}responses.Buffered messages retain their order, calls remain responsive, and replication feedback can still be sent while paused. Reconnecting clears buffered messages from the previous connection.
Tests cover TCP/TLS ordering, slot ownership, feedback while paused, and reconnect cleanup.