mirror of
https://github.com/Comfy-Org/ComfyUI_frontend.git
synced 2026-07-18 17:58:23 +00:00
Compare commits
1 Commits
matt/be-34
...
synap5e/fi
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
86121fb417 |
@@ -245,7 +245,7 @@ const focusAssetInSidebar = async (item: JobListItem) => {
|
||||
const assetId = String(jobId)
|
||||
openAssetsSidebar()
|
||||
await nextTick()
|
||||
await assetsStore.refreshHistoryHead()
|
||||
await assetsStore.updateHistory()
|
||||
const asset = assetsStore.historyAssets.find(
|
||||
(existingAsset) => existingAsset.id === assetId
|
||||
)
|
||||
|
||||
@@ -255,13 +255,13 @@ describe('resolveMissingMediaAssetSources', () => {
|
||||
1,
|
||||
expect.any(Function),
|
||||
200,
|
||||
{ offset: 0 }
|
||||
0
|
||||
)
|
||||
expect(mockFetchHistoryPage).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
expect.any(Function),
|
||||
200,
|
||||
{ offset: 200 }
|
||||
200
|
||||
)
|
||||
})
|
||||
|
||||
|
||||
@@ -176,7 +176,7 @@ async function fetchGeneratedHistoryAssets(
|
||||
const historyPage = await fetchHistoryPage(
|
||||
api.fetchApi.bind(api),
|
||||
HISTORY_MEDIA_ASSETS_PAGE_SIZE,
|
||||
{ offset: requestedOffset }
|
||||
requestedOffset
|
||||
)
|
||||
|
||||
signal?.throwIfAborted()
|
||||
|
||||
@@ -709,7 +709,7 @@ describe('verifyMediaCandidates', () => {
|
||||
expect(mockFetchHistoryPage).toHaveBeenCalledWith(
|
||||
expect.any(Function),
|
||||
200,
|
||||
{ offset: 0 }
|
||||
0
|
||||
)
|
||||
expect(candidates[0]).toMatchObject({
|
||||
name: 'subfolder/photo.png [output]',
|
||||
@@ -843,13 +843,13 @@ describe('verifyMediaCandidates', () => {
|
||||
1,
|
||||
expect.any(Function),
|
||||
200,
|
||||
{ offset: 0 }
|
||||
0
|
||||
)
|
||||
expect(mockFetchHistoryPage).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
expect.any(Function),
|
||||
200,
|
||||
{ offset: 200 }
|
||||
200
|
||||
)
|
||||
expect(candidates[0].isMissing).toBe(false)
|
||||
})
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import {
|
||||
JobsApiError,
|
||||
extractWorkflow,
|
||||
fetchHistory,
|
||||
fetchHistoryPage,
|
||||
@@ -40,8 +39,7 @@ function createMockResponse(
|
||||
offset: pagination.offset ?? 0,
|
||||
limit: pagination.limit ?? 200,
|
||||
total,
|
||||
has_more: pagination.has_more ?? false,
|
||||
next_cursor: pagination.next_cursor
|
||||
has_more: pagination.has_more ?? false
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -137,85 +135,23 @@ describe('fetchJobs', () => {
|
||||
expect(result[0].priority).toBe(999)
|
||||
})
|
||||
|
||||
it('propagates fetch errors', async () => {
|
||||
it('returns empty array on error', async () => {
|
||||
const mockFetch = vi.fn().mockRejectedValue(new Error('Network error'))
|
||||
|
||||
await expect(fetchHistory(mockFetch)).rejects.toThrow('Network error')
|
||||
const result = await fetchHistory(mockFetch)
|
||||
|
||||
expect(result).toEqual([])
|
||||
})
|
||||
|
||||
it('throws a JobsApiError carrying status, body, and parsed errorCode on non-ok response', async () => {
|
||||
it('returns empty array on non-ok response', async () => {
|
||||
const mockFetch = vi.fn().mockResolvedValue({
|
||||
ok: false,
|
||||
status: 400,
|
||||
text: () =>
|
||||
Promise.resolve(
|
||||
'{"code":"INVALID_CURSOR","message":"Invalid pagination cursor"}'
|
||||
)
|
||||
status: 500
|
||||
})
|
||||
|
||||
await expect(fetchHistory(mockFetch)).rejects.toBeInstanceOf(JobsApiError)
|
||||
await expect(fetchHistory(mockFetch)).rejects.toMatchObject({
|
||||
status: 400,
|
||||
errorCode: 'INVALID_CURSOR',
|
||||
message: expect.stringContaining('INVALID_CURSOR')
|
||||
})
|
||||
})
|
||||
const result = await fetchHistory(mockFetch)
|
||||
|
||||
it('leaves errorCode undefined when the body is not the structured error shape', async () => {
|
||||
const mockFetch = vi.fn().mockResolvedValue({
|
||||
ok: false,
|
||||
status: 502,
|
||||
text: () => Promise.resolve('<html>Bad Gateway</html>')
|
||||
})
|
||||
|
||||
const err = await fetchHistory(mockFetch).catch((e) => e)
|
||||
expect(err).toBeInstanceOf(JobsApiError)
|
||||
expect(err.errorCode).toBeUndefined()
|
||||
})
|
||||
|
||||
it('leaves errorCode undefined for an oversized body rather than parsing it', async () => {
|
||||
const oversized = `{"code":"INVALID_CURSOR","pad":"${'x'.repeat(20_000)}"}`
|
||||
const mockFetch = vi.fn().mockResolvedValue({
|
||||
ok: false,
|
||||
status: 400,
|
||||
text: () => Promise.resolve(oversized)
|
||||
})
|
||||
|
||||
const err = await fetchHistory(mockFetch).catch((e) => e)
|
||||
expect(err).toBeInstanceOf(JobsApiError)
|
||||
expect(err.errorCode).toBeUndefined()
|
||||
})
|
||||
|
||||
it('truncates oversized error bodies to 200 chars in the thrown message', async () => {
|
||||
const oversized = 'x'.repeat(500)
|
||||
const mockFetch = vi.fn().mockResolvedValue({
|
||||
ok: false,
|
||||
status: 500,
|
||||
text: () => Promise.resolve(oversized)
|
||||
})
|
||||
|
||||
const err = await fetchHistory(mockFetch).catch((e) => e)
|
||||
expect(err).toBeInstanceOf(JobsApiError)
|
||||
expect(err.message.length).toBeLessThanOrEqual(
|
||||
'[Jobs API] Failed to fetch jobs: 500 '.length + 200 + 1 // +1 for the ellipsis
|
||||
)
|
||||
expect(err.message).toContain('…')
|
||||
})
|
||||
|
||||
it('parses a null next_cursor as absent', async () => {
|
||||
const mockFetch = vi.fn().mockResolvedValue({
|
||||
ok: true,
|
||||
json: () =>
|
||||
Promise.resolve(
|
||||
createMockResponse([createMockJob('job1', 'completed')], 1, {
|
||||
next_cursor: null
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
const result = await fetchHistoryPage(mockFetch, 200, { offset: 0 })
|
||||
|
||||
expect(result.nextCursor).toBeUndefined()
|
||||
expect(result).toEqual([])
|
||||
})
|
||||
|
||||
it('parses batch containing text-only preview outputs', async () => {
|
||||
@@ -269,7 +205,7 @@ describe('fetchJobs', () => {
|
||||
)
|
||||
})
|
||||
|
||||
const result = await fetchHistoryPage(mockFetch, 2, { offset: 5 })
|
||||
const result = await fetchHistoryPage(mockFetch, 2, 5)
|
||||
|
||||
expect(mockFetch).toHaveBeenCalledWith(
|
||||
'/jobs?status=completed,failed,cancelled&limit=2&offset=5'
|
||||
@@ -282,79 +218,6 @@ describe('fetchJobs', () => {
|
||||
expect(result.jobs[0].priority).toBe(5)
|
||||
expect(result.jobs[1].priority).toBe(4)
|
||||
})
|
||||
|
||||
it('sends the cursor instead of offset and returns next_cursor', async () => {
|
||||
const mockFetch = vi
|
||||
.fn<(url: string) => Promise<Response>>()
|
||||
.mockResolvedValue(
|
||||
new Response(
|
||||
JSON.stringify(
|
||||
createMockResponse([createMockJob('job1', 'completed')], 10, {
|
||||
has_more: true,
|
||||
next_cursor: 'cursor-page-2'
|
||||
})
|
||||
),
|
||||
{ status: 200 }
|
||||
)
|
||||
)
|
||||
|
||||
const result = await fetchHistoryPage(mockFetch, 200, {
|
||||
after: 'cursor-page-1'
|
||||
})
|
||||
|
||||
expect(mockFetch).toHaveBeenCalledWith(
|
||||
'/jobs?status=completed,failed,cancelled&limit=200&after=cursor-page-1'
|
||||
)
|
||||
expect(result.nextCursor).toBe('cursor-page-2')
|
||||
expect(result.hasMore).toBe(true)
|
||||
})
|
||||
|
||||
it('uri-encodes the cursor', async () => {
|
||||
const mockFetch = vi.fn().mockResolvedValue({
|
||||
ok: true,
|
||||
json: () => Promise.resolve(createMockResponse([]))
|
||||
})
|
||||
|
||||
await fetchHistoryPage(mockFetch, 200, { after: 'a+b/c=' })
|
||||
|
||||
expect(mockFetch).toHaveBeenCalledWith(
|
||||
'/jobs?status=completed,failed,cancelled&limit=200&after=a%2Bb%2Fc%3D'
|
||||
)
|
||||
})
|
||||
|
||||
it('returns next_cursor from offset-mode responses for cursor bootstrap', async () => {
|
||||
const mockFetch = vi.fn().mockResolvedValue({
|
||||
ok: true,
|
||||
json: () =>
|
||||
Promise.resolve(
|
||||
createMockResponse([createMockJob('job1', 'completed')], 10, {
|
||||
has_more: true,
|
||||
next_cursor: 'minted-in-offset-mode'
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
const result = await fetchHistoryPage(mockFetch, 200, { offset: 0 })
|
||||
|
||||
expect(mockFetch).toHaveBeenCalledWith(
|
||||
'/jobs?status=completed,failed,cancelled&limit=200&offset=0'
|
||||
)
|
||||
expect(result.nextCursor).toBe('minted-in-offset-mode')
|
||||
})
|
||||
|
||||
it('omits nextCursor when the server does not mint one', async () => {
|
||||
const mockFetch = vi.fn().mockResolvedValue({
|
||||
ok: true,
|
||||
json: () =>
|
||||
Promise.resolve(
|
||||
createMockResponse([createMockJob('job1', 'completed')])
|
||||
)
|
||||
})
|
||||
|
||||
const result = await fetchHistoryPage(mockFetch, 200, { offset: 0 })
|
||||
|
||||
expect(result.nextCursor).toBeUndefined()
|
||||
})
|
||||
})
|
||||
|
||||
describe('fetchQueue', () => {
|
||||
@@ -405,10 +268,12 @@ describe('fetchJobs', () => {
|
||||
)
|
||||
})
|
||||
|
||||
it('propagates fetch errors', async () => {
|
||||
it('returns empty arrays on error', async () => {
|
||||
const mockFetch = vi.fn().mockRejectedValue(new Error('Network error'))
|
||||
|
||||
await expect(fetchQueue(mockFetch)).rejects.toThrow('Network error')
|
||||
const result = await fetchQueue(mockFetch)
|
||||
|
||||
expect(result).toEqual({ Running: [], Pending: [] })
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@@ -6,8 +6,6 @@
|
||||
* All distributions use the /jobs endpoint.
|
||||
*/
|
||||
|
||||
import { z } from 'zod'
|
||||
|
||||
import type { ComfyWorkflowJSON } from '@/platform/workflow/validation/schemas/workflowSchema'
|
||||
import { validateComfyWorkflow } from '@/platform/workflow/validation/schemas/workflowSchema'
|
||||
import type { JobId } from '@/schemas/apiSchema'
|
||||
@@ -20,64 +18,12 @@ import type {
|
||||
} from './jobTypes'
|
||||
import { zJobDetail, zJobsListResponse, zWorkflowContainer } from './jobTypes'
|
||||
|
||||
/**
|
||||
* Position of the page to fetch. `after` is an opaque keyset cursor from a
|
||||
* prior response's `nextCursor` and takes precedence over `offset`; `offset`
|
||||
* remains as the fallback for random access and for backends that don't mint
|
||||
* cursors.
|
||||
*/
|
||||
export type JobsPageRequest =
|
||||
| { after: string; offset?: never }
|
||||
| { offset?: number; after?: never }
|
||||
|
||||
/**
|
||||
* Non-ok response from the jobs API. Carries the HTTP status and the parsed
|
||||
* machine-readable `errorCode` (from the JSON error body) so callers can tell a
|
||||
* rejected cursor (`INVALID_CURSOR`) apart from other 400s and transient
|
||||
* failures. `errorCode` is undefined when the body isn't the structured error
|
||||
* shape (e.g. a proxy error page).
|
||||
*/
|
||||
const MAX_ERROR_BODY_LENGTH = 200
|
||||
|
||||
// Cap synchronous JSON.parse so an oversized error body can't block the UI
|
||||
// thread; the structured error is a few hundred bytes at most.
|
||||
const MAX_ERROR_PARSE_LENGTH = 10_000
|
||||
|
||||
const zJobsErrorBody = z.object({ code: z.string() })
|
||||
|
||||
function parseErrorCode(body: string): string | undefined {
|
||||
if (body.length > MAX_ERROR_PARSE_LENGTH) return undefined
|
||||
try {
|
||||
return zJobsErrorBody.safeParse(JSON.parse(body)).data?.code
|
||||
} catch {
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
export class JobsApiError extends Error {
|
||||
readonly errorCode?: string
|
||||
|
||||
constructor(
|
||||
readonly status: number,
|
||||
body: string
|
||||
) {
|
||||
const truncated =
|
||||
body.length > MAX_ERROR_BODY_LENGTH
|
||||
? `${body.slice(0, MAX_ERROR_BODY_LENGTH)}…`
|
||||
: body
|
||||
super(`[Jobs API] Failed to fetch jobs: ${status} ${truncated}`.trim())
|
||||
this.name = 'JobsApiError'
|
||||
this.errorCode = parseErrorCode(body)
|
||||
}
|
||||
}
|
||||
|
||||
interface FetchJobsRawResult {
|
||||
jobs: RawJobListItem[]
|
||||
total: number
|
||||
offset: number
|
||||
limit: number
|
||||
hasMore: boolean
|
||||
nextCursor?: string
|
||||
}
|
||||
|
||||
export interface FetchHistoryPageResult {
|
||||
@@ -86,39 +32,43 @@ export interface FetchHistoryPageResult {
|
||||
offset: number
|
||||
limit: number
|
||||
hasMore: boolean
|
||||
nextCursor?: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches raw jobs from /jobs endpoint.
|
||||
* Throws on failure so callers can tell a failed page apart from an empty
|
||||
* last page (e.g. a stale cursor rejected with 400 INVALID_CURSOR).
|
||||
* Fetches raw jobs from /jobs endpoint
|
||||
* @internal
|
||||
*/
|
||||
async function fetchJobsRaw(
|
||||
fetchApi: (url: string) => Promise<Response>,
|
||||
statuses: JobStatus[],
|
||||
maxItems: number = 200,
|
||||
page: JobsPageRequest = {}
|
||||
offset: number = 0
|
||||
): Promise<FetchJobsRawResult> {
|
||||
const statusParam = statuses.join(',')
|
||||
const pageParam =
|
||||
page.after != null
|
||||
? `after=${encodeURIComponent(page.after)}`
|
||||
: `offset=${page.offset ?? 0}`
|
||||
const url = `/jobs?status=${statusParam}&limit=${maxItems}&${pageParam}`
|
||||
const res = await fetchApi(url)
|
||||
if (!res.ok) {
|
||||
throw new JobsApiError(res.status, await res.text().catch(() => ''))
|
||||
}
|
||||
const data = zJobsListResponse.parse(await res.json())
|
||||
return {
|
||||
jobs: data.jobs,
|
||||
total: data.pagination.total,
|
||||
offset: data.pagination.offset,
|
||||
limit: data.pagination.limit,
|
||||
hasMore: data.pagination.has_more,
|
||||
nextCursor: data.pagination.next_cursor ?? undefined
|
||||
const url = `/jobs?status=${statusParam}&limit=${maxItems}&offset=${offset}`
|
||||
try {
|
||||
const res = await fetchApi(url)
|
||||
if (!res.ok) {
|
||||
console.error(`[Jobs API] Failed to fetch jobs: ${res.status}`)
|
||||
return {
|
||||
jobs: [],
|
||||
total: 0,
|
||||
offset,
|
||||
limit: maxItems,
|
||||
hasMore: false
|
||||
}
|
||||
}
|
||||
const data = zJobsListResponse.parse(await res.json())
|
||||
return {
|
||||
jobs: data.jobs,
|
||||
total: data.pagination.total,
|
||||
offset: data.pagination.offset,
|
||||
limit: data.pagination.limit,
|
||||
hasMore: data.pagination.has_more
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('[Jobs API] Error fetching jobs:', error)
|
||||
return { jobs: [], total: 0, offset, limit: maxItems, hasMore: false }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -148,7 +98,7 @@ export async function fetchHistory(
|
||||
maxItems: number = 200,
|
||||
offset: number = 0
|
||||
): Promise<JobListItem[]> {
|
||||
const { jobs } = await fetchHistoryPage(fetchApi, maxItems, { offset })
|
||||
const { jobs } = await fetchHistoryPage(fetchApi, maxItems, offset)
|
||||
return jobs
|
||||
}
|
||||
|
||||
@@ -158,13 +108,13 @@ export async function fetchHistory(
|
||||
export async function fetchHistoryPage(
|
||||
fetchApi: (url: string) => Promise<Response>,
|
||||
maxItems: number = 200,
|
||||
page: JobsPageRequest = {}
|
||||
offset: number = 0
|
||||
): Promise<FetchHistoryPageResult> {
|
||||
const result = await fetchJobsRaw(
|
||||
fetchApi,
|
||||
['completed', 'failed', 'cancelled'],
|
||||
maxItems,
|
||||
page
|
||||
offset
|
||||
)
|
||||
|
||||
// History gets priority based on total count (lower than queue)
|
||||
@@ -173,8 +123,7 @@ export async function fetchHistoryPage(
|
||||
total: result.total,
|
||||
offset: result.offset,
|
||||
limit: result.limit,
|
||||
hasMore: result.hasMore,
|
||||
nextCursor: result.nextCursor
|
||||
hasMore: result.hasMore
|
||||
}
|
||||
}
|
||||
|
||||
@@ -185,7 +134,12 @@ export async function fetchHistoryPage(
|
||||
export async function fetchQueue(
|
||||
fetchApi: (url: string) => Promise<Response>
|
||||
): Promise<{ Running: JobListItem[]; Pending: JobListItem[] }> {
|
||||
const { jobs } = await fetchJobsRaw(fetchApi, ['in_progress', 'pending'])
|
||||
const { jobs } = await fetchJobsRaw(
|
||||
fetchApi,
|
||||
['in_progress', 'pending'],
|
||||
200,
|
||||
0
|
||||
)
|
||||
|
||||
const running = jobs.filter((j) => j.status === 'in_progress')
|
||||
const pending = jobs.filter((j) => j.status === 'pending')
|
||||
|
||||
@@ -87,8 +87,7 @@ const zPaginationInfo = z.object({
|
||||
offset: z.number(),
|
||||
limit: z.number(),
|
||||
total: z.number(),
|
||||
has_more: z.boolean(),
|
||||
next_cursor: z.string().min(1).nullish()
|
||||
has_more: z.boolean()
|
||||
})
|
||||
|
||||
export const zJobsListResponse = z.object({
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -17,14 +17,6 @@ import {
|
||||
} from '@/platform/assets/services/assetService'
|
||||
import type { PaginationOptions } from '@/platform/assets/services/assetService'
|
||||
import { isCloud } from '@/platform/distribution/types'
|
||||
import {
|
||||
JobsApiError,
|
||||
fetchHistoryPage
|
||||
} from '@/platform/remote/comfyui/jobs/fetchJobs'
|
||||
import type {
|
||||
FetchHistoryPageResult,
|
||||
JobsPageRequest
|
||||
} from '@/platform/remote/comfyui/jobs/fetchJobs'
|
||||
import type { JobListItem } from '@/platform/remote/comfyui/jobs/jobTypes'
|
||||
import { api } from '@/scripts/api'
|
||||
|
||||
@@ -122,9 +114,8 @@ export const useAssetsStore = defineStore('assets', () => {
|
||||
return deletingAssetIds.has(assetId)
|
||||
}
|
||||
|
||||
// History pagination state
|
||||
// Pagination state
|
||||
const historyOffset = ref(0)
|
||||
const historyNextCursor = ref<string | null>(null)
|
||||
const hasMoreHistory = ref(true)
|
||||
const isLoadingMore = ref(false)
|
||||
|
||||
@@ -132,12 +123,6 @@ export const useAssetsStore = defineStore('assets', () => {
|
||||
|
||||
const loadedIds = shallowReactive(new Set<string>())
|
||||
|
||||
// Ids of every raw job walked so far, including ones that map to no
|
||||
// displayable asset (failed, cancelled, preview-less). Head-refresh gap
|
||||
// detection needs the full set: a burst of non-asset jobs at the top would
|
||||
// otherwise never overlap `loadedIds` and trigger a needless full reload.
|
||||
const loadedJobIds = new Set<string>()
|
||||
|
||||
const fetchInputFiles = isCloud
|
||||
? fetchInputFilesFromCloud
|
||||
: fetchInputFilesFromAPI
|
||||
@@ -162,141 +147,65 @@ export const useAssetsStore = defineStore('assets', () => {
|
||||
}
|
||||
|
||||
/**
|
||||
* Insert assets in sorted order (newest first), skipping already-loaded ids
|
||||
*/
|
||||
const mergeHistoryAssets = (newAssets: AssetItem[]) => {
|
||||
for (const asset of newAssets) {
|
||||
if (loadedIds.has(asset.id)) {
|
||||
continue
|
||||
}
|
||||
loadedIds.add(asset.id)
|
||||
|
||||
const assetTime = new Date(asset.created_at ?? 0).getTime()
|
||||
const insertIndex = allHistoryItems.value.findIndex(
|
||||
(item) => new Date(item.created_at ?? 0).getTime() < assetTime
|
||||
)
|
||||
|
||||
if (insertIndex === -1) {
|
||||
allHistoryItems.value.push(asset)
|
||||
} else {
|
||||
allHistoryItems.value.splice(insertIndex, 0, asset)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const trimHistoryToLimit = () => {
|
||||
if (allHistoryItems.value.length <= MAX_HISTORY_ITEMS) return
|
||||
|
||||
const removed = allHistoryItems.value.slice(MAX_HISTORY_ITEMS)
|
||||
allHistoryItems.value = allHistoryItems.value.slice(0, MAX_HISTORY_ITEMS)
|
||||
removed.forEach((item) => loadedIds.delete(item.id))
|
||||
}
|
||||
|
||||
const fetchHistoryJobsPage = (page: JobsPageRequest) =>
|
||||
fetchHistoryPage(api.fetchApi.bind(api), BATCH_SIZE, page)
|
||||
|
||||
// Invalidates in-flight history fetches whenever the list is replaced, so
|
||||
// a stale continuation can't merge into (or move the cursor of) the new walk.
|
||||
let historyFetchEpoch = 0
|
||||
|
||||
// Tracks whether the walk is keyset-paginated, independent of the current
|
||||
// cursor value: once a cursor has been minted the walk stays in cursor mode
|
||||
// even after it exhausts (`historyNextCursor` back to null), so head-refresh
|
||||
// merges keep preserving scroll-loaded items instead of replacing them.
|
||||
let historyCursorMode = false
|
||||
|
||||
const isRejectedCursorError = (err: unknown): boolean =>
|
||||
err instanceof JobsApiError &&
|
||||
err.status === 400 &&
|
||||
err.errorCode === 'INVALID_CURSOR'
|
||||
|
||||
const fetchHistoryPageWithCursorRecovery = async (
|
||||
after: string | null,
|
||||
epoch: number
|
||||
): Promise<FetchHistoryPageResult> => {
|
||||
if (after == null)
|
||||
return fetchHistoryJobsPage({ offset: historyOffset.value })
|
||||
try {
|
||||
return await fetchHistoryJobsPage({ after })
|
||||
} catch (err) {
|
||||
// Drop only a rejected cursor (e.g. stale across a restart) to the
|
||||
// offset fallback; transient failures and superseded-walk
|
||||
// continuations must propagate so a valid/newer cursor isn't lost.
|
||||
if (!isRejectedCursorError(err) || epoch !== historyFetchEpoch) throw err
|
||||
console.warn('Stale history cursor rejected, resuming via offset:', err)
|
||||
historyNextCursor.value = null
|
||||
historyCursorMode = false
|
||||
historyOffset.value = 0
|
||||
allHistoryItems.value = []
|
||||
loadedIds.clear()
|
||||
loadedJobIds.clear()
|
||||
return fetchHistoryJobsPage({ offset: 0 })
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch one page of history assets and update reactive state.
|
||||
*
|
||||
* Pagination model: the server starts in offset mode and mints a
|
||||
* `next_cursor` on any page that has one; subsequent requests pass that
|
||||
* cursor (keyset mode). The walk upgrades automatically — offset paging is
|
||||
* only used until the first cursor is received.
|
||||
*
|
||||
* An empty page with no cursor is treated as terminal regardless of
|
||||
* `has_more`, because offset paging would refetch the same page forever.
|
||||
* A cursor that hasn't advanced (the server echoed back the value it was
|
||||
* given) is also treated as terminal to prevent an infinite dedup loop.
|
||||
*
|
||||
* @param loadMore - When `true`, appends the next page to the existing list
|
||||
* (infinite-scroll continuation). When `false` (default), resets all
|
||||
* pagination state and replaces the list with the first page.
|
||||
* @returns The current accumulated list of history asset items.
|
||||
* Fetch history assets with pagination support
|
||||
* @param loadMore - true for pagination (append), false for initial load (replace)
|
||||
*/
|
||||
const fetchHistoryAssets = async (loadMore = false): Promise<AssetItem[]> => {
|
||||
// Reset state for initial load
|
||||
if (!loadMore) {
|
||||
historyFetchEpoch += 1
|
||||
historyOffset.value = 0
|
||||
historyNextCursor.value = null
|
||||
historyCursorMode = false
|
||||
hasMoreHistory.value = true
|
||||
allHistoryItems.value = []
|
||||
loadedIds.clear()
|
||||
loadedJobIds.clear()
|
||||
}
|
||||
|
||||
const epoch = historyFetchEpoch
|
||||
const requestedAfter = loadMore ? historyNextCursor.value : null
|
||||
const page = await fetchHistoryPageWithCursorRecovery(requestedAfter, epoch)
|
||||
if (epoch !== historyFetchEpoch) return allHistoryItems.value
|
||||
// Fetch from server with offset
|
||||
const history = await api.getHistory(BATCH_SIZE, {
|
||||
offset: historyOffset.value
|
||||
})
|
||||
|
||||
page.jobs.forEach((job) => loadedJobIds.add(job.id))
|
||||
const newAssets = mapHistoryToAssets(page.jobs)
|
||||
// Convert JobListItems to AssetItems
|
||||
const newAssets = mapHistoryToAssets(history)
|
||||
|
||||
if (loadMore) {
|
||||
mergeHistoryAssets(newAssets)
|
||||
// Filter out duplicates and insert in sorted order
|
||||
for (const asset of newAssets) {
|
||||
if (loadedIds.has(asset.id)) {
|
||||
continue // Skip duplicates
|
||||
}
|
||||
loadedIds.add(asset.id)
|
||||
|
||||
// Find insertion index to maintain sorted order (newest first)
|
||||
const assetTime = new Date(asset.created_at ?? 0).getTime()
|
||||
const insertIndex = allHistoryItems.value.findIndex(
|
||||
(item) => new Date(item.created_at ?? 0).getTime() < assetTime
|
||||
)
|
||||
|
||||
if (insertIndex === -1) {
|
||||
// Asset is oldest, append to end
|
||||
allHistoryItems.value.push(asset)
|
||||
} else {
|
||||
// Insert at the correct position
|
||||
allHistoryItems.value.splice(insertIndex, 0, asset)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Initial load: replace all
|
||||
allHistoryItems.value = newAssets
|
||||
newAssets.forEach((asset) => loadedIds.add(asset.id))
|
||||
}
|
||||
|
||||
const cursorStuck =
|
||||
page.nextCursor != null && page.nextCursor === requestedAfter
|
||||
if (page.nextCursor != null) historyCursorMode = true
|
||||
// The server ignores `offset` once the walk is keyset-paginated, so only
|
||||
// advance it while still in offset mode; otherwise the offset used by the
|
||||
// recovery fallback would drift past valid rows.
|
||||
if (!historyCursorMode) historyOffset.value += page.jobs.length
|
||||
hasMoreHistory.value =
|
||||
page.hasMore &&
|
||||
!cursorStuck &&
|
||||
(page.jobs.length > 0 || page.nextCursor != null)
|
||||
// Drop the cursor once paging terminates so state never carries a live
|
||||
// cursor alongside `hasMoreHistory === false`.
|
||||
historyNextCursor.value = hasMoreHistory.value
|
||||
? (page.nextCursor ?? null)
|
||||
: null
|
||||
// Update pagination state
|
||||
historyOffset.value += BATCH_SIZE
|
||||
hasMoreHistory.value = history.length === BATCH_SIZE
|
||||
|
||||
trimHistoryToLimit()
|
||||
if (allHistoryItems.value.length > MAX_HISTORY_ITEMS) {
|
||||
const removed = allHistoryItems.value.slice(MAX_HISTORY_ITEMS)
|
||||
allHistoryItems.value = allHistoryItems.value.slice(0, MAX_HISTORY_ITEMS)
|
||||
|
||||
// Clean up Set
|
||||
removed.forEach((item) => loadedIds.delete(item.id))
|
||||
}
|
||||
|
||||
return allHistoryItems.value
|
||||
}
|
||||
@@ -336,15 +245,13 @@ export const useAssetsStore = defineStore('assets', () => {
|
||||
isLoadingMore.value = true
|
||||
historyError.value = null
|
||||
|
||||
const epoch = historyFetchEpoch
|
||||
try {
|
||||
await fetchHistoryAssets(true)
|
||||
if (epoch !== historyFetchEpoch) return
|
||||
historyAssets.value = allHistoryItems.value
|
||||
} catch (err) {
|
||||
if (epoch !== historyFetchEpoch) return
|
||||
console.error('Error loading more history:', err)
|
||||
historyError.value = err
|
||||
// Keep existing data when error occurs (consistent with updateHistory)
|
||||
if (!historyAssets.value.length) {
|
||||
historyAssets.value = []
|
||||
}
|
||||
@@ -353,100 +260,6 @@ export const useAssetsStore = defineStore('assets', () => {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A head page with no further rows spans the whole timeline, so replacing
|
||||
* local state with it also prunes jobs deleted server-side (e.g. after the
|
||||
* queue history is cleared from another surface).
|
||||
*
|
||||
* Bumps `historyFetchEpoch`, which cancels any concurrent
|
||||
* `loadMoreHistory`/`fetchHistoryAssets` continuation.
|
||||
*/
|
||||
const replaceHistoryWithHeadPage = (page: FetchHistoryPageResult) => {
|
||||
historyFetchEpoch += 1
|
||||
const newAssets = mapHistoryToAssets(page.jobs)
|
||||
allHistoryItems.value = newAssets
|
||||
loadedIds.clear()
|
||||
newAssets.forEach((asset) => loadedIds.add(asset.id))
|
||||
loadedJobIds.clear()
|
||||
page.jobs.forEach((job) => loadedJobIds.add(job.id))
|
||||
historyOffset.value = page.jobs.length
|
||||
historyNextCursor.value = page.nextCursor ?? null
|
||||
historyCursorMode = page.nextCursor != null
|
||||
hasMoreHistory.value = page.hasMore
|
||||
}
|
||||
|
||||
let headRefreshInFlight: Promise<void> | null = null
|
||||
let headRefreshTrailing: Promise<void> | null = null
|
||||
|
||||
/**
|
||||
* Merge newly completed jobs into the top of the list without resetting
|
||||
* pagination state, so items loaded via infinite scroll survive the
|
||||
* refresh. Cursors only walk toward older items, so new completions are
|
||||
* picked up by re-fetching the head page and deduplicating. Bursts of
|
||||
* status events share the in-flight refresh, and a call arriving
|
||||
* mid-flight schedules exactly one trailing refresh — the shared response
|
||||
* was dispatched before that caller's event, so it could miss the very
|
||||
* completion the caller is reacting to.
|
||||
*/
|
||||
const refreshHistoryHead = (): Promise<void> => {
|
||||
if (!headRefreshInFlight) {
|
||||
headRefreshInFlight = doRefreshHistoryHead().finally(() => {
|
||||
headRefreshInFlight = null
|
||||
})
|
||||
return headRefreshInFlight
|
||||
}
|
||||
headRefreshTrailing ??= headRefreshInFlight.then(() => {
|
||||
headRefreshTrailing = null
|
||||
return refreshHistoryHead()
|
||||
})
|
||||
return headRefreshTrailing
|
||||
}
|
||||
|
||||
const doRefreshHistoryHead = async () => {
|
||||
historyError.value = null
|
||||
if (!allHistoryItems.value.length) {
|
||||
await updateHistory()
|
||||
return
|
||||
}
|
||||
|
||||
let epoch = historyFetchEpoch
|
||||
try {
|
||||
const page = await fetchHistoryJobsPage({ offset: 0 })
|
||||
if (epoch !== historyFetchEpoch) return
|
||||
|
||||
const reachesLoadedItems = page.jobs.some((job) =>
|
||||
loadedJobIds.has(job.id)
|
||||
)
|
||||
if (page.hasMore && !reachesLoadedItems) {
|
||||
await updateHistory()
|
||||
return
|
||||
}
|
||||
|
||||
// Merging only preserves scroll-loaded items safely in cursor mode,
|
||||
// including once the cursor has exhausted (historyNextCursor is null but
|
||||
// the loaded terminal pages must survive). In offset fallback mode,
|
||||
// prepending new head rows without advancing historyOffset would drift
|
||||
// the next offset request (the server timeline shifted down by the new
|
||||
// completions), so rebuild from the head page — which resets
|
||||
// historyOffset to a position consistent with that page.
|
||||
if (page.hasMore && historyCursorMode) {
|
||||
page.jobs.forEach((job) => loadedJobIds.add(job.id))
|
||||
mergeHistoryAssets(mapHistoryToAssets(page.jobs))
|
||||
trimHistoryToLimit()
|
||||
} else {
|
||||
replaceHistoryWithHeadPage(page)
|
||||
// replaceHistoryWithHeadPage bumps the epoch; re-sync so the catch
|
||||
// guard below suppresses stale continuations, not genuine errors.
|
||||
epoch = historyFetchEpoch
|
||||
}
|
||||
historyAssets.value = allHistoryItems.value
|
||||
} catch (err) {
|
||||
if (epoch !== historyFetchEpoch) return
|
||||
console.error('Error refreshing history:', err)
|
||||
historyError.value = err
|
||||
}
|
||||
}
|
||||
|
||||
const flatOutputAssets = ref<AssetItem[]>([])
|
||||
const flatOutputLoading = ref(false)
|
||||
const flatOutputError = ref<unknown>(null)
|
||||
@@ -1071,7 +884,6 @@ export const useAssetsStore = defineStore('assets', () => {
|
||||
updateInputs,
|
||||
updateHistory,
|
||||
loadMoreHistory,
|
||||
refreshHistoryHead,
|
||||
setAssetPreview,
|
||||
|
||||
// Flat output assets (cloud-only, tag-based)
|
||||
|
||||
@@ -100,10 +100,7 @@ vi.mock('@/composables/useAppMode', () => ({
|
||||
useAppMode: () => ({ isBuilderMode: ref(false) })
|
||||
}))
|
||||
vi.mock('@/stores/assetsStore', () => ({
|
||||
useAssetsStore: () => ({
|
||||
updateHistory: vi.fn(),
|
||||
refreshHistoryHead: vi.fn()
|
||||
})
|
||||
useAssetsStore: () => ({ updateHistory: vi.fn() })
|
||||
}))
|
||||
vi.mock('@/stores/commandStore', () => ({
|
||||
useCommandStore: () => ({ registerCommands: vi.fn() })
|
||||
|
||||
@@ -238,7 +238,7 @@ const onStatus = async (e: CustomEvent<StatusWsMessageStatus>) => {
|
||||
// Only update assets if the assets sidebar is currently open
|
||||
// When sidebar is closed, AssetsSidebarTab.vue will refresh on mount
|
||||
if (sidebarTabStore.activeSidebarTabId === 'assets' || linearMode.value) {
|
||||
await assetsStore.refreshHistoryHead()
|
||||
await assetsStore.updateHistory()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -247,7 +247,7 @@ const onExecutionSuccess = async () => {
|
||||
// Only update assets if the assets sidebar is currently open
|
||||
// When sidebar is closed, AssetsSidebarTab.vue will refresh on mount
|
||||
if (sidebarTabStore.activeSidebarTabId === 'assets' || linearMode.value) {
|
||||
await assetsStore.refreshHistoryHead()
|
||||
await assetsStore.updateHistory()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -706,6 +706,9 @@ export default defineConfig({
|
||||
test: {
|
||||
globals: true,
|
||||
environment: 'happy-dom',
|
||||
// Pin the timezone so date-formatting assertions are deterministic
|
||||
// regardless of the contributor's local timezone (CI runs in UTC).
|
||||
env: { TZ: 'UTC' },
|
||||
setupFiles: ['./vitest.setup.ts'],
|
||||
retry: process.env.CI ? 2 : 0,
|
||||
include: [
|
||||
|
||||
Reference in New Issue
Block a user