Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/e2e-webapp.yml
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ on:
jobs:
e2eTests:
name: "🧪 E2E Tests: Webapp"
runs-on: warp-ubuntu-latest-x64-8x
runs-on: warp-ubuntu-latest-x64-16x
timeout-minutes: 20
env:
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
Expand Down
2 changes: 1 addition & 1 deletion .github/workflows/e2e.yml
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ jobs:
strategy:
fail-fast: false
matrix:
os: [warp-ubuntu-latest-x64-4x, warp-windows-latest-x64-4x]
os: [warp-ubuntu-latest-x64-4x, warp-windows-latest-x64-8x]
package-manager: ["npm", "pnpm"]
steps:
- name: ⬇️ Checkout repo
Expand Down
4 changes: 2 additions & 2 deletions .github/workflows/release.yml
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ jobs:

release:
name: 🚀 Release npm packages
runs-on: ubuntu-latest
runs-on: ubuntu-latest # this cannot run on non-GH runner
environment: npm-publish
permissions:
contents: write
Expand Down Expand Up @@ -281,7 +281,7 @@ jobs:
# The prerelease job needs to be on the same workflow file due to a limitation related to how npm verifies OIDC claims.
prerelease:
name: 🧪 Prerelease
runs-on: ubuntu-latest
runs-on: ubuntu-latest # this cannot run on non-GH runner
environment: npm-publish
permissions:
contents: read
Expand Down
2 changes: 1 addition & 1 deletion .github/workflows/typecheck.yml
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ permissions:

jobs:
typecheck:
runs-on: warp-ubuntu-latest-x64-8x
runs-on: warp-ubuntu-latest-x64-16x

steps:
- name: ⬇️ Checkout repo
Expand Down
85 changes: 35 additions & 50 deletions .github/workflows/unit-tests-internal.yml
Original file line number Diff line number Diff line change
Expand Up @@ -14,17 +14,14 @@ on:
jobs:
unitTests:
name: "🧪 Unit Tests: Internal"
runs-on: warp-ubuntu-latest-x64-8x
strategy:
# one flaky shard shouldn't cancel its siblings - lets us re-run only the failed shard
fail-fast: false
matrix:
shardIndex: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]
shardTotal: [12]
# Single big machine instead of a 12-job matrix: the internal suites are serial
# (fileParallelism: false) and container-wait-bound, so 12 in-machine shard processes
# fit comfortably in 32 vCPUs while paying the setup cost (install, prisma generate,
# image pulls) once instead of 12 times.
runs-on: warp-ubuntu-latest-x64-32x
env:
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
SHARD_INDEX: ${{ matrix.shardIndex }}
SHARD_TOTAL: ${{ matrix.shardTotal }}
SHARD_TOTAL: 12
steps:
- name: 🔧 Disable IPv6
run: |
Expand Down Expand Up @@ -108,8 +105,34 @@ jobs:
- 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 }} --passWithNoTests
- name: 🏗️ Build test dependencies
# Build once up-front so the parallel shard runs below (turbo --only) never race
# to build or cache-restore the same outputs concurrently.
run: pnpm exec turbo run build --filter "@internal/*..."

- name: 🧪 Run Internal Unit Tests (${{ env.SHARD_TOTAL }} in-machine shards)
run: |
# Same shard partitioning as the old 12-job matrix (DurationShardingSequencer
# keys off --shard=i/N), but as parallel local processes. --only skips the
# ^build dependency handled by the step above.
status=0
declare -a pids
for i in $(seq 1 "$SHARD_TOTAL"); do
pnpm exec turbo run test --only --concurrency=1 --filter "@internal/*" -- \
--run --reporter=default --reporter=blob --shard="$i/$SHARD_TOTAL" --passWithNoTests \
> "/tmp/internal-shard-$i.log" 2>&1 &
pids[i]=$!
done
for i in $(seq 1 "$SHARD_TOTAL"); do
if ! wait "${pids[i]}"; then
status=1
echo "::error::internal unit test shard $i/$SHARD_TOTAL failed"
fi
echo "::group::🧪 shard $i/$SHARD_TOTAL"
cat "/tmp/internal-shard-$i.log"
echo "::endgroup::"
done
exit "$status"
Comment thread
carderne marked this conversation as resolved.

