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

fix(workbench): bound durable read fan-out (#1954)

Browse Source
This commit is contained in:
Lyon
2026-06-23 06:48:49 +08:00
committed by GitHub
parent bcb3a4d02e
commit 0d0fdd0b51
2 changed files with 21 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/server-workbench-http.ts
+9 -7
View File
@@ -582,14 +582,16 @@ async function queryFactsForSessionSummaries(readModel, sessions = []) {
const sessionIds = uniqueText(sessions.map(factSessionId));
const traceIds = uniqueText(sessions.map(factLastTraceId));
if (sessionIds.length === 0 && traceIds.length === 0) return { facts: emptyFactSet(), count: 0, durable: false, persistence: null };
const bySessionPromise = sessionIds.length > 0
? readModel.queryFacts({ sessionIds, families: ["messages", "turns"] })
: Promise.resolve({ facts: emptyFactSet(), count: 0, durable: false, persistence: null });
const byTracePromise = traceIds.length > 0
? readModel.queryFacts({ traceIds, families: ["checkpoints"] })
: Promise.resolve({ facts: emptyFactSet(), count: 0, durable: false, persistence: null });
const [bySession, byTrace] = await Promise.all([bySessionPromise, byTracePromise]);
// Session list summary is hot and is also sampled by the two-page web-probe
// observer. Avoid parallel fan-out here so one list request cannot consume
// multiple durable-read connections while another page is refreshing.
const bySession = sessionIds.length > 0
? await readModel.queryFacts({ sessionIds, families: ["messages", "turns"] })
: { facts: emptyFactSet(), count: 0, durable: false, persistence: null };
if (bySession.error) return bySession;
const byTrace = traceIds.length > 0
? await readModel.queryFacts({ traceIds, families: ["checkpoints"] })
: { facts: emptyFactSet(), count: 0, durable: false, persistence: null };
if (byTrace.error) return byTrace;
const facts = mergeFactSets(bySession.facts, byTrace.facts);
return { facts, count: Object.values(facts).reduce((sum, rows) => sum + rows.length, 0), durable: bySession.durable ?? byTrace.durable, persistence: bySession.persistence ?? byTrace.persistence ?? null };
internal/db/runtime-store.ts
+12 -8
View File
@@ -1342,14 +1342,18 @@ export class PostgresCloudRuntimeStore {
async queryWorkbenchFacts(params = {}) {
await this.assertReadyForDurableReads("workbench.facts.query");
const families = workbenchFactFamilySet(params);
const entries = await Promise.all([
["sessions", families.has("sessions") ? this.queryWorkbenchFactRows("workbench.sessions.query", "workbench_sessions", "session_json", params, { sessionId: "session_id", ownerUserId: "owner_user_id", projectId: "project_id", conversationId: "conversation_id", threadId: "thread_id", traceId: "last_trace_id", lastTraceId: "last_trace_id", status: "status" }) : Promise.resolve([])],
["messages", families.has("messages") ? this.queryWorkbenchFactRows("workbench.messages.query", "workbench_messages", "message_json", params, { messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", role: "role", status: "status" }) : Promise.resolve([])],
["parts", families.has("parts") ? this.queryWorkbenchFactRows("workbench.parts.query", "workbench_parts", "part_json", params, { partId: "part_id", messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", partType: "part_type", status: "status" }) : Promise.resolve([])],
["turns", families.has("turns") ? this.queryWorkbenchFactRows("workbench.turns.query", "workbench_turns", "turn_json", params, { turnId: "turn_id", sessionId: "session_id", traceId: "trace_id", messageId: "message_id", status: "status" }) : Promise.resolve([])],
["traceEvents", families.has("traceEvents") ? this.queryWorkbenchFactRows("workbench.trace-events.query", "workbench_trace_events", "event_json", params, { id: "id", traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", messageId: "message_id", eventType: "event_type" }) : Promise.resolve([])],
["checkpoints", families.has("checkpoints") ? this.queryWorkbenchFactRows("workbench.projection-checkpoints.query", "workbench_projection_checkpoints", "checkpoint_json", params, { traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", runId: "run_id", commandId: "command_id", projectionStatus: "projection_status", projectionHealth: "projection_health" }) : Promise.resolve([])]
].map(async ([family, rowsPromise]) => [family, await rowsPromise]));
// A single Workbench read can request several fact families. Running those
// SQL reads in parallel lets one HTTP request occupy the whole Postgres
// pool, which makes the control+observer pages amplify into 503s under
// periodic refresh. Keep this read model bounded to one connection per
// request; the page-level latency is preferable to pool starvation.
const entries = [];
entries.push(["sessions", families.has("sessions") ? await this.queryWorkbenchFactRows("workbench.sessions.query", "workbench_sessions", "session_json", params, { sessionId: "session_id", ownerUserId: "owner_user_id", projectId: "project_id", conversationId: "conversation_id", threadId: "thread_id", traceId: "last_trace_id", lastTraceId: "last_trace_id", status: "status" }) : []]);
entries.push(["messages", families.has("messages") ? await this.queryWorkbenchFactRows("workbench.messages.query", "workbench_messages", "message_json", params, { messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", role: "role", status: "status" }) : []]);
entries.push(["parts", families.has("parts") ? await this.queryWorkbenchFactRows("workbench.parts.query", "workbench_parts", "part_json", params, { partId: "part_id", messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", partType: "part_type", status: "status" }) : []]);
entries.push(["turns", families.has("turns") ? await this.queryWorkbenchFactRows("workbench.turns.query", "workbench_turns", "turn_json", params, { turnId: "turn_id", sessionId: "session_id", traceId: "trace_id", messageId: "message_id", status: "status" }) : []]);
entries.push(["traceEvents", families.has("traceEvents") ? await this.queryWorkbenchFactRows("workbench.trace-events.query", "workbench_trace_events", "event_json", params, { id: "id", traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", messageId: "message_id", eventType: "event_type" }) : []]);
entries.push(["checkpoints", families.has("checkpoints") ? await this.queryWorkbenchFactRows("workbench.projection-checkpoints.query", "workbench_projection_checkpoints", "checkpoint_json", params, { traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", runId: "run_id", commandId: "command_id", projectionStatus: "projection_status", projectionHealth: "projection_health" }) : []]);
const facts = Object.fromEntries(entries);
return withPersistence({ facts, count: Object.values(facts).reduce((sum, rows) => sum + rows.length, 0) }, this.summary());
}