fix(client): eager unblocking of in-flight operations (#3573)

+6

.changeset/tiny-pots-allow.md

··· 1 + --- 2 + '@urql/core': patch 3 + --- 4 + 5 + Fix issue where a reexecute on an in-flight operation would lead to multiple network-requests. 6 + For example, this issue presents itself when Graphcache is concurrently updating multiple, inter-dependent queries with shared entities. One query completing while others are still in-flight may lead to duplicate operations being issued.

+123 -3

packages/core/src/client.test.ts

··· 19 19 onPush, 20 20 tap, 21 21 take, 22 + fromPromise, 23 + fromValue, 24 + mergeMap, 22 25 } from 'wonka'; 23 26 24 27 import { gql } from './gql'; 25 28 import { Exchange, Operation, OperationResult } from './types'; 26 29 import { makeOperation } from './utils'; 27 30 import { Client, createClient } from './client'; 28 - import { queryOperation, subscriptionOperation } from './test-utils'; 31 + import { 32 + mutationOperation, 33 + queryOperation, 34 + subscriptionOperation, 35 + } from './test-utils'; 29 36 30 37 const url = 'https://hostname.com'; 31 38 ··· 701 708 expect(onOperation).toHaveBeenCalledTimes(2); 702 709 }); 703 710 704 - it('unblocks operations on call to reexecuteOperation', async () => { 711 + it('unblocks mutation operations on call to reexecuteOperation', async () => { 705 712 const onOperation = vi.fn(); 706 713 const onResult = vi.fn(); 707 714 ··· 724 731 exchanges: [exchange], 725 732 }); 726 733 734 + const operation = makeOperation('mutation', mutationOperation, { 735 + ...mutationOperation.context, 736 + requestPolicy: 'cache-first', 737 + }); 738 + 739 + pipe(client.executeRequestOperation(operation), subscribe(onResult)); 740 + 741 + expect(onOperation).toHaveBeenCalledTimes(1); 742 + expect(onResult).toHaveBeenCalledTimes(0); 743 + 744 + client.reexecuteOperation(operation); 745 + await Promise.resolve(); 746 + 747 + expect(onOperation).toHaveBeenCalledTimes(2); 748 + expect(onResult).toHaveBeenCalledTimes(1); 749 + }); 750 + 751 + // See https://github.com/urql-graphql/urql/issues/3254 752 + it('unblocks stale operations', async () => { 753 + const onOperation = vi.fn(); 754 + const onResult = vi.fn(); 755 + 756 + let sends = 0; 757 + const exchange: Exchange = () => ops$ => 758 + pipe( 759 + ops$, 760 + onPush(onOperation), 761 + map(op => ({ 762 + hasNext: false, 763 + stale: sends++ ? false : true, 764 + data: 'test', 765 + operation: op, 766 + })) 767 + ); 768 + 769 + const client = createClient({ 770 + url: 'test', 771 + exchanges: [exchange], 772 + }); 773 + 727 774 const operation = makeOperation('query', queryOperation, { 728 775 ...queryOperation.context, 729 776 requestPolicy: 'cache-first', ··· 732 779 pipe(client.executeRequestOperation(operation), subscribe(onResult)); 733 780 734 781 expect(onOperation).toHaveBeenCalledTimes(1); 782 + expect(onResult).toHaveBeenCalledTimes(1); 783 + 784 + client.reexecuteOperation(operation); 785 + await Promise.resolve(); 786 + 787 + expect(onOperation).toHaveBeenCalledTimes(2); 788 + expect(onResult).toHaveBeenCalledTimes(2); 789 + }); 790 + 791 + // See https://github.com/urql-graphql/urql/issues/3565 792 + it('blocks reexecuting operations that are in-flight', async () => { 793 + const onOperation = vi.fn(); 794 + const onResult = vi.fn(); 795 + 796 + let resolve; 797 + const exchange: Exchange = 798 + ({ client }) => 799 + ops$ => 800 + pipe( 801 + ops$, 802 + onPush(onOperation), 803 + mergeMap(op => { 804 + if (op.key === queryOperation.key) { 805 + const promise = new Promise<OperationResult>(res => { 806 + resolve = res; 807 + }); 808 + return fromPromise( 809 + promise.then(() => { 810 + return { 811 + hasNext: false, 812 + stale: false, 813 + data: 'test', 814 + operation: op, 815 + }; 816 + }) 817 + ); 818 + } else { 819 + client.reexecuteOperation(queryOperation); 820 + return fromValue({ 821 + hasNext: false, 822 + stale: false, 823 + data: 'test', 824 + operation: op, 825 + }); 826 + } 827 + }) 828 + ); 829 + 830 + const client = createClient({ 831 + url: 'test', 832 + exchanges: [exchange], 833 + }); 834 + 835 + const operation = makeOperation('query', queryOperation, { 836 + ...queryOperation.context, 837 + requestPolicy: 'cache-first', 838 + }); 839 + 840 + const mutation = makeOperation('mutation', mutationOperation, { 841 + ...mutationOperation.context, 842 + requestPolicy: 'cache-first', 843 + }); 844 + 845 + pipe(client.executeRequestOperation(operation), subscribe(onResult)); 846 + 847 + expect(onOperation).toHaveBeenCalledTimes(1); 735 848 expect(onResult).toHaveBeenCalledTimes(0); 736 849 737 - client.reexecuteOperation(operation); 850 + pipe(client.executeRequestOperation(mutation), subscribe(onResult)); 738 851 await Promise.resolve(); 739 852 740 853 expect(onOperation).toHaveBeenCalledTimes(2); 741 854 expect(onResult).toHaveBeenCalledTimes(1); 855 + 856 + resolve(); 857 + await Promise.resolve(); 858 + await Promise.resolve(); 859 + await Promise.resolve(); 860 + expect(onOperation).toHaveBeenCalledTimes(2); 861 + expect(onResult).toHaveBeenCalledTimes(2); 742 862 }); 743 863 }); 744 864

