feat: Periodicaly check if domain name changed and close connections to old database.

wip: periodic check for domain change

wip: periodic checks

wip: close sockets on instance closed

Author: hessjcg

src/cloud-sql-instance.ts

@@ -14,13 +14,23 @@
 
 import {IpAddressTypes, selectIpAddress} from './ip-addresses';
 import {InstanceConnectionInfo} from './instance-connection-info';
-import {resolveInstanceName} from './parse-instance-connection-name';
+import {isSameInstance, resolveInstanceName} from './parse-instance-connection-name';
 import {InstanceMetadata} from './sqladmin-fetcher';
 import {generateKeys} from './crypto';
 import {RSAKeys} from './rsa-keys';
 import {SslCert} from './ssl-cert';
 import {getRefreshInterval, isExpirationTimeValid} from './time';
 import {AuthTypes} from './auth-types';
+import {CloudSQLConnectorError} from './errors';
+
+// Private types that describe exactly the methods
+// needed from tls.Socket to be able to close
+// sockets when the DNS Name changes.
+type EventFn = () => void;
+type ClosableSocket = {
+  destroy: (error?: Error) => void;
+  once: (name: string, handler: EventFn) => void;
+};
 
 interface Fetcher {
   getInstanceMetadata({
@@ -42,6 +52,7 @@ interface CloudSQLInstanceOptions {
   ipType: IpAddressTypes;
   limitRateInterval?: number;
   sqlAdminFetcher: Fetcher;
+  checkDomainInterval?: number;
 }
 
 interface RefreshResult {
@@ -74,9 +85,13 @@ export class CloudSQLInstance {
   // The ongoing refresh promise is referenced by the `next` property
   private next?: Promise<RefreshResult>;
   private scheduledRefreshID?: ReturnType<typeof setTimeout> | null = undefined;
+  private checkDomainID?: ReturnType<typeof setInterval> | null = undefined;
   /* eslint-disable-next-line @typescript-eslint/no-explicit-any */
   private throttle?: any;
   private closed = false;
+  private checkDomainInterval: number;
+  private sockets = new Set<ClosableSocket>();
+
   public readonly instanceInfo: InstanceConnectionInfo;
   public ephemeralCert?: SslCert;
   public host?: string;
@@ -98,6 +113,7 @@ export class CloudSQLInstance {
     this.ipType = options.ipType || IpAddressTypes.PUBLIC;
     this.limitRateInterval = options.limitRateInterval || 30 * 1000; // 30 seconds
     this.sqlAdminFetcher = options.sqlAdminFetcher;
+    this.checkDomainInterval = options.checkDomainInterval || 30 * 1000;
   }
 
   // p-throttle library has to be initialized in an async scope in order to
@@ -142,7 +158,7 @@ export class CloudSQLInstance {
           // Else resolve immediately.
           resolve();
         }
-      }, 0);
+      });
     });
   }
 
@@ -152,6 +168,14 @@ export class CloudSQLInstance {
       this.next = undefined;
       return Promise.reject('closed');
     }
+    if (this?.instanceInfo?.domainName && !this.checkDomainID) {
+      this.checkDomainID = setInterval(
+        () => {
+          this.checkDomainChanged();
+        },
+        this.checkDomainInterval || 30 * 1000
+      );
+    }
 
     const currentRefreshId = this.scheduledRefreshID;
 
@@ -296,8 +320,8 @@ export class CloudSQLInstance {
     // If refresh has not yet started, then cancel the setTimeout
     if (this.scheduledRefreshID) {
       clearTimeout(this.scheduledRefreshID);
+      this.scheduledRefreshID = null;
     }
-    this.scheduledRefreshID = null;
   }
 
   // Mark this instance as having an active connection. This is important to
@@ -312,9 +336,48 @@ export class CloudSQLInstance {
   close(): void {
     this.closed = true;
     this.cancelRefresh();
+    if (this.checkDomainID) {
+      clearInterval(this.checkDomainID);
+      this.checkDomainID = null;
+    }
+    for (const socket of this.sockets) {
+      socket.destroy(
+        new CloudSQLConnectorError({
+          code: 'ERRCLOSED',
+          message: 'The connector was closed.',
+        })
+      );
+    }
   }
 
   isClosed(): boolean {
     return this.closed;
   }
+  async checkDomainChanged() {
+    if (!this.instanceInfo.domainName) {
+      return;
+    }
+
+    const newInfo = await resolveInstanceName(
+      undefined,
+      this.instanceInfo.domainName
+    );
+    if (!isSameInstance(this.instanceInfo, newInfo)) {
+      // Domain name changed. Close and remove, then create a new map entry.
+      this.close();
+    }
+  }
+  addSocket(socket: ClosableSocket) {
+    if (!this.instanceInfo.domainName) {
+      // This was not connected by domain name. Ignore all sockets.
+      return;
+    }
+
+    // Add the socket to the list
+    this.sockets.add(socket);
+    // When the socket is closed, remove it.
+    socket.once('closed', () => {
+      this.sockets.delete(socket);
+    });
+  }
 }

