resolve recursion issue and add more cleanup and logging/error handling

and refactor _checkRpcClient -> _checkSocket

lib/electrumx_rpc/subscribable_electrumx_client.dart

@@ -94,6 +94,7 @@ class SubscribableElectrumXClient {
     // Listen to global event bus for Tor status changes.
     _torStatusListener = bus.on<TorConnectionStatusChangedEvent>().listen(
       (event) async {
+        try {
           switch (event.newStatus) {
             case TorConnectionStatus.connecting:
               // If Tor is connecting, we need to wait.
@@ -110,6 +111,12 @@ class SubscribableElectrumXClient {
               _requireMutex = false;
               break;
           }
+        } finally {
+          // Ensure the lock is released.
+          if (_torConnectingLock.isLocked) {
+            _torConnectingLock.release();
+          }
+        }
       },
     );
 
@@ -166,7 +173,7 @@ class SubscribableElectrumXClient {
   /// Check if the RPC client is connected and connect if needed.
   ///
   /// If Tor is enabled but not running, it will attempt to start Tor.
-  Future<void> _checkRpcClient() async {
+  Future<void> _checkSocket({bool connecting = false}) async {
     if (_prefs.useTor) {
       // If we're supposed to use Tor...
       if (_torService.status != TorConnectionStatus.connected) {
@@ -193,6 +200,7 @@ class SubscribableElectrumXClient {
     }
 
     // Connect if needed.
+    if (!connecting) {
       if ((!_prefs.useTor && _socket == null) ||
           (_prefs.useTor && _socksSocket == null)) {
         if (currentFailoverIndex == -1) {
@@ -213,6 +221,7 @@ class SubscribableElectrumXClient {
         }
       }
     }
+  }
 
   /// Connect to the server.
   ///
@@ -221,6 +230,7 @@ class SubscribableElectrumXClient {
     required String host,
     required int port,
   }) async {
+    try {
       // Cache node information.
       _host = host;
       _port = port;
@@ -232,9 +242,10 @@ class SubscribableElectrumXClient {
 
       // If we're connecting to Tor, wait.
       if (_requireMutex) {
-      await _torConnectingLock.protect(() async => await _checkRpcClient());
+        await _torConnectingLock
+            .protect(() async => await _checkSocket(connecting: true));
       } else {
-      await _checkRpcClient();
+        await _checkSocket(connecting: true);
       }
 
       if (!Prefs.instance.useTor) {
@@ -309,6 +320,16 @@ class SubscribableElectrumXClient {
         _keepAlive,
         (_) async => _updateConnectionStatus(await ping()),
       );
+    } catch (e, s) {
+      final msg = "SubscribableElectrumXClient.connect: "
+          "failed to connect to $host:$port."
+          "\nError: $e\nStack trace: $s";
+      Logging.instance.log(msg, level: LogLevel.Fatal);
+
+      // Ensure cleanup is performed on failure to avoid resource leaks.
+      await disconnect(); // Use the disconnect method to clean up.
+      rethrow; // Rethrow the exception to handle it further up the call stack.
+    }
   }
 
   /// Connect to the server directly.
@@ -426,8 +447,28 @@ class SubscribableElectrumXClient {
   /// Disconnect from the server.
   Future<void> disconnect() async {
     _aliveTimer?.cancel();
+    _aliveTimer = null;
+
+    try {
       await _socket?.close();
+    } catch (e, s) {
+      Logging.instance.log(
+          "SubscribableElectrumXClient.disconnect: failed to close socket."
+          "\nError: $e\nStack trace: $s",
+          level: LogLevel.Warning);
+    }
+    _socket = null;
+
+    try {
       await _socksSocket?.close();
+    } catch (e, s) {
+      Logging.instance.log(
+          "SubscribableElectrumXClient.disconnect: failed to close SOCKS socket."
+          "\nError: $e\nStack trace: $s",
+          level: LogLevel.Warning);
+    }
+    _socksSocket = null;
+
     onConnectionStatusChanged = null;
   }
 
@@ -563,9 +604,9 @@ class SubscribableElectrumXClient {
   }) async {
     // If we're connecting to Tor, wait.
     if (_requireMutex) {
-      await _torConnectingLock.protect(() async => await _checkRpcClient());
+      await _torConnectingLock.protect(() async => await _checkSocket());
     } else {
-      await _checkRpcClient();
+      await _checkSocket();
     }
 
     // Check socket is connected.
@@ -629,9 +670,9 @@ class SubscribableElectrumXClient {
   }) async {
     // If we're connecting to Tor, wait.
     if (_requireMutex) {
-      await _torConnectingLock.protect(() async => await _checkRpcClient());
+      await _torConnectingLock.protect(() async => await _checkSocket());
     } else {
-      await _checkRpcClient();
+      await _checkSocket();
     }
 
     // Check socket is connected.
@@ -784,9 +825,9 @@ class SubscribableElectrumXClient {
   Future<bool> ping() async {
     // If we're connecting to Tor, wait.
     if (_requireMutex) {
-      await _torConnectingLock.protect(() async => await _checkRpcClient());
+      await _torConnectingLock.protect(() async => await _checkSocket());
     } else {
-      await _checkRpcClient();
+      await _checkSocket();
     }
 
     // Write to the socket.

lib/wallets/wallet/wallet_mixin_interfaces/electrumx_interface.dart

@@ -1019,6 +1019,12 @@ mixin ElectrumXInterface<T extends Bip39HDCurrency> on Bip39HDWallet<T> {
 
       // check and add appropriate addresses
       for (int k = 0; k < txCountBatchSize; k++) {
+        if (counts["${_id}_$k"] == null) {
+          print("121212");
+          print("${_id}_$k");
+          print("123123123");
+          print(counts);
+        }
         int count = counts["${_id}_$k"]!;
         if (count > 0) {
           iterationsAddressArray.add(txCountCallArgs["${_id}_$k"]!);