fix(otel): prevent infinite retry loops on unicode hex escape errors (#2337)
* prevent infinite retry loop for unicode errors * structured logs for prisma events * preserve all prisma event fields * either use structured logs or stdout, never both * split runs repo tests * decrease test shards to 8
This commit is contained in:
nicktrn
@@ -12,8 +12,8 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
matrix:
|
||||
shardIndex: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
|
||||
shardTotal: [10]
|
||||
shardIndex: [1, 2, 3, 4, 5, 6, 7, 8]
|
||||
shardTotal: [8]
|
||||
env:
|
||||
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
|
||||
SHARD_INDEX: ${{ matrix.shardIndex }}
|
||||
|
||||
+138
-20
@@ -122,30 +122,89 @@ function getClient() {
|
||||
url: databaseUrl.href,
|
||||
},
|
||||
},
|
||||
// @ts-expect-error
|
||||
log: [
|
||||
// events
|
||||
{
|
||||
emit: "stdout",
|
||||
emit: "event",
|
||||
level: "error",
|
||||
},
|
||||
{
|
||||
emit: "stdout",
|
||||
emit: "event",
|
||||
level: "info",
|
||||
},
|
||||
{
|
||||
emit: "stdout",
|
||||
emit: "event",
|
||||
level: "warn",
|
||||
},
|
||||
].concat(
|
||||
process.env.VERBOSE_PRISMA_LOGS === "1"
|
||||
// stdout
|
||||
...((process.env.PRISMA_LOG_TO_STDOUT === "1"
|
||||
? [
|
||||
{ emit: "event", level: "query" },
|
||||
{ emit: "stdout", level: "query" },
|
||||
{
|
||||
emit: "stdout",
|
||||
level: "error",
|
||||
},
|
||||
{
|
||||
emit: "stdout",
|
||||
level: "info",
|
||||
},
|
||||
{
|
||||
emit: "stdout",
|
||||
level: "warn",
|
||||
},
|
||||
]
|
||||
: []
|
||||
),
|
||||
: []) satisfies Prisma.LogDefinition[]),
|
||||
// verbose
|
||||
...((process.env.VERBOSE_PRISMA_LOGS === "1"
|
||||
? [
|
||||
{
|
||||
emit: "event",
|
||||
level: "query",
|
||||
},
|
||||
{
|
||||
emit: "stdout",
|
||||
level: "query",
|
||||
},
|
||||
]
|
||||
: []) satisfies Prisma.LogDefinition[]),
|
||||
],
|
||||
});
|
||||
|
||||
// Only use structured logging if we're not already logging to stdout
|
||||
if (process.env.PRISMA_LOG_TO_STDOUT !== "1") {
|
||||
client.$on("info", (log) => {
|
||||
logger.info("PrismaClient info", {
|
||||
clientType: "writer",
|
||||
event: {
|
||||
timestamp: log.timestamp,
|
||||
message: log.message,
|
||||
target: log.target,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
client.$on("warn", (log) => {
|
||||
logger.warn("PrismaClient warn", {
|
||||
clientType: "writer",
|
||||
event: {
|
||||
timestamp: log.timestamp,
|
||||
message: log.message,
|
||||
target: log.target,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
client.$on("error", (log) => {
|
||||
logger.error("PrismaClient error", {
|
||||
clientType: "writer",
|
||||
event: {
|
||||
timestamp: log.timestamp,
|
||||
message: log.message,
|
||||
target: log.target,
|
||||
},
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
// connect eagerly
|
||||
client.$connect();
|
||||
|
||||
@@ -174,30 +233,89 @@ function getReplicaClient() {
|
||||
url: replicaUrl.href,
|
||||
},
|
||||
},
|
||||
// @ts-expect-error
|
||||
log: [
|
||||
// events
|
||||
{
|
||||
emit: "stdout",
|
||||
emit: "event",
|
||||
level: "error",
|
||||
},
|
||||
{
|
||||
emit: "stdout",
|
||||
emit: "event",
|
||||
level: "info",
|
||||
},
|
||||
{
|
||||
emit: "stdout",
|
||||
emit: "event",
|
||||
level: "warn",
|
||||
},
|
||||
].concat(
|
||||
process.env.VERBOSE_PRISMA_LOGS === "1"
|
||||
// stdout
|
||||
...((process.env.PRISMA_LOG_TO_STDOUT === "1"
|
||||
? [
|
||||
{ emit: "event", level: "query" },
|
||||
{ emit: "stdout", level: "query" },
|
||||
{
|
||||
emit: "stdout",
|
||||
level: "error",
|
||||
},
|
||||
{
|
||||
emit: "stdout",
|
||||
level: "info",
|
||||
},
|
||||
{
|
||||
emit: "stdout",
|
||||
level: "warn",
|
||||
},
|
||||
]
|
||||
: []
|
||||
),
|
||||
: []) satisfies Prisma.LogDefinition[]),
|
||||
// verbose
|
||||
...((process.env.VERBOSE_PRISMA_LOGS === "1"
|
||||
? [
|
||||
{
|
||||
emit: "event",
|
||||
level: "query",
|
||||
},
|
||||
{
|
||||
emit: "stdout",
|
||||
level: "query",
|
||||
},
|
||||
]
|
||||
: []) satisfies Prisma.LogDefinition[]),
|
||||
],
|
||||
});
|
||||
|
||||
// Only use structured logging if we're not already logging to stdout
|
||||
if (process.env.PRISMA_LOG_TO_STDOUT !== "1") {
|
||||
replicaClient.$on("info", (log) => {
|
||||
logger.info("PrismaClient info", {
|
||||
clientType: "reader",
|
||||
event: {
|
||||
timestamp: log.timestamp,
|
||||
message: log.message,
|
||||
target: log.target,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
replicaClient.$on("warn", (log) => {
|
||||
logger.warn("PrismaClient warn", {
|
||||
clientType: "reader",
|
||||
event: {
|
||||
timestamp: log.timestamp,
|
||||
message: log.message,
|
||||
target: log.target,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
replicaClient.$on("error", (log) => {
|
||||
logger.error("PrismaClient error", {
|
||||
clientType: "reader",
|
||||
event: {
|
||||
timestamp: log.timestamp,
|
||||
message: log.message,
|
||||
target: log.target,
|
||||
},
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
// connect eagerly
|
||||
replicaClient.$connect();
|
||||
|
||||
|
||||
@@ -1230,25 +1230,23 @@ export class EventRepository {
|
||||
|
||||
return events;
|
||||
} catch (error) {
|
||||
if (error instanceof Prisma.PrismaClientUnknownRequestError) {
|
||||
logger.error("Failed to insert events, most likely because of null characters", {
|
||||
error: {
|
||||
name: error.name,
|
||||
message: error.message,
|
||||
stack: error.stack,
|
||||
clientVersion: error.clientVersion,
|
||||
},
|
||||
if (isRetriablePrismaError(error)) {
|
||||
const isKnownError = error instanceof Prisma.PrismaClientKnownRequestError;
|
||||
span.setAttribute("prisma_error_type", isKnownError ? "known" : "unknown");
|
||||
|
||||
const errorDetails = getPrismaErrorDetails(error);
|
||||
if (errorDetails.code) {
|
||||
span.setAttribute("prisma_error_code", errorDetails.code);
|
||||
}
|
||||
|
||||
logger.error("Failed to insert events, will attempt bisection", {
|
||||
error: errorDetails,
|
||||
});
|
||||
|
||||
if (events.length === 1) {
|
||||
logger.debug("Attempting to insert event individually and it failed", {
|
||||
event: events[0],
|
||||
error: {
|
||||
name: error.name,
|
||||
message: error.message,
|
||||
stack: error.stack,
|
||||
clientVersion: error.clientVersion,
|
||||
},
|
||||
error: errorDetails,
|
||||
});
|
||||
|
||||
span.setAttribute("failed_event_count", 1);
|
||||
@@ -1258,12 +1256,7 @@ export class EventRepository {
|
||||
|
||||
if (depth > MAX_FLUSH_DEPTH) {
|
||||
logger.error("Failed to insert events, reached maximum depth", {
|
||||
error: {
|
||||
name: error.name,
|
||||
message: error.message,
|
||||
stack: error.stack,
|
||||
clientVersion: error.clientVersion,
|
||||
},
|
||||
error: errorDetails,
|
||||
depth,
|
||||
eventsCount: events.length,
|
||||
});
|
||||
@@ -1917,3 +1910,62 @@ export async function recordRunDebugLog(
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Extracts error details from Prisma errors in a type-safe way.
|
||||
* Only includes 'code' property for PrismaClientKnownRequestError.
|
||||
*/
|
||||
function getPrismaErrorDetails(
|
||||
error: Prisma.PrismaClientUnknownRequestError | Prisma.PrismaClientKnownRequestError
|
||||
): {
|
||||
name: string;
|
||||
message: string;
|
||||
stack: string | undefined;
|
||||
clientVersion: string;
|
||||
code?: string;
|
||||
} {
|
||||
const base = {
|
||||
name: error.name,
|
||||
message: error.message,
|
||||
stack: error.stack,
|
||||
clientVersion: error.clientVersion,
|
||||
};
|
||||
|
||||
if (error instanceof Prisma.PrismaClientKnownRequestError) {
|
||||
return { ...base, code: error.code };
|
||||
}
|
||||
|
||||
return base;
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if a PrismaClientKnownRequestError is a Unicode/hex escape error.
|
||||
*/
|
||||
function isUnicodeError(error: Prisma.PrismaClientKnownRequestError): boolean {
|
||||
return (
|
||||
error.message.includes("lone leading surrogate in hex escape") ||
|
||||
error.message.includes("unexpected end of hex escape") ||
|
||||
error.message.includes("invalid Unicode") ||
|
||||
error.message.includes("invalid escape sequence")
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Determines if a Prisma error should be retried with bisection logic.
|
||||
* Returns true for errors that might be resolved by splitting the batch.
|
||||
*/
|
||||
function isRetriablePrismaError(
|
||||
error: unknown
|
||||
): error is Prisma.PrismaClientUnknownRequestError | Prisma.PrismaClientKnownRequestError {
|
||||
if (error instanceof Prisma.PrismaClientUnknownRequestError) {
|
||||
// Always retry unknown errors with bisection
|
||||
return true;
|
||||
}
|
||||
|
||||
if (error instanceof Prisma.PrismaClientKnownRequestError) {
|
||||
// Only retry known errors if they're Unicode/hex escape related
|
||||
return isUnicodeError(error);
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,749 @@
|
||||
import { describe, expect, vi } from "vitest";
|
||||
|
||||
// Mock the db prisma client
|
||||
vi.mock("~/db.server", () => ({
|
||||
prisma: {},
|
||||
$replica: {},
|
||||
}));
|
||||
|
||||
import { containerTest } from "@internal/testcontainers";
|
||||
import { setTimeout } from "node:timers/promises";
|
||||
import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
|
||||
import { setupClickhouseReplication } from "./utils/replicationUtils";
|
||||
|
||||
vi.setConfig({ testTimeout: 60_000 });
|
||||
|
||||
describe("RunsRepository (part 1/2)", () => {
|
||||
containerTest(
|
||||
"should list runs, using clickhouse as the source",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Now we insert a row into the table
|
||||
const taskRun = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_1234",
|
||||
taskIdentifier: "my-task",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
const { runs, pagination } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0].id).toBe(taskRun.id);
|
||||
expect(pagination.nextCursor).toBe(null);
|
||||
expect(pagination.previousCursor).toBe(null);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by task identifiers",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create runs with different task identifiers
|
||||
const taskRun1 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_task1",
|
||||
taskIdentifier: "task-1",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
const taskRun2 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_task2",
|
||||
taskIdentifier: "task-2",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
const taskRun3 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_task3",
|
||||
taskIdentifier: "task-3",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1236",
|
||||
spanId: "1236",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by specific tasks
|
||||
const { runs } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
tasks: ["task-1", "task-2"],
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(2);
|
||||
expect(runs.map((r) => r.taskIdentifier).sort()).toEqual(["task-1", "task-2"]);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by task versions",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create runs with different task versions
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_v1",
|
||||
taskIdentifier: "my-task",
|
||||
taskVersion: "1.0.0",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_v2",
|
||||
taskIdentifier: "my-task",
|
||||
taskVersion: "2.0.0",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_v3",
|
||||
taskIdentifier: "my-task",
|
||||
taskVersion: "3.0.0",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1236",
|
||||
spanId: "1236",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by specific versions
|
||||
const { runs } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
versions: ["1.0.0", "3.0.0"],
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(2);
|
||||
expect(runs.map((r) => r.taskVersion).sort()).toEqual(["1.0.0", "3.0.0"]);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by status",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create runs with different statuses
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_pending",
|
||||
taskIdentifier: "my-task",
|
||||
status: "PENDING",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_executing",
|
||||
taskIdentifier: "my-task",
|
||||
status: "EXECUTING",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_completed",
|
||||
taskIdentifier: "my-task",
|
||||
status: "COMPLETED_SUCCESSFULLY",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1236",
|
||||
spanId: "1236",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by specific statuses
|
||||
const { runs } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
statuses: ["PENDING", "COMPLETED_SUCCESSFULLY"],
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(2);
|
||||
expect(runs.map((r) => r.status).sort()).toEqual(["COMPLETED_SUCCESSFULLY", "PENDING"]);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by tags",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create runs with different tags
|
||||
const taskRun1 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_urgent",
|
||||
taskIdentifier: "my-task",
|
||||
runTags: ["urgent", "production"],
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
const taskRun2 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_regular",
|
||||
taskIdentifier: "my-task",
|
||||
runTags: ["regular", "development"],
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
const taskRun3 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_urgent_dev",
|
||||
taskIdentifier: "my-task",
|
||||
runTags: ["urgent", "development"],
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1236",
|
||||
spanId: "1236",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by tags
|
||||
const { runs } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
tags: ["urgent"],
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(2);
|
||||
expect(runs.map((r) => r.friendlyId).sort()).toEqual(["run_urgent", "run_urgent_dev"]);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by scheduleId",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create runs with different schedule IDs
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_scheduled_1",
|
||||
taskIdentifier: "my-task",
|
||||
scheduleId: "schedule_1",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_scheduled_2",
|
||||
taskIdentifier: "my-task",
|
||||
scheduleId: "schedule_2",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_unscheduled",
|
||||
taskIdentifier: "my-task",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1236",
|
||||
spanId: "1236",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by schedule ID
|
||||
const { runs } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
scheduleId: "schedule_1",
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0].friendlyId).toBe("run_scheduled_1");
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by isTest flag",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create test and non-test runs
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_test",
|
||||
taskIdentifier: "my-task",
|
||||
isTest: true,
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_production",
|
||||
taskIdentifier: "my-task",
|
||||
isTest: false,
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by isTest=true
|
||||
const testRuns = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
isTest: true,
|
||||
});
|
||||
|
||||
expect(testRuns.runs).toHaveLength(1);
|
||||
expect(testRuns.runs[0].friendlyId).toBe("run_test");
|
||||
|
||||
// Test filtering by isTest=false
|
||||
const productionRuns = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
isTest: false,
|
||||
});
|
||||
|
||||
expect(productionRuns.runs).toHaveLength(1);
|
||||
expect(productionRuns.runs[0].friendlyId).toBe("run_production");
|
||||
}
|
||||
);
|
||||
});
|
||||
+2
-735
@@ -13,740 +13,7 @@ import { setupClickhouseReplication } from "./utils/replicationUtils";
|
||||
|
||||
vi.setConfig({ testTimeout: 60_000 });
|
||||
|
||||
describe("RunsRepository", () => {
|
||||
containerTest(
|
||||
"should list runs, using clickhouse as the source",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Now we insert a row into the table
|
||||
const taskRun = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_1234",
|
||||
taskIdentifier: "my-task",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
const { runs, pagination } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0].id).toBe(taskRun.id);
|
||||
expect(pagination.nextCursor).toBe(null);
|
||||
expect(pagination.previousCursor).toBe(null);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by task identifiers",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create runs with different task identifiers
|
||||
const taskRun1 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_task1",
|
||||
taskIdentifier: "task-1",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
const taskRun2 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_task2",
|
||||
taskIdentifier: "task-2",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
const taskRun3 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_task3",
|
||||
taskIdentifier: "task-3",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1236",
|
||||
spanId: "1236",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by specific tasks
|
||||
const { runs } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
tasks: ["task-1", "task-2"],
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(2);
|
||||
expect(runs.map((r) => r.taskIdentifier).sort()).toEqual(["task-1", "task-2"]);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by task versions",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create runs with different task versions
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_v1",
|
||||
taskIdentifier: "my-task",
|
||||
taskVersion: "1.0.0",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_v2",
|
||||
taskIdentifier: "my-task",
|
||||
taskVersion: "2.0.0",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_v3",
|
||||
taskIdentifier: "my-task",
|
||||
taskVersion: "3.0.0",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1236",
|
||||
spanId: "1236",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by specific versions
|
||||
const { runs } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
versions: ["1.0.0", "3.0.0"],
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(2);
|
||||
expect(runs.map((r) => r.taskVersion).sort()).toEqual(["1.0.0", "3.0.0"]);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by status",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create runs with different statuses
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_pending",
|
||||
taskIdentifier: "my-task",
|
||||
status: "PENDING",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_executing",
|
||||
taskIdentifier: "my-task",
|
||||
status: "EXECUTING",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_completed",
|
||||
taskIdentifier: "my-task",
|
||||
status: "COMPLETED_SUCCESSFULLY",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1236",
|
||||
spanId: "1236",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by specific statuses
|
||||
const { runs } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
statuses: ["PENDING", "COMPLETED_SUCCESSFULLY"],
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(2);
|
||||
expect(runs.map((r) => r.status).sort()).toEqual(["COMPLETED_SUCCESSFULLY", "PENDING"]);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by tags",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create runs with different tags
|
||||
const taskRun1 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_urgent",
|
||||
taskIdentifier: "my-task",
|
||||
runTags: ["urgent", "production"],
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
const taskRun2 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_regular",
|
||||
taskIdentifier: "my-task",
|
||||
runTags: ["regular", "development"],
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
const taskRun3 = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_urgent_dev",
|
||||
taskIdentifier: "my-task",
|
||||
runTags: ["urgent", "development"],
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1236",
|
||||
spanId: "1236",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by tags
|
||||
const { runs } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
tags: ["urgent"],
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(2);
|
||||
expect(runs.map((r) => r.friendlyId).sort()).toEqual(["run_urgent", "run_urgent_dev"]);
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by scheduleId",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create runs with different schedule IDs
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_scheduled_1",
|
||||
taskIdentifier: "my-task",
|
||||
scheduleId: "schedule_1",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_scheduled_2",
|
||||
taskIdentifier: "my-task",
|
||||
scheduleId: "schedule_2",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_unscheduled",
|
||||
taskIdentifier: "my-task",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1236",
|
||||
spanId: "1236",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by schedule ID
|
||||
const { runs } = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
scheduleId: "schedule_1",
|
||||
});
|
||||
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0].friendlyId).toBe("run_scheduled_1");
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should filter runs by isTest flag",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
const { clickhouse } = await setupClickhouseReplication({
|
||||
prisma,
|
||||
databaseUrl: postgresContainer.getConnectionUri(),
|
||||
clickhouseUrl: clickhouseContainer.getConnectionUrl(),
|
||||
redisOptions,
|
||||
});
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test",
|
||||
slug: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test",
|
||||
slug: "test",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test",
|
||||
pkApiKey: "test",
|
||||
shortcode: "test",
|
||||
},
|
||||
});
|
||||
|
||||
// Create test and non-test runs
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_test",
|
||||
taskIdentifier: "my-task",
|
||||
isTest: true,
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_production",
|
||||
taskIdentifier: "my-task",
|
||||
isTest: false,
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
traceId: "1235",
|
||||
spanId: "1235",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
|
||||
await setTimeout(1000);
|
||||
|
||||
const runsRepository = new RunsRepository({
|
||||
prisma,
|
||||
clickhouse,
|
||||
});
|
||||
|
||||
// Test filtering by isTest=true
|
||||
const testRuns = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
isTest: true,
|
||||
});
|
||||
|
||||
expect(testRuns.runs).toHaveLength(1);
|
||||
expect(testRuns.runs[0].friendlyId).toBe("run_test");
|
||||
|
||||
// Test filtering by isTest=false
|
||||
const productionRuns = await runsRepository.listRuns({
|
||||
page: { size: 10 },
|
||||
projectId: project.id,
|
||||
environmentId: runtimeEnvironment.id,
|
||||
organizationId: organization.id,
|
||||
isTest: false,
|
||||
});
|
||||
|
||||
expect(productionRuns.runs).toHaveLength(1);
|
||||
expect(productionRuns.runs[0].friendlyId).toBe("run_production");
|
||||
}
|
||||
);
|
||||
|
||||
describe("RunsRepository (part 2/2)", () => {
|
||||
containerTest(
|
||||
"should filter runs by rootOnly flag",
|
||||
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
|
||||
@@ -1522,4 +789,4 @@ describe("RunsRepository", () => {
|
||||
expect(secondPage.pagination.previousCursor).toBeTruthy();
|
||||
}
|
||||
);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user