diff --git a/frontend/app/api/cities/detail-batch/route.ts b/frontend/app/api/cities/detail-batch/route.ts index 8588b43a..d15f816a 100644 --- a/frontend/app/api/cities/detail-batch/route.ts +++ b/frontend/app/api/cities/detail-batch/route.ts @@ -33,6 +33,15 @@ function parseRequestedCities(req: NextRequest) { } function buildCityDetailBatchTimeoutPayload(requestedCities: string[]) { + const city_status = Object.fromEntries( + requestedCities.map((city) => [ + city, + { + status: "proxy_timeout", + duration_ms: null, + }, + ]), + ); return { cities: requestedCities, details: {}, @@ -40,6 +49,18 @@ function buildCityDetailBatchTimeoutPayload(requestedCities: string[]) { missing: requestedCities, partial: true, timeout: true, + diagnostics: { + version: 1, + response_source: "next_proxy_timeout", + partial: true, + partial_reason: "proxy_timeout", + requested_count: requestedCities.length, + completed_count: 0, + missing_count: requestedCities.length, + error_count: 0, + proxy_timeout_ms: DETAIL_BATCH_PROXY_TIMEOUT_MS, + city_status, + }, }; } diff --git a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx index 084f9dd9..882da2db 100644 --- a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx +++ b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx @@ -33,6 +33,7 @@ import { mergeRowObservationIntoHourly, normObs, prefersHighFrequencyRunwayResolution, + readCityDetailBatchDiagnostics, readSessionCache, selectCompactSecondaryTemp, selectDisplayRunwayTemp, @@ -134,6 +135,115 @@ function rowObservationSignature(row: ScanOpportunityRow | null) { return hasObservation ? pieces.map((piece) => String(piece ?? "")).join("|") : ""; } +type ChartDetailStatus = "idle" | "loading" | "fresh" | "stale_cache" | "degraded"; +type ChartDetailSource = + | "none" + | "memory_cache" + | "session_cache" + | "network" + | "force_refresh" + | "partial_or_timeout"; + +type ChartFreshnessState = { + rowAppliedAtMs: number | null; + rowObservationTime: string | null; + ssePatchAppliedAtMs: number | null; + sseObservedAt: string | null; + sseRevision: number | null; + sseEmittedAtMs: number | null; + sseServerToClientLatencySec: number | null; + sseCollectorToClientLatencySec: number | null; + sseSourceToCollectorLatencySec: number | null; + detailRequestedAtMs: number | null; + detailResolvedAtMs: number | null; + detailErrorAtMs: number | null; + detailStatus: ChartDetailStatus; + detailSource: ChartDetailSource; +}; + +function createChartFreshnessState( + overrides: Partial = {}, +): ChartFreshnessState { + return { + rowAppliedAtMs: null, + rowObservationTime: null, + ssePatchAppliedAtMs: null, + sseObservedAt: null, + sseRevision: null, + sseEmittedAtMs: null, + sseServerToClientLatencySec: null, + sseCollectorToClientLatencySec: null, + sseSourceToCollectorLatencySec: null, + detailRequestedAtMs: null, + detailResolvedAtMs: null, + detailErrorAtMs: null, + detailStatus: "idle", + detailSource: "none", + ...overrides, + }; +} + +function ageSeconds(fromMs: number | null | undefined, nowMs: number) { + if (!Number.isFinite(fromMs)) return null; + return Math.max(0, Math.round((nowMs - Number(fromMs)) / 1000)); +} + +function buildChartFreshnessContext(state: ChartFreshnessState, nowMs = Date.now()) { + const rowAgeSec = ageSeconds(state.rowAppliedAtMs, nowMs); + const sseAgeSec = ageSeconds(state.ssePatchAppliedAtMs, nowMs); + let livePath: "none" | "row" | "sse" = "none"; + let liveAgeSec: number | null = null; + if (rowAgeSec !== null) { + livePath = "row"; + liveAgeSec = rowAgeSec; + } + if (sseAgeSec !== null && (liveAgeSec === null || sseAgeSec <= liveAgeSec)) { + livePath = "sse"; + liveAgeSec = sseAgeSec; + } + + return { + live_path: livePath, + live_age_sec: liveAgeSec, + row_applied_age_sec: rowAgeSec, + row_observation_time: state.rowObservationTime, + sse_patch_age_sec: sseAgeSec, + sse_observed_at: state.sseObservedAt, + sse_revision: state.sseRevision, + sse_emitted_at_ms: state.sseEmittedAtMs, + sse_server_to_client_latency_sec: state.sseServerToClientLatencySec, + sse_collector_to_client_latency_sec: state.sseCollectorToClientLatencySec, + sse_source_to_collector_latency_sec: state.sseSourceToCollectorLatencySec, + detail_status: state.detailStatus, + detail_source: state.detailSource, + detail_request_age_sec: ageSeconds(state.detailRequestedAtMs, nowMs), + detail_age_sec: ageSeconds(state.detailResolvedAtMs, nowMs), + detail_error_age_sec: ageSeconds(state.detailErrorAtMs, nowMs), + }; +} + +function rowObservationTimeForFreshness(row: ScanOpportunityRow | null) { + if (!row) return null; + const metarContext = row.metar_context || null; + return String( + metarContext?.airport_obs_time || + metarContext?.last_observation_time || + metarContext?.last_time || + row.local_time || + "", + ).trim() || null; +} + +function patchObservationTimeForFreshness(patch: { changes?: Record } | null | undefined) { + const changes = patch?.changes || {}; + return String( + changes.observed_at_utc || + changes.observed_at_local || + changes.obs_time || + "", + ).trim() || null; +} + function getWundergroundDailyHigh(hourly: HourlyForecast) { return validNumber(hourly?.wundergroundCurrent?.max_so_far) ?? null; } @@ -458,6 +568,9 @@ export function LiveTemperatureThresholdChart({ const [detailError, setDetailError] = useState(null); const [detailRetryNonce, setDetailRetryNonce] = useState(0); const [showingStaleDetail, setShowingStaleDetail] = useState(false); + const [chartFreshness, setChartFreshness] = useState(() => + createChartFreshnessState(), + ); const hasLoadedHourlyDetailRef = useRef(false); const chartVisibilityRef = useRef(null); const lastPatchAtRef = useRef(Date.now()); @@ -486,8 +599,28 @@ export function LiveTemperatureThresholdChart({ formatCityLocalDate(row?.tz_offset_seconds), ); const currentRowObservationSignature = useMemo(() => rowObservationSignature(row), [row]); + const markDetailRequest = useCallback((source: ChartDetailSource) => { + setChartFreshness((prev) => ({ + ...prev, + detailRequestedAtMs: Date.now(), + detailStatus: "loading", + detailSource: source, + })); + }, []); + const markDetailDegraded = useCallback((options?: { showUserError?: boolean }) => { + if (options?.showUserError) { + setDetailError(isEn ? "Detail temporarily unavailable." : "详情暂不可用"); + } + setChartFreshness((prev) => ({ + ...prev, + detailErrorAtMs: Date.now(), + detailStatus: "degraded", + detailSource: "partial_or_timeout", + })); + }, [isEn]); useEffect(() => { + const now = Date.now(); setUserToggledKeys({}); setZoomRange(null); setViewMode("full"); @@ -499,8 +632,15 @@ export function LiveTemperatureThresholdChart({ setDetailError(null); setDetailRetryNonce(0); setShowingStaleDetail(false); + setChartFreshness(createChartFreshnessState({ + rowAppliedAtMs: currentRowObservationSignature ? now : null, + rowObservationTime: rowObservationTimeForFreshness(row), + detailRequestedAtMs: city && detailLoadDelayMs === 0 ? now : null, + detailStatus: city && detailLoadDelayMs === 0 ? "loading" : "idle", + detailSource: city && detailLoadDelayMs === 0 ? "network" : "none", + })); hasLoadedHourlyDetailRef.current = false; - lastPatchAtRef.current = Date.now(); + lastPatchAtRef.current = now; lastAppliedPatchRevisionRef.current = 0; lastProbabilityRefreshAtRef.current = 0; lastForegroundRefreshAtRef.current = 0; @@ -551,6 +691,7 @@ export function LiveTemperatureThresholdChart({ const applySuccessfulHourlyDetail = useCallback((data: HourlyForecast, options?: { updateLiveTemp?: boolean }) => { if (!data) return; + const loadedAtMs = Date.now(); const rowSeed = seedHourlyForecastFromRow(row); const dataWithCurrentRow = mergeHourlyWithLiveObservations(data, rowSeed, row); hasLoadedHourlyDetailRef.current = true; @@ -561,16 +702,32 @@ export function LiveTemperatureThresholdChart({ setHourly((prev) => mergeHourlyWithLiveObservations(dataWithCurrentRow, prev, row)); setDetailError(null); setShowingStaleDetail(false); + setChartFreshness((prev) => ({ + ...prev, + detailResolvedAtMs: loadedAtMs, + detailErrorAtMs: null, + detailStatus: "fresh", + detailSource: + prev.detailSource === "none" || prev.detailSource === "partial_or_timeout" + ? "network" + : prev.detailSource, + })); }, [row]); useEffect(() => { if (!city || !currentRowObservationSignature) return; if (lastRowObservationSignatureRef.current === currentRowObservationSignature) return; + const now = Date.now(); lastRowObservationSignatureRef.current = currentRowObservationSignature; const rowSeed = seedHourlyForecastFromRow(row); const temp = getLiveTempFromHourly(rowSeed); if (temp !== null) setLiveTemp(temp); setHourly((prev) => mergeRowObservationIntoHourly(prev ?? rowSeed, row)); + setChartFreshness((prev) => ({ + ...prev, + rowAppliedAtMs: now, + rowObservationTime: rowObservationTimeForFreshness(row), + })); }, [city, currentRowObservationSignature, row]); useEffect(() => { @@ -581,10 +738,12 @@ export function LiveTemperatureThresholdChart({ const cacheKey = `${city}:${targetResolution}`; let cached = _hourlyCache.get(cacheKey); + let cachedSource: ChartDetailSource = cached ? "memory_cache" : "none"; if (!cached || Date.now() - Number(cached.ts || 0) >= HOURLY_CACHE_TTL_MS) { const sessionEntry = readSessionCache(cacheKey, { allowStale: true }); if (sessionEntry) { cached = sessionEntry; + cachedSource = "session_cache"; _hourlyCache.set(cacheKey, sessionEntry); } } @@ -595,6 +754,12 @@ export function LiveTemperatureThresholdChart({ hasLoadedHourlyDetailRef.current = true; setHourly(cached.data); setShowingStaleDetail(!hasFreshCache); + setChartFreshness((prev) => ({ + ...prev, + detailResolvedAtMs: Number(cached.ts || Date.now()), + detailStatus: hasFreshCache ? "fresh" : "stale_cache", + detailSource: cachedSource, + })); } if (hasFreshCache) { @@ -625,6 +790,7 @@ export function LiveTemperatureThresholdChart({ setShowingStaleDetail(false); } setIsHourlyLoading(true); + markDetailRequest("network"); let cancelled = false; const timer = setTimeout(() => { @@ -632,13 +798,15 @@ export function LiveTemperatureThresholdChart({ .then((data) => { if (cancelled) return; if (!data) { - setDetailError(isEn ? "Data temporarily unavailable." : "数据暂不可用"); + markDetailDegraded({ showUserError: true }); return; } applySuccessfulHourlyDetail(data); }) .catch(() => { - if (!cancelled) setDetailError(isEn ? "Data temporarily unavailable." : "数据暂不可用"); + if (!cancelled) { + markDetailDegraded({ showUserError: true }); + } }) .finally(() => { if (!cancelled) setIsHourlyLoading(false); @@ -660,17 +828,29 @@ export function LiveTemperatureThresholdChart({ slotIndex, detailLoadReady, detailRetryNonce, - isEn, + markDetailDegraded, + markDetailRequest, applySuccessfulHourlyDetail, ]); useEffect(() => { if (!latestPatch || latestPatch.revision <= lastAppliedPatchRevisionRef.current) return; + const patchAppliedAtMs = Date.now(); lastAppliedPatchRevisionRef.current = latestPatch.revision; - lastPatchAtRef.current = Date.now(); + lastPatchAtRef.current = patchAppliedAtMs; const tempValue = validNumber(latestPatch.changes.temp); if (tempValue !== null) setLiveTemp(tempValue); setHourly((prev) => mergePatchIntoHourly(prev ?? seedHourlyForecastFromRow(row), latestPatch)); + setChartFreshness((prev) => ({ + ...prev, + ssePatchAppliedAtMs: patchAppliedAtMs, + sseObservedAt: patchObservationTimeForFreshness(latestPatch), + sseRevision: latestPatch.revision, + sseEmittedAtMs: latestPatch.delivery?.sse_emitted_at_ms ?? null, + sseServerToClientLatencySec: latestPatch.delivery?.server_to_client_latency_sec ?? null, + sseCollectorToClientLatencySec: latestPatch.delivery?.collector_to_client_latency_sec ?? null, + sseSourceToCollectorLatencySec: latestPatch.delivery?.source_to_collector_latency_sec ?? null, + })); const hasObservationChange = tempValue !== null || @@ -684,36 +864,54 @@ export function LiveTemperatureThresholdChart({ let cancelled = false; const refreshProbabilityOverlayAfterPatch = () => { + markDetailRequest("force_refresh"); fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution }) .then((data) => { - if (cancelled || !data) return; + if (cancelled) return; + if (!data) { + markDetailDegraded(); + return; + } applySuccessfulHourlyDetail(data); }) - .catch(() => {}); + .catch(() => { + if (!cancelled) { + markDetailDegraded(); + } + }); }; refreshProbabilityOverlayAfterPatch(); return () => { cancelled = true; }; - }, [latestPatch, row, city, targetResolution, compact, isActive, isMaximized, applySuccessfulHourlyDetail]); + }, [latestPatch, row, city, targetResolution, compact, isActive, isMaximized, markDetailDegraded, markDetailRequest, applySuccessfulHourlyDetail]); useEffect(() => { if (!resyncVersion || !city) return; let cancelled = false; + markDetailRequest("force_refresh"); fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution }) .then((data) => { - if (cancelled || !data) return; + if (cancelled) return; + if (!data) { + markDetailDegraded(); + return; + } applySuccessfulHourlyDetail(data); }) - .catch(() => {}) + .catch(() => { + if (!cancelled) { + markDetailDegraded(); + } + }) .finally(() => { if (!cancelled) setIsHourlyLoading(false); }); return () => { cancelled = true; }; - }, [resyncVersion, city, targetResolution, applySuccessfulHourlyDetail]); + }, [resyncVersion, city, targetResolution, markDetailDegraded, markDetailRequest, applySuccessfulHourlyDetail]); // ── SSE fallback: only full-fetch if a visible chart has seen no patch for one METAR cadence ── useEffect(() => { @@ -721,14 +919,24 @@ export function LiveTemperatureThresholdChart({ let cancelled = false; const refreshFullDetail = () => { - lastPatchAtRef.current = Date.now(); + const now = Date.now(); + lastPatchAtRef.current = now; + markDetailRequest("force_refresh"); fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution }) .then((data) => { - if (cancelled || !data) return; + if (cancelled) return; + if (!data) { + markDetailDegraded(); + return; + } applySuccessfulHourlyDetail(data, { updateLiveTemp: true }); }) - .catch(() => {}) + .catch(() => { + if (!cancelled) { + markDetailDegraded(); + } + }) .finally(() => { if (!cancelled) setIsHourlyLoading(false); }); @@ -746,7 +954,7 @@ export function LiveTemperatureThresholdChart({ cancelled = true; clearInterval(id); }; - }, [city, compact, isActive, isMaximized, targetResolution, applySuccessfulHourlyDetail]); + }, [city, compact, isActive, isMaximized, targetResolution, markDetailDegraded, markDetailRequest, applySuccessfulHourlyDetail]); useEffect(() => { if (!shouldPollLiveChart({ city, compact, isActive, isMaximized })) return; @@ -766,13 +974,22 @@ export function LiveTemperatureThresholdChart({ lastForegroundRefreshAtRef.current = now; lastPatchAtRef.current = now; + markDetailRequest("force_refresh"); fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution }) .then((data) => { - if (cancelled || !data) return; + if (cancelled) return; + if (!data) { + markDetailDegraded(); + return; + } applySuccessfulHourlyDetail(data, { updateLiveTemp: true }); }) - .catch(() => {}); + .catch(() => { + if (!cancelled) { + markDetailDegraded(); + } + }); }; const handleVisibilityChange = () => { @@ -787,7 +1004,7 @@ export function LiveTemperatureThresholdChart({ document.removeEventListener("visibilitychange", handleVisibilityChange); window.removeEventListener("focus", refreshForegroundFullDetail); }; - }, [city, compact, isActive, isMaximized, targetResolution, applySuccessfulHourlyDetail]); + }, [city, compact, isActive, isMaximized, targetResolution, markDetailDegraded, markDetailRequest, applySuccessfulHourlyDetail]); useEffect(() => { if (!city || !currentCityLocalDate) return; @@ -797,19 +1014,27 @@ export function LiveTemperatureThresholdChart({ localDayRolloverFetchDateRef.current = currentCityLocalDate; let cancelled = false; + markDetailRequest("force_refresh"); fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution }) .then((data) => { - if (cancelled || !data) return; + if (cancelled) return; + if (!data) { + markDetailDegraded(); + return; + } applySuccessfulHourlyDetail(data); }) .catch(() => { - if (!cancelled) localDayRolloverFetchDateRef.current = ""; + if (!cancelled) { + localDayRolloverFetchDateRef.current = ""; + markDetailDegraded(); + } }); return () => { cancelled = true; }; - }, [city, currentCityLocalDate, hourly?.localDate, row?.local_date, targetResolution, applySuccessfulHourlyDetail]); + }, [city, currentCityLocalDate, hourly?.localDate, row?.local_date, targetResolution, markDetailDegraded, markDetailRequest, applySuccessfulHourlyDetail]); const chartHourly = useMemo(() => { if (!hourly) return hourly; @@ -1012,6 +1237,14 @@ export function LiveTemperatureThresholdChart({ const subtitle = row ? (isEn ? "Live & Forecast" : "实测与预测") : ""; const showDetailErrorBadge = !compact || isActive || isMaximized; + const chartFreshnessContext = useMemo( + () => buildChartFreshnessContext(chartFreshness), + [chartFreshness], + ); + const detailBatchDiagnostics = useMemo( + () => readCityDetailBatchDiagnostics(city, targetResolution), + [city, targetResolution, chartFreshness.detailRequestedAtMs, chartFreshness.detailErrorAtMs, chartFreshness.detailResolvedAtMs], + ); const handleZoomReset = useCallback(() => { setZoomRange(null); @@ -1072,8 +1305,9 @@ export function LiveTemperatureThresholdChart({ const handleRetryDetail = useCallback(() => { setDetailError(null); setIsHourlyLoading(true); + markDetailRequest("network"); setDetailRetryNonce((value) => value + 1); - }, []); + }, [markDetailRequest]); const handleReportIssue = useCallback((event: React.MouseEvent) => { event.stopPropagation(); @@ -1087,6 +1321,12 @@ export function LiveTemperatureThresholdChart({ is_active: isActive, is_maximized: isMaximized, detail_error: detailError, + freshness: chartFreshnessContext, + detail_batch_diagnostics: detailBatchDiagnostics, + live_path: chartFreshnessContext.live_path, + live_age_sec: chartFreshnessContext.live_age_sec, + detail_status: chartFreshnessContext.detail_status, + detail_source: chartFreshnessContext.detail_source, is_hourly_loading: isHourlyLoading, showing_stale_detail: showingStaleDetail, target_resolution: targetResolution, @@ -1104,11 +1344,13 @@ export function LiveTemperatureThresholdChart({ }, [ activeSeries, chartLocalDate, + chartFreshnessContext, chartSeries, city, compact, currentMetarTemp, currentRunwayTemp, + detailBatchDiagnostics, detailError, hasRunwayData, isActive, @@ -1304,6 +1546,7 @@ export function LiveTemperatureThresholdChart({ showRunwayDetails={showRunwayDetails} isHourlyLoading={isHourlyLoading} detailError={detailError} + detailStatus={chartFreshness.detailStatus} showingStaleDetail={showingStaleDetail} showDetailErrorBadge={showDetailErrorBadge} refAreaLeft={refAreaLeft} @@ -1342,6 +1585,7 @@ export const __getWundergroundDailyHighForTest = getWundergroundDailyHigh; export const __getInitialDetailLoadDelayMsForTest = getInitialDetailLoadDelayMs; export const __shouldFetchCityDetailForChartForTest = shouldFetchCityDetailForChart; export const __shouldPollLiveChartForTest = shouldPollLiveChart; +export const __buildChartFreshnessContextForTest = buildChartFreshnessContext; export const __mergePatchIntoHourlyForTest = mergePatchIntoHourly; export const __selectCompactSecondaryTempForTest = selectCompactSecondaryTemp; export const __selectDisplayRunwayTempForTest = selectDisplayRunwayTemp; diff --git a/frontend/components/dashboard/scan-terminal/TemperatureChartCanvas.tsx b/frontend/components/dashboard/scan-terminal/TemperatureChartCanvas.tsx index 4759c370..f3f51093 100644 --- a/frontend/components/dashboard/scan-terminal/TemperatureChartCanvas.tsx +++ b/frontend/components/dashboard/scan-terminal/TemperatureChartCanvas.tsx @@ -111,6 +111,7 @@ function TemperatureChartCanvasComponent({ showRunwayDetails, isHourlyLoading, detailError, + detailStatus, showingStaleDetail, showDetailErrorBadge = true, refAreaLeft, @@ -139,6 +140,7 @@ function TemperatureChartCanvasComponent({ showRunwayDetails: boolean; isHourlyLoading: boolean; detailError?: string | null; + detailStatus?: string | null; showingStaleDetail?: boolean; showDetailErrorBadge?: boolean; refAreaLeft: number | null; @@ -226,6 +228,10 @@ function TemperatureChartCanvasComponent({ const shouldShowUnavailableState = Boolean(row?.city) && Boolean(detailError) && !isHourlyLoading && !hasDrawableChartContent; const shouldShowBackgroundError = showDetailErrorBadge && Boolean(row?.city) && Boolean(detailError) && !isHourlyLoading && hasDrawableChartContent; + const backgroundErrorLabel = + showingStaleDetail || detailStatus === "stale_cache" + ? (isEn ? "Detail cache" : "详情缓存") + : (isEn ? "Detail degraded" : "详情降级"); return (
@@ -389,7 +395,7 @@ function TemperatureChartCanvasComponent({ {shouldShowUnavailableState && (
-
{isEn ? "Data temporarily unavailable" : "数据暂不可用"}
+
{isEn ? "Detail temporarily unavailable" : "详情暂不可用"}
)} + {freshnessBadges.length > 0 && ( +
+ {freshnessBadges.map((item) => ( + + {item.label} + + ))} +
+ )} {row.user_email || row.user_id || "—"} diff --git a/frontend/hooks/use-sse-patches.ts b/frontend/hooks/use-sse-patches.ts index 948425b5..6b2e4085 100644 --- a/frontend/hooks/use-sse-patches.ts +++ b/frontend/hooks/use-sse-patches.ts @@ -15,6 +15,13 @@ export type CityPatch = { changes: Record; revision: number; ts?: number; + delivery?: { + client_received_at_ms: number; + sse_emitted_at_ms?: number | null; + server_to_client_latency_sec?: number | null; + collector_to_client_latency_sec?: number | null; + source_to_collector_latency_sec?: number | null; + }; }; type ObservationPatchV1 = { @@ -30,6 +37,7 @@ type ObservationPatchV1 = { source_cadence_sec?: number | null; revision?: number; ts?: number; + sse_emitted_at_ms?: number | null; payload?: Record; }; @@ -105,6 +113,34 @@ function resolveSseReplayLimit(cityCount: number) { return Math.max(SSE_REPLAY_BASE_LIMIT, Math.min(SSE_REPLAY_MAX_LIMIT, requested)); } +function finiteNumber(value: unknown): number | null { + const number = Number(value); + return Number.isFinite(number) ? number : null; +} + +function latencySeconds(fromMs: number | null, toMs: number) { + if (fromMs === null) return null; + return Math.max(0, Math.round((toMs - fromMs) / 1000)); +} + +function buildPatchDelivery( + patch: Partial & ObservationPatchV1, + clientReceivedAtMs: number, + payload?: Record, +) { + const sseEmittedAtMs = finiteNumber(patch.sse_emitted_at_ms); + const collectorReceivedAtMs = finiteNumber(patch.ts); + const sourceToCollectorLatencySec = + finiteNumber(payload?.latency_sec) ?? finiteNumber((patch.changes as Record | undefined)?.latency_sec); + return { + client_received_at_ms: clientReceivedAtMs, + sse_emitted_at_ms: sseEmittedAtMs, + server_to_client_latency_sec: latencySeconds(sseEmittedAtMs, clientReceivedAtMs), + collector_to_client_latency_sec: latencySeconds(collectorReceivedAtMs, clientReceivedAtMs), + source_to_collector_latency_sec: sourceToCollectorLatencySec, + }; +} + function currentConnectionKey() { return `${useFallbackUrl ? "fallback" : "direct"}:${subscribedCityList().join("|")}:${lastRevision}`; } @@ -219,7 +255,10 @@ function registerCitySubscription(city: string) { }; } -function normalizeLegacyPatch(patch: Partial): CityPatch | null { +function normalizeLegacyPatch( + patch: Partial & ObservationPatchV1, + clientReceivedAtMs: number, +): CityPatch | null { const city = normalizeCityKey(patch.city); const changes = patch.changes; const revision = Number(patch.revision); @@ -232,10 +271,14 @@ function normalizeLegacyPatch(patch: Partial): CityPatch | null { changes: changes as Record, revision, ts: typeof patch.ts === "number" ? patch.ts : Date.now(), + delivery: buildPatchDelivery(patch, clientReceivedAtMs), }; } -function normalizeV1Patch(patch: ObservationPatchV1): CityPatch | null { +function normalizeV1Patch( + patch: ObservationPatchV1, + clientReceivedAtMs: number, +): CityPatch | null { const city = normalizeCityKey(patch.city); const revision = Number(patch.revision); const payload = patch.payload; @@ -262,17 +305,19 @@ function normalizeV1Patch(patch: ObservationPatchV1): CityPatch | null { changes, revision, ts: typeof patch.ts === "number" ? patch.ts : Date.now(), + delivery: buildPatchDelivery(patch, clientReceivedAtMs, payload), }; } function normalizeIncomingPatch(payload: unknown): CityPatch | null { if (!payload || typeof payload !== "object") return null; const patch = payload as Partial & ObservationPatchV1; + const clientReceivedAtMs = Date.now(); if (patch.type === "city_patch" || !patch.type) { - return normalizeLegacyPatch(patch); + return normalizeLegacyPatch(patch, clientReceivedAtMs); } if (patch.type === V1_EVENT_TYPE) { - return normalizeV1Patch(patch); + return normalizeV1Patch(patch, clientReceivedAtMs); } return null; } diff --git a/tests/test_sse_replay.py b/tests/test_sse_replay.py index bdbc86f6..0e1ea500 100644 --- a/tests/test_sse_replay.py +++ b/tests/test_sse_replay.py @@ -4,6 +4,7 @@ from fastapi.testclient import TestClient from web.app import app from web.routers import sse_router +from web.sse_manager import SseManager def _decode_sse_events(text: str): @@ -15,6 +16,25 @@ def _decode_sse_events(text: str): return events +def test_sse_format_event_stamps_emit_time_for_latency_diagnostics(monkeypatch): + monkeypatch.setattr("web.sse_manager.time.time", lambda: 1780750904.5) + + frame = SseManager._format_event( + { + "type": "city_observation_patch.v1", + "revision": 123, + "city": "busan", + "source": "amos", + "ts": 1780750864062, + "payload": {"temp": 23.0}, + } + ) + + event = _decode_sse_events(frame)[0] + assert event["sse_emitted_at_ms"] == 1780750904500 + assert event["payload"]["temp"] == 23.0 + + def test_events_endpoint_replays_only_requested_cities(monkeypatch): captured = {} diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 166b0073..ecfb8b8a 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -1012,6 +1012,13 @@ def test_city_detail_batch_returns_busy_when_global_builder_slot_is_full(monkeyp assert payload["busy"] is True assert payload["details"] == {} assert payload["missing"] == ["paris", "shanghai"] + assert payload["diagnostics"]["partial_reason"] == "busy" + assert payload["diagnostics"]["response_source"] == "busy" + assert payload["diagnostics"]["requested_count"] == 2 + assert payload["diagnostics"]["completed_count"] == 0 + assert payload["diagnostics"]["missing_count"] == 2 + assert payload["diagnostics"]["city_status"]["paris"]["status"] == "busy" + assert payload["diagnostics"]["city_status"]["shanghai"]["status"] == "busy" assert build_calls == 0 @@ -1102,6 +1109,17 @@ def test_city_detail_batch_returns_completed_details_when_one_city_is_slow(monke assert payload["partial"] is True assert payload["missing"] == ["slow"] assert payload["errors"] == {} + assert payload["diagnostics"]["partial_reason"] == "timeout" + assert payload["diagnostics"]["requested_count"] == 3 + assert payload["diagnostics"]["completed_count"] == 2 + assert payload["diagnostics"]["missing_count"] == 1 + assert payload["diagnostics"]["error_count"] == 0 + assert payload["diagnostics"]["batch_concurrency"] == 2 + assert payload["diagnostics"]["partial_timeout_ms"] == 20 + assert payload["diagnostics"]["city_status"]["fast"]["status"] == "ok" + assert payload["diagnostics"]["city_status"]["other"]["status"] == "ok" + assert payload["diagnostics"]["city_status"]["slow"]["status"] == "timeout" + assert isinstance(payload["diagnostics"]["city_status"]["fast"]["duration_ms"], (int, float)) assert "slow" not in completed diff --git a/web/services/city_api.py b/web/services/city_api.py index 7f339cd3..4147193e 100644 --- a/web/services/city_api.py +++ b/web/services/city_api.py @@ -973,6 +973,90 @@ def _city_detail_batch_partial_timeout_seconds() -> Optional[float]: return max(0.001, min(60.0, timeout_ms / 1000.0)) +def _city_detail_batch_partial_timeout_ms() -> Optional[int]: + timeout_sec = _city_detail_batch_partial_timeout_seconds() + if timeout_sec is None: + return None + return int(round(timeout_sec * 1000.0)) + + +def _city_detail_batch_partial_reason( + *, + busy: bool, + missing: List[str], + errors: Dict[str, str], +) -> Optional[str]: + if busy: + return "busy" + if missing and errors: + return "timeout_error" + if missing: + return "timeout" + if errors: + return "error" + return None + + +def _build_city_detail_batch_diagnostics( + *, + city_names: List[str], + details: Dict[str, Any], + errors: Dict[str, str], + missing: List[str], + resolution: Optional[str], + detail_scope: str, + force_refresh: bool, + response_source: str, + busy: bool = False, + city_durations_ms: Optional[Dict[str, float]] = None, +) -> Dict[str, Any]: + city_durations_ms = city_durations_ms or {} + missing_set = set(missing) + error_set = set(errors) + detail_set = set(details) + partial_reason = _city_detail_batch_partial_reason( + busy=busy, + missing=missing, + errors=errors, + ) + city_status: Dict[str, Dict[str, Any]] = {} + for city in city_names: + if busy: + status = "busy" + elif city in missing_set: + status = "timeout" + elif city in error_set: + status = "error" + elif city in detail_set: + status = "ok" + else: + status = "missing" + city_status[city] = { + "status": status, + "duration_ms": city_durations_ms.get(city), + } + if city in errors: + city_status[city]["error"] = errors[city] + + return { + "version": 1, + "response_source": response_source, + "partial": bool(partial_reason), + "partial_reason": partial_reason, + "requested_count": len(city_names), + "completed_count": len(details), + "missing_count": len(missing), + "error_count": len(errors), + "batch_concurrency": _city_detail_batch_concurrency(), + "global_concurrency": _city_detail_batch_global_concurrency(), + "partial_timeout_ms": _city_detail_batch_partial_timeout_ms(), + "force_refresh": force_refresh, + "resolution": resolution, + "scope": detail_scope, + "city_status": city_status, + } + + async def get_city_detail_batch_payload( request: Request, *, @@ -1014,30 +1098,52 @@ async def get_city_detail_batch_payload( async def _build_uncached_payload() -> Dict[str, Any]: build_semaphore = _city_detail_batch_build_semaphore() if not build_semaphore.acquire(blocking=False): + missing = list(city_names) + errors: Dict[str, str] = {} + details: Dict[str, Any] = {} return { "cities": city_names, - "details": {}, - "errors": {}, - "missing": list(city_names), + "details": details, + "errors": errors, + "missing": missing, "partial": True, "busy": True, "stale_reason": "city detail batch builder is busy", + "diagnostics": _build_city_detail_batch_diagnostics( + city_names=city_names, + details=details, + errors=errors, + missing=missing, + resolution=resolution, + detail_scope=detail_scope, + force_refresh=force_refresh, + response_source="busy", + busy=True, + ), } try: semaphore = asyncio.Semaphore(_city_detail_batch_concurrency()) + city_durations_ms: Dict[str, float] = {} async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]: async with semaphore: - return await _build_city_detail_batch_item_async( - city, - force_refresh=force_refresh, - market_slug=market_slug, - target_date=target_date, - resolution=resolution, - detail_scope=detail_scope, - timing_recorder=timer, - ) + started = time.perf_counter() + try: + return await _build_city_detail_batch_item_async( + city, + force_refresh=force_refresh, + market_slug=market_slug, + target_date=target_date, + resolution=resolution, + detail_scope=detail_scope, + timing_recorder=timer, + ) + finally: + city_durations_ms[city] = round( + (time.perf_counter() - started) * 1000.0, + 1, + ) task_by_city = { city: asyncio.create_task(_build_with_limit(city)) @@ -1076,6 +1182,17 @@ async def get_city_detail_batch_payload( "errors": errors, "missing": missing, "partial": bool(missing or errors), + "diagnostics": _build_city_detail_batch_diagnostics( + city_names=city_names, + details=details, + errors=errors, + missing=missing, + resolution=resolution, + detail_scope=detail_scope, + force_refresh=force_refresh, + response_source="fresh_build", + city_durations_ms=city_durations_ms, + ), } finally: build_semaphore.release() diff --git a/web/sse_manager.py b/web/sse_manager.py index c74b3006..c1f49ad4 100644 --- a/web/sse_manager.py +++ b/web/sse_manager.py @@ -162,6 +162,11 @@ class SseManager: @staticmethod def _format_event(event: dict[str, Any]) -> str: + if str(event.get("type") or "").startswith("city_observation_patch"): + event = { + **event, + "sse_emitted_at_ms": int(time.time() * 1000), + } return f"data: {json.dumps(event, ensure_ascii=False, separators=(',', ':'))}\n\n"