From 66d8564dd2f35f8a2750525af375a8327dcfa8f4 Mon Sep 17 00:00:00 2001 From: kuekhaoyang Date: Mon, 13 Jul 2026 21:09:47 +0800 Subject: [PATCH] fix issue 210 slow source search reliability --- CHANGELOG.md | 7 +++ app-release.json | 13 +++- app/api/ping/route.ts | 50 +++------------ app/page.tsx | 4 +- lib/api/http-utils.ts | 15 +++-- lib/api/source-latency.ts | 69 ++++++++++++++++++++ lib/hooks/useLatencyPing.ts | 92 ++++++++++++++++++--------- lib/hooks/useSearchAction.ts | 10 ++- lib/types/index.ts | 1 + lib/utils/latency.ts | 40 ++++++++++++ lib/utils/search-stream.ts | 93 ++++++++++++--------------- package-lock.json | 4 +- package.json | 2 +- tests/search-reliability.test.ts | 104 +++++++++++++++++++++++++++++++ 14 files changed, 364 insertions(+), 140 deletions(-) create mode 100644 lib/api/source-latency.ts create mode 100644 tests/search-reliability.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index a10d0bf..c320a4b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## 4.9.13 - 2026-07-13 + +- 修复慢视频源超过 3 秒未返回时，前端提前结束搜索并错误显示“未找到结果”的问题；搜索现在等待服务端明确完成或响应流真正关闭。 +- 实时延迟探测现在使用视频源的真实 `baseUrl`，不再把源 ID 当成 URL；`HEAD` 失败后的 `GET` 探测独立计时，避免把前一次失败耗时叠加为虚高延迟。 +- 延迟探测改为最多 4 路并发并在上一轮结束后再调度下一轮，避免 Docker 自托管实例因慢源产生重叠探测请求。 +- 搜索请求取消或超时后不再继续重试，并清理级联 AbortSignal 监听器；新增慢源、探测回退、并发上限和取消重试回归测试。 + ## 4.9.12 - 2026-07-12 - 修复窄屏导航栏隐藏整个用户区域、导致登录用户无法退出的问题；退出按钮现在在所有视口宽度下保持可见，用户身份详情仍在桌面端显示。 diff --git a/app-release.json b/app-release.json index 2f97356..6641cbd 100644 --- a/app-release.json +++ b/app-release.json @@ -4,8 +4,19 @@ "name": "KVideo", "branch": "main" }, - "currentVersion": "4.9.12", + "currentVersion": "4.9.13", "releases": [ + { + "version": "4.9.13", + "publishedAt": "2026-07-13", + "title": "修复慢源搜索与延迟探测", + "notes": [ + "慢视频源不再因 3 秒静默被前端提前判定为搜索完成，结果流会等待服务端明确结束。", + "实时延迟改用真实视频源地址并独立计算回退请求耗时，避免无效 URL 和累加计时造成虚高数值。", + "延迟探测限制为最多 4 路并发且轮次不重叠，搜索取消或超时后也不会继续无意义重试。", + "新增慢源搜索、延迟回退、探测并发和取消重试回归测试。" + ] + }, { "version": "4.9.12", "publishedAt": "2026-07-12", diff --git a/app/api/ping/route.ts b/app/api/ping/route.ts index 95d1176..e2ef59d 100644 --- a/app/api/ping/route.ts +++ b/app/api/ping/route.ts @@ -4,6 +4,7 @@ */ import { NextRequest, NextResponse } from 'next/server'; +import { probeSourceLatency } from '@/lib/api/source-latency'; export const runtime = 'edge'; @@ -18,53 +19,16 @@ export async function POST(request: NextRequest) { // Validate URL format try { - new URL(url); + const parsedUrl = new URL(url); + if (!['http:', 'https:'].includes(parsedUrl.protocol)) { + return NextResponse.json({ error: 'Unsupported URL protocol' }, { status: 400 }); + } } catch { return NextResponse.json({ error: 'Invalid URL format' }, { status: 400 }); } - const startTime = performance.now(); - - try { - // Use HEAD request for faster ping (less data transfer) - const controller = new AbortController(); - const timeoutId = setTimeout(() => controller.abort(), 5000); // 5s timeout - - await fetch(url, { - method: 'HEAD', - signal: controller.signal, - mode: 'no-cors', // Allow cross-origin requests - }); - - clearTimeout(timeoutId); - - const endTime = performance.now(); - const latency = Math.round(endTime - startTime); - - return NextResponse.json({ latency, success: true }); - } catch (fetchError) { - // If HEAD fails, try GET with timeout - const controller = new AbortController(); - const timeoutId = setTimeout(() => controller.abort(), 5000); - - try { - await fetch(url, { - method: 'GET', - signal: controller.signal, - }); - clearTimeout(timeoutId); - - const endTime = performance.now(); - const latency = Math.round(endTime - startTime); - return NextResponse.json({ latency, success: true }); - } catch { - clearTimeout(timeoutId); - const endTime = performance.now(); - const latency = Math.round(endTime - startTime); - // Still return latency even on error (timeout = slow) - return NextResponse.json({ latency, success: false, timeout: true }); - } - } + const result = await probeSourceLatency(url); + return NextResponse.json(result); } catch (error) { console.error('Ping error:', error); return NextResponse.json( diff --git a/app/page.tsx b/app/page.tsx index 12bdca5..ad2fdb9 100644 --- a/app/page.tsx +++ b/app/page.tsx @@ -26,7 +26,9 @@ function HomePage() { // Real-time latency pinging const sourceUrls = useMemo(() => - availableSources.map(s => ({ id: s.id, baseUrl: s.id })), // Using id as baseUrl if not available elsewhere + availableSources.flatMap((source) => + source.baseUrl ? [{ id: source.id, baseUrl: source.baseUrl }] : [] + ), [availableSources] ); diff --git a/lib/api/http-utils.ts b/lib/api/http-utils.ts index 9bec6b1..2b3ec8d 100644 --- a/lib/api/http-utils.ts +++ b/lib/api/http-utils.ts @@ -24,12 +24,12 @@ export async function fetchWithTimeout( // If an external signal is provided, propagate its abort const externalSignal = options.signal; + let onAbort: (() => void) | undefined; if (externalSignal) { if (externalSignal.aborted) { - clearTimeout(timeoutId); controller.abort(); } else { - const onAbort = () => controller.abort(); + onAbort = () => controller.abort(); externalSignal.addEventListener('abort', onAbort, { once: true }); } } @@ -39,11 +39,12 @@ export async function fetchWithTimeout( ...options, signal: controller.signal, }); - clearTimeout(timeoutId); return response; - } catch (error) { + } finally { clearTimeout(timeoutId); - throw error; + if (externalSignal && onAbort) { + externalSignal.removeEventListener('abort', onAbort); + } } } @@ -62,6 +63,10 @@ export async function withRetry( } catch (error) { lastError = error as Error; + if (lastError?.name === 'AbortError') { + throw lastError; + } + if (i < retries) { await new Promise(resolve => setTimeout(resolve, RETRY_DELAY * (i + 1))); } diff --git a/lib/api/source-latency.ts b/lib/api/source-latency.ts new file mode 100644 index 0000000..33b3095 --- /dev/null +++ b/lib/api/source-latency.ts @@ -0,0 +1,69 @@ +export interface SourceLatencyResult { + latency: number; + success: boolean; + timeout: boolean; + method: 'HEAD' | 'GET'; +} + +type ProbeAttemptResult = SourceLatencyResult; + +interface ProbeSourceLatencyOptions { + fetcher?: typeof fetch; + timeoutMs?: number; + now?: () => number; +} + +async function probeAttempt( + url: string, + method: 'HEAD' | 'GET', + fetcher: typeof fetch, + timeoutMs: number, + now: () => number, +): Promise { + const controller = new AbortController(); + let timedOut = false; + const timeoutId = setTimeout(() => { + timedOut = true; + controller.abort(); + }, timeoutMs); + const startedAt = now(); + + try { + await fetcher(url, { + method, + signal: controller.signal, + redirect: 'follow', + ...(method === 'GET' ? { headers: { Range: 'bytes=0-0' } } : {}), + }); + + return { + latency: Math.max(0, Math.round(now() - startedAt)), + success: true, + timeout: false, + method, + }; + } catch (error) { + return { + latency: Math.max(0, Math.round(now() - startedAt)), + success: false, + timeout: timedOut || (error instanceof Error && error.name === 'AbortError'), + method, + }; + } finally { + clearTimeout(timeoutId); + } +} + +export async function probeSourceLatency( + url: string, + options: ProbeSourceLatencyOptions = {}, +): Promise { + const fetcher = options.fetcher ?? fetch; + const timeoutMs = options.timeoutMs ?? 5000; + const now = options.now ?? (() => performance.now()); + + const headResult = await probeAttempt(url, 'HEAD', fetcher, timeoutMs, now); + if (headResult.success) return headResult; + + return probeAttempt(url, 'GET', fetcher, timeoutMs, now); +} diff --git a/lib/hooks/useLatencyPing.ts b/lib/hooks/useLatencyPing.ts index 42ec530..d5ddfb2 100644 --- a/lib/hooks/useLatencyPing.ts +++ b/lib/hooks/useLatencyPing.ts @@ -5,6 +5,7 @@ import { useState, useEffect, useCallback, useRef, useMemo } from 'react'; import { settingsStore } from '@/lib/store/settings-store'; +import { probeLatencyTargets } from '@/lib/utils/latency'; interface LatencyState { [sourceId: string]: number; @@ -23,17 +24,18 @@ export function useLatencyPing({ }: UseLatencyPingOptions) { const [latencies, setLatencies] = useState({}); const [isLoading, setIsLoading] = useState(false); - const intervalRef = useRef(null); + const intervalRef = useRef | null>(null); + const roundInFlightRef = useRef(false); const mountedRef = useRef(true); // Check if real-time latency is enabled in settings const [realtimeEnabled, setRealtimeEnabled] = useState(false); // Stabilize sourceUrls to prevent unnecessary effect re-runs if parent passes new array - const stableSourceUrls = useMemo(() => sourceUrls, [ - // Create a unique key for the sources array - sourceUrls.map(s => `${s.id}|${s.baseUrl}`).join(',') - ]); + const stableSourceUrls = useMemo( + () => sourceUrls.map((source) => ({ ...source })), + [sourceUrls], + ); useEffect(() => { const settings = settingsStore.getSettings(); @@ -50,47 +52,64 @@ export function useLatencyPing({ }; }, []); - const pingSource = useCallback(async (sourceId: string, baseUrl: string): Promise => { + const pingSource = useCallback(async (baseUrl: string): Promise => { + const controller = new AbortController(); + const timeoutId = setTimeout(() => controller.abort(), 12000); + try { const response = await fetch('/api/ping', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ url: baseUrl }), + signal: controller.signal, }); if (response.ok) { const data = await response.json(); - return data.latency || null; + return data.success && typeof data.latency === 'number' + ? data.latency + : null; } return null; } catch { return null; + } finally { + clearTimeout(timeoutId); } }, []); const pingAllSources = useCallback(async () => { - if (!mountedRef.current || stableSourceUrls.length === 0) return; + if ( + !mountedRef.current || + stableSourceUrls.length === 0 || + roundInFlightRef.current + ) return; + roundInFlightRef.current = true; setIsLoading(true); - const results = await Promise.all( - stableSourceUrls.map(async ({ id, baseUrl }) => { - const latency = await pingSource(id, baseUrl); - return { id, latency }; - }) - ); + try { + const results = await probeLatencyTargets( + stableSourceUrls, + ({ baseUrl }) => pingSource(baseUrl), + ); - if (mountedRef.current) { - setLatencies(prev => { - const newState = { ...prev }; - results.forEach(({ id, latency }) => { - if (latency !== null) { - newState[id] = latency; - } + if (mountedRef.current) { + setLatencies(prev => { + const newState = { ...prev }; + results.forEach(({ id, latency }) => { + if (latency !== null) { + newState[id] = latency; + } + }); + return newState; }); - return newState; - }); - setIsLoading(false); + } + } finally { + roundInFlightRef.current = false; + if (mountedRef.current) { + setIsLoading(false); + } } }, [stableSourceUrls, pingSource]); @@ -101,17 +120,30 @@ export function useLatencyPing({ const shouldPoll = enabled && realtimeEnabled && stableSourceUrls.length > 0; if (shouldPoll) { - // Initial ping - pingAllSources(); + let cancelled = false; + const poll = async () => { + await pingAllSources(); + if (!cancelled && mountedRef.current) { + intervalRef.current = setTimeout(poll, intervalMs); + } + }; - // Set up interval - intervalRef.current = setInterval(pingAllSources, intervalMs); + void poll(); + + return () => { + cancelled = true; + mountedRef.current = false; + if (intervalRef.current) { + clearTimeout(intervalRef.current); + intervalRef.current = null; + } + }; } return () => { mountedRef.current = false; if (intervalRef.current) { - clearInterval(intervalRef.current); + clearTimeout(intervalRef.current); intervalRef.current = null; } }; @@ -120,7 +152,7 @@ export function useLatencyPing({ const refreshLatency = useCallback((sourceId: string) => { const source = stableSourceUrls.find(s => s.id === sourceId); if (source) { - pingSource(sourceId, source.baseUrl).then(latency => { + pingSource(source.baseUrl).then(latency => { if (latency !== null && mountedRef.current) { setLatencies(prev => ({ ...prev, [sourceId]: latency })); } diff --git a/lib/hooks/useSearchAction.ts b/lib/hooks/useSearchAction.ts index ab05fe9..0e6277c 100644 --- a/lib/hooks/useSearchAction.ts +++ b/lib/hooks/useSearchAction.ts @@ -1,14 +1,13 @@ import { useRef, useCallback } from 'react'; -import { SOURCE_IDS } from '@/lib/utils/source-names'; import { sortVideos } from '@/lib/utils/sort'; import { binaryInsertVideos } from '@/lib/utils/sorted-insert'; import { processSearchStream } from '@/lib/utils/search-stream'; import type { SortOption } from '@/lib/store/settings-store'; import { settingsStore } from '@/lib/store/settings-store'; -import type { Video } from '@/lib/types'; import { useSearchState } from './useSearchState'; type SearchState = ReturnType; +type SearchSourceConfig = { id: string; baseUrl?: string }; interface UseSearchActionProps { state: SearchState; @@ -77,7 +76,10 @@ export function useSearchAction({ state, onCacheUpdate, onUrlUpdate }: UseSearch const reader = response.body?.getReader(); if (!reader) throw new Error('No response stream'); - const sourcesMap = new Map(); + const sourceConfigs = new Map( + targetSources.map((source: SearchSourceConfig) => [source.id, source]) + ); + const sourcesMap = new Map(); await processSearchStream({ reader, @@ -95,6 +97,7 @@ export function useSearchAction({ state, onCacheUpdate, onUrlUpdate }: UseSearch sourcesMap.set(sourceId, { count: newVideos.length, name: newVideos[0]?.sourceName || sourceId, + baseUrl: sourceConfigs.get(sourceId)?.baseUrl, }); } }, @@ -113,6 +116,7 @@ export function useSearchAction({ state, onCacheUpdate, onUrlUpdate }: UseSearch id: id, name: info.name, count: info.count, + ...(info.baseUrl ? { baseUrl: info.baseUrl } : {}), })); setAvailableSources(sources); diff --git a/lib/types/index.ts b/lib/types/index.ts index d7f5f31..1b022c4 100644 --- a/lib/types/index.ts +++ b/lib/types/index.ts @@ -51,6 +51,7 @@ export interface SourceBadge { id: string; name: string; count: number; + baseUrl?: string; typeName?: string; } diff --git a/lib/utils/latency.ts b/lib/utils/latency.ts index 4b42894..a3fb1ef 100644 --- a/lib/utils/latency.ts +++ b/lib/utils/latency.ts @@ -10,6 +10,46 @@ interface LatencyInfo { level: 'excellent' | 'good' | 'fair' | 'slow'; } +export interface LatencyProbeTarget { + id: string; + baseUrl: string; +} + +export interface LatencyProbeResult { + id: string; + latency: number | null; +} + +export async function probeLatencyTargets( + targets: LatencyProbeTarget[], + probe: (target: LatencyProbeTarget) => Promise, + concurrency: number = 4, +): Promise { + if (targets.length === 0) return []; + + const results = new Array(targets.length); + const workerCount = Math.min( + targets.length, + Math.max(1, Math.floor(concurrency)), + ); + let nextIndex = 0; + + const workers = Array.from({ length: workerCount }, async () => { + while (nextIndex < targets.length) { + const index = nextIndex; + nextIndex += 1; + const target = targets[index]; + results[index] = { + id: target.id, + latency: await probe(target), + }; + } + }); + + await Promise.all(workers); + return results; +} + /** * Get latency information with color coding * @param latency - Response time in milliseconds diff --git a/lib/utils/search-stream.ts b/lib/utils/search-stream.ts index 8b08c41..8887043 100644 --- a/lib/utils/search-stream.ts +++ b/lib/utils/search-stream.ts @@ -35,80 +35,65 @@ export async function processSearchStream({ }: StreamHandlerParams) { const decoder = new TextDecoder(); let buffer = ''; - let timeoutId: NodeJS.Timeout | null = null; let isCompleted = false; - // Auto-complete if no progress for 3 seconds - const resetTimeout = () => { - if (timeoutId) clearTimeout(timeoutId); + const blockedCategories = settingsStore.getSettings().blockedCategories; - timeoutId = setTimeout(() => { + while (true) { + const { done, value } = await reader.read(); + if (done) { if (!isCompleted) { - isCompleted = true; onComplete(); } - }, 3000); - }; + break; + } - try { - resetTimeout(); // Start initial timeout - const blockedCategories = settingsStore.getSettings().blockedCategories; + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split('\n'); + buffer = lines.pop() || ''; - while (true) { - const { done, value } = await reader.read(); - if (done) break; + for (const line of lines) { + if (!line.startsWith('data: ')) continue; - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split('\n'); - buffer = lines.pop() || ''; + try { + const data = JSON.parse(line.slice(6)); - for (const line of lines) { - if (!line.startsWith('data: ')) continue; - - try { - const data = JSON.parse(line.slice(6)); - - if (data.type === 'start') { - onStart(data.totalSources); - resetTimeout(); - } else if (data.type === 'videos') { - const newVideos: Video[] = data.videos - .filter((video: any) => hasMinimumMatch(video.vod_name, currentQuery)) - .filter((video: any) => !isCategoryBlocked(video, blockedCategories)) - .map((video: any) => ({ - ...video, - sourceName: video.sourceDisplayName || getSourceName(video.source), - isNew: true, - relevanceScore: calculateRelevanceScore(video, currentQuery), - })); - onVideos(newVideos, data.source); - if (data.pagecount && onPageInfo) { - onPageInfo(data.pagecount); - } - resetTimeout(); - } else if (data.type === 'progress') { - onProgress(data.completedSources, data.totalVideosFound); - resetTimeout(); - } else if (data.type === 'complete') { - if (timeoutId) clearTimeout(timeoutId); + if (data.type === 'start') { + onStart(data.totalSources); + } else if (data.type === 'videos') { + const newVideos: Video[] = data.videos + .filter((video: any) => hasMinimumMatch(video.vod_name, currentQuery)) + .filter((video: any) => !isCategoryBlocked(video, blockedCategories)) + .map((video: any) => ({ + ...video, + sourceName: video.sourceDisplayName || getSourceName(video.source), + isNew: true, + relevanceScore: calculateRelevanceScore(video, currentQuery), + })); + onVideos(newVideos, data.source); + if (data.pagecount && onPageInfo) { + onPageInfo(data.pagecount); + } + } else if (data.type === 'progress') { + onProgress(data.completedSources, data.totalVideosFound); + } else if (data.type === 'complete') { + if (!isCompleted) { isCompleted = true; if (data.maxPageCount && onPageInfo) { onPageInfo(data.maxPageCount); } onComplete(); - } else if (data.type === 'error') { + } + } else if (data.type === 'error') { + if (!isCompleted) { + isCompleted = true; onError(data.message); } - } catch (error) { - console.error('Error parsing stream data:', error); } + } catch (error) { + console.error('Error parsing stream data:', error); } } - } catch (error) { - if (timeoutId) clearTimeout(timeoutId); - throw error; - } finally { - if (timeoutId) clearTimeout(timeoutId); } } diff --git a/package-lock.json b/package-lock.json index 14c4294..f4dabde 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "kvideo", - "version": "4.9.12", + "version": "4.9.13", "lockfileVersion": 2, "requires": true, "packages": { "": { "name": "kvideo", - "version": "4.9.12", + "version": "4.9.13", "dependencies": { "@dnd-kit/core": "^6.3.1", "@dnd-kit/sortable": "^10.0.0", diff --git a/package.json b/package.json index 353c605..e20f61e 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "kvideo", - "version": "4.9.12", + "version": "4.9.13", "private": true, "scripts": { "dev": "node scripts/next-with-lan-access.mjs dev", diff --git a/tests/search-reliability.test.ts b/tests/search-reliability.test.ts new file mode 100644 index 0000000..6f922c5 --- /dev/null +++ b/tests/search-reliability.test.ts @@ -0,0 +1,104 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { withRetry } from '@/lib/api/http-utils'; +import { probeSourceLatency } from '@/lib/api/source-latency'; +import { processSearchStream } from '@/lib/utils/search-stream'; +import { probeLatencyTargets } from '@/lib/utils/latency'; + +test('search stream does not complete during a slow but active source search', async () => { + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('data: {"type":"start","totalSources":1}\n\n')); + + setTimeout(() => { + controller.enqueue(encoder.encode( + 'data: {"type":"videos","source":"slow","videos":[{"vod_id":1,"vod_name":"测试影片","source":"slow"}]}\n\n', + )); + controller.enqueue(encoder.encode( + 'data: {"type":"complete","totalVideosFound":1,"totalSources":1}\n\n', + )); + controller.close(); + }, 3200); + }, + }); + + let completionCount = 0; + const receivedTitles: string[] = []; + + await processSearchStream({ + reader: stream.getReader(), + currentQuery: '测试', + onStart: () => {}, + onVideos: (videos) => receivedTitles.push(...videos.map((video) => video.vod_name)), + onProgress: () => {}, + onComplete: () => { completionCount += 1; }, + onError: (message) => assert.fail(message), + }); + + assert.deepEqual(receivedTitles, ['测试影片']); + assert.equal(completionCount, 1); +}); + +test('GET fallback latency excludes the failed HEAD attempt duration', async () => { + const timestamps = [0, 5000, 5000, 5250]; + const methods: string[] = []; + const fetcher: typeof fetch = async (_input, init) => { + methods.push(init?.method ?? 'GET'); + if (init?.method === 'HEAD') { + throw new Error('HEAD unsupported'); + } + return new Response(null, { status: 200 }); + }; + + const result = await probeSourceLatency('https://example.com', { + fetcher, + now: () => timestamps.shift() ?? 5250, + }); + + assert.deepEqual(methods, ['HEAD', 'GET']); + assert.deepEqual(result, { + latency: 250, + success: true, + timeout: false, + method: 'GET', + }); +}); + +test('latency probes cap concurrent outbound requests', async () => { + const targets = Array.from({ length: 10 }, (_, index) => ({ + id: `source-${index}`, + baseUrl: `https://example.com/${index}`, + })); + let active = 0; + let maxActive = 0; + + const results = await probeLatencyTargets(targets, async () => { + active += 1; + maxActive = Math.max(maxActive, active); + await new Promise((resolve) => setTimeout(resolve, 10)); + active -= 1; + return 42; + }, 3); + + assert.equal(maxActive, 3); + assert.equal(results.length, targets.length); + assert.equal(results.every((result) => result.latency === 42), true); +}); + +test('aborted source requests are not retried', async () => { + let attempts = 0; + const abortError = new Error('aborted'); + abortError.name = 'AbortError'; + + await assert.rejects( + withRetry(async () => { + attempts += 1; + throw abortError; + }), + { name: 'AbortError' }, + ); + + assert.equal(attempts, 1); +});