src/connector.ts

@@ -44,6 +44,7 @@ export declare interface ConnectionOptions {
   ipType?: IpAddressTypes;
   instanceConnectionName: string;
   domainName?: string;
+  checkDomainInterval?: number;
   limitRateInterval?: number;
 }
 
@@ -125,6 +126,7 @@ class CloudSQLInstanceMap extends Map<string, CacheEntry> {
       } else {
         // Check the domain name, then if the instance is still open
         // return it
+        await entry.instance?.checkDomainChanged();
         if (!entry.instance?.isClosed()) {
           // The instance is open and the domain has not changed.
           // use the cached instance.
@@ -141,6 +143,7 @@ class CloudSQLInstanceMap extends Map<string, CacheEntry> {
       ipType: opts.ipType || IpAddressTypes.PUBLIC,
       limitRateInterval: opts.limitRateInterval || 30 * 1000, // 30 sec
       sqlAdminFetcher: this.sqlAdminFetcher,
+      checkDomainInterval: opts.checkDomainInterval,
     });
     this.set(this.cacheKey(opts), new CacheEntry(promise));
 
@@ -244,6 +247,9 @@ export class Connector {
           tlsSocket.once('secureConnect', async () => {
             cloudSqlInstance.setEstablishedConnection();
           });
+
+          cloudSqlInstance.addSocket(tlsSocket);
+
           return tlsSocket;
         }
 

test/cloud-sql-instance.ts

@@ -67,6 +67,7 @@ t.test('CloudSQLInstance', async t => {
       instanceConnectionName: 'my-project:us-east1:my-instance',
       sqlAdminFetcher: fetcher,
     });
+    t.after(() => instance.close());
 
     t.same(
       instance.ephemeralCert.cert,
@@ -115,6 +116,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
+    t.after(() => instance.close());
 
     await t.rejects(
       instance.refresh(),
@@ -135,6 +137,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
+    t.after(() => instance.close());
     instance.refresh = () => {
       if (refreshCount === 2) {
         const end = Date.now();
@@ -177,6 +180,7 @@ t.test('CloudSQLInstance', async t => {
           limitRateInterval: 50,
         },
       });
+      t.after(() => instance.close());
       await (() =>
         new Promise((res): void => {
           let refreshCount = 0;
@@ -233,6 +237,7 @@ t.test('CloudSQLInstance', async t => {
           limitRateInterval: 50,
         },
       });
+      t.after(() => instance.close());
       await (() =>
         new Promise((res): void => {
           let refreshCount = 0;
@@ -263,6 +268,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
+    t.after(() => instance.close());
 
     await instance.refresh();
 
@@ -301,6 +307,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
+    t.after(() => instance.close());
 
     let cancelRefreshCalled = false;
     let refreshCalled = false;
@@ -338,6 +345,7 @@ t.test('CloudSQLInstance', async t => {
         sqlAdminFetcher: fetcher,
       },
     });
+    t.after(() => instance.close());
 
     const start = Date.now();
     // starts regular refresh cycle
@@ -377,6 +385,7 @@ t.test('CloudSQLInstance', async t => {
         sqlAdminFetcher: fetcher,
       },
     });
+    t.after(() => instance.close());
     const start = Date.now();
     // starts out refresh logic
     let refreshCount = 1;
@@ -424,6 +433,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
+    t.after(() => instance.close());
 
     // starts a new refresh cycle but do not await on it
     instance.refresh();
@@ -451,6 +461,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
+    t.after(() => instance.close());
 
     // simulates an ongoing instance, already has data
     await instance.refresh();
@@ -487,6 +498,7 @@ t.test('CloudSQLInstance', async t => {
           limitRateInterval: 50,
         },
       });
+      t.after(() => instance.close());
 
       await instance.refresh();
       instance.setEstablishedConnection();
@@ -522,6 +534,7 @@ t.test('CloudSQLInstance', async t => {
           limitRateInterval: 50,
         },
       });
+      t.after(() => instance.close());
 
       await instance.refresh();
       instance.setEstablishedConnection();
@@ -589,6 +602,7 @@ t.test('CloudSQLInstance', async t => {
           limitRateInterval: 0,
         },
       });
+      t.after(() => instance.close());
       await (() =>
         new Promise((res): void => {
           let refreshCount = 0;