mirror of
https://github.com/whyour/qinglong.git
synced 2026-09-20 16:07:11 +08:00
832 lines
27 KiB
TypeScript
832 lines
27 KiB
TypeScript
import type {
|
|
RunAttemptRecord,
|
|
RunAttemptStatus,
|
|
RunEventRecord,
|
|
RunRecord,
|
|
} from '../../domain/run';
|
|
import {
|
|
EXECUTION_ORIGINS,
|
|
RUN_ATTEMPT_STATUSES,
|
|
RUN_CANCELLATION_REASONS,
|
|
RUN_EVENT_ACTOR_TYPES,
|
|
RUN_STATUSES,
|
|
} from '../../domain/run';
|
|
import {
|
|
assertRunRetryPolicyRecord,
|
|
RUN_RETRY_SAFETIES,
|
|
type RunRetryPolicyRecord,
|
|
} from '../../domain/runRetryPolicy';
|
|
import {
|
|
DuplicateIdempotencyKeyError,
|
|
DuplicateRunAttemptError,
|
|
DuplicateRunEventError,
|
|
RunEventPayloadTooLargeError,
|
|
RunRepositoryBusyError,
|
|
RunRepositoryConstraintError,
|
|
RunRepositoryError,
|
|
RunRepositoryOperationError,
|
|
} from '../../domain/repositoryErrors';
|
|
import type {
|
|
RunRepository,
|
|
RunRepositoryReader,
|
|
RunRepositoryTransaction,
|
|
} from '../../ports/runRepository';
|
|
import type {
|
|
PostgresClient as PostgresRunClient,
|
|
PostgresPool as PostgresRunPool,
|
|
PostgresQueryable as PostgresRunQueryable,
|
|
PostgresQueryResult as PostgresRunQueryResult,
|
|
} from '@qinglong/runtime-core';
|
|
|
|
export type {
|
|
PostgresClient as PostgresRunClient,
|
|
PostgresPool as PostgresRunPool,
|
|
PostgresQueryable as PostgresRunQueryable,
|
|
PostgresQueryResult as PostgresRunQueryResult,
|
|
} from '@qinglong/runtime-core';
|
|
import {
|
|
MAX_CANCELLATION_RECOVERY_PAGE_SIZE,
|
|
MAX_RUN_EVENT_PAGE_SIZE,
|
|
MAX_RUN_EVENT_PAYLOAD_BYTES,
|
|
} from '../../ports/runRepository';
|
|
|
|
interface ColumnDefinition {
|
|
readonly column: string;
|
|
readonly property: string;
|
|
}
|
|
|
|
type QueryRow = Record<string, unknown>;
|
|
|
|
const POSTGRES_RUNTIME_STATEMENT_TIMEOUT_MS = 5_000;
|
|
const POSTGRES_RUNTIME_LOCK_TIMEOUT_MS = 1_000;
|
|
const POSTGRES_RUNTIME_IDLE_TRANSACTION_TIMEOUT_MS = 10_000;
|
|
|
|
const RUN_COLUMNS: readonly ColumnDefinition[] = Object.freeze([
|
|
{ column: 'id', property: 'id' },
|
|
{ column: 'project_id', property: 'projectId' },
|
|
{ column: 'task_id', property: 'taskId' },
|
|
{ column: 'task_revision', property: 'taskRevision' },
|
|
{ column: 'task_name', property: 'taskName' },
|
|
{ column: 'task_snapshot_ref', property: 'taskSnapshotRef' },
|
|
{ column: 'legacy_cron_id', property: 'legacyCronId' },
|
|
{ column: 'parent_run_id', property: 'parentRunId' },
|
|
{ column: 'retry_of_run_id', property: 'retryOfRunId' },
|
|
{ column: 'trigger_id', property: 'triggerId' },
|
|
{ column: 'trigger_type', property: 'triggerType' },
|
|
{ column: 'execution_origin', property: 'executionOrigin' },
|
|
{ column: 'execution_owner', property: 'executionOwner' },
|
|
{ column: 'triggered_by', property: 'triggeredBy' },
|
|
{ column: 'request_id', property: 'requestId' },
|
|
{ column: 'scheduled_for_ms', property: 'scheduledForMs' },
|
|
{ column: 'status', property: 'status' },
|
|
{ column: 'version', property: 'version' },
|
|
{ column: 'event_sequence', property: 'eventSequence' },
|
|
{ column: 'priority', property: 'priority' },
|
|
{ column: 'idempotency_key', property: 'idempotencyKey' },
|
|
{ column: 'input_ref', property: 'inputRef' },
|
|
{ column: 'output_ref', property: 'outputRef' },
|
|
{ column: 'created_at_ms', property: 'createdAtMs' },
|
|
{ column: 'queued_at_ms', property: 'queuedAtMs' },
|
|
{ column: 'started_at_ms', property: 'startedAtMs' },
|
|
{ column: 'finished_at_ms', property: 'finishedAtMs' },
|
|
{ column: 'cancel_requested_at_ms', property: 'cancelRequestedAtMs' },
|
|
{ column: 'cancel_reason', property: 'cancelReason' },
|
|
{ column: 'error_code', property: 'errorCode' },
|
|
{ column: 'error_summary', property: 'errorSummary' },
|
|
]);
|
|
|
|
const ATTEMPT_COLUMNS: readonly ColumnDefinition[] = Object.freeze([
|
|
{ column: 'id', property: 'id' },
|
|
{ column: 'run_id', property: 'runId' },
|
|
{ column: 'step_run_id', property: 'stepRunId' },
|
|
{ column: 'attempt', property: 'attempt' },
|
|
{ column: 'status', property: 'status' },
|
|
{ column: 'executor_type', property: 'executorType' },
|
|
{ column: 'worker_id', property: 'workerId' },
|
|
{ column: 'executor_handle', property: 'executorHandle' },
|
|
{ column: 'pid', property: 'pid' },
|
|
{ column: 'log_artifact_id', property: 'logArtifactId' },
|
|
{ column: 'lease_token', property: 'leaseToken' },
|
|
{ column: 'lease_expires_at_ms', property: 'leaseExpiresAtMs' },
|
|
{ column: 'deadline_at_ms', property: 'deadlineAtMs' },
|
|
{ column: 'callback_token_hash', property: 'callbackTokenHash' },
|
|
{ column: 'callback_sequence', property: 'callbackSequence' },
|
|
{ column: 'created_at_ms', property: 'createdAtMs' },
|
|
{ column: 'started_at_ms', property: 'startedAtMs' },
|
|
{ column: 'finished_at_ms', property: 'finishedAtMs' },
|
|
{ column: 'exit_code', property: 'exitCode' },
|
|
{ column: 'error_code', property: 'errorCode' },
|
|
{ column: 'error_summary', property: 'errorSummary' },
|
|
]);
|
|
|
|
const EVENT_COLUMNS: readonly ColumnDefinition[] = Object.freeze([
|
|
{ column: 'id', property: 'id' },
|
|
{ column: 'run_id', property: 'runId' },
|
|
{ column: 'sequence', property: 'sequence' },
|
|
{ column: 'type', property: 'type' },
|
|
{ column: 'dedupe_key', property: 'dedupeKey' },
|
|
{ column: 'actor_type', property: 'actorType' },
|
|
{ column: 'actor_id', property: 'actorId' },
|
|
{ column: 'attempt_id', property: 'attemptId' },
|
|
{ column: 'step_run_id', property: 'stepRunId' },
|
|
{ column: 'payload', property: 'payload' },
|
|
{ column: 'created_at_ms', property: 'createdAtMs' },
|
|
]);
|
|
|
|
const RETRY_POLICY_COLUMNS: readonly ColumnDefinition[] = Object.freeze([
|
|
{ column: 'run_id', property: 'runId' },
|
|
{ column: 'max_attempts', property: 'maxAttempts' },
|
|
{ column: 'retry_on_lost', property: 'retryOnLost' },
|
|
{ column: 'safety', property: 'safety' },
|
|
{ column: 'backoff_base_ms', property: 'backoffBaseMs' },
|
|
{ column: 'backoff_max_ms', property: 'backoffMaxMs' },
|
|
{ column: 'next_attempt_at_ms', property: 'nextAttemptAtMs' },
|
|
{ column: 'version', property: 'version' },
|
|
{ column: 'created_at_ms', property: 'createdAtMs' },
|
|
{ column: 'updated_at_ms', property: 'updatedAtMs' },
|
|
]);
|
|
|
|
const TERMINAL_RUN_STATUSES = Object.freeze([
|
|
'succeeded',
|
|
'failed',
|
|
'cancelled',
|
|
'timed_out',
|
|
]);
|
|
|
|
const BUSY_SQL_STATES = new Set([
|
|
'08000',
|
|
'08001',
|
|
'08003',
|
|
'08004',
|
|
'08006',
|
|
'08007',
|
|
'08P01',
|
|
'40001',
|
|
'40P01',
|
|
'55P03',
|
|
'57014',
|
|
'57P01',
|
|
'57P02',
|
|
'57P03',
|
|
]);
|
|
|
|
const RUN_IDEMPOTENCY_CONSTRAINT = 'ql3_runs_project_idempotency_uidx';
|
|
const ATTEMPT_NUMBER_CONSTRAINT = 'ql3_run_attempts_run_attempt_uidx';
|
|
const EVENT_SEQUENCE_CONSTRAINT = 'ql3_run_events_run_sequence_uidx';
|
|
const EVENT_DEDUPE_CONSTRAINT = 'ql3_run_events_run_dedupe_uidx';
|
|
|
|
function quoted(identifier: string): string {
|
|
return `"${identifier}"`;
|
|
}
|
|
|
|
function selectColumns(columns: readonly ColumnDefinition[]): string {
|
|
return columns
|
|
.map(({ column, property }) => `${quoted(column)} AS ${quoted(property)}`)
|
|
.join(', ');
|
|
}
|
|
|
|
function insertSql(
|
|
tableName: string,
|
|
columns: readonly ColumnDefinition[],
|
|
): string {
|
|
return `INSERT INTO "ql3".${quoted(tableName)} (${columns
|
|
.map(({ column }) => quoted(column))
|
|
.join(', ')}) VALUES (${columns
|
|
.map((_, index) => `$${index + 1}`)
|
|
.join(', ')})`;
|
|
}
|
|
|
|
function updateSql(
|
|
tableName: string,
|
|
columns: readonly ColumnDefinition[],
|
|
predicate: string,
|
|
): string {
|
|
const mutableColumns = columns.slice(1);
|
|
return `UPDATE "ql3".${quoted(tableName)} SET ${mutableColumns
|
|
.map(({ column }, index) => `${quoted(column)} = $${index + 2}`)
|
|
.join(', ')} WHERE ${predicate} RETURNING ${quoted(columns[0].column)}`;
|
|
}
|
|
|
|
function writeValues(
|
|
record: object,
|
|
columns: readonly ColumnDefinition[],
|
|
): unknown[] {
|
|
const values = record as Record<string, unknown>;
|
|
return columns.map(({ property }) => values[property] ?? null);
|
|
}
|
|
|
|
function requiredString(row: QueryRow, property: string): string {
|
|
const value = row[property];
|
|
if (typeof value !== 'string' || value.length === 0) {
|
|
throw new RunRepositoryConstraintError(
|
|
`PostgreSQL Run row has an invalid ${property}`,
|
|
);
|
|
}
|
|
return value;
|
|
}
|
|
|
|
function optionalString(row: QueryRow, property: string): string | undefined {
|
|
const value = row[property];
|
|
if (value === null || value === undefined) return undefined;
|
|
if (typeof value !== 'string') {
|
|
throw new RunRepositoryConstraintError(
|
|
`PostgreSQL Run row has an invalid ${property}`,
|
|
);
|
|
}
|
|
return value;
|
|
}
|
|
|
|
function requiredInteger(row: QueryRow, property: string): number {
|
|
const value = row[property];
|
|
if (typeof value === 'number' && Number.isSafeInteger(value)) return value;
|
|
if (typeof value === 'string' && /^-?(0|[1-9]\d*)$/.test(value)) {
|
|
const parsed = Number(value);
|
|
if (Number.isSafeInteger(parsed)) return parsed;
|
|
}
|
|
throw new RunRepositoryConstraintError(
|
|
`PostgreSQL Run row has an invalid ${property}`,
|
|
);
|
|
}
|
|
|
|
function optionalInteger(row: QueryRow, property: string): number | undefined {
|
|
if (row[property] === null || row[property] === undefined) return undefined;
|
|
return requiredInteger(row, property);
|
|
}
|
|
|
|
function requiredBoolean(row: QueryRow, property: string): boolean {
|
|
const value = row[property];
|
|
if (typeof value !== 'boolean') {
|
|
throw new RunRepositoryConstraintError(
|
|
`PostgreSQL Run row has an invalid ${property}`,
|
|
);
|
|
}
|
|
return value;
|
|
}
|
|
|
|
function requiredEnum<T extends string>(
|
|
row: QueryRow,
|
|
property: string,
|
|
allowed: readonly T[],
|
|
): T {
|
|
const value = requiredString(row, property);
|
|
if (!allowed.includes(value as T)) {
|
|
throw new RunRepositoryConstraintError(
|
|
`PostgreSQL Run row has an unsupported ${property}`,
|
|
);
|
|
}
|
|
return value as T;
|
|
}
|
|
|
|
function assignOptional<T extends object, K extends keyof T>(
|
|
record: T,
|
|
key: K,
|
|
value: T[K] | undefined,
|
|
): void {
|
|
if (value !== undefined) record[key] = value;
|
|
}
|
|
|
|
function rowToRun(row: QueryRow): RunRecord {
|
|
const run: RunRecord = {
|
|
id: requiredString(row, 'id'),
|
|
projectId: requiredString(row, 'projectId'),
|
|
taskId: requiredString(row, 'taskId'),
|
|
taskRevision: requiredString(row, 'taskRevision'),
|
|
triggerType: requiredString(row, 'triggerType'),
|
|
executionOrigin: requiredEnum(row, 'executionOrigin', EXECUTION_ORIGINS),
|
|
executionOwner: requiredEnum(row, 'executionOwner', [
|
|
'legacy',
|
|
'runtime',
|
|
] as const),
|
|
status: requiredEnum(row, 'status', RUN_STATUSES),
|
|
version: requiredInteger(row, 'version'),
|
|
eventSequence: requiredInteger(row, 'eventSequence'),
|
|
priority: requiredInteger(row, 'priority'),
|
|
createdAtMs: requiredInteger(row, 'createdAtMs'),
|
|
};
|
|
assignOptional(run, 'taskName', optionalString(row, 'taskName'));
|
|
assignOptional(
|
|
run,
|
|
'taskSnapshotRef',
|
|
optionalString(row, 'taskSnapshotRef'),
|
|
);
|
|
assignOptional(run, 'legacyCronId', optionalInteger(row, 'legacyCronId'));
|
|
assignOptional(run, 'parentRunId', optionalString(row, 'parentRunId'));
|
|
assignOptional(run, 'retryOfRunId', optionalString(row, 'retryOfRunId'));
|
|
assignOptional(run, 'triggerId', optionalString(row, 'triggerId'));
|
|
assignOptional(run, 'triggeredBy', optionalString(row, 'triggeredBy'));
|
|
assignOptional(run, 'requestId', optionalString(row, 'requestId'));
|
|
assignOptional(run, 'scheduledForMs', optionalInteger(row, 'scheduledForMs'));
|
|
assignOptional(run, 'idempotencyKey', optionalString(row, 'idempotencyKey'));
|
|
assignOptional(run, 'inputRef', optionalString(row, 'inputRef'));
|
|
assignOptional(run, 'outputRef', optionalString(row, 'outputRef'));
|
|
assignOptional(run, 'queuedAtMs', optionalInteger(row, 'queuedAtMs'));
|
|
assignOptional(run, 'startedAtMs', optionalInteger(row, 'startedAtMs'));
|
|
assignOptional(run, 'finishedAtMs', optionalInteger(row, 'finishedAtMs'));
|
|
assignOptional(
|
|
run,
|
|
'cancelRequestedAtMs',
|
|
optionalInteger(row, 'cancelRequestedAtMs'),
|
|
);
|
|
if (row.cancelReason !== null && row.cancelReason !== undefined) {
|
|
run.cancelReason = requiredEnum(
|
|
row,
|
|
'cancelReason',
|
|
RUN_CANCELLATION_REASONS,
|
|
);
|
|
}
|
|
assignOptional(run, 'errorCode', optionalString(row, 'errorCode'));
|
|
assignOptional(run, 'errorSummary', optionalString(row, 'errorSummary'));
|
|
return run;
|
|
}
|
|
|
|
function rowToAttempt(row: QueryRow): RunAttemptRecord {
|
|
const attempt: RunAttemptRecord = {
|
|
id: requiredString(row, 'id'),
|
|
runId: requiredString(row, 'runId'),
|
|
attempt: requiredInteger(row, 'attempt'),
|
|
status: requiredEnum(row, 'status', RUN_ATTEMPT_STATUSES),
|
|
executorType: requiredString(row, 'executorType'),
|
|
callbackSequence: requiredInteger(row, 'callbackSequence'),
|
|
createdAtMs: requiredInteger(row, 'createdAtMs'),
|
|
};
|
|
assignOptional(attempt, 'stepRunId', optionalString(row, 'stepRunId'));
|
|
assignOptional(attempt, 'workerId', optionalString(row, 'workerId'));
|
|
assignOptional(
|
|
attempt,
|
|
'executorHandle',
|
|
optionalString(row, 'executorHandle'),
|
|
);
|
|
assignOptional(attempt, 'pid', optionalInteger(row, 'pid'));
|
|
assignOptional(
|
|
attempt,
|
|
'logArtifactId',
|
|
optionalString(row, 'logArtifactId'),
|
|
);
|
|
assignOptional(attempt, 'leaseToken', optionalString(row, 'leaseToken'));
|
|
assignOptional(
|
|
attempt,
|
|
'leaseExpiresAtMs',
|
|
optionalInteger(row, 'leaseExpiresAtMs'),
|
|
);
|
|
assignOptional(attempt, 'deadlineAtMs', optionalInteger(row, 'deadlineAtMs'));
|
|
assignOptional(
|
|
attempt,
|
|
'callbackTokenHash',
|
|
optionalString(row, 'callbackTokenHash'),
|
|
);
|
|
assignOptional(attempt, 'startedAtMs', optionalInteger(row, 'startedAtMs'));
|
|
assignOptional(attempt, 'finishedAtMs', optionalInteger(row, 'finishedAtMs'));
|
|
assignOptional(attempt, 'exitCode', optionalInteger(row, 'exitCode'));
|
|
assignOptional(attempt, 'errorCode', optionalString(row, 'errorCode'));
|
|
assignOptional(attempt, 'errorSummary', optionalString(row, 'errorSummary'));
|
|
return attempt;
|
|
}
|
|
|
|
function normalizePayload(payload: unknown): Readonly<Record<string, unknown>> {
|
|
let value = payload;
|
|
if (typeof value === 'string') {
|
|
try {
|
|
value = JSON.parse(value);
|
|
} catch (error) {
|
|
throw new RunRepositoryConstraintError(
|
|
'PostgreSQL RunEvent payload is invalid JSON',
|
|
error,
|
|
);
|
|
}
|
|
}
|
|
if (!value || typeof value !== 'object' || Array.isArray(value)) {
|
|
throw new RunRepositoryConstraintError(
|
|
'PostgreSQL RunEvent payload is not a JSON object',
|
|
);
|
|
}
|
|
return value as Readonly<Record<string, unknown>>;
|
|
}
|
|
|
|
function rowToEvent(row: QueryRow): RunEventRecord {
|
|
const event: RunEventRecord = {
|
|
id: requiredString(row, 'id'),
|
|
runId: requiredString(row, 'runId'),
|
|
sequence: requiredInteger(row, 'sequence'),
|
|
type: requiredString(row, 'type'),
|
|
actorType: requiredEnum(row, 'actorType', RUN_EVENT_ACTOR_TYPES),
|
|
payload: normalizePayload(row.payload),
|
|
createdAtMs: requiredInteger(row, 'createdAtMs'),
|
|
};
|
|
assignOptional(event, 'dedupeKey', optionalString(row, 'dedupeKey'));
|
|
assignOptional(event, 'actorId', optionalString(row, 'actorId'));
|
|
assignOptional(event, 'attemptId', optionalString(row, 'attemptId'));
|
|
assignOptional(event, 'stepRunId', optionalString(row, 'stepRunId'));
|
|
return event;
|
|
}
|
|
|
|
function rowToRetryPolicy(row: QueryRow): RunRetryPolicyRecord {
|
|
const policy: RunRetryPolicyRecord = {
|
|
runId: requiredString(row, 'runId'),
|
|
maxAttempts: requiredInteger(row, 'maxAttempts'),
|
|
retryOnLost: requiredBoolean(row, 'retryOnLost'),
|
|
safety: requiredEnum(row, 'safety', RUN_RETRY_SAFETIES),
|
|
backoffBaseMs: requiredInteger(row, 'backoffBaseMs'),
|
|
backoffMaxMs: requiredInteger(row, 'backoffMaxMs'),
|
|
version: requiredInteger(row, 'version'),
|
|
createdAtMs: requiredInteger(row, 'createdAtMs'),
|
|
updatedAtMs: requiredInteger(row, 'updatedAtMs'),
|
|
};
|
|
assignOptional(
|
|
policy,
|
|
'nextAttemptAtMs',
|
|
optionalInteger(row, 'nextAttemptAtMs'),
|
|
);
|
|
assertRunRetryPolicyRecord(policy);
|
|
return policy;
|
|
}
|
|
|
|
function sqlState(error: unknown): string | undefined {
|
|
if (!error || typeof error !== 'object') return undefined;
|
|
const value = (error as { code?: unknown }).code;
|
|
return typeof value === 'string' ? value : undefined;
|
|
}
|
|
|
|
function constraintName(error: unknown): string | undefined {
|
|
if (!error || typeof error !== 'object') return undefined;
|
|
const value = (error as { constraint?: unknown }).constraint;
|
|
return typeof value === 'string' ? value : undefined;
|
|
}
|
|
|
|
function mapPostgresError(error: unknown): RunRepositoryError {
|
|
if (error instanceof RunRepositoryError) return error;
|
|
const state = sqlState(error);
|
|
if (state && BUSY_SQL_STATES.has(state)) {
|
|
return new RunRepositoryBusyError(error);
|
|
}
|
|
if (state?.startsWith('23')) {
|
|
return new RunRepositoryConstraintError(
|
|
'PostgreSQL Run repository constraint violation',
|
|
error,
|
|
);
|
|
}
|
|
return new RunRepositoryOperationError(error);
|
|
}
|
|
|
|
function affectedOneOrNone(result: PostgresRunQueryResult): boolean {
|
|
const count = result.rowCount ?? result.rows.length;
|
|
if (count === 0) return false;
|
|
if (count === 1) return true;
|
|
throw new RunRepositoryConstraintError(
|
|
'PostgreSQL compare-and-set affected more than one row',
|
|
);
|
|
}
|
|
|
|
function assertEventPayloadSize(event: RunEventRecord): void {
|
|
let serialized: string;
|
|
try {
|
|
serialized = JSON.stringify(event.payload);
|
|
} catch (error) {
|
|
throw new RunRepositoryConstraintError(
|
|
'RunEvent payload is not JSON serializable',
|
|
error,
|
|
);
|
|
}
|
|
const bytes = Buffer.byteLength(serialized, 'utf8');
|
|
if (bytes > MAX_RUN_EVENT_PAYLOAD_BYTES) {
|
|
throw new RunEventPayloadTooLargeError(bytes, MAX_RUN_EVENT_PAYLOAD_BYTES);
|
|
}
|
|
}
|
|
|
|
async function queryMapped<TRow extends QueryRow = QueryRow>(
|
|
queryable: PostgresRunQueryable,
|
|
text: string,
|
|
values?: readonly unknown[],
|
|
): Promise<PostgresRunQueryResult<TRow>> {
|
|
try {
|
|
return await queryable.query<TRow>(text, values);
|
|
} catch (error) {
|
|
throw mapPostgresError(error);
|
|
}
|
|
}
|
|
|
|
function singleRow<TRow extends QueryRow>(
|
|
result: PostgresRunQueryResult<TRow>,
|
|
): TRow | null {
|
|
if (result.rows.length === 0) return null;
|
|
if (result.rows.length !== 1) {
|
|
throw new RunRepositoryConstraintError(
|
|
'PostgreSQL Run repository returned duplicate identity rows',
|
|
);
|
|
}
|
|
return result.rows[0];
|
|
}
|
|
|
|
const RUN_SELECT = selectColumns(RUN_COLUMNS);
|
|
const ATTEMPT_SELECT = selectColumns(ATTEMPT_COLUMNS);
|
|
const EVENT_SELECT = selectColumns(EVENT_COLUMNS);
|
|
const RETRY_POLICY_SELECT = selectColumns(RETRY_POLICY_COLUMNS);
|
|
|
|
const INSERT_RUN_SQL = insertSql('runs', RUN_COLUMNS);
|
|
const INSERT_ATTEMPT_SQL = insertSql('run_attempts', ATTEMPT_COLUMNS);
|
|
const INSERT_EVENT_SQL = insertSql('run_events', EVENT_COLUMNS);
|
|
const INSERT_RETRY_POLICY_SQL = insertSql(
|
|
'run_retry_policies',
|
|
RETRY_POLICY_COLUMNS,
|
|
);
|
|
const UPDATE_RUN_SQL = updateSql(
|
|
'runs',
|
|
RUN_COLUMNS,
|
|
`"id" = $1 AND "version" = $${RUN_COLUMNS.length + 1}`,
|
|
);
|
|
const UPDATE_ATTEMPT_SQL = updateSql(
|
|
'run_attempts',
|
|
ATTEMPT_COLUMNS,
|
|
`"id" = $1 AND "status" = $${
|
|
ATTEMPT_COLUMNS.length + 1
|
|
} AND "callback_sequence" = $${ATTEMPT_COLUMNS.length + 2}`,
|
|
);
|
|
const UPDATE_RETRY_POLICY_SQL = updateSql(
|
|
'run_retry_policies',
|
|
RETRY_POLICY_COLUMNS,
|
|
`"run_id" = $1 AND "version" = $${RETRY_POLICY_COLUMNS.length + 1}`,
|
|
);
|
|
|
|
class PostgresRunReader implements RunRepositoryReader {
|
|
constructor(protected readonly queryable: PostgresRunQueryable) {}
|
|
|
|
async findRunById(runId: string): Promise<RunRecord | null> {
|
|
const row = singleRow(
|
|
await queryMapped(
|
|
this.queryable,
|
|
`SELECT ${RUN_SELECT} FROM "ql3"."runs" WHERE "id" = $1`,
|
|
[runId],
|
|
),
|
|
);
|
|
return row ? rowToRun(row) : null;
|
|
}
|
|
|
|
async findAttemptById(attemptId: string): Promise<RunAttemptRecord | null> {
|
|
const row = singleRow(
|
|
await queryMapped(
|
|
this.queryable,
|
|
`SELECT ${ATTEMPT_SELECT} FROM "ql3"."run_attempts" WHERE "id" = $1`,
|
|
[attemptId],
|
|
),
|
|
);
|
|
return row ? rowToAttempt(row) : null;
|
|
}
|
|
|
|
async findLatestAttemptByRunId(
|
|
runId: string,
|
|
): Promise<RunAttemptRecord | null> {
|
|
const row = singleRow(
|
|
await queryMapped(
|
|
this.queryable,
|
|
`SELECT ${ATTEMPT_SELECT} FROM "ql3"."run_attempts" WHERE "run_id" = $1 ORDER BY "attempt" DESC, "id" DESC LIMIT 1`,
|
|
[runId],
|
|
),
|
|
);
|
|
return row ? rowToAttempt(row) : null;
|
|
}
|
|
|
|
async findRetryPolicyByRunId(
|
|
runId: string,
|
|
): Promise<RunRetryPolicyRecord | null> {
|
|
const row = singleRow(
|
|
await queryMapped(
|
|
this.queryable,
|
|
`SELECT ${RETRY_POLICY_SELECT} FROM "ql3"."run_retry_policies" WHERE "run_id" = $1`,
|
|
[runId],
|
|
),
|
|
);
|
|
return row ? rowToRetryPolicy(row) : null;
|
|
}
|
|
|
|
async listEvents(
|
|
runId: string,
|
|
options: { afterSequence?: number; limit?: number } = {},
|
|
): Promise<RunEventRecord[]> {
|
|
const afterSequence = options.afterSequence ?? 0;
|
|
const limit = options.limit ?? 100;
|
|
if (!Number.isInteger(afterSequence) || afterSequence < 0) {
|
|
throw new RangeError('afterSequence must be a non-negative integer');
|
|
}
|
|
if (
|
|
!Number.isInteger(limit) ||
|
|
limit < 1 ||
|
|
limit > MAX_RUN_EVENT_PAGE_SIZE
|
|
) {
|
|
throw new RangeError(
|
|
'limit must be between 1 and MAX_RUN_EVENT_PAGE_SIZE',
|
|
);
|
|
}
|
|
const result = await queryMapped(
|
|
this.queryable,
|
|
`SELECT ${EVENT_SELECT} FROM "ql3"."run_events" WHERE "run_id" = $1 AND "sequence" > $2 ORDER BY "sequence", "id" LIMIT $3`,
|
|
[runId, afterSequence, limit],
|
|
);
|
|
return result.rows.map(rowToEvent);
|
|
}
|
|
|
|
async listCancellationRequested(
|
|
options: { beforeMs?: number; limit?: number } = {},
|
|
): Promise<RunRecord[]> {
|
|
const beforeMs = options.beforeMs;
|
|
const limit = options.limit ?? 100;
|
|
if (
|
|
beforeMs !== undefined &&
|
|
(!Number.isSafeInteger(beforeMs) || beforeMs < 0)
|
|
) {
|
|
throw new RangeError('beforeMs must be a non-negative safe integer');
|
|
}
|
|
if (
|
|
!Number.isSafeInteger(limit) ||
|
|
limit < 1 ||
|
|
limit > MAX_CANCELLATION_RECOVERY_PAGE_SIZE
|
|
) {
|
|
throw new RangeError(
|
|
'limit must be between 1 and MAX_CANCELLATION_RECOVERY_PAGE_SIZE',
|
|
);
|
|
}
|
|
const result = await queryMapped(
|
|
this.queryable,
|
|
`SELECT ${RUN_SELECT} FROM "ql3"."runs" WHERE "status" <> ALL($1::text[]) AND "cancel_requested_at_ms" IS NOT NULL AND ($2::bigint IS NULL OR "cancel_requested_at_ms" <= $2) ORDER BY "cancel_requested_at_ms", "id" LIMIT $3`,
|
|
[TERMINAL_RUN_STATUSES, beforeMs ?? null, limit],
|
|
);
|
|
return result.rows.map(rowToRun);
|
|
}
|
|
}
|
|
|
|
export class PostgresRunTransaction
|
|
extends PostgresRunReader
|
|
implements RunRepositoryTransaction
|
|
{
|
|
async insertRun(run: RunRecord): Promise<void> {
|
|
try {
|
|
await this.queryable.query(INSERT_RUN_SQL, writeValues(run, RUN_COLUMNS));
|
|
} catch (error) {
|
|
if (
|
|
sqlState(error) === '23505' &&
|
|
constraintName(error) === RUN_IDEMPOTENCY_CONSTRAINT &&
|
|
run.idempotencyKey
|
|
) {
|
|
throw new DuplicateIdempotencyKeyError(
|
|
run.projectId,
|
|
run.idempotencyKey,
|
|
);
|
|
}
|
|
throw mapPostgresError(error);
|
|
}
|
|
}
|
|
|
|
async insertAttempt(attempt: RunAttemptRecord): Promise<void> {
|
|
try {
|
|
await this.queryable.query(
|
|
INSERT_ATTEMPT_SQL,
|
|
writeValues(attempt, ATTEMPT_COLUMNS),
|
|
);
|
|
} catch (error) {
|
|
if (
|
|
sqlState(error) === '23505' &&
|
|
constraintName(error) === ATTEMPT_NUMBER_CONSTRAINT
|
|
) {
|
|
throw new DuplicateRunAttemptError(attempt.runId, attempt.attempt);
|
|
}
|
|
throw mapPostgresError(error);
|
|
}
|
|
}
|
|
|
|
async insertRetryPolicy(policy: RunRetryPolicyRecord): Promise<void> {
|
|
assertRunRetryPolicyRecord(policy);
|
|
try {
|
|
await this.queryable.query(
|
|
INSERT_RETRY_POLICY_SQL,
|
|
writeValues(policy, RETRY_POLICY_COLUMNS),
|
|
);
|
|
} catch (error) {
|
|
throw mapPostgresError(error);
|
|
}
|
|
}
|
|
|
|
async compareAndSetRun(
|
|
run: RunRecord,
|
|
expectedVersion: number,
|
|
): Promise<boolean> {
|
|
if (run.version !== expectedVersion + 1) {
|
|
throw new RunRepositoryConstraintError(
|
|
'A compare-and-set Run write must increment version exactly once',
|
|
);
|
|
}
|
|
const result = await queryMapped(this.queryable, UPDATE_RUN_SQL, [
|
|
...writeValues(run, RUN_COLUMNS),
|
|
expectedVersion,
|
|
]);
|
|
return affectedOneOrNone(result);
|
|
}
|
|
|
|
async compareAndSetAttempt(
|
|
attempt: RunAttemptRecord,
|
|
expected: {
|
|
status: RunAttemptStatus;
|
|
callbackSequence: number;
|
|
},
|
|
): Promise<boolean> {
|
|
const result = await queryMapped(this.queryable, UPDATE_ATTEMPT_SQL, [
|
|
...writeValues(attempt, ATTEMPT_COLUMNS),
|
|
expected.status,
|
|
expected.callbackSequence,
|
|
]);
|
|
return affectedOneOrNone(result);
|
|
}
|
|
|
|
async compareAndSetRetryPolicy(
|
|
policy: RunRetryPolicyRecord,
|
|
expectedVersion: number,
|
|
): Promise<boolean> {
|
|
if (policy.version !== expectedVersion + 1) {
|
|
throw new RunRepositoryConstraintError(
|
|
'A compare-and-set retry policy write must increment version exactly once',
|
|
);
|
|
}
|
|
assertRunRetryPolicyRecord(policy);
|
|
const result = await queryMapped(this.queryable, UPDATE_RETRY_POLICY_SQL, [
|
|
...writeValues(policy, RETRY_POLICY_COLUMNS),
|
|
expectedVersion,
|
|
]);
|
|
return affectedOneOrNone(result);
|
|
}
|
|
|
|
async appendEvent(event: RunEventRecord): Promise<void> {
|
|
assertEventPayloadSize(event);
|
|
try {
|
|
await this.queryable.query(
|
|
INSERT_EVENT_SQL,
|
|
writeValues(event, EVENT_COLUMNS),
|
|
);
|
|
} catch (error) {
|
|
if (
|
|
sqlState(error) === '23505' &&
|
|
(constraintName(error) === EVENT_SEQUENCE_CONSTRAINT ||
|
|
constraintName(error) === EVENT_DEDUPE_CONSTRAINT)
|
|
) {
|
|
throw new DuplicateRunEventError(event.runId, event.dedupeKey);
|
|
}
|
|
throw mapPostgresError(error);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Driver-neutral PostgreSQL Run Repository. The cluster-only package owns the
|
|
* concrete pg.Pool binding; edge/standalone builds never import the driver.
|
|
*/
|
|
export class PostgresRunRepository
|
|
extends PostgresRunReader
|
|
implements RunRepository
|
|
{
|
|
constructor(private readonly pool: PostgresRunPool) {
|
|
super(pool);
|
|
}
|
|
|
|
async transaction<T>(
|
|
work: (transaction: RunRepositoryTransaction) => Promise<T>,
|
|
): Promise<T> {
|
|
let client: PostgresRunClient;
|
|
try {
|
|
client = await this.pool.connect();
|
|
} catch (error) {
|
|
throw mapPostgresError(error);
|
|
}
|
|
let began = false;
|
|
let phase: 'begin' | 'work' | 'commit' = 'begin';
|
|
try {
|
|
await client.query('BEGIN');
|
|
began = true;
|
|
await client.query('SET TRANSACTION ISOLATION LEVEL READ COMMITTED');
|
|
await client.query(`SELECT set_config('statement_timeout', $1, true)`, [
|
|
`${POSTGRES_RUNTIME_STATEMENT_TIMEOUT_MS}ms`,
|
|
]);
|
|
await client.query(`SELECT set_config('lock_timeout', $1, true)`, [
|
|
`${POSTGRES_RUNTIME_LOCK_TIMEOUT_MS}ms`,
|
|
]);
|
|
await client.query(
|
|
`SELECT set_config('idle_in_transaction_session_timeout', $1, true)`,
|
|
[`${POSTGRES_RUNTIME_IDLE_TRANSACTION_TIMEOUT_MS}ms`],
|
|
);
|
|
phase = 'work';
|
|
const result = await work(new PostgresRunTransaction(client));
|
|
phase = 'commit';
|
|
await client.query('COMMIT');
|
|
began = false;
|
|
return result;
|
|
} catch (error) {
|
|
if (began) {
|
|
try {
|
|
await client.query('ROLLBACK');
|
|
} catch {
|
|
// Preserve the work/commit failure; release discards broken clients.
|
|
}
|
|
}
|
|
if (phase === 'work') throw error;
|
|
throw mapPostgresError(error);
|
|
} finally {
|
|
client.release();
|
|
}
|
|
}
|
|
}
|