Compare commits

...

29 Commits

Author SHA1 Message Date
Matt Miller
6d711e2b72 fix(assets): close trailing-run gap and normalize sync throws in coalescer
Return an already-queued trailing run before anything else so a call
landing in the microtask gap after the leading run clears inFlight but
before the trailing run clears its slot coalesces into the trailing run
instead of starting a second leading run, preserving the
at-most-one-trailing guarantee.

Normalize a synchronous throw from run() into a rejected promise so the
exported coalescer never throws out of invoke() and never latches a
half-initialized slot.
2026-07-17 20:43:14 -07:00
Matt Miller
2f6b4a7ba1 fix(assets): harden history head-refresh trailing chain against a rejected leading refresh 2026-07-17 18:56:55 -07:00
Matt Miller
4d9f89a40f fix(assets): harden history epoch guards and cursor consistency
- Guard loadMoreHistory's success assignment against a superseding
  reload so a stale continuation can't blank historyAssets mid-reload
- Re-sync the epoch after replaceHistoryWithHeadPage so post-replace
  errors surface instead of being swallowed by the staleness guard
- Drop historyNextCursor once paging terminates, so state never carries
  a live cursor alongside hasMoreHistory === false
- Reset historyError before the empty-list early return in
  doRefreshHistoryHead; document replaceHistoryWithHeadPage's epoch bump
2026-07-17 17:48:05 -07:00
Matt Miller
8e9baaad71 fix(assets): base head-refresh gap detection on raw jobs, not assets
Track every walked job id (including failed/cancelled/preview-less jobs that
map to no displayable asset) in a separate `loadedJobIds` set and use it for
head-refresh gap detection. Previously a burst of non-asset jobs at the top of
the timeline never overlapped `loadedIds` and forced a full reload that dropped
scroll position, even though the timeline was contiguous.

Also stop advancing `historyOffset` once the walk is keyset-paginated, so the
offset used by the recovery fallback can't drift past valid rows.

Adds regression coverage for the non-asset-overlap gap case and for discarding
a stale head refresh when a concurrent reset bumps the fetch epoch.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-17 17:19:36 -07:00
Matt Miller
9b9b458e4f fix(assets): preserve loaded pages after cursor exhausts on head refresh
Track keyset mode with a dedicated historyCursorMode flag instead of
inferring it from historyNextCursor. Once a cursor has been minted the
walk stays in cursor mode even after the cursor exhausts (nextCursor
back to null), so a completion-triggered head refresh merges the new
job into the loaded list rather than replacing the whole timeline with
just the head page and dropping already-loaded terminal pages.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-01 14:34:16 -07:00
Matt Miller
c9c0083834 Merge branch 'main' into matt/fe-962-fe-adopt-cursor-pagination-in-the-generated-tab-jobs-sourced 2026-06-30 21:04:14 -07:00
Matt Miller
60fa89d6cf Merge branch 'main' into matt/fe-962-fe-adopt-cursor-pagination-in-the-generated-tab-jobs-sourced 2026-06-30 14:34:53 -07:00
Matt Miller
e4d56565c1 test: assert concrete 500 status on the transient history error
Locks the recover-vs-not classification — only a JobsApiError with
status 400 drops to the offset fallback; a 500 must surface as a
recorded error without resetting the cursor.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-29 11:46:43 -07:00
Matt Miller
007a5cce8e fix: rebuild head page in offset mode to prevent history offset drift
In offset-fallback (cursorless) mode, refreshHistoryHead merged newly
completed jobs into the head without advancing historyOffset. The server
timeline shifts down by the new completions, so the next offset-mode
loadMoreHistory re-requested an overlapping window and silently skipped
rows past the shift. Rebuild from the already-fetched head page instead,
which resets historyOffset to a position consistent with that page.
Cursor mode keeps the merge, where offset isn't used for continuation.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-29 11:45:08 -07:00
Matt Miller
e471b64e68 Merge remote-tracking branch 'origin/main' into HEAD
# Conflicts:
#	src/stores/assetsStore.test.ts
2026-06-22 14:03:15 -07:00
Matt Miller
8b40f6a161 fix: reset offset to 0 on cursor-recovery fallback to prevent page drift
When a pagination cursor is rejected, fall back to offset 0 and replace
the loaded list rather than using the stale client-side offset. The
previous offset drifted when items were deleted server-side, causing the
recovery page to skip or duplicate rows.

