Merge 7680a062b4 into 3f7b9667c6

apps/web/hooks/processing-poll-cycle.test.ts

@@ -0,0 +1,129 @@
+import { describe, expect, it } from "bun:test"
+import {
+	decideProcessingPoll,
+	MAX_PROCESSING_POLLS,
+	type ProcessingPollCycle,
+} from "./processing-poll-cycle"
+
+function activeSnapshot(
+	documentIds: readonly string[],
+	dataUpdateCount: number,
+	queryHash = "processing-scope-a",
+) {
+	return {
+		documentIds,
+		totalCount: documentIds.length,
+		dataUpdateCount,
+		queryHash,
+	}
+}
+
+describe("processing document poll cycles", () => {
+	it("stops the same processing job at the 60-update cap", () => {
+		let cycle: ProcessingPollCycle | null = null
+
+		for (
+			let updateCount = 1;
+			updateCount < MAX_PROCESSING_POLLS;
+			updateCount++
+		) {
+			const decision = decideProcessingPoll(
+				cycle,
+				activeSnapshot(
+					updateCount % 2 === 0
+						? ["document-a", "document-b"]
+						: ["document-b", "document-a"],
+					updateCount,
+				),
+			)
+			cycle = decision.cycle
+			expect(decision.shouldPoll).toBe(true)
+		}
+
+		const capped = decideProcessingPoll(
+			cycle,
+			activeSnapshot(["document-a", "document-b"], MAX_PROCESSING_POLLS),
+		)
+		expect(capped.shouldPoll).toBe(false)
+	})
+
+	it("restarts after an idle response even when the absolute count exceeds 60", () => {
+		let decision = decideProcessingPoll(null, activeSnapshot(["document-a"], 1))
+		decision = decideProcessingPoll(
+			decision.cycle,
+			activeSnapshot(["document-a"], MAX_PROCESSING_POLLS),
+		)
+		expect(decision.shouldPoll).toBe(false)
+
+		decision = decideProcessingPoll(decision.cycle, {
+			documentIds: [],
+			totalCount: 0,
+			dataUpdateCount: MAX_PROCESSING_POLLS + 1,
+			queryHash: "processing-scope-a",
+		})
+		expect(decision.shouldPoll).toBe(false)
+
+		decision = decideProcessingPoll(
+			decision.cycle,
+			activeSnapshot(["document-a"], MAX_PROCESSING_POLLS + 2),
+		)
+		expect(decision.shouldPoll).toBe(true)
+	})
+
+	it("starts a fresh budget when the query scope changes", () => {
+		let decision = decideProcessingPoll(
+			null,
+			activeSnapshot(["shared-document-id"], 1, "processing-scope-a"),
+		)
+		decision = decideProcessingPoll(
+			decision.cycle,
+			activeSnapshot(
+				["shared-document-id"],
+				MAX_PROCESSING_POLLS,
+				"processing-scope-a",
+			),
+		)
+		expect(decision.shouldPoll).toBe(false)
+
+		decision = decideProcessingPoll(
+			decision.cycle,
+			activeSnapshot(
+				["shared-document-id"],
+				MAX_PROCESSING_POLLS,
+				"processing-scope-b",
+			),
+		)
+		expect(decision.shouldPoll).toBe(true)
+	})
+
+	it("starts a fresh budget when the processing document IDs change", () => {
+		let decision = decideProcessingPoll(null, activeSnapshot(["document-a"], 1))
+		decision = decideProcessingPoll(
+			decision.cycle,
+			activeSnapshot(["document-a"], MAX_PROCESSING_POLLS),
+		)
+		expect(decision.shouldPoll).toBe(false)
+
+		decision = decideProcessingPoll(
+			decision.cycle,
+			activeSnapshot(["document-b"], MAX_PROCESSING_POLLS + 1),
+		)
+		expect(decision.shouldPoll).toBe(true)
+	})
+
+	it("does not spend the budget when the callback re-evaluates unchanged data", () => {
+		let decision = decideProcessingPoll(null, activeSnapshot(["document-a"], 1))
+
+		for (
+			let evaluation = 0;
+			evaluation < MAX_PROCESSING_POLLS * 2;
+			evaluation++
+		) {
+			decision = decideProcessingPoll(
+				decision.cycle,
+				activeSnapshot(["document-a"], 1),
+			)
+			expect(decision.shouldPoll).toBe(true)
+		}
+	})
+})

apps/web/hooks/processing-poll-cycle.ts