- name: Gather all reports
if: ${{ !cancelled() }}
Expand All @@ -118,44 +141,6 @@ jobs:
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
- name: 📊 Merge reports
if: ${{ !cancelled() }}
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
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: warp-ubuntu-latest-x64-2x
steps:
- name: ⬇️ Checkout repo
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
with:
fetch-depth: 1
persist-credentials: false

- name: ⎔ Setup pnpm
uses: pnpm/action-setup@fc06bc1257f339d1d5d8b3a19a8cae5388b55320 # v5.0.0
with:
version: 10.33.2

- name: ⎔ Setup node
uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0
with:
node-version: 22.23.1
# no cache enabled, we're not installing deps

- name: Download blob reports from GitHub Actions Artifacts
uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
with:
path: .vitest-reports
pattern: internal-blob-report-*
merge-multiple: true

- name: Merge reports
run: pnpm dlx vitest@4.1.7 run --merge-reports --pass-with-no-tests
11 changes: 8 additions & 3 deletions .github/workflows/unit-tests-webapp.yml
Original file line number Diff line number Diff line change
Expand Up @@ -14,13 +14,18 @@ on:
jobs:
unitTests:
name: "🧪 Unit Tests: Webapp"
runs-on: warp-ubuntu-latest-x64-8x
# 10 shards on 16x machines: webapp test throughput is limited per-machine (one
# docker daemon + disk absorbing all the per-file Postgres/ClickHouse container
# spin-up), so many machines beats few big ones - fewer/bigger (3x32) measured
# SLOWER than 10x8. The 16x (vs 8x) gives the fork pool the CPU headroom the 8x
# runners lacked. Setup overhead per machine is ~1 min on warm runners.
runs-on: warp-ubuntu-latest-x64-16x
strategy:
# one flaky shard shouldn't cancel its siblings - lets us re-run only the failed shard
fail-fast: false
matrix:
shardIndex: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
shardTotal: [10]
shardIndex: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]
shardTotal: [12]
env:
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
SHARD_INDEX: ${{ matrix.shardIndex }}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,191 @@
// Split from triggerTask.server.test.ts (combined parent + locked-worker read)
// so CI's duration-based sharding can balance the container-heavy tests.
import { describe, expect, vi } from "vitest";

// Mock the db prisma client. The service is constructed against a real
// testcontainer prisma instead — these empty singletons only satisfy the
// module-level imports of the production wiring (infrastructure boundary).
vi.mock("~/db.server", () => ({
prisma: {},
$replica: {},
runOpsNewPrisma: {},
runOpsLegacyPrisma: {},
runOpsNewReplica: {},
runOpsLegacyReplica: {},
}));
// Inherited harness boilerplate. The parent read under test takes the
// findRun(where, client) overload with this.prisma, so it does not consult this
// flag; the mock only satisfies other wiring imported transitively.
vi.mock("~/v3/runOpsMigration/splitMode.server", () => ({ isSplitEnabled: async () => false }));

vi.mock("~/services/platform.v3.server", async (importOriginal) => {
const actual = (await importOriginal()) as Record<string, unknown>;
return {
...actual,
getEntitlement: vi.fn(),
};
});

import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "@internal/run-engine/tests";
import { assertNonNullable, containerTest } from "@internal/testcontainers";
import { trace } from "@opentelemetry/api";
import { IdempotencyKeyConcern } from "~/runEngine/concerns/idempotencyKeys.server";
import { DefaultQueueManager } from "~/runEngine/concerns/queues.server";
import { RunEngineTriggerTaskService } from "./triggerTask.server";
import {
buildEngine,
CapturingParentRunValidator,
MockPayloadProcessor,
MockTraceEventConcern,
} from "./triggerTask.server.test.helpers";

vi.setConfig({ testTimeout: 60_000 }); // 60 seconds timeout

