I have a Broadway Pipeline with a custom producer that is just using postgres to store tasks in normal table.

When I restart my application, I’m losing access to MyApp.Repo in different ways. Sometimes individual Processors cannot write to the database, but they will get acknowledged fine. Other times, acknowledgements will get lost. No matter what always with errors:

[error]
[info] Postgrex.Protocol (#PID<0.3666.0>) disconnected: ** (DBConnection.ConnectionError) client #PID<0.4048.0> exited

This makes sense why it’s happening — MyApp.Repo is getting shut down prior to MyApp.Pipeline. However, MyApp.Repo comes before MyApp.Pipeline in my supervision tree, so I’m not sure what the issue is. I assume this is because it takes time to drain the messages, but that doesn’t help me much in finding a solution.

I imagine there is something incredibly simple I can do with OTP to resolve this, but I’m unsure of what. My current attempt is to add the following to my application.ex

@impl true
def prep_stop(state) do
  :ok = Broadway.stop(MyApp.Pipeline, :shutdown)
  state
end

but based on how little information I can find on the prep_stop callback, I’m assuming this is not the best OTP way to go about things.