Allow subscriptions to be deduplicated #12663

jerelmiller · 2025-05-30T22:00:08Z

Ensures that subscriptions can be deduplicated properly. Disable deduplication with queryDeduplication like queries.

pkg-pr-new · 2025-05-30T22:01:32Z

npm i https://pkg.pr.new/apollographql/apollo-client/@apollo/client@12663

commit: 38c7792

changeset-bot · 2025-05-30T22:01:35Z

🦋 Changeset detected

Latest commit: 61a4c51

The changes in this PR will be included in the next version bump.

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

jerelmiller · 2025-05-30T22:01:48Z

src/utilities/internal/onAnyEvent.ts

Yay for finalize!

github-actions · 2025-05-30T22:02:02Z

size-limit report 📦

Path	Size
import { ApolloClient, InMemoryCache, HttpLink } from "@apollo/client" (CJS)	42.32 KB (-0.15% 🔽)
import { ApolloClient, InMemoryCache, HttpLink } from "@apollo/client" (production) (CJS)	37.53 KB (+0.03% 🔺)
import { ApolloClient, InMemoryCache, HttpLink } from "@apollo/client"	32.63 KB (+0.05% 🔺)
import { ApolloClient, InMemoryCache, HttpLink } from "@apollo/client" (production)	27.2 KB (+0.35% 🔺)
import { ApolloProvider } from "@apollo/client/react"	5.6 KB (+0.41% 🔺)
import { ApolloProvider } from "@apollo/client/react" (production)	963 B (0%)
import { useQuery } from "@apollo/client/react"	11.69 KB (+0.19% 🔺)
import { useQuery } from "@apollo/client/react" (production)	7.08 KB (0%)
import { useLazyQuery } from "@apollo/client/react"	6.79 KB (+0.66% 🔺)
import { useLazyQuery } from "@apollo/client/react" (production)	2.13 KB (0%)
import { useMutation } from "@apollo/client/react"	6.13 KB (+0.42% 🔺)
import { useMutation } from "@apollo/client/react" (production)	1.48 KB (0%)
import { useSubscription } from "@apollo/client/react"	6.47 KB (+0.26% 🔺)
import { useSubscription } from "@apollo/client/react" (production)	1.8 KB (-1.92% 🔽)
import { useSuspenseQuery } from "@apollo/client/react"	8.23 KB (+0.39% 🔺)
import { useSuspenseQuery } from "@apollo/client/react" (production)	3.59 KB (+0.03% 🔺)
import { useBackgroundQuery } from "@apollo/client/react"	8.03 KB (+0.25% 🔺)
import { useBackgroundQuery } from "@apollo/client/react" (production)	3.42 KB (+0.18% 🔺)
import { useLoadableQuery } from "@apollo/client/react"	8.01 KB (+0.26% 🔺)
import { useLoadableQuery } from "@apollo/client/react" (production)	3.39 KB (0%)
import { useReadQuery } from "@apollo/client/react"	6.25 KB (+0.4% 🔺)
import { useReadQuery } from "@apollo/client/react" (production)	1.62 KB (0%)
import { useFragment } from "@apollo/client/react"	6.31 KB (+0.42% 🔺)
import { useFragment } from "@apollo/client/react" (production)	1.68 KB (0%)

…ion correctly

jerelmiller · 2025-06-05T06:14:30Z

src/core/QueryManager.ts

                inFlightLinkObservables.peek(printedServerQuery, varJson) ===
-                  entry
+                entry
              ) {
                inFlightLinkObservables.remove(printedServerQuery, varJson);


Turns out that previously this deduplicated request was not removed from inFlightObservables when a request was unsubscribed before a value was emitted. Now that we are using finalize instead of onAnyEvent, this ensures the request is removed from inFlightObservables when unsubscribed.

.changeset/green-ways-jam.md

src/core/ApolloClient.ts

phryneas · 2025-06-05T10:38:41Z

src/react/hooks/useSubscription.ts

    },
  };