Acceptable tradeoff: this rare fallback resets scroll position.
2026-06-22 13:50:34 -07:00
Matt Miller
ea0f8a9040 test: add body-truncation coverage and historyError-null assertion for superseded loadMore 2026-06-22 13:34:51 -07:00
Matt Miller
6c3ead5b81 Update src/stores/assetsStore.ts
Co-authored-by: Alexander Brown <drjkl@comfy.org>
2026-06-22 13:23:01 -07:00
Matt Miller
6f7686b952 Update src/platform/remote/comfyui/jobs/fetchJobs.ts
Co-authored-by: Alexander Brown <drjkl@comfy.org>
2026-06-22 13:22:49 -07:00
Matt Miller
c23814551a Update src/platform/remote/comfyui/jobs/jobTypes.ts
Co-authored-by: Alexander Brown <drjkl@comfy.org>
2026-06-22 13:22:34 -07:00
Matt Miller
89fdbcd913 Update src/stores/assetsStore.ts
Co-authored-by: Alexander Brown <drjkl@comfy.org>
2026-06-22 13:22:16 -07:00
Matt Miller
c207f5699d Merge branch 'main' into matt/fe-962-fe-adopt-cursor-pagination-in-the-generated-tab-jobs-sourced 2026-06-19 15:33:51 -07:00
GitHub Action
4290619af0 [automated] Apply ESLint and Oxfmt fixes 2026-06-19 02:40:07 +00:00
mattmiller
753b0b4c9b fix(assets): harden epoch guards and truncate JobsApiError body
- Hoist epoch snapshot to before the try block in doRefreshHistoryHead
  so the catch can check it and suppress stale failures
- Same pattern in loadMoreHistory: captures epoch before the await so a
  superseded loadMore that throws doesn't overwrite the healthy walk's
  error state
- Truncate JobsApiError body at 200 chars to avoid unbounded memory reads
  and leaking raw backend error detail into logs/error trackers
- Replace truthy cursor check (page.after ? ...) with != null in both
  fetchJobsRaw and fetchHistoryPageWithCursorRecovery so an empty-string
  cursor doesn't silently fall back to offset mode
2026-06-18 19:36:22 -07:00
Matt Miller
f318718521 docs(assets): add JSDoc to assetsStore per review 2026-06-17 11:49:04 -07:00
Matt Miller
69a4d78cba fix(assets): guard the history jobs walk against a non-advancing cursor
Mirror the flat-output walk's stuck-cursor guard so a backend that
returns the same next_cursor it was given (with has_more:true and a
non-empty page) can't loop the history walk forever. Without this guard,
mergeHistoryAssets dedupes every already-seen row, historyNextCursor
stays set to the stuck value, and hasMoreHistory remains true because
jobs.length > 0 — causing every subsequent loadMoreHistory to re-fetch
the identical page indefinitely.

Fix: capture requestedAfter before the fetch, detect cursorStuck when
page.nextCursor equals it, and in that case force hasMoreHistory=false
and drop the cursor. The initial (!loadMore) path sets requestedAfter to
null, so cursorStuck is always false there — no regression to the happy
path. Add a regression test that simulates the stuck-cursor shape and
asserts no further fetch is issued after the guard fires.
2026-06-17 11:49:04 -07:00
Matt Miller
1f810a1373 fix(assets): harden cursor pagination against review findings
Applies verified findings from multi-model review of #12769:

- next_cursor accepts null (z.string().nullish()) so a backend serializing
  absent-as-null can't crash the whole history fetch at the zod boundary
- JobsApiError carries the HTTP status and response body; the store's
  cursor recovery now drops the cursor for the offset fallback only on a
  400 (stale/rejected cursor) and only while its walk is still current —
  transient failures and superseded continuations propagate so a valid
  cursor is never discarded
- an empty page without a cursor is terminal (offset paging cannot
  advance by zero), while an empty page that still mints a cursor keeps
  walking
- refreshHistoryHead coalesces event bursts into the in-flight refresh
  plus exactly one trailing refresh, so a job completing mid-refresh is
  picked up instead of being satisfied by a response dispatched before
  it existed
2026-06-17 11:49:03 -07:00
Matt Miller
5ef89c73e6 feat(assets): adopt cursor pagination in the Generated tab jobs walk
The Generated tab walks GET /jobs with offset paging, which drifts when
jobs complete mid-scroll, and every status event resets the loaded list
back to page one. Core and cloud now share a keyset cursor contract on
/jobs (after / next_cursor), so:

- fetchHistoryPage accepts { after } and sends it instead of offset;
  next_cursor is surfaced from the pagination response
