Compare commits

...

2 Commits

Author SHA1 Message Date
2569718930@qq.com 78ea0326a5 Return partial city detail batches 2026-05-31 19:51:28 +08:00
2569718930@qq.com d3f444dbf6 Bind PolyWeather services to loopback 2026-05-31 19:46:00 +08:00
9 changed files with 150 additions and 18 deletions
+2 -2
View File
@@ -79,7 +79,7 @@ services:
timeout: 5s timeout: 5s
image: ghcr.io/yangyuan-zhen/polyweather-frontend:${IMAGE_TAG:-latest} image: ghcr.io/yangyuan-zhen/polyweather-frontend:${IMAGE_TAG:-latest}
ports: ports:
- 3001:3000 - "127.0.0.1:3001:3000"
restart: unless-stopped restart: unless-stopped
polyweather_web: polyweather_web:
command: python web/app.py command: python web/app.py
@@ -104,7 +104,7 @@ services:
timeout: 5s timeout: 5s
image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest} image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest}
ports: ports:
- 8000:8000 - "127.0.0.1:8000:8000"
restart: unless-stopped restart: unless-stopped
user: ${UID:-1000}:${GID:-1000} user: ${UID:-1000}:${GID:-1000}
volumes: volumes:
@@ -42,8 +42,13 @@ export async function GET(req: NextRequest) {
try { try {
return await proxyBackendJsonGet(req, { return await proxyBackendJsonGet(req, {
cacheControl: cachePolicy.responseCacheControl, cacheControl: cachePolicy.responseCacheControl,
fetchCache: cacheControlForData: (data) =>
cachePolicy.fetchMode === "no-store" ? "no-store" : undefined, data &&
typeof data === "object" &&
(data as { partial?: unknown }).partial === true
? "no-store, max-age=0"
: cachePolicy.responseCacheControl,
fetchCache: "no-store",
publicMessage: "Failed to fetch city detail batch", publicMessage: "Failed to fetch city detail batch",
revalidateSeconds: cachePolicy.revalidateSeconds, revalidateSeconds: cachePolicy.revalidateSeconds,
signal: controller.signal, signal: controller.signal,
@@ -96,6 +96,18 @@ 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",
); );
assert(
chartLogicSource.includes("partial?: boolean") &&
chartLogicSource.includes("missing?: string[]"),
"frontend city detail batch payload should understand partial responses and missing city markers",
);
const flushCityDetailBatchBlock = chartLogicSource.match(/async function flushCityDetailBatch[\s\S]*?\r?\n}\r?\n\r?\nfunction fetchCityDetailBatchWithTimeout/)?.[0] || "";
assert(
flushCityDetailBatchBlock.includes("partialMissingCities") &&
flushCityDetailBatchBlock.includes("resolveBatchWaiters(waiters, null)") &&
flushCityDetailBatchBlock.includes("payload?.partial === true"),
"partial detail-batch misses should resolve without immediately issuing single-city fallback requests",
);
const fetchHourlyBlock = chartLogicSource.match(/async function fetchHourlyForecastForCity[\s\S]*?\r?\n}\r?\n\r?\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)") &&
@@ -966,8 +966,11 @@ type HourlyForecastFetchOptions = {
}; };
type CityDetailBatchPayload = { type CityDetailBatchPayload = {
cities?: string[];
details?: Record<string, CityDetail | null | undefined>; details?: Record<string, CityDetail | null | undefined>;
errors?: Record<string, string>; errors?: Record<string, string>;
missing?: string[];
partial?: boolean;
}; };
type CityDetailBatchWaiter = { type CityDetailBatchWaiter = {
@@ -1112,6 +1115,10 @@ async function flushCityDetailBatch(resolution: string) {
try { try {
const payload = await fetchCityDetailBatchWithTimeout(cities, resolution); const payload = await fetchCityDetailBatchWithTimeout(cities, resolution);
const details = payload?.details || {}; const details = payload?.details || {};
const partialMissingCities =
payload?.partial === true
? new Set((payload.missing || []).map((city) => normalizeCityKey(city)))
: new Set<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);
@@ -1121,6 +1128,10 @@ async function flushCityDetailBatch(resolution: string) {
resolveBatchWaiters(waiters, data); resolveBatchWaiters(waiters, data);
return; return;
} }
if (partialMissingCities.has(normalizeCityKey(city))) {
resolveBatchWaiters(waiters, null);
return;
}
try { try {
resolveBatchWaiters( resolveBatchWaiters(
waiters, waiters,
@@ -41,6 +41,21 @@ export function runTests() {
const detailBatchProxy = readFrontend("app", "api", "cities", "detail-batch", "route.ts"); const detailBatchProxy = readFrontend("app", "api", "cities", "detail-batch", "route.ts");
assert.match(detailBatchProxy, /createProxyTimer\(req,\s*"city_detail_batch"\)/); assert.match(detailBatchProxy, /createProxyTimer\(req,\s*"city_detail_batch"\)/);
assert.match(detailBatchProxy, /timing:\s*timer/); assert.match(detailBatchProxy, /timing:\s*timer/);
assert.match(
detailBatchProxy,
/fetchCache:\s*"no-store"/,
"city detail batch proxy should avoid caching partial backend fetches in the Next data cache",
);
assert.match(
detailBatchProxy,
/cacheControlForData/,
"city detail batch proxy should be able to suppress response caching for partial payloads",
);
assert.match(
apiProxySource,
/cacheControlForData\?:/,
"generic backend JSON proxy should allow response cache policy to depend on parsed data",
);
const scanTerminalProxy = readFrontend("app", "api", "scan", "terminal", "route.ts"); const scanTerminalProxy = readFrontend("app", "api", "scan", "terminal", "route.ts");
assert.match(scanTerminalProxy, /createProxyTimer\(req,\s*"scan_terminal"\)/); assert.match(scanTerminalProxy, /createProxyTimer\(req,\s*"scan_terminal"\)/);
+7 -4
View File
@@ -85,6 +85,7 @@ export async function proxyBackendJsonGet(
req: NextRequest, req: NextRequest,
options: { options: {
cacheControl?: string; cacheControl?: string;
cacheControlForData?: (data: unknown) => string | undefined;
conditionalResponse?: boolean; conditionalResponse?: boolean;
detailLimit?: number; detailLimit?: number;
error?: string; error?: string;
@@ -148,12 +149,14 @@ export async function proxyBackendJsonGet(
const data = await (timing const data = await (timing
? timing.measure("backend_read", () => res.json()) ? timing.measure("backend_read", () => res.json())
: res.json()); : res.json());
const responseCacheControl =
options.cacheControlForData?.(data) ?? options.cacheControl;
const response = const response =
options.cacheControl && options.conditionalResponse !== false responseCacheControl && options.conditionalResponse !== false
? buildCachedJsonResponse(req, data, options.cacheControl) ? buildCachedJsonResponse(req, data, responseCacheControl)
: NextResponse.json(data, { : NextResponse.json(data, {
headers: options.cacheControl headers: responseCacheControl
? { "Cache-Control": options.cacheControl } ? { "Cache-Control": responseCacheControl }
: undefined, : undefined,
}); });
const withCookies = applyAuthResponseCookies(response, auth.response); const withCookies = applyAuthResponseCookies(response, auth.response);
+9
View File
@@ -80,6 +80,15 @@ def test_deploy_script_retries_startup_smoke_checks():
assert 'smoke_check "frontend" "https://www.polyweather.top/" 15 3 5' in script assert 'smoke_check "frontend" "https://www.polyweather.top/" 15 3 5' in script
def test_docker_compose_keeps_polyweather_ports_on_loopback():
compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8")
assert "127.0.0.1:3001:3000" in compose
assert "127.0.0.1:8000:8000" in compose
assert "\n - 3001:3000" not in compose
assert "\n - 8000:8000" not in compose
def test_city_detail_builds_deb_hourly_consensus_before_peak_window(): def test_city_detail_builds_deb_hourly_consensus_before_peak_window():
source = (ROOT / "web" / "analysis_service.py").read_text(encoding="utf-8") source = (ROOT / "web" / "analysis_service.py").read_text(encoding="utf-8")
+38
View File
@@ -618,6 +618,44 @@ def test_city_detail_batch_endpoint_limits_backend_concurrency(monkeypatch):
assert max_active <= 2 assert max_active <= 2
def test_city_detail_batch_returns_completed_details_when_one_city_is_slow(monkeypatch):
import asyncio
completed = []
async def build_batch_item(city, **kwargs):
if city == "slow":
await asyncio.sleep(0.08)
completed.append(city)
return city, {
"city": city,
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
"resolution": kwargs.get("resolution"),
}
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", "20")
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, "_build_city_detail_batch_item_async", build_batch_item)
payload = asyncio.run(
city_api.get_city_detail_batch_payload(
object(),
cities="fast,slow,other",
resolution="10m",
limit=3,
)
)
assert payload["cities"] == ["fast", "slow", "other"]
assert sorted(payload["details"]) == ["fast", "other"]
assert payload["details"]["fast"]["resolution"] == "10m"
assert payload["partial"] is True
assert payload["missing"] == ["slow"]
assert payload["errors"] == {}
assert "slow" not in completed
def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypatch): def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypatch):
import asyncio import asyncio
+49 -10
View File
@@ -500,6 +500,19 @@ def _city_detail_batch_concurrency() -> int:
return max(1, min(6, value)) return max(1, min(6, value))
def _city_detail_batch_partial_timeout_seconds() -> Optional[float]:
try:
timeout_ms = int(
os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", "8500")
or "8500"
)
except ValueError:
timeout_ms = 8500
if timeout_ms <= 0:
return None
return max(0.001, min(60.0, timeout_ms / 1000.0))
async def get_city_detail_batch_payload( async def get_city_detail_batch_payload(
request: Request, request: Request,
*, *,
@@ -528,7 +541,13 @@ async def get_city_detail_batch_payload(
), ),
) )
if not city_names: if not city_names:
return {"cities": [], "details": {}, "errors": {}} return {
"cities": [],
"details": {},
"errors": {},
"missing": [],
"partial": False,
}
semaphore = asyncio.Semaphore(_city_detail_batch_concurrency()) semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())
@@ -543,27 +562,47 @@ async def get_city_detail_batch_payload(
timing_recorder=timer, timing_recorder=timer,
) )
tasks = [ task_by_city = {
_build_with_limit(city) city: asyncio.create_task(_build_with_limit(city))
for city in city_names for city in city_names
] }
results = await timer.measure_async( task_city_lookup = {task: city for city, task in task_by_city.items()}
done, pending = await timer.measure_async(
"build_details", "build_details",
lambda: asyncio.gather(*tasks, return_exceptions=True), lambda: asyncio.wait(
task_by_city.values(),
timeout=_city_detail_batch_partial_timeout_seconds(),
),
) )
details: Dict[str, Any] = {} details: Dict[str, Any] = {}
errors: Dict[str, str] = {} errors: Dict[str, str] = {}
for city, result in zip(city_names, results): missing: List[str] = []
if isinstance(result, Exception): for task in done:
errors[city] = str(result) city = task_city_lookup[task]
try:
result_city, payload = task.result()
except Exception as exc:
errors[city] = str(exc)
continue continue
result_city, payload = result
details[result_city] = payload details[result_city] = payload
for task in pending:
city = task_city_lookup[task]
missing.append(city)
task.cancel()
missing_set = set(missing)
missing = [city for city in city_names if city in missing_set]
partial = bool(missing or errors)
if partial:
outcome = "partial"
return { return {
"cities": city_names, "cities": city_names,
"details": details, "details": details,
"errors": errors, "errors": errors,
"missing": missing,
"partial": partial,
} }
except HTTPException as exc: except HTTPException as exc:
outcome = f"http_{exc.status_code}" outcome = f"http_{exc.status_code}"