-  let observable: Observable<SubscribeResult<MaybeMasked<TData>>> | null = null;
-  return Object.assign(
-    new Observable<SubscribeResult<MaybeMasked<TData>>>((observer) => {
-      // lazily start the subscription when the first observer subscribes
-      // to get around strict mode
-      if (!observable) {
-        observable = client.subscribe(options);
-      }
-      const sub = observable.subscribe(observer);
-      return () => sub.unsubscribe();
-    }),


Wouldn't StrictMode still be a concern if someone disabled deduplication?

It seems I get the same result (executing a request through the link chain twice) whether I use the old code or new code since React 18 will run the useSyncExternalStore callback twice in React 18 which will observable.subscribe twice. About the only thing this does is prevent creating multiple observables returned from client.subscribe itself rather than the two wrapper observables since the useState callback is called twice in strict mode as well.

The connection to the link chain doesn't start until you call subscribe on the returned observable (perhaps it eagerly started the subscription in the past?)

Yeah, I think it did that eagerly on creation - isn't that the case anymore?

Nope! I was thinking I should add an explicit test for this, but it is covered by this test https://github.com/apollographql/apollo-client/pull/12663/files/4d0ed15213cff36837bb4b5c58d529f3476bcfae#diff-61c0c1641f292ca56586caa27870be6570b63d3fed8d2f9eb1e3ff36d36f4c35R937-R1013, specifically this line:

// Ensure we aren't eagerly subscribing expect(onSubscribe).not.toHaveBeenCalled();

Added an explicit test in fe947b8 to make sure the request is lazily created.

phryneas · 2025-06-05T11:02:13Z

src/core/QueryManager.ts

+        entry.observable = execute(
+          link,
+          operation,
+          executeContext
+        ) as Observable<FetchResult<TData>>;


Wouldn't this need extra logic so non-duplicated subscriptions could be restarted, too?

Oops thanks! Updated in 5931152

phryneas · 2025-06-05T11:19:08Z

src/react/hooks/useSubscription.ts

@@ -331,8 +329,12 @@ export function useSubscription<
      !optionsRef.current.skip,
      "A subscription that is skipped cannot be restarted."
    );
