transport: reconnect to the last endpoint without restarting tasks

Avoid restarting transport tasks during client reconnect.
Reuse the existing client transport and try the last successful endpoint
first.
If that fails, continue with the configured connect locators or scouted
locators. Restore the connection without replacing the transport object.

Closes: #1005
Closes: #1053

Author: steils

.github/workflows/ci.yml

@@ -352,9 +352,10 @@ jobs:
       - name: Build project and run tests
         run: |
           sudo apt install -y ninja-build
-          CMAKE_GENERATOR=Ninja ASAN=ON make BUILD_TYPE=Debug
+          CMAKE_GENERATOR=Ninja ASAN=ON ZENOH_DEBUG=3 make BUILD_TYPE=Debug
           ninja -C build/ test
           python3 ./build/tests/single_thread.py
+          python3 -u ./build/tests/connection_restore.py ./zenoh-standalone/zenohd --single-thread
         timeout-minutes: 15
         env:
           Z_FEATURE_MULTI_THREAD: 0

include/zenoh-pico/net/session.h

@@ -148,6 +148,7 @@ typedef struct _z_session_t {
 
 #if Z_FEATURE_AUTO_RECONNECT == 1
     _z_network_message_slist_t *_declaration_cache;
+    _z_string_t _last_connect_locator;
 #endif
 
     // Session subscriptions

include/zenoh-pico/transport/manager.h

@@ -31,6 +31,8 @@ enum _z_peer_op_e {
 
 z_result_t _z_new_transport(_z_transport_t *zt, const _z_id_t *bs, const _z_string_t *locator, z_whatami_t mode,
                             int peer_op, const _z_config_t *session_cfg, _z_runtime_t *runtime);
+z_result_t _z_reconnect_unicast_client_transport(_z_transport_unicast_t *ztu, const _z_string_t *locator,
+                                                 const _z_id_t *local_zid, const _z_config_t *session_cfg);
 z_result_t _z_new_peer(_z_transport_t *zt, const _z_id_t *session_id, const _z_string_t *locator,
                        const _z_config_t *session_cfg);
 bool _z_transport_open_error_is_retryable(z_result_t ret);

include/zenoh-pico/transport/unicast/transport.h

@@ -31,6 +31,8 @@ z_result_t _z_unicast_open_peer(_z_transport_unicast_establish_param_t *param, c
                                 const _z_id_t *local_zid, int peer_op, _z_sys_net_socket_t *socket);
 z_result_t _z_unicast_send_close(_z_transport_unicast_t *ztu, uint8_t reason, bool link_only);
 z_result_t _z_unicast_transport_close(_z_transport_unicast_t *ztu, uint8_t reason);
+void _z_unicast_transport_clear_connection(_z_transport_unicast_t *ztu);
+void _z_unicast_transport_replace_connection(_z_transport_unicast_t *dst, _z_transport_unicast_t *src);
 void _z_unicast_transport_clear(_z_transport_unicast_t *ztu);
 
 #ifdef __cplusplus

src/net/session.c

@@ -39,6 +39,7 @@
 #include "zenoh-pico/transport/unicast.h"
 #include "zenoh-pico/transport/unicast/lease.h"
 #include "zenoh-pico/transport/unicast/read.h"
+#include "zenoh-pico/transport/unicast/transport.h"
 #include "zenoh-pico/utils/config.h"
 #include "zenoh-pico/utils/logging.h"
 #include "zenoh-pico/utils/result.h"
@@ -130,6 +131,17 @@ static z_result_t _z_config_get_mode(const _z_config_t *config, z_whatami_t *mod
     return ret;
 }
 
+#if Z_FEATURE_AUTO_RECONNECT == 1
+static void _z_session_update_last_connect_locator(_z_session_t *s, const _z_string_t *locator) {
+    _z_string_t copy = _z_string_null();
+    if (_z_string_copy(&copy, locator) != _Z_RES_OK) {
+        return;
+    }
+    _z_string_clear(&s->_last_connect_locator);
+    (void)_z_string_move(&s->_last_connect_locator, &copy);
+}
+#endif
+
 static z_result_t _z_open_inner(_z_session_rc_t *zs, _z_string_t *locator, const _z_id_t *zid, int peer_op,
                                 const _z_config_t *config) {
     z_result_t ret = _Z_RES_OK;
@@ -141,6 +153,11 @@ static z_result_t _z_open_inner(_z_session_rc_t *zs, _z_string_t *locator, const
     }
     _z_transport_get_common(&zn->_tp)->_session = _z_session_rc_clone_as_weak(zs);
     _z_transport_get_common(&zn->_tp)->_state = _Z_TRANSPORT_STATE_OPEN;
+#if Z_FEATURE_AUTO_RECONNECT == 1
+    if ((zn->_mode == Z_WHATAMI_CLIENT) && (peer_op == _Z_PEER_OP_OPEN)) {
+        _z_session_update_last_connect_locator(zn, locator);
+    }
+#endif
 #if Z_FEATURE_MULTICAST_DECLARATIONS == 1
     if (zn->_tp._type == _Z_TRANSPORT_MULTICAST_TYPE) {
         ret = _z_interest_pull_resource_from_peers(zn);
@@ -542,6 +559,116 @@ z_result_t _z_open(_z_session_rc_t *zn, _z_config_t *config, const _z_id_t *zid)
 }
 
 #if Z_FEATURE_AUTO_RECONNECT == 1
+static z_result_t _z_reconnect_append_unique_locator(_z_string_svec_t *dst, const _z_string_t *locator) {
+    if ((locator == NULL) || !_z_string_check(locator)) {
+        return _Z_RES_OK;
+    }
+
+    size_t len = _z_string_svec_len(dst);
+    for (size_t i = 0; i < len; i++) {
+        if (_z_string_equals(_z_string_svec_get(dst, i), locator)) {
+            return _Z_RES_OK;
+        }
+    }
+
+    _z_string_t copy = _z_string_null();
+    _Z_RETURN_IF_ERR(_z_string_copy(&copy, locator));
+    z_result_t ret = _z_string_svec_append(dst, &copy, false);
+    if (ret != _Z_RES_OK) {
+        _z_string_clear(&copy);
+    }
+    return ret;
+}
+
+static z_result_t _z_client_reconnect_candidates(_z_session_t *s, _z_string_svec_t *candidates) {
+    _z_string_svec_t listen_locators = _z_string_svec_null();
+    _z_string_svec_t connect_locators = _z_string_svec_null();
+    _z_string_svec_t scouted_locators = _z_string_svec_null();
+
+    *candidates = _z_string_svec_null();
+    z_result_t ret = _z_reconnect_append_unique_locator(candidates, &s->_last_connect_locator);
+    if (ret != _Z_RES_OK) {
+        return ret;
+    }
+
+    ret = _z_config_get_all(&s->_config, &listen_locators, Z_CONFIG_LISTEN_KEY);
+    if (ret != _Z_RES_OK) {
+        goto exit;
+    }
+    ret = _z_config_get_all(&s->_config, &connect_locators, Z_CONFIG_CONNECT_KEY);
+    if (ret != _Z_RES_OK) {
+        goto exit;
+    }
+
+    size_t connect_len = _z_string_svec_len(&connect_locators);
+    for (size_t i = 0; i < connect_len; i++) {
+        ret = _z_reconnect_append_unique_locator(candidates, _z_string_svec_get(&connect_locators, i));
+        if (ret != _Z_RES_OK) {
+            goto exit;
+        }
+    }
+
+    if ((connect_len == 0) && (_z_string_svec_len(&listen_locators) == 0)) {
+        ret = _z_locators_by_scout(&s->_config, &s->_local_zid, &scouted_locators);
+        if (ret != _Z_RES_OK) {
+            goto exit;
+        }
+        size_t scouted_len = _z_string_svec_len(&scouted_locators);
+        for (size_t i = 0; i < scouted_len; i++) {
+            ret = _z_reconnect_append_unique_locator(candidates, _z_string_svec_get(&scouted_locators, i));
+            if (ret != _Z_RES_OK) {
+                goto exit;
+            }
+        }
+    }
+
+exit:
+    _z_string_svec_clear(&listen_locators);
+    _z_string_svec_clear(&connect_locators);
+    _z_string_svec_clear(&scouted_locators);
+
+    if (ret != _Z_RES_OK) {
+        _z_string_svec_clear(candidates);
+        return ret;
+    }
+    return _z_string_svec_is_empty(candidates) ? _Z_ERR_CONFIG_LOCATOR_INVALID : _Z_RES_OK;
+}
+
+static z_result_t _z_client_reopen_unicast(_z_session_t *s) {
+    _z_string_svec_t candidates = _z_string_svec_null();
+    _Z_RETURN_IF_ERR(_z_client_reconnect_candidates(s, &candidates));
+
+    z_result_t ret = _Z_ERR_TRANSPORT_OPEN_FAILED;
+    z_result_t last_retryable_ret = _Z_RES_OK;
+    z_result_t last_non_retryable_ret = _Z_RES_OK;
+    size_t len = _z_string_svec_len(&candidates);
+    for (size_t i = 0; i < len; i++) {
+        _z_string_t *locator = _z_string_svec_get(&candidates, i);
+        ret = _z_reconnect_unicast_client_transport(&s->_tp._transport._unicast, locator, &s->_local_zid, &s->_config);
+        if (ret == _Z_RES_OK) {
+            _z_session_update_last_connect_locator(s, locator);
+            break;
+        }
+        if (_z_transport_open_error_is_retryable(ret)) {
+            last_retryable_ret = ret;
+            continue;
+        }
+        last_non_retryable_ret = ret;
+    }
+
+    _z_string_svec_clear(&candidates);
+    if (ret == _Z_RES_OK) {
+        return _Z_RES_OK;
+    }
+    if (last_retryable_ret != _Z_RES_OK) {
+        return last_retryable_ret;
+    }
+    if (last_non_retryable_ret != _Z_RES_OK) {
+        return last_non_retryable_ret;
+    }
+    return _Z_ERR_TRANSPORT_OPEN_FAILED;
+}
+
 void _z_client_reopen_task_drop(void *ztc_arg) {
     _z_transport_common_t *tc = (_z_transport_common_t *)ztc_arg;
     if (tc->_state == _Z_TRANSPORT_STATE_RECONNECTING) {
@@ -555,47 +682,66 @@ _z_fut_fn_result_t _z_client_reopen_task_fn(void *ztc_arg, _z_executor_t *execut
     _z_transport_common_t *tc = (_z_transport_common_t *)ztc_arg;
     _z_transport_tasks_t tasks_handles = tc->_tasks;
     _z_session_rc_t zs = _z_session_weak_upgrade(&tc->_session);  // should not fail
+    if (_Z_RC_IS_NULL(&zs)) {
+        return _z_fut_fn_result_ready();
+    }
     _z_session_t *s = _Z_RC_IN_VAL(&zs);
-    _z_session_weak_drop(&tc->_session);
+    bool in_place_unicast = (s->_mode == Z_WHATAMI_CLIENT) && (s->_tp._type == _Z_TRANSPORT_UNICAST_TYPE) &&
+                            (tc == &s->_tp._transport._unicast._common);
+    if (!in_place_unicast) {
+        _z_session_weak_drop(&tc->_session);
+    }
 
     if (_z_config_is_empty(&s->_config)) {
         _z_session_rc_drop(&zs);
         return _z_fut_fn_result_ready();
     }
+
     _z_session_transport_mutex_lock(s);
-    z_result_t ret = _z_open(&zs, &s->_config, &s->_local_zid);
-    _z_session_transport_mutex_unlock(s);
+    z_result_t ret = in_place_unicast ? _z_client_reopen_unicast(s) : _z_open(&zs, &s->_config, &s->_local_zid);
     if (ret != _Z_RES_OK) {
-        if (ret == _Z_ERR_TRANSPORT_OPEN_FAILED || ret == _Z_ERR_SCOUT_NO_RESULTS ||
-            ret == _Z_ERR_TRANSPORT_TX_FAILED || ret == _Z_ERR_TRANSPORT_RX_FAILED ||
-            ret == _Z_ERR_TRANSPORT_RX_DURATION_EXPIRED || ret == _Z_ERR_TRANSPORT_OPEN_PARTIAL_CONNECTIVITY) {
+        _z_session_transport_mutex_unlock(s);
+        if ((ret == _Z_ERR_TRANSPORT_OPEN_FAILED) || (ret == _Z_ERR_SCOUT_NO_RESULTS) ||
+            (ret == _Z_ERR_TRANSPORT_TX_FAILED) || (ret == _Z_ERR_TRANSPORT_RX_FAILED) ||
+            (ret == _Z_ERR_TRANSPORT_RX_DURATION_EXPIRED) || (ret == _Z_ERR_TRANSPORT_OPEN_PARTIAL_CONNECTIVITY)) {
             _Z_DEBUG("Reopen failed, next try in 1s");
-            tc->_session = _z_session_rc_clone_as_weak(&zs);
+            if (!in_place_unicast) {
+                tc->_session = _z_session_rc_clone_as_weak(&zs);
+            }
             tc->_state = _Z_TRANSPORT_STATE_RECONNECTING;
             tc->_tasks = tasks_handles;
             _z_session_rc_drop(&zs);
             return _z_fut_fn_result_wake_up_after(1000);
-        } else {
-            _Z_ERROR("Reopen failed, will not retry");
-            tc->_state = _Z_TRANSPORT_STATE_CLOSED;
-            _z_session_rc_drop(&zs);
-            return _z_fut_fn_result_ready();
         }
+        _Z_ERROR("Reopen failed, will not retry");
+        tc->_state = _Z_TRANSPORT_STATE_CLOSED;
+        _z_session_rc_drop(&zs);
+        return _z_fut_fn_result_ready();
     }
-
+    tc = _z_transport_get_common(&s->_tp);
     tc->_tasks = tasks_handles;
+    _z_session_transport_mutex_unlock(s);
+
     if (!_z_network_message_slist_is_empty(s->_declaration_cache)) {
         _z_network_message_slist_t *iter = s->_declaration_cache;
         while (iter != NULL) {
             _z_network_message_t *n_msg = _z_network_message_slist_value(iter);
             ret = _z_send_n_msg(s, n_msg, Z_RELIABILITY_RELIABLE, Z_CONGESTION_CONTROL_BLOCK, NULL);
             if (ret != _Z_RES_OK) {
                 _Z_DEBUG("Send message during reopen failed: %i", ret);
-                _z_transport_clear(&s->_tp);
-                tc->_session = _z_session_rc_clone_as_weak(&zs);
-                tc->_state = _Z_TRANSPORT_STATE_RECONNECTING;
+                _z_session_transport_mutex_lock(s);
+                if (in_place_unicast && s->_tp._type == _Z_TRANSPORT_UNICAST_TYPE) {
+                    tc = &s->_tp._transport._unicast._common;
+                    _z_unicast_transport_clear_connection(&s->_tp._transport._unicast);
+                } else {
+                    _z_transport_clear(&s->_tp);
+                    tc->_session = _z_session_rc_clone_as_weak(&zs);
+                    tc->_state = _Z_TRANSPORT_STATE_RECONNECTING;
+                }
+                tc->_tasks = tasks_handles;
+                _z_session_transport_mutex_unlock(s);
                 _z_session_rc_drop(&zs);
-                return _z_fut_fn_result_continue();
+                return _z_fut_fn_result_wake_up_after(1000);
             }
 
             iter = _z_network_message_slist_next(iter);

src/session/utils.c

@@ -84,6 +84,7 @@ z_result_t _z_session_init(_z_session_t *zn, const _z_id_t *zid) {
     _z_config_init(&zn->_config);
 #if Z_FEATURE_AUTO_RECONNECT == 1
     zn->_declaration_cache = NULL;
+    zn->_last_connect_locator = _z_string_null();
 #endif
 
     // Initialize the data structs
@@ -180,6 +181,7 @@ z_result_t _z_session_close(_z_session_t *zn) {
     _Z_RETURN_IF_ERR(_z_session_mutex_lock(zn));
 #if Z_FEATURE_AUTO_RECONNECT == 1
     _z_network_message_slist_free(&zn->_declaration_cache);
+    _z_string_clear(&zn->_last_connect_locator);
 #endif
     _z_session_mutex_unlock(zn);
     _z_flush_local_resources(zn);

src/transport/common/tx.c

@@ -273,8 +273,12 @@ z_result_t _z_transport_tx_send_t_msg(_z_transport_common_t *ztc, const _z_trans
     // If sending to a peer list, make sure the peer mutex is locked
     _z_transport_tx_mutex_lock(ztc, true);
 
-    ret = _z_transport_tx_send_t_msg_inner(ztc, t_msg, peers);
+    if ((ztc->_state != _Z_TRANSPORT_STATE_OPEN) || (ztc->_link == NULL)) {
+        _z_transport_tx_mutex_unlock(ztc);
+        return _Z_ERR_TRANSPORT_NOT_AVAILABLE;
+    }
 
+    ret = _z_transport_tx_send_t_msg_inner(ztc, t_msg, peers);
     _z_transport_tx_mutex_unlock(ztc);
     return ret;
 }
@@ -298,6 +302,13 @@ static z_result_t _z_transport_tx_send_n_msg(_z_transport_common_t *ztc, const _
         return ret;
     }
     // Process message
+    if ((ztc->_state != _Z_TRANSPORT_STATE_OPEN) || (ztc->_link == NULL)) {
+        if (!_z_transport_batch_hold_tx_mutex()) {
+            _z_transport_tx_mutex_unlock(ztc);
+        }
+        return _Z_ERR_TRANSPORT_NOT_AVAILABLE;
+    }
+
     ret = _z_transport_tx_send_n_msg_inner(ztc, n_msg, reliability, peers);
     if (!_z_transport_batch_hold_tx_mutex()) {
         _z_transport_tx_mutex_unlock(ztc);
@@ -320,6 +331,13 @@ static z_result_t _z_transport_tx_send_n_batch(_z_transport_common_t *ztc, z_con
             return ret;
         }
         // Send batch
+        if ((ztc->_state != _Z_TRANSPORT_STATE_OPEN) || (ztc->_link == NULL)) {
+            if (!_z_transport_batch_hold_tx_mutex()) {
+                _z_transport_tx_mutex_unlock(ztc);
+            }
+            return _Z_ERR_TRANSPORT_NOT_AVAILABLE;
+        }
+
         _Z_DEBUG("Send network batch");
         ret = _z_transport_tx_flush_buffer(ztc, peers);
         if (!_z_transport_batch_hold_tx_mutex()) {
@@ -391,6 +409,10 @@ z_result_t _z_send_t_msg(_z_transport_t *zt, const _z_transport_message_t *t_msg
             ret = _z_transport_tx_send_t_msg(&zt->_transport._multicast._common, t_msg, NULL);
             break;
         case _Z_TRANSPORT_RAWETH_TYPE:
+            if (zt->_transport._raweth._common._state != _Z_TRANSPORT_STATE_OPEN) {
+                ret = _Z_ERR_TRANSPORT_NOT_AVAILABLE;
+                break;
+            }
             ret = _z_raweth_send_t_msg(&zt->_transport._raweth._common, t_msg);
             break;
         default:
@@ -530,6 +552,10 @@ z_result_t _z_send_n_msg(_z_session_t *zn, const _z_network_message_t *z_msg, z_
                 _z_transport_tx_send_n_msg(&zn->_tp._transport._multicast._common, z_msg, reliability, cong_ctrl, NULL);
             break;
         case _Z_TRANSPORT_RAWETH_TYPE:
+            if (zn->_tp._transport._raweth._common._state != _Z_TRANSPORT_STATE_OPEN) {
+                ret = _Z_ERR_TRANSPORT_NOT_AVAILABLE;
+                break;
+            }
             ret = _z_raweth_send_n_msg(zn, z_msg, reliability, cong_ctrl);
             break;
         default: