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

Merge pull request #1612 from pikasTech/fix/workbench-bg-db-pressure-1610

Browse Source 
fix: stagger workbench background db scans
This commit is contained in:
Lyon
2026-06-19 16:55:34 +08:00
committed by GitHub
parent 2e52043edd 063633f8e2
commit 1db6016ffc
5 changed files with 80 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/deploy.yaml
+4
View File
@@ -315,6 +315,8 @@ lanes:
HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS: "60000"
HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS: "30000"
HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE: "100"
HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_JITTER_MS: "15000"
HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_FAILURE_BACKOFF_MS: "120000"
observable: true
hwlab-user-billing:
runtimeKind: go-service
@@ -465,6 +467,8 @@ lanes:
HWLAB_WORKBENCH_EMPTY_SESSION_GC_INTERVAL_MS: "60000"
HWLAB_WORKBENCH_EMPTY_SESSION_GC_INITIAL_DELAY_MS: "5000"
HWLAB_WORKBENCH_EMPTY_SESSION_GC_BATCH_SIZE: "100"
HWLAB_WORKBENCH_EMPTY_SESSION_GC_JITTER_MS: "15000"
HWLAB_WORKBENCH_EMPTY_SESSION_GC_FAILURE_BACKOFF_MS: "120000"
- serviceId: hwlab-user-billing
env:
HWLAB_USER_BILLING_DB_URL: secretRef:hwlab-cloud-api-v03-db/database-url
internal/cloud/access-control.ts
+2
View File
@@ -108,6 +108,8 @@ const ACCESS_SCHEMA_STATEMENTS = Object.freeze([
`CREATE INDEX IF NOT EXISTS idx_agent_sessions_owner ON agent_sessions(owner_user_id)`,
`CREATE INDEX IF NOT EXISTS idx_agent_sessions_conversation ON agent_sessions(conversation_id)`,
`CREATE INDEX IF NOT EXISTS idx_agent_sessions_last_trace ON agent_sessions(last_trace_id)`,
`CREATE INDEX IF NOT EXISTS idx_agent_sessions_active_updated ON agent_sessions(updated_at DESC, id DESC) WHERE status <> 'archived'`,
`CREATE INDEX IF NOT EXISTS idx_agent_sessions_owner_active_updated ON agent_sessions(owner_user_id, updated_at DESC, id DESC) WHERE status <> 'archived'`,
`CREATE TABLE IF NOT EXISTS account_workspaces (
id TEXT PRIMARY KEY,
owner_user_id TEXT NOT NULL REFERENCES users(id),
internal/cloud/server-code-agent-http.ts
+34 -11
View File
@@ -44,6 +44,8 @@ const DEFAULT_AGENTRUN_PROJECTION_RESUME_INITIAL_DELAY_MS = 5000;
const DEFAULT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS = 60000;
const DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS = 30000;
const DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE = 100;
const DEFAULT_AGENTRUN_PROJECTION_RESUME_JITTER_MS = 15000;
const DEFAULT_AGENTRUN_PROJECTION_RESUME_FAILURE_BACKOFF_MS = 120000;
const CODE_AGENT_TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled"]);
const CODE_AGENT_PROVIDER_PROFILE_ALIASES = Object.freeze({ codex: "codex-api" });
const CODE_AGENT_PROVIDER_PROFILE_ID_PATTERN = /^[a-z0-9][a-z0-9-]{0,63}$/u;
@@ -879,30 +881,44 @@ export function startAgentRunProjectionResume({ options = {}, traceStore = defau
const intervalMs = parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS);
const minAgeMs = parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS);
const batchSize = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE, DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE);
const jitterMs = intervalMs > 0 ? Math.min(parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_JITTER_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_JITTER_MS), intervalMs) : 0;
const failureBackoffMs = Math.max(intervalMs, parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_FAILURE_BACKOFF_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_FAILURE_BACKOFF_MS));
let stopped = false;
let active = false;
let timer = null;
let nextOffset = 0;
const schedule = (delayMs, reason) => {
if (stopped) return;
const scheduledDelayMs = Math.max(0, Number(delayMs) || 0) + randomJitterMs(jitterMs);
timer = setTimeout(() => {
timer = null;
void runAgentRunProjectionResumePass({ options, traceStore, logger, batchSize, minAgeMs, reason })
.catch((error) => logAgentRunProjectionResume(logger, {
const offset = nextOffset;
active = true;
let failed = false;
void runAgentRunProjectionResumePass({ options, traceStore, logger, batchSize, minAgeMs, offset, reason })
.then((result) => {
failed = result?.failed > 0;
nextOffset = result?.scannedSessions >= batchSize ? offset + batchSize : 0;
})
.catch((error) => {
failed = true;
logAgentRunProjectionResume(logger, {
event: "workbench_projection_resume",
ok: false,
status: "failed",
reason,
offset,
errorCode: error?.code ?? "projection_resume_pass_failed",
message: error?.message ?? "AgentRun projection resume pass failed.",
valuesRedacted: true
}))
});
})
.finally(() => {
active = false;
if (!stopped && intervalMs > 0) schedule(intervalMs, "interval");
if (!stopped && intervalMs > 0) schedule(failed ? failureBackoffMs : intervalMs, failed ? "failure_backoff" : "interval");
});
active = true;
}, Math.max(0, Number(delayMs) || 0));
}, scheduledDelayMs);
timer.unref?.();
};
@@ -913,6 +929,8 @@ export function startAgentRunProjectionResume({ options = {}, traceStore = defau
intervalMs,
minAgeMs,
batchSize,
jitterMs,
failureBackoffMs,
valuesPrinted: false,
stop() {
stopped = true;
@@ -923,8 +941,8 @@ export function startAgentRunProjectionResume({ options = {}, traceStore = defau
};
}
async function runAgentRunProjectionResumePass({ options = {}, traceStore = defaultCodeAgentTraceStore, logger = console, batchSize, minAgeMs, reason = "manual" } = {}) {
const sessions = await listAgentRunProjectionResumeSessions(options, batchSize);
async function runAgentRunProjectionResumePass({ options = {}, traceStore = defaultCodeAgentTraceStore, logger = console, batchSize, minAgeMs, offset = 0, reason = "manual" } = {}) {
const sessions = await listAgentRunProjectionResumeSessions(options, batchSize, offset);
const candidates = sessions.flatMap((session) => agentRunProjectionResumeCandidates(session, { minAgeMs }));
let resumed = 0;
let terminal = 0;
@@ -944,6 +962,7 @@ async function runAgentRunProjectionResumePass({ options = {}, traceStore = defa
ok: failed === 0,
status: failed === 0 ? "completed" : "degraded",
reason,
offset,
scannedSessions: sessions.length,
candidates: candidates.length,
resumed,
@@ -951,10 +970,10 @@ async function runAgentRunProjectionResumePass({ options = {}, traceStore = defa
failed,
valuesRedacted: true
});
return { scannedSessions: sessions.length, candidates: candidates.length, resumed, terminal, failed };
return { offset, scannedSessions: sessions.length, candidates: candidates.length, resumed, terminal, failed };
}
async function listAgentRunProjectionResumeSessions(options = {}, batchSize = DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE) {
async function listAgentRunProjectionResumeSessions(options = {}, batchSize = DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE, offset = 0) {
const accessStore = options.accessController?.store ?? options.accessController ?? null;
if (typeof accessStore?.listAgentSessionsForUser !== "function") return [];
return await accessStore.listAgentSessionsForUser({
@@ -962,7 +981,7 @@ async function listAgentRunProjectionResumeSessions(options = {}, batchSize = DE
ownerScoped: false,
includeArchived: false,
limit: batchSize,
offset: 0
offset
}) ?? [];
}
@@ -1106,6 +1125,10 @@ function parseNonNegativeInteger(value, fallback) {
return Number.isInteger(parsed) && parsed >= 0 ? parsed : fallback;
}
function randomJitterMs(maxJitterMs) {
return maxJitterMs > 0 ? Math.floor(Math.random() * (maxJitterMs + 1)) : 0;
}
function agentRunProjectionActivitySignature(result, runnerTrace) {
const agentRun = result?.agentRun && typeof result.agentRun === "object" ? result.agentRun : {};
const trace = runnerTrace && typeof runnerTrace === "object" ? runnerTrace : result?.runnerTrace;
internal/cloud/workbench-empty-session-gc.ts
+38 -11
View File
@@ -7,6 +7,8 @@ export const DEFAULT_WORKBENCH_EMPTY_SESSION_TTL_MS = 10 * 60 * 1000;
export const DEFAULT_WORKBENCH_EMPTY_SESSION_GC_INTERVAL_MS = 60 * 1000;
export const DEFAULT_WORKBENCH_EMPTY_SESSION_GC_INITIAL_DELAY_MS = 5 * 1000;
export const DEFAULT_WORKBENCH_EMPTY_SESSION_GC_BATCH_SIZE = 100;
export const DEFAULT_WORKBENCH_EMPTY_SESSION_GC_JITTER_MS = 15 * 1000;
export const DEFAULT_WORKBENCH_EMPTY_SESSION_GC_FAILURE_BACKOFF_MS = 2 * 60 * 1000;
const RUNNING_STATUSES = new Set(["active", "accepted", "busy", "creating", "dispatching", "pending", "queued", "running", "streaming"]);
const TERMINAL_RESULT_STATUSES = new Set(["blocked", "canceled", "cancelled", "completed", "failed", "timeout"]);
@@ -19,23 +21,29 @@ export function startWorkbenchEmptySessionGc(options: any = {}) {
const intervalMs = boundedInteger(env.HWLAB_WORKBENCH_EMPTY_SESSION_GC_INTERVAL_MS, DEFAULT_WORKBENCH_EMPTY_SESSION_GC_INTERVAL_MS, { min: 1_000, max: 3_600_000 });
const initialDelayMs = boundedInteger(env.HWLAB_WORKBENCH_EMPTY_SESSION_GC_INITIAL_DELAY_MS, DEFAULT_WORKBENCH_EMPTY_SESSION_GC_INITIAL_DELAY_MS, { min: 0, max: 3_600_000 });
const batchSize = boundedInteger(env.HWLAB_WORKBENCH_EMPTY_SESSION_GC_BATCH_SIZE, DEFAULT_WORKBENCH_EMPTY_SESSION_GC_BATCH_SIZE, { min: 1, max: 100 });
const jitterMs = boundedInteger(env.HWLAB_WORKBENCH_EMPTY_SESSION_GC_JITTER_MS, DEFAULT_WORKBENCH_EMPTY_SESSION_GC_JITTER_MS, { min: 0, max: intervalMs });
const failureBackoffMs = boundedInteger(env.HWLAB_WORKBENCH_EMPTY_SESSION_GC_FAILURE_BACKOFF_MS, DEFAULT_WORKBENCH_EMPTY_SESSION_GC_FAILURE_BACKOFF_MS, { min: intervalMs, max: 3_600_000 });
const canMutate = typeof store?.listAgentSessionsForUser === "function" && typeof store?.archiveAgentConversation === "function";
if (!enabled || !canMutate) {
return disabledGcHandle({ enabled, reason: enabled ? "session_store_missing_lifecycle_mutation" : "disabled", ttlMs, intervalMs, initialDelayMs, batchSize });
return disabledGcHandle({ enabled, reason: enabled ? "session_store_missing_lifecycle_mutation" : "disabled", ttlMs, intervalMs, initialDelayMs, batchSize, jitterMs, failureBackoffMs });
}
let stopped = false;
let running = false;
let timer: ReturnType<typeof setTimeout> | null = null;
let nextOffset = 0;
const tick = async (reason = "manual") => {
if (stopped) return { ok: false, status: "stopped", reason };
if (running) return { ok: true, status: "skipped", reason: "already_running" };
running = true;
const offset = nextOffset;
try {
const result = await runWorkbenchEmptySessionGc({ store, ttlMs, batchSize, now: options.now, reason });
const result = await runWorkbenchEmptySessionGc({ store, ttlMs, batchSize, offset, now: options.now, reason });
if (result?.ok) nextOffset = result.scanned >= batchSize ? offset + batchSize : 0;
logGcResult(options.logger, result);
return result;
} catch (error) {
const result = { ok: false, status: "failed", reason, error: error instanceof Error ? error.message : String(error), valuesRedacted: true };
const result = { ok: false, status: "failed", reason, offset, error: error instanceof Error ? error.message : String(error), valuesRedacted: true };
logGcResult(options.logger, result);
return result;
} finally {
@@ -43,10 +51,21 @@ export function startWorkbenchEmptySessionGc(options: any = {}) {
}
};
const startupTimer = setTimeout(() => { void tick("startup"); }, initialDelayMs);
startupTimer.unref?.();
const interval = setInterval(() => { void tick("interval"); }, intervalMs);
interval.unref?.();
const schedule = (delayMs: number, reason: string) => {
if (stopped) return;
const actualDelayMs = Math.max(0, delayMs) + randomJitterMs(jitterMs);
timer = setTimeout(() => {
timer = null;
void tick(reason).then((result: any) => {
if (stopped) return;
const failed = result?.ok === false || result?.status === "failed";
schedule(failed ? failureBackoffMs : intervalMs, failed ? "failure_backoff" : "interval");
});
}, actualDelayMs);
timer.unref?.();
};
schedule(initialDelayMs, "startup");
return {
enabled: true,
@@ -55,16 +74,18 @@ export function startWorkbenchEmptySessionGc(options: any = {}) {
intervalMs,
initialDelayMs,
batchSize,
jitterMs,
failureBackoffMs,
tick,
stop() {
stopped = true;
clearTimeout(startupTimer);
clearInterval(interval);
if (timer) clearTimeout(timer);
timer = null;
}
};
}
export async function runWorkbenchEmptySessionGc({ store, ttlMs = DEFAULT_WORKBENCH_EMPTY_SESSION_TTL_MS, batchSize = DEFAULT_WORKBENCH_EMPTY_SESSION_GC_BATCH_SIZE, now, reason = "manual" }: any = {}) {
export async function runWorkbenchEmptySessionGc({ store, ttlMs = DEFAULT_WORKBENCH_EMPTY_SESSION_TTL_MS, batchSize = DEFAULT_WORKBENCH_EMPTY_SESSION_GC_BATCH_SIZE, offset = 0, now, reason = "manual" }: any = {}) {
if (typeof store?.listAgentSessionsForUser !== "function" || typeof store?.archiveAgentConversation !== "function") {
return { ok: false, status: "blocked", reason: "session_store_missing_lifecycle_mutation", ttlMs, batchSize, valuesRedacted: true };
}
@@ -74,7 +95,8 @@ export async function runWorkbenchEmptySessionGc({ store, ttlMs = DEFAULT_WORKBE
ownerScoped: false,
actorRole: "admin",
includeArchived: false,
limit: batchSize
limit: batchSize,
offset
}) ?? [];
const keptByReason: Record<string, number> = {};
const archivedSessionIds: string[] = [];
@@ -100,6 +122,7 @@ export async function runWorkbenchEmptySessionGc({ store, ttlMs = DEFAULT_WORKBE
cutoffAt: new Date(nowMs - ttlMs).toISOString(),
ttlMs,
batchSize,
offset,
scanned: sessions.length,
archived: archivedSessionIds.length,
archivedSessionIds,
@@ -167,6 +190,10 @@ function boundedInteger(value: unknown, fallback: number, { min, max }: { min: n
return Math.min(Math.max(candidate, min), max);
}
function randomJitterMs(maxJitterMs: number) {
return maxJitterMs > 0 ? Math.floor(Math.random() * (maxJitterMs + 1)) : 0;
}
function timestampFor(now: unknown) {
const value = typeof now === "function" ? (now as () => unknown)() : now;
if (value instanceof Date && !Number.isNaN(value.valueOf())) return value.toISOString();
internal/db/migrations/0001_cloud_core_skeleton.sql
+2
View File
@@ -121,6 +121,8 @@ ALTER TABLE agent_sessions ADD COLUMN IF NOT EXISTS session_json TEXT NOT NULL D
ALTER TABLE agent_sessions ADD COLUMN IF NOT EXISTS updated_at TEXT;
CREATE INDEX IF NOT EXISTS idx_agent_sessions_owner ON agent_sessions(owner_user_id);
CREATE INDEX IF NOT EXISTS idx_agent_sessions_conversation ON agent_sessions(conversation_id);
CREATE INDEX IF NOT EXISTS idx_agent_sessions_active_updated ON agent_sessions(updated_at DESC, id DESC) WHERE status <> 'archived';
CREATE INDEX IF NOT EXISTS idx_agent_sessions_owner_active_updated ON agent_sessions(owner_user_id, updated_at DESC, id DESC) WHERE status <> 'archived';
CREATE TABLE IF NOT EXISTS account_workspaces (
id TEXT PRIMARY KEY,