wip: close sockets on instance closed

Author: hessjcg

src/cloud-sql-instance.ts

@@ -24,6 +24,16 @@ 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({
@@ -45,7 +55,7 @@ interface CloudSQLInstanceOptions {
   ipType: IpAddressTypes;
   limitRateInterval?: number;
   sqlAdminFetcher: Fetcher;
-  checkDomainInterval?: number
+  checkDomainInterval?: number;
 }
 
 interface RefreshResult {
@@ -78,11 +88,12 @@ 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 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 checkDomainInterval: number;
+  private sockets = new Set<ClosableSocket>();
 
   public readonly instanceInfo: InstanceConnectionInfo;
   public ephemeralCert?: SslCert;
@@ -105,7 +116,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;
+    this.checkDomainInterval = options.checkDomainInterval || 30 * 1000;
   }
 
   // p-throttle library has to be initialized in an async scope in order to
@@ -137,9 +148,9 @@ export class CloudSQLInstance {
   // in progress, the promise will resolve immediately.
   refreshComplete(): Promise<void> {
     return new Promise(resolve => {
-      // setTimeout() to yield execution to allow other refresh background
+      // setImmediate() to yield execution to allow other refresh background
       // tasks to start.
-      setTimeout(() => {
+      setImmediate(() => {
         if (this.next) {
           // If there is a refresh promise in progress, resolve this promise
           // when the refresh is complete.
@@ -148,7 +159,7 @@ export class CloudSQLInstance {
           // Else resolve immediately.
           resolve();
         }
-      }, 0);
+      });
     });
   }
 
@@ -158,10 +169,13 @@ 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)
+    if (this?.instanceInfo?.domainName && !this.checkDomainID) {
+      this.checkDomainID = setInterval(
+        () => {
+          this.checkDomainChanged();
+        },
+        this.checkDomainInterval || 30 * 1000
+      );
     }
 
     const currentRefreshId = this.scheduledRefreshID;
@@ -307,7 +321,7 @@ 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;
     }
   }
 
@@ -323,9 +337,17 @@ export class CloudSQLInstance {
   close(): void {
     this.closed = true;
     this.cancelRefresh();
-    if(this.checkDomainID){
+    if (this.checkDomainID) {
       clearInterval(this.checkDomainID);
-      this.checkDomainID = null
+      this.checkDomainID = null;
+    }
+    for (const socket of this.sockets) {
+      socket.destroy(
+        new CloudSQLConnectorError({
+          code: 'ERRCLOSED',
+          message: 'The connector was closed.',
+        })
+      );
     }
   }
 
@@ -346,4 +368,17 @@ export class CloudSQLInstance {
       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,8 +44,8 @@ export declare interface ConnectionOptions {
   ipType?: IpAddressTypes;
   instanceConnectionName: string;
   domainName?: string;
-  checkDomainInterval?: number
-  limitRateInterval?: number
+  checkDomainInterval?: number;
+  limitRateInterval?: number;
 }
 
 export declare interface SocketConnectionOptions extends ConnectionOptions {
@@ -143,7 +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
+      checkDomainInterval: opts.checkDomainInterval,
     });
     this.set(this.cacheKey(opts), new CacheEntry(promise));
 
@@ -248,6 +248,9 @@ export class Connector {
           tlsSocket.once('secureConnect', async () => {
             cloudSqlInstance.setEstablishedConnection();
           });
+
+          cloudSqlInstance.addSocket(tlsSocket);
+
           return tlsSocket;
         }
 

test/cloud-sql-instance.ts

@@ -67,7 +67,7 @@ t.test('CloudSQLInstance', async t => {
       instanceConnectionName: 'my-project:us-east1:my-instance',
       sqlAdminFetcher: fetcher,
     });
-    t.after(() =>instance.close())
+    t.after(() => instance.close());
 
     t.same(
       instance.ephemeralCert.cert,
@@ -116,7 +116,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
-    t.after(() =>instance.close())
+    t.after(() => instance.close());
 
     await t.rejects(
       instance.refresh(),
@@ -137,7 +137,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
-    t.after(() =>instance.close())
+    t.after(() => instance.close());
     instance.refresh = () => {
       if (refreshCount === 2) {
         const end = Date.now();
@@ -180,7 +180,7 @@ t.test('CloudSQLInstance', async t => {
           limitRateInterval: 50,
         },
       });
-      t.after(() =>instance.close())
+      t.after(() => instance.close());
       await (() =>
         new Promise((res): void => {
           let refreshCount = 0;
@@ -237,7 +237,7 @@ t.test('CloudSQLInstance', async t => {
           limitRateInterval: 50,
         },
       });
-      t.after(() =>instance.close())
+      t.after(() => instance.close());
       await (() =>
         new Promise((res): void => {
           let refreshCount = 0;
@@ -268,7 +268,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
-    t.after(() =>instance.close())
+    t.after(() => instance.close());
 
     await instance.refresh();
 
@@ -308,7 +308,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
-    t.after(() =>instance.close())
+    t.after(() => instance.close());
 
     let cancelRefreshCalled = false;
     let refreshCalled = false;
@@ -347,7 +347,7 @@ t.test('CloudSQLInstance', async t => {
         sqlAdminFetcher: fetcher,
       },
     });
-    t.after(() =>instance.close())
+    t.after(() => instance.close());
 
     const start = Date.now();
     // starts regular refresh cycle
@@ -387,7 +387,7 @@ t.test('CloudSQLInstance', async t => {
         sqlAdminFetcher: fetcher,
       },
     });
-    t.after(()=> instance.close())
+    t.after(() => instance.close());
     const start = Date.now();
     // starts out refresh logic
     let refreshCount = 1;
@@ -435,7 +435,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
-    t.after(() =>instance.close())
+    t.after(() => instance.close());
 
     // starts a new refresh cycle but do not await on it
     instance.refresh();
@@ -463,7 +463,7 @@ t.test('CloudSQLInstance', async t => {
         limitRateInterval: 50,
       },
     });
-    t.after(() =>instance.close())
+    t.after(() => instance.close());
 
     // simulates an ongoing instance, already has data
     await instance.refresh();
@@ -500,7 +500,7 @@ t.test('CloudSQLInstance', async t => {
           limitRateInterval: 50,
         },
       });
-      t.after(() =>instance.close())
+      t.after(() => instance.close());
 
       await instance.refresh();
       instance.setEstablishedConnection();
@@ -536,7 +536,7 @@ t.test('CloudSQLInstance', async t => {
           limitRateInterval: 50,
         },
       });
-      t.after(() =>instance.close())
+      t.after(() => instance.close());
 
       await instance.refresh();
       instance.setEstablishedConnection();
@@ -605,7 +605,7 @@ t.test('CloudSQLInstance', async t => {
           limitRateInterval: 0,
         },
       });
-      t.after(() =>instance.close())
+      t.after(() => instance.close());
       await (() =>
         new Promise((res): void => {
           let refreshCount = 0;
@@ -629,5 +629,6 @@ t.test('CloudSQLInstance', async t => {
           instance.refresh();
           instance.setEstablishedConnection();
         }))();
-    });
+    }
+  );
 });

test/connector.ts

@@ -639,7 +639,6 @@ t.test('Connector by domain resolves and creates instance', async t => {
   connector.close();
 });
 
-
 t.test('Connector by domain resolves and creates instance', async t => {
   const th = setupConnectorModule(t);
   const connector = new th.Connector();
@@ -681,38 +680,47 @@ t.test('Connector by domain resolves and creates instance', async t => {
   connector.close();
 });
 
+t.test(
+  'Connector checks if name changes in background and closes connector',
+  async t => {
+    const th = setupConnectorModule(t);
+    const connector = new th.Connector();
+    t.after(() => {
+      connector.close();
+    });
 
-t.test('Connector checks if name changes in background and closes connector', async t => {
-  const th = setupConnectorModule(t);
-  const connector = new th.Connector();
-  t.after(() => {
-    connector.close();
-  });
-
-  // Get options loads the instance
-  await connector.getOptions({
-    ipType: 'PUBLIC',
-    authType: 'PASSWORD',
-    domainName: 'db.example.com',
-    checkDomainInterval: 10, // 10ms for testing
-  });
-
-  // Ensure there is only one entry.
-  t.same(connector.instances.size, 1);
-  const oldInstance = connector.instances.get(
-    'db.example.com-PASSWORD-PUBLIC'
-  ).instance;
-  t.same(oldInstance.instanceInfo.domainName, 'db.example.com');
-  t.same(oldInstance.instanceInfo.instanceId, 'instance');
-
-  // getOptions after DNS response changes closes the old instance
-  // and loads a new one.
-  th.resolveTxtResponse = 'project:region2:instance2';
-  await new Promise((res) =>{
-    setTimeout(res, 50);
-  })
+    // Get options loads the instance
+    await connector.getOptions({
+      ipType: 'PUBLIC',
+      authType: 'PASSWORD',
+      domainName: 'db.example.com',
+      checkDomainInterval: 10, // 10ms for testing
+    });
 
-  t.same(oldInstance.isClosed(), true, 'old instance is closed');
+    // Ensure there is only one entry.
+    t.same(connector.instances.size, 1);
+    const oldInstance = connector.instances.get(
+      'db.example.com-PASSWORD-PUBLIC'
+    ).instance;
+    t.same(oldInstance.instanceInfo.domainName, 'db.example.com');
+    t.same(oldInstance.instanceInfo.instanceId, 'instance');
+
+    // add a mock socket to the old instance
+    const mockSocket = {
+      destroyed: false,
+      once(e:string, fn:()=>void) {},
+      destroy(){this.destroyed = true}
+    }
+    oldInstance.addSocket(mockSocket);
+
+    // getOptions after DNS response changes closes the old instance
+    // and loads a new one.
+    th.resolveTxtResponse = 'project:region2:instance2';
+    await new Promise(res => {
+      setTimeout(res, 50);
+    });
 
-  connector.close();
-});
+    t.same(oldInstance.isClosed(), true, 'old instance is closed');
+    t.same(mockSocket.destroyed, true, 'old instance closed its sockets');
+  }
+);