- the assets store walks history by cursor, bootstraps into cursor mode
  from the first offset page (the server mints a cursor either way), and
  keeps offset as the fallback for backends that don't mint cursors
- a rejected cursor (e.g. 400 INVALID_CURSOR after a server restart)
  drops the cursor and resumes once via the offset fallback
- fetchJobsRaw now throws on failure instead of fabricating an empty
  hasMore:false page, so a failed fetch is distinguishable from the end
  of the timeline and infinite scroll stays resumable after errors
- job-completion refreshes call refreshHistoryHead, which merge-prepends
  the head page instead of resetting scroll state; it replaces the list
  when the page spans the whole timeline (pruning server-side deletions)
  and restarts the walk when more than a page of new jobs would leave an
  unfillable gap
- in-flight page fetches are invalidated by an epoch counter when the
  list resets, so stale continuations can't pollute a fresh walk

hasMoreHistory is now server-authoritative instead of inferred from
batch length.
2026-06-17 11:49:02 -07:00
Matt Miller
1fee4490d4 Merge branch 'main' into ci/cursor-review-workflow 2026-06-17 11:26:20 -07:00
Matt Miller
8d23daa33d fix: resolve review feedback 2026-06-17 11:25:19 -07:00
GitHub Action
880582ab5d [automated] Apply ESLint and Oxfmt fixes 2026-06-17 18:21:48 +00:00
Matt Miller
e498c4ae0d ci: bump cursor-review SHA to github-workflows#9, drop judge_model override 2026-06-17 11:17:58 -07:00
Matt Miller
4cadbb8af9 fix: resolve review feedback 2026-06-15 15:42:26 -07:00
Matt Miller
cc048464fa ci: add team-gated Cursor review (thin caller for github-workflows)
Calls the reusable Comfy-Org/github-workflows cursor-review.yml (single source of truth for panel, judge, prompts, scripts) instead of a standalone copy. Label-gated to the team; secret-bearing jobs skip fork PRs. Judge overridden to Opus 4.8.
2026-06-15 14:36:26 -07:00
11 changed files with 1579 additions and 152 deletions

View File

@@ -245,7 +245,7 @@ const focusAssetInSidebar = async (item: JobListItem) => {
const assetId = String(jobId)
openAssetsSidebar()
await nextTick()
await assetsStore.updateHistory()
await assetsStore.refreshHistoryHead()
const asset = assetsStore.historyAssets.find(
(existingAsset) => existingAsset.id === assetId
)

View File

@@ -255,13 +255,13 @@ describe('resolveMissingMediaAssetSources', () => {
1,
expect.any(Function),
200,
0
{ offset: 0 }
)
expect(mockFetchHistoryPage).toHaveBeenNthCalledWith(
2,
expect.any(Function),
200,
200
{ offset: 200 }
)
})

View File

