github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Improve unit test workflow performance (#2096)

Browse Source 
* shard unit tests

* temp enable for all pushes

* fix test workflow

* update to latest vitest and only add to root package.json

* additionally use default reporter

* gather reports before uploading

* split up slow replication tests

* split up unit tests workflow

* move workflows to parent dir

* use new paths in parent workflow

* prevent artifact clashes

* we always need to create the reports dir

* speed up merge reports

* gather reports even when tests fail

* fix artifact patterns

* increase shards

* disable push trigger again

* improve dequeue snapshot test reliability
This commit is contained in:
nicktrn
2025-05-23 12:22:14 +01:00
committed by GitHub
parent cf3735976e
commit 7c791dd519
21 changed files with 1114 additions and 1521 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/unit-tests-internal.yml
+130
View File
@@ -0,0 +1,130 @@
name: "🧪 Unit Tests: Internal"
permissions:
contents: read
on:
workflow_call:
jobs:
unitTests:
name: "🧪 Unit Tests: Internal"
runs-on: ubuntu-latest
strategy:
matrix:
shardIndex: [1, 2, 3, 4, 5, 6, 7, 8]
shardTotal: [8]
env:
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
SHARD_INDEX: ${{ matrix.shardIndex }}
SHARD_TOTAL: ${{ matrix.shardTotal }}
steps:
- name: 🔧 Disable IPv6
run: |
sudo sysctl -w net.ipv6.conf.all.disable_ipv6=1
sudo sysctl -w net.ipv6.conf.default.disable_ipv6=1
sudo sysctl -w net.ipv6.conf.lo.disable_ipv6=1
- name: 🔧 Configure docker address pool
run: |
CONFIG='{
"default-address-pools" : [
{
"base" : "172.17.0.0/12",
"size" : 20
},
{
"base" : "192.168.0.0/16",
"size" : 24
}
]
}'
mkdir -p /etc/docker
echo "$CONFIG" | sudo tee /etc/docker/daemon.json
- name: 🔧 Restart docker daemon
run: sudo systemctl restart docker
- name: ⬇️ Checkout repo
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
cache: "pnpm"
# ..to avoid rate limits when pulling images
- name: 🐳 Login to DockerHub
if: ${{ env.DOCKERHUB_USERNAME }}
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: 🐳 Skipping DockerHub login (no secrets available)
if: ${{ !env.DOCKERHUB_USERNAME }}
run: echo "DockerHub login skipped because secrets are not available."
- name: 📥 Download deps
run: pnpm install --frozen-lockfile
- name: 📀 Generate Prisma Client
run: pnpm run generate
- name: 🧪 Run Internal Unit Tests
run: pnpm run test:internal --reporter=default --reporter=blob --shard=${{ matrix.shardIndex }}/${{ matrix.shardTotal }}
- name: Gather all reports
if: ${{ !cancelled() }}
run: |
mkdir -p .vitest-reports
find . -type f -path '*/.vitest-reports/blob-*.json' \
-exec bash -c 'src="$1"; basename=$(basename "$src"); pkg=$(dirname "$src" | sed "s|^\./||;s|/\.vitest-reports$||;s|/|_|g"); cp "$src" ".vitest-reports/${pkg}-${basename}"' _ {} \;
- name: Upload blob reports to GitHub Actions Artifacts
if: ${{ !cancelled() }}
uses: actions/upload-artifact@v4
with:
name: internal-blob-report-${{ matrix.shardIndex }}
path: .vitest-reports/*
include-hidden-files: true
retention-days: 1
merge-reports:
name: "📊 Merge Reports"
if: ${{ !cancelled() }}
needs: [unitTests]
runs-on: ubuntu-latest
steps:
- name: ⬇️ Checkout repo
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
# no cache enabled, we're not installing deps
- name: Download blob reports from GitHub Actions Artifacts
uses: actions/download-artifact@v4
with:
path: .vitest-reports
pattern: internal-blob-report-*
merge-multiple: true
- name: Merge reports
run: pnpm dlx vitest run --merge-reports
.github/workflows/unit-tests-packages.yml
+130
View File
@@ -0,0 +1,130 @@
name: "🧪 Unit Tests: Packages"
permissions:
contents: read
on:
workflow_call:
jobs:
unitTests:
name: "🧪 Unit Tests: Packages"
runs-on: ubuntu-latest
strategy:
matrix:
shardIndex: [1]
shardTotal: [1]
env:
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
SHARD_INDEX: ${{ matrix.shardIndex }}
SHARD_TOTAL: ${{ matrix.shardTotal }}
steps:
- name: 🔧 Disable IPv6
run: |
sudo sysctl -w net.ipv6.conf.all.disable_ipv6=1
sudo sysctl -w net.ipv6.conf.default.disable_ipv6=1
sudo sysctl -w net.ipv6.conf.lo.disable_ipv6=1
- name: 🔧 Configure docker address pool
run: |
CONFIG='{
"default-address-pools" : [
{
"base" : "172.17.0.0/12",
"size" : 20
},
{
"base" : "192.168.0.0/16",
"size" : 24
}
]
}'
mkdir -p /etc/docker
echo "$CONFIG" | sudo tee /etc/docker/daemon.json
- name: 🔧 Restart docker daemon
run: sudo systemctl restart docker
- name: ⬇️ Checkout repo
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
cache: "pnpm"
# ..to avoid rate limits when pulling images
- name: 🐳 Login to DockerHub
if: ${{ env.DOCKERHUB_USERNAME }}
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: 🐳 Skipping DockerHub login (no secrets available)
if: ${{ !env.DOCKERHUB_USERNAME }}
run: echo "DockerHub login skipped because secrets are not available."
- name: 📥 Download deps
run: pnpm install --frozen-lockfile
- name: 📀 Generate Prisma Client
run: pnpm run generate
- name: 🧪 Run Package Unit Tests
run: pnpm run test:packages --reporter=default --reporter=blob --shard=${{ matrix.shardIndex }}/${{ matrix.shardTotal }}
- name: Gather all reports
if: ${{ !cancelled() }}
run: |
mkdir -p .vitest-reports
find . -type f -path '*/.vitest-reports/blob-*.json' \
-exec bash -c 'src="$1"; basename=$(basename "$src"); pkg=$(dirname "$src" | sed "s|^\./||;s|/\.vitest-reports$||;s|/|_|g"); cp "$src" ".vitest-reports/${pkg}-${basename}"' _ {} \;
- name: Upload blob reports to GitHub Actions Artifacts
if: ${{ !cancelled() }}
uses: actions/upload-artifact@v4
with:
name: packages-blob-report-${{ matrix.shardIndex }}
path: .vitest-reports/*
include-hidden-files: true
retention-days: 1
merge-reports:
name: "📊 Merge Reports"
if: ${{ !cancelled() }}
needs: [unitTests]
runs-on: ubuntu-latest
steps:
- name: ⬇️ Checkout repo
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
# no cache enabled, we're not installing deps
- name: Download blob reports from GitHub Actions Artifacts
uses: actions/download-artifact@v4
with:
path: .vitest-reports
pattern: packages-blob-report-*
merge-multiple: true
- name: Merge reports
run: pnpm dlx vitest run --merge-reports
.github/workflows/unit-tests-webapp.yml
+136
View File
@@ -0,0 +1,136 @@
name: "🧪 Unit Tests: Webapp"
permissions:
contents: read
on:
workflow_call:
jobs:
unitTests:
name: "🧪 Unit Tests: Webapp"
runs-on: ubuntu-latest
strategy:
matrix:
shardIndex: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
shardTotal: [10]
env:
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
SHARD_INDEX: ${{ matrix.shardIndex }}
SHARD_TOTAL: ${{ matrix.shardTotal }}
steps:
- name: 🔧 Disable IPv6
run: |
sudo sysctl -w net.ipv6.conf.all.disable_ipv6=1
sudo sysctl -w net.ipv6.conf.default.disable_ipv6=1
sudo sysctl -w net.ipv6.conf.lo.disable_ipv6=1
- name: 🔧 Configure docker address pool
run: |
CONFIG='{
"default-address-pools" : [
{
"base" : "172.17.0.0/12",
"size" : 20
},
{
"base" : "192.168.0.0/16",
"size" : 24
}
]
}'
mkdir -p /etc/docker
echo "$CONFIG" | sudo tee /etc/docker/daemon.json
- name: 🔧 Restart docker daemon
run: sudo systemctl restart docker
- name: ⬇️ Checkout repo
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
cache: "pnpm"
# ..to avoid rate limits when pulling images
- name: 🐳 Login to DockerHub
if: ${{ env.DOCKERHUB_USERNAME }}
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: 🐳 Skipping DockerHub login (no secrets available)
if: ${{ !env.DOCKERHUB_USERNAME }}
run: echo "DockerHub login skipped because secrets are not available."
- name: 📥 Download deps
run: pnpm install --frozen-lockfile
- name: 📀 Generate Prisma Client
run: pnpm run generate
- name: 🧪 Run Webapp Unit Tests
run: pnpm run test:webapp --reporter=default --reporter=blob --shard=${{ matrix.shardIndex }}/${{ matrix.shardTotal }}
env:
DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres
DIRECT_URL: postgresql://postgres:postgres@localhost:5432/postgres
SESSION_SECRET: "secret"
MAGIC_LINK_SECRET: "secret"
ENCRYPTION_KEY: "secret"
- name: Gather all reports
if: ${{ !cancelled() }}
run: |
mkdir -p .vitest-reports
find . -type f -path '*/.vitest-reports/blob-*.json' \
-exec bash -c 'src="$1"; basename=$(basename "$src"); pkg=$(dirname "$src" | sed "s|^\./||;s|/\.vitest-reports$||;s|/|_|g"); cp "$src" ".vitest-reports/${pkg}-${basename}"' _ {} \;
- name: Upload blob reports to GitHub Actions Artifacts
if: ${{ !cancelled() }}
uses: actions/upload-artifact@v4
with:
name: webapp-blob-report-${{ matrix.shardIndex }}
path: .vitest-reports/*
include-hidden-files: true
retention-days: 1
merge-reports:
name: "📊 Merge Reports"
if: ${{ !cancelled() }}
needs: [unitTests]
runs-on: ubuntu-latest
steps:
- name: ⬇️ Checkout repo
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
# no cache enabled, we're not installing deps
- name: Download blob reports from GitHub Actions Artifacts
uses: actions/download-artifact@v4
with:
path: .vitest-reports
pattern: webapp-blob-report-*
merge-multiple: true
- name: Merge reports
run: pnpm dlx vitest run --merge-reports
.github/workflows/unit-tests.yml
+9 -79
View File
@@ -7,82 +7,12 @@ on:
workflow_call:
jobs:
unitTests:
name: "🧪 Unit Tests"
runs-on: ubuntu-latest
env:
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
steps:
- name: 🔧 Disable IPv6
run: |
sudo sysctl -w net.ipv6.conf.all.disable_ipv6=1
sudo sysctl -w net.ipv6.conf.default.disable_ipv6=1
sudo sysctl -w net.ipv6.conf.lo.disable_ipv6=1
- name: 🔧 Configure docker address pool
run: |
CONFIG='{
"default-address-pools" : [
{
"base" : "172.17.0.0/12",
"size" : 20
},
{
"base" : "192.168.0.0/16",
"size" : 24
}
]
}'
mkdir -p /etc/docker
echo "$CONFIG" | sudo tee /etc/docker/daemon.json
- name: 🔧 Restart docker daemon
run: sudo systemctl restart docker
- name: ⬇️ Checkout repo
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
cache: "pnpm"
# ..to avoid rate limits when pulling images
- name: 🐳 Login to DockerHub
if: ${{ env.DOCKERHUB_USERNAME }}
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: 🐳 Skipping DockerHub login (no secrets available)
if: ${{ !env.DOCKERHUB_USERNAME }}
run: echo "DockerHub login skipped because secrets are not available."
- name: 📥 Download deps
run: pnpm install --frozen-lockfile
- name: 📀 Generate Prisma Client
run: pnpm run generate
- name: 🧪 Run Webapp Unit Tests
run: pnpm run test:webapp
env:
DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres
DIRECT_URL: postgresql://postgres:postgres@localhost:5432/postgres
SESSION_SECRET: "secret"
MAGIC_LINK_SECRET: "secret"
ENCRYPTION_KEY: "secret"
- name: 🧪 Run Package Unit Tests
run: pnpm run test:packages
- name: 🧪 Run Internal Unit Tests
run: pnpm run test:internal
webapp:
uses: ./.github/workflows/unit-tests-webapp.yml
secrets: inherit
packages:
uses: ./.github/workflows/unit-tests-packages.yml
secrets: inherit
internal:
uses: ./.github/workflows/unit-tests-internal.yml
secrets: inherit
apps/supervisor/package.json
+1 -2
View File
@@ -22,7 +22,6 @@
"zod": "3.23.8"
},
"devDependencies": {
"@types/dockerode": "^3.3.33",
"vitest": "^1.4.0"
"@types/dockerode": "^3.3.33"
}
}
apps/webapp/package.json
+1 -2
View File
@@ -254,8 +254,7 @@
"tailwindcss": "3.4.1",
"ts-node": "^10.7.0",
"tsconfig-paths": "^3.14.1",
"vite-tsconfig-paths": "^4.0.5",
"vitest": "^1.4.0"
"vite-tsconfig-paths": "^4.0.5"
},
"engines": {
"node": ">=16.0.0"
apps/webapp/test/fairDequeuingStrategy.test.ts
+5 -2
View File
@@ -256,12 +256,15 @@ describe("FairDequeuingStrategy", () => {
"consumer-1"
);
const tolerance = 0.15;
const withTolerance = (value: number) => value * (1 + tolerance);
const distribute2Duration = performance.now() - startDistribute2;
console.log("Second distribution took", distribute2Duration, "ms");
// Make sure the second call is more than 9 times faster than the first
expect(distribute2Duration).toBeLessThan(distribute1Duration / 9);
expect(distribute2Duration).toBeLessThan(withTolerance(distribute1Duration / 9));
const startDistribute3 = performance.now();
@@ -275,7 +278,7 @@ describe("FairDequeuingStrategy", () => {
console.log("Third distribution took", distribute3Duration, "ms");
// Make sure the third call is more than 4 times the second
expect(distribute3Duration).toBeGreaterThan(distribute2Duration * 4);
expect(withTolerance(distribute3Duration)).toBeGreaterThan(distribute2Duration * 4);
}
);
apps/webapp/test/runsReplicationService.test.ts → apps/webapp/test/runsReplicationService.part1.test.ts
+1 -602
View File
@@ -10,7 +10,7 @@ import superjson from "superjson";
vi.setConfig({ testTimeout: 60_000 });
describe("RunsReplicationService", () => {
describe("RunsReplicationService (part 1/2)", () => {
containerTest(
"should replicate runs to clickhouse",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
@@ -1174,605 +1174,4 @@ describe("RunsReplicationService", () => {
await runsReplicationServiceB.stop();
}
);
containerTest(
"should handover leadership to a second service, and the second service should be able to extend the leader lock",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "runs-replication-shutdown-handover",
});
// Service A
const runsReplicationServiceA = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-shutdown-handover",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 1,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
leaderLockAcquireAdditionalTimeMs: 10_000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-shutdown-handover-a", "debug"),
});
await runsReplicationServiceA.start();
// Service A
const runsReplicationServiceB = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-shutdown-handover",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 1,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
leaderLockAcquireAdditionalTimeMs: 10_000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-shutdown-handover-b", "debug"),
});
// Now we need to initiate starting the second service, and after 6 seconds, we need to shutdown the first service
await Promise.all([
setTimeout(6000).then(() => runsReplicationServiceA.stop()),
runsReplicationServiceB.start(),
]);
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(10_000);
// Check that the row was replicated to clickhouse
const queryRuns = clickhouse.reader.query({
name: "runs-replication",
query: "SELECT * FROM trigger_dev.task_runs_v2",
schema: z.any(),
});
const [queryError, result] = await queryRuns({});
expect(queryError).toBeNull();
expect(result?.length).toBe(1);
expect(result?.[0]).toEqual(
expect.objectContaining({
run_id: taskRun.id,
friendly_id: taskRun.friendlyId,
task_identifier: taskRun.taskIdentifier,
environment_id: runtimeEnvironment.id,
project_id: project.id,
organization_id: organization.id,
environment_type: "DEVELOPMENT",
engine: "V2",
})
);
await runsReplicationServiceB.stop();
}
);
containerTest(
"should replicate all 1,000 TaskRuns inserted in bulk to ClickHouse",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "runs-replication-stress-bulk-insert",
});
const runsReplicationService = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-stress-bulk-insert",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 10,
flushIntervalMs: 100,
flushBatchSize: 50,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-stress-bulk-insert", "info"),
});
await runsReplicationService.start();
const organization = await prisma.organization.create({
data: {
title: "test-stress-bulk-insert",
slug: "test-stress-bulk-insert",
},
});
const project = await prisma.project.create({
data: {
name: "test-stress-bulk-insert",
slug: "test-stress-bulk-insert",
organizationId: organization.id,
externalRef: "test-stress-bulk-insert",
},
});
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: "test-stress-bulk-insert",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test-stress-bulk-insert",
pkApiKey: "test-stress-bulk-insert",
shortcode: "test-stress-bulk-insert",
},
});
// Prepare 1,000 unique TaskRuns
const now = Date.now();
const runsData = Array.from({ length: 1000 }, (_, i) => ({
friendlyId: `run_bulk_${now}_${i}`,
taskIdentifier: `my-task-bulk`,
payload: JSON.stringify({ bulk: i }),
payloadType: "application/json",
traceId: `bulk-${i}`,
spanId: `bulk-${i}`,
queue: "test-stress-bulk-insert",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT" as const,
engine: "V2" as const,
status: "PENDING" as const,
attemptNumber: 1,
createdAt: new Date(now + i),
updatedAt: new Date(now + i),
}));
// Bulk insert
const created = await prisma.taskRun.createMany({ data: runsData });
expect(created.count).toBe(1000);
// Wait for replication
await setTimeout(5000);
// Query ClickHouse for all runs using FINAL
const queryRuns = clickhouse.reader.query({
name: "runs-replication-stress-bulk-insert",
query: `SELECT run_id, friendly_id, trace_id, task_identifier FROM trigger_dev.task_runs_v2 FINAL`,
schema: z.any(),
});
const [queryError, result] = await queryRuns({});
expect(queryError).toBeNull();
expect(result?.length).toBe(1000);
// Check a few random runs for correctness
for (let i = 0; i < 10; i++) {
const idx = Math.floor(Math.random() * 1000);
const expected = runsData[idx];
const found = result?.find((r: any) => r.friendly_id === expected.friendlyId);
expect(found).toBeDefined();
expect(found).toEqual(
expect.objectContaining({
friendly_id: expected.friendlyId,
trace_id: expected.traceId,
task_identifier: expected.taskIdentifier,
})
);
}
await runsReplicationService.stop();
}
);
containerTest(
"should replicate all 1,000 TaskRuns inserted in bulk to ClickHouse with updates",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "runs-replication-stress-bulk-insert",
});
const runsReplicationService = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-stress-bulk-insert",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 10,
flushIntervalMs: 100,
flushBatchSize: 50,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-stress-bulk-insert", "info"),
});
await runsReplicationService.start();
const organization = await prisma.organization.create({
data: {
title: "test-stress-bulk-insert",
slug: "test-stress-bulk-insert",
},
});
const project = await prisma.project.create({
data: {
name: "test-stress-bulk-insert",
slug: "test-stress-bulk-insert",
organizationId: organization.id,
externalRef: "test-stress-bulk-insert",
},
});
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: "test-stress-bulk-insert",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test-stress-bulk-insert",
pkApiKey: "test-stress-bulk-insert",
shortcode: "test-stress-bulk-insert",
},
});
// Prepare 1,000 unique TaskRuns
const now = Date.now();
const runsData = Array.from({ length: 1000 }, (_, i) => ({
friendlyId: `run_bulk_${now}_${i}`,
taskIdentifier: `my-task-bulk`,
payload: JSON.stringify({ bulk: i }),
payloadType: "application/json",
traceId: `bulk-${i}`,
spanId: `bulk-${i}`,
queue: "test-stress-bulk-insert",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT" as const,
engine: "V2" as const,
status: "PENDING" as const,
attemptNumber: 1,
createdAt: new Date(now + i),
updatedAt: new Date(now + i),
}));
// Bulk insert
const created = await prisma.taskRun.createMany({ data: runsData });
expect(created.count).toBe(1000);
// Update all the runs
await prisma.taskRun.updateMany({
data: { status: "COMPLETED_SUCCESSFULLY" },
});
// Wait for replication
await setTimeout(5000);
// Query ClickHouse for all runs using FINAL
const queryRuns = clickhouse.reader.query({
name: "runs-replication-stress-bulk-insert",
query: `SELECT * FROM trigger_dev.task_runs_v2 FINAL`,
schema: z.any(),
});
const [queryError, result] = await queryRuns({});
expect(queryError).toBeNull();
expect(result?.length).toBe(1000);
// Check a few random runs for correctness
for (let i = 0; i < 10; i++) {
const idx = Math.floor(Math.random() * 1000);
const expected = runsData[idx];
const found = result?.find((r: any) => r.friendly_id === expected.friendlyId);
expect(found).toBeDefined();
expect(found).toEqual(
expect.objectContaining({
friendly_id: expected.friendlyId,
trace_id: expected.traceId,
task_identifier: expected.taskIdentifier,
status: "COMPLETED_SUCCESSFULLY",
})
);
}
await runsReplicationService.stop();
}
);
containerTest(
"should replicate all events in a single transaction (insert, update)",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "runs-replication-multi-event-tx",
});
const runsReplicationService = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-multi-event-tx",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 10,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
});
await runsReplicationService.start();
const organization = await prisma.organization.create({
data: {
title: "test-multi-event-tx",
slug: "test-multi-event-tx",
},
});
const project = await prisma.project.create({
data: {
name: "test-multi-event-tx",
slug: "test-multi-event-tx",
organizationId: organization.id,
externalRef: "test-multi-event-tx",
},
});
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: "test-multi-event-tx",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test-multi-event-tx",
pkApiKey: "test-multi-event-tx",
shortcode: "test-multi-event-tx",
},
});
// Start a transaction
const [run1, run2] = await prisma.$transaction(async (tx) => {
const run1 = await tx.taskRun.create({
data: {
friendlyId: `run_multi_event_1_${Date.now()}`,
taskIdentifier: "my-task-multi-event-1",
payload: JSON.stringify({ multi: 1 }),
payloadType: "application/json",
traceId: `multi-1-${Date.now()}`,
spanId: `multi-1-${Date.now()}`,
queue: "test-multi-event-tx",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT",
engine: "V2",
status: "PENDING",
attemptNumber: 1,
createdAt: new Date(),
updatedAt: new Date(),
},
});
const run2 = await tx.taskRun.create({
data: {
friendlyId: `run_multi_event_2_${Date.now()}`,
taskIdentifier: "my-task-multi-event-2",
payload: JSON.stringify({ multi: 2 }),
payloadType: "application/json",
traceId: `multi-2-${Date.now()}`,
spanId: `multi-2-${Date.now()}`,
queue: "test-multi-event-tx",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT",
engine: "V2",
status: "PENDING",
attemptNumber: 1,
createdAt: new Date(),
updatedAt: new Date(),
},
});
await tx.taskRun.update({
where: { id: run1.id },
data: { status: "COMPLETED_SUCCESSFULLY" },
});
return [run1, run2];
});
// Wait for replication
await setTimeout(1000);
// Query ClickHouse for both runs using FINAL
const queryRuns = clickhouse.reader.query({
name: "runs-replication-multi-event-tx",
query: `SELECT * FROM trigger_dev.task_runs_v2 FINAL WHERE run_id IN ({run_id_1:String}, {run_id_2:String})`,
schema: z.any(),
params: z.object({ run_id_1: z.string(), run_id_2: z.string() }),
});
const [queryError, result] = await queryRuns({ run_id_1: run1.id, run_id_2: run2.id });
expect(queryError).toBeNull();
expect(result?.length).toBe(2);
const run1Result = result?.find((r: any) => r.run_id === run1.id);
const run2Result = result?.find((r: any) => r.run_id === run2.id);
expect(run1Result).toBeDefined();
expect(run1Result).toEqual(
expect.objectContaining({ run_id: run1.id, status: "COMPLETED_SUCCESSFULLY" })
);
expect(run2Result).toBeDefined();
expect(run2Result).toEqual(expect.objectContaining({ run_id: run2.id }));
await runsReplicationService.stop();
}
);
containerTest(
"should be able to handle processing transactions for a long period of time",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "runs-replication-long-tx",
});
const runsReplicationService = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-long-tx",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 10,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-long-tx", "info"),
});
await runsReplicationService.start();
const organization = await prisma.organization.create({
data: {
title: "test-long-tx",
slug: "test-long-tx",
},
});
const project = await prisma.project.create({
data: {
name: "test-long-tx",
slug: "test-long-tx",
organizationId: organization.id,
externalRef: "test-long-tx",
},
});
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: "test-long-tx",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test-long-tx",
pkApiKey: "test-long-tx",
shortcode: "test-long-tx",
},
});
// Start an interval that will create a new run every 500ms for 4 minutes
const interval = setInterval(async () => {
await prisma.taskRun.create({
data: {
friendlyId: `run_long_tx_${Date.now()}`,
taskIdentifier: "my-task-long-tx",
payload: JSON.stringify({ long: 1 }),
payloadType: "application/json",
traceId: `long-${Date.now()}`,
spanId: `long-${Date.now()}`,
queue: "test-long-tx",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT",
engine: "V2",
status: "PENDING",
attemptNumber: 1,
createdAt: new Date(),
updatedAt: new Date(),
},
});
}, 500);
// Wait for 1 minute
await setTimeout(1 * 60 * 1000);
// Stop the interval
clearInterval(interval);
// Wait for replication
await setTimeout(1000);
// Query ClickHouse for all runs using FINAL
const queryRuns = clickhouse.reader.query({
name: "runs-replication-long-tx",
query: `SELECT * FROM trigger_dev.task_runs_v2 FINAL`,
schema: z.any(),
});
const [queryError, result] = await queryRuns({});
expect(queryError).toBeNull();
expect(result?.length).toBeGreaterThanOrEqual(50);
await runsReplicationService.stop();
},
{ timeout: 60_000 * 5 }
);
});
apps/webapp/test/runsReplicationService.part2.test.ts
+614
View File
@@ -0,0 +1,614 @@
import { ClickHouse } from "@internal/clickhouse";
import { containerTest } from "@internal/testcontainers";
import { Logger } from "@trigger.dev/core/logger";
import { setTimeout } from "node:timers/promises";
import { z } from "zod";
import { TaskRunStatus } from "~/database-types";
import { RunsReplicationService } from "~/services/runsReplicationService.server";
import { createInMemoryTracing } from "./utils/tracing";
import superjson from "superjson";
vi.setConfig({ testTimeout: 60_000 });
describe("RunsReplicationService (part 2/2)", () => {
containerTest(
"should handover leadership to a second service, and the second service should be able to extend the leader lock",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "runs-replication-shutdown-handover",
});
// Service A
const runsReplicationServiceA = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-shutdown-handover",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 1,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
leaderLockAcquireAdditionalTimeMs: 10_000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-shutdown-handover-a", "debug"),
});
await runsReplicationServiceA.start();
// Service A
const runsReplicationServiceB = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-shutdown-handover",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 1,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
leaderLockAcquireAdditionalTimeMs: 10_000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-shutdown-handover-b", "debug"),
});
// Now we need to initiate starting the second service, and after 6 seconds, we need to shutdown the first service
await Promise.all([
setTimeout(6000).then(() => runsReplicationServiceA.stop()),
runsReplicationServiceB.start(),
]);
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(10_000);
// Check that the row was replicated to clickhouse
const queryRuns = clickhouse.reader.query({
name: "runs-replication",
query: "SELECT * FROM trigger_dev.task_runs_v2",
schema: z.any(),
});
const [queryError, result] = await queryRuns({});
expect(queryError).toBeNull();
expect(result?.length).toBe(1);
expect(result?.[0]).toEqual(
expect.objectContaining({
run_id: taskRun.id,
friendly_id: taskRun.friendlyId,
task_identifier: taskRun.taskIdentifier,
environment_id: runtimeEnvironment.id,
project_id: project.id,
organization_id: organization.id,
environment_type: "DEVELOPMENT",
engine: "V2",
})
);
await runsReplicationServiceB.stop();
}
);
containerTest(
"should replicate all 1,000 TaskRuns inserted in bulk to ClickHouse",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "runs-replication-stress-bulk-insert",
});
const runsReplicationService = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-stress-bulk-insert",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 10,
flushIntervalMs: 100,
flushBatchSize: 50,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-stress-bulk-insert", "info"),
});
await runsReplicationService.start();
const organization = await prisma.organization.create({
data: {
title: "test-stress-bulk-insert",
slug: "test-stress-bulk-insert",
},
});
const project = await prisma.project.create({
data: {
name: "test-stress-bulk-insert",
slug: "test-stress-bulk-insert",
organizationId: organization.id,
externalRef: "test-stress-bulk-insert",
},
});
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: "test-stress-bulk-insert",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test-stress-bulk-insert",
pkApiKey: "test-stress-bulk-insert",
shortcode: "test-stress-bulk-insert",
},
});
// Prepare 1,000 unique TaskRuns
const now = Date.now();
const runsData = Array.from({ length: 1000 }, (_, i) => ({
friendlyId: `run_bulk_${now}_${i}`,
taskIdentifier: `my-task-bulk`,
payload: JSON.stringify({ bulk: i }),
payloadType: "application/json",
traceId: `bulk-${i}`,
spanId: `bulk-${i}`,
queue: "test-stress-bulk-insert",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT" as const,
engine: "V2" as const,
status: "PENDING" as const,
attemptNumber: 1,
createdAt: new Date(now + i),
updatedAt: new Date(now + i),
}));
// Bulk insert
const created = await prisma.taskRun.createMany({ data: runsData });
expect(created.count).toBe(1000);
// Wait for replication
await setTimeout(5000);
// Query ClickHouse for all runs using FINAL
const queryRuns = clickhouse.reader.query({
name: "runs-replication-stress-bulk-insert",
query: `SELECT run_id, friendly_id, trace_id, task_identifier FROM trigger_dev.task_runs_v2 FINAL`,
schema: z.any(),
});
const [queryError, result] = await queryRuns({});
expect(queryError).toBeNull();
expect(result?.length).toBe(1000);
// Check a few random runs for correctness
for (let i = 0; i < 10; i++) {
const idx = Math.floor(Math.random() * 1000);
const expected = runsData[idx];
const found = result?.find((r: any) => r.friendly_id === expected.friendlyId);
expect(found).toBeDefined();
expect(found).toEqual(
expect.objectContaining({
friendly_id: expected.friendlyId,
trace_id: expected.traceId,
task_identifier: expected.taskIdentifier,
})
);
}
await runsReplicationService.stop();
}
);
containerTest(
"should replicate all 1,000 TaskRuns inserted in bulk to ClickHouse with updates",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "runs-replication-stress-bulk-insert",
});
const runsReplicationService = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-stress-bulk-insert",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 10,
flushIntervalMs: 100,
flushBatchSize: 50,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-stress-bulk-insert", "info"),
});
await runsReplicationService.start();
const organization = await prisma.organization.create({
data: {
title: "test-stress-bulk-insert",
slug: "test-stress-bulk-insert",
},
});
const project = await prisma.project.create({
data: {
name: "test-stress-bulk-insert",
slug: "test-stress-bulk-insert",
organizationId: organization.id,
externalRef: "test-stress-bulk-insert",
},
});
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: "test-stress-bulk-insert",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test-stress-bulk-insert",
pkApiKey: "test-stress-bulk-insert",
shortcode: "test-stress-bulk-insert",
},
});
// Prepare 1,000 unique TaskRuns
const now = Date.now();
const runsData = Array.from({ length: 1000 }, (_, i) => ({
friendlyId: `run_bulk_${now}_${i}`,
taskIdentifier: `my-task-bulk`,
payload: JSON.stringify({ bulk: i }),
payloadType: "application/json",
traceId: `bulk-${i}`,
spanId: `bulk-${i}`,
queue: "test-stress-bulk-insert",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT" as const,
engine: "V2" as const,
status: "PENDING" as const,
attemptNumber: 1,
createdAt: new Date(now + i),
updatedAt: new Date(now + i),
}));
// Bulk insert
const created = await prisma.taskRun.createMany({ data: runsData });
expect(created.count).toBe(1000);
// Update all the runs
await prisma.taskRun.updateMany({
data: { status: "COMPLETED_SUCCESSFULLY" },
});
// Wait for replication
await setTimeout(5000);
// Query ClickHouse for all runs using FINAL
const queryRuns = clickhouse.reader.query({
name: "runs-replication-stress-bulk-insert",
query: `SELECT * FROM trigger_dev.task_runs_v2 FINAL`,
schema: z.any(),
});
const [queryError, result] = await queryRuns({});
expect(queryError).toBeNull();
expect(result?.length).toBe(1000);
// Check a few random runs for correctness
for (let i = 0; i < 10; i++) {
const idx = Math.floor(Math.random() * 1000);
const expected = runsData[idx];
const found = result?.find((r: any) => r.friendly_id === expected.friendlyId);
expect(found).toBeDefined();
expect(found).toEqual(
expect.objectContaining({
friendly_id: expected.friendlyId,
trace_id: expected.traceId,
task_identifier: expected.taskIdentifier,
status: "COMPLETED_SUCCESSFULLY",
})
);
}
await runsReplicationService.stop();
}
);
containerTest(
"should replicate all events in a single transaction (insert, update)",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "runs-replication-multi-event-tx",
});
const runsReplicationService = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-multi-event-tx",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 10,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
});
await runsReplicationService.start();
const organization = await prisma.organization.create({
data: {
title: "test-multi-event-tx",
slug: "test-multi-event-tx",
},
});
const project = await prisma.project.create({
data: {
name: "test-multi-event-tx",
slug: "test-multi-event-tx",
organizationId: organization.id,
externalRef: "test-multi-event-tx",
},
});
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: "test-multi-event-tx",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test-multi-event-tx",
pkApiKey: "test-multi-event-tx",
shortcode: "test-multi-event-tx",
},
});
// Start a transaction
const [run1, run2] = await prisma.$transaction(async (tx) => {
const run1 = await tx.taskRun.create({
data: {
friendlyId: `run_multi_event_1_${Date.now()}`,
taskIdentifier: "my-task-multi-event-1",
payload: JSON.stringify({ multi: 1 }),
payloadType: "application/json",
traceId: `multi-1-${Date.now()}`,
spanId: `multi-1-${Date.now()}`,
queue: "test-multi-event-tx",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT",
engine: "V2",
status: "PENDING",
attemptNumber: 1,
createdAt: new Date(),
updatedAt: new Date(),
},
});
const run2 = await tx.taskRun.create({
data: {
friendlyId: `run_multi_event_2_${Date.now()}`,
taskIdentifier: "my-task-multi-event-2",
payload: JSON.stringify({ multi: 2 }),
payloadType: "application/json",
traceId: `multi-2-${Date.now()}`,
spanId: `multi-2-${Date.now()}`,
queue: "test-multi-event-tx",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT",
engine: "V2",
status: "PENDING",
attemptNumber: 1,
createdAt: new Date(),
updatedAt: new Date(),
},
});
await tx.taskRun.update({
where: { id: run1.id },
data: { status: "COMPLETED_SUCCESSFULLY" },
});
return [run1, run2];
});
// Wait for replication
await setTimeout(1000);
// Query ClickHouse for both runs using FINAL
const queryRuns = clickhouse.reader.query({
name: "runs-replication-multi-event-tx",
query: `SELECT * FROM trigger_dev.task_runs_v2 FINAL WHERE run_id IN ({run_id_1:String}, {run_id_2:String})`,
schema: z.any(),
params: z.object({ run_id_1: z.string(), run_id_2: z.string() }),
});
const [queryError, result] = await queryRuns({ run_id_1: run1.id, run_id_2: run2.id });
expect(queryError).toBeNull();
expect(result?.length).toBe(2);
const run1Result = result?.find((r: any) => r.run_id === run1.id);
const run2Result = result?.find((r: any) => r.run_id === run2.id);
expect(run1Result).toBeDefined();
expect(run1Result).toEqual(
expect.objectContaining({ run_id: run1.id, status: "COMPLETED_SUCCESSFULLY" })
);
expect(run2Result).toBeDefined();
expect(run2Result).toEqual(expect.objectContaining({ run_id: run2.id }));
await runsReplicationService.stop();
}
);
containerTest(
"should be able to handle processing transactions for a long period of time",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "runs-replication-long-tx",
});
const runsReplicationService = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-long-tx",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 10,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-long-tx", "info"),
});
await runsReplicationService.start();
const organization = await prisma.organization.create({
data: {
title: "test-long-tx",
slug: "test-long-tx",
},
});
const project = await prisma.project.create({
data: {
name: "test-long-tx",
slug: "test-long-tx",
organizationId: organization.id,
externalRef: "test-long-tx",
},
});
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: "test-long-tx",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test-long-tx",
pkApiKey: "test-long-tx",
shortcode: "test-long-tx",
},
});
// Start an interval that will create a new run every 500ms for 4 minutes
const interval = setInterval(async () => {
await prisma.taskRun.create({
data: {
friendlyId: `run_long_tx_${Date.now()}`,
taskIdentifier: "my-task-long-tx",
payload: JSON.stringify({ long: 1 }),
payloadType: "application/json",
traceId: `long-${Date.now()}`,
spanId: `long-${Date.now()}`,
queue: "test-long-tx",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT",
engine: "V2",
status: "PENDING",
attemptNumber: 1,
createdAt: new Date(),
updatedAt: new Date(),
},
});
}, 500);
// Wait for 1 minute
await setTimeout(1 * 60 * 1000);
// Stop the interval
clearInterval(interval);
// Wait for replication
await setTimeout(1000);
// Query ClickHouse for all runs using FINAL
const queryRuns = clickhouse.reader.query({
name: "runs-replication-long-tx",
query: `SELECT * FROM trigger_dev.task_runs_v2 FINAL`,
schema: z.any(),
});
const [queryError, result] = await queryRuns({});
expect(queryError).toBeNull();
expect(result?.length).toBeGreaterThanOrEqual(50);
await runsReplicationService.stop();
},
{ timeout: 60_000 * 5 }
);
});
internal-packages/clickhouse/package.json
+1 -3
View File
@@ -14,9 +14,7 @@
},
"devDependencies": {
"@internal/testcontainers": "workspace:*",
"@vitest/coverage-v8": "^3.0.8",
"rimraf": "6.0.1",
"vitest": "^3.0.8"
"rimraf": "6.0.1"
},
"scripts": {
"clean": "rimraf dist",
internal-packages/redis/package.json
-3
View File
@@ -9,9 +9,6 @@
"ioredis": "^5.3.2",
"@trigger.dev/core": "workspace:*"
},
"devDependencies": {
"vitest": "^1.4.0"
},
"scripts": {
"typecheck": "tsc --noEmit"
}
internal-packages/replication/package.json
-2
View File
@@ -14,9 +14,7 @@
},
"devDependencies": {
"@internal/testcontainers": "workspace:*",
"@vitest/coverage-v8": "^3.0.8",
"rimraf": "6.0.1",
"vitest": "^3.0.8",
"@types/pg": "8.11.14"
},
"scripts": {
internal-packages/run-engine/package.json
+1 -3
View File
@@ -35,9 +35,7 @@
"devDependencies": {
"@internal/testcontainers": "workspace:*",
"@types/seedrandom": "^3.0.8",
"@vitest/coverage-v8": "^3.0.8",
"rimraf": "6.0.1",
"vitest": "^3.0.8"
"rimraf": "6.0.1"
},
"scripts": {
"clean": "rimraf dist",
internal-packages/testcontainers/package.json
+1 -2
View File
@@ -16,8 +16,7 @@
"@trigger.dev/core": "workspace:*",
"std-env": "^3.9.0",
"testcontainers": "^10.25.0",
"tinyexec": "^0.3.0",
"vitest": "^1.4.0"
"tinyexec": "^0.3.0"
},
"scripts": {
"typecheck": "tsc --noEmit"
internal-packages/tracing/package.json
-3
View File
@@ -11,9 +11,6 @@
"@opentelemetry/semantic-conventions": "^1.27.0",
"@trigger.dev/core": "workspace:*"
},
"devDependencies": {
"vitest": "^1.4.0"
},
"scripts": {
"typecheck": "tsc --noEmit"
}
internal-packages/zod-worker/package.json
+1 -2
View File
@@ -14,8 +14,7 @@
},
"devDependencies": {
"@types/lodash.omit": "^4.5.7",
"@types/pg": "8.6.6",
"vitest": "^1.4.0"
"@types/pg": "8.6.6"
},
"scripts": {
"typecheck": "tsc --noEmit"
package.json
+2 -1
View File
@@ -54,6 +54,7 @@
"@playwright/test": "^1.36.2",
"@trigger.dev/database": "workspace:*",
"@types/node": "20.14.14",
"@vitest/coverage-v8": "3.1.4",
"autoprefixer": "^10.4.12",
"eslint-plugin-turbo": "^2.0.4",
"lefthook": "^1.11.3",
@@ -64,7 +65,7 @@
"typescript": "5.5.4",
"vite": "^4.1.1",
"vite-tsconfig-paths": "^4.0.5",
"vitest": "^0.28.4"
"vitest": "3.1.4"
},
"packageManager": "pnpm@8.15.5",
"dependencies": {
packages/cli-v3/package.json
+1 -2
View File
@@ -65,8 +65,7 @@
"rimraf": "^5.0.7",
"ts-essentials": "10.0.1",
"tshy": "^3.0.2",
"tsx": "4.17.0",
"vitest": "^2.0.5"
"tsx": "4.17.0"
},
"scripts": {
"clean": "rimraf dist .tshy .tshy-build .turbo",
packages/core/package.json
+1 -2
View File
@@ -212,8 +212,7 @@
"rimraf": "^3.0.2",
"ts-essentials": "10.0.1",
"tshy": "^3.0.2",
"tsx": "4.17.0",
"vitest": "^1.6.0"
"tsx": "4.17.0"
},
"engines": {
"node": ">=18.20.0"
packages/redis-worker/package.json
+1 -2
View File
@@ -37,8 +37,7 @@
"@types/lodash.omit": "^4.5.7",
"rimraf": "6.0.1",
"tsup": "^8.4.0",
"tsx": "4.17.0",
"vitest": "^1.4.0"
"tsx": "4.17.0"
},
"engines": {
"node": ">=18.20.0"
pnpm-lock.yaml
Generated
+78 -809
View File
File diff suppressed because it is too large Load Diff