-    setObservable(recreateRef.current());
-  }, [optionsRef, recreateRef]);
+    if (observable?.__.completed) {


Wouldn't we also have to recreate if the subscription is still running, but options have changed?

Also if the subscription was not deduplicated (if that isn't handled elsewhere)?

Isn't that covered by this test?

apollo-client/src/react/hooks/__tests__/useSubscription.test.tsx

Lines 2346 to 2431 in 5931152

it("will use the most recently passed in options", async () => {

using _disabledAct = disableActEnvironment();

const {

link,

takeSnapshot,

getCurrentSnapshot,

onSubscribe,

onUnsubscribe,

rerender,

} = await setup({

variables: { id: "1" },

});

{

const snapshot = await takeSnapshot();

expect(snapshot).toStrictEqualTyped({

loading: true,

data: undefined,

error: undefined,

});

}

link.simulateResult({ result: { data: { totalLikes: 1 } } });

{

const snapshot = await takeSnapshot();

expect(snapshot).toStrictEqualTyped({

loading: false,

data: { totalLikes: 1 },

error: undefined,

});

}

await expect(takeSnapshot).not.toRerender({ timeout: 20 });

expect(onUnsubscribe).toHaveBeenCalledTimes(0);

expect(onSubscribe).toHaveBeenCalledTimes(1);

void rerender({ variables: { id: "2" } });

await waitFor(() => expect(onUnsubscribe).toHaveBeenCalledTimes(1));

expect(onSubscribe).toHaveBeenCalledTimes(2);

expect(link.operation?.variables).toStrictEqual({ id: "2" });

{

const snapshot = await takeSnapshot();

expect(snapshot).toStrictEqualTyped({

loading: true,

data: undefined,

error: undefined,

});

}

link.simulateResult({ result: { data: { totalLikes: 1000 } } });

{

const snapshot = await takeSnapshot();

expect(snapshot).toStrictEqualTyped({

loading: false,

data: { totalLikes: 1000 },

error: undefined,

});

}

expect(onUnsubscribe).toHaveBeenCalledTimes(1);

expect(onSubscribe).toHaveBeenCalledTimes(2);

expect(link.operation?.variables).toStrictEqual({ id: "2" });

getCurrentSnapshot().restart();

await waitFor(() => expect(onUnsubscribe).toHaveBeenCalledTimes(2));

expect(onSubscribe).toHaveBeenCalledTimes(3);

expect(link.operation?.variables).toStrictEqual({ id: "2" });

await expect(takeSnapshot).not.toRerender();

link.simulateResult({ result: { data: { totalLikes: 1005 } } });

{

const snapshot = await takeSnapshot();

expect(snapshot).toStrictEqualTyped({

loading: false,

data: { totalLikes: 1005 },

error: undefined,

});

}

});

The observable gets recreated when you pass in new options by

apollo-client/src/react/hooks/useSubscription.ts

Lines 227 to 238 in 5931152

!observable ||

((client !== observable.__.client ||

subscription !== observable.__.query ||

fetchPolicy !== observable.__.fetchPolicy ||

errorPolicy !== observable.__.errorPolicy ||

!equal(variables, observable.__.variables)) &&

(typeof shouldResubscribe === "function" ?

!!shouldResubscribe(options!)

: shouldResubscribe) !== false)

) {

setObservable((observable = recreate()));

}

so wouldn't this cover it since that would unsubscribe from the old observable, then resubscribe with the new observable?

Oh dang you're right, in my mind for some reason useSubscription wouldn't change options until you called restart but of course that's not the case.

Copilot

Pull Request Overview

This PR adds support for deduplicating subscription operations over shared observables and exposes a restart method on subscription observables to re-establish connections, with opt-out via queryDeduplication.

Introduces a SubscriptionObservable<T> interface (with restart) and updates return types in core and React hooks
Updates QueryManager to track deduped subscriptions, wrap link observables with restart logic, and choose share vs shareReplay based on operation type
Adjusts useSubscription hook and existing tests to set a completion flag and expose restart behavior

Reviewed Changes

Copilot reviewed 22 out of 22 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
src/react/hooks/useSubscription.ts	Tracks completion and calls `observable.restart()` in effect
src/core/QueryManager.ts	Adds `withRestart` wrapper, stores restart in `inFlightLinkObservables`, uses `share` for subscriptions
src/core/types.ts	Defines `SubscriptionObservable<T>` interface
src/core/ApolloClient.ts	Returns `SubscriptionObservable` and preserves `restart` method
src/utilities/internal/onAnyEvent.ts	Removes deprecated utility
src/utilities/internal/index.ts	Removes `onAnyEvent` export
Tests and snapshots	Updated to assert `restart` presence and add restart tests
.changeset & .api-reports	Reflect updated API surface and release notes

Comments suppressed due to low confidence (3)

src/core/ApolloClient.ts:547

[nitpick] Attaching restart only to the mapped observable means users lose restart if they chain further RxJS operators (e.g. .pipe(map(...))). Consider providing a custom observable class or operator that preserves the restart method across operator chains.

return Object.assign(mapped, { restart: observable.restart });

src/core/QueryManager.ts:1284

The branch where deduplication is disabled (queryDeduplication: false) does not appear to have a corresponding test. Adding a test that verifies no dedup and correct restart behavior when deduplication is off would ensure coverage for this use case.

} else {

src/react/hooks/useSubscription.ts:332

The useEffect will call observable.restart() on every render when the subscription is active, potentially causing repeated reconnections. Consider introducing a user-controlled trigger or additional state to guard restarts so they only occur on explicit requests or specific dependency changes.

if (observable?.__.completed) {

Copilot · 2025-06-05T16:32:39Z

src/core/QueryManager.ts

+      function withRestart(source: Observable<FetchResult>) {
+        return new Observable<FetchResult>((observer) => {


[nitpick] The withRestart wrapper embeds complex subscription and restart logic inline. Extracting it into a dedicated utility or method with clear documentation would improve readability and reuse.

Suggested change

function withRestart(source: Observable<FetchResult>) {

return new Observable<FetchResult>((observer) => {

/**

* Wraps an observable with restartable subscription logic.

* @param source The source observable to wrap.

* @param entry An object containing the restart function.

* @returns A new observable with restartable subscription logic.

*/

function withRestart<T>(source: Observable<T>, entry: { restart?: () => void }): Observable<T> {

return new Observable<T>((observer) => {

Appreciate the suggestion, but nah its internal

phryneas

Looks good to me now.

jerelmiller requested a review from phryneas May 30, 2025 22:00

jerelmiller commented May 30, 2025

View reviewed changes

src/utilities/internal/onAnyEvent.ts Outdated

Copy link

Member Author

jerelmiller May 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yay for finalize!

jerelmiller linked an issue Jun 3, 2025 that may be closed by this pull request

Tracking issue: deduplicate running subscriptions in 4.0 #12348

Closed

jerelmiller added 24 commits June 4, 2025 23:14

Add tests for deduplicating subscriptions

637bcbb

Use tap to implement onAnyEvent

1bf0768

Fix wrong assertion

1a7a02d

Don't replay shared observable for subscriptions

b8690e6

Wait until complete or error to remove in flight observable

be2bf2c

Fix issue where unsubscribing didn't remove obs from in flight

6057bc4

Remove onAnyEvent helper

ee6a174

Update api report

cf8c945

Only call getOperationDefinition once

bb71f73

Update size limits

e303be7

Add changeset

1722c50

Add additional tests for mixing deduplicated subscriptions

a550193

Update exports snapshot

de5d466

Fix typo in changeset

1a3429b

Remove unused import

6b0a432

Ensure useSubscription restart can sync unsubscribe when calling restart

be67500

Write failing test to restart a subscription

2f3e363

Add handler to remove observer in tests

763d664

Add test to ensure multiplexed subscriptions unsubscribe from connect…

6f618c9

…ion correctly

Add test to restart subscription with multiple observers

8efe370

Add ability to restart subscription

1ff1347

Add test to restart deduped subscriptions

b6ef837

Use mergeMap to fix local state export

e888239

Copy restart instead of pipe

b0803ef

jerelmiller force-pushed the jerel/dedupe-subscriptions branch from a6576bd to 77c96b3 Compare June 5, 2025 05:17

jerelmiller added 11 commits June 4, 2025 23:18

Add comment

a63797b

Rename to SubscriptionObservable. Add doc block

c7e9737

Minor tweak to comment

0c5ee00

Export SubscriptionObservable from core

25570d1

Remove unneeded as any cast

d0f3a01

Fix lint warning

8c45eec

Update maxUsageCount for failing test

16dc3be

Add comment

7295a5a

Add another changeset

fd560c1

Only wrap promise in from

fe38fd0

Remove ref to track sync unsubscribe

eee79e8

jerelmiller commented Jun 5, 2025

View reviewed changes

Update size limits

4d0ed15

phryneas reviewed Jun 5, 2025

View reviewed changes

jerelmiller added 6 commits June 5, 2025 09:33

Fix lint warning

6cb5431

Switch to test

fefa680

Ensure non-deduplicatd subscriptions can be restarted

5931152

Add test to ensure connection is lazily created

fe947b8

Merge changesets

61a4c51

Update size limits

38c7792

phryneas requested a review from Copilot June 5, 2025 16:31

Copilot AI reviewed Jun 5, 2025

View reviewed changes

jerelmiller requested a review from phryneas June 5, 2025 16:40

phryneas approved these changes Jun 5, 2025

View reviewed changes

github-actions bot added the auto-cleanup 🤖 label Jun 5, 2025

jerelmiller merged commit 01512f2 into release-4.0 Jun 5, 2025
46 of 48 checks passed

jerelmiller deleted the jerel/dedupe-subscriptions branch June 5, 2025 16:57

github-actions bot mentioned this pull request Jun 5, 2025

Version Packages (alpha) #12676

Merged

jerelmiller mentioned this pull request Jun 5, 2025

Tracking issue: deduplicate running subscriptions in 4.0 #12348

Closed

	it("will use the most recently passed in options", async () => {
	using _disabledAct = disableActEnvironment();
	const {
	link,
	takeSnapshot,
	getCurrentSnapshot,
	onSubscribe,
	onUnsubscribe,
	rerender,
	} = await setup({
	variables: { id: "1" },
	});

	{
	const snapshot = await takeSnapshot();
	expect(snapshot).toStrictEqualTyped({
	loading: true,
	data: undefined,
	error: undefined,
	});
	}

	link.simulateResult({ result: { data: { totalLikes: 1 } } });

	{
	const snapshot = await takeSnapshot();
	expect(snapshot).toStrictEqualTyped({
	loading: false,
	data: { totalLikes: 1 },
	error: undefined,
	});
	}

	await expect(takeSnapshot).not.toRerender({ timeout: 20 });
	expect(onUnsubscribe).toHaveBeenCalledTimes(0);
	expect(onSubscribe).toHaveBeenCalledTimes(1);

	void rerender({ variables: { id: "2" } });

	await waitFor(() => expect(onUnsubscribe).toHaveBeenCalledTimes(1));
	expect(onSubscribe).toHaveBeenCalledTimes(2);
	expect(link.operation?.variables).toStrictEqual({ id: "2" });

	{
	const snapshot = await takeSnapshot();
	expect(snapshot).toStrictEqualTyped({
	loading: true,
	data: undefined,
	error: undefined,
	});
	}

	link.simulateResult({ result: { data: { totalLikes: 1000 } } });

	{
	const snapshot = await takeSnapshot();
	expect(snapshot).toStrictEqualTyped({
	loading: false,
	data: { totalLikes: 1000 },
	error: undefined,
	});
	}

	expect(onUnsubscribe).toHaveBeenCalledTimes(1);
	expect(onSubscribe).toHaveBeenCalledTimes(2);
	expect(link.operation?.variables).toStrictEqual({ id: "2" });

	getCurrentSnapshot().restart();

	await waitFor(() => expect(onUnsubscribe).toHaveBeenCalledTimes(2));
	expect(onSubscribe).toHaveBeenCalledTimes(3);
	expect(link.operation?.variables).toStrictEqual({ id: "2" });

	await expect(takeSnapshot).not.toRerender();

	link.simulateResult({ result: { data: { totalLikes: 1005 } } });

	{
	const snapshot = await takeSnapshot();
	expect(snapshot).toStrictEqualTyped({
	loading: false,
	data: { totalLikes: 1005 },
	error: undefined,
	});
	}
	});

	!observable \|\|
	((client !== observable.__.client \|\|
	subscription !== observable.__.query \|\|
	fetchPolicy !== observable.__.fetchPolicy \|\|
	errorPolicy !== observable.__.errorPolicy \|\|
	!equal(variables, observable.__.variables)) &&
	(typeof shouldResubscribe === "function" ?
	!!shouldResubscribe(options!)
	: shouldResubscribe) !== false)
	) {
	setObservable((observable = recreate()));
	}

		function withRestart(source: Observable<FetchResult>) {
		return new Observable<FetchResult>((observer) => {

-      function withRestart(source: Observable<FetchResult>) {
-        return new Observable<FetchResult>((observer) => {
+      /**
+       * Wraps an observable with restartable subscription logic.
+       * @param source The source observable to wrap.
+       * @param entry An object containing the restart function.
+       * @returns A new observable with restartable subscription logic.
+       */
+      function withRestart<T>(source: Observable<T>, entry: { restart?: () => void }): Observable<T> {
+        return new Observable<T>((observer) => {

Allow subscriptions to be deduplicated #12663

Allow subscriptions to be deduplicated #12663

Uh oh!

Conversation

jerelmiller commented May 30, 2025

Uh oh!

pkg-pr-new bot commented May 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

changeset-bot bot commented May 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

🦋 Changeset detected

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented May 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

size-limit report 📦

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

phryneas left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

pkg-pr-new bot commented May 30, 2025 •

edited

Loading

changeset-bot bot commented May 30, 2025 •

edited

Loading

github-actions bot commented May 30, 2025 •

edited

Loading