Improve subscription performance by 10-20 times

packages/query-core/src/__tests__/mutationCache.test.tsx

@@ -416,7 +416,7 @@ describe('mutationCache', () => {
 
       unsubscribe()
 
-      await vi.advanceTimersByTimeAsync(10)
+      await vi.advanceTimersByTimeAsync(11)
       expect(queryClient.getMutationCache().getAll()).toHaveLength(0)
       expect(onSuccess).toHaveBeenCalledTimes(1)
     })

packages/query-core/src/gcManager.ts

@@ -0,0 +1,204 @@
+import { timeoutManager } from './timeoutManager'
+import type { Removable } from './removable'
+import type { ManagedTimerId } from './timeoutManager'
+
+/**
+ * Configuration for the GC manager
+ */
+export interface GCManagerConfig {
+  /**
+   * Force disable garbage collection.
+   * @default false
+   */
+  forceDisable?: boolean
+}
+
+/**
+ * Manages periodic garbage collection across all caches.
+ *
+ * Instead of each query/mutation having its own timeout,
+ * the GCManager runs a single interval that scans all
+ * registered caches for items eligible for removal.
+ *
+ * @example
+ * ```typescript
+ * // Register a cache for GC
+ * gcManager.registerCache(queryCache)
+ *
+ * // Start scanning
+ * gcManager.startScanning()
+ *
+ * // Change scan interval
+ * gcManager.setScanInterval(60000) // 1 minute
+ *
+ * // Stop scanning
+ * gcManager.stopScanning()
+ * ```
+ */
+export class GCManager {
+  #isScanning = false
+  #forceDisable = false
+  #eligibleItems = new Set<Removable>()
+  #scheduledScanTimeoutId: ManagedTimerId | null = null
+  #isScheduledScan = false
+
+  constructor(config: GCManagerConfig = {}) {
+    this.#forceDisable = config.forceDisable ?? false
+  }
+
+  #scheduleScan(): void {
+    if (this.#forceDisable || this.#isScheduledScan) {
+      return
+    }
+
+    this.#isScheduledScan = true
+
+    queueMicrotask(() => {
+      if (!this.#isScheduledScan) {
+        return
+      }
+
+      this.#isScheduledScan = false
+
+      let minTimeUntilGc = Infinity
+
+      for (const item of this.#eligibleItems) {
+        const timeUntilGc = getTimeUntilGc(item)
+
+        if (timeUntilGc < minTimeUntilGc) {
+          minTimeUntilGc = timeUntilGc
+        }
+      }
+
+      if (minTimeUntilGc === Infinity) {
+        return
+      }
+
+      if (this.#scheduledScanTimeoutId !== null) {
+        timeoutManager.clearTimeout(this.#scheduledScanTimeoutId)
+      }
+
+      this.#isScanning = true
+      this.#scheduledScanTimeoutId = timeoutManager.setTimeout(() => {
+        this.#isScanning = false
+        this.#scheduledScanTimeoutId = null
+
+        this.#performScan()
+
+        // If there are still eligible items, schedule the next scan
+        if (this.#eligibleItems.size > 0) {
+          this.#scheduleScan()
+        }
+      }, minTimeUntilGc)
+    })
+  }
+
+  /**
+   * Stop periodic scanning. Safe to call multiple times.
+   */
+  stopScanning(): void {
+    this.#isScanning = false
+    this.#isScheduledScan = false
+
+    if (this.#scheduledScanTimeoutId === null) {
+      return
+    }
+
+    timeoutManager.clearTimeout(this.#scheduledScanTimeoutId)
+
+    this.#scheduledScanTimeoutId = null
+  }
+
+  /**
+   * Check if scanning is active
+   */
+  isScanning(): boolean {
+    return this.#isScanning
+  }
+
+  /**
+   * Track an item that has been marked for garbage collection.
+   * Automatically starts scanning if not already running.
+   *
+   * @param item - The query or mutation marked for GC
+   */
+  trackEligibleItem(item: Removable): void {
+    if (this.#forceDisable) {
+      return
+    }
+
+    if (this.#eligibleItems.has(item)) {
+      return
+    }
+
+    this.#eligibleItems.add(item)
+
+    this.#scheduleScan()
+  }
+
+  /**
+   * Untrack an item that is no longer eligible for garbage collection.
+   * Automatically stops scanning if no items remain eligible.
+   *
+   * @param item - The query or mutation no longer eligible for GC
+   */
+  untrackEligibleItem(item: Removable): void {
+    if (this.#forceDisable) {
+      return
+    }
+
+    if (!this.#eligibleItems.has(item)) {
+      return
+    }
+
+    this.#eligibleItems.delete(item)
+
+    if (this.isScanning()) {
+      if (this.getEligibleItemCount() === 0) {
+        this.stopScanning()
+      } else {
+        this.#scheduleScan()
+      }
+    }
+  }
+
+  /**
+   * Get the number of items currently eligible for garbage collection.
+   */
+  getEligibleItemCount(): number {
+    return this.#eligibleItems.size
+  }
+
+  #performScan(): void {
+    // Iterate through all eligible items and attempt to collect them
+    for (const item of this.#eligibleItems) {
+      try {
+        if (item.isEligibleForGc()) {
+          const wasCollected = item.optionalRemove()
+
+          if (wasCollected) {
+            this.#eligibleItems.delete(item)
+          }
+        }
+      } catch (error) {
+        // Log but don't throw - one cache error shouldn't stop others
+        if (process.env.NODE_ENV !== 'production') {
+          console.error('[GCManager] Error during garbage collection:', error)
+        }
+      }
+    }
+  }
+
+  clear(): void {
+    this.#eligibleItems.clear()
+    this.stopScanning()
+  }
+}
+
+function getTimeUntilGc(item: Removable): number {
+  const gcAt = item.getGcAtTimestamp()
+  if (gcAt === null) {
+    return Infinity
+  }
+  return Math.max(0, gcAt - Date.now())
+}

