Repository: urql-graphql/urql (ts)

Issue:
RFC: Retry Exchange - Reset Delay
<!--
  🚨 RFCs are for proposed changes (not bugs or questions)
  Specifically they are whenever you'd like to see new features
  being added to urql, or enable new use-cases.

  Please open a Bug Report for issues/bugs, and use GitHub Discussions
  or the Discord channel for questions instead.
-->

## Summary

<!--
  Describe in a couple of words *what* you're proposing.
  If relevant, include *why* this should be addressed now.
  The problem should be clearly stated and the solution
  should be summarised.
-->

I believe it is not currently possible to reset the retry delay. For queries and mutations this makes sense, but when I have a subscription that I want to always be available, I would like to always retry it (`retryIf => true` for my retry exchange). 

When there are network errors or for some reason my subscription breaks, I use a map exchange to check if its due to my JWT expiring (I'm using Hasura and it forcefully closes the websocket connection when the access token expires) - if it is, I trigger a refresh to get a new access token, and proceed to let the retry exchange attempt to retry the subscription. This works fine, however, after a few token expiries, I always end up hitting the `maxDelayMs`, which I'd like to be able to reset under certain conditions (for example, on a token refresh), otherwise I will always end up with a large gap in time where I'm not receiving my precious subscriptions.

## Proposed Solution

<!--
  Explain the solution you're proposing in detail.
  *How* will this change be implemented, and how does it work?
-->

If it is already possible, it would be nice if someone explained to me how to achieve this =)

Right now, it seems the retries simply accumulate the `delayAmount * a random backoff factor` until it reaches the `maxDelayMs`. If this `delayAmount` could be accessed from the operation context or something, I could modify it in other exchanges to reset it to the initial value.

Edit: I literally just noticed `context.retryDelay` in the source code. Sorry, I'm going to try messing with that value now, but anyway, this might be something useful to add into the docs for people with the same problem as me (which I believe should be common for subscriptions).

Edit Edit: Changing `operation.context.retryDelay` works but doing so asynchronously might cause some issues (it seems retryExchange makes a new operation and copies over the context values via spread operator, so the reference to it might get lost in some cases).

Pull request:
fix(retry): Reset delay/count state if an operation succeeds
Resolves #3220

## Summary

This resets operations’ `delay` and `count` states if they succeed. This may be important to recover from failing subscriptions, since a new result should be able to restore the retry state to where it was before the subscription failed.

## Set of changes

- Add resetting object state to `retryExchange`
- Add test case


Patch:
diff --git a/.changeset/breezy-melons-deny.md b/.changeset/breezy-melons-deny.md
new file mode 100644
index 00000000..cc2f9dbb
--- /dev/null
+++ b/.changeset/breezy-melons-deny.md
@@ -0,0 +1,5 @@
+---
+'@urql/core': patch
+---
+
+Respect `additionalTypenames` on subscriptions and re-execute queries for them as well, as one would intuitively expect.
diff --git a/.changeset/curvy-poems-compare.md b/.changeset/curvy-poems-compare.md
new file mode 100644
index 00000000..8faaf8db
--- /dev/null
+++ b/.changeset/curvy-poems-compare.md
@@ -0,0 +1,5 @@
+---
+'@urql/exchange-retry': minor
+---
+
+Reset `retryExchange`’s previous attempts and delay if an operation succeeds. This prevents the exchange from keeping its old retry count and delay if the operation delivered a result in the meantime. This is important for it to help recover from failing subscriptions.
diff --git a/.changeset/hungry-cars-hide.md b/.changeset/hungry-cars-hide.md
new file mode 100644
index 00000000..abe1679a
--- /dev/null
+++ b/.changeset/hungry-cars-hide.md
@@ -0,0 +1,5 @@
+---
+'@urql/vue': patch
+---
+
+Prevent multiple operations being executed in a row when multiple inputs change simultaneously (e.g. `isPaused` and query inputs)
diff --git a/exchanges/retry/src/retryExchange.ts b/exchanges/retry/src/retryExchange.ts
index 9a0d3936..aafff0fe 100644
--- a/exchanges/retry/src/retryExchange.ts
+++ b/exchanges/retry/src/retryExchange.ts
@@ -78,6 +78,11 @@ export interface RetryExchangeOptions {
   ): Operation | null | undefined;
 }
 
