feat(data-connect): add de-duplication, resume, and intelligent reconnection with backoff (#9905)

* chore(data-connect): update documentation and names in stream transport (#9846)

* rename in preparation for future changes

* finish renaming + updating comments

* minor typo fix

* feat(data-connect): add query de-duplication to stream transport (#9850)

* rename in preparation for future changes

* finish renaming + updating comments

* minor typo fix

* add de-duping - now for tests

* add de-duplication tests for invokeQuery

* finish invoke operation dedupe tests

* add subscribe de-duping as well

* add more query dedupe tests

* chore(data-connect): clean up streaming tests a bit (#9851)

* rename in preparation for future changes

* finish renaming + updating comments

* minor typo fix

* add de-duping - now for tests

* add de-duplication tests for invokeQuery

* finish invoke operation dedupe tests

* add subscribe de-duping as well

* add more query dedupe tests

* clean up stream tests

* clean up promise types in streaming tests

* Pasta/bucatini resume refactor (#9885)

* add test changes

* add transportchanges

* finalize changes, run yarn format

* finalize changes based on reviewer feedback

* feat(data-connect): add prefer resume over execute optimization to stream transport (#9879)

* initial implementation - now for testing

* finish tests

* merge pasta/bucatini-resume-refactor

* address reviewer comments

* remove old comments

* address reviewer comments

* feat(data-connect): add intelligent reconnection to streaming transport and harden disconnect/cleanup functionality (#9893)

* WIP

* first pass

* fixing names

* tidying up, fixing tests

* fixed tests

* address reviewer comments

* address reviewer comments and harden close/cleanup lifecycle

* address reviewer comments

* make listener registering and unregistering more robust

* merge main, add changeset

.changeset/nervous-carpets-approve.md

@@ -0,0 +1,6 @@
+---
+'@firebase/data-connect': minor
+'firebase': minor
+---
+
+Hardened the Firebase SQL Connect streaming transport with intelligent reconnection, query de-duplication, and resume optimizations.

packages/data-connect/src/network/manager.ts

@@ -86,7 +86,7 @@ export class DataConnectTransportManager
           this.transportOptions.sslEnabled
         );
       }
-      this.streamTransport.onGracefulStreamClose = () => {
+      this.streamTransport.onCloseCallback = () => {
         this.streamTransport = undefined;
       };
     }

packages/data-connect/src/network/stream/websocket.ts

@@ -15,17 +15,10 @@
  * limitations under the License.
  */
 
-import { DataConnectOptions, TransportOptions } from '../../api/DataConnect';
-import { AppCheckTokenProvider } from '../../core/AppCheckTokenProvider';
 import { Code, DataConnectError } from '../../core/error';
-import { AuthTokenProvider } from '../../core/FirebaseAuthProvider';
 import { logError } from '../../logger';
 import { websocketUrlBuilder } from '../../util/url';
-import {
-  CallerSdkType,
-  CallerSdkTypeEnum,
-  DataConnectResponse
-} from '../transport';
+import { DataConnectResponse } from '../transport';
 
 import { AbstractDataConnectStreamTransport } from './streamTransport';
 import { DataConnectStreamRequest, DataConnectStreamResponse } from './wire';
@@ -99,28 +92,6 @@ export class WebSocketTransport extends AbstractDataConnectStreamTransport {
    */
   private connectionAttempt: Promise<void> | null = null;
 
-  constructor(
-    options: DataConnectOptions,
-    protected apiKey?: string | undefined,
-    protected appId?: string | null,
-    protected authProvider?: AuthTokenProvider | undefined,
-    protected appCheckProvider?: AppCheckTokenProvider | undefined,
-    transportOptions?: TransportOptions | undefined,
-    protected _isUsingGen = false,
-    protected _callerSdkType: CallerSdkType = CallerSdkTypeEnum.Base
-  ) {
-    super(
-      options,
-      apiKey,
-      appId,
-      authProvider,
-      appCheckProvider,
-      transportOptions,
-      _isUsingGen,
-      _callerSdkType
-    );
-  }
-
   protected ensureConnection(): Promise<void> {
     try {
       if (this.streamIsReady) {
@@ -336,6 +307,6 @@ export class WebSocketTransport extends AbstractDataConnectStreamTransport {
         'WebSocket message did not include requestId'
       );
     }
-    return result as DataConnectStreamResponse<Data>;
+    return result;
   }
 }

packages/data-connect/test/unit/transportManager.test.ts

@@ -38,6 +38,7 @@ use(sinonChai);
 /** Interface that exposes private fields of stream transport for testing purposes. */
 interface StreamTransportWithInternals {
   onStreamClose(code: number, reason: string): void;
+  rejectAllRequests(code: string, reason: string): void;
 }
 
 /** Interface that exposes private fields of TransportManager for testing purposes. */
@@ -504,9 +505,13 @@ describe('DataConnectTransportManager', () => {
       };
       transportWithInternals.subscribeObservers = new Map([['1', observer]]);
 
+      // rejectAllRequests simulates a disconnect without reconnects!
       (
         streamTransport as unknown as StreamTransportWithInternals
-      ).onStreamClose(1006, 'Abnormal Closure');
+      ).rejectAllRequests(
+        Code.OTHER,
+        'Stream disconnected with code 1006: Abnormal Closure'
+      );
 
       expect(observer.onDisconnect).to.have.been.calledOnceWith(
         Code.OTHER,

packages/data-connect/test/unit/websocketTransport.test.ts

@@ -48,7 +48,7 @@ interface TransportWithInternals {
     requestId: string,
     response: DataConnectResponse<Data>
   ): Promise<void>;
-  rejectAllActiveRequests(code: Code, reason: string): void;
+  rejectAllRequests(code: Code, reason: string): void;
 }
 
 describe('WebSocketTransport', () => {
@@ -181,19 +181,19 @@ describe('WebSocketTransport', () => {
       expect(transport.connection).to.be.undefined;
     });
 
-    it('should call rejectAllActiveRequests and clean up when closed externally', async () => {
+    it('should call rejectAllRequests and clean up when closed externally', async () => {
       const openPromise = transport.openConnection();
       await transport.connection!.simulateOpen();
       await openPromise;
 
-      const rejectSpy = sinon.spy(transport, 'rejectAllActiveRequests');
+      const rejectSpy = sinon.spy(transport, 'rejectAllRequests');
 
       await transport.connection!.simulateClose();
 
       expect(rejectSpy).to.have.been.calledOnce;
       expect(rejectSpy.firstCall.args[0]).to.equal(Code.OTHER);
       expect(rejectSpy.firstCall.args[1]).to.equal(
-        `Stream disconnected with code ${WEBSOCKET_CLOSE_CODE}: Normal Closure`
+        `Stream disconnected while idle with code ${WEBSOCKET_CLOSE_CODE}: Normal Closure`
       );
     });
   });