packages/query-core/src/mutation.ts

@@ -12,6 +12,7 @@ import type { MutationCache } from './mutationCache'
 import type { MutationObserver } from './mutationObserver'
 import type { Retryer } from './retryer'
 import type { QueryClient } from './queryClient'
+import type { GCManager } from './gcManager'
 
 // TYPES
 
@@ -108,9 +109,8 @@ export class Mutation<
     this.#mutationCache = config.mutationCache
     this.#observers = []
     this.state = config.state || getDefaultState()
-
     this.setOptions(config.options)
-    this.scheduleGc()
+    this.markForGc()
   }
 
   setOptions(
@@ -125,12 +125,16 @@ export class Mutation<
     return this.options.meta
   }
 
+  protected getGcManager(): GCManager {
+    return this.#client.getGcManager()
+  }
+
   addObserver(observer: MutationObserver<any, any, any, any>): void {
     if (!this.#observers.includes(observer)) {
       this.#observers.push(observer)
 
       // Stop the mutation from being garbage collected
-      this.clearGcTimeout()
+      this.clearGcMark()
 
       this.#mutationCache.notify({
         type: 'observerAdded',
@@ -143,7 +147,9 @@ export class Mutation<
   removeObserver(observer: MutationObserver<any, any, any, any>): void {
     this.#observers = this.#observers.filter((x) => x !== observer)
 
-    this.scheduleGc()
+    if (this.isSafeToRemove()) {
+      this.markForGc()
+    }
 
     this.#mutationCache.notify({
       type: 'observerRemoved',
@@ -152,14 +158,21 @@ export class Mutation<
     })
   }
 
-  protected optionalRemove() {
+  private isSafeToRemove(): boolean {
+    return this.state.status !== 'pending' && this.#observers.length === 0
+  }
+
+  optionalRemove(): boolean {
     if (!this.#observers.length) {
       if (this.state.status === 'pending') {
-        this.scheduleGc()
+        this.markForGc()
       } else {
         this.#mutationCache.remove(this)
+        return true
       }
     }
+
+    return false
   }
 
   continue(): Promise<unknown> {
@@ -370,6 +383,10 @@ export class Mutation<
     }
     this.state = reducer(this.state)
 
+    if (this.isSafeToRemove()) {
+      this.markForGc()
+    }
+
     notifyManager.batch(() => {
       this.#observers.forEach((observer) => {
         observer.onMutationUpdate(action)