fix: remove race condition bug in refresh logic

core/src/main/java/com/google/cloud/sql/core/CloudSqlInstance.java

@@ -28,7 +28,6 @@
 import dev.failsafe.RateLimiter;
 import java.io.IOException;
 import java.security.KeyPair;
-import java.time.Duration;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.List;
@@ -57,8 +56,7 @@ class CloudSqlInstance {
   private final ListenableFuture<KeyPair> keyPair;
   private final Object instanceDataGuard = new Object();
   // Limit forced refreshes to 1 every minute.
-  private final RateLimiter<Object> forcedRenewRateLimiter =
-      RateLimiter.burstyBuilder(2, Duration.ofSeconds(30)).build();
+  private final RateLimiter<Object> forcedRenewRateLimiter;
 
   private final RefreshCalculator refreshCalculator = new RefreshCalculator();
 
@@ -68,6 +66,9 @@ class CloudSqlInstance {
   @GuardedBy("instanceDataGuard")
   private ListenableFuture<InstanceData> nextInstanceData;
 
+  @GuardedBy("instanceDataGuard")
+  private boolean forceRefreshRunning;
+
   /**
    * Initializes a new Cloud SQL instance based on the given connection name.
    *
@@ -82,12 +83,14 @@ class CloudSqlInstance {
       AuthType authType,
       CredentialFactory tokenSourceFactory,
       ListeningScheduledExecutorService executor,
-      ListenableFuture<KeyPair> keyPair) {
+      ListenableFuture<KeyPair> keyPair,
+      RateLimiter<Object> forcedRenewRateLimiter) {
     this.instanceName = new CloudSqlInstanceName(connectionName);
     this.instanceDataSupplier = instanceDataSupplier;
     this.authType = authType;
     this.executor = executor;
     this.keyPair = keyPair;
+    this.forcedRenewRateLimiter = forcedRenewRateLimiter;
 
     if (authType == AuthType.IAM) {
       HttpRequestInitializer source = tokenSourceFactory.create();
@@ -159,20 +162,21 @@ String getPreferredIp(List<String> preferredTypes) {
    */
   void forceRefresh() {
     synchronized (instanceDataGuard) {
-      nextInstanceData.cancel(false);
-      if (nextInstanceData.isCancelled()) {
-        logger.fine(
-            "Force Refresh: the next refresh operation was cancelled."
-                + " Scheduling new refresh operation immediately.");
-        currentInstanceData = executor.submit(this::performRefresh);
-        nextInstanceData = currentInstanceData;
-      } else {
-        logger.fine(
-            "Force Refresh: the next refresh operation is already running."
-                + " Marking it as the current operation.");
-        // Otherwise it's already running, so just move next to current.
-        currentInstanceData = nextInstanceData;
+      // Don't force a refresh until the current forceRefresh operation
+      // has produced a successful refresh.
+      if (forceRefreshRunning) {
+        return;
       }
+
+      forceRefreshRunning = true;
+      nextInstanceData.cancel(false);
+      logger.fine(
+          String.format(
+              "[%s] Force Refresh: the next refresh operation was cancelled."
+                  + " Scheduling new refresh operation immediately.",
+              instanceName));
+      currentInstanceData = executor.submit(this::performRefresh);
+      nextInstanceData = currentInstanceData;
     }
   }
 
@@ -182,10 +186,14 @@ void forceRefresh() {
    * would expire.
    */
   private InstanceData performRefresh() throws InterruptedException, ExecutionException {
-    logger.fine("Refresh Operation: Acquiring rate limiter permit.");
+    logger.fine(
+        String.format("[%s] Refresh Operation: Acquiring rate limiter permit.", instanceName));
     // To avoid unreasonable SQL Admin API usage, use a rate limit to throttle our usage.
     forcedRenewRateLimiter.acquirePermit();
-    logger.fine("Refresh Operation: Acquired rate limiter permit. Starting refresh...");
+    logger.fine(
+        String.format(
+            "[%s] Refresh Operation: Acquired rate limiter permit. Starting refresh...",
+            instanceName));
 
     try {
       InstanceData data =
@@ -194,15 +202,16 @@ private InstanceData performRefresh() throws InterruptedException, ExecutionExce
 
       logger.fine(
           String.format(
-              "Refresh Operation: Completed refresh with new certificate expiration at %s.",
-              data.getExpiration().toInstant().toString()));
+              "[%s] Refresh Operation: Completed refresh with new certificate expiration at %s.",
+              instanceName, data.getExpiration().toInstant().toString()));
       long secondsToRefresh =
           refreshCalculator.calculateSecondsUntilNextRefresh(
               Instant.now(), data.getExpiration().toInstant());
 
       logger.fine(
           String.format(
-              "Refresh Operation: Next operation scheduled at %s.",
+              "[%s] Refresh Operation: Next operation scheduled at %s.",
+              instanceName,
               Instant.now()
                   .plus(secondsToRefresh, ChronoUnit.SECONDS)
                   .truncatedTo(ChronoUnit.SECONDS)
@@ -212,12 +221,17 @@ private InstanceData performRefresh() throws InterruptedException, ExecutionExce
         currentInstanceData = Futures.immediateFuture(data);
         nextInstanceData =
             executor.schedule(this::performRefresh, secondsToRefresh, TimeUnit.SECONDS);
+        // Refresh completed successfully, reset forceRefreshRunning.
+        forceRefreshRunning = false;
       }
-
       return data;
     } catch (ExecutionException | InterruptedException e) {
       logger.log(
-          Level.FINE, "Refresh Operation: Failed! Starting next refresh operation immediately.", e);
+          Level.FINE,
+          String.format(
+              "[%s] Refresh Operation: Failed! Starting next refresh operation immediately.",
+              instanceName),
+          e);
       synchronized (instanceDataGuard) {
         nextInstanceData = executor.submit(this::performRefresh);
       }
@@ -228,4 +242,20 @@ private InstanceData performRefresh() throws InterruptedException, ExecutionExce
   SslData getSslData() {
     return getInstanceData().getSslData();
   }
+
+  ListenableFuture<InstanceData> getNext() {
+    synchronized (instanceDataGuard) {
+      return this.nextInstanceData;
+    }
+  }
+
+  ListenableFuture<InstanceData> getCurrent() {
+    synchronized (instanceDataGuard) {
+      return this.currentInstanceData;
+    }
+  }
+
+  public CloudSqlInstanceName getInstanceName() {
+    return instanceName;
+  }
 }

core/src/main/java/com/google/cloud/sql/core/CloudSqlInstanceName.java

@@ -70,4 +70,9 @@ String getRegionId() {
   String getInstanceId() {
     return instanceId;
   }
+
+  @Override
+  public String toString() {
+    return connectionName;
+  }
 }

core/src/main/java/com/google/cloud/sql/core/CoreSocketFactory.java

@@ -24,13 +24,15 @@
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningScheduledExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
+import dev.failsafe.RateLimiter;
 import java.io.File;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.security.KeyPair;
 import java.security.KeyPairGenerator;
 import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
@@ -125,6 +127,7 @@ static ListeningScheduledExecutorService getDefaultExecutor() {
     // there should be enough free threads so that there will not be a deadlock. Most users
     // configure 3 or fewer instances, requiring 6 threads during refresh. By setting
     // this to 8, it's enough threads for most users, plus a safety factor of 2.
+
     ScheduledThreadPoolExecutor executor =
         (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(8);
 
@@ -350,6 +353,12 @@ CloudSqlInstance getCloudSqlInstance(String instanceName, AuthType authType) {
         instanceName,
         k ->
             new CloudSqlInstance(
-                k, adminApiService, authType, credentialFactory, executor, localKeyPair));
+                k,
+                adminApiService,
+                authType,
+                credentialFactory,
+                executor,
+                localKeyPair,
+                RateLimiter.burstyBuilder(2, Duration.ofSeconds(30)).build()));
   }
 }

core/src/main/java/com/google/cloud/sql/core/SqlAdminApiFetcher.java

@@ -103,8 +103,7 @@ public InstanceData getInstanceData(
       ListenableFuture<KeyPair> keyPair)
       throws ExecutionException, InterruptedException {
 
-    ListenableFuture<Optional<AccessToken>> token =
-        executor.submit(() -> accessTokenSupplier.get());
+    ListenableFuture<Optional<AccessToken>> token = executor.submit(accessTokenSupplier::get);
 
     // Fetch the metadata
     ListenableFuture<Metadata> metadataFuture =
@@ -152,14 +151,18 @@ public InstanceData getInstanceData(
                             .orElse(x509Certificate.getNotAfter());
                   }
 
+                  logger.fine(String.format("[%s] INSTANCE DATA DONE", instanceName));
+
                   return new InstanceData(
                       Futures.getDone(metadataFuture),
                       Futures.getDone(sslContextFuture),
                       expiration);
                 },
                 executor);
 
