fix(query-core): prevent reducer from being called twice in streamedQuery (#9970)

* fix(query-core): prevent reducer from being called twice in streamedQuery #9787

- Fix bug where reducer was called twice for each chunk (once in setQueryData, once in loop)
- Store reducer result in variable to avoid duplicate calls
- Add test case to verify reducer is only called once per chunk

* fix(streamed-query): prevent reducer from being called twice

Fixed an issue where the reducer was being invoked twice when refetchMode is set to 'replace'.
Added a regression test to verify the fix.

Fixes #9787

* docs: add changeset and lockfile update

* fix(changeset): change version bump to patch

* Update streamedQuery.test.tsx

fix: lint

* fix: support refetchMode append and prevent double reducer calls

- Fix refetchMode append to preserve existing data on refetch
- Fix reducer being called twice by calling it once and storing result
- Update pnpm-lock.yaml to match main branch

* chore: update pnpm-lock.yaml to include size-limit dependencies

* ci: apply automated fixes

* refactor: simplify streamedQuery logic with clearer isReplaceRefetch condition

- Extract isReplaceRefetch variable for better readability
- Use explicit if-else instead of negated conditions
- Make it clearer when reducer is called in each branch

---------

Co-authored-by: TkDodo <[email protected]>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: 3 people

.changeset/fair-peaches-deny.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/query-core': patch
+---
+
+Fix streamedQuery reducer being called twice

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

@@ -538,4 +538,41 @@ describe('streamedQuery', () => {
 
     unsubscribe()
   })
+
+  test('should not call reducer twice when refetchMode is replace', async () => {
+    const key = queryKey()
+    const arr: Array<number> = []
+
+    const observer = new QueryObserver(queryClient, {
+      queryKey: key,
+      queryFn: streamedQuery({
+        streamFn: async function* () {
+          const v = [1, 2, 3]
+          yield* v
+        },
+        reducer: (oldStream, newChunk) => {
+          arr.push(newChunk)
+          return [...oldStream, newChunk]
+        },
+        initialValue: [] as Array<number>,
+        refetchMode: 'replace',
+      }),
+    })
+
+    const unsubscribe = observer.subscribe(vi.fn())
+
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(arr).toEqual([1, 2, 3])
+    expect(observer.getCurrentResult().data).toEqual([1, 2, 3])
+
+    void observer.refetch()
+
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(arr).toEqual([1, 2, 3, 1, 2, 3])
+    expect(observer.getCurrentResult().data).toEqual([1, 2, 3])
+
+    unsubscribe()
+  })
 })

packages/query-core/src/streamedQuery.ts

@@ -95,22 +95,25 @@ export function streamedQuery<
 
     const stream = await streamFn(streamFnContext)
 
+    const isReplaceRefetch = isRefetch && refetchMode === 'replace'
+
     for await (const chunk of stream) {
       if (cancelled) {
         break
       }
 
-      // don't append to the cache directly when replace-refetching
-      if (!isRefetch || refetchMode !== 'replace') {
+      if (isReplaceRefetch) {
+        // don't append to the cache directly when replace-refetching
+        result = reducer(result, chunk)
+      } else {
         context.client.setQueryData<TData>(context.queryKey, (prev) =>
           reducer(prev === undefined ? initialValue : prev, chunk),
         )
       }
-      result = reducer(result, chunk)
     }
 
     // finalize result: replace-refetching needs to write to the cache
-    if (isRefetch && refetchMode === 'replace' && !cancelled) {
+    if (isReplaceRefetch && !cancelled) {
       context.client.setQueryData<TData>(context.queryKey, result)
     }