conn_mgr: fix duplicate connection and double-release bugs

- conn_mgr_connect_node: check instance->connections globally to reuse
  existing ETCP_CONN across groups instead of creating a new one (DUP addr)
- cm_start_phase_direct: close dead connection when etcp_link_new fails
  to prevent shadowing live connection in instance->connections hash
- cm_cleanup_db_node: remove redundant topo_node_registry_release
  (already done inside topo_node_free_lists)

src/routing_layer/conn_mgr.c

@@ -229,7 +229,6 @@ static void cm_cleanup_db_node(struct CONN_MGR_ENTRY* entry) {
     }
     DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: removing db_node 0x%016llx from group (no paths)", (unsigned long long)entry->node_id);
     queue_remove_data(group->nodes, &nq->ll);
-    topo_node_registry_release(group->instance->topo_groups, nq->node);
     topo_node_free_lists(group, nq);
     queue_entry_free(&nq->ll);
 }
@@ -270,6 +269,18 @@ int conn_mgr_connect_node(struct CONN_MGR* mgr, uint64_t node_id, uint32_t idle_
         return CONN_MGR_OK;
     }
     struct ETCP_CONN* existing = topo_group_find_conn_for_node(group, node_id);
+    if (!existing) {
+        struct ll_entry* ce = queue_find_data_by_index(mgr->instance->connections, (const uint8_t*)&node_id);
+        while (ce) {
+            struct conn_queue_entry* cqe = (struct conn_queue_entry*)ce->data;
+            if (cqe->conn && cqe->conn->peer_node_id == node_id && cqe->conn->links_up) {
+                existing = cqe->conn;
+                topo_group_new_conn(group, existing);
+                break;
+            }
+            ce = queue_find_next_by_index(mgr->instance->connections, (const uint8_t*)&node_id, ce);
+        }
+    }
     if (existing && existing->peer_node_id == node_id && existing->links) {
         struct ETCP_LINK* l = existing->links;
         while (l) { if (l->link_status && l->initialized) break; l = l->next; }
@@ -562,6 +573,7 @@ static void cm_start_phase_direct(struct CONN_MGR_ENTRY* entry) {
                             entry->main.timer = uasync_set_timeout(entry->mgr->instance->ua, (int)(entry->mgr->direct_timeout_ms * 10), entry, cm_direct_timeout_cb, "conn_mgr_db_direct");
                             return;
                         }
+                        etcp_connection_close(conn);
                     }
                 }
                 s = s->next;
@@ -595,6 +607,7 @@ static void cm_start_phase_direct(struct CONN_MGR_ENTRY* entry) {
                                 entry->main.timer = uasync_set_timeout(entry->mgr->instance->ua, (int)(entry->mgr->direct_timeout_ms * 10), entry, cm_direct_timeout_cb, "conn_mgr_direct");
                                 return;
                             }
+                            etcp_connection_close(conn);
                         }
                     }
                     s = s->next;