feat(query-core): expose fetch source attribution 🤖🤖🤖

.changeset/query-core-fetch-attribution.md

@@ -0,0 +1,12 @@
+---
+'@tanstack/query-core': minor
+'@tanstack/query-broadcast-client-experimental': patch
+---
+
+Expose cache-wide fetch acceptance and tracked query-function invocation positions. Integrations can reserve positions for direct writes and read operation-local positions from cache events, query-function contexts, and observer results. `fetchSourceStarted` reports an invocation even when a custom persister or behavior ignores its result. Cached data has a `dataSourceSequence` only when Query Core controls the path from query function to result; it is the fetch's first invocation and a lower bound when the fetch retries. Opaque custom persisters and replaced fetch functions leave it undefined.
+
+If behavior setup invokes `queryFn` and then fails, the source event carries a rejected setup-failure promise without a `fetchStarted` event, because no fetch was installed. The Query publishes a terminal error to observers and QueryCache callbacks, except for silent or reverting cancellations. Invalidation and state writes during setup still allow that error; data written during setup remains cached. Reset, a newer fetch, or cache removal retires the failed attempt. Quiet cancellations settle the fetching state, and a reverted cancellation with cached data resolves with that data. A silently cancelled fetch follows the full promise of the attempt that directly replaced it, including setup and result-validation failures. Query Core observes abandoned eager query-function promises before aborting them, while the returned promises remain unchanged for the behavior. Immutable behavior contexts still allow query-function invocation and source attribution.
+
+Ordinary `context.options` writes reach the Query's options, and stable `queryFn` reads through writable data properties or configurable accessors return a tracked wrapper. Reflective reads of frozen or nonconfigurable descriptors, and accessors that replace their own descriptor while being read, are outside that attribution guarantee. A function captured outside that context can run without a source position. Reset or cache removal during option initial-data publication or behavior setup retires the fetch before retryer installation: its promise rejects, its signal aborts, and `fetchCancelled` is reported without `fetchStarted`. A previously invoked eager source remains observable with a rejected promise. `Query.fetch` continues returning a promise if a cancellation listener throws. Cache removal announces the old Query before a cancellation listener can add a replacement under the same key.
+
+The broadcast client uses the observer state recorded when removal begins, so an earlier listener detaching the last observer cannot suppress the outgoing removal. It replays a replacement created by an earlier removal listener. A replacement created while receiving a removal is broadcast to other clients.

packages/angular-query-experimental/src/__tests__/inject-query.test.ts

@@ -208,9 +208,11 @@ describe('injectQuery', () => {
     // should call queryFn with context containing the new queryKey
     expect(queryFn).toHaveBeenNthCalledWith(2, {
       client: queryClient,
+      fetchSequence: expect.any(Number),
       meta: undefined,
       queryKey: key2,
       signal: expect.anything(),
+      sourceSequence: expect.any(Number),
     })
   })
 

packages/preact-query/src/__tests__/useInfiniteQuery.test.tsx

