Implement reconnect handling on error

[ported-pw]

I need a way to have persistent sessions, so I am currently looking into what a good design is for reconnect handling (potentially with exponential back-off and picking new servers) on error.

[ported-pw]

I have spent like 2 days on this now and I have yet to reach a satisfactory approach for this, especially one that doesn't require big changes to the current connection API.
My goal was to implement retry logic fully transparently, e.g. a call would wait until the connection is established.

[ported-pw]

use futures_util::future::BoxFuture;
use futures_util::FutureExt;
use human_repr::HumanDuration;
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
use steam_vent::{Connection, ConnectionError};
use tokio::sync::Mutex;
use tokio::time::sleep;
use tracing::{error, info};

#[derive(Default)]
enum ConnectionState {
    #[default]
    NeedReconnect,
    Connected(Connection),
}

type ConnectionResult = Result<Connection, ConnectionError>;

#[derive(Clone)]
pub struct ConnectionHolder {
    max_attempts: Option<u32>,
    state: Arc<Mutex<ConnectionState>>,
    creation_fn: Arc<Box<dyn Fn() -> BoxFuture<'static, ConnectionResult> + Sync + Send>>,
}

impl ConnectionHolder {
    pub fn new<CreationFnFut>(
        max_attempts: Option<u32>,
        creation_fn: impl Fn() -> CreationFnFut + Send + Sync + 'static,
    ) -> Self
    where
        for<'any> CreationFnFut: Future<Output = ConnectionResult> + Send + 'any,
    {
        Self {
            max_attempts,
            state: Default::default(),
            creation_fn: Arc::new(Box::new(move || creation_fn().boxed())),
        }
    }

    pub async fn force_reconnect(&self) {
        let mut guard = self.state.lock().await;
        *guard = ConnectionState::NeedReconnect;
    }

    pub async fn checkout(&self) -> Result<Connection, ConnectionError> {
        let mut guard = self.state.lock().await;
        match &*guard {
            ConnectionState::Connected(connection) => Ok(connection.clone()),
            ConnectionState::NeedReconnect => {
                info!("Reconnect required");
                let mut current_attempt = 0;
                loop {
                    match (self.creation_fn)().await {
                        Err(e) => {
                            if let Some(max_attempts) = self.max_attempts {
                                if current_attempt >= max_attempts {
                                    error!(error = ?e, "Failed to connect");
                                    return Err(e);
                                }
                                current_attempt += 1;
                                // TODO add jitter
                                let delay = Duration::from_secs(2u64.pow(current_attempt));
                                error!(error = ?e, "Failed to connect, retrying in {}", delay.human_duration());
                                sleep(delay).await;
                            }
                        }
                        Ok(connection) => {
                            info!("Connected successfully");
                            *guard = ConnectionState::Connected(connection.clone());
                            return Ok(connection);
                        }
                    }
                }
            }
        }
    }
}

This is pretty much what I came up with so far, for at least doing this "from the outside" with explicit checkout and error handling. I think to integrate it, we probably need to split the Connection traits some more, maybe pull out the heartbeat as well and use that for checking the connection?

[icewind1991]

(Haven't had time to properly dig into this or look into the wip PR, so I might be saying things you already considered)

Pulling out the heartbeat handling probably makes sense.

I don't think it would be great to use that for reconnect detection though. As there would be a decent chance of it being "to late" and only detecting the issue after a request has already failed.

My initial approach would be to have a method on NetworkError to check if it's a connection issue (and we should thus reconnect) or a logic issue. And have a wrapper around Connection that will automatically reconnect+retry.

Having said that I'm not a 100% if retrying on connection failures should be done automatically, make it opt in per request, or let the user handle it themselves. The main worry would be cases where the connection dies between sending the request and handling the response, there steam has received the command and we end up sending the same command twice.

[ported-pw]

I would certainly want to make it in a way that it is an entirely optional feature, as in, a wrapper/another implementor of something.

My initial approach would be to have a method on NetworkError to check if it's a connection issue (and we should thus reconnect) or a logic issue. And have a wrapper around Connection that will automatically reconnect+retry.

Basically, I have had trouble with getting exactly that working using the current interfaces, and I couldn't find a decent approach.

My first idea was to replace Sender and MessageFilter with error-handling versions, somehow await a reconnect in the main struct, and continue.
This might still be a viable solution, I didn't pursue it fully because it also needs to track connection state properly (so that e.g. a send during the login process won't cause an immediate retry as well if it fails), and I didn't realise that until I had already switched approaches.

The idea I went with is more what you described, in this case I attempted to wrap a Connection, but implementing ConnectionTrait on that didn't really work out as the required sender, filter, and session functions can't really be implemented in a meaningful way, so some connection API changes probably need to happen to make this work nicely (but I'm not entirely sure what those are).

I think next I would go back to my first idea and include a state a bit like what I posted above, but I don't have the time right now to complete this in a nice way for the library, so I will probably come back to this later.

[icewind1991]

but implementing ConnectionTrait on that didn't really work out as the required sender, filter, and session functions can't really be implemented in a meaningful way

Moving the common implementation of ConnectionTrait from the trait to a blanket impl<T: ConnectionImpl> ConnectionTrait for T should allow getting rid of that requirement

[icewind1991]

but implementing ConnectionTrait on that didn't really work out as the required sender, filter, and session functions can't really be implemented in a meaningful way

Moving the common implementation of ConnectionTrait from the trait to a blanket impl<T: ConnectionImpl> ConnectionTrait for T should allow getting rid of that requirement

488d0e8