@@ -176,7 +176,7 @@ async function fetchGeneratedHistoryAssets(
const historyPage = await fetchHistoryPage(
api.fetchApi.bind(api),
HISTORY_MEDIA_ASSETS_PAGE_SIZE,
requestedOffset
{ offset: requestedOffset }
)
signal?.throwIfAborted()

View File

@@ -709,7 +709,7 @@ describe('verifyMediaCandidates', () => {
expect(mockFetchHistoryPage).toHaveBeenCalledWith(
expect.any(Function),
200,
0
{ offset: 0 }
)
expect(candidates[0]).toMatchObject({
name: 'subfolder/photo.png [output]',
@@ -843,13 +843,13 @@ describe('verifyMediaCandidates', () => {
1,
expect.any(Function),
200,
0
{ offset: 0 }
)
expect(mockFetchHistoryPage).toHaveBeenNthCalledWith(
2,
expect.any(Function),
200,
200
{ offset: 200 }
)
expect(candidates[0].isMissing).toBe(false)
})

View File

@@ -1,6 +1,7 @@
import { describe, expect, it, vi } from 'vitest'
import {
JobsApiError,
extractWorkflow,
fetchHistory,
fetchHistoryPage,
@@ -39,7 +40,8 @@ function createMockResponse(
offset: pagination.offset ?? 0,
limit: pagination.limit ?? 200,
total,
has_more: pagination.has_more ?? false
has_more: pagination.has_more ?? false,
next_cursor: pagination.next_cursor
}
}
}
@@ -135,23 +137,57 @@ describe('fetchJobs', () => {
expect(result[0].priority).toBe(999)
})
it('returns empty array on error', async () => {
it('propagates fetch errors', async () => {
const mockFetch = vi.fn().mockRejectedValue(new Error('Network error'))
const result = await fetchHistory(mockFetch)
expect(result).toEqual([])
await expect(fetchHistory(mockFetch)).rejects.toThrow('Network error')
})
it('returns empty array on non-ok response', async () => {
it('throws a JobsApiError carrying status and body on non-ok response', async () => {
const mockFetch = vi.fn().mockResolvedValue({
ok: false,
status: 500
status: 400,
text: () =>
Promise.resolve('{"error":"Invalid cursor","code":"INVALID_CURSOR"}')
})
const result = await fetchHistory(mockFetch)
await expect(fetchHistory(mockFetch)).rejects.toBeInstanceOf(JobsApiError)
await expect(fetchHistory(mockFetch)).rejects.toMatchObject({
status: 400,
message: expect.stringContaining('INVALID_CURSOR')
})
})
expect(result).toEqual([])
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()
})
it('parses batch containing text-only preview outputs', async () => {
@@ -205,7 +241,7 @@ describe('fetchJobs', () => {
)
})
const result = await fetchHistoryPage(mockFetch, 2, 5)
const result = await fetchHistoryPage(mockFetch, 2, { offset: 5 })
expect(mockFetch).toHaveBeenCalledWith(
'/jobs?status=completed,failed,cancelled&limit=2&offset=5'
@@ -218,6 +254,79 @@ 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', () => {
@@ -268,12 +377,10 @@ describe('fetchJobs', () => {
)
})
it('returns empty arrays on error', async () => {
it('propagates fetch errors', async () => {
const mockFetch = vi.fn().mockRejectedValue(new Error('Network error'))
const result = await fetchQueue(mockFetch)
expect(result).toEqual({ Running: [], Pending: [] })
await expect(fetchQueue(mockFetch)).rejects.toThrow('Network error')
})
})

View File

@@ -18,12 +18,43 @@ 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 so callers can
* tell a rejected cursor (400 INVALID_CURSOR) apart from transient failures.
*/
const MAX_ERROR_BODY_LENGTH = 200
export class JobsApiError extends Error {
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'
}
}
interface FetchJobsRawResult {
jobs: RawJobListItem[]
total: number
offset: number
limit: number
hasMore: boolean
nextCursor?: string
}
export interface FetchHistoryPageResult {
@@ -32,43 +63,39 @@ export interface FetchHistoryPageResult {
offset: number
limit: number
hasMore: boolean
nextCursor?: string
}
/**
* Fetches raw jobs from /jobs endpoint
* 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).
* @internal
*/
async function fetchJobsRaw(
fetchApi: (url: string) => Promise<Response>,
statuses: JobStatus[],
maxItems: number = 200,
offset: number = 0
page: JobsPageRequest = {}
): Promise<FetchJobsRawResult> {
const statusParam = statuses.join(',')
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 }
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
}
}
@@ -98,7 +125,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
}
@@ -108,13 +135,13 @@ export async function fetchHistory(
export async function fetchHistoryPage(
fetchApi: (url: string) => Promise<Response>,
maxItems: number = 200,
offset: number = 0
page: JobsPageRequest = {}
): Promise<FetchHistoryPageResult> {
const result = await fetchJobsRaw(
fetchApi,
['completed', 'failed', 'cancelled'],
maxItems,
offset
page
)
// History gets priority based on total count (lower than queue)
@@ -123,7 +150,8 @@ export async function fetchHistoryPage(
total: result.total,
offset: result.offset,
limit: result.limit,
hasMore: result.hasMore
hasMore: result.hasMore,
nextCursor: result.nextCursor
}
}
@@ -134,12 +162,7 @@ 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'],
200,
0
)
const { jobs } = await fetchJobsRaw(fetchApi, ['in_progress', 'pending'])
const running = jobs.filter((j) => j.status === 'in_progress')
const pending = jobs.filter((j) => j.status === 'pending')

View File

@@ -87,7 +87,8 @@ const zPaginationInfo = z.object({
offset: z.number(),
limit: z.number(),
total: z.number(),
has_more: z.boolean()
has_more: z.boolean(),
next_cursor: z.string().min(1).nullish()
})
export const zJobsListResponse = z.object({

File diff suppressed because it is too large Load Diff

View File

@@ -17,6 +17,14 @@ 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'
@@ -95,6 +103,54 @@ const BATCH_SIZE = 200
const MAX_HISTORY_ITEMS = 1000 // Maximum items to keep in memory
const FLAT_OUTPUT_PAGE_SIZE = 200
/**
* Coalesce concurrent calls to an async `run` into a single leading run plus
* at most one trailing run. Calls arriving while a run is in flight share that
* run; the first such call schedules exactly one trailing run to pick up any
* state the in-flight run was dispatched too early to observe.
*
* The leading rejection is swallowed before scheduling the trailing run so a
* failed run can never latch a settled-rejected promise in the trailing slot,
* which would freeze every future call. The leading caller still observes the
* rejection via the returned promise.
*
* A queued trailing run is returned before anything else so that a call landing
* in the microtask gap after the leading run clears `inFlight` but before the
* trailing run clears its slot coalesces into that trailing run rather than
* starting a second leading run, preserving the at-most-one-trailing guarantee.
*/
export function createTrailingRefreshCoalescer(
run: () => Promise<void>
): () => Promise<void> {
let inFlight: Promise<void> | null = null
let trailing: Promise<void> | null = null
const invoke = (): Promise<void> => {
if (trailing) return trailing
if (!inFlight) {
let started: Promise<void>
try {
started = run()
} catch (error) {
started = Promise.reject(error)
}
inFlight = started.finally(() => {
inFlight = null
})
return inFlight
}
trailing = inFlight
.catch(() => {})
.then(() => {
trailing = null
return invoke()
})
return trailing
}
return invoke
}
export const useAssetsStore = defineStore('assets', () => {
const assetDownloadStore = useAssetDownloadStore()
const modelToNodeStore = useModelToNodeStore()
@@ -114,8 +170,9 @@ export const useAssetsStore = defineStore('assets', () => {
return deletingAssetIds.has(assetId)
}
// Pagination state
// History pagination state
const historyOffset = ref(0)
const historyNextCursor = ref<string | null>(null)
const hasMoreHistory = ref(true)
const isLoadingMore = ref(false)
@@ -123,6 +180,12 @@ 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
@@ -147,65 +210,139 @@ export const useAssetsStore = defineStore('assets', () => {
}
/**
* Fetch history assets with pagination support
* @param loadMore - true for pagination (append), false for initial load (replace)
* 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
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.
*/
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()
}
// Fetch from server with offset
const history = await api.getHistory(BATCH_SIZE, {
offset: historyOffset.value
})
const epoch = historyFetchEpoch
const requestedAfter = loadMore ? historyNextCursor.value : null
const page = await fetchHistoryPageWithCursorRecovery(requestedAfter, epoch)
if (epoch !== historyFetchEpoch) return allHistoryItems.value
// Convert JobListItems to AssetItems
const newAssets = mapHistoryToAssets(history)
page.jobs.forEach((job) => loadedJobIds.add(job.id))
const newAssets = mapHistoryToAssets(page.jobs)
if (loadMore) {
// 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)
}
}
mergeHistoryAssets(newAssets)
} else {
// Initial load: replace all
allHistoryItems.value = newAssets
newAssets.forEach((asset) => loadedIds.add(asset.id))
}
// Update pagination state
historyOffset.value += BATCH_SIZE
hasMoreHistory.value = history.length === BATCH_SIZE
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
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))
}
trimHistoryToLimit()
return allHistoryItems.value
}
@@ -245,13 +382,15 @@ 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 = []
}
@@ -260,6 +399,86 @@ 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
}
/**
* 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 = createTrailingRefreshCoalescer(() =>
doRefreshHistoryHead()
)
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)
@@ -884,6 +1103,7 @@ export const useAssetsStore = defineStore('assets', () => {
updateInputs,
updateHistory,
loadMoreHistory,
refreshHistoryHead,
setAssetPreview,
// Flat output assets (cloud-only, tag-based)

View File

@@ -100,7 +100,10 @@ vi.mock('@/composables/useAppMode', () => ({
useAppMode: () => ({ isBuilderMode: ref(false) })
}))
vi.mock('@/stores/assetsStore', () => ({
useAssetsStore: () => ({ updateHistory: vi.fn() })
useAssetsStore: () => ({
updateHistory: vi.fn(),
refreshHistoryHead: vi.fn()
})
}))
vi.mock('@/stores/commandStore', () => ({
useCommandStore: () => ({ registerCommands: vi.fn() })

View File

@@ -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.updateHistory()
await assetsStore.refreshHistoryHead()
}
}
@@ -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.updateHistory()
await assetsStore.refreshHistoryHead()
}
}