mirror of
https://github.com/openclaw/openclaw.git
synced 2026-10-03 17:53:39 +00:00
fix(gateway): reduce worker overhead with large session stores (#149700)
* fix(gateway): avoid full-store hydration for worker session lookup Punchcard-Session: brisk-summit-brook-mr * improve(bench): measure worker transcript commit scaling Punchcard-Session: brisk-summit-brook-mr * fix(gateway): avoid unrelated prompt reads in worker target lookup Use current-main list discovery and register the real worker commit benchmark through the supported source runner. Punchcard-Session: clear-workshop-brook-w2 * test(gateway): verify worker lookups skip unrelated prompts Punchcard-Session: clear-workshop-brook-w2 * fix(bench): use retained session key reader for fixture counts Keep the seeded session-count assertion outside measured commit work after the runtime count API retirement. Punchcard-Session: ember-river-harbor-t4 * fix(ci): settle streaming fixtures and update native test expectations Punchcard-Session: ember-river-harbor-t4 * fix(test): keep child pagination and dashboard state coherent Punchcard-Session: ember-river-harbor-t4 * test(ui): split session fixtures and child-query cases Punchcard-Session: ember-river-harbor-t4 * test(ui): reuse serialized record predicate Punchcard-Session: ember-river-harbor-t4 * test(ui): report stalled inactive-agent outbox stage Punchcard-Session: ember-river-harbor-t4
This commit is contained in:
Vincent Koc
GitHub
parent
ace5752a4d
commit
73775cb8a2
13 changed files with 985 additions and 136 deletions
387
scripts/bench-worker-transcript-commit.ts
Normal file
387
scripts/bench-worker-transcript-commit.ts
Normal file
|
|
@ -0,0 +1,387 @@
|
|||
// Run with pnpm test:gateway:worker-commit.
|
||||
// Measures the real committer owner with synthetic admission, not worker transport or UI delivery.
|
||||
import assert from "node:assert/strict";
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import { Session as InspectorSession } from "node:inspector/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { monitorEventLoopDelay, performance } from "node:perf_hooks";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import { parseArgs } from "node:util";
|
||||
import type {
|
||||
WorkerTranscriptCommitParams,
|
||||
WorkerTranscriptMessage,
|
||||
} from "../packages/gateway-protocol/src/schema/worker-admission.js";
|
||||
import { SessionManager } from "../src/agents/sessions/session-manager.js";
|
||||
import { createZeroUsageFixture } from "../src/agents/test-helpers/usage-fixtures.js";
|
||||
import {
|
||||
listSessionEntryKeysReadOnly,
|
||||
loadSessionEntry,
|
||||
persistSessionTranscriptTurn,
|
||||
upsertSessionEntryCore,
|
||||
} from "../src/config/sessions/session-accessor.js";
|
||||
import { waitForSessionTranscriptIndexReconcilesInStateDir } from "../src/config/sessions/session-transcript-reconcile.js";
|
||||
import type { OpenClawConfig } from "../src/config/types.openclaw.js";
|
||||
import type { WorkerConnectionIdentity } from "../src/gateway/worker-environments/connection-identity.js";
|
||||
import { createWorkerTranscriptCommitStore } from "../src/gateway/worker-environments/transcript-commit-store.js";
|
||||
import { createWorkerTranscriptCommitter } from "../src/gateway/worker-environments/transcript-commit.js";
|
||||
import { onSessionTranscriptUpdate } from "../src/sessions/transcript-events.js";
|
||||
import { openOpenClawStateDatabase } from "../src/state/openclaw-state-db.js";
|
||||
import {
|
||||
withOpenClawTestState,
|
||||
type OpenClawTestState,
|
||||
} from "../src/test-utils/openclaw-test-state.js";
|
||||
|
||||
const { values } = parseArgs({
|
||||
options: {
|
||||
"active-events": { type: "string", default: "1000" },
|
||||
"abandoned-events": { type: "string", default: "0" },
|
||||
"session-count": { type: "string", default: "1" },
|
||||
"message-chars": { type: "string", default: "1024" },
|
||||
commits: { type: "string", default: "12" },
|
||||
runs: { type: "string", default: "3" },
|
||||
output: { type: "string" },
|
||||
"cpu-profile": { type: "string" },
|
||||
},
|
||||
});
|
||||
|
||||
function integer(value: string, name: string, minimum: number, maximum: number): number {
|
||||
const number = Number(value);
|
||||
if (!Number.isSafeInteger(number) || number < minimum || number > maximum) {
|
||||
throw new Error(`${name} must be an integer between ${minimum} and ${maximum}`);
|
||||
}
|
||||
return number;
|
||||
}
|
||||
|
||||
const fixture = {
|
||||
activeEvents: integer(values["active-events"], "active-events", 2, 10_000),
|
||||
abandonedEvents: integer(values["abandoned-events"], "abandoned-events", 0, 10_000),
|
||||
sessionCount: integer(values["session-count"], "session-count", 1, 1000),
|
||||
messageChars: integer(values["message-chars"], "message-chars", 64, 4096),
|
||||
commits: integer(values.commits, "commits", 1, 100),
|
||||
};
|
||||
const runs = integer(values.runs, "runs", 1, 10);
|
||||
if (values["cpu-profile"] && runs !== 1) {
|
||||
throw new Error(
|
||||
"cpu-profile requires runs=1; keep diagnostic profiles separate from timing runs",
|
||||
);
|
||||
}
|
||||
const sessionId = "worker-commit-benchmark";
|
||||
const sessionKey = `agent:main:${sessionId}`;
|
||||
const runEpoch = 7;
|
||||
const skillsPrompt = "Benchmark skill instructions. ".repeat(600).slice(0, 16 * 1024);
|
||||
const identity: WorkerConnectionIdentity = {
|
||||
environmentId: "benchmark-environment",
|
||||
credentialHash: "benchmark-credential-hash",
|
||||
bundleHash: "b".repeat(64),
|
||||
sessionId,
|
||||
runId: "benchmark-run",
|
||||
turnClaim: {
|
||||
sessionId,
|
||||
claimId: "benchmark-claim",
|
||||
runId: "benchmark-run",
|
||||
placementGeneration: 4,
|
||||
owner: { kind: "worker", environmentId: "benchmark-environment", ownerEpoch: runEpoch },
|
||||
},
|
||||
ownerEpoch: runEpoch,
|
||||
rpcSetVersion: 1,
|
||||
protocolFeatures: ["worker-transcript-commit-v1"],
|
||||
credentialExpiresAtMs: 10_000,
|
||||
};
|
||||
|
||||
function messages(seq: number): WorkerTranscriptMessage[] {
|
||||
const toolCallId = `benchmark-read-${seq}`;
|
||||
const assistant = {
|
||||
role: "assistant" as const,
|
||||
api: "openai-responses",
|
||||
provider: "openai",
|
||||
model: "gpt-4.1",
|
||||
usage: createZeroUsageFixture(),
|
||||
timestamp: seq * 10 + 1,
|
||||
};
|
||||
return [
|
||||
{
|
||||
role: "user",
|
||||
content: [{ type: "text", text: `Inspect item ${seq}.` }],
|
||||
timestamp: seq * 10,
|
||||
},
|
||||
{
|
||||
...assistant,
|
||||
content: [
|
||||
{ type: "toolCall", id: toolCallId, name: "read", arguments: { path: "README.md" } },
|
||||
],
|
||||
stopReason: "toolUse",
|
||||
},
|
||||
{
|
||||
role: "toolResult",
|
||||
toolCallId,
|
||||
toolName: "read",
|
||||
content: [{ type: "text", text: "x".repeat(fixture.messageChars) }],
|
||||
isError: false,
|
||||
timestamp: seq * 10 + 2,
|
||||
},
|
||||
{
|
||||
...assistant,
|
||||
content: [{ type: "text", text: `Item ${seq} inspected.` }],
|
||||
stopReason: "stop",
|
||||
timestamp: seq * 10 + 3,
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
function percentile(samples: number[], fraction: number): number {
|
||||
const sorted = samples.toSorted((left, right) => left - right);
|
||||
return sorted[Math.min(sorted.length - 1, Math.ceil(sorted.length * fraction) - 1)] ?? 0;
|
||||
}
|
||||
|
||||
async function seedFixture(state: OpenClawTestState, shape: typeof fixture) {
|
||||
const storePath = path.join(state.sessionsDir("main"), "sessions.json");
|
||||
const config: OpenClawConfig = {
|
||||
agents: { list: [{ id: "main", default: true }] },
|
||||
session: {
|
||||
mainKey: "main",
|
||||
store: path.join(state.stateDir, "agents", "{agentId}", "sessions", "sessions.json"),
|
||||
},
|
||||
};
|
||||
for (let index = 0; index < shape.sessionCount; index += 1) {
|
||||
await upsertSessionEntryCore(
|
||||
{
|
||||
agentId: "main",
|
||||
sessionKey: index === 0 ? sessionKey : `agent:main:unrelated-${index}`,
|
||||
storePath,
|
||||
},
|
||||
{
|
||||
sessionId: index === 0 ? sessionId : `unrelated-${index}`,
|
||||
lifecycleRevision: "benchmark-revision",
|
||||
updatedAt: 1,
|
||||
skillsSnapshot: { prompt: skillsPrompt, skills: [] },
|
||||
},
|
||||
);
|
||||
}
|
||||
const target = { agentId: "main", sessionId, sessionKey, storePath };
|
||||
const content = "x".repeat(shape.messageChars);
|
||||
const seed = Array.from(
|
||||
{ length: shape.activeEvents - 1 + shape.abandonedEvents },
|
||||
(_, index) => ({
|
||||
eventId: `seed-${index}`,
|
||||
parentId: index === 0 ? null : `seed-${index - 1}`,
|
||||
message: { role: index % 2 === 0 ? "user" : "assistant", content },
|
||||
}),
|
||||
);
|
||||
await persistSessionTranscriptTurn(target, { messages: seed, touchSessionEntry: false });
|
||||
// branch() is an in-memory cursor move. Appending an anchor persists the selected branch.
|
||||
const seeded = SessionManager.open(target);
|
||||
seeded.branch(`seed-${shape.activeEvents - 2}`);
|
||||
const baseLeafId = seeded.appendMessage({
|
||||
role: "user",
|
||||
content: "Benchmark anchor",
|
||||
timestamp: 1,
|
||||
});
|
||||
await waitForSessionTranscriptIndexReconcilesInStateDir(state.stateDir);
|
||||
const verifiedSeed = SessionManager.open(target);
|
||||
assert.equal(verifiedSeed.getLeafId(), baseLeafId);
|
||||
assert.equal(
|
||||
verifiedSeed.getBranch().filter((entry) => entry.type === "message").length,
|
||||
shape.activeEvents,
|
||||
);
|
||||
assert.equal(
|
||||
verifiedSeed.getEntries().filter((entry) => entry.type === "message").length,
|
||||
shape.activeEvents + shape.abandonedEvents,
|
||||
);
|
||||
const initialDagEvents = verifiedSeed.getEntries().length;
|
||||
const sessionRows = (await listSessionEntryKeysReadOnly({ agentId: "main", storePath })).length;
|
||||
assert.equal(sessionRows, shape.sessionCount);
|
||||
const seedSha256 = createHash("sha256").update(JSON.stringify(seed)).digest("hex");
|
||||
return { target, config, baseLeafId, initialDagEvents, sessionRows, seedSha256 };
|
||||
}
|
||||
|
||||
async function runSample(shape: typeof fixture, profilePath?: string) {
|
||||
const sampleStart = performance.now();
|
||||
const result = await withOpenClawTestState(
|
||||
{ label: "worker-commit-benchmark" },
|
||||
async (state) => {
|
||||
const { target, config, baseLeafId, initialDagEvents, sessionRows, seedSha256 } =
|
||||
await seedFixture(state, shape);
|
||||
const committer = createWorkerTranscriptCommitter({
|
||||
getConfig: () => config,
|
||||
store: createWorkerTranscriptCommitStore({ database: openOpenClawStateDatabase() }),
|
||||
});
|
||||
const updates: Array<{ messageId: string | undefined; messageSeq: number | undefined }> = [];
|
||||
const unsubscribe = onSessionTranscriptUpdate((update) => {
|
||||
if (update.target.sessionId === sessionId) {
|
||||
updates.push({ messageId: update.messageId, messageSeq: update.messageSeq });
|
||||
}
|
||||
});
|
||||
const batches = Array.from({ length: shape.commits }, (_, index) => messages(index + 1));
|
||||
const inspector = profilePath ? new InspectorSession() : undefined;
|
||||
const loopDelay = monitorEventLoopDelay({ resolution: 1 });
|
||||
try {
|
||||
inspector?.connect();
|
||||
if (inspector) {
|
||||
await inspector.post("Profiler.enable");
|
||||
await inspector.post("Profiler.start");
|
||||
}
|
||||
loopDelay.enable();
|
||||
await delay(5);
|
||||
loopDelay.reset();
|
||||
const memoryBefore = process.memoryUsage();
|
||||
const cpuBefore = process.cpuUsage();
|
||||
const loopBefore = performance.eventLoopUtilization();
|
||||
const started = performance.now();
|
||||
const durationsMs: number[] = [];
|
||||
const entryIds: string[] = [];
|
||||
let leaf = baseLeafId;
|
||||
let lastRequest: WorkerTranscriptCommitParams | undefined;
|
||||
let lastOutcome: Awaited<ReturnType<typeof committer.commit>> | undefined;
|
||||
for (let index = 0; index < batches.length; index += 1) {
|
||||
const request = { runEpoch, seq: index + 1, baseLeafId: leaf, messages: batches[index]! };
|
||||
const start = performance.now();
|
||||
const outcome = await committer.commit({
|
||||
identity,
|
||||
request,
|
||||
assertCurrent: () => undefined,
|
||||
});
|
||||
durationsMs.push(performance.now() - start);
|
||||
assert.equal(
|
||||
outcome.ok,
|
||||
true,
|
||||
`commit ${index + 1} rejected: ${JSON.stringify(outcome)}`,
|
||||
);
|
||||
if (!outcome.ok) {
|
||||
throw new Error("unreachable rejected commit");
|
||||
}
|
||||
assert.equal(outcome.result.entryIds.length, 4);
|
||||
assert.equal(outcome.result.newLeafId, outcome.result.entryIds.at(-1));
|
||||
entryIds.push(...outcome.result.entryIds);
|
||||
leaf = outcome.result.newLeafId;
|
||||
lastRequest = request;
|
||||
lastOutcome = outcome;
|
||||
}
|
||||
const elapsedMs = performance.now() - started;
|
||||
const cpu = process.cpuUsage(cpuBefore);
|
||||
const loopUtilization = performance.eventLoopUtilization(loopBefore);
|
||||
const memoryAfter = process.memoryUsage();
|
||||
await delay(5);
|
||||
loopDelay.disable();
|
||||
if (inspector && profilePath) {
|
||||
const { profile } = await inspector.post("Profiler.stop");
|
||||
await fs.writeFile(profilePath, JSON.stringify(profile));
|
||||
}
|
||||
assert.equal(new Set(entryIds).size, shape.commits * 4);
|
||||
assert.deepEqual(
|
||||
updates,
|
||||
entryIds.map((messageId, index) => ({
|
||||
messageId,
|
||||
messageSeq: shape.activeEvents + index + 1,
|
||||
})),
|
||||
);
|
||||
const publicationCount = updates.length;
|
||||
assert.ok(lastRequest);
|
||||
assert.deepEqual(
|
||||
await committer.commit({
|
||||
identity,
|
||||
request: lastRequest,
|
||||
assertCurrent: () => undefined,
|
||||
}),
|
||||
lastOutcome,
|
||||
);
|
||||
assert.equal(updates.length, publicationCount, "ledger replay must not publish again");
|
||||
const final = SessionManager.open(target);
|
||||
assert.equal(final.getLeafId(), leaf);
|
||||
assert.equal(loadSessionEntry(target)?.skillsSnapshot?.prompt, skillsPrompt);
|
||||
assert.deepEqual(
|
||||
final
|
||||
.getBranch()
|
||||
.slice(-entryIds.length)
|
||||
.map((entry) => entry.id),
|
||||
entryIds,
|
||||
);
|
||||
assert.equal(
|
||||
final.getBranch().filter((entry) => entry.type === "message").length,
|
||||
shape.activeEvents + entryIds.length,
|
||||
);
|
||||
assert.equal(
|
||||
final.getEntries().filter((entry) => entry.type === "message").length,
|
||||
shape.activeEvents + shape.abandonedEvents + entryIds.length,
|
||||
);
|
||||
const drainStart = performance.now();
|
||||
await waitForSessionTranscriptIndexReconcilesInStateDir(state.stateDir);
|
||||
const reconcileDrainMs = performance.now() - drainStart;
|
||||
return {
|
||||
requestedCommits: shape.commits,
|
||||
completedCommits: durationsMs.length,
|
||||
appendedMessages: entryIds.length,
|
||||
emittedUpdates: publicationCount,
|
||||
initialDagEvents,
|
||||
sessionRows,
|
||||
seedSha256,
|
||||
batchSha256: createHash("sha256").update(JSON.stringify(batches)).digest("hex"),
|
||||
elapsedMs,
|
||||
commitsPerSecond: shape.commits / (elapsedMs / 1000),
|
||||
commitMs: {
|
||||
samples: durationsMs,
|
||||
p50: percentile(durationsMs, 0.5),
|
||||
p95: percentile(durationsMs, 0.95),
|
||||
p99: percentile(durationsMs, 0.99),
|
||||
},
|
||||
processCpuMs: { user: cpu.user / 1000, system: cpu.system / 1000 },
|
||||
eventLoop: {
|
||||
utilization: loopUtilization.utilization,
|
||||
delayP99Ms: loopDelay.percentile(99) / 1e6,
|
||||
delayMaxMs: loopDelay.max / 1e6,
|
||||
samples: loopDelay.count,
|
||||
},
|
||||
memory: {
|
||||
before: memoryBefore,
|
||||
after: memoryAfter,
|
||||
processLifetimePeakRssBytes: process.resourceUsage().maxRSS * 1024,
|
||||
},
|
||||
reconcileDrainMs,
|
||||
};
|
||||
} finally {
|
||||
loopDelay.disable();
|
||||
inspector?.disconnect();
|
||||
unsubscribe();
|
||||
}
|
||||
},
|
||||
);
|
||||
return { ...result, fixtureAndCleanupMs: performance.now() - sampleStart - result.elapsedMs };
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
// Load the lazy committer runtime through its real entrypoint, using a separate state and ledger.
|
||||
await runSample({ ...fixture, activeEvents: 2, abandonedEvents: 0, sessionCount: 1, commits: 1 });
|
||||
const samples = [];
|
||||
for (let run = 0; run < runs; run += 1) {
|
||||
samples.push(await runSample(fixture, values["cpu-profile"]));
|
||||
}
|
||||
const output = {
|
||||
scope:
|
||||
"committer owner; synthetic admission; synchronous event emission, no worker transport or UI delivery",
|
||||
warmup: "one real commit on separate state before fresh-state samples",
|
||||
fixtureIdentity:
|
||||
"deterministic seed and request hashes; runtime-generated entry IDs and timestamps vary",
|
||||
fixture,
|
||||
sessionSkillsPromptBytes: Buffer.byteLength(skillsPrompt),
|
||||
runtime: {
|
||||
node: process.version,
|
||||
versions: process.versions,
|
||||
availableParallelism: os.availableParallelism(),
|
||||
logicalCpus: os.cpus().length,
|
||||
},
|
||||
profiled: Boolean(values["cpu-profile"]),
|
||||
samples,
|
||||
};
|
||||
const json = `${JSON.stringify(output, null, 2)}\n`;
|
||||
if (values.output) {
|
||||
await fs.writeFile(values.output, json);
|
||||
}
|
||||
process.stdout.write(json);
|
||||
}
|
||||
|
||||
await main().catch((error: unknown) => {
|
||||
console.error(error);
|
||||
console.error("[bench-worker-transcript-commit] FAILED (exit 1)");
|
||||
process.exitCode = 1;
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue