服务器长跑不再堆积取消监听,数据库少走几趟。

报告工人的等待到点就摘掉取消监听。本地数据库缓存表结构,并把设置身份合成一条语句,对照通过后默认打开,写成 off 仍走原来的语句。交付、重算和限流的内存表会过期。登录读缓存默认关闭,要开再设 IDENTITY_SESSION_COOKIE_CACHE=on。

自测:tsc 通过,lint 0 error,点名测试 134 通过,test:db 开关关和开都只有已知的 13 条失败,往返 49 降到 32。

Co-Authored-By: Grok (agent loop)
This commit is contained in:
jesse-ux
2026-10-14 03:32:39 +08:00
co-authored by Grok (agent loop)
parent 66cd60081f
commit 4c3485bc23
28 changed files with 1614 additions and 34 deletions
@@ -21,7 +21,7 @@ export const dynamic = "force-dynamic";
*/
async function currentUserId(): Promise<string | null> {
const supabase = await createServerSupabaseClient();
const supabase = await createServerSupabaseClient({ disableCookieCache: true });
const { data: { user }, error } = await supabase.auth.getUser();
return error || !user ? null : user.id;
}
@@ -14,7 +14,10 @@ async function userSession(request: Request) {
return null;
}
const services = getIdentityAuthServices();
const session = await services.user.api.getSession({ headers: request.headers });
const session = await services.user.api.getSession({
headers: request.headers,
query: { disableCookieCache: true },
});
return session ? { services, session } : null;
}
+1
View File
@@ -60,6 +60,7 @@ export async function requirePermission(
requireIdentityServerSession(
getIdentityAuthServices().user.api,
adminHeaders,
{ disableCookieCache: true },
),
loadAdminSession,
});
@@ -34,10 +34,14 @@ export async function runBirthTimeScoringPoll(input: PollInput): Promise<PollRes
export function scoringPollDelay(attempt: number, signal: AbortSignal): Promise<void> {
const duration = Math.min(400 * (2 ** attempt), 3_200);
return new Promise((resolve) => {
const timer = window.setTimeout(resolve, duration);
signal.addEventListener("abort", () => {
const onAbort = () => {
window.clearTimeout(timer);
resolve();
}, { once: true });
};
const timer = window.setTimeout(() => {
signal.removeEventListener("abort", onAbort);
resolve();
}, duration);
signal.addEventListener("abort", onAbort, { once: true });
});
}
+201 -11
View File
@@ -169,7 +169,18 @@ export function upsertConflictColumns(options?: { onConflict?: string }): string
.filter(Boolean);
}
async function primaryKeyColumns(client: PoolClient, table: string): Promise<string[]> {
async function primaryKeyColumns(
client: PoolClient,
table: string,
scope: MetadataScope,
): Promise<string[]> {
const cacheKey = metadataCacheEnabled()
? `${scope.connectionString}|${table}`
: null;
if (cacheKey) {
const cached = readTimed(metadataCache().primaryKeys, cacheKey);
if (cached) return [...cached];
}
identifier(table);
const result = await client.query<{ column_name: string }>(
`
@@ -182,7 +193,9 @@ async function primaryKeyColumns(client: PoolClient, table: string): Promise<str
`,
[table],
);
return result.rows.map((row) => row.column_name);
const columns = result.rows.map((row) => row.column_name);
if (cacheKey && columns.length > 0) writeTimed(metadataCache().primaryKeys, cacheKey, [...columns]);
return columns;
}
function queryError(error: unknown): QueryError {
@@ -220,7 +233,115 @@ function localDataPool(connectionString: string): Pool {
return pool;
}
const METADATA_TTL_MS = 300_000;
const METADATA_INVALIDATING_CODES = new Set(["42703", "42P01", "42883", "42804"]);
type MetadataScope = Readonly<{
connectionString: string;
role: LocalDatabaseRole | null;
}>;
type TimedCacheEntry<T> = Readonly<{ value: T; storedAt: number }>;
type LocalPostgresMetadataCache = {
columns: Map<string, TimedCacheEntry<Map<string, string>>>;
primaryKeys: Map<string, TimedCacheEntry<string[]>>;
functions: Map<string, TimedCacheEntry<FunctionMetadata>>;
};
const metadataGlobal = globalThis as typeof globalThis & {
jyotishaLocalPostgresMetadata?: LocalPostgresMetadataCache;
jyotishaLocalPostgresMetadataNow?: () => number;
};
function metadataCache(): LocalPostgresMetadataCache {
metadataGlobal.jyotishaLocalPostgresMetadata ??= {
columns: new Map(),
primaryKeys: new Map(),
functions: new Map(),
};
return metadataGlobal.jyotishaLocalPostgresMetadata;
}
function metadataNow(): number {
return metadataGlobal.jyotishaLocalPostgresMetadataNow?.() ?? Date.now();
}
function metadataCacheEnabled(): boolean {
// 2026-10-14 A/B passed. Unset means on. Exact "off" keeps the old query sequence.
return process.env.JYOTISHA_PG_METADATA_CACHE !== "off";
}
function mergedSessionSetupEnabled(): boolean {
return process.env.JYOTISHA_PG_MERGED_SESSION_SETUP !== "off";
}
function roleCacheKey(role: LocalDatabaseRole | null): string {
return role ?? "owner";
}
function readTimed<T>(
store: Map<string, TimedCacheEntry<T>>,
key: string,
): T | undefined {
const entry = store.get(key);
if (!entry) return undefined;
if (metadataNow() - entry.storedAt >= METADATA_TTL_MS) {
store.delete(key);
return undefined;
}
return entry.value;
}
function writeTimed<T>(
store: Map<string, TimedCacheEntry<T>>,
key: string,
value: T,
): void {
store.set(key, { value, storedAt: metadataNow() });
}
function dropMetadataForConnection(connectionString: string): void {
const cache = metadataGlobal.jyotishaLocalPostgresMetadata;
if (!cache) return;
const prefix = `${connectionString}|`;
for (const store of [cache.columns, cache.primaryKeys, cache.functions]) {
for (const key of store.keys()) {
if (key.startsWith(prefix)) store.delete(key);
}
}
}
function postgresErrorCode(error: unknown): string | undefined {
if (!error || typeof error !== "object") return undefined;
const code = (error as { code?: unknown }).code;
return typeof code === "string" ? code : undefined;
}
export function clearLocalPostgresMetadataCacheForTests(): void {
metadataGlobal.jyotishaLocalPostgresMetadata = {
columns: new Map(),
primaryKeys: new Map(),
functions: new Map(),
};
}
export function setLocalPostgresMetadataClockForTests(now: (() => number) | null): void {
if (now) metadataGlobal.jyotishaLocalPostgresMetadataNow = now;
else delete metadataGlobal.jyotishaLocalPostgresMetadataNow;
}
export function installLocalPostgresPoolForTests(connectionString: string, pool: Pool): void {
poolGlobal.jyotishaLocalDataPools ??= new Map();
poolGlobal.jyotishaLocalDataPools.set(connectionString, pool);
}
export function localPostgresPoolForTests(connectionString: string): Pool {
return localDataPool(connectionString);
}
export async function closeLocalPostgresDataPool(connectionString: string): Promise<void> {
dropMetadataForConnection(connectionString);
const pools = poolGlobal.jyotishaLocalDataPools;
const pool = pools?.get(connectionString);
if (!pool) return;
@@ -229,6 +350,11 @@ export async function closeLocalPostgresDataPool(connectionString: string): Prom
}
export async function closeLocalPostgresDataPools(): Promise<void> {
metadataGlobal.jyotishaLocalPostgresMetadata = {
columns: new Map(),
primaryKeys: new Map(),
functions: new Map(),
};
const pools = poolGlobal.jyotishaLocalDataPools;
poolGlobal.jyotishaLocalDataPools = new Map();
if (!pools) return;
@@ -237,6 +363,7 @@ export async function closeLocalPostgresDataPools(): Promise<void> {
async function inBusinessTransaction<T>(
pool: Pool,
connectionString: string,
identity: LocalIdentity,
role: LocalDatabaseRole | null,
run: (client: PoolClient) => Promise<T>,
@@ -244,15 +371,32 @@ async function inBusinessTransaction<T>(
const client = await pool.connect();
try {
await client.query("begin");
if (role !== null) await client.query(`set local role ${role}`);
await client.query(
"select set_config('request.jwt.claim.sub', $1, true), set_config('request.jwt.claim.email', $2, true)",
[identity?.id ?? "", identity?.email ?? ""],
);
if (mergedSessionSetupEnabled()) {
if (role !== null) {
await client.query(
"select set_config('role', $1, true), set_config('request.jwt.claim.sub', $2, true), set_config('request.jwt.claim.email', $3, true)",
[role, identity?.id ?? "", identity?.email ?? ""],
);
} else {
await client.query(
"select set_config('request.jwt.claim.sub', $1, true), set_config('request.jwt.claim.email', $2, true)",
[identity?.id ?? "", identity?.email ?? ""],
);
}
} else {
if (role !== null) await client.query(`set local role ${role}`);
await client.query(
"select set_config('request.jwt.claim.sub', $1, true), set_config('request.jwt.claim.email', $2, true)",
[identity?.id ?? "", identity?.email ?? ""],
);
}
const result = await run(client);
await client.query("commit");
return result;
} catch (error) {
if (METADATA_INVALIDATING_CODES.has(postgresErrorCode(error) ?? "")) {
dropMetadataForConnection(connectionString);
}
await client.query("rollback").catch(() => undefined);
throw error;
} finally {
@@ -263,7 +407,15 @@ async function inBusinessTransaction<T>(
async function columnTypes(
client: PoolClient,
table: string,
scope: MetadataScope,
): Promise<Map<string, string>> {
const cacheKey = metadataCacheEnabled()
? `${scope.connectionString}|${roleCacheKey(scope.role)}|${table}`
: null;
if (cacheKey) {
const cached = readTimed(metadataCache().columns, cacheKey);
if (cached) return new Map(cached);
}
const result = await client.query<{ column_name: string; udt_name: string }>(
`
select column_name, udt_name
@@ -272,7 +424,9 @@ async function columnTypes(
`,
[table],
);
return new Map(result.rows.map((row) => [row.column_name, row.udt_name]));
const types = new Map(result.rows.map((row) => [row.column_name, row.udt_name]));
if (cacheKey && types.size > 0) writeTimed(metadataCache().columns, cacheKey, new Map(types));
return types;
}
function databaseValue(type: string | undefined, value: unknown): unknown {
@@ -314,6 +468,7 @@ class LocalPostgresQueryBuilder implements PromiseLike<QueryResult> {
constructor(
private readonly pool: Pool,
private readonly connectionString: string,
private readonly identity: LocalIdentity,
private readonly role: LocalDatabaseRole | null,
private readonly table: string,
@@ -516,10 +671,14 @@ class LocalPostgresQueryBuilder implements PromiseLike<QueryResult> {
try {
return await inBusinessTransaction(
this.pool,
this.connectionString,
this.identity,
this.role,
async (client) => {
const types = await columnTypes(client, this.table);
const types = await columnTypes(client, this.table, {
connectionString: this.connectionString,
role: this.role,
});
const parameters: unknown[] = [];
let sql: string;
@@ -563,7 +722,10 @@ class LocalPostgresQueryBuilder implements PromiseLike<QueryResult> {
if (this.mutation.kind === "upsert") {
let conflict = this.mutation.conflict;
if (conflict.length === 0) {
conflict = await primaryKeyColumns(client, this.table);
conflict = await primaryKeyColumns(client, this.table, {
connectionString: this.connectionString,
role: this.role,
});
}
if (conflict.length === 0) {
throw new Error("upsert requires a conflict target");
@@ -648,7 +810,21 @@ async function functionMetadata(
client: PoolClient,
functionName: string,
argumentNames: readonly string[],
scope: MetadataScope,
): Promise<FunctionMetadata> {
const cacheKey = metadataCacheEnabled()
? `${scope.connectionString}|${roleCacheKey(scope.role)}|${functionName}|${[...argumentNames].sort().join(",")}`
: null;
if (cacheKey) {
const cached = readTimed(metadataCache().functions, cacheKey);
if (cached) {
return {
...cached,
argument_names: cached.argument_names ? [...cached.argument_names] : null,
argument_types: [...cached.argument_types],
};
}
}
const result = await client.query<FunctionMetadata>(
`
select
@@ -671,6 +847,13 @@ async function functionMetadata(
);
const metadata = result.rows[0];
if (!metadata) throw new Error("database function not found");
if (cacheKey) {
writeTimed(metadataCache().functions, cacheKey, {
...metadata,
argument_names: metadata.argument_names ? [...metadata.argument_names] : null,
argument_types: [...metadata.argument_types],
});
}
return metadata;
}
@@ -691,12 +874,14 @@ export class LocalPostgresDataClient {
}>;
private readonly pool: Pool;
private readonly connectionString: string;
constructor(
connectionString: string,
private readonly identity: LocalIdentity,
private readonly role: LocalDatabaseRole | null,
) {
this.connectionString = connectionString;
if (role !== null && role !== "authenticated" && role !== "service_role") {
throw new Error("unsupported database role");
}
@@ -709,6 +894,7 @@ export class LocalPostgresDataClient {
from(table: string) {
return new LocalPostgresQueryBuilder(
this.pool,
this.connectionString,
this.identity,
this.role,
table,
@@ -723,12 +909,16 @@ export class LocalPostgresDataClient {
identifier(functionName);
return await inBusinessTransaction(
this.pool,
this.connectionString,
this.identity,
this.role,
async (client) => {
const names = Object.keys(args);
names.forEach(identifier);
const metadata = await functionMetadata(client, functionName, names);
const metadata = await functionMetadata(client, functionName, names, {
connectionString: this.connectionString,
role: this.role,
});
const typeByName = new Map(
(metadata.argument_names ?? []).map((name, index) => [
name,
@@ -173,13 +173,17 @@ function defaultRetryDelayMs(job: PersonalReportJobRecord): number {
async function defaultSleep(milliseconds: number, signal?: AbortSignal): Promise<void> {
if (signal?.aborted) return;
await new Promise<void>((resolve) => {
const timer = globalThis.setTimeout(resolve, milliseconds);
const candidate = timer as ReturnType<typeof globalThis.setTimeout> & { unref?: () => void };
candidate.unref?.();
signal?.addEventListener("abort", () => {
const onAbort = () => {
globalThis.clearTimeout(timer);
resolve();
}, { once: true });
};
const timer = globalThis.setTimeout(() => {
signal?.removeEventListener("abort", onAbort);
resolve();
}, milliseconds);
const candidate = timer as ReturnType<typeof globalThis.setTimeout> & { unref?: () => void };
candidate.unref?.();
signal?.addEventListener("abort", onAbort, { once: true });
});
}
@@ -188,12 +188,47 @@ export async function rescoreMinuteAfterWindowChange(
export { persistBlockScanPayload } from "./score-persist.ts";
const rescoreAttempts = new Map<string, true>();
const RESCORE_ATTEMPT_TTL_MS = 6 * 60 * 60 * 1000;
const RESCORE_ATTEMPT_CAP = 5_000;
const rescoreAttempts = new Map<string, number>();
export function resetStaleMinuteRescoreAttemptsForTests(): void {
rescoreAttempts.clear();
}
function pruneRescoreAttempts(now: number): void {
for (const [key, at] of rescoreAttempts) {
if (now - at > RESCORE_ATTEMPT_TTL_MS) rescoreAttempts.delete(key);
}
while (rescoreAttempts.size > RESCORE_ATTEMPT_CAP) {
const oldest = rescoreAttempts.keys().next().value;
if (oldest === undefined) return;
rescoreAttempts.delete(oldest);
}
}
/** Returns true when this key should be scored now. A fresh key is remembered. */
function claimRescoreAttempt(key: string, now: number): boolean {
pruneRescoreAttempts(now);
const existing = rescoreAttempts.get(key);
if (existing !== undefined && now - existing <= RESCORE_ATTEMPT_TTL_MS) return false;
if (!rescoreAttempts.has(key) && rescoreAttempts.size >= RESCORE_ATTEMPT_CAP) {
const oldest = rescoreAttempts.keys().next().value;
if (oldest !== undefined) rescoreAttempts.delete(oldest);
}
rescoreAttempts.set(key, now);
return true;
}
export function noteRescoreAttemptForTests(key: string, at: number): void {
claimRescoreAttempt(key, at);
}
export function rescoreAttemptCountForTests(): number {
return rescoreAttempts.size;
}
function rescoreAttemptKey(caseId: string, fingerprint: string, askedTurnId?: string | null): string {
return `${caseId}:${askedTurnId ?? ""}:${fingerprint}`;
}
@@ -219,10 +254,9 @@ export async function rescoreStaleMinuteSnapshotIfNeeded(input: {
}
const fingerprint = evidenceLedgerFingerprint(dossier.evidence);
const key = rescoreAttemptKey(input.caseId, fingerprint, input.askedTurnId);
if (rescoreAttempts.has(key)) {
if (!claimRescoreAttempt(key, Date.now())) {
return { dossier, snapshotCurrent: false, rescoreAttempted: false };
}
rescoreAttempts.set(key, true);
try {
await rescoreMinuteAfterWindowChange(input.accounting, input.userId, input.caseId, {
keepAnswers: true,
@@ -6,11 +6,30 @@
*/
const WINDOW_MS = 60_000;
const DELIVERY_MARK_CAP = 10_000;
type DeliveryMark = Readonly<{ resultId: string; at: number; narration?: string }>;
const delivered = new Map<string, DeliveryMark>();
function pruneDelivered(now: number): void {
for (const [caseId, mark] of delivered) {
if (now - mark.at > WINDOW_MS) delivered.delete(caseId);
}
}
function makeRoomForDeliveryMark(): void {
while (delivered.size >= DELIVERY_MARK_CAP) {
const oldest = delivered.keys().next().value;
if (oldest === undefined) return;
delivered.delete(oldest);
}
}
export function deliveryTurnGuardSizeForTests(): number {
return delivered.size;
}
export type DeliveryTurnTrigger =
| "finalizeSuccessfulTurnExit"
| "persistExhaustionGateTurn"
@@ -39,8 +58,10 @@ export function markDeliveryTurn(input: {
}): void {
if (!input.caseId || !input.resultId) return;
const now = input.now ?? Date.now();
pruneDelivered(now);
const narration = input.narration?.replace(/\s+/g, " ").trim() || undefined;
const existing = delivered.get(input.caseId);
if (!existing) makeRoomForDeliveryMark();
if (existing && existing.resultId === input.resultId && now - existing.at <= WINDOW_MS) {
if (existing.narration) return;
delivered.set(input.caseId, {
+27
View File
@@ -32,9 +32,36 @@ function defaultStore(): RequestRateLimitStore {
return state.jyotishaRequestRateLimits;
}
let seenMaxWindowMs = 0;
let consumeCalls = 0;
export function resetRequestRateLimitMaintenanceForTests(): void {
seenMaxWindowMs = 0;
consumeCalls = 0;
}
function sweepExpiredRateLimitKeys(
store: RequestRateLimitStore,
now: number,
maxWindowMs: number,
): void {
for (const [key, stamps] of store) {
let newest = Number.NEGATIVE_INFINITY;
for (const stamp of stamps) {
if (stamp > newest) newest = stamp;
}
if (!Number.isFinite(newest) || newest <= now - maxWindowMs) store.delete(key);
}
}
export function consumeRequestRateLimit(input: RequestRateLimitInput): RequestRateLimitDecision {
if (input.windowMs > seenMaxWindowMs) seenMaxWindowMs = input.windowMs;
consumeCalls += 1;
const now = input.now ?? Date.now();
const store = input.store ?? defaultStore();
if (consumeCalls % 1_000 === 0 || store.size > 5_000) {
sweepExpiredRateLimitKeys(store, now, seenMaxWindowMs);
}
const windowStart = now - input.windowMs;
const recent = (store.get(input.key) ?? []).filter((timestamp) => timestamp > windowStart);
if (recent.length >= input.limit) {
+4 -1
View File
@@ -9,12 +9,15 @@ import { getIdentityAuthServices } from "@/modules/identity/auth";
import { readIdentitySession } from "@/modules/identity/session";
import { getSupabasePublicConfig } from "./config";
export async function createServerSupabaseClient() {
export async function createServerSupabaseClient(options?: {
readonly disableCookieCache?: boolean;
}) {
if (process.env.AUTH_PROVIDER?.trim() === "self-hosted") {
const requestHeaders = new Headers(await headers());
const session = await readIdentitySession(
getIdentityAuthServices().user.api,
requestHeaders,
options?.disableCookieCache ? { disableCookieCache: true } : undefined,
);
return createLocalPostgresDataClient(
readDatabaseUrl(process.env, "APP_DATABASE_URL"),
+19 -3
View File
@@ -1,5 +1,4 @@
import { createHmac } from "node:crypto";
import type { Pool } from "pg";
import type { BetterAuthOptions } from "better-auth";
import { admin, emailOTP, twoFactor, type EmailOTPOptions } from "better-auth/plugins";
@@ -11,10 +10,19 @@ export type AdminUserAuthorizer = (userId: string) => Promise<boolean>;
interface BuildAuthOptionsInput {
config: SelfHostedIdentityConfig;
database: Pool;
database: BetterAuthOptions["database"];
emailSender: EmailOtpSender;
}
export const IDENTITY_SESSION_COOKIE_CACHE_MAX_AGE_SECONDS = 300;
/** Server-only. Unset or any value other than "on" keeps today's uncached session reads. */
export function identitySessionCookieCacheEnabled(
env: NodeJS.ProcessEnv = process.env,
): boolean {
return env.IDENTITY_SESSION_COOKIE_CACHE === "on";
}
function otpIdempotencyKey(
secret: string,
email: string,
@@ -86,7 +94,15 @@ export function buildAuthOptions({
trustedOrigins: [config.userOrigin, config.adminOrigin],
telemetry: { enabled: false },
user: identityModelMapping.user,
session: identityModelMapping.session,
session: identitySessionCookieCacheEnabled()
? {
...identityModelMapping.session,
cookieCache: {
enabled: true,
maxAge: IDENTITY_SESSION_COOKIE_CACHE_MAX_AGE_SECONDS,
},
}
: identityModelMapping.session,
account: identityModelMapping.account,
verification: identityModelMapping.verification,
rateLimit: {
+17 -4
View File
@@ -18,8 +18,15 @@ export interface IdentityServerSession extends IdentitySession {
sessionToken: string;
}
export interface IdentitySessionQuery {
readonly disableCookieCache?: boolean;
}
export interface IdentitySessionReader {
getSession(input: { headers: Headers }): Promise<RawIdentitySession | null>;
getSession(input: {
headers: Headers;
query?: IdentitySessionQuery;
}): Promise<RawIdentitySession | null>;
}
export class IdentityAuthorizationError extends Error {
@@ -43,8 +50,12 @@ function parseRoles(role: string | null | undefined): string[] {
export async function readIdentityServerSession(
reader: IdentitySessionReader,
requestHeaders: Headers,
options?: IdentitySessionQuery,
): Promise<IdentityServerSession | null> {
const value = await reader.getSession({ headers: requestHeaders });
const value = await reader.getSession({
headers: requestHeaders,
...(options?.disableCookieCache ? { query: { disableCookieCache: true } } : {}),
});
if (!value) return null;
const expiresAt = new Date(value.session.expiresAt);
@@ -71,16 +82,18 @@ export async function readIdentityServerSession(
export async function readIdentitySession(
reader: IdentitySessionReader,
requestHeaders: Headers,
options?: IdentitySessionQuery,
): Promise<IdentitySession | null> {
const session = await readIdentityServerSession(reader, requestHeaders);
const session = await readIdentityServerSession(reader, requestHeaders, options);
return session ? { user: session.user, expiresAt: session.expiresAt } : null;
}
export async function requireIdentityServerSession(
reader: IdentitySessionReader,
requestHeaders: Headers,
options?: IdentitySessionQuery,
): Promise<IdentityServerSession> {
const session = await readIdentityServerSession(reader, requestHeaders);
const session = await readIdentityServerSession(reader, requestHeaders, options);
if (!session) throw new IdentityAuthorizationError("Authentication required", 401);
return session;
}