github/InvokeAI
1
1
Fork 0
mirror of https://github.com/invoke-ai/InvokeAI.git synced 2026-04-23 03:00:31 -04:00
Code Issues Packages Projects Releases Wiki Activity

fix(ui): queue tab list of queue items

Browse Source 
Reverted incomplete change to how queue items are listed. In the future
I think we should redo it to work like the gallery. For now, it is back
the way it was in v5.
This commit is contained in:
psychedelicious
2025-07-08 14:21:25 +10:00
committed by Kent Keirsey
parent db6af134b7
commit b1e2cb8401
9 changed files with 266 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

204
invokeai/frontend/web/src/services/api/endpoints/queue.ts
View File

@@ -1,4 +1,8 @@
import type { EntityState, ThunkDispatch, UnknownAction } from '@reduxjs/toolkit';
import { createEntityAdapter } from '@reduxjs/toolkit';
import { getSelectorsOptions } from 'app/store/createMemoizedSelector';
import { $queueId } from 'app/store/nanostores/queueId';
import { listParamsReset } from 'features/queue/store/queueSlice';
import queryString from 'query-string';
import type { components, paths } from 'services/api/schema';
@@ -31,6 +35,30 @@ export type SessionQueueItemStatus = NonNullable<
NonNullable<paths['/api/v1/queue/{queue_id}/list']['get']['parameters']['query']>['status']
>;
export const queueItemsAdapter = createEntityAdapter<components['schemas']['SessionQueueItem'], string>({
selectId: (queueItem) => String(queueItem.item_id),
sortComparer: (a, b) => {
// Sort by priority in descending order
if (a.priority > b.priority) {
return -1;
}
if (a.priority < b.priority) {
return 1;
}
// If priority is the same, sort by id in ascending order
if (a.item_id < b.item_id) {
return -1;
}
if (a.item_id > b.item_id) {
return 1;
}
return 0;
},
});
export const queueItemsAdapterSelectors = queueItemsAdapter.getSelectors(undefined, getSelectorsOptions);
export const queueApi = api.injectEndpoints({
endpoints: (build) => ({
enqueueBatch: build.mutation<
@@ -50,6 +78,57 @@ export const queueApi = api.injectEndpoints({
{ type: 'SessionQueueItem', id: LIST_TAG },
{ type: 'SessionQueueItem', id: LIST_ALL_TAG },
],
onQueryStarted: async (arg, api) => {
const { dispatch, queryFulfilled } = api;
try {
const { data } = await queryFulfilled;
resetListQueryData(dispatch);
/**
* When a batch is enqueued, we need to update the queue status. While it might be templting to invalidate the
* `SessionQueueStatus` tag here, this can introduce a race condition when the queue item executes quickly:
*
* - Enqueue via this query
* - On success, we invalidate `SessionQueueStatus` tag - network request sent to server
* - The server gets the queue status request and responds, but this takes some time... in the meantime:
* - The new queue item starts executing, and we receive a socket queue item status changed event
* - We optimistically update the queue status in the queue item status changed socket handler
* - At this point, the queue status is correct
* - Finally, we get the queue status from the tag invalidation request - but it's reporting the queue status
* from _before_ the last queue event
* - The queue status is now incorrect!
*
* Ok, what if we just never did optimistic updates and invalidated the tag in the queue event handlers instead?
* It's much simpler that way, but it causes a lot of network requests - 3 per queue item, as it moves from
* pending -> in_progress -> completed/failed/canceled.
*
* We can do a bit of extra work here, incrementing the pending and total counts in the queue status, and do
* similar optimistic updates in the socket handler. Because this optimistic update runs immediately after the
* enqueue network request, it should always occur _before_ the next queue event, so no race condition:
*
* - Enqueue batch via this query
* - On success, optimistically update - this happens immediately on the HTTP OK - before the next queue event
* - At this point, the queue status is correct
* - A queue item status changes and we receive a socket event w/ updated status
* - Update status optimistically in socket handler
* - Queue status is still correct
*
* This problem occurs most commonly with canvas filters like Canny edge detection, which are single-node
* graphs that execute very quickly. Image generation graphs take long enough to not trigger this race
* condition - even when all nodes are cached on the server.
*/
dispatch(
queueApi.util.updateQueryData('getQueueStatus', undefined, (draft) => {
if (!draft) {
return;
}
draft.queue.pending += data.enqueued;
draft.queue.total += data.enqueued;
})
);
} catch {
// no-op
}
},
}),
resumeProcessor: build.mutation<
paths['/api/v1/queue/{queue_id}/processor/resume']['put']['responses']['200']['content']['application/json'],
@@ -85,6 +164,15 @@ export const queueApi = api.injectEndpoints({
{ type: 'SessionQueueItem', id: LIST_TAG },
{ type: 'SessionQueueItem', id: LIST_ALL_TAG },
],
onQueryStarted: async (arg, api) => {
const { dispatch, queryFulfilled } = api;
try {
await queryFulfilled;
resetListQueryData(dispatch);
} catch {
// no-op
}
},
}),
clearQueue: build.mutation<
paths['/api/v1/queue/{queue_id}/clear']['put']['responses']['200']['content']['application/json'],
@@ -104,6 +192,15 @@ export const queueApi = api.injectEndpoints({
{ type: 'SessionQueueItem', id: LIST_TAG },
{ type: 'SessionQueueItem', id: LIST_ALL_TAG },
],
onQueryStarted: async (arg, api) => {
const { dispatch, queryFulfilled } = api;
try {
await queryFulfilled;
resetListQueryData(dispatch);
} catch {
// no-op
}
},
}),
getCurrentQueueItem: build.query<
paths['/api/v1/queue/{queue_id}/current']['get']['responses']['200']['content']['application/json'],
@@ -187,6 +284,25 @@ export const queueApi = api.injectEndpoints({
url: buildQueueUrl(`i/${item_id}/cancel`),
method: 'PUT',
}),
onQueryStarted: async (item_id, { dispatch, queryFulfilled }) => {
try {
const { data } = await queryFulfilled;
dispatch(
queueApi.util.updateQueryData('listQueueItems', undefined, (draft) => {
queueItemsAdapter.updateOne(draft, {
id: String(item_id),
changes: {
status: data.status,
completed_at: data.completed_at,
updated_at: data.updated_at,
},
});
})
);
} catch {
// no-op
}
},
invalidatesTags: (result) => {
if (!result) {
return [];
@@ -210,6 +326,15 @@ export const queueApi = api.injectEndpoints({
method: 'PUT',
body,
}),
onQueryStarted: async (arg, api) => {
const { dispatch, queryFulfilled } = api;
try {
await queryFulfilled;
resetListQueryData(dispatch);
} catch {
// no-op
}
},
invalidatesTags: (result, error, { batch_ids }) => {
if (!result) {
return [];
@@ -256,16 +381,6 @@ export const queueApi = api.injectEndpoints({
}),
invalidatesTags: ['SessionQueueStatus', 'BatchStatus', 'QueueCountsByDestination', 'SessionQueueItem'],
}),
deleteAllExceptCurrent: build.mutation<
paths['/api/v1/queue/{queue_id}/delete_all_except_current']['put']['responses']['200']['content']['application/json'],
void
>({
query: () => ({
url: buildQueueUrl('delete_all_except_current'),
method: 'PUT',
}),
invalidatesTags: ['SessionQueueStatus', 'BatchStatus', 'QueueCountsByDestination', 'SessionQueueItem'],
}),
retryItemsById: build.mutation<
paths['/api/v1/queue/{queue_id}/retry_items_by_id']['put']['responses']['200']['content']['application/json'],
paths['/api/v1/queue/{queue_id}/retry_items_by_id']['put']['requestBody']['content']['application/json']
@@ -275,6 +390,15 @@ export const queueApi = api.injectEndpoints({
method: 'PUT',
body,
}),
onQueryStarted: async (arg, api) => {
const { dispatch, queryFulfilled } = api;
try {
await queryFulfilled;
resetListQueryData(dispatch);
} catch {
// no-op
}
},
invalidatesTags: (result, error, item_ids) => {
if (!result) {
return [];
@@ -290,24 +414,31 @@ export const queueApi = api.injectEndpoints({
},
}),
listQueueItems: build.query<
components['schemas']['CursorPaginatedResults_SessionQueueItem_'],
{ cursor?: number; priority?: number; destination?: string } | undefined
EntityState<components['schemas']['SessionQueueItem'], string> & {
has_more: boolean;
},
{ cursor?: number; priority?: number } | undefined
>({
query: (queryArgs) => ({
url: getListQueueItemsUrl(queryArgs),
method: 'GET',
}),
keepUnusedDataFor: 60 * 5, // 5 minutes
providesTags: (result, _error, _args) => {
if (!result) {
return [];
}
return [
'FetchOnReconnect',
{ type: 'SessionQueueItem', id: LIST_TAG },
...result.items.map(({ item_id }) => ({ type: 'SessionQueueItem', id: item_id }) satisfies ApiTagDescription),
];
serializeQueryArgs: () => {
return buildQueueUrl('list');
},
transformResponse: (response: components['schemas']['CursorPaginatedResults_SessionQueueItem_']) =>
queueItemsAdapter.addMany(
queueItemsAdapter.getInitialState({
has_more: response.has_more,
}),
response.items
),
merge: (cache, response) => {
queueItemsAdapter.addMany(cache, queueItemsAdapterSelectors.selectAll(response));
cache.has_more = response.has_more;
},
forceRefetch: ({ currentArg, previousArg }) => currentArg !== previousArg,
keepUnusedDataFor: 60 * 5, // 5 minutes
}),
listAllQueueItems: build.query<
paths['/api/v1/queue/{queue_id}/list_all']['get']['responses']['200']['content']['application/json'],
@@ -356,6 +487,16 @@ export const queueApi = api.injectEndpoints({
{ type: 'SessionQueueItem', id: LIST_ALL_TAG },
],
}),
deleteAllExceptCurrent: build.mutation<
paths['/api/v1/queue/{queue_id}/delete_all_except_current']['put']['responses']['200']['content']['application/json'],
void
>({
query: () => ({
url: buildQueueUrl('delete_all_except_current'),
method: 'PUT',
}),
invalidatesTags: ['SessionQueueStatus', 'BatchStatus', 'QueueCountsByDestination', 'SessionQueueItem'],
}),
getQueueCountsByDestination: build.query<
paths['/api/v1/queue/{queue_id}/counts_by_destination']['get']['responses']['200']['content']['application/json'],
paths['/api/v1/queue/{queue_id}/counts_by_destination']['get']['parameters']['query']
@@ -378,6 +519,7 @@ export const {
useClearQueueMutation,
usePruneQueueMutation,
useGetQueueStatusQuery,
useGetQueueItemQuery,
useListQueueItemsQuery,
useCancelQueueItemMutation,
useCancelQueueItemsByDestinationMutation,
@@ -392,6 +534,24 @@ export const {
export const selectQueueStatus = queueApi.endpoints.getQueueStatus.select();
const resetListQueryData = (
// eslint-disable-next-line @typescript-eslint/no-explicit-any
dispatch: ThunkDispatch<any, any, UnknownAction>
) => {
dispatch(
queueApi.util.updateQueryData('listQueueItems', undefined, (draft) => {
// remove all items from the list
queueItemsAdapter.removeAll(draft);
// reset the has_more flag
draft.has_more = false;
})
);
// set the list cursor and priority to undefined
dispatch(listParamsReset());
// we have to manually kick off another query to get the first page and re-initialize the list
dispatch(queueApi.endpoints.listQueueItems.initiate(undefined));
};
export const enqueueMutationFixedCacheKeyOptions = {
fixedCacheKey: 'enqueueBatch',
} as const;