fix: check if first connection after failover is stale (#416)

Co-authored-by: Sophia Chu <112967780+sophia-bq@users.noreply.github.com>
Co-authored-by: joyc-bq <95259163+joyc-bq@users.noreply.github.com>

Author: 3 people

common/lib/host_list_provider/monitoring/cluster_topology_monitor.ts

@@ -202,7 +202,7 @@ export class ClusterTopologyMonitorImpl implements ClusterTopologyMonitor {
       try {
         client = await this._pluginService.forceConnect(this.initialHostInfo, this._monitoringProperties);
       } catch {
-        logger.error(Messages.get("ClusterTopologyMonitor.unableToConnect", this.initialHostInfo.hostId));
+        // Unable to connect to host;
         return null;
       }
 
@@ -214,7 +214,7 @@ export class ClusterTopologyMonitorImpl implements ClusterTopologyMonitor {
           if (writerId) {
             this.isVerifiedWriterConnection = true;
             this.writerHostInfo = this.initialHostInfo;
-            logger.info(Messages.get("ClusterTopologyMonitor.writerMonitoringConnection", this.initialHostInfo.hostId));
+            logger.info(Messages.get("ClusterTopologyMonitor.writerMonitoringConnection", this.initialHostInfo.host));
             writerVerifiedByThisTask = true;
           }
         } catch (error) {
@@ -449,11 +449,19 @@ export class HostMonitor {
             client = null;
           }
 
+          if (writerId) {
+            // First connection after failover may be stale.
+            if ((await this.monitor.pluginService.getHostRole(client)) !== HostRole.WRITER) {
+              logger.debug(Messages.get("HostMonitor.writerIsStale", writerId));
+              writerId = null;
+            }
+          }
+
           if (writerId) {
             if (this.monitor.hostMonitorsWriterClient) {
               await this.monitor.closeConnection(client);
             } else {
-              logger.debug(Messages.get("HostMonitor.detectedWriter", writerId));
+              logger.debug(Messages.get("HostMonitor.detectedWriter", writerId, this.hostInfo.host));
               const updatedHosts: HostInfo[] = await this.monitor.fetchTopologyAndUpdateCache(client);
               if (updatedHosts && this.monitor.hostMonitorsWriterClient === null) {
                 this.monitor.hostMonitorsWriterClient = client;

common/lib/plugins/failover2/failover2_plugin.ts

@@ -403,12 +403,12 @@ export class Failover2Plugin extends AbstractConnectionPlugin implements CanRele
           try {
             writerCandidateClient = await this.createConnectionForHost(writerCandidateHostInfo);
           } catch (err) {
-            this.logAndThrowError("Failover.unableToConnectToWriter");
+            this.logAndThrowError(Messages.get("Failover.unableToConnectToWriterDueToError", writerCandidateHostInfo.host, err.message));
           }
         }
 
         if (!writerCandidateClient) {
-          this.logAndThrowError("Failover.unableToConnectToWriter");
+          this.logAndThrowError(Messages.get("Failover.unableToConnectToWriter"));
         }
 
         if ((await this.pluginService.getHostRole(writerCandidateClient)) !== HostRole.WRITER) {
@@ -417,7 +417,7 @@ export class Failover2Plugin extends AbstractConnectionPlugin implements CanRele
           } catch (error) {
             // Do nothing.
           }
-          this.logAndThrowError(Messages.get("Failover2.failoverWriterConnectedToReader"));
+          this.logAndThrowError(Messages.get("Failover2.failoverWriterConnectedToReader", writerCandidateHostInfo.host));
         }
 
         await this.pluginService.abortCurrentClient();

common/lib/utils/locales/en.json

@@ -55,6 +55,7 @@
   "Failover.connectionChangedError": "The active SQL connection has changed due to a connection failure. Please re-configure session state if required.",
   "Failover.parameterValue": "%s = %s",
   "Failover.unableToConnectToWriter": "Unable to establish SQL connection to the writer instance.",
+  "Failover.unableToConnectToWriterDueToError": "Unable to establish SQL connection to the writer instance: %s due to error: %s.",
   "Failover.unableToConnectToReader": "Unable to establish SQL connection to the reader instance.",
   "Failover.unableToDetermineWriter": "Unable to determine the current writer instance.",
   "Failover.detectedError": "Detected an error while executing a command: %s",
@@ -209,9 +210,8 @@
   "ConfigurationProfileBuilder.canNotUpdateKnownPreset": "Can't add or update a built-in preset configuration profile '%s'.",
   "AwsClient.configurationProfileNotFound": "Configuration profile '%s' not found.",
   "AwsClient.targetClientNotDefined": "AwsClient targetClient not defined.",
-  "Failover2.unableToConnect": "Unable to establish SQL connection to the instance.",
   "Failover2.failoverReaderNotConnectedToReader": "Unable to establish SQL connection to the instance '%s' as a reader.",
-  "Failover2.failoverWriterConnectedToReader": "Unable to establish SQL connection to a writer instance.",
+  "Failover2.failoverWriterConnectedToReader": "The new writer was identified to be '%s', but querying the instance for its role returned a reader.",
   "Failover2.unableToFetchTopology": "Unable to establish SQL connection and fetch topology.",
   "Failover2.errorSelectingReaderHost": "An error occurred while attempting to select a reader host candidate: '%s'.",
   "Failover2.readerCandidateNull": "Reader candidate unable to be selected: '%s'.",
@@ -230,9 +230,10 @@
   "ClusterTopologyMonitor.errorDuringMonitoring": "Error thrown during cluster topology monitoring: '%s'.",
   "ClusterTopologyMonitor.endMonitoring": "Stop cluster topology monitoring.",
   "HostMonitor.startMonitoring": "Host monitor '%s' started.",
-  "HostMonitor.detectedWriter": "Detected writer: '%s'.",
+  "HostMonitor.detectedWriter": "Detected writer: '%s' - '%s'.",
   "HostMonitor.endMonitoring": "Host monitor '%s' completed in '%s'.",
   "HostMonitor.writerHostChanged": "Writer host has changed from '%s' to '%s'.",
+  "HostMonitor.writerIsStale": "Connected writer instance '%s' is stale.",
   "SlidingExpirationCacheWithCleanupTask.cleaningUp": "Cleanup interval of '%s' minutes has passed, cleaning up sliding expiration cache '%s'.",
   "SlidingExpirationCacheWithCleanupTask.cleanUpTaskInterrupted": "Sliding expiration cache '%s' cleanup task has been interrupted and is exiting.",
   "SlidingExpirationCacheWithCleanupTask.cleanUpTaskStopped": "Sliding expiration cache '%s' cleanup task has been stopped and is exiting.",

mysql/lib/dialect/rds_multi_az_mysql_database_dialect.ts

@@ -146,9 +146,8 @@ export class RdsMultiAZMySQLDatabaseDialect extends MySQLDatabaseDialect impleme
       if (!writerHostId) {
         const currentConnection = await this.identifyConnection(targetClient);
         return currentConnection ?? null;
-      } else {
-        return null;
       }
+      return writerHostId;
     } catch (error: any) {
       throw new AwsWrapperError(Messages.get("RdsMultiAZMySQLDatabaseDialect.invalidQuery", error.message));
     }

tests/integration/container/tests/aurora_failover2.test.ts

@@ -53,7 +53,6 @@ async function initDefaultConfig(host: string, port: number, connectToProxy: boo
     password: env.databaseInfo.password,
     port: port,
     plugins: "failover2",
-    failoverTimeoutMs: 250000,
     enableTelemetry: true,
     telemetryTracesBackend: "OTLP",
     telemetryMetricsBackend: "OTLP"