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

fix: close realtime initialization visibility gaps

Browse Source
This commit is contained in:
root
2026-07-10 06:59:16 +02:00
parent efe4b62a0b
commit d5230cb923
4 changed files with 159 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/server-workbench-realtime-http.test.ts
+110
View File
@@ -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";
internal/cloud/server-workbench-realtime-http.ts
+30 -6
View File
@@ -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(() => {
scripts/src/dev-runtime-migration.mjs
+8 -5
View File
@@ -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",
scripts/src/dev-runtime-migration.test.mjs
+11
View File
@@ -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"));