revert: keep only minimal review fixes
This commit is contained in:
1 parent
caf1f6621a
commit
396b78aa3e
39 files changed
+421
-958
No files matched your search
@@ -5,4 +5,3 @@ export {
|
||||
serializeCustomWidgetDefinition,
|
||||
} from "./router/custom-widget/stored-definition";
|
||||
export { configureCustomWidgetSourceFromRequest } from "./router/custom-widget/secret-persistence";
|
||||
export { invalidateCustomWidgetResponseCache } from "./router/custom-widget/request-executor";
|
||||
@@ -64,7 +64,7 @@ export const previewActionProcedures = {
|
||||
kind: "action",
|
||||
});
|
||||
if (response.ok && request.invalidates?.length) {
|
||||
await invalidateCustomWidgetResponseCache(
|
||||
invalidateCustomWidgetResponseCache(
|
||||
request.invalidates.map((requestId) => `custom-jsx:preview:${session.id}:${requestId}:`),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -19,7 +19,7 @@ export const previewQueryProcedures = {
|
||||
.input(z.object({ sessionId: z.string().min(1) }))
|
||||
.mutation(async ({ ctx, input }) => {
|
||||
const session = await getPreviewSession(input.sessionId, ctx.session.user.id);
|
||||
await invalidateCustomWidgetResponseCache([`custom-jsx:preview:${session.id}:`]);
|
||||
invalidateCustomWidgetResponseCache([`custom-jsx:preview:${session.id}:`]);
|
||||
}),
|
||||
|
||||
previewQuery: permissionRequiredProcedure
|
||||
|
||||
@@ -16,7 +16,6 @@ import {
|
||||
import type { CustomWidgetHttpRequest, CustomWidgetHttpResponse } from "@homarr/custom-widgets/server";
|
||||
|
||||
import { toTrpcError } from "./domain-error";
|
||||
import { executeWithCustomWidgetResponseCache, invalidateSharedCustomWidgetResponseCache } from "./response-cache";
|
||||
|
||||
export type {
|
||||
CustomWidgetAuthConfig,
|
||||
@@ -75,18 +74,14 @@ export async function resolveAndValidateHost(hostname: string, scope: CustomWidg
|
||||
|
||||
export async function executeCustomWidgetRequest(input: CustomWidgetHttpRequest): Promise<CustomWidgetHttpResponse> {
|
||||
try {
|
||||
return await executeWithCustomWidgetResponseCache(input, (request) =>
|
||||
executeDomainRequest({
|
||||
...request,
|
||||
logError: (event) => logger.error("Custom widget request failed", event),
|
||||
}),
|
||||
);
|
||||
return await executeDomainRequest({
|
||||
...input,
|
||||
logError: (event) => logger.error("Custom widget request failed", event),
|
||||
});
|
||||
} catch (error) {
|
||||
toTrpcError(error);
|
||||
}
|
||||
}
|
||||
|
||||
export async function invalidateCustomWidgetResponseCache(prefixes: readonly string[]) {
|
||||
export const invalidateCustomWidgetResponseCache = (prefixes: readonly string[]) =>
|
||||
invalidateDomainResponseCache(prefixes);
|
||||
await invalidateSharedCustomWidgetResponseCache(prefixes);
|
||||
}
|
||||
@@ -1,214 +0,0 @@
|
||||
import { Buffer } from "node:buffer";
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
import { createLogger } from "@homarr/core/infrastructure/logs";
|
||||
import { createRedisClient } from "@homarr/core/infrastructure/redis";
|
||||
import type { RedisClient } from "@homarr/core/infrastructure/redis";
|
||||
import type { CustomWidgetHttpRequest, CustomWidgetHttpResponse } from "@homarr/custom-widgets/server";
|
||||
|
||||
const ENTRY_PREFIX = "custom-widget:response-cache:v2:entry:";
|
||||
const GENERATION_PREFIX = "custom-widget:response-cache:v2:generation:";
|
||||
const LRU_KEY = "custom-widget:response-cache:v2:lru";
|
||||
const SIZE_KEY = "custom-widget:response-cache:v2:sizes";
|
||||
const TOTAL_BYTES_KEY = "custom-widget:response-cache:v2:bytes";
|
||||
// Outlive the one-hour response TTL and 45-second request deadline so a
|
||||
// generation cannot reset while an entry from the previous one is reachable.
|
||||
const GENERATION_RETENTION_SECONDS = 2 * 60 * 60;
|
||||
const MAX_ENTRIES = 1_000;
|
||||
const MAX_BYTES = 64 * 1024 * 1024;
|
||||
|
||||
const INVALIDATE_GENERATIONS_SCRIPT = `
|
||||
for _, key in ipairs(KEYS) do
|
||||
redis.call('INCR', key)
|
||||
redis.call('EXPIRE', key, ARGV[1])
|
||||
end
|
||||
return #KEYS
|
||||
`;
|
||||
|
||||
// Reading has to touch the LRU on a hit and reclaim the byte/LRU/size
|
||||
// bookkeeping on a miss: entry keys carry their own PX expiry, so a TTL-expired
|
||||
// response would otherwise keep occupying a slot and its bytes forever and
|
||||
// shrink the usable cache to nothing over a long uptime.
|
||||
const READ_SCRIPT = `
|
||||
local value = redis.call('GET', KEYS[1])
|
||||
if value then
|
||||
redis.call('ZADD', KEYS[2], ARGV[1], KEYS[1])
|
||||
return value
|
||||
end
|
||||
local size = tonumber(redis.call('HGET', KEYS[3], KEYS[1]) or '0')
|
||||
if size > 0 then redis.call('DECRBY', KEYS[4], size) end
|
||||
redis.call('ZREM', KEYS[2], KEYS[1])
|
||||
redis.call('HDEL', KEYS[3], KEYS[1])
|
||||
return false
|
||||
`;
|
||||
|
||||
const FORGET_SCRIPT = `
|
||||
local size = tonumber(redis.call('HGET', KEYS[3], KEYS[1]) or '0')
|
||||
if size > 0 then redis.call('DECRBY', KEYS[4], size) end
|
||||
redis.call('DEL', KEYS[1])
|
||||
redis.call('ZREM', KEYS[2], KEYS[1])
|
||||
redis.call('HDEL', KEYS[3], KEYS[1])
|
||||
return 1
|
||||
`;
|
||||
|
||||
const STORE_SCRIPT = `
|
||||
local previousSize = tonumber(redis.call('HGET', KEYS[2], ARGV[1]) or '0')
|
||||
if previousSize > 0 then redis.call('DECRBY', KEYS[3], previousSize) end
|
||||
redis.call('SET', ARGV[1], ARGV[2], 'PX', ARGV[3])
|
||||
redis.call('ZADD', KEYS[1], ARGV[4], ARGV[1])
|
||||
redis.call('HSET', KEYS[2], ARGV[1], ARGV[5])
|
||||
local totalBytes = redis.call('INCRBY', KEYS[3], ARGV[5])
|
||||
while redis.call('ZCARD', KEYS[1]) > tonumber(ARGV[6]) or totalBytes > tonumber(ARGV[7]) do
|
||||
local evicted = redis.call('ZPOPMIN', KEYS[1], 1)
|
||||
if #evicted == 0 then break end
|
||||
local evictedKey = evicted[1]
|
||||
local evictedSize = tonumber(redis.call('HGET', KEYS[2], evictedKey) or '0')
|
||||
redis.call('DEL', evictedKey)
|
||||
redis.call('HDEL', KEYS[2], evictedKey)
|
||||
if evictedSize > 0 then totalBytes = redis.call('DECRBY', KEYS[3], evictedSize) end
|
||||
end
|
||||
return totalBytes
|
||||
`;
|
||||
|
||||
const logger = createLogger({ module: "custom-widget:response-cache" });
|
||||
const inFlight = new Map<string, { cacheKey: string; request: Promise<CustomWidgetHttpResponse> }>();
|
||||
let redis: RedisClient | undefined;
|
||||
|
||||
const useSharedCache = () => process.env.NODE_ENV !== "test";
|
||||
|
||||
const getRedis = () => {
|
||||
redis ??= createRedisClient();
|
||||
return redis;
|
||||
};
|
||||
|
||||
const getGenerationKey = (prefix: string) => `${GENERATION_PREFIX}${createHash("sha256").update(prefix).digest("hex")}`;
|
||||
|
||||
const getCacheKeyPrefixes = (cacheKey: string) => {
|
||||
const prefixes: string[] = [];
|
||||
let separatorIndex = cacheKey.indexOf(":");
|
||||
while (separatorIndex >= 0) {
|
||||
prefixes.push(cacheKey.slice(0, separatorIndex + 1));
|
||||
separatorIndex = cacheKey.indexOf(":", separatorIndex + 1);
|
||||
}
|
||||
prefixes.push(cacheKey);
|
||||
return prefixes;
|
||||
};
|
||||
|
||||
const getGeneration = async (client: RedisClient, cacheKey: string) => {
|
||||
const keys = getCacheKeyPrefixes(cacheKey).map(getGenerationKey);
|
||||
const generations = await client.mget(...keys);
|
||||
return generations.map((generation) => generation ?? "0").join(":");
|
||||
};
|
||||
|
||||
const getEntryKey = (generation: string, cacheKey: string) => {
|
||||
const digest = createHash("sha256").update(cacheKey).digest("hex");
|
||||
return `${ENTRY_PREFIX}${generation}:${digest}`;
|
||||
};
|
||||
|
||||
const readCachedResponse = async (client: RedisClient, entryKey: string) => {
|
||||
const serialized = (await client.eval(READ_SCRIPT, 4, entryKey, LRU_KEY, SIZE_KEY, TOTAL_BYTES_KEY, Date.now())) as
|
||||
| string
|
||||
| null;
|
||||
if (serialized === null) return undefined;
|
||||
try {
|
||||
return JSON.parse(serialized) as CustomWidgetHttpResponse;
|
||||
} catch {
|
||||
await client.eval(FORGET_SCRIPT, 4, entryKey, LRU_KEY, SIZE_KEY, TOTAL_BYTES_KEY);
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
|
||||
const storeCachedResponse = async (
|
||||
client: RedisClient,
|
||||
entryKey: string,
|
||||
response: CustomWidgetHttpResponse,
|
||||
ttlSeconds: number,
|
||||
) => {
|
||||
const serialized = JSON.stringify(response);
|
||||
const size = Buffer.byteLength(serialized, "utf8");
|
||||
await client.eval(
|
||||
STORE_SCRIPT,
|
||||
3,
|
||||
LRU_KEY,
|
||||
SIZE_KEY,
|
||||
TOTAL_BYTES_KEY,
|
||||
entryKey,
|
||||
serialized,
|
||||
ttlSeconds * 1_000,
|
||||
Date.now(),
|
||||
size,
|
||||
MAX_ENTRIES,
|
||||
MAX_BYTES,
|
||||
);
|
||||
};
|
||||
|
||||
export async function executeWithCustomWidgetResponseCache(
|
||||
input: CustomWidgetHttpRequest,
|
||||
execute: (input: CustomWidgetHttpRequest) => Promise<CustomWidgetHttpResponse>,
|
||||
): Promise<CustomWidgetHttpResponse> {
|
||||
const cacheKey = input.kind === "query" ? input.cacheKey : undefined;
|
||||
if (!cacheKey) return execute(input);
|
||||
if (!useSharedCache()) return execute(input);
|
||||
|
||||
let client: RedisClient;
|
||||
let generation: string;
|
||||
try {
|
||||
client = getRedis();
|
||||
generation = await getGeneration(client, cacheKey);
|
||||
if ((input.cacheTtlSeconds ?? 0) > 0) {
|
||||
const cached = await readCachedResponse(client, getEntryKey(generation, cacheKey));
|
||||
if (cached) return cached;
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error("Shared custom widget response cache is unavailable", {
|
||||
errorName: error instanceof Error ? error.name : "UnknownError",
|
||||
});
|
||||
// Keep the cache key so the caller's process-local cache still absorbs
|
||||
// polling while Redis is down. Dropping it would send every refresh of
|
||||
// every placed widget straight to the upstream API.
|
||||
return execute(input);
|
||||
}
|
||||
|
||||
const pendingKey = `${generation}:${cacheKey}`;
|
||||
const pending = inFlight.get(pendingKey);
|
||||
if (pending) return pending.request;
|
||||
|
||||
let request: Promise<CustomWidgetHttpResponse>;
|
||||
request = execute({ ...input, cacheKey: undefined })
|
||||
.then(async (response) => {
|
||||
const ttlSeconds = input.cacheTtlSeconds ?? 0;
|
||||
if (!response.ok || ttlSeconds <= 0) return response;
|
||||
try {
|
||||
const currentGeneration = await getGeneration(client, cacheKey);
|
||||
if (currentGeneration === generation) {
|
||||
await storeCachedResponse(client, getEntryKey(generation, cacheKey), response, ttlSeconds);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error("Failed to store a custom widget response", {
|
||||
errorName: error instanceof Error ? error.name : "UnknownError",
|
||||
});
|
||||
}
|
||||
return response;
|
||||
})
|
||||
.finally(() => {
|
||||
if (inFlight.get(pendingKey)?.request === request) inFlight.delete(pendingKey);
|
||||
});
|
||||
inFlight.set(pendingKey, { cacheKey, request });
|
||||
return request;
|
||||
}
|
||||
|
||||
export async function invalidateSharedCustomWidgetResponseCache(prefixes: readonly string[]) {
|
||||
if (prefixes.length === 0 || !useSharedCache()) return;
|
||||
const uniquePrefixes = [...new Set(prefixes)];
|
||||
for (const [key, pending] of inFlight) {
|
||||
if (uniquePrefixes.some((prefix) => pending.cacheKey.startsWith(prefix))) inFlight.delete(key);
|
||||
}
|
||||
try {
|
||||
const keys = uniquePrefixes.map(getGenerationKey);
|
||||
await getRedis().eval(INVALIDATE_GENERATIONS_SCRIPT, keys.length, ...keys, GENERATION_RETENTION_SECONDS);
|
||||
} catch (error) {
|
||||
logger.error("Failed to invalidate the shared custom widget response cache", {
|
||||
errorName: error instanceof Error ? error.name : "UnknownError",
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -118,8 +118,8 @@ describe("Custom JSX v2 board router", () => {
|
||||
});
|
||||
|
||||
test("refreshes the server response cache for a placed widget", async () => {
|
||||
const { db, itemId, ownerId } = await setup();
|
||||
const caller = customApiRouter.createCaller({ db, deviceType: undefined, session: createSession(ownerId) });
|
||||
const { db, itemId } = await setup();
|
||||
const caller = customApiRouter.createCaller({ db, deviceType: undefined, session: null });
|
||||
|
||||
await caller.refresh({ itemId });
|
||||
|
||||
|
||||
@@ -2,7 +2,6 @@ import { TRPCError } from "@trpc/server";
|
||||
import { parse as parseSuperJson } from "superjson";
|
||||
import { z } from "zod/v4";
|
||||
|
||||
import { mapWithConcurrency } from "@homarr/common";
|
||||
import { decryptSecret } from "@homarr/common/server";
|
||||
import { eq } from "@homarr/db";
|
||||
import { boards, customWidgetDefinitions, items, legacyCustomWidgetDefinitions } from "@homarr/db/schema";
|
||||
@@ -14,7 +13,6 @@ import {
|
||||
validateCustomWidgetOptions,
|
||||
} from "@homarr/custom-widgets/core";
|
||||
import type { CustomJsxRequest, CustomWidgetSource } from "@homarr/custom-widgets/core";
|
||||
import { CUSTOM_WIDGET_USER_ITEM_CONCURRENCY_LIMIT } from "@homarr/custom-widgets/core";
|
||||
|
||||
import { createTRPCRouter, protectedProcedure, publicProcedure } from "../../trpc";
|
||||
import { throwIfActionForbiddenAsync } from "../board/board-access";
|
||||
@@ -42,7 +40,7 @@ interface CustomWidgetItemOptions {
|
||||
refreshInterval?: number;
|
||||
}
|
||||
|
||||
type RouterContext = Parameters<typeof throwIfActionForbiddenAsync>[0] & { ipAddress?: string };
|
||||
type RouterContext = Parameters<typeof throwIfActionForbiddenAsync>[0];
|
||||
type ResolvedDefinition = Awaited<ReturnType<typeof resolvePlacedDefinitionAsync>>;
|
||||
|
||||
const parseItemOptions = (raw: string): CustomWidgetItemOptions => {
|
||||
@@ -156,7 +154,6 @@ const withRequestLimit = async <T>(
|
||||
const release = await acquireCustomWidgetRequestLimit({
|
||||
category: request.kind === "query" ? "query" : request.method === "DELETE" ? "delete" : "action",
|
||||
userId: ctx.session?.user.id,
|
||||
anonymousId: ctx.ipAddress,
|
||||
itemId: resolved.item.id,
|
||||
definitionId: resolved.stored.id,
|
||||
});
|
||||
@@ -197,9 +194,9 @@ const executeRequest = async (
|
||||
};
|
||||
|
||||
export const customApiRouter = createTRPCRouter({
|
||||
refresh: protectedProcedure.input(itemInputSchema).mutation(async ({ ctx, input }) => {
|
||||
refresh: publicProcedure.input(itemInputSchema).mutation(async ({ ctx, input }) => {
|
||||
const resolved = await resolvePlacedDefinitionAsync(ctx, input.itemId);
|
||||
await invalidateCustomWidgetResponseCache([
|
||||
invalidateCustomWidgetResponseCache([
|
||||
`custom-jsx:${resolved.item.id}:${getCustomWidgetCacheVersion(resolved.stored)}:`,
|
||||
]);
|
||||
}),
|
||||
@@ -209,10 +206,8 @@ export const customApiRouter = createTRPCRouter({
|
||||
const loadRequests = Object.entries(resolved.definition.requests).filter(
|
||||
([, request]) => request.kind === "query" && request.trigger === "load",
|
||||
);
|
||||
const entries = await mapWithConcurrency(
|
||||
loadRequests,
|
||||
CUSTOM_WIDGET_USER_ITEM_CONCURRENCY_LIMIT,
|
||||
async ([requestId, request]) => {
|
||||
const entries = await Promise.all(
|
||||
loadRequests.map(async ([requestId, request]) => {
|
||||
try {
|
||||
const response = await executeRequest(ctx, resolved, { id: requestId, ...request }, {});
|
||||
return [
|
||||
@@ -242,13 +237,12 @@ export const customApiRouter = createTRPCRouter({
|
||||
},
|
||||
] as const;
|
||||
}
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
return {
|
||||
type: "customJsx" as const,
|
||||
template: resolved.definition.template,
|
||||
canRefresh: ctx.session !== null,
|
||||
data: Object.fromEntries(entries.map(([id, result]) => [id, result.data])),
|
||||
status: Object.fromEntries(entries.map(([id, result]) => [id, result.status])),
|
||||
options: resolved.configuration,
|
||||
@@ -291,7 +285,7 @@ export const customApiRouter = createTRPCRouter({
|
||||
}
|
||||
const response = await executeRequest(ctx, resolved, request, input.params);
|
||||
if (response.ok && request.invalidates?.length) {
|
||||
await invalidateCustomWidgetResponseCache(
|
||||
invalidateCustomWidgetResponseCache(
|
||||
request.invalidates.flatMap((requestId) => [
|
||||
`custom-jsx:${resolved.item.id}:${getCustomWidgetCacheVersion(resolved.stored)}:${requestId}:`,
|
||||
`custom-widget:options:${resolved.stored.id}:${getCustomWidgetCacheVersion(resolved.stored)}:${requestId}:`,
|
||||
|
||||
@@ -13,7 +13,7 @@ import { ZodError } from "zod/v4";
|
||||
|
||||
import type { Session } from "@homarr/auth";
|
||||
import { extractBaseUrlFromHeaders, FlattenError } from "@homarr/common";
|
||||
import { trustedIpAddressFromHeaders, userAgent } from "@homarr/common/server";
|
||||
import { userAgent } from "@homarr/common/server";
|
||||
import type { DeviceType } from "@homarr/common/server";
|
||||
import { createLogger } from "@homarr/core/infrastructure/logs";
|
||||
import { db } from "@homarr/db";
|
||||
@@ -43,8 +43,6 @@ interface ApiContext {
|
||||
deviceType: DeviceType;
|
||||
baseUrl?: `${string}://${string}`;
|
||||
onboardingClaimToken?: string;
|
||||
/** Trusted transport peer, safe to key rate limits and quotas on. */
|
||||
ipAddress?: string;
|
||||
db: typeof db;
|
||||
}
|
||||
|
||||
@@ -62,7 +60,6 @@ export const createTRPCContext = (opts: { headers: Headers; session: Session | n
|
||||
session,
|
||||
deviceType: userAgent(opts.headers).device.type,
|
||||
baseUrl: extractBaseUrlFromHeaders(opts.headers),
|
||||
ipAddress: trustedIpAddressFromHeaders(opts.headers) ?? undefined,
|
||||
onboardingClaimToken: getOnboardingClaimTokenFromCookieHeader(opts.headers.get("cookie")),
|
||||
db,
|
||||
};
|
||||
|
||||
Reference in new issue
Block a user