Merge pull request #1776 from pikasTech/fix/1770-postgres-timeout-boundary
fix: contain postgres projection timeouts
This commit is contained in:
@@ -248,19 +248,27 @@ export function createDurableCodeAgentTraceStore({ traceStore = defaultCodeAgent
|
|||||||
}
|
}
|
||||||
|
|
||||||
function persistTraceEvent(traceEventStore, event, meta = {}, workbenchEventWriter = null) {
|
function persistTraceEvent(traceEventStore, event, meta = {}, workbenchEventWriter = null) {
|
||||||
|
void persistTraceEventAsync(traceEventStore, event, meta, workbenchEventWriter);
|
||||||
|
}
|
||||||
|
|
||||||
|
async function persistTraceEventAsync(traceEventStore, event, meta = {}, workbenchEventWriter = null) {
|
||||||
try {
|
try {
|
||||||
const projectionContext = {
|
const projectionContext = {
|
||||||
traceId: event.traceId,
|
traceId: event.traceId,
|
||||||
agentSessionId: event.agentSessionId ?? event.sessionId ?? meta.agentSessionId ?? meta.sessionId,
|
agentSessionId: event.agentSessionId ?? event.sessionId ?? meta.agentSessionId ?? meta.sessionId,
|
||||||
sessionId: event.sessionId ?? meta.sessionId
|
sessionId: event.sessionId ?? meta.sessionId
|
||||||
};
|
};
|
||||||
Promise.resolve(traceEventStore.writeAgentTraceEvent({ event }, projectionContext)).catch((error) => {
|
try {
|
||||||
|
await traceEventStore.writeAgentTraceEvent({ event }, projectionContext);
|
||||||
|
} catch (error) {
|
||||||
console.error(`[workbench-projection] durable trace event write failed: traceId=${event.traceId} sourceEventId=${event.sourceEventId ?? "none"} error=${error?.message ?? error}`);
|
console.error(`[workbench-projection] durable trace event write failed: traceId=${event.traceId} sourceEventId=${event.sourceEventId ?? "none"} error=${error?.message ?? error}`);
|
||||||
});
|
}
|
||||||
if (typeof workbenchEventWriter === "function") {
|
if (typeof workbenchEventWriter === "function") {
|
||||||
Promise.resolve(workbenchEventWriter(event, meta)).catch((error) => {
|
try {
|
||||||
|
await workbenchEventWriter(event, meta);
|
||||||
|
} catch (error) {
|
||||||
console.error(`[workbench-projection] workbench event writer failed: traceId=${event.traceId} error=${error?.message ?? error}`);
|
console.error(`[workbench-projection] workbench event writer failed: traceId=${event.traceId} error=${error?.message ?? error}`);
|
||||||
});
|
}
|
||||||
}
|
}
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
// Trace capture must not fail the user turn, but projection write errors must be visible for diagnostics.
|
// Trace capture must not fail the user turn, but projection write errors must be visible for diagnostics.
|
||||||
|
|||||||
@@ -1775,34 +1775,61 @@ export class PostgresCloudRuntimeStore {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
recordRuntimeDbOperationFailure(label, error) {
|
||||||
|
const classified = classifyRuntimeDbError(error);
|
||||||
|
this.lastReadiness = this.blockedSummary({
|
||||||
|
...classified,
|
||||||
|
reason: `Postgres durable runtime adapter could not complete ${label}`
|
||||||
|
});
|
||||||
|
this.logger?.warn?.({
|
||||||
|
event: "postgres_runtime_operation_failed",
|
||||||
|
operation: label,
|
||||||
|
blocker: classified.blocker,
|
||||||
|
queryResult: classified.connection?.queryResult ?? "query_blocked",
|
||||||
|
errorCode: classified.connection?.errorCode ?? "UNKNOWN",
|
||||||
|
valuesRedacted: true,
|
||||||
|
endpointRedacted: true
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
async query(sql, params = []) {
|
async query(sql, params = []) {
|
||||||
const client = await this.getQueryClient();
|
try {
|
||||||
return client.query(sql, params);
|
const client = await this.getQueryClient();
|
||||||
|
return await client.query(sql, params);
|
||||||
|
} catch (error) {
|
||||||
|
this.recordRuntimeDbOperationFailure("query", error);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async withDurableTransaction(label, fn) {
|
async withDurableTransaction(label, fn) {
|
||||||
const client = await this.getQueryClient();
|
try {
|
||||||
if (typeof client.connect === "function") {
|
const client = await this.getQueryClient();
|
||||||
const tx = await client.connect();
|
if (typeof client.connect === "function") {
|
||||||
|
const tx = await client.connect();
|
||||||
|
try {
|
||||||
|
await tx.query("BEGIN");
|
||||||
|
const result = await fn(tx);
|
||||||
|
await tx.query("COMMIT");
|
||||||
|
return result;
|
||||||
|
} catch (error) {
|
||||||
|
try { await tx.query("ROLLBACK"); } catch {}
|
||||||
|
throw error;
|
||||||
|
} finally {
|
||||||
|
tx.release?.();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
await client.query("BEGIN");
|
||||||
try {
|
try {
|
||||||
await tx.query("BEGIN");
|
const result = await fn(client);
|
||||||
const result = await fn(tx);
|
await client.query("COMMIT");
|
||||||
await tx.query("COMMIT");
|
|
||||||
return result;
|
return result;
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
try { await tx.query("ROLLBACK"); } catch {}
|
try { await client.query("ROLLBACK"); } catch {}
|
||||||
throw error;
|
throw error;
|
||||||
} finally {
|
|
||||||
tx.release?.();
|
|
||||||
}
|
}
|
||||||
}
|
|
||||||
await client.query("BEGIN");
|
|
||||||
try {
|
|
||||||
const result = await fn(client);
|
|
||||||
await client.query("COMMIT");
|
|
||||||
return result;
|
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
try { await client.query("ROLLBACK"); } catch {}
|
this.recordRuntimeDbOperationFailure(label || "transaction", error);
|
||||||
throw error;
|
throw error;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user