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

fix(workbench): replay SSE outbox through runtime service (#2070)

Browse Source
This commit is contained in:
Lyon
2026-06-24 23:21:02 +08:00
committed by GitHub
parent 51f2f2cd47
commit 2651a6ee55
2 changed files with 11 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/server-workbench-http.test.ts
+2 -2
View File
@@ -1585,7 +1585,7 @@ test("workbench realtime stream replays durable outbox after requested seq", asy
session: { sessionStatus: "running", lastTraceId: traceId }
};
const outboxQueries = [];
const runtimeStore = {
const workbenchRuntime = {
async readWorkbenchProjectionOutbox(params = {}) {
outboxQueries.push({ ...params });
return [
@@ -1601,7 +1601,7 @@ test("workbench realtime stream replays durable outbox after requested seq", asy
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, runtimeStore, env: { HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000" } });
const server = createCloudApiServer({ accessController, workbenchRuntime, env: { HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000" } });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
internal/cloud/server-workbench-http.ts
+9 -5
View File
@@ -112,11 +112,12 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
const requestedSessionId = safeSessionId(url.searchParams.get("sessionId") ?? url.searchParams.get("includeSessionId"));
const requestedTraceId = safeTraceId(url.searchParams.get("traceId"));
const heartbeatMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_HEARTBEAT_MS, DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS);
const realtimeOutboxReader = workbenchRuntimeOutboxReader(request, options);
attachWorkbenchRealtimeOtelContext(request, {
sessionId: requestedSessionId,
traceId: requestedTraceId,
heartbeatMs,
outboxMode: typeof (options.runtimeStore ?? null)?.readWorkbenchProjectionOutbox === "function"
outboxMode: typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function"
});
const perf = options.backendPerformance;
const auth = perf ? await perf.measure("workbench_auth", () => authenticateWorkbenchRead(request, response, options)) : await authenticateWorkbenchRead(request, response, options);
@@ -217,24 +218,23 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
traceId: activeTraceId,
threadId: streamThreadId,
heartbeatMs,
outboxMode: typeof (options.runtimeStore ?? null)?.readWorkbenchProjectionOutbox === "function"
outboxMode: typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function"
});
emitWorkbenchRealtimeAcceptedOtelSpan(request, options.env);
if (activeTraceId && requestedAfterSeq <= 0) {
await (perf ? perf.measure("workbench_initial_trace", () => writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" })) : writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" }));
}
const runtimeStore = options.runtimeStore ?? null;
const outboxPollMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_OUTBOX_POLL_MS, 500);
let outboxCursor = requestedAfterSeq;
const outboxTraceSnapshots = new Set(activeTraceId ? [activeTraceId] : []);
if (typeof runtimeStore?.readWorkbenchProjectionOutbox === "function" && (streamSessionId || activeTraceId)) {
if (typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function" && (streamSessionId || activeTraceId)) {
const pollOutbox = async () => {
if (closed || response.destroyed) return;
try {
const query = { afterSeq: outboxCursor, limit: 50 };
if (streamSessionId) query.sessionId = streamSessionId;
else query.traceId = activeTraceId;
const rows = await runtimeStore.readWorkbenchProjectionOutbox(query);
const rows = await realtimeOutboxReader.readWorkbenchProjectionOutbox(query);
for (const row of rows) {
if (closed || response.destroyed) break;
outboxCursor = row.outboxSeq;
@@ -737,6 +737,10 @@ function workbenchRuntimeFactsReadModel(request, options, actor) {
};
}
function workbenchRuntimeOutboxReader(request, options) {
return options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger, traceparent: request?.hwlabHttpRequestContext?.traceparent });
}
async function handleWorkbenchSessionListBunLegacy(response, url, options, actor) {
if (url.searchParams.has("projectId") || url.searchParams.has("workspaceId")) return sendJson(response, 400, workbenchError("workbench_authority_removed", "Workbench session list is keyed by sessionId only."));
const perf = options.backendPerformance;