chore: support RPC consistency mechanism

.changeset/green-radios-taste.md

@@ -0,0 +1,5 @@
+---
+"@fuel-ts/account": patch
+---
+
+chore: support RPC consistency mechanism

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

@@ -9,21 +9,30 @@ type FuelGraphQLSubscriberOptions = {
   query: DocumentNode;
   variables?: Record<string, unknown>;
   fetchFn: typeof fetch;
+  operationName: string;
+  onEvent?: (event: FuelGraphqlSubscriberEvent) => void;
 };
 
+export interface FuelGraphqlSubscriberEvent {
+  data: unknown;
+  errors?: { message: string }[];
+  extensions?: Record<string, unknown>;
+}
+
 export class FuelGraphqlSubscriber implements AsyncIterator<unknown> {
   public static incompatibleNodeVersionMessage: string | false = false;
   private static textDecoder = new TextDecoder();
 
   private constructor(private stream: ReadableStreamDefaultReader<Uint8Array>) {}
 
   public static async create(options: FuelGraphQLSubscriberOptions) {
-    const { url, query, variables, fetchFn } = options;
+    const { url, query, variables, fetchFn, operationName, onEvent } = options;
     const response = await fetchFn(`${url}-sub`, {
       method: 'POST',
       body: JSON.stringify({
         query: print(query),
         variables,
+        operationName,
       }),
       headers: {
         'Content-Type': 'application/json',
@@ -35,7 +44,7 @@ export class FuelGraphqlSubscriber implements AsyncIterator<unknown> {
     const [backgroundStream, resultStream] = response.body!.tee();
 
     // eslint-disable-next-line no-void
-    void this.readInBackground(backgroundStream.getReader());
+    void this.readInBackground(backgroundStream.getReader(), onEvent);
 
     const [errorReader, resultReader] = resultStream.tee().map((stream) => stream.getReader());
 
@@ -49,6 +58,60 @@ export class FuelGraphqlSubscriber implements AsyncIterator<unknown> {
     return new FuelGraphqlSubscriber(resultReader);
   }
 
+  public static async readEvent(
+    reader: ReadableStreamDefaultReader<Uint8Array>,
+    parsingLeftover: string = ''
+  ): Promise<{
+    event: FuelGraphqlSubscriberEvent | undefined;
+    done: boolean;
+    parsingLeftover: string;
+  }> {
+    let text = parsingLeftover;
+    const regex = /data:.*\n\n/g;
+
+    // eslint-disable-next-line no-constant-condition
+    while (true) {
+      const matches = [...text.matchAll(regex)].flatMap((match) => match);
+
+      if (matches.length > 0) {
+        try {
+          const event = JSON.parse(matches[0].replace(/^data:/, ''));
+
+          return {
+            event,
+            done: false,
+            parsingLeftover: text.replace(matches[0], ''),
+          };
+        } catch (e) {
+          throw new FuelError(
+            ErrorCode.STREAM_PARSING_ERROR,
+            `Error while parsing stream data response: ${text}`
+          );
+        }
+      }
+
+      const { value, done } = await reader.read();
+
+      if (done) {
+        return { event: undefined, done, parsingLeftover: '' };
+      }
+
+      /**
+       * We don't care about keep-alive messages.
+       * The only responses that I came across from the node are either 200 responses with "data:.*" or keep-alive messages.
+       * You can find the keep-alive message in the fuel-core codebase (latest permalink as of writing):
+       * https://github.com/FuelLabs/fuel-core/blob/e1e631902f762081d2124d9c457ddfe13ac366dc/crates/fuel-core/src/graphql_api/service.rs#L247
+       * To get the actual latest info you need to check out the master branch (might change):
+       * https://github.com/FuelLabs/fuel-core/blob/master/crates/fuel-core/src/graphql_api/service.rs#L247
+       * */
+      const decoded = FuelGraphqlSubscriber.textDecoder
+        .decode(value)
+        .replace(':keep-alive-text\n\n', '');
+
+      text += decoded;
+    }
+  }
+
   /**
    * Reads the stream in the background,
    * thereby preventing the stream from not being read
@@ -57,17 +120,29 @@ export class FuelGraphqlSubscriber implements AsyncIterator<unknown> {
    * it is still available in the other streams
    * via internal mechanisms related to teeing.
    */
-  private static async readInBackground(reader: ReadableStreamDefaultReader<Uint8Array>) {
+  private static async readInBackground(
+    reader: ReadableStreamDefaultReader<Uint8Array>,
+    onEvent?: (event: FuelGraphqlSubscriberEvent) => void
+  ) {
+    let leftoverText = '';
     // eslint-disable-next-line no-constant-condition
     while (true) {
-      const { done } = await reader.read();
+      const { event, done, parsingLeftover } = await FuelGraphqlSubscriber.readEvent(
+        reader,
+        leftoverText
+      );
+
       if (done) {
         return;
       }
+
+      onEvent?.(event as FuelGraphqlSubscriberEvent);
+      leftoverText = parsingLeftover;
     }
   }
 
-  private events: Array<{ data: unknown; errors?: { message: string }[] }> = [];
+  private events: Array<FuelGraphqlSubscriberEvent> = [];
+
   private parsingLeftover = '';
 
   async next(): Promise<IteratorResult<unknown, unknown>> {
@@ -79,44 +154,18 @@ export class FuelGraphqlSubscriber implements AsyncIterator<unknown> {
         assertGqlResponseHasNoErrors(errors, FuelGraphqlSubscriber.incompatibleNodeVersionMessage);
         return { value: data, done: false };
       }
-      const { value, done } = await this.stream.read();
-      if (done) {
-        return { value, done };
-      }
 
-      /**
-       * We don't care about keep-alive messages.
-       * The only responses that I came across from the node are either 200 responses with "data:.*" or keep-alive messages.
-       * You can find the keep-alive message in the fuel-core codebase (latest permalink as of writing):
-       * https://github.com/FuelLabs/fuel-core/blob/e1e631902f762081d2124d9c457ddfe13ac366dc/crates/fuel-core/src/graphql_api/service.rs#L247
-       * To get the actual latest info you need to check out the master branch (might change):
-       * https://github.com/FuelLabs/fuel-core/blob/master/crates/fuel-core/src/graphql_api/service.rs#L247
-       * */
-      const decoded = FuelGraphqlSubscriber.textDecoder
-        .decode(value)
-        .replace(':keep-alive-text\n\n', '');
-
-      if (decoded === '') {
-        continue;
-      }
+      const { event, done, parsingLeftover } = await FuelGraphqlSubscriber.readEvent(
+        this.stream,
+        this.parsingLeftover
+      );
 
-      const text = `${this.parsingLeftover}${decoded}`;
-      const regex = /data:.*\n\n/g;
-
-      const matches = [...text.matchAll(regex)].flatMap((match) => match);
-
-      matches.forEach((match) => {
-        try {
-          this.events.push(JSON.parse(match.replace(/^data:/, '')));
-        } catch (e) {
-          throw new FuelError(
-            ErrorCode.STREAM_PARSING_ERROR,
-            `Error while parsing stream data response: ${text}`
-          );
-        }
-      });
+      this.parsingLeftover = parsingLeftover;
 
-      this.parsingLeftover = text.replace(matches.join(), '');
+      if (done) {
+        return { value: undefined, done: true };
+      }
+      this.events.push(event as FuelGraphqlSubscriberEvent);
     }
   }
 

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

@@ -933,6 +933,66 @@ describe('Provider', () => {
     expect(spyOperation).toHaveBeenCalledTimes(1);
   });
 
+  test('clearing cache based on URL only clears the cache for that URL', async () => {
+    using launched1 = await setupTestProviderAndWallets({
+      nodeOptions: {
+        args: ['--poa-instant', 'false', '--poa-interval-period', '10ms'],
+        loggingEnabled: false,
+      },
+    });
+    using launched2 = await setupTestProviderAndWallets({
+      nodeOptions: {
+        args: ['--poa-instant', 'false', '--poa-interval-period', '10ms'],
+        loggingEnabled: false,
+      },
+    });
+    const { provider: provider1 } = launched1;
+    const { provider: provider2 } = launched2;
+
+    // allow for block production
+    await sleep(200);
+
+    // update block height cache for both providers
+    await provider1.getLatestGasPrice();
+    await provider2.getLatestGasPrice();
+
+    Provider.clearChainAndNodeCaches(provider1.url);
+
+    // verify block height cache got reset correctly
+    const fetchSpy = vi.spyOn(global, 'fetch');
+
+    try {
+      await provider1.operations.submit({ encodedTransaction: '0x123' });
+    } catch (error) {
+      //
+    }
+    try {
+      await provider2.operations.submit({ encodedTransaction: '0x123' });
+    } catch (error) {
+      //
+    }
+
+    const {
+      extensions: { required_fuel_block_height: cache1BlockHeight },
+    } = JSON.parse(fetchSpy.mock.calls[0][1]?.body as string);
+    const {
+      extensions: { required_fuel_block_height: cache2BlockHeight },
+    } = JSON.parse(fetchSpy.mock.calls[1][1]?.body as string);
+
+    expect(cache1BlockHeight).toEqual(0);
+    expect(cache2BlockHeight).toBeGreaterThan(0);
+
+    // verify nodeInfo and chainInfo caches got reset correctly
+    const getChainAndNodeInfoSpy1 = vi.spyOn(provider1.operations, 'getChainAndNodeInfo');
+    const getChainAndNodeInfoSpy2 = vi.spyOn(provider2.operations, 'getChainAndNodeInfo');
+
+    await provider1.fetchChainAndNodeInfo();
+    await provider2.fetchChainAndNodeInfo();
+
+    expect(getChainAndNodeInfoSpy1).toHaveBeenCalledTimes(1);
+    expect(getChainAndNodeInfoSpy2).toHaveBeenCalledTimes(0);
+  });
+
   it('should ensure getGasConfig return essential gas related data', async () => {
     using launched = await setupTestProviderAndWallets();
     const { provider } = launched;