-    return done.get();
+    InstanceData instanceData = done.get();
+    logger.fine(String.format("[%s] ALL FUTURES DONE", instanceName));
+    return instanceData;
   }
 
   String getApplicationName() {
@@ -219,6 +222,9 @@ private Metadata fetchMetadata(CloudSqlInstanceName instanceName, AuthType authT
       try {
         Certificate instanceCaCertificate =
             createCertificate(instanceMetadata.getServerCaCert().getCert());
+
+        logger.fine(String.format("[%s] METADATA DONE", instanceName));
+
         return new Metadata(ipAddrs, instanceCaCertificate);
       } catch (CertificateException ex) {
         throw new RuntimeException(
@@ -288,6 +294,8 @@ private Certificate fetchEphemeralCertificate(
           ex);
     }
 
+    logger.fine(String.format("[%s %d] CERT DONE", instanceName, Thread.currentThread().getId()));
+
     return ephemeralCertificate;
   }
 
@@ -339,6 +347,9 @@ private SslData createSslData(
 
       sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), new SecureRandom());
 
+      logger.fine(
+          String.format("[%s %d] SSL CONTEXT", instanceName, Thread.currentThread().getId()));
+
       return new SslData(sslContext, kmf, tmf);
     } catch (GeneralSecurityException | IOException ex) {
       throw new RuntimeException(