@@ -0,0 +1,61 @@
+export const MAX_PROCESSING_POLLS = 60
+
+export type ProcessingPollCycle = {
+	documentIds: readonly string[]
+	firstDataUpdateCount: number
+	queryHash: string
+}
+
+export type ProcessingPollSnapshot = {
+	documentIds: readonly string[]
+	totalCount: number
+	dataUpdateCount: number
+	queryHash: string
+}
+
+export type ProcessingPollDecision = {
+	cycle: ProcessingPollCycle | null
+	shouldPoll: boolean
+}
+
+function normalizedDocumentIds(documentIds: readonly string[]) {
+	return [...new Set(documentIds)].sort()
+}
+
+function hasSameDocumentIds(left: readonly string[], right: readonly string[]) {
+	return (
+		left.length === right.length &&
+		left.every((documentId, index) => documentId === right[index])
+	)
+}
+
+export function decideProcessingPoll(
+	cycle: ProcessingPollCycle | null,
+	snapshot: ProcessingPollSnapshot,
+): ProcessingPollDecision {
+	if (snapshot.totalCount === 0) {
+		return { cycle: null, shouldPoll: false }
+	}
+
+	const documentIds = normalizedDocumentIds(snapshot.documentIds)
+	const startsNewCycle =
+		cycle === null ||
+		cycle.queryHash !== snapshot.queryHash ||
+		!hasSameDocumentIds(cycle.documentIds, documentIds) ||
+		snapshot.dataUpdateCount < cycle.firstDataUpdateCount
+	const nextCycle = startsNewCycle
+		? {
+				documentIds,
+				firstDataUpdateCount: snapshot.dataUpdateCount,
+				queryHash: snapshot.queryHash,
+			}
+		: cycle
+	// React Query's count spans the cache lifetime, so make it cycle-relative.
+	const updatesInCycle =
+		snapshot.dataUpdateCount - nextCycle.firstDataUpdateCount + 1
+
+	return {
+		cycle: nextCycle,
+		shouldPoll: updatesInCycle < MAX_PROCESSING_POLLS,
+	}
+}

apps/web/hooks/use-processing-documents.ts

@@ -5,15 +5,29 @@ import { useQuery, useQueryClient } from "@tanstack/react-query"
 import { $fetch } from "@lib/api"
 import { useProject } from "@/stores"
 import { useAuth } from "@lib/auth-context"
+import {
+	decideProcessingPoll,
+	type ProcessingPollCycle,
+} from "./processing-poll-cycle"
 
-const MAX_POLLS = 60
 const POLL_INTERVAL_MS = 5_000
 
+type ProcessingDocument = {
+	id?: string | null
+	status?: string | null
+}
+
+type ProcessingDocumentsData = {
+	documents?: ProcessingDocument[]
+	totalCount?: number
+}
+
 export function useProcessingDocuments() {
 	const { user } = useAuth()
 	const { effectiveContainerTags } = useProject()
 	const queryClient = useQueryClient()
 	const prevIdsRef = useRef<Set<string>>(new Set())
+	const pollCycleRef = useRef<ProcessingPollCycle | null>(null)
 
 	const { data } = useQuery({
 		queryKey: ["processing-documents", effectiveContainerTags],
@@ -27,12 +41,17 @@ export function useProcessingDocuments() {
 		},
 		enabled: !!user,
 		refetchInterval: (query) => {
-			const count =
-				(query.state.data as { totalCount?: number } | undefined)?.totalCount ??
-				0
-			const polls = query.state.dataUpdateCount
-			if (count === 0 || polls >= MAX_POLLS) return false
-			return POLL_INTERVAL_MS
+			const queryData = query.state.data as ProcessingDocumentsData | undefined
+			const decision = decideProcessingPoll(pollCycleRef.current, {
+				documentIds: (queryData?.documents ?? []).flatMap((document) =>
+					document.id ? [document.id] : [],
+				),
+				totalCount: queryData?.totalCount ?? 0,
+				dataUpdateCount: query.state.dataUpdateCount,
+				queryHash: query.queryHash,
+			})
+			pollCycleRef.current = decision.cycle
+			return decision.shouldPoll ? POLL_INTERVAL_MS : false
 		},
 		staleTime: 0,
 	})
@@ -41,14 +60,7 @@ export function useProcessingDocuments() {
 	// Query's structural sharing) so `processingMap` only changes identity
 	// when the poll payload actually changes — the effect below depends on it.
 	const docs = useMemo(
-		() =>
-			(
-				data as
-					| {
-							documents?: Array<{ id?: string | null; status?: string | null }>
-					  }
-					| undefined
-			)?.documents ?? [],
+		() => (data as ProcessingDocumentsData | undefined)?.documents ?? [],
 		[data],
 	)