Fix operation queue to only affect reexecuteOperation (#662)

* Fix operation queue to only affect reexecuteOperation

Previously any call that dispatches an operation could
lead to a queued dispatch, which stops peeking from working
since the effect (exchange pipeline) isn't actually executed
on the spot. In such a case the queue is flushed later which
can lead to undefined behaviour, e.g. looping onOperationEnd,
which means that we think a query is active while it's not.

Instead we want to apply the operation queuing only to
client.reexecuteOperation, which is what's called from inside
exchanges.

The new behaviour: We now queue up operations from
reexecuteOperation when we're currently dispatching another
operation. Otherwise every operation executes immediately.

* Update snapshot tests

.changeset/early-yaks-impress.md

@@ -0,0 +1,5 @@
+---
+'@urql/core': patch
+---
+
+Fix critical bug in operation queueing that can lead to unexpected teardowns and swallowed operations. This would happen when a teardown operation kicks off the queue.

packages/core/src/__snapshots__/client.test.ts.snap

@@ -15,6 +15,7 @@ Client {
   "maskTypename": false,
   "operations$": [Function],
   "preferGetMethod": false,
+  "queue": Array [],
   "reexecuteOperation": [Function],
   "requestPolicy": "cache-first",
   "results$": [Function],

packages/core/src/client.ts

@@ -84,10 +84,11 @@ export class Client {
   maskTypename: boolean;
 
   // These are internals to be used to keep track of operations
-  dispatchOperation: (operation: Operation) => void;
+  dispatchOperation: (operation?: Operation) => void;
   operations$: Source<Operation>;
   results$: Source<OperationResult>;
   activeOperations = Object.create(null) as ActiveOperations;
+  queue: Operation[] = [];
 
   constructor(opts: ClientOptions) {
     if (process.env.NODE_ENV !== 'production' && !opts.url) {
@@ -109,20 +110,13 @@ export class Client {
     >();
     this.operations$ = operations$;
 
-    // Internally operations aren't always dispatched immediately
-    // Since exchanges can dispatch and reexecute operations themselves,
-    // if we're inside an exchange we instead queue the operation and flush
-    // them in order after
-    const queuedOperations: Operation[] = [];
     let isDispatching = false;
-
-    this.dispatchOperation = (operation: Operation) => {
-      queuedOperations.push(operation);
+    this.dispatchOperation = (operation?: Operation) => {
+      if (operation) nextOperation(operation);
       if (!isDispatching) {
         isDispatching = true;
-        let queued;
-        while ((queued = queuedOperations.shift()) !== undefined)
-          nextOperation(queued);
+        let queued: Operation | void;
+        while ((queued = this.queue.shift())) nextOperation(queued);
         isDispatching = false;
       }
     };
@@ -246,7 +240,8 @@ export class Client {
     // Reexecute operation only if any subscribers are still subscribed to the
     // operation's exchange results
     if ((this.activeOperations[operation.key] || 0) > 0) {
-      this.dispatchOperation(operation);
+      this.queue.push(operation);
+      this.dispatchOperation();
     }
   };