[redis-rs][core] Move connection refresh to the background

[GilboaAWS]

Issue link

This Pull Request is linked to issue (URL): [#2910]

Checklist

Before submitting the PR make sure the following are checked:

• This Pull Request is related to one issue.
• Commit message has a detailed description of what changed and why.
• Tests are added or updated.
• CHANGELOG.md and documentation files are updated.
• Destination branch is correct - main or release
• Create merge commit if merging release branch into main, squash otherwise.

[ikolomi]

This is a substantial change in the core mechanics. It must include tests that fail w/o this change to prove it necessity

[nihohit]

instead of splitting your state between multiple maps and leaving room for error in reading the wrong map, consider just use an enum representing each connection state (Usable, Reconnecting, etc.) in the main connection map.

See

valkey-glide/glide-core/src/client/reconnecting_connection.rs

Line 43 in ffc679a

enum ConnectionState {

for an example on how to do this.

[barshaul]

First comments (stopped before fn update_refreshed_connection).
Please see all inline comments and see how it behaves with cases where refresh connections is being called only with management/user/both. Also, maybe I haven't got to it yet - but the refresh task should be bounded to the lifetime of the clusterNode / address, so it would be cancelled when refresh_slots removes it from the topology

[barshaul]

as we talked - instead of using RefreshState, use a ConnectionState that internally holds the operation / connection (same as is reconnectingConnection). For example -

enum ConnectionState<Connection> {
     Connected(ConnectionDetails<Connection>>),
     Reconnecting(JoinHandle<()>),
}

pub struct ClusterNode<Connection> {
    pub user_connection: ConnectionState<Connection>,
    pub management_connection: Option<ConnectionState<Connection>>,
}

or

enum ConnectionState {
     Connected(ClusterNode<Connection>>),
     Reconnecting(handle),
}

connection_map: DashMap<String, ConnectionState<Connection>>,

the refresh_connections can be called only for user/management connection or for both, so you should make sure this solution covers all cases

[barshaul]

I don't see a good reason for using DashSet or DashMap, these structs are only going to be used from a single point without concurrency

[barshaul]

1. You can use the completed addresses to hold the newly created cluster node.
2. Instead of throwing all refresh structs directly into the connection container we might be better by wrapping all updates with some struct, so later on we'll be able to add refresh slots or other updates there too.
   For example:

struct RefreshUpdates {
    in_progress_addresses: HashSet<String>, 
    completed_addresses: HashMap<String, ClusterNode<Connection>>, 
     // refreshed_slot_map: Option<SlotMap>,
}

impl RefreshUpdates {
    pub(crate) fn in_progress(&self) -> &HashSet<String> {
        &self.in_progress_addresses
    }

    pub(crate) fn add_refresh(&mut self, address: String) {
        self.in_progress_addresses.insert(address);
    }

    pub(crate) fn complete_refresh(&mut self, address: String, node: ClusterNode<Connection>) {
        if !self.in_progress_addresses.remove(&address) {
            warn!("Attempted to complete refresh for an address not in progress: {}", address);
        }
        self.completed_addresses.insert(address, node);
    }
    
    pub(crate) fn completed_addresses(&mut self) -> HashMap<String, ClusterNode<Connection>> {
        std::mem::take(&mut self.completed_addresses)
    }
}

[barshaul]

lets have an API in the connection_container for "is_reconnecting(address)".
And you can use filter and for each:

        addresses
            .into_iter()
            .filter(|address| !connections_container.is_reconnecting(address))
            .for_each(|address| {
                let handle = tokio::spawn(async move {
                    info!("Refreshing connection task to {:?} started", address);
                    ...
                });
            });

[barshaul]

why do you wrap it with async block again?

[barshaul]

Suggested change

	debug!("refresh connection tasts initiated");
	debug!("refresh connection tasks initiated");

[barshaul]

it creates an unnecessary copy. instead you can change failed_connections to be a set to begin with