Reduce terminal auth and analytics stalls

This commit is contained in:
2569718930@qq.com
2026-05-31 19:21:53 +08:00
parent 5083b7c433
commit 8d26afdec0
8 changed files with 295 additions and 31 deletions
+80 -24
View File
@@ -7,25 +7,46 @@ import {
buildProxyExceptionResponse, buildProxyExceptionResponse,
buildUpstreamErrorResponse, buildUpstreamErrorResponse,
} from "@/lib/api-proxy"; } from "@/lib/api-proxy";
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 ANALYTICS_ENABLED = const ANALYTICS_ENABLED =
process.env.NEXT_PUBLIC_POLYWEATHER_APP_ANALYTICS !== "false"; process.env.NEXT_PUBLIC_POLYWEATHER_APP_ANALYTICS !== "false";
const ANALYTICS_PROXY_TIMEOUT_MS = Math.max(
250,
Number(process.env.POLYWEATHER_ANALYTICS_PROXY_TIMEOUT_MS || "1500") || 1500,
);
export async function POST(req: NextRequest) { export async function POST(req: NextRequest) {
const timer = createProxyTimer(req, "analytics_events");
if (!ANALYTICS_ENABLED) { if (!ANALYTICS_ENABLED) {
return new NextResponse(null, { status: 204 }); return finishProxyTimedResponse(
} new NextResponse(null, { status: 204 }),
timer,
if (!API_BASE) { "disabled",
return NextResponse.json(
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 },
); );
} }
if (!API_BASE) {
return finishProxyTimedResponse(
NextResponse.json(
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 },
),
timer,
"missing_api_base",
);
}
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), ANALYTICS_PROXY_TIMEOUT_MS);
let auth: Awaited<ReturnType<typeof buildBackendRequestHeaders>> | null = null;
try { try {
const body = await req.json(); const body = await timer.measure("request_read", () => req.json());
const payload = const payload =
body && typeof body.payload === "object" && body.payload != null body && typeof body.payload === "object" && body.payload != null
? body.payload ? body.payload
@@ -42,30 +63,65 @@ export async function POST(req: NextRequest) {
referer_header: req.headers.get("referer") || "", referer_header: req.headers.get("referer") || "",
}, },
}; };
const auth = await buildBackendRequestHeaders(req, { auth = await timer.measure("auth_headers", () =>
includeSupabaseIdentity: false, buildBackendRequestHeaders(req, {
}); includeSupabaseIdentity: false,
}),
);
const headers = new Headers(auth.headers); const headers = new Headers(auth.headers);
headers.set("Content-Type", "application/json"); headers.set("Content-Type", "application/json");
const res = await fetch(`${API_BASE}/api/analytics/events`, { const res = await timer.measure("backend_fetch", () =>
method: "POST", fetch(`${API_BASE}/api/analytics/events`, {
headers, method: "POST",
body: JSON.stringify(enrichedBody), headers,
cache: "no-store", body: JSON.stringify(enrichedBody),
}); cache: "no-store",
signal: controller.signal,
}),
);
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, {
detailLimit: 260, detailLimit: 260,
}); });
return applyAuthResponseCookies(response, auth.response); return finishProxyTimedResponse(
applyAuthResponseCookies(response, auth.response),
timer,
`upstream_${res.status}`,
{ backendServerTiming },
);
} }
const data = await res.json(); const data = await timer.measure("backend_read", () => res.json());
const response = NextResponse.json(data); const response = NextResponse.json(data);
return applyAuthResponseCookies(response, auth.response); return finishProxyTimedResponse(
applyAuthResponseCookies(response, auth.response),
timer,
"ok",
{ backendServerTiming },
);
} catch (error) { } catch (error) {
return buildProxyExceptionResponse(error, { const timedOut = controller.signal.aborted;
publicMessage: "Failed to track analytics event", const response = timedOut
}); ? NextResponse.json(
{
ok: false,
accepted: true,
dropped: true,
reason: "timeout",
},
{ status: 202 },
)
: buildProxyExceptionResponse(error, {
publicMessage: "Failed to track analytics event",
});
const withCookies = auth ? applyAuthResponseCookies(response, auth.response) : response;
return finishProxyTimedResponse(
withCookies,
timer,
timedOut ? "timeout_accepted" : "exception",
);
} finally {
clearTimeout(timeoutId);
} }
} }
@@ -55,7 +55,10 @@ import {
mergeAccessStateWithAuthPayload, mergeAccessStateWithAuthPayload,
type AuthProfilePayload, type AuthProfilePayload,
} from "@/components/dashboard/scan-terminal/terminal-access-state"; } from "@/components/dashboard/scan-terminal/terminal-access-state";
import { loadTerminalAuthProfile } from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap"; import {
createAuthProfileRequestCache,
loadTerminalAuthProfile,
} from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap";
import { import {
cityListItemsToScanRows, cityListItemsToScanRows,
mergeScanRowsWithCityFallbackRows, mergeScanRowsWithCityFallbackRows,
@@ -962,7 +965,7 @@ function ScanTerminalScreen() {
createEmptyAccess(true), createEmptyAccess(true),
); );
const loadAuthProfile = useCallback( const rawLoadAuthProfile = useCallback(
async ( async (
accessToken?: string | null, accessToken?: string | null,
options?: { preferSnapshot?: boolean }, options?: { preferSnapshot?: boolean },
@@ -984,6 +987,28 @@ function ScanTerminalScreen() {
}, },
[], [],
); );
const authProfileRequestCacheRef = useRef<{
load: typeof rawLoadAuthProfile;
cached: ReturnType<typeof createAuthProfileRequestCache>;
} | null>(null);
const loadAuthProfile = useCallback(
(
accessToken?: string | null,
options?: { preferSnapshot?: boolean },
): Promise<AuthProfilePayload> => {
const current = authProfileRequestCacheRef.current;
if (current?.load === rawLoadAuthProfile) {
return current.cached(accessToken, options);
}
const next = {
load: rawLoadAuthProfile,
cached: createAuthProfileRequestCache(rawLoadAuthProfile),
};
authProfileRequestCacheRef.current = next;
return next.cached(accessToken, options);
},
[rawLoadAuthProfile],
);
const refreshLiveAuthProfile = useCallback(async () => { const refreshLiveAuthProfile = useCallback(async () => {
const supabaseEnabled = hasSupabasePublicEnv(); const supabaseEnabled = hasSupabasePublicEnv();
@@ -96,7 +96,7 @@ 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]*?\n}\n\nfunction fetchCityDetailWithTimeout/)?.[0] || ""; const fetchHourlyBlock = chartLogicSource.match(/async function fetchHourlyForecastForCity[\s\S]*?\r?\n}\r?\n\r?\nfunction fetchCityDetailWithTimeout/)?.[0] || "";
assert( assert(
fetchHourlyBlock.includes("queueCityDetailBatch(city, resParam)") && fetchHourlyBlock.includes("queueCityDetailBatch(city, resParam)") &&
!fetchHourlyBlock.includes("runQueuedHourlyDetailRequest"), !fetchHourlyBlock.includes("runQueuedHourlyDetailRequest"),
@@ -1,4 +1,5 @@
import { import {
createAuthProfileRequestCache,
loadTerminalAuthProfile, loadTerminalAuthProfile,
type TerminalAuthProfilePayload, type TerminalAuthProfilePayload,
} from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap"; } from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap";
@@ -27,6 +28,41 @@ async function flushMicrotasks() {
} }
export async function runTests() { export async function runTests() {
const pendingProfile = deferred<TerminalAuthProfilePayload>();
let underlyingProfileLoads = 0;
const cachedLoadAuthProfile = createAuthProfileRequestCache((
accessToken?: string | null,
options?: { preferSnapshot?: boolean },
) => {
underlyingProfileLoads += 1;
assert(
accessToken === "shared-token" && options?.preferSnapshot === true,
"auth profile request cache should pass through the original token and snapshot preference",
);
return pendingProfile.promise;
});
const firstCachedProfile = cachedLoadAuthProfile("shared-token", {
preferSnapshot: true,
});
const secondCachedProfile = cachedLoadAuthProfile("shared-token", {
preferSnapshot: true,
});
assert(
firstCachedProfile === secondCachedProfile && underlyingProfileLoads === 1,
"auth profile request cache should dedupe identical in-flight profile loads",
);
pendingProfile.resolve({
authenticated: true,
user_id: "shared-user",
subscription_active: true,
});
await firstCachedProfile;
await cachedLoadAuthProfile("shared-token", { preferSnapshot: true });
assert(
underlyingProfileLoads === 2,
"auth profile request cache should clear a key after the in-flight load settles",
);
const slowCookieProfile = deferred<TerminalAuthProfilePayload>(); const slowCookieProfile = deferred<TerminalAuthProfilePayload>();
const fastSession = deferred<{ data: { session: { access_token: string } } }>(); const fastSession = deferred<{ data: { session: { access_token: string } } }>();
const calls: string[] = []; const calls: string[] = [];
@@ -57,6 +57,32 @@ function canResolveProfileImmediately(
return payload?.authenticated === true && payload.subscription_active === true; return payload?.authenticated === true && payload.subscription_active === true;
} }
function authProfileRequestCacheKey(
accessToken?: string | null,
options?: { preferSnapshot?: boolean },
) {
const token = String(accessToken || "").trim();
const scope = token ? `bearer:${token}` : "cookie";
const mode = options?.preferSnapshot ? "snapshot" : "live";
return `${mode}:${scope}`;
}
export function createAuthProfileRequestCache(
loadAuthProfile: LoadTerminalAuthProfileOptions["loadAuthProfile"],
): LoadTerminalAuthProfileOptions["loadAuthProfile"] {
const pending = new Map<string, Promise<TerminalAuthProfilePayload>>();
return (accessToken, options) => {
const key = authProfileRequestCacheKey(accessToken, options);
const existing = pending.get(key);
if (existing) return existing;
const request = loadAuthProfile(accessToken, options).finally(() => {
pending.delete(key);
});
pending.set(key, request);
return request;
};
}
export async function loadTerminalAuthProfile({ export async function loadTerminalAuthProfile({
getSession, getSession,
hasSupabasePublicEnv, hasSupabasePublicEnv,
@@ -57,4 +57,26 @@ export function runTests() {
for (const stage of ["auth_headers", "ops_auth", "backend_fetch", "backend_read"]) { for (const stage of ["auth_headers", "ops_auth", "backend_fetch", "backend_read"]) {
assert.match(onlineUsersProxy, new RegExp(stage)); assert.match(onlineUsersProxy, new RegExp(stage));
} }
const analyticsProxy = readFrontend("app", "api", "analytics", "events", "route.ts");
assert.match(
analyticsProxy,
/ANALYTICS_PROXY_TIMEOUT_MS/,
"analytics event proxy should use a short dedicated timeout instead of waiting for long backend stalls",
);
assert.match(
analyticsProxy,
/createProxyTimer\(req,\s*"analytics_events"\)/,
"analytics event proxy should expose Server-Timing for HAR inspection",
);
assert.match(
analyticsProxy,
/AbortController/,
"analytics event proxy should abort slow upstream tracking requests",
);
assert.match(
analyticsProxy,
/status:\s*202/,
"analytics event proxy timeout should stay non-blocking for the fire-and-forget client event",
);
} }
+64
View File
@@ -678,6 +678,70 @@ def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypat
assert build_calls == 1 assert build_calls == 1
def test_stale_city_detail_uses_cached_full_payload_while_refreshing(monkeypatch):
import asyncio
refresh_calls = 0
build_inputs = []
class FakeCache:
def get_city_cache(self, kind, city):
assert kind == "full"
assert city == "paris"
return {
"payload": {
"city": "paris",
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
},
}
async def fake_run_in_threadpool(fn, *args, **kwargs):
if fn is city_api.legacy_routes._refresh_city_full_cache:
await asyncio.sleep(0.01)
return fn(*args, **kwargs)
def refresh_full(city, force_refresh):
nonlocal refresh_calls
refresh_calls += 1
return {
"city": city,
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [21.0]},
}
def build_detail(data, market_slug, target_date, resolution):
build_inputs.append(data["hourly"]["temps"][0])
return {
"city": data["city"],
"live_temp": data["hourly"]["temps"][0],
"resolution": resolution,
}
city_api._CITY_FULL_REFRESH_INFLIGHT.clear()
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, "_city_cache_is_fresh", lambda entry, ttl: False)
monkeypatch.setattr(city_api.legacy_routes, "_overlay_latest_wunderground_current", lambda city, payload: payload)
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_request():
payload = await city_api.get_city_detail_aggregate_payload(object(), "Paris", resolution="10m")
await asyncio.sleep(0.03)
return payload
result = asyncio.run(run_request())
assert result["live_temp"] == 20.0
assert build_inputs == [20.0]
assert refresh_calls == 1
def test_force_refresh_invalidates_short_city_detail_payload_cache(monkeypatch): def test_force_refresh_invalidates_short_city_detail_payload_cache(monkeypatch):
import asyncio import asyncio
+39 -4
View File
@@ -24,6 +24,7 @@ _RECENT_DEB_CACHE_TTL_SEC = max(
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_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {}
_CITY_FULL_STALE_REFRESH_TASKS: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {}
_CITY_FULL_REFRESH_LOCK = asyncio.Lock() _CITY_FULL_REFRESH_LOCK = asyncio.Lock()
CityDetailPayloadCacheKey = Tuple[str, str, str, str, str, int] CityDetailPayloadCacheKey = Tuple[str, str, str, str, str, int]
_CITY_DETAIL_PAYLOAD_CACHE: Dict[CityDetailPayloadCacheKey, Dict[str, Any]] = {} _CITY_DETAIL_PAYLOAD_CACHE: Dict[CityDetailPayloadCacheKey, Dict[str, Any]] = {}
@@ -54,9 +55,17 @@ async def _refresh_city_full_cache_singleflight(city: str, force_refresh: bool)
async with _CITY_FULL_REFRESH_LOCK: async with _CITY_FULL_REFRESH_LOCK:
task = _CITY_FULL_REFRESH_INFLIGHT.get(key) task = _CITY_FULL_REFRESH_INFLIGHT.get(key)
if task is None: if task is None:
task = asyncio.create_task( async def _run_refresh() -> Dict[str, Any]:
run_in_threadpool(legacy_routes._refresh_city_full_cache, city, force_refresh), try:
) return await run_in_threadpool(
legacy_routes._refresh_city_full_cache,
city,
force_refresh,
)
finally:
await _invalidate_city_detail_payload_cache(city)
task = asyncio.create_task(_run_refresh())
_CITY_FULL_REFRESH_INFLIGHT[key] = task _CITY_FULL_REFRESH_INFLIGHT[key] = task
try: try:
return await task return await task
@@ -84,14 +93,40 @@ async def _refresh_city_full_data(city: str, force_refresh: bool) -> Dict[str, A
return await _refresh_city_full_cache_singleflight(city, force_refresh) return await _refresh_city_full_cache_singleflight(city, force_refresh)
def _start_city_full_stale_refresh(city: str) -> None:
normalized = str(city or "").strip().lower()
if not normalized:
return
existing = _CITY_FULL_STALE_REFRESH_TASKS.get(normalized)
if existing is not None and not existing.done():
return
task = asyncio.create_task(_refresh_city_full_data(city, False))
_CITY_FULL_STALE_REFRESH_TASKS[normalized] = task
def _cleanup(done: "asyncio.Task[Dict[str, Any]]") -> None:
if _CITY_FULL_STALE_REFRESH_TASKS.get(normalized) is done:
_CITY_FULL_STALE_REFRESH_TASKS.pop(normalized, None)
try:
done.result()
except Exception as exc: # pragma: no cover - defensive background guard
logger.warning("city full stale refresh failed city={}: {}", city, exc)
task.add_done_callback(_cleanup)
async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, Any]: async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, Any]:
if force_refresh: if force_refresh:
return await _refresh_city_full_data(city, True) return await _refresh_city_full_data(city, True)
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city) cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city)
if cached_entry: if cached_entry:
payload = cached_entry.get("payload") or {}
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC): if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC):
if payload:
_start_city_full_stale_refresh(city)
return await _overlay_cached_wunderground(city, payload)
return await _refresh_city_full_data(city, False) return await _refresh_city_full_data(city, False)
return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) return await _overlay_cached_wunderground(city, payload)
return await _refresh_city_full_data(city, False) return await _refresh_city_full_data(city, False)