+32 -11

packages/core/src/client.ts

··· 643 643 // Store replay result 644 644 onPush(result => { 645 645 if (result.stale) { 646 - // If the current result has queued up an operation of the same 647 - // key, then `stale` refers to it 648 - for (const operation of queue) { 649 - if (operation.key === result.operation.key) { 650 - dispatched.delete(operation.key); 651 - break; 646 + if (!result.hasNext) { 647 + // we are dealing with an optimistic mutation or a partial result 648 + dispatched.delete(operation.key); 649 + } else { 650 + // If the current result has queued up an operation of the same 651 + // key, then `stale` refers to it 652 + for (const operation of queue) { 653 + if (operation.key === result.operation.key) { 654 + dispatched.delete(operation.key); 655 + break; 656 + } 652 657 } 653 658 } 654 659 } else if (!result.hasNext) { ··· 697 702 // operation's exchange results 698 703 if (operation.kind === 'teardown') { 699 704 dispatchOperation(operation); 700 - } else if (operation.kind === 'mutation' || active.has(operation.key)) { 701 - let queued = false; 702 - for (let i = 0; i < queue.length; i++) 703 - queued = queued || queue[i].key === operation.key; 704 - if (!queued) dispatched.delete(operation.key); 705 + } else if (operation.kind === 'mutation') { 705 706 queue.push(operation); 706 707 Promise.resolve().then(dispatchOperation); 708 + } else if (active.has(operation.key)) { 709 + let queued = false; 710 + for (let i = 0; i < queue.length; i++) { 711 + if (queue[i].key === operation.key) { 712 + queue[i] = operation; 713 + queued = true; 714 + } 715 + } 716 + 717 + if ( 718 + !queued && 719 + (!dispatched.has(operation.key) || 720 + operation.context.requestPolicy === 'network-only') 721 + ) { 722 + queue.push(operation); 723 + Promise.resolve().then(dispatchOperation); 724 + } else { 725 + dispatched.delete(operation.key); 726 + Promise.resolve().then(dispatchOperation); 727 + } 707 728 } 708 729 }, 709 730

Configure Feed

Configure Feed