@@ -69,6 +69,9 @@ describe('useInfiniteQuery', () => {
     expect(states[0]).toEqual({
       data: undefined,
       dataUpdatedAt: 0,
+      dataUpdateCount: 0,
+      dataFetchSequence: undefined,
+      dataSourceSequence: undefined,
       error: null,
       errorUpdatedAt: 0,
       failureCount: 0,
@@ -104,6 +107,9 @@ describe('useInfiniteQuery', () => {
     expect(states[1]).toEqual({
       data: { pages: [0], pageParams: [0] },
       dataUpdatedAt: expect.any(Number),
+      dataUpdateCount: 1,
+      dataFetchSequence: expect.any(Number),
+      dataSourceSequence: expect.any(Number),
       error: null,
       errorUpdatedAt: 0,
       failureCount: 0,

packages/preact-query/src/__tests__/useQuery.test.tsx

@@ -109,6 +109,9 @@ describe('useQuery', () => {
     expect(states[0]).toEqual({
       data: undefined,
       dataUpdatedAt: 0,
+      dataUpdateCount: 0,
+      dataFetchSequence: undefined,
+      dataSourceSequence: undefined,
       error: null,
       errorUpdatedAt: 0,
       failureCount: 0,
@@ -137,6 +140,9 @@ describe('useQuery', () => {
     expect(states[1]).toEqual({
       data: 'test',
       dataUpdatedAt: expect.any(Number),
+      dataUpdateCount: 1,
+      dataFetchSequence: expect.any(Number),
+      dataSourceSequence: expect.any(Number),
       error: null,
       errorUpdatedAt: 0,
       failureCount: 0,
@@ -197,6 +203,9 @@ describe('useQuery', () => {
     expect(states[0]).toEqual({
       data: undefined,
       dataUpdatedAt: 0,
+      dataUpdateCount: 0,
+      dataFetchSequence: undefined,
+      dataSourceSequence: undefined,
       error: null,
       errorUpdatedAt: 0,
       failureCount: 0,
@@ -225,6 +234,9 @@ describe('useQuery', () => {
     expect(states[1]).toEqual({
       data: undefined,
       dataUpdatedAt: 0,
+      dataUpdateCount: 0,
+      dataFetchSequence: undefined,
+      dataSourceSequence: undefined,
       error: null,
       errorUpdatedAt: 0,
       failureCount: 1,
@@ -253,6 +265,9 @@ describe('useQuery', () => {
     expect(states[2]).toEqual({
       data: undefined,
       dataUpdatedAt: 0,
+      dataUpdateCount: 0,
+      dataFetchSequence: undefined,
+      dataSourceSequence: undefined,
       error: new Error('rejected #2'),
       errorUpdatedAt: expect.any(Number),
       failureCount: 2,

packages/query-broadcast-client-experimental/src/__tests__/index.test.ts

@@ -1,4 +1,4 @@
-import { QueryClient } from '@tanstack/query-core'
+import { QueryClient, QueryObserver } from '@tanstack/query-core'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { queryKey, sleep } from '@tanstack/query-test-utils'
 import { broadcastQueryClient } from '..'
@@ -8,6 +8,7 @@ import type { QueryCache } from '@tanstack/query-core'
 const mockPostMessage = vi.fn().mockResolvedValue(undefined)
 const mockClose = vi.fn()
 let lastCreatedChannel: { onmessage: ((action: any) => void) | null }
+const createdChannels: Array<{ onmessage: ((action: any) => void) | null }> = []
 
 vi.mock('broadcast-channel', async (importOriginal) => {
   const actual = await importOriginal()
@@ -19,6 +20,7 @@ vi.mock('broadcast-channel', async (importOriginal) => {
       close = mockClose
       constructor() {
         lastCreatedChannel = this
+        createdChannels.push(this)
       }
     },
   }
@@ -31,6 +33,7 @@ describe('broadcastQueryClient', () => {
   beforeEach(() => {
     queryClient = new QueryClient()
     queryCache = queryClient.getQueryCache()
+    createdChannels.length = 0
     mockPostMessage.mockResolvedValue(undefined)
     mockClose.mockReset()
   })
@@ -52,6 +55,218 @@ describe('broadcastQueryClient', () => {
     expect(queryCache.hasListeners()).toBe(false)
   })
 
+  // A removal cancellation listener may create a new Query with the same
+  // key. Broadcast recipients replay by hash, so they must see the old
+  // removal before the replacement addition.
+  it('keeps a same-key replacement on another client after removal reentry', () => {
+    const otherClient = new QueryClient()
+    const key = ['broadcast-removal-reentry']
+    otherClient.setQueryData(key, 'old peer data')
+    const oldPeerQuery = otherClient.getQueryCache().find({ queryKey: key })
+    expect(oldPeerQuery).toBeDefined()
+    queryClient.setQueryData(key, 'seed')
+    const oldQuery = queryCache.find<string>({ queryKey: key })
+    expect(oldQuery).toBeDefined()
+    const observer = new QueryObserver(queryClient, {
+      queryKey: key,
+      queryFn: () => 'unused',
+      enabled: false,
+    })
+    const stopObserver = observer.subscribe(() => {})
+    const oldFetch = oldQuery!.fetch({
+      queryKey: key,
+      queryFn: () => new Promise<string>(() => {}),
+    })
+    void oldFetch.catch(() => {})
+    const stopFirst = broadcastQueryClient({ queryClient })
+    const stopOther = broadcastQueryClient({ queryClient: otherClient })
+    const stopReplacement = queryCache.subscribe((event) => {
+      if (event.type === 'fetchCancelled' && event.query === oldQuery) {
+        void queryClient
+          .query({
+            queryKey: key,
+            queryFn: () => new Promise<string>(() => {}),
+          })
+          .catch(() => {})
+      }
+    })
+    mockPostMessage.mockClear()
+
+    queryCache.remove(oldQuery!)
+    expect(queryCache.find({ queryKey: key })).toBeDefined()
+    const messages = mockPostMessage.mock.calls.map(([message]) => message)
+    for (const message of messages) {
+      createdChannels[1]?.onmessage?.(message)
+    }
+    const replacement = otherClient.getQueryCache().find({ queryKey: key })
+    expect(replacement).toBeDefined()
+    expect(replacement).not.toBe(oldPeerQuery)
+    expect(replacement?.state.data).toBeUndefined()
+    expect(messages.map((message) => message.type)).toEqual([
+      'removed',
+      'added',
+    ])
+
+    stopReplacement()
+    stopFirst()
+    stopOther()
+    stopObserver()
+    queryClient.clear()
+    otherClient.clear()
+  })
+
+  // A cache listener registered before broadcasting can add a replacement
+  // while the old removal is still being delivered. The remote cache must
+  // converge to the source even when it hears that addition first.
+  it.each(['kept', 'detached'] as const)(
+    'replays the current replacement after a removal listener adds it with old observer %s',
+    (observerState) => {
+      const otherClient = new QueryClient()
+      const key = ['broadcast-removed-listener-reentry']
+      queryClient.setQueryData(key, 'old source data')
+      otherClient.setQueryData(key, 'old peer data')
+      const oldSourceQuery = queryCache.find({ queryKey: key })
+      const oldPeerQuery = otherClient.getQueryCache().find({ queryKey: key })
+      expect(oldSourceQuery).toBeDefined()
+      expect(oldPeerQuery).toBeDefined()
+      const observer = new QueryObserver(queryClient, {
+        queryKey: key,
+        queryFn: () => 'unused',
+        enabled: false,
+      })
+      const stopObserver = observer.subscribe(() => {})
+      const stopReplacement = queryCache.subscribe((event) => {
+        if (event.type === 'removed' && event.query === oldSourceQuery) {
+          if (observerState === 'detached') stopObserver()
+          queryCache.build(queryClient, {
+            queryKey: key,
+            queryFn: () => 'new source data',
+          })
+        }
+      })
+      const stopFirst = broadcastQueryClient({ queryClient })
+      const stopOther = broadcastQueryClient({ queryClient: otherClient })
+      mockPostMessage.mockClear()
+
+      queryCache.remove(oldSourceQuery!)
+      const sourceReplacement = queryCache.find({ queryKey: key })
+      expect(sourceReplacement).toBeDefined()
+      expect(sourceReplacement).not.toBe(oldSourceQuery)
+      const messages = mockPostMessage.mock.calls.map(([message]) => message)
+      for (const message of messages) {
+        createdChannels[1]?.onmessage?.(message)
+      }
+      const peerReplacement = otherClient
+        .getQueryCache()
+        .find({ queryKey: key })
+      expect(peerReplacement).toBeDefined()
+      expect(peerReplacement).not.toBe(oldPeerQuery)
+      expect(peerReplacement?.state.data).toBe(sourceReplacement?.state.data)
+
+      stopReplacement()
+      stopFirst()
+      stopOther()
+      stopObserver()
+      queryClient.clear()
+      otherClient.clear()
+    },
+  )
+
+  // Broadcast uses the observer state at the start of removal. A cache
+  // listener can detach the final observer before broadcast receives the
+  // event, but the peer must still retire its old query.
+  it('broadcasts an observed removal when an earlier listener detaches its observer', () => {
+    const otherClient = new QueryClient()
+    const key = ['broadcast-removal-observer-detach']
+    queryClient.setQueryData(key, 'old source data')
+    otherClient.setQueryData(key, 'old peer data')
+    const oldSourceQuery = queryCache.find({ queryKey: key })
+    const oldPeerQuery = otherClient.getQueryCache().find({ queryKey: key })
+    expect(oldSourceQuery).toBeDefined()
+    expect(oldPeerQuery).toBeDefined()
+    const observer = new QueryObserver(queryClient, {
+      queryKey: key,
+      queryFn: () => 'unused',
+      enabled: false,
+    })
+    const stopObserver = observer.subscribe(() => {})
+    const stopDetachingListener = queryCache.subscribe((event) => {
+      if (event.type === 'removed' && event.query === oldSourceQuery) {
+        stopObserver()
+      }
+    })
+    const stopSource = broadcastQueryClient({ queryClient })
+    const stopOther = broadcastQueryClient({ queryClient: otherClient })
+    mockPostMessage.mockClear()
+
+    queryCache.remove(oldSourceQuery!)
+    const messages = mockPostMessage.mock.calls.map(([message]) => message)
+    expect(messages.map((message) => message.type)).toEqual(['removed'])
+    for (const message of messages) {
+      createdChannels[1]?.onmessage?.(message)
+    }
+    expect(otherClient.getQueryCache().find({ queryKey: key })).toBeUndefined()
+
+    stopDetachingListener()
+    stopSource()
+    stopOther()
+    stopObserver()
+    queryClient.clear()
+    otherClient.clear()
+  })
+
+  // A received removal can also trigger a local cache listener. That local
+  // replacement must travel back to the sender after the incoming removal is
+  // applied, while the incoming removal itself is not echoed.
+  it('broadcasts a local replacement created while receiving a removal', () => {
+    const otherClient = new QueryClient()
+    const otherCache = otherClient.getQueryCache()
+    const key = ['broadcast-receiver-removal-reentry']
+    queryClient.setQueryData(key, 'old source data')
+    otherClient.setQueryData(key, 'old peer data')
+    const oldSourceQuery = queryCache.find({ queryKey: key })
+    const oldPeerQuery = otherCache.find({ queryKey: key })
+    expect(oldSourceQuery).toBeDefined()
+    expect(oldPeerQuery).toBeDefined()
+    const observer = new QueryObserver(queryClient, {
+      queryKey: key,
+      queryFn: () => 'unused',
+      enabled: false,
+    })
+    const stopObserver = observer.subscribe(() => {})
+    const stopSource = broadcastQueryClient({ queryClient })
+    const stopReplacement = otherCache.subscribe((event) => {
+      if (event.type === 'removed' && event.query === oldPeerQuery) {
+        otherCache.build(otherClient, {
+          queryKey: key,
+          queryFn: () => 'new peer data',
+        })
+      }
+    })
+    const stopOther = broadcastQueryClient({ queryClient: otherClient })
+    mockPostMessage.mockClear()
+
+    queryCache.remove(oldSourceQuery!)
+    const removal = mockPostMessage.mock.calls.map(([message]) => message)
+    expect(removal.map((message) => message.type)).toEqual(['removed'])
+    mockPostMessage.mockClear()
+    createdChannels[1]?.onmessage?.(removal[0])
+    const peerReplacement = otherCache.find({ queryKey: key })
+    expect(peerReplacement).toBeDefined()
+    expect(peerReplacement).not.toBe(oldPeerQuery)
+    const outgoing = mockPostMessage.mock.calls.map(([message]) => message)
+    expect(outgoing.map((message) => message.type)).toEqual(['added'])
+    createdChannels[0]?.onmessage?.(outgoing[0])
+    expect(queryCache.find({ queryKey: key })).toBeDefined()
+
+    stopSource()
+    stopReplacement()
+    stopOther()
+    stopObserver()
+    queryClient.clear()
+    otherClient.clear()
+  })
+
   describe('incoming message handling', () => {
     it('should keep broadcasting local changes after applying an incoming message throws', () => {
       const remoteKey = queryKey()