PgListener interface improvements

[Diggsey]

Whilst you can use try_recv to be notified of the connection dropping (and possible lost messages) you get the notification before the connection has been re-established, making it fairly useless. This is because more messages could be dropped between receiving None and the next time you call recv.

Here is some pseudocode for how I think a PgListener could be used:

let mut listener = PgListener::new(pool);
let mut maybe_work_item = None;

listener.listen("work_channel").await;

loop {
    if let Some(work_item) = maybe_work_item {
        // Busy state
        spawn_with_limit(work_item).await;

        maybe_work_item  = look_for_work().await;
        if maybe_work_item.is_none() {
            listener.listen("work_channel").await;
        }
    } else {
        // Idle state
        maybe_work_item  = look_for_work().await;
        if maybe_work_item.is_none() {
            // Wait to be notified of new work
            listener.recv().await;
        } else {
            listener.unlisten("work_channel");
        }
    }
}

Here, recv would return an enum with the following variants:

enum ListenerEvent {
    ConnectionLost,
    ConnectionEstablished,
    Notified(Notification),
}

ConnectionEstablished would only be sent after the LISTEN request, so that you can guarantee at-least-once delivery by waiting for either a Notified event or a ConnectionEstablished event.

[swlynch99]

Having run into this issue myself, a workaround to the connection only being re-established after the next call to try_recv is to run a no-op query on the connection before doing any recovery related to missing a message.

This looks something like this:

let mut listener = ...;

loop {
    let event = match listener.try_recv().await? {
        Some(event) => event,
        None => {
            // make the listener reconnect here, before we do anything
            sqlx::query("SELECT 1").execute(&mut listener).await?;
         
            recover_from_missing_event().await?
            continue;
         }
     };
     
     handle_event(event).await?;
}

[leoyvens]

Triage: I believe this was fixed in #3585

[abonander]

Great catch @leoyvens, I believe that PR does address this. If anyone comes across this and disagrees, please open a new issue.