Repository: FormidableLabs/urql (ts)

Issue:
Re-executing query doesn't work
Hi there. I'm trying to upgrade a project from `urql@1.11.6` to `urql@2.0.3`, and I'm running into an issue when trying to re-execute a query using the `'network-only'` request policy. I've created a minimal reproduction here https://github.com/bausmeier/urql-repro, and the request in question is [in the `AccountInfo` component](https://github.com/bausmeier/urql-repro/blob/b329302bc47af6dbecd1fc0222528b505785035c/src/AccountInfo.jsx#L20).

Effectively what I've got is two components which render alongside each other:
1. A panel which shows some details of the viewer's account.
2. A login form which on success [stores an access token](https://github.com/bausmeier/urql-repro/blob/b329302bc47af6dbecd1fc0222528b505785035c/src/LoginForm.jsx#L11) so that it can be added to the `Authorization` header of subsequent requests [in `fetchOptions`](https://github.com/bausmeier/urql-repro/blob/b329302bc47af6dbecd1fc0222528b505785035c/src/UrqlProvider.jsx#L9-L19).

The rough reproduction looks like this:

![2021-05-08-193100_grim](https://user-images.githubusercontent.com/116090/117548307-8fa9f400-b034-11eb-9c5e-5494a846ce4c.png)

The `viewer` query which I'm executing will return `null` if there is no `Authorization` header, and an object containing an id if there is an `Authorization` header.

What I expect to happen, and what _does_ happen with `urql@1.11.6`, is that when the "Refresh" button is pressed, a GraphQL network request is made to get the viewer. If an access token has been set by "logging in" then it should be included in the request, and the account info should display the viewer's id.

What actually happens with `urql@2.0.3` is that after the initial request is made on page load, pressing the "Refresh" button doesn't fire off any more GraphQL requests at all.

I may be doing something completely stupid, but this used to work with the older version of urql, and a similar approach works with Apollo, so I'm a bit perplexed as to why it's not working with the latest urql. I'm happy to provide more info if needed.

Pull request description:
(core) - Fix concurrent operations not respecting request policies
Resolves #1621

## Summary

We discovered a slight regression in `@urql/core@2.1.0` that caused reexecuting an operation to not be possible under the usual circumstances anymore. The reason for this was that the `replayOnStart` utility was effective but we'd share the entire source too eagerly. Since this source was shared, no new operation would be created for future concurrent runs which makes changing the `requestPolicy` of an operation, e.g. for the `executeQuery` function in bindings, impossible.

Instead, I've now moved the implementation back to `Client` and cleaned it up a little, which should make it easier for us to predict how it'll behave. Sources aren't completely shared anymore and instead (which was the same in a prior implementation) we only share the result source and store that internally. The `OperationResult` stream and `onStart` behaviour isn't shared any longer.


Patch:
diff --git a/.changeset/new-dingos-clean.md b/.changeset/new-dingos-clean.md
new file mode 100644
index 00000000..d65f7097
--- /dev/null
+++ b/.changeset/new-dingos-clean.md
@@ -0,0 +1,5 @@
+---
+'@urql/core': patch
+---
+
+Fix a regression in `@urql/core@2.1.1` that prevented concurrent operations from being dispatched with differing request policies, which for instance prevented the explicit `executeQuery` calls on bindings to fail.
diff --git a/packages/core/src/client.ts b/packages/core/src/client.ts
index 74af03ab..d278bed0 100755
--- a/packages/core/src/client.ts
+++ b/packages/core/src/client.ts
@@ -4,6 +4,7 @@ import {
   filter,
   makeSubject,
   onEnd,
+  onPush,
   onStart,
   pipe,
   share,
@@ -41,7 +42,6 @@ import {
 import {
   createRequest,
   withPromise,
-  replayOnStart,
   maskTypename,
   noop,
   makeOperation,
@@ -69,6 +69,11 @@ export interface ClientOptions {
 
 export const createClient = (opts: ClientOptions) => new Client(opts);
 
+interface ActiveOperation {
+  source: Source<OperationResult>;
+  replay: OperationResult | null;
+}
+
 /** The URQL application-wide client library. Each execute method starts a GraphQL request and returns a stream of results. */
 export class Client {
   /** Start an operation from an exchange */
@@ -90,7 +95,7 @@ export class Client {
   dispatchOperation: (operation?: Operation | void) => void;
   operations$: Source<Operation>;
   results$: Source<OperationResult>;
-  activeOperations: Map<number, Source<OperationResult>> = new Map();
+  activeOperations: Map<number, ActiveOperation> = new Map();
   queue: Operation[] = [];
 
   constructor(opts: ClientOptions) {
@@ -198,96 +203,113 @@ export class Client {
   executeRequestOperation<Data = any, Variables = object>(
     operation: Operation<Data, Variables>
   ): Source<OperationResult<Data, Variables>> {
-    let active = this.activeOperations.get(operation.key);
-    if (active) return active;
-
-    let result$ = pipe(
-      this.results$,
-      filter((res: OperationResult) => res.operation.key === operation.key)
-    ) as Source<OperationResult<Data, Variables>>;
-    if (this.maskTypename) {
-      result$ = pipe(
-        result$,
-        map(res => ({ ...res, data: maskTypename(res.data) }))
+    let activeOperation = this.activeOperations.get(operation.key);
+    if (!activeOperation) {
+      let result$ = pipe(
+        this.results$,
+        filter(
+          (res: OperationResult) =>
+            res.operation.kind === operation.kind &&
+            res.operation.key === operation.key
+        )
       );
-    }
 
-    // A mutation is always limited to just a single result and is never shared
-    if (operation.kind === 'mutation') {
-      return pipe(
-        result$,
-        onStart<OperationResult>(() => this.dispatchOperation(operation)),
-        take(1)
-      );
-    }
-
-    const teardown$ = pipe(
-      this.operations$,
-      filter(
-        (op: Operation) => op.kind === 'teardown' && op.key === operation.key
-      )
-    );
+      // Mask typename properties if the option for it is turned on
+      if (this.maskTypename) {
+        result$ = pipe(
+          result$,
+          map(res => ({ ...res, data: maskTypename(res.data) }))
+        );
+      }
 
-    const refetch$ = pipe(
-      this.operations$,
-      filter(
-        (op: Operation) =>
-          op.kind === operation.kind &&
-          op.key === operation.key &&
-          (op.context.requestPolicy === 'network-only' ||
-            op.context.requestPolicy === 'cache-and-network')
-      ),
-      take(1)
-    );
+      // A mutation is always limited to just a single result and is never shared
+      if (operation.kind === 'mutation') {
+        return pipe(
+          result$,
+          onStart(() => this.dispatchOperation(operation)),
+          take(1)
+        );
+      }
 
-    result$ = pipe(
-      result$,
-      takeUntil(teardown$),
-      switchMap(result => {
-        if (result.stale) return fromValue(result);
-        return merge([
-          fromValue(result),
-          pipe(
-            refetch$,
-            map(() => ({ ...result, stale: true }))
+      activeOperation = {
+        source: pipe(
+          result$,
+          // End the results stream when an active teardown event is sent
+          takeUntil(
+            pipe(
+              this.operations$,
+              filter(op => op.kind === 'teardown' && op.key === operation.key)
+            )
           ),
-        ]);
-      }),
-      onEnd<OperationResult>(() => {
-        this.activeOperations.delete(operation.key);
-        for (let i = this.queue.length - 1; i >= 0; i--)
-          if (this.queue[i].key === operation.key) this.queue.splice(i, 1);
-        this.dispatchOperation(
-          makeOperation('teardown', operation, operation.context)
-        );
-      })
-    );
+          switchMap(result => {
+            if (result.stale) {
+              return fromValue(result);
+            }
+
+            return merge([
+              fromValue(result),
+              // Mark a result as stale when a new operation is sent for it
+              pipe(
+                this.operations$,
+                filter(op => {
+                  return (
+                    op.kind === operation.kind &&
+                    op.key === operation.key &&
+                    (op.context.requestPolicy === 'network-only' ||
+                      op.context.requestPolicy === 'cache-and-network')
+                  );
+                }),
+                take(1),
+                map(() => ({ ...result, stale: true }))
+              ),
+            ]);
+          }),
+          onPush(result => {
+            activeOperation!.replay = result;
+          }),
+          onEnd(() => {
+            // Delete the active operation handle
+            activeOperation!.replay = null;
+            this.activeOperations.delete(operation.key);
+            // Delete all queued up operations of the same key on end
+            for (let i = this.queue.length - 1; i >= 0; i--)
+              if (this.queue[i].key === operation.key) this.queue.splice(i, 1);
+            // Dispatch a teardown signal for the stopped operation
+            this.dispatchOperation(
+              makeOperation('teardown', operation, operation.context)
+            );
+          }),
+          share
+        ),
+        replay: null,
+      };
+    }
 
-    if (operation.kind === 'subscription') {
-      active = pipe(
-        result$,
-        onStart(() => {
-          this.activeOperations.set(operation.key, active!);
+    const prevReplay = activeOperation.replay;
+    const subject = makeSubject<OperationResult>();
+    const isNetworkOperation =
+      operation.context.requestPolicy === 'cache-and-network' ||
+      operation.context.requestPolicy === 'network-only';
+
+    return pipe(
+      merge([subject.source, activeOperation.source]),
+      onStart(() => {
+        this.activeOperations.set(operation.key, activeOperation!);
+        if (operation.kind === 'subscription') {
+          return this.dispatchOperation(operation);
+        } else if (isNetworkOperation) {
           this.dispatchOperation(operation);
-        }),
-        share
-      );
-    } else {
-      const mode =
-        operation.context.requestPolicy === 'cache-and-network' ||
-        operation.context.requestPolicy === 'network-only'
-          ? 'pre'
-          : 'post';
-      active = pipe(
-        result$,
-        replayOnStart(mode, () => {
-          this.activeOperations.set(operation.key, active!);
-          this.dispatchOperation(operation);
-        })
-      );
-    }
+        }
 
-    return active;
+        if (prevReplay != null && prevReplay === activeOperation!.replay) {
+          subject.next(
+            isNetworkOperation ? { ...prevReplay, stale: true } : prevReplay
+          );
+        } else if (!isNetworkOperation) {
+          this.dispatchOperation(operation);
+        }
+      })
+    );
   }
 
   query<Data = any, Variables extends object = {}>(
diff --git a/packages/core/src/utils/streamUtils.ts b/packages/core/src/utils/streamUtils.ts
index 148d0a4b..33f1d36e 100644
--- a/packages/core/src/utils/streamUtils.ts
+++ b/packages/core/src/utils/streamUtils.ts
@@ -1,66 +1,9 @@
-import {
-  Operator,
-  Source,
-  pipe,
-  toPromise,
-  take,
-  share,
-  onPush,
-  onStart,
-  onEnd,
-  subscribe,
-  make,
-} from 'wonka';
+import { Source, pipe, toPromise, take } from 'wonka';
 
-import { OperationResult, PromisifiedSource } from '../types';
+import { PromisifiedSource } from '../types';
 
 export function withPromise<T>(source$: Source<T>): PromisifiedSource<T> {
   (source$ as PromisifiedSource<T>).toPromise = () =>
     pipe(source$, take(1), toPromise);
   return source$ as PromisifiedSource<T>;
 }
-
-export type ReplayMode = 'pre' | 'post';
-
-export function replayOnStart<T extends OperationResult>(
-  mode: ReplayMode,
-  start: () => void
-): Operator<T, T> {
-  return source$ => {
-    let replay: T | void;
-
-    const shared$ = pipe(
-      source$,
-      onEnd(() => {
-        replay = undefined;
-      }),
-      onPush(value => {
-        replay = value;
-      }),
-      share
-    );
-
-    return make<T>(observer => {
-      const prevReplay = replay;
-
-      return pipe(
-        shared$,
-        onEnd(observer.complete),
-        onStart(() => {
-          if (mode === 'pre') {
-            start();
-          }
-
-          if (prevReplay !== undefined && prevReplay === replay) {
-            observer.next(
-              mode === 'pre' ? { ...prevReplay, stale: true } : prevReplay
-            );
-          } else if (mode === 'post') {
-            start();
-          }
-        }),
-        subscribe(observer.next)
-      ).unsubscribe;
-    });
-  };
-}


Test patch:
diff --git a/packages/core/src/client.test.ts b/packages/core/src/client.test.ts
index 09678edb..e7e8dfbb 100755
--- a/packages/core/src/client.test.ts
+++ b/packages/core/src/client.test.ts
@@ -565,6 +565,63 @@ describe('shared sources behavior', () => {
     expect(resultTwo).toHaveBeenCalledTimes(1);
   });
 
+  it('dispatches the correct request policy on subsequent sources', async () => {
+    const exchange: Exchange = () => ops$ => {
+      let i = 0;
+      return pipe(
+        ops$,
+        map(op => ({
+          data: ++i,
+          operation: op,
+        })),
+        delay(1)
+      );
+    };
+
+    const client = createClient({
+      url: 'test',
+      exchanges: [exchange],
+    });
+
+    const resultOne = jest.fn();
+    const resultTwo = jest.fn();
+    const operationOne = makeOperation('query', queryOperation, {
+      ...queryOperation.context,
+      requestPolicy: 'cache-first',
+    });
+    const operationTwo = makeOperation('query', queryOperation, {
+      ...queryOperation.context,
+      requestPolicy: 'network-only',
+    });
+
+    pipe(client.executeRequestOperation(operationOne), subscribe(resultOne));
+
+    expect(resultOne).toHaveBeenCalledTimes(0);
+
+    jest.advanceTimersByTime(1);
+
+    expect(resultOne).toHaveBeenCalledTimes(1);
+    expect(resultOne).toHaveBeenCalledWith({
+      data: 1,
+      operation: operationOne,
+    });
+
+    pipe(client.executeRequestOperation(operationTwo), subscribe(resultTwo));
+
+    expect(resultTwo).toHaveBeenCalledWith({
+      data: 1,
+      operation: operationOne,
+      stale: true,
+    });
+
+    jest.advanceTimersByTime(1);
+
+    expect(resultTwo).toHaveBeenCalledWith({
+      data: 2,
+      operation: operationTwo,
+    });
+  });
+
   it('replays results from prior operation result as needed (network-only)', async () => {
     const exchange: Exchange = () => ops$ => {
       let i = 0;