diff --git a/internal/cloud/server-workbench-realtime-http.test.ts b/internal/cloud/server-workbench-realtime-http.test.ts index c22d0040..0e422577 100644 --- a/internal/cloud/server-workbench-realtime-http.test.ts +++ b/internal/cloud/server-workbench-realtime-http.test.ts @@ -92,6 +92,116 @@ test("workbench realtime initial connection emits current snapshot without repla } }); +test("synchronous projection notification is buffered until the unique initial snapshot completes", async () => { + const sessionId = "ses_realtime_initial_barrier"; + const traceId = "trc_realtime_initial_barrier"; + const calls = []; + let releaseInitial; + let unsubscribeCount = 0; + const initialGate = new Promise((resolve) => { releaseInitial = resolve; }); + const kafkaEventBridge = { + subscribeProjectionCommits(listener) { + listener({ sessionId, traceId }); + return () => { unsubscribeCount += 1; }; + }, + async stop() {} + }; + const runtime = { + async readAtomicWorkbenchProjectionSync(params = {}) { + calls.push({ ...params }); + if (calls.length === 1) { + await initialGate; + return { + facts: { + sessions: [{ sessionId, lastTraceId: traceId }], + messages: [{ messageId: "msg_initial_barrier", sessionId, traceId, role: "agent", text: "snapshot before delta", projectedSeq: 10 }], + turns: [] + }, + events: [], + cutoffOutboxSeq: 10, + cursorOutboxSeq: 10, + hasMore: false + }; + } + const event = { id: "wte_initial_barrier_delta", sessionId, traceId, projectedSeq: 11, message: "delta after snapshot" }; + return { + facts: {}, + events: [{ outboxSeq: 11, entityFamily: "traceEvents", entityId: event.id, projectedSeq: 11, projectionRevision: 11, sessionId, traceId, commitType: "event", payload: { family: "traceEvents", fact: event } }], + cutoffOutboxSeq: 11, + cursorOutboxSeq: 11, + hasMore: false + }; + } + }; + const server = createCloudApiServer({ + accessController: realtimeAccessController(), + workbenchRuntime: runtime, + kafkaEventBridge, + env: { HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000", HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE: "100" } + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + try { + const eventsPromise = getSseEvents(server.address().port, `/v1/workbench/events?sessionId=${sessionId}`, 3); + await waitForCondition(() => calls.length === 1); + assert.equal(calls[0].snapshotOnly, true); + assert.equal(calls[0].deltaOnly, false); + assert.equal(calls.length, 1); + releaseInitial(); + const events = await eventsPromise; + assert.deepEqual(events.map((event) => event.event), ["workbench.connected", "workbench.message.snapshot", "workbench.trace.event"]); + assert.equal(events[1].data.message.text, "snapshot before delta"); + assert.equal(events[2].data.event.message, "delta after snapshot"); + assert.equal(calls.length, 2); + assert.equal(calls[1].snapshotOnly, false); + assert.equal(calls[1].deltaOnly, true); + assert.equal(calls[1].afterOutboxSeq, 10); + } finally { + releaseInitial?.(); + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } + assert.equal(unsubscribeCount, 1); +}); + +test("initial atomic snapshot failure emits an error and closes SSE for reconnect", async () => { + const sessionId = "ses_realtime_initial_failure"; + let unsubscribeCount = 0; + const server = createCloudApiServer({ + accessController: realtimeAccessController(), + workbenchRuntime: { + async readAtomicWorkbenchProjectionSync() { + const error = new Error("initial snapshot failed"); + error.code = "workbench_initial_snapshot_failed"; + throw error; + } + }, + kafkaEventBridge: { + subscribeProjectionCommits() { return () => { unsubscribeCount += 1; }; }, + async stop() {} + }, + env: { HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000", HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE: "100" } + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const controller = new AbortController(); + let timeout = null; + try { + const response = await fetch(`http://127.0.0.1:${server.address().port}/v1/workbench/events?sessionId=${sessionId}`, { signal: controller.signal }); + assert.equal(response.status, 200); + const body = await Promise.race([ + response.text(), + new Promise((_, reject) => { timeout = setTimeout(() => reject(new Error("initial snapshot failure left SSE open")), 500); }) + ]); + const events = body.trim().split("\n\n").filter(Boolean).map(parseSseBlock); + assert.deepEqual(events.map((event) => event.event), ["workbench.connected", "workbench.error"]); + assert.equal(events[1].data.error.code, "workbench_initial_snapshot_failed"); + assert.equal(body.includes("workbench.heartbeat"), false); + await waitForCondition(() => unsubscribeCount === 1); + } finally { + if (timeout) clearTimeout(timeout); + controller.abort(); + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + test("projector commit during initial snapshot wakes one coalesced delta scan without idle polling", async () => { const sessionId = "ses_realtime_commit_wakeup"; const traceId = "trc_realtime_commit_wakeup"; diff --git a/internal/cloud/server-workbench-realtime-http.ts b/internal/cloud/server-workbench-realtime-http.ts index fce8e3d8..65fd37f6 100644 --- a/internal/cloud/server-workbench-realtime-http.ts +++ b/internal/cloud/server-workbench-realtime-http.ts @@ -222,6 +222,8 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option let outboxCursor = requestedAfterSeq; let scanRunning = false; let scanRequested = false; + let initialScanState = "pending"; + let initialScanWakeRequested = false; let realtimeConnection = null; const isActive = () => !closed && !response.destroyed && !response.writableEnded; @@ -364,28 +366,37 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option hasMore = snapshot?.hasMore === true; } }; - const scanProjectionOutbox = async ({ initial = false } = {}) => { + const emitProjectionScanError = async (error) => { + await writeEvent("workbench.error", { type: "error", realtimeSource: "projection-outbox", sessionId: streamSessionId, traceId: activeTraceId, error: { code: error?.code ?? "workbench_outbox_scan_failed", message: error?.message ?? "Workbench projection outbox scan failed.", valuesRedacted: true } }); + }; + const scanProjectionOutbox = async () => { if (!isActive()) return; + if (initialScanState !== "complete") { + if (initialScanState === "pending") initialScanWakeRequested = true; + return; + } if (scanRunning) { scanRequested = true; return; } scanRunning = true; - let initialPass = initial; try { do { scanRequested = false; - await runProjectionScan({ initial: initialPass }); - initialPass = false; + await runProjectionScan({ initial: false }); } while (scanRequested && isActive()); } catch (error) { - await writeEvent("workbench.error", { type: "error", realtimeSource: "projection-outbox", sessionId: streamSessionId, traceId: activeTraceId, error: { code: error?.code ?? "workbench_outbox_scan_failed", message: error?.message ?? "Workbench projection outbox scan failed.", valuesRedacted: true } }); + await emitProjectionScanError(error); } finally { scanRunning = false; } }; const requestProjectionScan = () => { if (!isActive()) return; + if (initialScanState !== "complete") { + if (initialScanState === "pending") initialScanWakeRequested = true; + return; + } scanRequested = true; void scanProjectionOutbox(); }; @@ -394,7 +405,20 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option }); if (typeof unsubscribeProjectionCommits === "function") cleanup.push(unsubscribeProjectionCommits); - await scanProjectionOutbox({ initial: true }); + try { + await runProjectionScan({ initial: true }); + initialScanState = "complete"; + } catch (error) { + initialScanState = "failed"; + await emitProjectionScanError(error); + if (!response.writableEnded) response.end(); + return; + } + if (initialScanState === "complete" && initialScanWakeRequested && isActive()) { + initialScanWakeRequested = false; + scanRequested = true; + await scanProjectionOutbox(); + } if (!isActive()) return; const heartbeat = setInterval(() => { diff --git a/scripts/src/dev-runtime-migration.mjs b/scripts/src/dev-runtime-migration.mjs index 2c75b53f..ffda0968 100644 --- a/scripts/src/dev-runtime-migration.mjs +++ b/scripts/src/dev-runtime-migration.mjs @@ -29,7 +29,7 @@ const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../ const migrationPaths = CLOUD_CORE_MIGRATIONS.map((migration) => migration.path); const latestMigrationPath = migrationPaths.at(-1); const defaultReportPath = tempReportPath("dev-runtime-migration-report.json"); -const issue = "pikasTech/HWLAB#164"; +const issue = "pikasTech/HWLAB#2464"; export async function runDevRuntimeMigrationCli(argv = process.argv.slice(2), options = {}) { const args = parseArgs(argv); @@ -495,6 +495,12 @@ function summarizeDbEnv(env = {}) { } function buildApplyBoundary(args) { + const schemaWriteScope = migrationPaths.map((migrationPath) => + `DEV Postgres schema objects declared by ${migrationPath}` + ); + const ledgerWriteScope = CLOUD_CORE_MIGRATIONS.map((migration) => + `DEV ${CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE} ledger row ${migration.id}` + ); return { defaultMode: "check", mode: args.mode, @@ -503,10 +509,7 @@ function buildApplyBoundary(args) { requiresForLiveDbRead: ["--confirm-dev", "HWLAB_CLOUD_DB_URL from hwlab-cloud-api-dev-db/database-url"], requiresForApply: ["--apply", "--confirm-dev", "--confirmed-non-production", "HWLAB_CLOUD_DB_URL from hwlab-cloud-api-dev-db/database-url"], writeScope: args.mode === "apply" - ? [ - "DEV Postgres schema objects declared by internal/db/migrations/0001_cloud_core_skeleton.sql", - "DEV hwlab_schema_migrations ledger row 0001_cloud_core_skeleton" - ] + ? [...schemaWriteScope, ...ledgerWriteScope] : [], noWriteScope: [ "Kubernetes Secret resources", diff --git a/scripts/src/dev-runtime-migration.test.mjs b/scripts/src/dev-runtime-migration.test.mjs index 1baf9721..632a2b5e 100644 --- a/scripts/src/dev-runtime-migration.test.mjs +++ b/scripts/src/dev-runtime-migration.test.mjs @@ -9,8 +9,10 @@ import { parseArgs } from "./dev-runtime-migration.mjs"; import { + CLOUD_CORE_MIGRATIONS, CLOUD_RUNTIME_DURABLE_MIGRATION_ID, CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION, + CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE, CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS } from "../../internal/db/schema.ts"; @@ -34,6 +36,8 @@ test("source check is non-secret and does not attempt live DB access", async () assert.equal(report.gates.schema.status, "ready"); assert.equal(report.gates.migration.status, "ready"); assert.equal(report.gates.readiness.status, "not_checked"); + assert.equal(report.issue, "pikasTech/HWLAB#2464"); + assert.deepEqual(report.applyBoundary.writeScope, []); assert.equal(report.safety.sourceStoresSecretValues, false); assert.equal(report.safety.printsSecretValues, false); assert.equal(JSON.stringify(report).includes("secret-password"), false); @@ -308,6 +312,13 @@ test("apply mode records ledger and verifies durable runtime readiness", async ( assert.equal(report.gates.readiness.status, "ready"); assert.equal(report.runtime.migration.appliedMigrationId, CLOUD_RUNTIME_DURABLE_MIGRATION_ID); assert.equal(report.runtime.migration.appliedSchemaVersion, CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION); + assert.deepEqual(report.applyBoundary.writeScope, [ + ...CLOUD_CORE_MIGRATIONS.map((migration) => `DEV Postgres schema objects declared by ${migration.path}`), + ...CLOUD_CORE_MIGRATIONS.map((migration) => `DEV ${CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE} ledger row ${migration.id}`) + ]); + assert.equal(report.applyBoundary.writeScope.length, CLOUD_CORE_MIGRATIONS.length * 2); + assert.ok(report.applyBoundary.writeScope.some((entry) => entry.includes("0008_workbench_kafka_realtime.sql"))); + assert.ok(report.applyBoundary.writeScope.some((entry) => entry.endsWith(`ledger row ${CLOUD_RUNTIME_DURABLE_MIGRATION_ID}`))); assert.ok(queryClient.calls.some((call) => call.sql === "BEGIN")); assert.ok(queryClient.calls.some((call) => call.sql.includes("INSERT INTO hwlab_schema_migrations"))); assert.ok(queryClient.calls.some((call) => call.sql === "COMMIT"));