This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new e5f72388f fix(ai): isolate timeline pagination requests (#5589)
e5f72388f is described below
commit e5f72388f2829a646cac22200a7e026cd3e6e85e
Author: aias00 <[email protected]>
AuthorDate: Thu Oct 8 20:26:38 2026 -0700
fix(ai): isolate timeline pagination requests (#5589)
A single `requestId` ref was shared by `refetch` and `loadMore`, so one
"load more" silently
cancelled the full refresh that `useAgentRun` awaits in its finally block,
with nothing surfaced to
the user. Give the two paths their own ownership counters, let a complete
snapshot invalidate an
incremental page that started from the old cursor, and clear `loading` only
once both have settled.
---
.../pages/ai/hooks/useConversationTimeline.test.ts | 94 ++++++++++++++++++++++
web/src/pages/ai/hooks/useConversationTimeline.ts | 45 ++++++++---
2 files changed, 128 insertions(+), 11 deletions(-)
diff --git a/web/src/pages/ai/hooks/useConversationTimeline.test.ts
b/web/src/pages/ai/hooks/useConversationTimeline.test.ts
index c43dd09e0..eed5e2156 100644
--- a/web/src/pages/ai/hooks/useConversationTimeline.test.ts
+++ b/web/src/pages/ai/hooks/useConversationTimeline.test.ts
@@ -46,6 +46,14 @@ function page(
return { items, nextAfter, activeRun };
}
+function deferred<T>() {
+ let resolve: (value: T) => void = () => undefined;
+ const promise = new Promise<T>((release) => {
+ resolve = release;
+ });
+ return { promise, resolve };
+}
+
function render(conversationId: number | null = 7, limit?: number, maxPages?:
number) {
return renderHook(
({ id }: { id: number | null }) => useConversationTimeline(id, { limit,
maxPages }),
@@ -195,6 +203,92 @@ describe('useConversationTimeline', () => {
expect(result.current.items).toHaveLength(2);
});
+ it('loadMoreDoesNotCancelAnInFlightRefetchTest', async () => {
+ const refreshed = deferred<AiTimelineVO>();
+ const nextPage = deferred<AiTimelineVO>();
+ timelineMock
+ .mockResolvedValueOnce(page([item(1, { type: 'text', text: 'before' })],
1))
+ .mockReturnValueOnce(refreshed.promise)
+ .mockReturnValueOnce(nextPage.promise);
+
+ const { result } = render(7, 1, 1);
+ await waitFor(() => expect(result.current.hasMore).toBe(true));
+
+ let refetchPromise: Promise<void> = Promise.resolve();
+ act(() => {
+ refetchPromise = result.current.refetch();
+ });
+ await waitFor(() => expect(timelineMock).toHaveBeenCalledTimes(2));
+
+ let loadMorePromise: Promise<void> = Promise.resolve();
+ act(() => {
+ loadMorePromise = result.current.loadMore();
+ });
+ await waitFor(() => expect(timelineMock).toHaveBeenCalledTimes(3));
+
+ await act(async () => {
+ nextPage.resolve(page([item(2, { type: 'text', text: 'incremental' })],
null));
+ await loadMorePromise;
+ });
+ expect(result.current.items.map((row) => row.seq)).toEqual([1, 2]);
+ expect(result.current.loading).toBe(true);
+
+ await act(async () => {
+ refreshed.resolve(
+ page(
+ [item(1, { type: 'text', text: 'before' }), item(3, { type: 'text',
text: 'persisted' })],
+ null,
+ ),
+ );
+ await refetchPromise;
+ });
+
+ expect(result.current.items.map((row) => row.seq)).toEqual([1, 3]);
+ expect(result.current.loading).toBe(false);
+ });
+
+ it('refetchSupersedesAnOlderInFlightLoadMoreTest', async () => {
+ const nextPage = deferred<AiTimelineVO>();
+ const refreshed = deferred<AiTimelineVO>();
+ timelineMock
+ .mockResolvedValueOnce(page([item(1, { type: 'text', text: 'before' })],
1))
+ .mockReturnValueOnce(nextPage.promise)
+ .mockReturnValueOnce(refreshed.promise);
+
+ const { result } = render(7, 1, 1);
+ await waitFor(() => expect(result.current.hasMore).toBe(true));
+
+ let loadMorePromise: Promise<void> = Promise.resolve();
+ act(() => {
+ loadMorePromise = result.current.loadMore();
+ });
+ await waitFor(() => expect(timelineMock).toHaveBeenCalledTimes(2));
+
+ let refetchPromise: Promise<void> = Promise.resolve();
+ act(() => {
+ refetchPromise = result.current.refetch();
+ });
+ await waitFor(() => expect(timelineMock).toHaveBeenCalledTimes(3));
+
+ await act(async () => {
+ refreshed.resolve(
+ page(
+ [item(1, { type: 'text', text: 'before' }), item(3, { type: 'text',
text: 'persisted' })],
+ null,
+ ),
+ );
+ await refetchPromise;
+ });
+
+ await act(async () => {
+ nextPage.resolve(page([item(2, { type: 'text', text: 'stale incremental'
})], null));
+ await loadMorePromise;
+ });
+
+ expect(result.current.items.map((row) => row.seq)).toEqual([1, 3]);
+ expect(result.current.loading).toBe(false);
+ });
+
it('reportsTheServerMessageOnFailureTest', async () => {
timelineMock.mockRejectedValue(new Error('会话不存在'));
diff --git a/web/src/pages/ai/hooks/useConversationTimeline.ts
b/web/src/pages/ai/hooks/useConversationTimeline.ts
index 33ae3adb0..fc34091df 100644
--- a/web/src/pages/ai/hooks/useConversationTimeline.ts
+++ b/web/src/pages/ai/hooks/useConversationTimeline.ts
@@ -105,7 +105,9 @@ export function useConversationTimeline(
const [runSpeeds, setRunSpeeds] = useState<Map<number, number>>(new Map());
const [loading, setLoading] = useState(false);
const [error, setError] = useState('');
- const requestId = useRef(0);
+ const refetchRequestId = useRef(0);
+ const activeRefetchRef = useRef<number | null>(null);
+ const loadMoreRequestId = useRef(0);
/**
* Request id of the in-flight `loadMore`, or null. Guards `loadMore` only —
`refetch` must stay
* callable while another load is in flight, because `useAgentRun` awaits it
in a finally block.
@@ -116,6 +118,10 @@ export function useConversationTimeline(
const refetch = useCallback(async (): Promise<void> => {
if (conversationId === null) {
+ refetchRequestId.current += 1;
+ activeRefetchRef.current = null;
+ loadMoreRequestId.current += 1;
+ loadingMoreRef.current = null;
setItems([]);
setLoadedActiveRun(null);
setNextAfter(null);
@@ -124,7 +130,11 @@ export function useConversationTimeline(
return;
}
- const id = ++requestId.current;
+ const id = ++refetchRequestId.current;
+ activeRefetchRef.current = id;
+ // A newer full snapshot supersedes an incremental page that started from
the old cursor.
+ loadMoreRequestId.current += 1;
+ loadingMoreRef.current = null;
setLoading(true);
setError('');
try {
@@ -136,7 +146,7 @@ export function useConversationTimeline(
for (let page = 0; page < maxPages; page += 1) {
const result = await getConversationTimeline(conversationId, { after,
limit });
- if (id !== requestId.current) return;
+ if (id !== refetchRequestId.current) return;
collected = collected.concat(result.items);
run = result.activeRun;
cursor = result.nextAfter;
@@ -145,16 +155,23 @@ export function useConversationTimeline(
after = cursor;
}
+ // A load-more call may have started while this request was in flight.
The complete snapshot is
+ // authoritative, so its commit also invalidates that incremental write.
+ loadMoreRequestId.current += 1;
+ loadingMoreRef.current = null;
setItems(collected);
setLoadedActiveRun({ conversationId, run });
setNextAfter(cursor);
setRunSpeeds(speeds);
} catch (loadError) {
- if (id !== requestId.current) return;
+ if (id !== refetchRequestId.current) return;
setError(describeThrownMessage(loadError));
throw loadError;
} finally {
- if (id === requestId.current) setLoading(false);
+ if (activeRefetchRef.current === id) {
+ activeRefetchRef.current = null;
+ if (loadingMoreRef.current === null) setLoading(false);
+ }
}
}, [conversationId, limit, maxPages]);
@@ -162,13 +179,14 @@ export function useConversationTimeline(
if (conversationId === null || nextAfter === null ||
loadingMoreRef.current !== null) return;
const after = nextAfter;
- const id = ++requestId.current;
+ const id = ++loadMoreRequestId.current;
+ const refetchId = refetchRequestId.current;
loadingMoreRef.current = id;
setLoading(true);
setError('');
try {
const result = await getConversationTimeline(conversationId, { after,
limit });
- if (id !== requestId.current) return;
+ if (id !== loadMoreRequestId.current || refetchId !==
refetchRequestId.current) return;
setItems((previous) => previous.concat(result.items));
setLoadedActiveRun({ conversationId, run: result.activeRun });
setNextAfter(result.nextAfter);
@@ -178,11 +196,13 @@ export function useConversationTimeline(
return merged;
});
} catch (loadError) {
- if (id !== requestId.current) return;
+ if (id !== loadMoreRequestId.current || refetchId !==
refetchRequestId.current) return;
setError(describeThrownMessage(loadError));
} finally {
- if (loadingMoreRef.current === id) loadingMoreRef.current = null;
- if (id === requestId.current) setLoading(false);
+ if (loadingMoreRef.current === id) {
+ loadingMoreRef.current = null;
+ if (activeRefetchRef.current === null) setLoading(false);
+ }
}
}, [conversationId, limit, nextAfter]);
@@ -191,7 +211,10 @@ export function useConversationTimeline(
// eslint-disable-next-line react-hooks/set-state-in-effect
void refetch().catch(() => undefined);
return () => {
- requestId.current += 1;
+ refetchRequestId.current += 1;
+ activeRefetchRef.current = null;
+ loadMoreRequestId.current += 1;
+ loadingMoreRef.current = null;
};
}, [refetch]);