describe("RunEngineTriggerTaskService combined parent + locked-worker reads", () => {
containerTest(
"issues two independent single-table reads when one call supplies both parentRunId and lockToVersion",
async ({ prisma, redisOptions }) => {
const engine = buildEngine(prisma, redisOptions);

try {
const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const taskIdentifier = "test-task";
const { worker } = await setupBackgroundWorker(engine, environment, taskIdentifier);

const workerRow = await prisma.backgroundWorker.findUniqueOrThrow({
where: { id: worker.id },
});

// Count BOTH reads issued by the service on the control-plane client:
// the parent read (runStore.findRun → taskRun.findFirst) and the
// locked-worker read (backgroundWorker.findFirst). Capture every
// findFirst arg so we can assert no read carries a cross-seam include.
let taskRunFindFirstCalls = 0;
let backgroundWorkerFindFirstCalls = 0;
const findFirstArgs: any[] = [];
const countingPrisma = new Proxy(prisma, {
get(target, prop, receiver) {
if (prop === "backgroundWorker") {
const delegate = Reflect.get(target, prop, receiver);
return new Proxy(delegate, {
get(bwTarget, bwProp, bwReceiver) {
if (bwProp === "findFirst") {
return async (args: any) => {
backgroundWorkerFindFirstCalls += 1;
findFirstArgs.push(args);
return (delegate as any).findFirst(args);
};
}
const value = Reflect.get(bwTarget, bwProp, bwReceiver);
return typeof value === "function" ? value.bind(bwTarget) : value;
},
});
}
if (prop === "taskRun") {
const delegate = Reflect.get(target, prop, receiver);
return new Proxy(delegate, {
get(trTarget, trProp, trReceiver) {
if (trProp === "findFirst") {
return async (args: any) => {
taskRunFindFirstCalls += 1;
findFirstArgs.push(args);
return (delegate as any).findFirst(args);
};
}
const value = Reflect.get(trTarget, trProp, trReceiver);
return typeof value === "function" ? value.bind(trTarget) : value;
},
});
}
const value = Reflect.get(target, prop, receiver);
return typeof value === "function" ? value.bind(target) : value;
},
}) as typeof prisma;

const triggerTaskService = new RunEngineTriggerTaskService({
engine,
prisma: countingPrisma,
payloadProcessor: new MockPayloadProcessor(),
// queueConcern/idempotency get the real unproxied prisma so the
// counting proxy only observes reads issued by the service itself.
queueConcern: new DefaultQueueManager(prisma, engine),
idempotencyKeyConcern: new IdempotencyKeyConcern(
prisma,
engine,
new MockTraceEventConcern()
),
validator: new CapturingParentRunValidator(),
traceEventConcern: new MockTraceEventConcern(),
tracer: trace.getTracer("test", "0.0.0"),
metadataMaximumSize: 1024 * 1024 * 1,
});

// ROOT parent first (uses the unproxied prisma via a separate service so
// its internal reads don't pollute the child's counts).
const parentService = new RunEngineTriggerTaskService({
engine,
prisma,
payloadProcessor: new MockPayloadProcessor(),
queueConcern: new DefaultQueueManager(prisma, engine),
idempotencyKeyConcern: new IdempotencyKeyConcern(
prisma,
engine,
new MockTraceEventConcern()
),
validator: new CapturingParentRunValidator(),
traceEventConcern: new MockTraceEventConcern(),
tracer: trace.getTracer("test", "0.0.0"),
metadataMaximumSize: 1024 * 1024 * 1,
});
const parentResult = await parentService.call({
taskId: taskIdentifier,
environment,
body: { payload: { kind: "parent" } },
});
assertNonNullable(parentResult);

// CHILD supplying BOTH parentRunId AND lockToVersion in one call.
const childResult = await triggerTaskService.call({
taskId: taskIdentifier,
environment,
body: {
payload: { kind: "child" },
options: {
parentRunId: parentResult.run.friendlyId,
lockToVersion: workerRow.version,
},
},
});
assertNonNullable(childResult);

const parentRow = await prisma.taskRun.findUniqueOrThrow({
where: { id: parentResult.run.id },
});
const childRow = await prisma.taskRun.findUniqueOrThrow({
where: { id: childResult.run.id },
});

// Child resolved the parent (single-table parent read).
expect(childRow.parentTaskRunId).toBe(parentRow.id);
expect(childRow.depth).toBe(parentRow.depth + 1);

// Child locked to the worker (single-table worker read).
expect(childRow.lockedToVersionId).toBe(workerRow.id);
expect(childRow.taskVersion).toBe(workerRow.version);

// Exactly one backgroundWorker.findFirst fired for the locked-worker read,
// and at least one taskRun.findFirst fired for the parent read.
expect(backgroundWorkerFindFirstCalls).toBe(1);
expect(taskRunFindFirstCalls).toBeGreaterThanOrEqual(1);

// NO-JOIN proof: no captured read carried an `include` joining
// taskRun <-> backgroundWorker. Every findFirst arg has include undefined.
for (const args of findFirstArgs) {
expect(args?.include).toBeUndefined();
}
} finally {
await engine.quit();
}
}
);
});
Loading