fix: ignored subscriptions not being read (#3695)

Author: nedsalk

.changeset/silent-points-punch.md

@@ -0,0 +1,5 @@
+---
+"@fuel-ts/account": patch
+---
+
+fix: ignored subscriptions not being read

packages/account/src/providers/fuel-graphql-subscriber.ts

@@ -32,7 +32,12 @@ export class FuelGraphqlSubscriber implements AsyncIterator<unknown> {
     });
 
     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-    const [errorReader, resultReader] = response.body!.tee().map((stream) => stream.getReader());
+    const [backgroundStream, resultStream] = response.body!.tee();
+
+    // eslint-disable-next-line no-void
+    void this.readInBackground(backgroundStream.getReader());
+
+    const [errorReader, resultReader] = resultStream.tee().map((stream) => stream.getReader());
 
     /**
      * If the node threw an error, read it and throw it to the user
@@ -44,6 +49,24 @@ export class FuelGraphqlSubscriber implements AsyncIterator<unknown> {
     return new FuelGraphqlSubscriber(resultReader);
   }
 
+  /**
+   * Reads the stream in the background,
+   * thereby preventing the stream from not being read
+   * if the user ignores the subscription.
+   * Even though the read data is ignored in this function,
+   * it is still available in the other streams
+   * via internal mechanisms related to teeing.
+   */
+  private static async readInBackground(reader: ReadableStreamDefaultReader<Uint8Array>) {
+    // eslint-disable-next-line no-constant-condition
+    while (true) {
+      const { done } = await reader.read();
+      if (done) {
+        return;
+      }
+    }
+  }
+
   private events: Array<{ data: unknown; errors?: { message: string }[] }> = [];
   private parsingLeftover = '';
 

packages/account/src/providers/provider.test.ts

@@ -9,6 +9,7 @@ import { InputType, OutputType, ReceiptType } from '@fuel-ts/transactions';
 import { DateTime, arrayify, hexlify, sleep } from '@fuel-ts/utils';
 import { ASSET_A, ASSET_B } from '@fuel-ts/utils/test-utils';
 import { versions } from '@fuel-ts/versions';
+import type { MockInstance } from 'vitest';
 
 import { Wallet } from '..';
 import {
@@ -1753,6 +1754,53 @@ describe('Provider', () => {
     expect(numberOfEvents).toEqual(2);
   });
 
+  it('subscriptions: streams are consumed even if the async iterator is not', async () => {
+    using launched = await setupTestProviderAndWallets();
+    const { provider } = launched;
+
+    const sseResponse = new TextEncoder().encode(`data:{"field":"not-relevant"}\n\n`);
+
+    let pullCallNum = 0;
+
+    const underlyingSource: UnderlyingDefaultSource = {
+      pull: (controller) => {
+        pullCallNum += 1;
+        controller.enqueue(sseResponse);
+        if (pullCallNum === 20) {
+          controller.close();
+        }
+      },
+    };
+
+    const pullSpy: MockInstance = vi.spyOn(underlyingSource, 'pull');
+
+    vi.spyOn(global, 'fetch').mockImplementationOnce(() =>
+      Promise.resolve(
+        new Response(
+          new ReadableStream(
+            underlyingSource,
+            /**
+             * Only pull when .read() is called.
+             * Don't do any behind-the-scenes buffering
+             * so that we can test that the sdk itself is pulling
+             * even if the user isn't reading.
+             */
+            { highWaterMark: 0 }
+          )
+        )
+      )
+    );
+
+    await provider.operations.submitAndAwaitStatus({
+      encodedTransaction: "it's mocked so doesn't matter",
+    });
+
+    // give time for the pulls to be called in the background
+    await sleep(500);
+
+    expect(pullSpy).toHaveBeenCalledTimes(20);
+  });
+
   it('subscriptions: throws if the stream data string parsing fails for some reason', async () => {
     using launched = await setupTestProviderAndWallets();
     const { provider } = launched;