Matt Aitken
ef04cc39ef
## Problem
`ClickHouseRunsRepository.listRunIds` / `listRuns` order results by the
composite key `(created_at, run_id)`, but the cursor predicate cut on
`run_id` **alone**:
```ts
.where("run_id < {runId: String}", { runId: cursor })
.orderBy("created_at DESC, run_id DESC")
```
This is only sound when `run_id` lexicographic order matches
`created_at` order. `run_id`s are cuids — only coarsely time-sortable —
so when a burst of runs is created within a sub-second window, the two
orders can diverge. When they do, the next-page predicate (`run_id <
cursor`, where `cursor` is the *last* page element = the smallest
`created_at`, not necessarily the smallest `run_id`):
- **re-includes** rows already returned on a previous page (duplicates),
and
- **skips** rows it should have returned (silent data loss).
For bulk **replay** this caused runs to be replayed more than once
(replay has no idempotency guard). For the dashboard and the `runs.list`
API it could silently repeat or skip runs at page boundaries.
## Fix
Make the cursor predicate match the composite ordering:
- Cursors now encode the full `(created_at, run_id)` key as an **opaque
URL-safe base64 token**
(`base64url({"c":<createdAtMs>,"r":"<runId>"})`), and the query cuts on
the matching tuple — `(created_at, run_id) < (…)` forward / `> (…)`
backward.
- The `ORDER BY` is unchanged, so the query stays aligned with the
table's primary key — no performance regression (the tuple range
predicate is actually more index-friendly than `run_id <` alone).
- Cursors are **server-issued opaque tokens** (the SDK only echoes
`pagination.next` / `pagination.previous` back), so this needs **no
client/SDK update**. Legacy cursors were the bare internal `run_id`;
they're detected by decode failure (a cuid isn't a valid base64-wrapped
JSON payload) and fall back to the old `run_id`-only predicate, so
in-flight cursors keep working and drain naturally. New cursors also no
longer expose a bare internal run id.
- `listRunIds` is now the single cursor-aware list primitive: it returns
`{ runIds, pagination: { nextCursor, previousCursor } }`, and `listRuns`
builds on it (one place constructs cursors). Bulk actions consume the
same method and advance by `pagination.nextCursor`, finishing when it's
`null`.
- `getTaskRunsQueryBuilder` now also selects
`toUnixTimestamp64Milli(created_at) AS created_at_ms`, using a dedicated
`TaskRunListQueryResult` schema. The shared `TaskRunV2QueryResult` stays
`run_id`-only so the run-engine pending-version lookup
(`getPendingVersionIdsQueryBuilder`, which selects only `run_id`)
doesn't fail validation on a column it doesn't query.
## Tests
New `runsRepositoryCursor.test.ts` (testcontainer-backed, real
Postgres→ClickHouse replication):
- **forward** pagination returns every run exactly once when `run_id`
order is the reverse of `created_at` order (reproduces the
duplicate/skip bug — fails on `main`; this
walk-until-`nextCursor`-null-and-assert-complete is exactly the bulk
action's iteration),
- **backward** pagination round-trips to the previous page across a
boundary,
- **legacy** bare-`run_id` cursor still uses the old predicate
(backwards compatibility).
The existing `runsRepository` suites (part1–4) still pass; `part4`'s
`count new runs with listRunIds` test was updated for the new `{ runIds,
pagination }` return shape, and the `clickhouse` `taskRuns`
query-builder snapshots were regenerated for the added `created_at_ms`
column.
## Notes
- Separate, pre-existing issue (out of scope, not introduced here):
`listRuns`' backward display-slicing (`rows.slice(1, size+1)` when
`hasMore`) has an off-by-one that can return a straddled page. Tracked
separately.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
456 lines
13 KiB
TypeScript
456 lines
13 KiB
TypeScript
import { describe, expect, vi } from "vitest";
|
|
|
|
// Mock the db prisma client
|
|
vi.mock("~/db.server", () => ({
|
|
prisma: {},
|
|
$replica: {},
|
|
}));
|
|
|
|
import { replicationContainerTest } 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 4/4)", () => {
|
|
replicationContainerTest(
|
|
"should filter runs by date range (from/to)",
|
|
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",
|
|
},
|
|
});
|
|
|
|
const now = new Date();
|
|
const yesterday = new Date(now.getTime() - 24 * 60 * 60 * 1000);
|
|
const tomorrow = new Date(now.getTime() + 24 * 60 * 60 * 1000);
|
|
|
|
// Create runs with different creation dates
|
|
await prisma.taskRun.create({
|
|
data: {
|
|
friendlyId: "run_yesterday",
|
|
taskIdentifier: "my-task",
|
|
createdAt: yesterday,
|
|
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_today",
|
|
taskIdentifier: "my-task",
|
|
createdAt: now,
|
|
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_tomorrow",
|
|
taskIdentifier: "my-task",
|
|
createdAt: tomorrow,
|
|
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 date range (from yesterday to today)
|
|
const { runs } = await runsRepository.listRuns({
|
|
page: { size: 10 },
|
|
projectId: project.id,
|
|
environmentId: runtimeEnvironment.id,
|
|
organizationId: organization.id,
|
|
from: yesterday.getTime(),
|
|
to: now.getTime(),
|
|
});
|
|
|
|
expect(runs).toHaveLength(2);
|
|
expect(runs.map((r) => r.friendlyId).sort()).toEqual(["run_today", "run_yesterday"]);
|
|
}
|
|
);
|
|
|
|
replicationContainerTest(
|
|
"should handle multiple filters combined",
|
|
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 combinations of properties
|
|
await prisma.taskRun.create({
|
|
data: {
|
|
friendlyId: "run_match",
|
|
taskIdentifier: "task-1",
|
|
taskVersion: "1.0.0",
|
|
status: "COMPLETED_SUCCESSFULLY",
|
|
isTest: false,
|
|
runTags: ["urgent"],
|
|
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_no_match_task",
|
|
taskIdentifier: "task-2", // Different task
|
|
taskVersion: "1.0.0",
|
|
status: "COMPLETED_SUCCESSFULLY",
|
|
isTest: false,
|
|
runTags: ["urgent"],
|
|
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_no_match_status",
|
|
taskIdentifier: "task-1",
|
|
taskVersion: "1.0.0",
|
|
status: "PENDING", // Different status
|
|
isTest: false,
|
|
runTags: ["urgent"],
|
|
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 combining multiple filters
|
|
const { runs } = await runsRepository.listRuns({
|
|
page: { size: 10 },
|
|
projectId: project.id,
|
|
environmentId: runtimeEnvironment.id,
|
|
organizationId: organization.id,
|
|
tasks: ["task-1"],
|
|
versions: ["1.0.0"],
|
|
statuses: ["COMPLETED_SUCCESSFULLY"],
|
|
isTest: false,
|
|
tags: ["urgent"],
|
|
});
|
|
|
|
expect(runs).toHaveLength(1);
|
|
expect(runs[0].friendlyId).toBe("run_match");
|
|
}
|
|
);
|
|
|
|
replicationContainerTest(
|
|
"should handle pagination correctly",
|
|
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 multiple runs for pagination testing
|
|
const runs = [];
|
|
for (let i = 1; i <= 5; i++) {
|
|
const run = await prisma.taskRun.create({
|
|
data: {
|
|
friendlyId: `run_${i}`,
|
|
taskIdentifier: "my-task",
|
|
payload: JSON.stringify({ foo: "bar" }),
|
|
traceId: `123${i}`,
|
|
spanId: `123${i}`,
|
|
queue: "test",
|
|
runtimeEnvironmentId: runtimeEnvironment.id,
|
|
projectId: project.id,
|
|
organizationId: organization.id,
|
|
environmentType: "DEVELOPMENT",
|
|
engine: "V2",
|
|
},
|
|
});
|
|
runs.push(run);
|
|
}
|
|
|
|
await setTimeout(1000);
|
|
|
|
const runsRepository = new RunsRepository({
|
|
prisma,
|
|
clickhouse,
|
|
});
|
|
|
|
// Test first page
|
|
const firstPage = await runsRepository.listRuns({
|
|
page: { size: 2 },
|
|
projectId: project.id,
|
|
environmentId: runtimeEnvironment.id,
|
|
organizationId: organization.id,
|
|
});
|
|
|
|
expect(firstPage.runs).toHaveLength(2);
|
|
expect(firstPage.pagination.nextCursor).toBeTruthy();
|
|
expect(firstPage.pagination.previousCursor).toBe(null);
|
|
|
|
// Test next page using cursor
|
|
const secondPage = await runsRepository.listRuns({
|
|
page: {
|
|
size: 2,
|
|
cursor: firstPage.pagination.nextCursor!,
|
|
direction: "forward",
|
|
},
|
|
projectId: project.id,
|
|
environmentId: runtimeEnvironment.id,
|
|
organizationId: organization.id,
|
|
});
|
|
|
|
expect(secondPage.runs).toHaveLength(2);
|
|
expect(secondPage.pagination.nextCursor).toBeTruthy();
|
|
expect(secondPage.pagination.previousCursor).toBeTruthy();
|
|
}
|
|
);
|
|
|
|
replicationContainerTest(
|
|
"should count new runs with listRunIds",
|
|
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",
|
|
},
|
|
});
|
|
|
|
const taskRun = await prisma.taskRun.create({
|
|
data: {
|
|
friendlyId: "run_has_new",
|
|
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 baseOptions = {
|
|
projectId: project.id,
|
|
environmentId: runtimeEnvironment.id,
|
|
organizationId: organization.id,
|
|
};
|
|
|
|
const createdAtMs = taskRun.createdAt.getTime();
|
|
|
|
const newRunIdsBefore = await runsRepository.listRunIds({
|
|
...baseOptions,
|
|
from: createdAtMs - 1,
|
|
page: { size: 100 },
|
|
});
|
|
expect(newRunIdsBefore.runIds.length).toBeGreaterThanOrEqual(1);
|
|
|
|
const newRunIdsAfter = await runsRepository.listRunIds({
|
|
...baseOptions,
|
|
from: createdAtMs + 60_000,
|
|
page: { size: 100 },
|
|
});
|
|
expect(newRunIdsAfter.runIds).toHaveLength(0);
|
|
|
|
const fromBeforeRun = createdAtMs - 1;
|
|
|
|
const matchingTaskIds = await runsRepository.listRunIds({
|
|
...baseOptions,
|
|
from: fromBeforeRun,
|
|
tasks: ["my-task"],
|
|
page: { size: 100 },
|
|
});
|
|
expect(matchingTaskIds.runIds.length).toBeGreaterThanOrEqual(1);
|
|
|
|
const otherTaskIds = await runsRepository.listRunIds({
|
|
...baseOptions,
|
|
from: fromBeforeRun,
|
|
tasks: ["other-task"],
|
|
page: { size: 100 },
|
|
});
|
|
expect(otherTaskIds.runIds).toHaveLength(0);
|
|
}
|
|
);
|
|
});
|