+interface RetryState {
+  count: number;
+  delay: number | null;
+}
+
 /** Exchange factory that retries failed operations.
  *
  * @param options - A {@link RetriesExchangeOptions} configuration object.
@@ -117,10 +122,14 @@ export const retryExchange = (options: RetryExchangeOptions): Exchange => {
 
       const retryWithBackoff$ = pipe(
         retry$,
-        mergeMap((op: Operation) => {
-          const { key, context } = op;
-          const retryCount = (context.retryCount || 0) + 1;
-          let delayAmount = context.retryDelay || MIN_DELAY;
+        mergeMap((operation: Operation) => {
+          const retry: RetryState = operation.context.retry || {
+            count: 0,
+            delay: null,
+          };
+
+          const retryCount = ++retry.count;
+          let delayAmount = retry.delay || MIN_DELAY;
 
           const backoffFactor = Math.random() + 1.5;
           // if randomDelay is enabled and it won't exceed the max delay, apply a random
@@ -137,7 +146,7 @@ export const retryExchange = (options: RetryExchangeOptions): Exchange => {
             filter(op => {
               return (
                 (op.kind === 'query' || op.kind === 'teardown') &&
-                op.key === key
+                op.key === operation.key
               );
             })
           );
@@ -145,7 +154,7 @@ export const retryExchange = (options: RetryExchangeOptions): Exchange => {
           dispatchDebug({
             type: 'retryAttempt',
             message: `The operation has failed and a retry has been triggered (${retryCount} / ${MAX_ATTEMPTS})`,
-            operation: op,
+            operation,
             data: {
               retryCount,
             },
@@ -154,10 +163,9 @@ export const retryExchange = (options: RetryExchangeOptions): Exchange => {
           // Add new retryDelay and retryCount to operation
           return pipe(
             fromValue(
-              makeOperation(op.kind, op, {
-                ...op.context,
-                retryDelay: delayAmount,
-                retryCount,
+              makeOperation(operation.kind, operation, {
+                ...operation.context,
+                retry,
               })
             ),
             debounce(() => delayAmount),
@@ -171,6 +179,7 @@ export const retryExchange = (options: RetryExchangeOptions): Exchange => {
         merge([operations$, retryWithBackoff$]),
         forward,
         filter(res => {
+          const retry = res.operation.context.retry as RetryState | undefined;
           // Only retry if the error passes the conditional retryIf function (if passed)
           // or if the error contains a networkError
           if (
@@ -179,12 +188,16 @@ export const retryExchange = (options: RetryExchangeOptions): Exchange => {
               ? !retryIf(res.error, res.operation)
               : !retryWith && !res.error.networkError)
           ) {
+            // Reset the delay state for a successful operation
+            if (retry) {
+              retry.count = 0;
+              retry.delay = null;
+            }
             return true;
           }
 
           const maxNumberAttemptsExceeded =
-            (res.operation.context.retryCount || 0) >= MAX_ATTEMPTS - 1;
-
+            ((retry && retry.count) || 0) >= MAX_ATTEMPTS - 1;
           if (!maxNumberAttemptsExceeded) {
             const operation = retryWith
               ? retryWith(res.error, res.operation)
diff --git a/packages/core/src/exchanges/cache.ts b/packages/core/src/exchanges/cache.ts
index 64e711e3..87873fbc 100755
--- a/packages/core/src/exchanges/cache.ts
+++ b/packages/core/src/exchanges/cache.ts
@@ -114,12 +114,21 @@ export const cacheExchange: Exchange = ({ forward, client, dispatchDebug }) => {
         let { operation } = response;
         if (!operation) return;
 
-        const typenames = collectTypesFromResponse(response.data).concat(
-          operation.context.additionalTypenames || []
-        );
+        let typenames = operation.context.additionalTypenames || [];
+        // NOTE: For now, we only respect `additionalTypenames` from subscriptions to
+        // avoid unexpected breaking changes
+        // We'd expect live queries or other update mechanisms to be more suitable rather
+        // than using subscriptions as “signals” to reexecute queries. However, if they’re
+        // just used as signals, it’s intuitive to hook them up using `additionalTypenames`
+        if (response.operation.kind !== 'subscription') {
+          typenames = collectTypesFromResponse(response.data).concat(typenames);
+        }
 
         // Invalidates the cache given a mutation's response
-        if (response.operation.kind === 'mutation') {
+        if (
+          response.operation.kind === 'mutation' ||
+          response.operation.kind === 'subscription'
+        ) {
           const pendingOperations = new Set<number>();
 
           dispatchDebug({
diff --git a/packages/vue-urql/src/useQuery.ts b/packages/vue-urql/src/useQuery.ts
index 62339442..860653be 100644
--- a/packages/vue-urql/src/useQuery.ts
+++ b/packages/vue-urql/src/useQuery.ts
@@ -1,6 +1,14 @@
 /* eslint-disable react-hooks/rules-of-hooks */
 
