Compare commits

..

2 Commits

Author SHA1 Message Date
2569718930@qq.com 372d4366a8 Instrument API timing and reduce detail fallbacks 2026-05-31 19:01:07 +08:00
2569718930@qq.com 668f4d9bd3 Optimize terminal batching and cache guards 2026-05-31 18:28:50 +08:00
25 changed files with 1287 additions and 196 deletions
+13 -3
View File
@@ -1,6 +1,10 @@
import { NextRequest, NextResponse } from "next/server"; import { NextRequest, NextResponse } from "next/server";
import { proxyBackendJsonGet } from "@/lib/api-proxy"; import { proxyBackendJsonGet } from "@/lib/api-proxy";
import { buildCityDetailProxyCachePolicy } from "@/lib/proxy-cache-policy"; import { buildCityDetailProxyCachePolicy } from "@/lib/proxy-cache-policy";
import {
createProxyTimer,
finishProxyTimedResponse,
} from "@/lib/proxy-timing";
const API_BASE = process.env.POLYWEATHER_API_BASE_URL; const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
const DETAIL_BATCH_PROXY_TIMEOUT_MS = Number( const DETAIL_BATCH_PROXY_TIMEOUT_MS = Number(
@@ -8,10 +12,15 @@ const DETAIL_BATCH_PROXY_TIMEOUT_MS = Number(
); );
export async function GET(req: NextRequest) { export async function GET(req: NextRequest) {
const timer = createProxyTimer(req, "city_detail_batch");
if (!API_BASE) { if (!API_BASE) {
return NextResponse.json( return finishProxyTimedResponse(
{ error: "POLYWEATHER_API_BASE_URL is not configured" }, NextResponse.json(
{ status: 500 }, { error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 },
),
timer,
"missing_api_base",
); );
} }
@@ -39,6 +48,7 @@ export async function GET(req: NextRequest) {
revalidateSeconds: cachePolicy.revalidateSeconds, revalidateSeconds: cachePolicy.revalidateSeconds,
signal: controller.signal, signal: controller.signal,
timeoutPublicMessage: "City detail batch request timed out", timeoutPublicMessage: "City detail batch request timed out",
timing: timer,
url: `${API_BASE}/api/cities/detail-batch?${searchParams.toString()}`, url: `${API_BASE}/api/cities/detail-batch?${searchParams.toString()}`,
}); });
} finally { } finally {
+38 -16
View File
@@ -9,6 +9,10 @@ import {
} from "@/lib/api-proxy"; } from "@/lib/api-proxy";
import { buildCachedJsonResponse } from "@/lib/http-cache"; import { buildCachedJsonResponse } from "@/lib/http-cache";
import { buildCityDetailProxyCachePolicy } from "@/lib/proxy-cache-policy"; import { buildCityDetailProxyCachePolicy } from "@/lib/proxy-cache-policy";
import {
createProxyTimer,
finishProxyTimedResponse,
} from "@/lib/proxy-timing";
const API_BASE = process.env.POLYWEATHER_API_BASE_URL; const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
@@ -34,15 +38,16 @@ export async function GET(
req: NextRequest, req: NextRequest,
context: { params: Promise<{ name: string }> }, context: { params: Promise<{ name: string }> },
) { ) {
const timer = createProxyTimer(req, "city_detail");
if (!API_BASE) { if (!API_BASE) {
const response = NextResponse.json( const response = NextResponse.json(
{ error: "POLYWEATHER_API_BASE_URL is not configured" }, { error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 }, { status: 500 },
); );
return response; return finishProxyTimedResponse(response, timer, "missing_api_base");
} }
const { name } = await context.params; const { name } = await timer.measure("route_params", () => context.params);
const forceRefresh = req.nextUrl.searchParams.get("force_refresh") ?? "false"; const forceRefresh = req.nextUrl.searchParams.get("force_refresh") ?? "false";
const cachePolicy = buildCityDetailProxyCachePolicy(forceRefresh, 15); const cachePolicy = buildCityDetailProxyCachePolicy(forceRefresh, 15);
const depth = req.nextUrl.searchParams.get("depth"); const depth = req.nextUrl.searchParams.get("depth");
@@ -67,31 +72,48 @@ export async function GET(
const url = `${API_BASE}/api/city/${encodeURIComponent(name)}/detail?${searchParams.toString()}`; const url = `${API_BASE}/api/city/${encodeURIComponent(name)}/detail?${searchParams.toString()}`;
try { try {
const auth = await buildBackendRequestHeaders(req, { const auth = await timer.measure("auth_headers", () =>
includeSupabaseIdentity: false, buildBackendRequestHeaders(req, {
}); includeSupabaseIdentity: false,
const res = await fetch(url, { }),
headers: auth.headers, );
...(cachePolicy.fetchMode === "no-store" const res = await timer.measure("backend_fetch", () =>
? { cache: "no-store" as const } fetch(url, {
: { next: { revalidate: cachePolicy.revalidateSeconds ?? 15 } }), headers: auth.headers,
}); ...(cachePolicy.fetchMode === "no-store"
? { cache: "no-store" as const }
: { next: { revalidate: cachePolicy.revalidateSeconds ?? 15 } }),
}),
);
const backendServerTiming = res.headers.get("server-timing") || "";
if (!res.ok) { if (!res.ok) {
const raw = await res.text(); const raw = await timer.measure("backend_read", () => res.text());
const response = buildUpstreamErrorResponse(res.status, raw); const response = buildUpstreamErrorResponse(res.status, raw);
return applyAuthResponseCookies(response, auth.response); return finishProxyTimedResponse(
applyAuthResponseCookies(response, auth.response),
timer,
`upstream_${res.status}`,
{ backendServerTiming },
);
} }
const data = normalizeCityDetailPayload(await res.json()); const data = normalizeCityDetailPayload(
await timer.measure("backend_read", () => res.json()),
);
const response = buildCachedJsonResponse( const response = buildCachedJsonResponse(
req, req,
data, data,
cachePolicy.responseCacheControl, cachePolicy.responseCacheControl,
); );
return applyAuthResponseCookies(response, auth.response); return finishProxyTimedResponse(
applyAuthResponseCookies(response, auth.response),
timer,
"ok",
{ backendServerTiming },
);
} catch (error) { } catch (error) {
const response = buildProxyExceptionResponse(error, { const response = buildProxyExceptionResponse(error, {
publicMessage: "Failed to fetch city detail aggregate", publicMessage: "Failed to fetch city detail aggregate",
}); });
return response; return finishProxyTimedResponse(response, timer, "exception");
} }
} }
+42 -15
View File
@@ -5,27 +5,45 @@ import {
} from "@/lib/backend-auth"; } from "@/lib/backend-auth";
import { buildProxyExceptionResponse } from "@/lib/api-proxy"; import { buildProxyExceptionResponse } from "@/lib/api-proxy";
import { requireOpsProxyAuth } from "@/lib/ops-proxy-auth"; import { requireOpsProxyAuth } from "@/lib/ops-proxy-auth";
import {
createProxyTimer,
finishProxyTimedResponse,
} from "@/lib/proxy-timing";
const API_BASE = process.env.POLYWEATHER_API_BASE_URL; const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
export async function GET(req: NextRequest) { export async function GET(req: NextRequest) {
const timer = createProxyTimer(req, "ops_online_users");
if (!API_BASE) { if (!API_BASE) {
return NextResponse.json( return finishProxyTimedResponse(
{ error: "POLYWEATHER_API_BASE_URL is not configured" }, NextResponse.json(
{ status: 500 }, { error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 },
),
timer,
"missing_api_base",
); );
} }
try { try {
const auth = await buildBackendRequestHeaders(req); const auth = await timer.measure("auth_headers", () =>
const authError = requireOpsProxyAuth(req, auth); buildBackendRequestHeaders(req),
if (authError) return authError; );
const authError = timer.measureSync("ops_auth", () =>
requireOpsProxyAuth(req, auth),
);
if (authError) {
return finishProxyTimedResponse(authError, timer, "ops_auth_error");
}
const res = await fetch(`${API_BASE}/api/ops/online-users`, { const res = await timer.measure("backend_fetch", () =>
headers: auth.headers, fetch(`${API_BASE}/api/ops/online-users`, {
cache: "no-store", headers: auth.headers,
}); cache: "no-store",
const raw = await res.text(); }),
);
const backendServerTiming = res.headers.get("server-timing") || "";
const raw = await timer.measure("backend_read", () => res.text());
const response = new NextResponse(raw, { const response = new NextResponse(raw, {
status: res.status, status: res.status,
headers: { headers: {
@@ -33,10 +51,19 @@ export async function GET(req: NextRequest) {
"Cache-Control": "no-store", "Cache-Control": "no-store",
}, },
}); });
return applyAuthResponseCookies(response, auth.response); return finishProxyTimedResponse(
applyAuthResponseCookies(response, auth.response),
timer,
res.ok ? "ok" : `upstream_${res.status}`,
{ backendServerTiming },
);
} catch (error) { } catch (error) {
return buildProxyExceptionResponse(error, { return finishProxyTimedResponse(
publicMessage: "Failed to fetch online users", buildProxyExceptionResponse(error, {
}); publicMessage: "Failed to fetch online users",
}),
timer,
"exception",
);
} }
} }
+13 -3
View File
@@ -2,6 +2,10 @@ import { NextRequest, NextResponse } from "next/server";
import { proxyBackendJsonGet } from "@/lib/api-proxy"; import { proxyBackendJsonGet } from "@/lib/api-proxy";
import { buildForceRefreshProxyCachePolicy } from "@/lib/proxy-cache-policy"; import { buildForceRefreshProxyCachePolicy } from "@/lib/proxy-cache-policy";
import { DASHBOARD_REFRESH_POLICY_SEC } from "@/lib/refresh-policy"; import { DASHBOARD_REFRESH_POLICY_SEC } from "@/lib/refresh-policy";
import {
createProxyTimer,
finishProxyTimedResponse,
} from "@/lib/proxy-timing";
const API_BASE = process.env.POLYWEATHER_API_BASE_URL; const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
const SCAN_TERMINAL_PROXY_TIMEOUT_MS = Number( const SCAN_TERMINAL_PROXY_TIMEOUT_MS = Number(
@@ -11,10 +15,15 @@ const SCAN_TERMINAL_PROXY_TIMEOUT_MS = Number(
export const maxDuration = 45; export const maxDuration = 45;
export async function GET(req: NextRequest) { export async function GET(req: NextRequest) {
const timer = createProxyTimer(req, "scan_terminal");
if (!API_BASE) { if (!API_BASE) {
return NextResponse.json( return finishProxyTimedResponse(
{ error: "POLYWEATHER_API_BASE_URL is not configured" }, NextResponse.json(
{ status: 500 }, { error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 },
),
timer,
"missing_api_base",
); );
} }
@@ -61,6 +70,7 @@ export async function GET(req: NextRequest) {
revalidateSeconds: cachePolicy.revalidateSeconds, revalidateSeconds: cachePolicy.revalidateSeconds,
signal: controller.signal, signal: controller.signal,
timeoutPublicMessage: "Scan terminal request timed out", timeoutPublicMessage: "Scan terminal request timed out",
timing: timer,
url, url,
}); });
} finally { } finally {
@@ -77,6 +77,8 @@ const TrainingDashboard = dynamic(
}, },
); );
const ONLINE_USERS_REFRESH_MS = 5 * 60_000;
function createEmptyAccess(loading = true): ProAccessState { function createEmptyAccess(loading = true): ProAccessState {
return { return {
loading, loading,
@@ -425,14 +427,22 @@ function PolyWeatherTerminal({
useEffect(() => { useEffect(() => {
const fetchOnline = () => { const fetchOnline = () => {
if (typeof document !== "undefined" && document.visibilityState === "hidden") return;
fetch("/api/ops/online-users", { headers: { Accept: "application/json" } }) fetch("/api/ops/online-users", { headers: { Accept: "application/json" } })
.then((r) => (r.ok ? r.json() : null)) .then((r) => (r.ok ? r.json() : null))
.then((d) => { if (d?.online != null) setOnlineCount(d.online); }) .then((d) => { if (d?.online != null) setOnlineCount(d.online); })
.catch(() => {}); .catch(() => {});
}; };
fetchOnline(); fetchOnline();
const id = setInterval(fetchOnline, 60_000); const handleVisibilityChange = () => {
return () => clearInterval(id); if (document.visibilityState === "visible") fetchOnline();
};
document.addEventListener("visibilitychange", handleVisibilityChange);
const id = setInterval(fetchOnline, ONLINE_USERS_REFRESH_MS);
return () => {
document.removeEventListener("visibilitychange", handleVisibilityChange);
clearInterval(id);
};
}, []); }, []);
const [gridCols, setGridCols] = useState<number>(() => { const [gridCols, setGridCols] = useState<number>(() => {
@@ -101,6 +101,10 @@ function formatCityLocalDate(tzOffsetSeconds: number | null | undefined) {
return `${y}-${m}-${d}`; return `${y}-${m}-${d}`;
} }
function getLiveTempFromHourly(data: HourlyForecast) {
return validNumber(data?.airportCurrent?.temp) ?? validNumber(data?.airportPrimary?.temp) ?? null;
}
function getWundergroundDailyHigh(hourly: HourlyForecast) { function getWundergroundDailyHigh(hourly: HourlyForecast) {
return validNumber(hourly?.wundergroundCurrent?.max_so_far) ?? null; return validNumber(hourly?.wundergroundCurrent?.max_so_far) ?? null;
} }
@@ -365,6 +369,8 @@ export function LiveTemperatureThresholdChart({
.then((data) => { .then((data) => {
if (cancelled || !data) return; if (cancelled || !data) return;
hasLoadedHourlyDetailRef.current = true; hasLoadedHourlyDetailRef.current = true;
const temp = getLiveTempFromHourly(data);
if (temp !== null) setLiveTemp(temp);
setHourly(data); setHourly(data);
}) })
.catch(() => {}) .catch(() => {})
@@ -377,15 +383,6 @@ export function LiveTemperatureThresholdChart({
if (typeof document !== "undefined" && document.visibilityState === "hidden") return; if (typeof document !== "undefined" && document.visibilityState === "hidden") return;
if (Date.now() - lastPatchAtRef.current < 2 * 60_000) return; if (Date.now() - lastPatchAtRef.current < 2 * 60_000) return;
fetch(`/api/city/${encodeURIComponent(city)}/summary`)
.then((res) => (res.ok ? res.json() : null))
.then((payload) => {
if (cancelled || !payload) return;
const temp = validNumber(payload?.current?.temp);
if (temp !== null) setLiveTemp(temp);
})
.catch(() => {});
refreshFullDetail(); refreshFullDetail();
}; };
@@ -403,19 +400,12 @@ export function LiveTemperatureThresholdChart({
const refreshForegroundFullDetail = () => { const refreshForegroundFullDetail = () => {
lastPatchAtRef.current = Date.now(); lastPatchAtRef.current = Date.now();
fetch(`/api/city/${encodeURIComponent(city)}/summary`)
.then((res) => (res.ok ? res.json() : null))
.then((payload) => {
if (cancelled || !payload) return;
const temp = validNumber(payload?.current?.temp);
if (temp !== null) setLiveTemp(temp);
})
.catch(() => {});
fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution }) fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution })
.then((data) => { .then((data) => {
if (cancelled || !data) return; if (cancelled || !data) return;
hasLoadedHourlyDetailRef.current = true; hasLoadedHourlyDetailRef.current = true;
const temp = getLiveTempFromHourly(data);
if (temp !== null) setLiveTemp(temp);
setHourly(data); setHourly(data);
}) })
.catch(() => {}); .catch(() => {});
@@ -12,6 +12,7 @@ import {
import { import {
MAX_HOURLY_DETAIL_CONCURRENT_REQUESTS, MAX_HOURLY_DETAIL_CONCURRENT_REQUESTS,
HOURLY_CACHE_TTL_MS, HOURLY_CACHE_TTL_MS,
__resolveCityDetailFromBatchForTest,
__readHourlyCacheEntryForTest, __readHourlyCacheEntryForTest,
__resetHourlyDetailRequestQueueForTest, __resetHourlyDetailRequestQueueForTest,
__runQueuedHourlyDetailRequestForTest, __runQueuedHourlyDetailRequestForTest,
@@ -49,6 +50,10 @@ export async function runTests() {
path.join(projectRoot, "components", "dashboard", "scan-terminal", "temperature-chart-logic.ts"), path.join(projectRoot, "components", "dashboard", "scan-terminal", "temperature-chart-logic.ts"),
"utf8", "utf8",
); );
const dashboardSource = fs.readFileSync(
path.join(projectRoot, "components", "dashboard", "ScanTerminalDashboard.tsx"),
"utf8",
);
assert( assert(
querySource.includes("useSsePatchVersion") && querySource.includes("useSsePatchVersion") &&
@@ -91,10 +96,16 @@ export async function runTests() {
chartLogicSource.includes("primeCityDetailCache"), chartLogicSource.includes("primeCityDetailCache"),
"visible terminal chart detail fetches should be coalesced into one batch request and prime the shared chart cache", "visible terminal chart detail fetches should be coalesced into one batch request and prime the shared chart cache",
); );
const fetchHourlyBlock = chartLogicSource.match(/async function fetchHourlyForecastForCity[\s\S]*?\r?\n}\r?\n\r?\nfunction fetchCityDetailWithTimeout/)?.[0] || "";
assert( assert(
chartLogicSource.includes("options.ignoreCache\n ? runQueuedHourlyDetailRequest") && fetchHourlyBlock.includes("queueCityDetailBatch(city, resParam)") &&
chartLogicSource.includes(": queueCityDetailBatch(city, resParam)"), !fetchHourlyBlock.includes("runQueuedHourlyDetailRequest"),
"normal first-paint city detail requests should enter the batch queue before single-request concurrency limiting", "first-paint and background city detail refreshes should both enter the batch queue before falling back to single requests",
);
assert(
dashboardSource.includes("ONLINE_USERS_REFRESH_MS = 5 * 60_000") &&
dashboardSource.includes('document.visibilityState === "hidden"'),
"terminal online-user presence should refresh slowly and pause while the browser tab is hidden",
); );
assert( assert(
chartSource.includes("IntersectionObserver") && chartSource.includes("IntersectionObserver") &&
@@ -120,6 +131,19 @@ export async function runTests() {
__shouldFetchCityDetailForChartForTest({ city: "paris", documentHidden: true, isChartVisible: true }) === false, __shouldFetchCityDetailForChartForTest({ city: "paris", documentHidden: true, isChartVisible: true }) === false,
"hidden browser tabs should not prefetch city detail", "hidden browser tabs should not prefetch city detail",
); );
const normalizedBatchDetail = __resolveCityDetailFromBatchForTest(
{
"hong kong": {
city: "hong kong",
timeseries: { hourly: { times: ["00:00"], temps: [32] } },
},
} as any,
"Hong Kong",
) as any;
assert(
normalizedBatchDetail?.city === "hong kong",
"frontend detail batch lookup should accept backend-normalized city keys before falling back to single-city requests",
);
__resetHourlyDetailRequestQueueForTest(); __resetHourlyDetailRequestQueueForTest();
let activeRequests = 0; let activeRequests = 0;
@@ -182,6 +182,10 @@ export function runTests() {
!foregroundRefreshBlock.includes("setIsHourlyLoading(true)"), !foregroundRefreshBlock.includes("setIsHourlyLoading(true)"),
"foreground resume refresh should update full detail immediately in the background without showing the loading overlay", "foreground resume refresh should update full detail immediately in the background without showing the loading overlay",
); );
assert(
!chart.includes("/api/city/${encodeURIComponent(city)}/summary"),
"visible chart fallback and foreground refresh should not issue a separate summary request after requesting full detail",
);
assert(chart.includes("viewMode"), "temperature chart must expose a view mode for DEB-peak auto view versus full-day view"); assert(chart.includes("viewMode"), "temperature chart must expose a view mode for DEB-peak auto view versus full-day view");
assert(chart.includes('useState<"auto" | "full">("full")'), "temperature chart must default every city panel to the all-day view"); assert(chart.includes('useState<"auto" | "full">("full")'), "temperature chart must default every city panel to the all-day view");
assert( assert(
@@ -1073,6 +1073,30 @@ function rejectBatchWaiters(
(waiters || []).forEach((waiter) => waiter.reject(reason)); (waiters || []).forEach((waiter) => waiter.reject(reason));
} }
function resolveCityDetailFromBatch(
details: Record<string, CityDetail | null | undefined> | undefined,
city: string,
) {
if (!details) return undefined;
const trimmed = String(city || "").trim();
const direct =
details[city] ||
details[trimmed] ||
details[trimmed.toLowerCase()] ||
details[normalizeCityKey(trimmed)];
if (direct) return direct;
const requestedKey = normalizeCityKey(trimmed);
if (!requestedKey) return undefined;
for (const [key, detail] of Object.entries(details)) {
if (!detail) continue;
if (normalizeCityKey(key) === requestedKey) return detail;
const detailCity = (detail as any).city || detail.name || detail.display_name;
if (normalizeCityKey(detailCity) === requestedKey) return detail;
}
return undefined;
}
async function flushCityDetailBatch(resolution: string) { async function flushCityDetailBatch(resolution: string) {
const queue = _cityDetailBatchQueues.get(resolution); const queue = _cityDetailBatchQueues.get(resolution);
if (!queue) return; if (!queue) return;
@@ -1091,14 +1115,17 @@ async function flushCityDetailBatch(resolution: string) {
await Promise.all( await Promise.all(
cities.map(async (city) => { cities.map(async (city) => {
const waiters = queue.waiters.get(city); const waiters = queue.waiters.get(city);
const detail = details[city]; const detail = resolveCityDetailFromBatch(details, city);
const data = primeCityDetailCache(city, resolution, detail); const data = primeCityDetailCache(city, resolution, detail);
if (data) { if (data) {
resolveBatchWaiters(waiters, data); resolveBatchWaiters(waiters, data);
return; return;
} }
try { try {
resolveBatchWaiters(waiters, await fetchSingleHourlyForecastForCity(city, resolution)); resolveBatchWaiters(
waiters,
await runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resolution)),
);
} catch (error) { } catch (error) {
rejectBatchWaiters(waiters, error); rejectBatchWaiters(waiters, error);
} }
@@ -1109,7 +1136,10 @@ async function flushCityDetailBatch(resolution: string) {
cities.map(async (city) => { cities.map(async (city) => {
const waiters = queue.waiters.get(city); const waiters = queue.waiters.get(city);
try { try {
resolveBatchWaiters(waiters, await fetchSingleHourlyForecastForCity(city, resolution)); resolveBatchWaiters(
waiters,
await runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resolution)),
);
} catch (fallbackError) { } catch (fallbackError) {
rejectBatchWaiters(waiters, fallbackError || error); rejectBatchWaiters(waiters, fallbackError || error);
} }
@@ -1158,11 +1188,7 @@ async function fetchHourlyForecastForCity(
const pending = _hourlyRequestCache.get(requestKey); const pending = _hourlyRequestCache.get(requestKey);
if (pending) return pending; if (pending) return pending;
const request = ( const request = queueCityDetailBatch(city, resParam)
options.ignoreCache
? runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resParam))
: queueCityDetailBatch(city, resParam)
)
.finally(() => { .finally(() => {
_hourlyRequestCache.delete(requestKey); _hourlyRequestCache.delete(requestKey);
}); });
@@ -2424,6 +2450,7 @@ export {
HOURLY_CACHE_TTL_MS, HOURLY_CACHE_TTL_MS,
_hourlyCache, _hourlyCache,
__readHourlyCacheEntryForTest, __readHourlyCacheEntryForTest,
resolveCityDetailFromBatch as __resolveCityDetailFromBatchForTest,
__resetHourlyDetailRequestQueueForTest, __resetHourlyDetailRequestQueueForTest,
__runQueuedHourlyDetailRequestForTest, __runQueuedHourlyDetailRequestForTest,
buildChartDomain, buildChartDomain,
@@ -0,0 +1,60 @@
import assert from "node:assert/strict";
import fs from "node:fs";
import path from "node:path";
function readFrontend(...parts: string[]) {
return fs.readFileSync(path.join(process.cwd(), ...parts), "utf8");
}
export function runTests() {
const timingSource = readFrontend("lib", "proxy-timing.ts");
assert.match(
timingSource,
/createProxyTimer/,
"shared proxy timing helper should create timers for slow API proxies",
);
assert.match(
timingSource,
/Server-Timing/,
"shared proxy timing helper should write Server-Timing headers for HAR inspection",
);
assert.doesNotMatch(
timingSource,
/authUserId|authEmail|userId|email/,
"proxy timing logs must avoid raw user ids or emails",
);
const apiProxySource = readFrontend("lib", "api-proxy.ts");
assert.match(
apiProxySource,
/timing\?: ProxyTimer/,
"generic backend JSON proxy should accept an optional timer",
);
for (const stage of ["auth_headers", "backend_fetch", "backend_read"]) {
assert.match(
apiProxySource,
new RegExp(stage),
`generic backend JSON proxy should measure ${stage}`,
);
}
const detailBatchProxy = readFrontend("app", "api", "cities", "detail-batch", "route.ts");
assert.match(detailBatchProxy, /createProxyTimer\(req,\s*"city_detail_batch"\)/);
assert.match(detailBatchProxy, /timing:\s*timer/);
const scanTerminalProxy = readFrontend("app", "api", "scan", "terminal", "route.ts");
assert.match(scanTerminalProxy, /createProxyTimer\(req,\s*"scan_terminal"\)/);
assert.match(scanTerminalProxy, /timing:\s*timer/);
const cityDetailProxy = readFrontend("app", "api", "city", "[name]", "detail", "route.ts");
assert.match(cityDetailProxy, /createProxyTimer\(req,\s*"city_detail"\)/);
for (const stage of ["auth_headers", "backend_fetch", "backend_read"]) {
assert.match(cityDetailProxy, new RegExp(stage));
}
const onlineUsersProxy = readFrontend("app", "api", "ops", "online-users", "route.ts");
assert.match(onlineUsersProxy, /createProxyTimer\(req,\s*"ops_online_users"\)/);
for (const stage of ["auth_headers", "ops_auth", "backend_fetch", "backend_read"]) {
assert.match(onlineUsersProxy, new RegExp(stage));
}
}
+55 -15
View File
@@ -5,6 +5,10 @@ import {
buildBackendRequestHeaders, buildBackendRequestHeaders,
} from "@/lib/backend-auth"; } from "@/lib/backend-auth";
import { buildCachedJsonResponse } from "@/lib/http-cache"; import { buildCachedJsonResponse } from "@/lib/http-cache";
import {
finishProxyTimedResponse,
type ProxyTimer,
} from "@/lib/proxy-timing";
const PASSTHROUGH_UPSTREAM_STATUSES = new Set([ const PASSTHROUGH_UPSTREAM_STATUSES = new Set([
400, 400,
@@ -91,31 +95,59 @@ export async function proxyBackendJsonGet(
signal?: AbortSignal; signal?: AbortSignal;
statusOnException?: number; statusOnException?: number;
timeoutPublicMessage?: string; timeoutPublicMessage?: string;
timing?: ProxyTimer;
url: string; url: string;
}, },
) { ) {
let auth: Awaited<ReturnType<typeof buildBackendRequestHeaders>> | null = null; let auth: Awaited<ReturnType<typeof buildBackendRequestHeaders>> | null = null;
const timing = options.timing;
try { try {
auth = await buildBackendRequestHeaders(req, { auth = await (timing
includeSupabaseIdentity: options.includeSupabaseIdentity ?? false, ? timing.measure("auth_headers", () =>
}); buildBackendRequestHeaders(req, {
const res = await fetch(options.url, { includeSupabaseIdentity: options.includeSupabaseIdentity ?? false,
headers: auth.headers, }),
...(options.fetchCache )
? { cache: options.fetchCache } : buildBackendRequestHeaders(req, {
: { next: { revalidate: options.revalidateSeconds ?? 30 } }), includeSupabaseIdentity: options.includeSupabaseIdentity ?? false,
signal: options.signal, }));
}); const res = await (timing
? timing.measure("backend_fetch", () =>
fetch(options.url, {
headers: auth!.headers,
...(options.fetchCache
? { cache: options.fetchCache }
: { next: { revalidate: options.revalidateSeconds ?? 30 } }),
signal: options.signal,
}),
)
: fetch(options.url, {
headers: auth.headers,
...(options.fetchCache
? { cache: options.fetchCache }
: { next: { revalidate: options.revalidateSeconds ?? 30 } }),
signal: options.signal,
}));
const backendServerTiming = res.headers.get("server-timing") || "";
if (!res.ok) { if (!res.ok) {
const raw = await res.text(); const raw = await (timing
? timing.measure("backend_read", () => res.text())
: res.text());
const response = buildUpstreamErrorResponse(res.status, raw, { const response = buildUpstreamErrorResponse(res.status, raw, {
detailLimit: options.detailLimit, detailLimit: options.detailLimit,
error: options.error, error: options.error,
}); });
return applyAuthResponseCookies(response, auth.response); const withCookies = applyAuthResponseCookies(response, auth.response);
return timing
? finishProxyTimedResponse(withCookies, timing, `upstream_${res.status}`, {
backendServerTiming,
})
: withCookies;
} }
const data = await res.json(); const data = await (timing
? timing.measure("backend_read", () => res.json())
: res.json());
const response = const response =
options.cacheControl && options.conditionalResponse !== false options.cacheControl && options.conditionalResponse !== false
? buildCachedJsonResponse(req, data, options.cacheControl) ? buildCachedJsonResponse(req, data, options.cacheControl)
@@ -124,7 +156,12 @@ export async function proxyBackendJsonGet(
? { "Cache-Control": options.cacheControl } ? { "Cache-Control": options.cacheControl }
: undefined, : undefined,
}); });
return applyAuthResponseCookies(response, auth.response); const withCookies = applyAuthResponseCookies(response, auth.response);
return timing
? finishProxyTimedResponse(withCookies, timing, "ok", {
backendServerTiming,
})
: withCookies;
} catch (error) { } catch (error) {
const timedOut = options.signal?.aborted === true; const timedOut = options.signal?.aborted === true;
const response = buildProxyExceptionResponse(error, { const response = buildProxyExceptionResponse(error, {
@@ -134,6 +171,9 @@ export async function proxyBackendJsonGet(
: options.publicMessage, : options.publicMessage,
status: timedOut ? 504 : options.statusOnException, status: timedOut ? 504 : options.statusOnException,
}); });
return auth ? applyAuthResponseCookies(response, auth.response) : response; const withCookies = auth ? applyAuthResponseCookies(response, auth.response) : response;
return timing
? finishProxyTimedResponse(withCookies, timing, timedOut ? "timeout" : "exception")
: withCookies;
} }
} }
+1
View File
@@ -716,6 +716,7 @@ export interface AiAnalysisStructured {
} }
export interface CityDetail { export interface CityDetail {
city?: string;
name: string; name: string;
display_name: string; display_name: string;
detail_depth?: "panel" | "market" | "nearby" | "full"; detail_depth?: "panel" | "market" | "nearby" | "full";
+113
View File
@@ -0,0 +1,113 @@
import type { NextRequest, NextResponse } from "next/server";
export type ProxyTimingStage = {
durationMs: number;
name: string;
};
export type ProxyTimer = {
hasAuthorization: boolean;
hasSupabaseCookie: boolean;
measure<T>(name: string, action: () => Promise<T>): Promise<T>;
measureSync<T>(name: string, action: () => T): T;
route: string;
stages: ProxyTimingStage[];
totalMs(): number;
};
function proxyNowMs() {
return typeof performance !== "undefined" ? performance.now() : Date.now();
}
function hasRequestSupabaseSessionCookie(req: NextRequest) {
return req.cookies.getAll().some((cookie) => {
const name = cookie.name.toLowerCase();
const value = String(cookie.value || "").trim();
return Boolean(
value &&
(name === "supabase-auth-token" ||
(name.startsWith("sb-") && name.includes("-auth-token"))),
);
});
}
export function createProxyTimer(req: NextRequest, route: string): ProxyTimer {
const startedAt = proxyNowMs();
const stages: ProxyTimingStage[] = [];
const recordStage = (name: string, stageStartedAt: number) => {
stages.push({
durationMs: Math.round((proxyNowMs() - stageStartedAt) * 10) / 10,
name,
});
};
return {
hasAuthorization: Boolean(req.headers.get("authorization")),
hasSupabaseCookie: hasRequestSupabaseSessionCookie(req),
async measure<T>(name: string, action: () => Promise<T>) {
const stageStartedAt = proxyNowMs();
try {
return await action();
} finally {
recordStage(name, stageStartedAt);
}
},
measureSync<T>(name: string, action: () => T) {
const stageStartedAt = proxyNowMs();
try {
return action();
} finally {
recordStage(name, stageStartedAt);
}
},
route,
stages,
totalMs() {
return Math.round((proxyNowMs() - startedAt) * 10) / 10;
},
};
}
function formatServerTiming(stages: ProxyTimingStage[]) {
return stages
.map(({ durationMs, name }) => {
const safeName = name.replace(/[^A-Za-z0-9_-]/g, "_");
return `${safeName};dur=${Math.max(0, durationMs).toFixed(1)}`;
})
.join(", ");
}
export function finishProxyTimedResponse(
response: NextResponse,
timer: ProxyTimer,
outcome: string,
extra?: { backendServerTiming?: string },
) {
const total = timer.totalMs();
const ownServerTiming = formatServerTiming(
[...timer.stages, { durationMs: total, name: "total" }].map((stage) => ({
...stage,
name: `${timer.route}_${stage.name}`,
})),
);
const backendServerTiming = String(extra?.backendServerTiming || "").trim();
response.headers.set(
"Server-Timing",
backendServerTiming
? `${ownServerTiming}, ${backendServerTiming}`
: ownServerTiming,
);
console.info(
"[api-proxy-timing]",
JSON.stringify({
hasAuthorization: timer.hasAuthorization,
hasSupabaseCookie: timer.hasSupabaseCookie,
outcome,
route: timer.route,
stages: timer.stages,
status: response.status,
totalMs: total,
}),
);
return response;
}
+3
View File
@@ -946,6 +946,9 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
for key in list(self._wunderground_historical_cache.keys()): for key in list(self._wunderground_historical_cache.keys()):
if key.startswith(prefix): if key.startswith(prefix):
self._wunderground_historical_cache.pop(key, None) self._wunderground_historical_cache.pop(key, None)
for key in list(getattr(self, "_wunderground_negative_cache", {}).keys()):
if key.startswith(prefix):
self._wunderground_negative_cache.pop(key, None)
def _uses_fahrenheit(self, city_lower: str) -> bool: def _uses_fahrenheit(self, city_lower: str) -> bool:
return city_lower in self.US_CITIES return city_lower in self.US_CITIES
@@ -73,6 +73,13 @@ class WundergroundHistoricalMixin:
self._wunderground_historical_cache = {} self._wunderground_historical_cache = {}
if not hasattr(self, "_wunderground_historical_cache_lock"): if not hasattr(self, "_wunderground_historical_cache_lock"):
self._wunderground_historical_cache_lock = threading.Lock() self._wunderground_historical_cache_lock = threading.Lock()
if not hasattr(self, "_wunderground_negative_cache"):
self._wunderground_negative_cache = {}
if not hasattr(self, "wunderground_negative_cache_ttl_sec"):
self.wunderground_negative_cache_ttl_sec = max(
60,
int(os.getenv("WUNDERGROUND_NEGATIVE_CACHE_TTL_SEC", "900")),
)
if not hasattr(self, "wunderground_historical_cache_ttl_sec"): if not hasattr(self, "wunderground_historical_cache_ttl_sec"):
self.wunderground_historical_cache_ttl_sec = max( self.wunderground_historical_cache_ttl_sec = max(
30, 30,
@@ -239,6 +246,15 @@ class WundergroundHistoricalMixin:
cached = self._wunderground_historical_cache.get(cache_key) cached = self._wunderground_historical_cache.get(cache_key)
if cached and now_ts - float(cached.get("t", 0)) < self.wunderground_historical_cache_ttl_sec: if cached and now_ts - float(cached.get("t", 0)) < self.wunderground_historical_cache_ttl_sec:
return dict(cached["d"]) return dict(cached["d"])
negative_cached = self._wunderground_negative_cache.get(cache_key)
if negative_cached and now_ts - float(negative_cached.get("t", 0)) < self.wunderground_negative_cache_ttl_sec:
record_source_call(
"wunderground",
"historical",
"negative_cached",
(time.perf_counter() - started) * 1000.0,
)
return None
current_url = ( current_url = (
"https://api.weather.com/v1/location/" "https://api.weather.com/v1/location/"
@@ -286,6 +302,16 @@ class WundergroundHistoricalMixin:
location_id, location_id,
exc, exc,
) )
if (
isinstance(exc, httpx.HTTPStatusError)
and exc.response is not None
and 400 <= exc.response.status_code < 500
):
with self._wunderground_historical_cache_lock:
self._wunderground_negative_cache[cache_key] = {
"t": time.time(),
"status": exc.response.status_code,
}
record_source_call( record_source_call(
"wunderground", "wunderground",
"historical", "historical",
@@ -396,6 +422,7 @@ class WundergroundHistoricalMixin:
"t": time.time(), "t": time.time(),
"d": payload, "d": payload,
} }
self._wunderground_negative_cache.pop(cache_key, None)
record_source_call( record_source_call(
"wunderground", "wunderground",
"historical", "historical",
+116
View File
@@ -0,0 +1,116 @@
from pathlib import Path
from fastapi.testclient import TestClient
from web.app import app
import web.services.city_api as city_api
import web.services.scan_api as scan_api
ROOT = Path(__file__).resolve().parents[1]
client = TestClient(app)
def test_backend_shared_timing_helper_avoids_sensitive_identity_fields():
source = (ROOT / "web" / "services" / "request_timing.py").read_text(
encoding="utf-8"
)
assert "ServerTimingRecorder" in source
assert "server_timing_value" in source
assert "user_id" not in source
assert "email" not in source
def test_city_detail_batch_response_includes_backend_server_timing(monkeypatch):
class FakeCache:
def get_city_cache(self, kind, city):
assert kind == "full"
return {"payload": {"city": city, "hourly": {"times": [], "temps": []}}}
def build_detail(data, market_slug, target_date, resolution):
return {"city": data["city"], "resolution": resolution}
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
monkeypatch.setattr(
city_api.legacy_routes,
"_normalize_city_or_404",
lambda name: name.strip().lower(),
)
monkeypatch.setattr(city_api.legacy_routes, "_city_cache_is_fresh", lambda entry, ttl: True)
monkeypatch.setattr(
city_api.legacy_routes,
"_overlay_latest_wunderground_current",
lambda city, payload: payload,
)
monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache())
monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
response = client.get("/api/cities/detail-batch?cities=Paris&resolution=10m")
assert response.status_code == 200
server_timing = response.headers["server-timing"]
assert "city_detail_batch_assert_entitlement" in server_timing
assert "city_detail_batch_full_data_paris" in server_timing
assert "city_detail_batch_detail_payload_paris" in server_timing
assert "city_detail_batch_total" in server_timing
def test_city_detail_response_includes_backend_server_timing(monkeypatch):
class FakeCache:
def get_city_cache(self, kind, city):
assert kind == "full"
return {"payload": {"city": city, "hourly": {"times": [], "temps": []}}}
def build_detail(data, market_slug, target_date, resolution):
return {"city": data["city"], "resolution": resolution}
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
monkeypatch.setattr(
city_api.legacy_routes,
"_normalize_city_or_404",
lambda name: name.strip().lower(),
)
monkeypatch.setattr(city_api.legacy_routes, "_city_cache_is_fresh", lambda entry, ttl: True)
monkeypatch.setattr(
city_api.legacy_routes,
"_overlay_latest_wunderground_current",
lambda city, payload: payload,
)
monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache())
monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
response = client.get("/api/city/Paris/detail?resolution=10m")
assert response.status_code == 200
server_timing = response.headers["server-timing"]
assert "city_detail_assert_entitlement" in server_timing
assert "city_detail_full_data" in server_timing
assert "city_detail_detail_payload" in server_timing
assert "city_detail_total" in server_timing
def test_scan_terminal_response_includes_backend_server_timing(monkeypatch):
monkeypatch.setattr(scan_api.legacy_routes, "_assert_entitlement", lambda request: None)
monkeypatch.setattr(
scan_api.legacy_routes,
"build_scan_terminal_payload",
lambda filters, force_refresh=False, timing_recorder=None: {"rows": [], "filters": filters},
)
response = client.get("/api/scan/terminal?limit=1")
assert response.status_code == 200
server_timing = response.headers["server-timing"]
assert "scan_terminal_assert_entitlement" in server_timing
assert "scan_terminal_build_payload" in server_timing
assert "scan_terminal_total" in server_timing
def test_online_users_response_includes_backend_server_timing():
response = client.get("/api/ops/online-users")
assert response.status_code == 200
server_timing = response.headers["server-timing"]
assert "ops_online_users_online_count" in server_timing
assert "ops_online_users_total" in server_timing
+177 -1
View File
@@ -564,7 +564,183 @@ def test_city_detail_batch_endpoint_builds_multiple_cached_details(monkeypatch):
assert sorted(payload["details"]) == ["paris", "shanghai"] assert sorted(payload["details"]) == ["paris", "shanghai"]
assert payload["details"]["shanghai"]["resolution"] == "10m" assert payload["details"]["shanghai"]["resolution"] == "10m"
assert payload["details"]["paris"]["overlay_city"] == "paris" assert payload["details"]["paris"]["overlay_city"] == "paris"
assert calls == [("shanghai", "10m"), ("paris", "10m")] assert sorted(calls) == [("paris", "10m"), ("shanghai", "10m")]
def test_city_detail_batch_endpoint_limits_backend_concurrency(monkeypatch):
import asyncio
active = 0
max_active = 0
async def fake_run_in_threadpool(fn, *args, **kwargs):
nonlocal active, max_active
active += 1
max_active = max(max_active, active)
try:
await asyncio.sleep(0.01)
return fn(*args, **kwargs)
finally:
active -= 1
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "2")
monkeypatch.setattr(city_api, "run_in_threadpool", fake_run_in_threadpool)
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower())
monkeypatch.setattr(city_api.legacy_routes, "_city_cache_is_fresh", lambda entry, ttl: False)
class FakeCache:
def get_city_cache(self, kind, city):
assert kind == "full"
return None
def refresh_full(city, force_refresh):
return {
"city": city,
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
}
def build_detail(data, market_slug, target_date, resolution):
return {
"city": data["city"],
"hourly": data["hourly"],
"resolution": resolution,
}
monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache())
monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_full_cache", refresh_full)
monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
response = client.get("/api/cities/detail-batch?cities=a,b,c,d,e&resolution=10m&limit=5")
assert response.status_code == 200
assert response.json()["cities"] == ["a", "b", "c", "d", "e"]
assert max_active <= 2
def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypatch):
import asyncio
refresh_calls = 0
build_calls = 0
class FakeCache:
payload = None
def get_city_cache(self, kind, city):
assert kind == "full"
if self.payload is None:
return None
return {"payload": self.payload}
fake_cache = FakeCache()
async def fake_run_in_threadpool(fn, *args, **kwargs):
if fn is city_api.legacy_routes._refresh_city_full_cache:
await asyncio.sleep(0.02)
return fn(*args, **kwargs)
def refresh_full(city, force_refresh):
nonlocal refresh_calls
refresh_calls += 1
fake_cache.payload = {
"city": city,
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
}
return fake_cache.payload
def build_detail(data, market_slug, target_date, resolution):
nonlocal build_calls
build_calls += 1
return {
"city": data["city"],
"hourly": data["hourly"],
"resolution": resolution,
}
monkeypatch.setattr(city_api, "run_in_threadpool", fake_run_in_threadpool)
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower())
monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", fake_cache)
monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_full_cache", refresh_full)
monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
async def run_two_requests():
return await asyncio.gather(
city_api.get_city_detail_aggregate_payload(object(), "Paris", resolution="10m"),
city_api.get_city_detail_aggregate_payload(object(), "Paris", resolution="10m"),
)
results = asyncio.run(run_two_requests())
assert [item["city"] for item in results] == ["paris", "paris"]
assert refresh_calls == 1
assert build_calls == 1
def test_force_refresh_invalidates_short_city_detail_payload_cache(monkeypatch):
import asyncio
build_calls = 0
refreshed_payloads = [
{
"city": "paris",
"local_date": "2026-05-30",
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
},
{
"city": "paris",
"local_date": "2026-05-30",
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [21.0]},
},
]
class FakeCache:
def get_city_cache(self, kind, city):
assert kind == "full"
return None
async def fake_run_in_threadpool(fn, *args, **kwargs):
return fn(*args, **kwargs)
def refresh_full(city, force_refresh):
assert city == "paris"
assert refreshed_payloads
return refreshed_payloads.pop(0)
def build_detail(data, market_slug, target_date, resolution):
nonlocal build_calls
build_calls += 1
return {
"city": data["city"],
"live_temp": data["hourly"]["temps"][0],
"resolution": resolution,
}
city_api._CITY_DETAIL_PAYLOAD_CACHE.clear()
city_api._CITY_DETAIL_PAYLOAD_CACHE_TS.clear()
city_api._CITY_DETAIL_PAYLOAD_INFLIGHT.clear()
monkeypatch.setattr(city_api, "run_in_threadpool", fake_run_in_threadpool)
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower())
monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache())
monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_full_cache", refresh_full)
monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
first = asyncio.run(city_api.get_city_detail_aggregate_payload(object(), "Paris", resolution="10m"))
second = asyncio.run(
city_api.get_city_detail_aggregate_payload(
object(),
"Paris",
resolution="10m",
force_refresh=True,
),
)
assert first["live_temp"] == 20.0
assert second["live_temp"] == 21.0
assert build_calls == 2
def test_payment_runtime_endpoint_returns_shape(): def test_payment_runtime_endpoint_returns_shape():
+49
View File
@@ -180,6 +180,55 @@ def test_fetch_wunderground_historical_keeps_fahrenheit_for_us_cities(monkeypatc
assert payload["max_so_far"] == 75 assert payload["max_so_far"] == 75
def test_fetch_wunderground_historical_negative_caches_client_errors(monkeypatch, tmp_path):
collector = _collector(monkeypatch, tmp_path)
monkeypatch.setenv("WUNDERGROUND_NEGATIVE_CACHE_TTL_SEC", "900")
calls = []
def fake_get(url: str, **_kwargs):
calls.append(url)
return httpx.Response(
400,
json={"errors": [{"message": "invalid location"}]},
request=httpx.Request("GET", url),
)
monkeypatch.setattr(collector, "_http_get", fake_get)
first = collector.fetch_wunderground_historical(
"istanbul",
use_fahrenheit=False,
utc_offset=10800,
local_date="2026-05-31",
)
second = collector.fetch_wunderground_historical(
"istanbul",
use_fahrenheit=False,
utc_offset=10800,
local_date="2026-05-31",
)
assert first is None
assert second is None
assert len(calls) == 2
collector._evict_city_caches(
city="istanbul",
lat=None,
lon=None,
use_fahrenheit=False,
)
third = collector.fetch_wunderground_historical(
"istanbul",
use_fahrenheit=False,
utc_offset=10800,
local_date="2026-05-31",
)
assert third is None
assert len(calls) == 4
def test_fetch_all_sources_attaches_wunderground_historical(monkeypatch, tmp_path): def test_fetch_all_sources_attaches_wunderground_historical(monkeypatch, tmp_path):
collector = _collector(monkeypatch, tmp_path) collector = _collector(monkeypatch, tmp_path)
called = {} called = {}
+10 -3
View File
@@ -2,7 +2,7 @@
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
from fastapi import APIRouter, BackgroundTasks, Query, Request from fastapi import APIRouter, BackgroundTasks, Query, Request, Response
from web.services.city_api import ( from web.services.city_api import (
get_city_detail_batch_payload, get_city_detail_batch_payload,
@@ -12,6 +12,7 @@ from web.services.city_api import (
list_cities_payload, list_cities_payload,
) )
from web.services.city_realtime_stream import get_realtime_stream_payload from web.services.city_realtime_stream import get_realtime_stream_payload
from web.services.request_timing import attach_server_timing_header
router = APIRouter(tags=["city"]) router = APIRouter(tags=["city"])
@@ -108,6 +109,7 @@ async def cities_model_range(
@router.get("/api/cities/detail-batch") @router.get("/api/cities/detail-batch")
async def city_detail_batch( async def city_detail_batch(
request: Request, request: Request,
response: Response,
cities: str = "", cities: str = "",
force_refresh: bool = False, force_refresh: bool = False,
market_slug: Optional[str] = None, market_slug: Optional[str] = None,
@@ -115,7 +117,7 @@ async def city_detail_batch(
resolution: Optional[str] = "10m", resolution: Optional[str] = "10m",
limit: int = 12, limit: int = 12,
): ):
return await get_city_detail_batch_payload( payload = await get_city_detail_batch_payload(
request, request,
cities=cities, cities=cities,
force_refresh=force_refresh, force_refresh=force_refresh,
@@ -124,6 +126,8 @@ async def city_detail_batch(
resolution=resolution, resolution=resolution,
limit=limit, limit=limit,
) )
attach_server_timing_header(response, request, "city_detail_batch_server_timing")
return payload
@router.get("/api/city/{name}") @router.get("/api/city/{name}")
@@ -159,13 +163,14 @@ async def city_summary(
@router.get("/api/city/{name}/detail") @router.get("/api/city/{name}/detail")
async def city_detail_aggregate( async def city_detail_aggregate(
request: Request, request: Request,
response: Response,
name: str, name: str,
force_refresh: bool = False, force_refresh: bool = False,
market_slug: Optional[str] = None, market_slug: Optional[str] = None,
target_date: Optional[str] = None, target_date: Optional[str] = None,
resolution: Optional[str] = "10m", resolution: Optional[str] = "10m",
): ):
return await get_city_detail_aggregate_payload( payload = await get_city_detail_aggregate_payload(
request, request,
name, name,
force_refresh=force_refresh, force_refresh=force_refresh,
@@ -173,6 +178,8 @@ async def city_detail_aggregate(
target_date=target_date, target_date=target_date,
resolution=resolution, resolution=resolution,
) )
attach_server_timing_header(response, request, "city_detail_server_timing")
return payload
@router.get("/api/city/{name}/realtime-stream") @router.get("/api/city/{name}/realtime-stream")
+25 -4
View File
@@ -1,6 +1,6 @@
"""Operations/admin API routes.""" """Operations/admin API routes."""
from fastapi import APIRouter, Request from fastapi import APIRouter, Request, Response
from web.core import GrantPointsRequest from web.core import GrantPointsRequest
from web.services.ops_api import ( from web.services.ops_api import (
@@ -30,14 +30,35 @@ from web.services.ops_api import (
get_ops_training_accuracy, get_ops_training_accuracy,
get_ops_telegram_audit, get_ops_telegram_audit,
) )
from web.services.request_timing import ServerTimingRecorder, attach_server_timing_header
router = APIRouter(tags=["ops"]) router = APIRouter(tags=["ops"])
@router.get("/api/ops/online-users") @router.get("/api/ops/online-users")
async def ops_online_users(): async def ops_online_users(request: Request, response: Response):
from src.utils.online_tracker import online_count timer = ServerTimingRecorder(
return {"online": online_count()} request,
log_name="ops_online_users_timing",
prefix="ops_online_users",
state_attr="ops_online_users_server_timing",
)
outcome = "ok"
status_code = 200
try:
from src.utils.online_tracker import online_count
return {"online": timer.measure("online_count", online_count)}
except Exception:
outcome = "exception"
status_code = 500
raise
finally:
timer.finish(outcome=outcome, status_code=status_code)
attach_server_timing_header(
response,
request,
"ops_online_users_server_timing",
)
@router.get("/api/ops/users") @router.get("/api/ops/users")
+4 -1
View File
@@ -9,6 +9,7 @@ from web.services.scan_api import (
get_scan_terminal_overview_payload, get_scan_terminal_overview_payload,
get_scan_terminal_payload, get_scan_terminal_payload,
) )
from web.services.request_timing import attach_server_timing_header
router = APIRouter(tags=["scan"]) router = APIRouter(tags=["scan"])
@@ -45,12 +46,14 @@ async def scan_terminal(
region=region or trading_region or None, region=region or trading_region or None,
timezone_offset_seconds=timezone_offset_seconds, timezone_offset_seconds=timezone_offset_seconds,
) )
return JSONResponse( response = JSONResponse(
content=payload, content=payload,
headers={ headers={
"Cache-Control": "public, s-maxage=30, stale-while-revalidate=120", "Cache-Control": "public, s-maxage=30, stale-while-revalidate=120",
}, },
) )
attach_server_timing_header(response, request, "scan_terminal_server_timing")
return response
@router.post("/api/scan/terminal/overview") @router.post("/api/scan/terminal/overview")
+40 -5
View File
@@ -235,16 +235,41 @@ def build_scan_terminal_payload(
raw_filters: Optional[Dict[str, Any]] = None, raw_filters: Optional[Dict[str, Any]] = None,
*, *,
force_refresh: bool = False, force_refresh: bool = False,
timing_recorder: Any = None,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
filters = _normalize_scan_terminal_filters(raw_filters) filters = (
timing_recorder.measure(
"normalize_filters",
lambda: _normalize_scan_terminal_filters(raw_filters),
)
if timing_recorder is not None
else _normalize_scan_terminal_filters(raw_filters)
)
if not force_refresh: if not force_refresh:
cached = get_cached_scan_terminal_payload( cached = (
filters, ttl_sec=SCAN_TERMINAL_PAYLOAD_TTL_SEC timing_recorder.measure(
"cache_lookup",
lambda: get_cached_scan_terminal_payload(
filters,
ttl_sec=SCAN_TERMINAL_PAYLOAD_TTL_SEC,
),
)
if timing_recorder is not None
else get_cached_scan_terminal_payload(
filters, ttl_sec=SCAN_TERMINAL_PAYLOAD_TTL_SEC
)
) )
if cached is not None: if cached is not None:
return cached return cached
cached_entry = get_scan_terminal_cache_entry(filters) or {} cached_entry = (
timing_recorder.measure(
"stale_cache_lookup",
lambda: get_scan_terminal_cache_entry(filters) or {},
)
if timing_recorder is not None
else get_scan_terminal_cache_entry(filters) or {}
)
success_payload = cached_entry.get("success_payload") success_payload = cached_entry.get("success_payload")
if isinstance(success_payload, dict) and success_payload: if isinstance(success_payload, dict) and success_payload:
started = _start_scan_terminal_background_refresh(filters) started = _start_scan_terminal_background_refresh(filters)
@@ -257,7 +282,17 @@ def build_scan_terminal_payload(
failed_at=cached_entry.get("last_failed_at"), failed_at=cached_entry.get("last_failed_at"),
) )
return _build_scan_terminal_payload_uncached(filters, force_refresh=force_refresh) return (
timing_recorder.measure(
"uncached_build",
lambda: _build_scan_terminal_payload_uncached(
filters,
force_refresh=force_refresh,
),
)
if timing_recorder is not None
else _build_scan_terminal_payload_uncached(filters, force_refresh=force_refresh)
)
_SCAN_PREWARM_STARTED = False _SCAN_PREWARM_STARTED = False
+272 -78
View File
@@ -13,6 +13,7 @@ from fastapi.concurrency import run_in_threadpool
from loguru import logger from loguru import logger
import web.routes as legacy_routes import web.routes as legacy_routes
from web.services.request_timing import ServerTimingRecorder
_RECENT_DEB_CACHE: Optional[Dict[str, Dict[str, object]]] = None _RECENT_DEB_CACHE: Optional[Dict[str, Dict[str, object]]] = None
_RECENT_DEB_CACHE_TS = 0.0 _RECENT_DEB_CACHE_TS = 0.0
@@ -22,6 +23,22 @@ _RECENT_DEB_CACHE_TTL_SEC = max(
60, 60,
int(os.getenv("POLYWEATHER_CITIES_DEB_RECENT_CACHE_TTL_SEC", "300") or "300"), int(os.getenv("POLYWEATHER_CITIES_DEB_RECENT_CACHE_TTL_SEC", "300") or "300"),
) )
_CITY_FULL_REFRESH_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {}
_CITY_FULL_REFRESH_LOCK = asyncio.Lock()
CityDetailPayloadCacheKey = Tuple[str, str, str, str, str, int]
_CITY_DETAIL_PAYLOAD_CACHE: Dict[CityDetailPayloadCacheKey, Dict[str, Any]] = {}
_CITY_DETAIL_PAYLOAD_CACHE_TS: Dict[CityDetailPayloadCacheKey, float] = {}
_CITY_DETAIL_PAYLOAD_INFLIGHT: Dict[CityDetailPayloadCacheKey, "asyncio.Task[Dict[str, Any]]"] = {}
_CITY_DETAIL_PAYLOAD_EPOCH: Dict[str, int] = {}
_CITY_DETAIL_PAYLOAD_LOCK = asyncio.Lock()
def _city_detail_payload_cache_ttl() -> float:
try:
value = float(os.getenv("POLYWEATHER_CITY_DETAIL_PAYLOAD_CACHE_TTL_SEC", "8") or "8")
except ValueError:
value = 8.0
return max(0.0, min(30.0, value))
async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Dict[str, Any]: async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Dict[str, Any]:
@@ -32,6 +49,134 @@ async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Di
) )
async def _refresh_city_full_cache_singleflight(city: str, force_refresh: bool) -> Dict[str, Any]:
key = f"{city}:{bool(force_refresh)}"
async with _CITY_FULL_REFRESH_LOCK:
task = _CITY_FULL_REFRESH_INFLIGHT.get(key)
if task is None:
task = asyncio.create_task(
run_in_threadpool(legacy_routes._refresh_city_full_cache, city, force_refresh),
)
_CITY_FULL_REFRESH_INFLIGHT[key] = task
try:
return await task
finally:
if task.done():
async with _CITY_FULL_REFRESH_LOCK:
if _CITY_FULL_REFRESH_INFLIGHT.get(key) is task:
_CITY_FULL_REFRESH_INFLIGHT.pop(key, None)
async def _invalidate_city_detail_payload_cache(city: str) -> None:
normalized = str(city or "").strip().lower()
if not normalized:
return
async with _CITY_DETAIL_PAYLOAD_LOCK:
_CITY_DETAIL_PAYLOAD_EPOCH[normalized] = _CITY_DETAIL_PAYLOAD_EPOCH.get(normalized, 0) + 1
old_keys = [key for key in _CITY_DETAIL_PAYLOAD_CACHE if key[0] == normalized]
for key in old_keys:
_CITY_DETAIL_PAYLOAD_CACHE.pop(key, None)
_CITY_DETAIL_PAYLOAD_CACHE_TS.pop(key, None)
async def _refresh_city_full_data(city: str, force_refresh: bool) -> Dict[str, Any]:
await _invalidate_city_detail_payload_cache(city)
return await _refresh_city_full_cache_singleflight(city, force_refresh)
async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, Any]:
if force_refresh:
return await _refresh_city_full_data(city, True)
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city)
if cached_entry:
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC):
return await _refresh_city_full_data(city, False)
return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {})
return await _refresh_city_full_data(city, False)
def _city_detail_payload_cache_key(
data: Dict[str, Any],
market_slug: Optional[str],
target_date: Optional[str],
resolution: Optional[str],
) -> CityDetailPayloadCacheKey:
city = str(data.get("city") or data.get("name") or "").strip().lower()
fingerprint = str(
data.get("updated_at_ts")
or data.get("updated_at")
or data.get("local_time")
or data.get("local_date")
or id(data)
)
generation = _CITY_DETAIL_PAYLOAD_EPOCH.get(city, 0)
return (
city,
str(resolution or "10m"),
str(market_slug or ""),
str(target_date or ""),
fingerprint,
generation,
)
async def _build_city_detail_payload_cached(
data: Dict[str, Any],
market_slug: Optional[str],
target_date: Optional[str],
resolution: Optional[str],
) -> Dict[str, Any]:
ttl = _city_detail_payload_cache_ttl()
if ttl <= 0:
return await run_in_threadpool(
legacy_routes._build_city_detail_payload,
data,
market_slug,
target_date,
resolution,
)
key = _city_detail_payload_cache_key(data, market_slug, target_date, resolution)
now_ts = time.time()
async with _CITY_DETAIL_PAYLOAD_LOCK:
cached = _CITY_DETAIL_PAYLOAD_CACHE.get(key)
cached_ts = _CITY_DETAIL_PAYLOAD_CACHE_TS.get(key, 0.0)
if cached is not None and now_ts - cached_ts < ttl:
return cached
task = _CITY_DETAIL_PAYLOAD_INFLIGHT.get(key)
if task is None:
task = asyncio.create_task(
run_in_threadpool(
legacy_routes._build_city_detail_payload,
data,
market_slug,
target_date,
resolution,
),
)
_CITY_DETAIL_PAYLOAD_INFLIGHT[key] = task
try:
payload = await task
finally:
if task.done():
async with _CITY_DETAIL_PAYLOAD_LOCK:
if _CITY_DETAIL_PAYLOAD_INFLIGHT.get(key) is task:
_CITY_DETAIL_PAYLOAD_INFLIGHT.pop(key, None)
async with _CITY_DETAIL_PAYLOAD_LOCK:
_CITY_DETAIL_PAYLOAD_CACHE[key] = payload
_CITY_DETAIL_PAYLOAD_CACHE_TS[key] = time.time()
if len(_CITY_DETAIL_PAYLOAD_CACHE) > 256:
oldest_keys = sorted(
_CITY_DETAIL_PAYLOAD_CACHE_TS,
key=lambda item: _CITY_DETAIL_PAYLOAD_CACHE_TS.get(item, 0.0),
)[:64]
for old_key in oldest_keys:
_CITY_DETAIL_PAYLOAD_CACHE.pop(old_key, None)
_CITY_DETAIL_PAYLOAD_CACHE_TS.pop(old_key, None)
return payload
def _default_deb_recent() -> Dict[str, object]: def _default_deb_recent() -> Dict[str, object]:
return { return {
"tier": "other", "tier": "other",
@@ -167,14 +312,7 @@ async def get_city_detail_payload(
else: else:
detail_mode = "panel" detail_mode = "panel"
if detail_mode == "full": if detail_mode == "full":
if force_refresh: return await _get_city_full_data(city, force_refresh=force_refresh)
return await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, True)
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city)
if cached_entry:
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC):
return await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False)
return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {})
return await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False)
if detail_mode == "panel": if detail_mode == "panel":
if force_refresh: if force_refresh:
return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, True) return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, True)
@@ -231,27 +369,41 @@ async def get_city_detail_aggregate_payload(
target_date: Optional[str] = None, target_date: Optional[str] = None,
resolution: Optional[str] = "10m", resolution: Optional[str] = "10m",
) -> Dict[str, Any]: ) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request) timer = ServerTimingRecorder(
city = legacy_routes._normalize_city_or_404(name) request,
if force_refresh: log_name="city_detail_timing",
data = await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, True) prefix="city_detail",
else: state_attr="city_detail_server_timing",
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city)
if cached_entry:
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC):
data = await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False)
else:
data = await _overlay_cached_wunderground(city, cached_entry.get("payload") or {})
else:
data = await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False)
return await run_in_threadpool(
legacy_routes._build_city_detail_payload,
data,
market_slug,
target_date,
resolution,
) )
outcome = "ok"
status_code = 200
try:
timer.measure("assert_entitlement", lambda: legacy_routes._assert_entitlement(request))
city = timer.measure("normalize_city", lambda: legacy_routes._normalize_city_or_404(name))
data = await timer.measure_async(
"full_data",
lambda: _get_city_full_data(city, force_refresh=force_refresh),
)
return await timer.measure_async(
"detail_payload",
lambda: _build_city_detail_payload_cached(
data,
market_slug,
target_date,
resolution,
),
)
except HTTPException as exc:
outcome = f"http_{exc.status_code}"
status_code = exc.status_code
raise
except Exception:
outcome = "exception"
status_code = 500
raise
finally:
timer.finish(outcome=outcome, status_code=status_code)
def _parse_batch_city_names(raw_cities: str, *, limit: int) -> List[str]: def _parse_batch_city_names(raw_cities: str, *, limit: int) -> List[str]:
@@ -271,38 +423,48 @@ def _parse_batch_city_names(raw_cities: str, *, limit: int) -> List[str]:
return out return out
def _build_city_detail_batch_item( async def _build_city_detail_batch_item_async(
city: str, city: str,
*, *,
force_refresh: bool, force_refresh: bool,
market_slug: Optional[str], market_slug: Optional[str],
target_date: Optional[str], target_date: Optional[str],
resolution: Optional[str], resolution: Optional[str],
timing_recorder: Optional[ServerTimingRecorder] = None,
) -> Tuple[str, Dict[str, Any]]: ) -> Tuple[str, Dict[str, Any]]:
if force_refresh: if timing_recorder is not None:
data = legacy_routes._refresh_city_full_cache(city, True) data = await timing_recorder.measure_async(
f"full_data_{city}",
lambda: _get_city_full_data(city, force_refresh=force_refresh),
)
detail = await timing_recorder.measure_async(
f"detail_payload_{city}",
lambda: _build_city_detail_payload_cached(
data,
market_slug,
target_date,
resolution,
),
)
else: else:
cached_entry = legacy_routes._CACHE_DB.get_city_cache("full", city) data = await _get_city_full_data(city, force_refresh=force_refresh)
if cached_entry and legacy_routes._city_cache_is_fresh( detail = await _build_city_detail_payload_cached(
cached_entry, data,
legacy_routes.CITY_FULL_CACHE_TTL_SEC, market_slug,
): target_date,
data = legacy_routes._overlay_latest_wunderground_current( resolution,
city, )
cached_entry.get("payload") or {},
)
else:
data = legacy_routes._refresh_city_full_cache(city, False)
detail = legacy_routes._build_city_detail_payload(
data,
market_slug,
target_date,
resolution,
)
return city, detail return city, detail
def _city_detail_batch_concurrency() -> int:
try:
value = int(os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "3") or "3")
except ValueError:
value = 3
return max(1, min(6, value))
async def get_city_detail_batch_payload( async def get_city_detail_batch_payload(
request: Request, request: Request,
*, *,
@@ -313,36 +475,68 @@ async def get_city_detail_batch_payload(
resolution: Optional[str] = "10m", resolution: Optional[str] = "10m",
limit: int = 12, limit: int = 12,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request) timer = ServerTimingRecorder(
city_names = _parse_batch_city_names(cities, limit=max(1, min(24, int(limit or 12)))) request,
if not city_names: log_name="city_detail_batch_timing",
return {"cities": [], "details": {}, "errors": {}} prefix="city_detail_batch",
state_attr="city_detail_batch_server_timing",
tasks = [ )
run_in_threadpool( outcome = "ok"
_build_city_detail_batch_item, status_code = 200
city, try:
force_refresh=force_refresh, timer.measure("assert_entitlement", lambda: legacy_routes._assert_entitlement(request))
market_slug=market_slug, city_names = timer.measure(
target_date=target_date, "parse_cities",
resolution=resolution, lambda: _parse_batch_city_names(
cities,
limit=max(1, min(24, int(limit or 12))),
),
) )
for city in city_names if not city_names:
] return {"cities": [], "details": {}, "errors": {}}
results = await asyncio.gather(*tasks, return_exceptions=True)
details: Dict[str, Any] = {}
errors: Dict[str, str] = {}
for city, result in zip(city_names, results):
if isinstance(result, Exception):
errors[city] = str(result)
continue
result_city, payload = result
details[result_city] = payload
return { semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())
"cities": city_names,
"details": details,
"errors": errors,
}
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,
timing_recorder=timer,
)
tasks = [
_build_with_limit(city)
for city in city_names
]
results = await timer.measure_async(
"build_details",
lambda: asyncio.gather(*tasks, return_exceptions=True),
)
details: Dict[str, Any] = {}
errors: Dict[str, str] = {}
for city, result in zip(city_names, results):
if isinstance(result, Exception):
errors[city] = str(result)
continue
result_city, payload = result
details[result_city] = payload
return {
"cities": city_names,
"details": details,
"errors": errors,
}
except HTTPException as exc:
outcome = f"http_{exc.status_code}"
status_code = exc.status_code
raise
except Exception:
outcome = "exception"
status_code = 500
raise
finally:
timer.finish(outcome=outcome, status_code=status_code)
+87
View File
@@ -0,0 +1,87 @@
"""Small helpers for exposing request stage timings via Server-Timing."""
from __future__ import annotations
import re
import threading
import time
from typing import Awaitable, Callable, Dict, Optional, TypeVar
from fastapi import Request, Response
from loguru import logger
T = TypeVar("T")
class ServerTimingRecorder:
def __init__(
self,
request: Optional[Request],
*,
log_name: str,
prefix: str,
state_attr: str,
) -> None:
self.request = request
self.log_name = log_name
self.prefix = prefix
self.state_attr = state_attr
self.started = time.perf_counter()
self.timings_ms: Dict[str, float] = {}
self._lock = threading.Lock()
def _record(self, stage: str, started: float) -> None:
elapsed_ms = round((time.perf_counter() - started) * 1000.0, 1)
with self._lock:
self.timings_ms[stage] = elapsed_ms
def measure(self, stage: str, action: Callable[[], T]) -> T:
started = time.perf_counter()
try:
return action()
finally:
self._record(stage, started)
async def measure_async(self, stage: str, action: Callable[[], Awaitable[T]]) -> T:
started = time.perf_counter()
try:
return await action()
finally:
self._record(stage, started)
def server_timing_value(self) -> str:
with self._lock:
items = list(self.timings_ms.items())
return ", ".join(
f"{self._metric_name(stage)};dur={max(0.0, duration):.1f}"
for stage, duration in items
)
def finish(self, *, outcome: str, status_code: int) -> None:
self._record("total", self.started)
value = self.server_timing_value()
state = getattr(self.request, "state", None)
if state is not None:
setattr(state, self.state_attr, value)
logger.info(
"{} outcome={} status_code={} timings_ms={}",
self.log_name,
outcome,
status_code,
dict(self.timings_ms),
)
def _metric_name(self, stage: str) -> str:
raw = f"{self.prefix}_{stage}"
return re.sub(r"[^A-Za-z0-9_-]", "_", raw)
def attach_server_timing_header(
response: Response,
request: Request,
state_attr: str,
) -> None:
value = str(getattr(request.state, state_attr, "") or "").strip()
if value:
response.headers["Server-Timing"] = value
+56 -21
View File
@@ -2,12 +2,25 @@
from __future__ import annotations from __future__ import annotations
from inspect import Parameter, signature
from typing import Any, Dict from typing import Any, Dict
from fastapi import Request from fastapi import HTTPException, Request
from fastapi.concurrency import run_in_threadpool from fastapi.concurrency import run_in_threadpool
import web.routes as legacy_routes import web.routes as legacy_routes
from web.services.request_timing import ServerTimingRecorder
def _supports_timing_recorder(func: Any) -> bool:
try:
params = signature(func).parameters.values()
except (TypeError, ValueError):
return True
return any(
param.name == "timing_recorder" or param.kind == Parameter.VAR_KEYWORD
for param in params
)
async def get_scan_terminal_payload( async def get_scan_terminal_payload(
@@ -26,27 +39,49 @@ async def get_scan_terminal_payload(
region: str = "", region: str = "",
timezone_offset_seconds: int | None = None, timezone_offset_seconds: int | None = None,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request) timer = ServerTimingRecorder(
filters: Dict[str, Any] = { request,
"scan_mode": scan_mode, log_name="scan_terminal_timing",
"min_price": min_price, prefix="scan_terminal",
"max_price": max_price, state_attr="scan_terminal_server_timing",
"min_edge_pct": min_edge_pct,
"min_liquidity": min_liquidity,
"high_liquidity_only": high_liquidity_only,
"market_type": market_type,
"time_range": time_range,
"limit": limit,
}
if timezone_offset_seconds is not None:
filters["timezone_offset_seconds"] = timezone_offset_seconds
if region:
filters["trading_region"] = region
return await run_in_threadpool(
legacy_routes.build_scan_terminal_payload,
filters,
force_refresh=force_refresh,
) )
outcome = "ok"
status_code = 200
try:
timer.measure("assert_entitlement", lambda: legacy_routes._assert_entitlement(request))
filters: Dict[str, Any] = {
"scan_mode": scan_mode,
"min_price": min_price,
"max_price": max_price,
"min_edge_pct": min_edge_pct,
"min_liquidity": min_liquidity,
"high_liquidity_only": high_liquidity_only,
"market_type": market_type,
"time_range": time_range,
"limit": limit,
}
if timezone_offset_seconds is not None:
filters["timezone_offset_seconds"] = timezone_offset_seconds
if region:
filters["trading_region"] = region
async def build_payload():
builder = legacy_routes.build_scan_terminal_payload
kwargs: Dict[str, Any] = {"force_refresh": force_refresh}
if _supports_timing_recorder(builder):
kwargs["timing_recorder"] = timer
return await run_in_threadpool(builder, filters, **kwargs)
return await timer.measure_async("build_payload", build_payload)
except HTTPException as exc:
outcome = f"http_{exc.status_code}"
status_code = exc.status_code
raise
except Exception:
outcome = "exception"
status_code = 500
raise
finally:
timer.finish(outcome=outcome, status_code=status_code)
async def get_scan_terminal_overview_payload(request: Request) -> Dict[str, Any]: async def get_scan_terminal_overview_payload(request: Request) -> Dict[str, Any]: