mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: stabilize code queue overview proxy

Browse Source
This commit is contained in:
Codex
2026-05-16 15:16:21 +00:00
parent 694f1d39bf
commit 42798b63bb
2 changed files with 78 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/backend-core/src/index.ts
+3
View File
@@ -18,6 +18,7 @@ import type { WsData, ProviderSocket } from "./types";
const runtimeConfig = readConfig();
const { logger: loggerFn, recentLogs } = createLogger("backend-core", runtimeConfig.logFile);
const serverIdleTimeoutSeconds = 120;
ctx.config = runtimeConfig;
ctx.logger = loggerFn;
@@ -119,6 +120,7 @@ async function providerRoute(req: Request, server: Server<WsData>): Promise<Resp
const apiServer = Bun.serve<WsData>({
port: config().port,
idleTimeout: serverIdleTimeoutSeconds,
fetch(req, server) {
return route(req, server);
},
@@ -143,6 +145,7 @@ const apiServer = Bun.serve<WsData>({
const providerServer = Bun.serve<WsData>({
port: config().providerPort,
idleTimeout: serverIdleTimeoutSeconds,
async fetch(req, server) {
const started = performance.now();
const url = new URL(req.url);
src/components/backend-core/src/microservice-proxy.ts
+75 -10
View File
@@ -11,6 +11,7 @@ import { getNodes, getNodeDockerStatuses } from "./db";
const microserviceProxyMaxBodyTextLength = 8 * 1024 * 1024;
const microserviceAvailabilityTtlMs = 30_000;
const codeQueueOverviewPathFallbackStaleMs = 30_000;
const microserviceForwardRequestHeaders = [
"accept",
"content-type",
@@ -416,7 +417,19 @@ function microserviceCacheKey(service: MicroserviceConfig, method: string, targe
return JSON.stringify([service.id, method, targetPath, proxyOptions.query, proxyOptions.jsonArrayLimits]);
}
function responseFromMicroserviceCache(entry: MicroserviceProxyCacheEntry, state: "hit" | "stale"): Response {
function microservicePathFallbackCacheKey(service: MicroserviceConfig, method: string, targetPath: string): string {
return JSON.stringify([service.id, method, targetPath, "__path_fallback__"]);
}
function canUseMicroservicePathFallback(service: MicroserviceConfig, method: string, targetPath: string): boolean {
return service.id === "code-queue" && targetPath === "/api/tasks/overview" && (method === "GET" || method === "HEAD");
}
function cloneMicroserviceCacheEntry(entry: MicroserviceProxyCacheEntry): MicroserviceProxyCacheEntry {
return { ...entry };
}
function responseFromMicroserviceCache(entry: MicroserviceProxyCacheEntry, state: "hit" | "stale" | "path-stale"): Response {
return new Response(entry.bodyText, {
status: entry.status,
headers: {
@@ -448,6 +461,18 @@ function readStaleMicroserviceCache(key: string): Response | null {
return responseFromMicroserviceCache(entry, "stale");
}
function readMicroservicePathFallback(service: MicroserviceConfig, method: string, targetPath: string): Response | null {
if (!canUseMicroservicePathFallback(service, method, targetPath)) return null;
const key = microservicePathFallbackCacheKey(service, method, targetPath);
const entry = ctx.microserviceProxyCache.get(key);
if (entry === undefined) return null;
if (entry.staleExpiresAt <= Date.now()) {
ctx.microserviceProxyCache.delete(key);
return null;
}
return responseFromMicroserviceCache(entry, "path-stale");
}
async function cacheableResponseSnapshot(response: Response): Promise<MicroserviceProxyCacheEntry | null> {
if (response.status < 200 || response.status >= 300) return null;
if (response.headers.get("x-unidesk-response-truncated") === "true") return null;
@@ -464,9 +489,10 @@ async function cacheableResponseSnapshot(response: Response): Promise<Microservi
function rememberMicroserviceCache(key: string, ttlMs: number, entry: MicroserviceProxyCacheEntry | null): void {
if (entry === null || ttlMs <= 0) return;
const keyParts = JSON.parse(key) as string[];
entry.expiresAt = Date.now() + ttlMs;
entry.staleExpiresAt = entry.expiresAt + microserviceCacheStaleMs(String(keyParts[0] ?? ""), String(keyParts[2] ?? ""));
ctx.microserviceProxyCache.set(key, entry);
const stored = cloneMicroserviceCacheEntry(entry);
stored.expiresAt = Date.now() + ttlMs;
stored.staleExpiresAt = stored.expiresAt + microserviceCacheStaleMs(String(keyParts[0] ?? ""), String(keyParts[2] ?? ""));
ctx.microserviceProxyCache.set(key, stored);
if (ctx.microserviceProxyCache.size > 300) {
const now = Date.now();
for (const [cacheKey, cacheEntry] of ctx.microserviceProxyCache) {
@@ -475,6 +501,20 @@ function rememberMicroserviceCache(key: string, ttlMs: number, entry: Microservi
}
}
function rememberMicroservicePathFallback(
service: MicroserviceConfig,
method: string,
targetPath: string,
entry: MicroserviceProxyCacheEntry | null,
): void {
if (entry === null || !canUseMicroservicePathFallback(service, method, targetPath)) return;
const now = Date.now();
const stored = cloneMicroserviceCacheEntry(entry);
stored.expiresAt = now;
stored.staleExpiresAt = now + codeQueueOverviewPathFallbackStaleMs;
ctx.microserviceProxyCache.set(microservicePathFallbackCacheKey(service, method, targetPath), stored);
}
function invalidateMicroserviceCache(serviceId: string): void {
const prefix = `["${serviceId}",`;
for (const key of ctx.microserviceProxyCache.keys()) {
@@ -607,18 +647,23 @@ function refreshMicroserviceCacheInBackground(
cacheKey: string,
ttlMs: number,
fetchResponse: () => Promise<Response>,
fallbackContext?: { service: MicroserviceConfig; method: string; targetPath: string },
refreshKey = cacheKey,
): void {
if (ctx.microserviceProxyRefreshes.has(cacheKey)) return;
if (ctx.microserviceProxyRefreshes.has(refreshKey)) return;
const refresh = fetchResponse()
.then((response) => cacheableResponseSnapshot(response))
.then((entry) => rememberMicroserviceCache(cacheKey, ttlMs, entry))
.then((entry) => {
rememberMicroserviceCache(cacheKey, ttlMs, entry);
if (fallbackContext !== undefined) rememberMicroservicePathFallback(fallbackContext.service, fallbackContext.method, fallbackContext.targetPath, entry);
})
.catch((error) => {
logger("warn", "microservice_cache_revalidate_failed", { cacheKey, error: errorToJson(error) });
})
.finally(() => {
ctx.microserviceProxyRefreshes.delete(cacheKey);
ctx.microserviceProxyRefreshes.delete(refreshKey);
});
ctx.microserviceProxyRefreshes.set(cacheKey, refresh);
ctx.microserviceProxyRefreshes.set(refreshKey, refresh);
}
// ---------------------------------------------------------------------------
@@ -785,11 +830,31 @@ export async function microserviceRoute(req: Request, url: URL): Promise<Respons
if (method === "GET" || method === "HEAD") {
const stale = readStaleMicroserviceCache(cacheKey);
if (stale !== null) {
refreshMicroserviceCacheInBackground(cacheKey, cacheTtlMs, () => fetchMicroserviceUpstreamResponse(service, method, targetPath, proxyOptions, requestHeaders, bodyText));
refreshMicroserviceCacheInBackground(
cacheKey,
cacheTtlMs,
() => fetchMicroserviceUpstreamResponse(service, method, targetPath, proxyOptions, requestHeaders, bodyText),
{ service, method, targetPath },
);
return stale;
}
const fallback = readMicroservicePathFallback(service, method, targetPath);
if (fallback !== null) {
refreshMicroserviceCacheInBackground(
cacheKey,
cacheTtlMs,
() => fetchMicroserviceUpstreamResponse(service, method, targetPath, proxyOptions, requestHeaders, bodyText),
{ service, method, targetPath },
microservicePathFallbackCacheKey(service, method, targetPath),
);
return fallback;
}
}
const response = await fetchMicroserviceUpstreamResponse(service, method, targetPath, proxyOptions, requestHeaders, bodyText, req.signal);
if ((method === "GET" || method === "HEAD") && cacheTtlMs > 0) rememberMicroserviceCache(cacheKey, cacheTtlMs, await cacheableResponseSnapshot(response));
if ((method === "GET" || method === "HEAD") && cacheTtlMs > 0) {
const snapshot = await cacheableResponseSnapshot(response);
rememberMicroserviceCache(cacheKey, cacheTtlMs, snapshot);
rememberMicroservicePathFallback(service, method, targetPath, snapshot);
}
return response;
}