-import { WatchStopHandle, Ref, ref, watchEffect, reactive, isRef } from 'vue';
+import {
+  WatchStopHandle,
+  Ref,
+  shallowRef,
+  ref,
+  watchEffect,
+  reactive,
+  isRef,
+} from 'vue';
 
 import { Subscription, Source, pipe, subscribe, onEnd } from 'wonka';
 
@@ -14,11 +22,10 @@ import {
   RequestPolicy,
   Operation,
   createRequest,
-  GraphQLRequest,
 } from '@urql/core';
 
 import { useClient } from './useClient';
-import { unwrapPossibleProxy } from './utils';
+import { unwrapPossibleProxy, updateShallowRef } from './utils';
 
 type MaybeRef<T> = T | Ref<T>;
 type MaybeRefObj<T extends {}> = { [K in keyof T]: MaybeRef<T[K]> };
@@ -258,31 +265,34 @@ export function callUseQuery<T = any, V extends AnyVariables = AnyVariables>(
     ? _args.pause
     : ref(!!_args.pause);
 
-  const request: Ref<GraphQLRequest<T, V>> = ref(
-    createRequest<T, V>(
+  const input = shallowRef({
+    request: createRequest<T, V>(
       unwrapPossibleProxy(args.query as any),
       unwrapPossibleProxy<V>(args.variables as V)
-    ) as any
-  );
+    ),
+    requestPolicy: unwrapPossibleProxy(args.requestPolicy),
+    isPaused: isPaused.value,
+  });
 
   const source: Ref<Source<OperationResult<T, V>> | undefined> = ref();
 
   stops.push(
     watchEffect(() => {
-      const newRequest = createRequest<T, V>(
-        unwrapPossibleProxy(args.query as any),
-        unwrapPossibleProxy<V>(args.variables as V)
-      );
-      if (request.value.key !== newRequest.key) {
-        request.value = newRequest;
-      }
+      updateShallowRef(input, {
+        request: createRequest<T, V>(
+          unwrapPossibleProxy(args.query as any),
+          unwrapPossibleProxy<V>(args.variables as V)
+        ),
+        requestPolicy: unwrapPossibleProxy(args.requestPolicy),
+        isPaused: isPaused.value,
+      });
     }, watchOptions)
   );
 
   stops.push(
     watchEffect(() => {
-      source.value = !isPaused.value
-        ? client.value.executeQuery<T, V>(request.value, {
+      source.value = !input.value.isPaused
+        ? client.value.executeQuery<T, V>(input.value.request, {
             requestPolicy: unwrapPossibleProxy(
               args.requestPolicy
             ) as RequestPolicy,
@@ -301,11 +311,16 @@ export function callUseQuery<T = any, V extends AnyVariables = AnyVariables>(
     fetching,
     isPaused,
     executeQuery(opts?: Partial<OperationContext>): UseQueryResponse<T, V> {
-      const s = (source.value = client.value.executeQuery<T, V>(request.value, {
-        requestPolicy: unwrapPossibleProxy(args.requestPolicy) as RequestPolicy,
-        ...args.context,
-        ...opts,
-      }));
+      const s = (source.value = client.value.executeQuery<T, V>(
+        input.value.request,
+        {
+          requestPolicy: unwrapPossibleProxy(
+            args.requestPolicy
+          ) as RequestPolicy,
+          ...args.context,
+          ...opts,
+        }
+      ));
 
       return {
         ...response,
diff --git a/packages/vue-urql/src/useSubscription.ts b/packages/vue-urql/src/useSubscription.ts
index 33078eb3..f353231d 100644
--- a/packages/vue-urql/src/useSubscription.ts
+++ b/packages/vue-urql/src/useSubscription.ts
@@ -2,7 +2,15 @@
 
 import { Source, pipe, subscribe, onEnd } from 'wonka';
 
-import { WatchStopHandle, Ref, ref, watchEffect, reactive, isRef } from 'vue';
+import {
+  WatchStopHandle,
+  Ref,
+  ref,
+  shallowRef,
+  watchEffect,
+  reactive,
+  isRef,
+} from 'vue';
 
 import {
   Client,
@@ -13,11 +21,10 @@ import {
   OperationContext,
   Operation,
   createRequest,
-  GraphQLRequest,
 } from '@urql/core';
 
 import { useClient } from './useClient';
-import { unwrapPossibleProxy } from './utils';
+import { unwrapPossibleProxy, updateShallowRef } from './utils';
 
 type MaybeRef<T> = Exclude<T, void> | Ref<Exclude<T, void>>;
 type MaybeRefObj<T extends {}> = { [K in keyof T]: MaybeRef<T[K]> };
@@ -259,31 +266,32 @@ export function callUseSubscription<
     ? _args.pause
     : ref(!!_args.pause);
 
-  const request: Ref<GraphQLRequest<T, V>> = ref(
-    createRequest<T, V>(
+  const input = shallowRef({
+    request: createRequest<T, V>(
       unwrapPossibleProxy(args.query as any),
       unwrapPossibleProxy<V>(args.variables as V)
-    ) as any
-  );
+    ),
+    isPaused: isPaused.value,
+  });
 
   const source: Ref<Source<OperationResult<T, V>> | undefined> = ref();
 
   stops.push(
     watchEffect(() => {
-      const newRequest = createRequest<T, V>(
-        unwrapPossibleProxy(args.query as any),
-        unwrapPossibleProxy<V>(args.variables as V)
-      );
-      if (request.value.key !== newRequest.key) {
-        request.value = newRequest;
-      }
+      updateShallowRef(input, {
+        request: createRequest<T, V>(
+          unwrapPossibleProxy(args.query as any),
+          unwrapPossibleProxy<V>(args.variables as V)
+        ),
+        isPaused: isPaused.value,
+      });
     }, watchOptions)
   );
 
   stops.push(
     watchEffect(() => {
       source.value = !isPaused.value
-        ? client.value.executeSubscription<T, V>(request.value, {
+        ? client.value.executeSubscription<T, V>(input.value.request, {
             ...(unwrapPossibleProxy(args.context) as Partial<OperationContext>),
           })
         : undefined;
@@ -333,10 +341,13 @@ export function callUseSubscription<
     executeSubscription(
       opts?: Partial<OperationContext>
     ): UseSubscriptionResponse<T, R, V> {
-      source.value = client.value.executeSubscription<T, V>(request.value, {
-        ...unwrapPossibleProxy(args.context),
-        ...opts,
-      });
+      source.value = client.value.executeSubscription<T, V>(
+        input.value.request,
+        {
+          ...unwrapPossibleProxy(args.context),
+          ...opts,
+        }
+      );
 
       return state;
     },
diff --git a/packages/vue-urql/src/utils.ts b/packages/vue-urql/src/utils.ts
index 161d2d3a..0e3ad4e4 100644
--- a/packages/vue-urql/src/utils.ts
+++ b/packages/vue-urql/src/utils.ts
@@ -1,7 +1,38 @@
-import { Ref, isRef } from 'vue';
+import { GraphQLRequest, AnyVariables } from '@urql/core';
+import { Ref, ShallowRef, isRef } from 'vue';
 
 export function unwrapPossibleProxy<V>(possibleProxy: V | Ref<V>): V {
   return possibleProxy && isRef(possibleProxy)
     ? possibleProxy.value
     : possibleProxy;
 }
+
+export interface RequestState<
+  Data = any,
+  Variables extends AnyVariables = AnyVariables
+> {
+  request: GraphQLRequest<Data, Variables>;
+  isPaused: boolean;
+}
+
+export function createRequestState<
+  Data = any,
+  Variables extends AnyVariables = AnyVariables
+>(
+  request: GraphQLRequest<Data, Variables>,
+  isPaused: boolean
+): RequestState<Data, Variables> {
+  return { request, isPaused };
+}
+
+export const updateShallowRef = <T extends Record<string, any>>(
+  ref: ShallowRef<T>,
+  next: T
+) => {
+  for (const key in next) {
+    if (ref.value[key] !== next[key]) {
+      ref.value = next;
+      return;
+    }
+  }
+};


Test patch:
diff --git a/exchanges/retry/src/retryExchange.test.ts b/exchanges/retry/src/retryExchange.test.ts
index 2a418bea..13de90f8 100644
--- a/exchanges/retry/src/retryExchange.test.ts
+++ b/exchanges/retry/src/retryExchange.test.ts
@@ -1,4 +1,14 @@
-import { pipe, map, makeSubject, publish, tap } from 'wonka';
+import {
+  Source,
+  pipe,
+  map,
+  makeSubject,
+  mergeMap,
+  fromValue,
+  fromArray,
+  publish,
+  tap,
+} from 'wonka';
 import { vi, expect, it, beforeEach, afterEach } from 'vitest';
 
 import {
@@ -67,7 +77,7 @@ beforeEach(() => {
   ({ source: ops$, next } = makeSubject<Operation>());
 });
 
-it(`retries if it hits an error and works for multiple concurrent operations`, () => {
+it('retries if it hits an error and works for multiple concurrent operations', () => {
   const queryTwo = gql`
     {
       films {
@@ -149,7 +159,7 @@ it('should retry x number of times and then return the successful result', () =>
     // @ts-ignore
     return {
       operation: forwardOp,
-      ...(forwardOp.context.retryCount! >= numberRetriesBeforeSuccess
+      ...(forwardOp.context.retry?.count >= numberRetriesBeforeSuccess
         ? { data: queryOneData }
         : { error: queryOneError }),
     };
@@ -186,6 +196,68 @@ it('should retry x number of times and then return the successful result', () =>
   expect(result).toHaveBeenCalledTimes(1);
 });
 
+it('should reset the retry counter if an operation succeeded first', () => {
+  let call = 0;
+  const response = vi.fn((forwardOp: Operation): Source<any> => {
+    expect(forwardOp.key).toBe(op.key);
+    if (call === 0) {
+      call++;
+      return fromValue({
+        operation: forwardOp,
+        error: queryOneError,
+      } as any);
+    } else if (call === 1) {
+      call++;
+      return fromArray([
+        {
+          operation: forwardOp,
+          data: queryOneData,
+        } as any,
+        {
+          operation: forwardOp,
+          error: queryOneError,
+        } as any,
+      ]);
+    } else {
+      expect(forwardOp.context.retry).toEqual({ count: 1, delay: null });
+
+      return fromValue({
+        operation: forwardOp,
+        data: queryOneData,
+      } as any);
+    }
+  });
+
+  const result = vi.fn();
+  const forward: ExchangeIO = ops$ => {
+    return pipe(ops$, mergeMap(response));
+  };
+
+  const mockRetryIf = vi.fn(() => true);
+
+  pipe(
+    retryExchange({
+      ...mockOptions,
+      retryIf: mockRetryIf,
+    })({
+      forward,
+      client,
+      dispatchDebug,
+    })(ops$),
+    tap(result),
+    publish
+  );
+
+  next(op);
+  vi.runAllTimers();
+
+  expect(mockRetryIf).toHaveBeenCalledTimes(2);
+  expect(mockRetryIf).toHaveBeenCalledWith(queryOneError as any, op);
+
+  expect(response).toHaveBeenCalledTimes(3);
+  expect(result).toHaveBeenCalledTimes(2);
+});
+
 it(`should still retry if retryIf undefined but there is a networkError`, () => {
   const errorWithNetworkError = {
     ...queryOneError,