Merge pull request #211 from KuekHaoYang/agent/fix-issue-210-search-latency

修复慢源搜索提前结束与延迟虚高
2026-08-13 07:43:43 +08:00 · 2026-07-13 21:15:17 +08:00
parent 2ae0a03366 66d8564dd2
commit 8d558c79cf
14 changed files with 364 additions and 140 deletions
@@ -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

 - 修复窄屏导航栏隐藏整个用户区域、导致登录用户无法退出的问题；退出按钮现在在所有视口宽度下保持可见，用户身份详情仍在桌面端显示。
@@ -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",
@@ -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(
@@ -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]
  );

@@ -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<T>(
        } 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)));
            }
@@ -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<ProbeAttemptResult> {
+  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<SourceLatencyResult> {
+  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);
+}
@@ -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<LatencyState>({});
    const [isLoading, setIsLoading] = useState(false);
-    const intervalRef = useRef<NodeJS.Timeout | null>(null);
+    const intervalRef = useRef<ReturnType<typeof setTimeout> | 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<number | null> => {
+    const pingSource = useCallback(async (baseUrl: string): Promise<number | null> => {
+        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 }));
                }
@@ -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<typeof useSearchState>;
+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<string, { count: number; name: string }>();
+            const sourceConfigs = new Map<string, SearchSourceConfig>(
+                targetSources.map((source: SearchSourceConfig) => [source.id, source])
+            );
+            const sourcesMap = new Map<string, { count: number; name: string; baseUrl?: string }>();

            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);

@@ -51,6 +51,7 @@ export interface SourceBadge {
  id: string;
  name: string;
  count: number;
+  baseUrl?: string;
  typeName?: string;
 }

@@ -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<number | null>,
+  concurrency: number = 4,
+): Promise<LatencyProbeResult[]> {
+  if (targets.length === 0) return [];
+
+  const results = new Array<LatencyProbeResult>(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
@@ -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);
    }
 }
@@ -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",
@@ -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",
@@ -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<Uint8Array>({
+    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);
+});